x0x 0.32.0

Agent-to-agent gossip network for AI systems — no winners, no losers, just cooperation
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

//! Connect diagnostics — counter surface for connect allow/deny decisions.
//!
//! Minimal-scope mirror of `exec/diagnostics.rs`. The forwarder (shipped in
//! #183) calls [`ConnectDiagnostics::record_allowed`] /
//! [`record_denied`](ConnectDiagnostics::record_denied) on each authorized or
//! denied forward when connect is enabled, so the counters reflect live
//! traffic; they read `0` when connect is disabled (the default). Intentionally
//! no warnings/sessions machinery (connect diagnostics is counters + summary
//! only).

use std::collections::HashMap;
use std::sync::Mutex;

use serde::Serialize;

use crate::connect::acl::ConnectAclSummary;
use crate::connect::gate::ConnectDenialReason;

/// Atomic allow/deny counters + per-reason denial breakdown.
#[derive(Debug)]
pub struct ConnectDiagnostics {
    streams_allowed: std::sync::atomic::AtomicU64,
    streams_denied: std::sync::atomic::AtomicU64,
    denial_breakdown: Mutex<HashMap<ConnectDenialReason, u64>>,
    acl_summary: ConnectAclSummary,
}

impl ConnectDiagnostics {
    /// Construct from the loaded policy's summary.
    #[must_use]
    pub fn new(summary: ConnectAclSummary) -> Self {
        Self {
            streams_allowed: std::sync::atomic::AtomicU64::new(0),
            streams_denied: std::sync::atomic::AtomicU64::new(0),
            denial_breakdown: Mutex::new(HashMap::new()),
            acl_summary: summary,
        }
    }

    /// Record an allowed connect stream.
    pub fn record_allowed(&self) {
        self.streams_allowed
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    }

    /// Record a denied connect stream, bucketed by reason.
    pub fn record_denied(&self, reason: ConnectDenialReason) {
        self.streams_denied
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        if let Ok(mut map) = self.denial_breakdown.lock() {
            *map.entry(reason).or_insert(0) += 1;
        }
    }

    /// Snapshot for the `/diagnostics/connect` endpoint.
    #[must_use]
    pub fn snapshot(&self) -> ConnectDiagnosticsSnapshot {
        let allowed = self
            .streams_allowed
            .load(std::sync::atomic::Ordering::Relaxed);
        let denied = self
            .streams_denied
            .load(std::sync::atomic::Ordering::Relaxed);
        let breakdown = self
            .denial_breakdown
            .lock()
            .map(|m| m.iter().map(|(k, v)| (*k, *v)).collect())
            .unwrap_or_default();
        ConnectDiagnosticsSnapshot {
            streams_allowed: allowed,
            streams_denied: denied,
            denial_breakdown: breakdown,
            acl_summary: self.acl_summary.clone(),
        }
    }
}

/// Serializable diagnostics snapshot.
#[derive(Debug, Clone, Serialize)]
pub struct ConnectDiagnosticsSnapshot {
    pub streams_allowed: u64,
    pub streams_denied: u64,
    pub denial_breakdown: HashMap<ConnectDenialReason, u64>,
    pub acl_summary: ConnectAclSummary,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn record_allowed_and_denied_updates_counters_and_breakdown() {
        let diag = ConnectDiagnostics::new(ConnectAclSummary {
            enabled: true,
            loaded_from: "/tmp/x.toml".to_string(),
            loaded_at_unix_ms: 0,
            allow_entry_count: 1,
            target_entry_count: 1,
            disabled_reason: None,
        });
        diag.record_allowed();
        diag.record_allowed();
        diag.record_denied(ConnectDenialReason::ConnectDisabled);
        diag.record_denied(ConnectDenialReason::TargetNotAllowed);

        let snap = diag.snapshot();
        assert_eq!(snap.streams_allowed, 2);
        assert_eq!(snap.streams_denied, 2);
        assert_eq!(
            snap.denial_breakdown
                .get(&ConnectDenialReason::ConnectDisabled),
            Some(&1)
        );
        assert_eq!(
            snap.denial_breakdown
                .get(&ConnectDenialReason::TargetNotAllowed),
            Some(&1)
        );
        assert!(snap.acl_summary.enabled);
    }
}