monitord/

dbus_stats.rs

//! # dbus_stats module
//!
//! Handle getting statistics of our Dbus daemon/broker

use std::collections::HashMap;
use std::fs;
use std::io;
use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock};

use thiserror::Error;
use tokio::sync::RwLock;
use tokio::sync::Semaphore;
use tracing::Instrument;
use tracing::{debug, error};
use uzers::get_user_by_uid;
use zbus::fdo::{DBusProxy, StatsProxy};
use zbus::names::BusName;
use zvariant::{Dict, OwnedValue, Value};

use crate::MachineStats;

#[derive(Error, Debug)]
pub enum MonitordDbusStatsError {
    #[error("D-Bus error: {0}")]
    ZbusError(#[from] zbus::Error),
    #[error("D-Bus fdo error: {0}")]
    FdoError(#[from] zbus::fdo::Error),
    #[error("Task join error: {0}")]
    JoinError(#[from] tokio::task::JoinError),
}

// Unfortunately, various DBus daemons (ex: dbus-broker and dbus-daemon)
// represent stats differently. Moreover, the stats vary across versions of the same daemon.
// Hence, the code uses flexible approach providing max available information.

/// Per-peer resource accounting from dbus-broker's PeerAccounting stats.
/// Each peer represents a single D-Bus connection identified by a unique bus name.
#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, PartialEq)]
pub struct DBusBrokerPeerAccounting {
    /// Unique D-Bus connection name (e.g. ":1.42")
    pub id: String,
    /// Well-known bus name owned by this peer, if any (e.g. "org.freedesktop.NetworkManager")
    pub well_known_name: Option<String>,

    // credentials
    /// Unix UID of the process owning this D-Bus connection
    pub unix_user_id: Option<u32>,
    /// PID of the process owning this D-Bus connection
    pub process_id: Option<u32>,
    /// Unix supplementary group IDs of the process owning this connection
    pub unix_group_ids: Option<Vec<u32>>,
    // ignoring LinuxSecurityLabel
    // pub linux_security_label: Option<String>,

    // stats
    /// Number of bus name objects held by this peer
    pub name_objects: Option<u32>,
    /// Bytes consumed by match rules registered by this peer
    pub match_bytes: Option<u32>,
    /// Number of match rules registered by this peer for signal filtering
    pub matches: Option<u32>,
    /// Number of pending reply objects (outstanding method calls awaiting replies)
    pub reply_objects: Option<u32>,
    /// Total bytes received by this peer from the bus
    pub incoming_bytes: Option<u32>,
    /// Total file descriptors received by this peer via D-Bus fd-passing
    pub incoming_fds: Option<u32>,
    /// Total bytes sent by this peer to the bus
    pub outgoing_bytes: Option<u32>,
    /// Total file descriptors sent by this peer via D-Bus fd-passing
    pub outgoing_fds: Option<u32>,
    /// Bytes used for D-Bus activation requests by this peer
    pub activation_request_bytes: Option<u32>,
    /// File descriptors used for D-Bus activation requests by this peer
    pub activation_request_fds: Option<u32>,
}

impl DBusBrokerPeerAccounting {
    /// Returns true if the peer has a well-known name
    pub fn has_well_known_name(&self) -> bool {
        self.well_known_name.is_some()
    }

    /// Returns the well-known name if present, otherwise falls back to the unique D-Bus connection ID
    pub fn get_name(&self) -> &str {
        self.well_known_name.as_deref().unwrap_or(&self.id)
    }

    pub fn get_cgroup_name(&self) -> Result<String, io::Error> {
        let pid = self
            .process_id
            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "missing process_id"))?;

        let path = format!("/proc/{}/cgroup", pid);
        let content = fs::read_to_string(&path)?;

        // ex: 0::/system.slice/metalos.classic.metald.service
        let cgroup = content.strip_prefix("0::").ok_or_else(|| {
            io::Error::new(io::ErrorKind::InvalidData, "unexpected cgroup format")
        })?;

        Ok(cgroup.trim().trim_matches('/').replace('/', "-"))
    }
}

/// Aggregated D-Bus resource accounting grouped by cgroup.
/// Not directly present in dbus-broker stats; computed by summing peer stats that share a cgroup.
/// Grouping by cgroup reduces metric cardinality while still identifying abusive clients.
#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, PartialEq)]
pub struct DBusBrokerCGroupAccounting {
    /// Cgroup path with slashes replaced by dashes (e.g. "system.slice-sshd.service")
    pub name: String,

    // stats (aggregated sums across all peers in this cgroup)
    /// Total bus name objects held by peers in this cgroup
    pub name_objects: Option<u32>,
    /// Total bytes consumed by match rules from peers in this cgroup
    pub match_bytes: Option<u32>,
    /// Total match rules registered by peers in this cgroup
    pub matches: Option<u32>,
    /// Total pending reply objects from peers in this cgroup
    pub reply_objects: Option<u32>,
    /// Total bytes received by peers in this cgroup
    pub incoming_bytes: Option<u32>,
    /// Total file descriptors received by peers in this cgroup
    pub incoming_fds: Option<u32>,
    /// Total bytes sent by peers in this cgroup
    pub outgoing_bytes: Option<u32>,
    /// Total file descriptors sent by peers in this cgroup
    pub outgoing_fds: Option<u32>,
    /// Total activation request bytes from peers in this cgroup
    pub activation_request_bytes: Option<u32>,
    /// Total activation request file descriptors from peers in this cgroup
    pub activation_request_fds: Option<u32>,
}

impl DBusBrokerCGroupAccounting {
    pub fn combine_with_peer(&mut self, peer: &DBusBrokerPeerAccounting) {
        fn sum(a: &mut Option<u32>, b: &Option<u32>) {
            *a = match (a.take(), b) {
                (Some(x), Some(y)) => Some(x + y),
                (Some(x), None) => Some(x),
                (None, Some(y)) => Some(*y),
                (None, None) => None,
            };
        }

        sum(&mut self.name_objects, &peer.name_objects);
        sum(&mut self.match_bytes, &peer.match_bytes);
        sum(&mut self.matches, &peer.matches);
        sum(&mut self.reply_objects, &peer.reply_objects);
        sum(&mut self.incoming_bytes, &peer.incoming_bytes);
        sum(&mut self.incoming_fds, &peer.incoming_fds);
        sum(&mut self.outgoing_bytes, &peer.outgoing_bytes);
        sum(&mut self.outgoing_fds, &peer.outgoing_fds);
        sum(
            &mut self.activation_request_bytes,
            &peer.activation_request_bytes,
        );
        sum(
            &mut self.activation_request_fds,
            &peer.activation_request_fds,
        );
    }
}

/// Current/maximum resource pair as reported by dbus-broker's UserAccounting.
/// Note: dbus-broker stores the current value in inverted form; actual usage = max - cur.
#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, PartialEq)]
pub struct CurMaxPair {
    /// Remaining quota (inverted: actual usage = max - cur)
    pub cur: u32,
    /// Maximum allowed quota for this resource
    pub max: u32,
}

impl CurMaxPair {
    pub fn get_usage(&self) -> u32 {
        // There is a theoretical possibility of max < cur due to various factors.
        // I'll leave it for now to avoid premature optimizations.
        self.max - self.cur
    }
}

/// Per-user aggregated D-Bus resource limits and usage from dbus-broker's UserAccounting.
/// Each entry tracks quota consumption across all connections belonging to a Unix user.
#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, PartialEq)]
pub struct DBusBrokerUserAccounting {
    /// Unix user ID this accounting entry belongs to
    pub uid: u32,
    /// Username resolved from `uid` at parse time; falls back to the numeric UID string if unknown.
    pub username: String,

    /// Message byte quota: remaining (cur) and maximum (max) allowed bytes across all connections
    pub bytes: Option<CurMaxPair>,
    /// File descriptor quota: remaining (cur) and maximum (max) allowed FDs across all connections
    pub fds: Option<CurMaxPair>,
    /// Unattributed stale pidfd descriptors held by the system dbus-broker.
    ///
    /// dbus-broker exposes these through procfs, not D-Bus UserAccounting. The kernel
    /// reports dead pidfds as `Pid: -1` in `/proc/<dbus-broker>/fdinfo/<fd>`.
    /// The stale pidfd no longer exposes the original process credentials through
    /// procfs, so monitord surfaces the system-broker count on uid 0/root for metric
    /// compatibility. This does not mean root owns those stale pidfds.
    pub stale_fds: Option<u32>,
    /// Match rule quota: remaining (cur) and maximum (max) allowed match rules across all connections
    pub matches: Option<CurMaxPair>,
    /// Object quota: remaining (cur) and maximum (max) allowed objects (names, replies) across all connections
    pub objects: Option<CurMaxPair>,
    // UserUsage provides detailed breakdown of the aggregated numbers.
    // However, dbus-broker exposes usage as real values (not inverted, see CurMaxPair).
}

impl DBusBrokerUserAccounting {
    fn new(uid: u32) -> Self {
        let username = match get_user_by_uid(uid) {
            Some(user) => user.name().to_string_lossy().into_owned(),
            None => uid.to_string(),
        };

        Self {
            uid,
            username,
            ..Default::default()
        }
    }
}

/// D-Bus daemon/broker statistics from org.freedesktop.DBus.Debug.Stats.
/// Works with both dbus-daemon and dbus-broker; broker-specific fields are in separate maps.
#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default, Eq, PartialEq)]
pub struct DBusStats {
    /// Current D-Bus message serial number (monotonically increasing message counter)
    pub serial: Option<u32>,
    /// Number of fully authenticated active D-Bus connections
    pub active_connections: Option<u32>,
    /// Number of D-Bus connections still in the authentication handshake phase
    pub incomplete_connections: Option<u32>,
    /// Current number of registered bus names (well-known + unique)
    pub bus_names: Option<u32>,
    /// Peak (high-water mark) number of bus names ever registered simultaneously
    pub peak_bus_names: Option<u32>,
    /// Peak number of bus names registered by a single connection
    pub peak_bus_names_per_connection: Option<u32>,
    /// Current number of active signal match rules across all connections
    pub match_rules: Option<u32>,
    /// Peak number of match rules ever registered simultaneously
    pub peak_match_rules: Option<u32>,
    /// Peak number of match rules registered by a single connection
    pub peak_match_rules_per_connection: Option<u32>,
    /// Stale pidfd descriptors held by the system dbus-broker process.
    pub stale_fds: Option<u32>,

    /// Per-peer resource accounting (dbus-broker only), keyed by unique connection name
    pub dbus_broker_peer_accounting: Option<HashMap<String, DBusBrokerPeerAccounting>>,
    /// Per-cgroup resource accounting (dbus-broker only), keyed by cgroup name
    pub dbus_broker_cgroup_accounting: Option<HashMap<String, DBusBrokerCGroupAccounting>>,
    /// Per-user resource quota accounting (dbus-broker only), keyed by Unix UID
    pub dbus_broker_user_accounting: Option<HashMap<u32, DBusBrokerUserAccounting>>,
}

impl DBusStats {
    pub fn peer_accounting(&self) -> Option<&HashMap<String, DBusBrokerPeerAccounting>> {
        self.dbus_broker_peer_accounting.as_ref()
    }

    pub fn cgroup_accounting(&self) -> Option<&HashMap<String, DBusBrokerCGroupAccounting>> {
        self.dbus_broker_cgroup_accounting.as_ref()
    }

    pub fn user_accounting(&self) -> Option<&HashMap<u32, DBusBrokerUserAccounting>> {
        self.dbus_broker_user_accounting.as_ref()
    }
}

fn parse_ppid_from_stat(stat: &str) -> Option<u32> {
    let (_, after_comm) = stat.rsplit_once(") ")?;
    let mut fields = after_comm.split_whitespace();
    let _state = fields.next()?;
    fields.next()?.parse().ok()
}

fn proc_cmdline_args(path: &Path) -> io::Result<Vec<String>> {
    let bytes = fs::read(path)?;
    Ok(bytes
        .split(|b| *b == b'\0')
        .filter(|arg| !arg.is_empty())
        .map(|arg| String::from_utf8_lossy(arg).into_owned())
        .collect())
}

fn proc_pid_dirs(proc_root: &Path) -> io::Result<Vec<(u32, std::path::PathBuf)>> {
    let mut result = Vec::new();
    for entry in fs::read_dir(proc_root)? {
        let entry = entry?;
        let file_name = entry.file_name();
        let Some(name) = file_name.to_str() else {
            continue;
        };
        let Ok(pid) = name.parse::<u32>() else {
            continue;
        };
        result.push((pid, entry.path()));
    }
    Ok(result)
}

fn find_system_dbus_broker_pid(proc_root: &Path) -> io::Result<Option<u32>> {
    let pid_dirs = proc_pid_dirs(proc_root)?;
    let mut launcher_pids = Vec::new();

    for (pid, path) in &pid_dirs {
        let args = match proc_cmdline_args(&path.join("cmdline")) {
            Ok(args) => args,
            Err(_) => continue,
        };

        let is_system_launcher = args
            .first()
            .map(|arg| arg.ends_with("dbus-broker-launch"))
            .unwrap_or(false)
            && (args
                .windows(2)
                .any(|window| window[0] == "--scope" && window[1] == "system")
                || args.iter().any(|arg| arg == "--scope=system"));

        if is_system_launcher {
            launcher_pids.push(*pid);
        }
    }

    for (pid, path) in &pid_dirs {
        let args = match proc_cmdline_args(&path.join("cmdline")) {
            Ok(args) => args,
            Err(_) => continue,
        };

        let is_broker = args
            .first()
            .map(|arg| arg.ends_with("dbus-broker"))
            .unwrap_or(false);
        if !is_broker {
            continue;
        }

        let stat = match fs::read_to_string(path.join("stat")) {
            Ok(stat) => stat,
            Err(_) => continue,
        };
        let Some(ppid) = parse_ppid_from_stat(&stat) else {
            continue;
        };
        if launcher_pids.contains(&ppid) {
            return Ok(Some(*pid));
        }
    }

    Ok(None)
}

fn is_stale_pidfd(fd_path: &Path, fdinfo_path: &Path) -> bool {
    let Ok(target) = fs::read_link(fd_path) else {
        return false;
    };
    if target != Path::new("anon_inode:[pidfd]") {
        return false;
    }

    let Ok(fdinfo) = fs::read_to_string(fdinfo_path) else {
        return false;
    };
    fdinfo.lines().any(|line| {
        let Some(pid) = line.strip_prefix("Pid:") else {
            return false;
        };
        pid.trim() == "-1"
    })
}

fn count_stale_pidfds(proc_root: &Path, pid: u32) -> io::Result<u32> {
    let fd_dir = proc_root.join(pid.to_string()).join("fd");
    let fdinfo_dir = proc_root.join(pid.to_string()).join("fdinfo");
    let mut count: u32 = 0;

    for entry in fs::read_dir(fdinfo_dir)? {
        let entry = entry?;
        let fd_name = entry.file_name();
        let fd_path = fd_dir.join(&fd_name);
        if is_stale_pidfd(&fd_path, &entry.path()) {
            count = count.saturating_add(1);
        }
    }

    Ok(count)
}

#[cfg(test)]
fn collect_system_dbus_broker_stale_fds_from_proc(proc_root: &Path) -> io::Result<Option<u32>> {
    let Some(pid) = find_system_dbus_broker_pid(proc_root)? else {
        return Ok(None);
    };

    Ok(Some(count_stale_pidfds(proc_root, pid)?))
}

fn collect_system_dbus_broker_stale_fds_with_cache(
    proc_root: &Path,
    broker_pid_cache: &Mutex<Option<u32>>,
) -> io::Result<Option<u32>> {
    let cached_pid = *broker_pid_cache
        .lock()
        .map_err(|_| io::Error::other("dbus-broker pid cache poisoned"))?;

    if let Some(pid) = cached_pid {
        match count_stale_pidfds(proc_root, pid) {
            Ok(count) => return Ok(Some(count)),
            Err(err) if err.kind() == io::ErrorKind::NotFound => {
                if let Ok(mut cached_pid) = broker_pid_cache.lock() {
                    if *cached_pid == Some(pid) {
                        *cached_pid = None;
                    }
                }
            }
            Err(err) => return Err(err),
        }
    }

    let Some(pid) = find_system_dbus_broker_pid(proc_root)? else {
        return Ok(None);
    };

    let count = count_stale_pidfds(proc_root, pid)?;
    let mut cached_pid = broker_pid_cache
        .lock()
        .map_err(|_| io::Error::other("dbus-broker pid cache poisoned"))?;
    *cached_pid = Some(pid);

    Ok(Some(count))
}

fn is_expected_procfs_error(err: &io::Error) -> bool {
    matches!(
        err.kind(),
        io::ErrorKind::NotFound | io::ErrorKind::PermissionDenied
    )
}

fn collect_system_dbus_broker_stale_fds() -> Option<u32> {
    static SYSTEM_DBUS_BROKER_PID: OnceLock<Mutex<Option<u32>>> = OnceLock::new();

    let broker_pid_cache = SYSTEM_DBUS_BROKER_PID.get_or_init(|| Mutex::new(None));

    match collect_system_dbus_broker_stale_fds_with_cache(Path::new("/proc"), broker_pid_cache) {
        Ok(stale_fds) => stale_fds,
        Err(err) => {
            if is_expected_procfs_error(&err) {
                debug!("could not collect dbus-broker stale fd stats: {}", err);
            } else {
                error!("failed to collect dbus-broker stale fd stats: {}", err);
            }
            None
        }
    }
}

fn get_u32(dict: &Dict, key: &str) -> Option<u32> {
    let value_key: Value = key.into();
    dict.get(&value_key).ok().and_then(|v| match v.flatten() {
        Some(Value::U32(val)) => Some(*val),
        _ => None,
    })
}

fn get_u32_vec(dict: &Dict, key: &str) -> Option<Vec<u32>> {
    let value_key: Value = key.into();
    dict.get(&value_key).ok().and_then(|v| match v.flatten() {
        Some(Value::Array(array)) => {
            let vec: Vec<u32> = array
                .iter()
                .filter_map(|item| {
                    if let Value::U32(num) = item {
                        Some(*num)
                    } else {
                        None
                    }
                })
                .collect();

            Some(vec)
        }
        _ => None,
    })
}

/* Parse DBusBrokerPeerAccounting from OwnedValue.
 * Expected structure:
 * struct {
 *     string ":1.2197907"
 *     array [
 *         dict entry(
 *              string "UnixUserID"
 *              variant uint32 0
 *         )
 *         ... other fields
 *     ]
 *     array [
 *         dict entry(
 *              string "NameObjects"
 *              uint32 1
 *         )
 *         ... other fields
 *     ]
 * }
 */

fn parse_peer_struct(
    peer_value: &Value,
    well_known_to_peer_names: &HashMap<String, String>,
) -> Option<DBusBrokerPeerAccounting> {
    let peer_struct = match peer_value {
        Value::Structure(peer_struct) => peer_struct,
        _ => return None,
    };

    match peer_struct.fields() {
        [Value::Str(id), Value::Dict(credentials), Value::Dict(stats), ..] => {
            Some(DBusBrokerPeerAccounting {
                id: id.to_string(),
                well_known_name: well_known_to_peer_names.get(id.as_str()).cloned(),
                unix_user_id: get_u32(credentials, "UnixUserID"),
                process_id: get_u32(credentials, "ProcessID"),
                unix_group_ids: get_u32_vec(credentials, "UnixGroupIDs"),
                name_objects: get_u32(stats, "NameObjects"),
                match_bytes: get_u32(stats, "MatchBytes"),
                matches: get_u32(stats, "Matches"),
                reply_objects: get_u32(stats, "ReplyObjects"),
                incoming_bytes: get_u32(stats, "IncomingBytes"),
                incoming_fds: get_u32(stats, "IncomingFds"),
                outgoing_bytes: get_u32(stats, "OutgoingBytes"),
                outgoing_fds: get_u32(stats, "OutgoingFds"),
                activation_request_bytes: get_u32(stats, "ActivationRequestBytes"),
                activation_request_fds: get_u32(stats, "ActivationRequestFds"),
            })
        }
        _ => None,
    }
}

async fn parse_peer_accounting(
    connection: &zbus::Connection,
    config: &crate::config::Config,
    owned_value: Option<&OwnedValue>,
) -> Result<Option<Vec<DBusBrokerPeerAccounting>>, MonitordDbusStatsError> {
    // need to keep collecting peer stats when cgroup_stats=true
    // since cgroup_stats is a derivative of peer stats
    if !config.dbus_stats.peer_stats && !config.dbus_stats.cgroup_stats {
        return Ok(None);
    }

    let value: &Value = match owned_value {
        Some(v) => v,
        None => return Ok(None),
    };

    let peers_value = match value {
        Value::Array(peers_value) => peers_value,
        _ => return Ok(None),
    };

    let well_known_to_peer_names = get_well_known_to_peer_names(connection, config).await?;

    let result = peers_value
        .iter()
        .filter_map(|peer| parse_peer_struct(peer, &well_known_to_peer_names))
        .collect();

    Ok(Some(result))
}

fn filter_and_collect_peer_accounting(
    config: &crate::config::Config,
    peers: Option<&Vec<DBusBrokerPeerAccounting>>,
) -> Option<HashMap<String, DBusBrokerPeerAccounting>> {
    // reject collecting peer stats when told so
    if !config.dbus_stats.peer_stats {
        return None;
    }

    let result = peers?
        .iter()
        .filter(|peer| {
            if config.dbus_stats.peer_well_known_names_only && !peer.has_well_known_name() {
                return false;
            }

            let id = peer.id.as_str();
            let name = peer.get_name();
            if config.dbus_stats.peer_blocklist.contains(id)
                || config.dbus_stats.peer_blocklist.contains(name)
            {
                return false;
            }

            if !config.dbus_stats.peer_allowlist.is_empty()
                && !config.dbus_stats.peer_allowlist.contains(id)
                && !config.dbus_stats.peer_allowlist.contains(name)
            {
                return false;
            }

            true
        })
        .map(|peer| (peer.id.clone(), peer.clone()))
        .collect();

    Some(result)
}

fn filter_and_collect_cgroup_accounting(
    config: &crate::config::Config,
    peers: Option<&Vec<DBusBrokerPeerAccounting>>,
) -> Option<HashMap<String, DBusBrokerCGroupAccounting>> {
    // reject collecting cgroup stats when told so
    if !config.dbus_stats.cgroup_stats {
        return None;
    }

    let mut result: HashMap<String, DBusBrokerCGroupAccounting> = HashMap::new();

    for peer in peers?.iter() {
        let cgroup_name = match peer.get_cgroup_name() {
            Ok(name) => name,
            Err(err) => {
                error!("Failed to get cgroup name for peer {}: {}", peer.id, err);
                continue;
            }
        };

        if config.dbus_stats.cgroup_blocklist.contains(&cgroup_name) {
            continue;
        }

        if !config.dbus_stats.cgroup_allowlist.is_empty()
            && !config.dbus_stats.cgroup_allowlist.contains(&cgroup_name)
        {
            continue;
        }

        let entry =
            result
                .entry(cgroup_name.clone())
                .or_insert_with(|| DBusBrokerCGroupAccounting {
                    name: cgroup_name,
                    ..Default::default()
                });

        entry.combine_with_peer(peer);
    }

    Some(result)
}

/* Parse DBusBrokerUserAccounting from OwnedValue.
 * Expected structure:
 * struct {
 *     uint32 0
 *     array [
 *         struct {
 *             string "Bytes"
 *             uint32 536843240
 *             uint32 536870912
 *         }
 *         ... more fields
 *     ]
 *     # TODO parse usages, ignoring for now
 *     # see src/bus/driver.c:2258
 *     # the part below is not parsed
 *     array [
 *         dict entry(
 *             uint32 0
 *             array [
 *             dict entry(
 *                 string "Bytes"
 *                 uint32 27672
 *             )
 *             ... more fields
 *             ]
 *         )
 *     ]
 * }
 */

fn parse_user_struct(user_value: &Value) -> Option<DBusBrokerUserAccounting> {
    let user_struct = match user_value {
        Value::Structure(user_struct) => user_struct,
        _ => return None,
    };

    match user_struct.fields() {
        [Value::U32(uid), Value::Array(user_stats), ..] => {
            let mut user = DBusBrokerUserAccounting::new(*uid);
            for user_stat in user_stats.iter() {
                if let Value::Structure(user_stat) = user_stat {
                    if let [Value::Str(name), Value::U32(cur), Value::U32(max), ..] =
                        user_stat.fields()
                    {
                        let pair = CurMaxPair {
                            cur: *cur,
                            max: *max,
                        };
                        match name.as_str() {
                            "Bytes" => user.bytes = Some(pair),
                            "Fds" => user.fds = Some(pair),
                            "Matches" => user.matches = Some(pair),
                            "Objects" => user.objects = Some(pair),
                            _ => {} // ignore other fields
                        }
                    }
                }
            }

            Some(user)
        }
        _ => None,
    }
}

fn parse_user_accounting(
    config: &crate::config::Config,
    owned_value: &OwnedValue,
) -> Option<HashMap<u32, DBusBrokerUserAccounting>> {
    // reject collecting user stats when told so
    if !config.dbus_stats.user_stats {
        return None;
    }

    let value: &Value = owned_value;
    let users_value = match value {
        Value::Array(users_value) => users_value,
        _ => return None,
    };

    let result = users_value
        .iter()
        .filter_map(parse_user_struct)
        .filter(|user| {
            let uid = user.uid.to_string();
            if config.dbus_stats.user_blocklist.contains(&uid)
                || config.dbus_stats.user_blocklist.contains(&user.username)
            {
                return false;
            }

            if !config.dbus_stats.user_allowlist.is_empty()
                && !config.dbus_stats.user_allowlist.contains(&uid)
                && !config.dbus_stats.user_allowlist.contains(&user.username)
            {
                return false;
            }

            true
        })
        .map(|user| (user.uid, user))
        .collect();

    Some(result)
}

async fn get_well_known_to_peer_names(
    connection: &zbus::Connection,
    config: &crate::config::Config,
) -> Result<HashMap<String, String>, MonitordDbusStatsError> {
    let dbus_proxy: DBusProxy<'static> = DBusProxy::builder(connection)
        .cache_properties(zbus::proxy::CacheProperties::No)
        .build()
        .await?;

    let dbus_names = dbus_proxy.list_names().await?;

    // Each well-known name's owner is an independent D-Bus round-trip. A
    // semaphore (rather than an unbounded join) caps how many lookups are in
    // flight at once: a burst of simultaneous D-Bus calls could itself worsen
    // host-level IPC contention on hosts where per-call latency is already
    // elevated — the same failure mode units.rs's per-unit loop guards against.
    let semaphore = Arc::new(Semaphore::new(
        config.dbus_stats.peer_name_concurrency.max(1) as usize,
    ));
    let parent_span = tracing::Span::current();
    let mut join_set = tokio::task::JoinSet::new();
    for owned_busname in dbus_names {
        if let BusName::WellKnown(_) = &*owned_busname {
            let dbus_proxy = dbus_proxy.clone();
            let semaphore = Arc::clone(&semaphore);
            let parent_span = parent_span.clone();
            join_set.spawn(async move {
                let _permit = semaphore
                    .acquire()
                    .await
                    .expect("semaphore closed unexpectedly");
                let span = tracing::debug_span!(
                    parent: &parent_span,
                    "peer_name_lookup",
                    name = %owned_busname
                );
                async {
                    let owner = dbus_proxy.get_name_owner((&owned_busname).into()).await?;
                    Ok::<_, MonitordDbusStatsError>((owner.to_string(), owned_busname.to_string()))
                }
                .instrument(span)
                .await
            });
        }
    }

    let mut result = HashMap::new();
    while let Some(joined) = join_set.join_next().await {
        let (owner, name) = joined??;
        result.insert(owner, name);
    }

    Ok(result)
}

/// Pull all units from dbus and count how system is setup and behaving
async fn parse_dbus_stats_inner(
    config: &crate::config::Config,
    connection: &zbus::Connection,
    collect_stale_fds: bool,
) -> Result<DBusStats, MonitordDbusStatsError> {
    let stats_proxy = StatsProxy::builder(connection)
        .cache_properties(zbus::proxy::CacheProperties::No)
        .build()
        .await?;

    let stale_fds_task = if collect_stale_fds && config.dbus_stats.stale_fd_stats {
        Some(tokio::task::spawn_blocking(
            collect_system_dbus_broker_stale_fds,
        ))
    } else {
        None
    };

    let stats = stats_proxy.get_stats().await?;
    let peers = parse_peer_accounting(
        connection,
        config,
        stats.rest().get("org.bus1.DBus.Debug.Stats.PeerAccounting"),
    )
    .await?;

    let stale_fds = match stale_fds_task {
        Some(task) => match task.await {
            Ok(stale_fds) => stale_fds,
            Err(err) => {
                error!("dbus-broker stale fd collection task failed: {}", err);
                None
            }
        },
        None => None,
    };
    let mut dbus_broker_user_accounting = stats
        .rest()
        .get("org.bus1.DBus.Debug.Stats.UserAccounting")
        .map(|user| parse_user_accounting(config, user))
        .unwrap_or_default();

    if let (Some(stale_fds), Some(user_accounting)) =
        (stale_fds, dbus_broker_user_accounting.as_mut())
    {
        if let Some(root) = user_accounting.get_mut(&0) {
            root.stale_fds = Some(stale_fds);
        }
    }

    let dbus_stats = DBusStats {
        serial: stats.serial(),
        active_connections: stats.active_connections(),
        incomplete_connections: stats.incomplete_connections(),
        bus_names: stats.bus_names(),
        peak_bus_names: stats.peak_bus_names(),
        peak_bus_names_per_connection: stats.peak_bus_names_per_connection(),
        match_rules: stats.match_rules(),
        peak_match_rules: stats.peak_match_rules(),
        peak_match_rules_per_connection: stats.peak_match_rules_per_connection(),
        stale_fds,

        // attempt to parse dbus-broker specific stats
        dbus_broker_peer_accounting: filter_and_collect_peer_accounting(config, peers.as_ref()),
        dbus_broker_cgroup_accounting: filter_and_collect_cgroup_accounting(config, peers.as_ref()),
        dbus_broker_user_accounting,
    };

    Ok(dbus_stats)
}

/// Pull all units from dbus and count how system is setup and behaving
pub async fn parse_dbus_stats(
    config: &crate::config::Config,
    connection: &zbus::Connection,
) -> Result<DBusStats, MonitordDbusStatsError> {
    parse_dbus_stats_inner(config, connection, true).await
}

/// Async wrapper than can update dbus stats when passed a locked struct
pub async fn update_dbus_stats(
    config: Arc<crate::config::Config>,
    connection: zbus::Connection,
    locked_machine_stats: Arc<RwLock<MachineStats>>,
) -> anyhow::Result<()> {
    update_dbus_stats_inner(config, connection, locked_machine_stats, true).await
}

/// Async wrapper for nested machine/container buses.
///
/// Stale pidfd accounting is read from host procfs and currently cannot be
/// attributed to nested machine D-Bus brokers, so machine stats skip it.
pub async fn update_machine_dbus_stats(
    config: Arc<crate::config::Config>,
    connection: zbus::Connection,
    locked_machine_stats: Arc<RwLock<MachineStats>>,
) -> anyhow::Result<()> {
    update_dbus_stats_inner(config, connection, locked_machine_stats, false).await
}

async fn update_dbus_stats_inner(
    config: Arc<crate::config::Config>,
    connection: zbus::Connection,
    locked_machine_stats: Arc<RwLock<MachineStats>>,
    collect_stale_fds: bool,
) -> anyhow::Result<()> {
    match parse_dbus_stats_inner(&config, &connection, collect_stale_fds).await {
        Ok(dbus_stats) => {
            let mut machine_stats = locked_machine_stats.write().await;
            machine_stats.dbus_stats = Some(dbus_stats)
        }
        Err(err) => error!("dbus stats failed: {:?}", err),
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::os::unix::fs::symlink;
    use zvariant::{Array, OwnedValue, Str, Structure, Value};

    fn write_fake_proc_process(proc_root: &Path, pid: u32, ppid: u32, cmdline: &[&str]) {
        let pid_dir = proc_root.join(pid.to_string());
        fs::create_dir_all(pid_dir.join("fd")).expect("create fake fd dir");
        fs::create_dir_all(pid_dir.join("fdinfo")).expect("create fake fdinfo dir");

        let mut cmdline_bytes = Vec::new();
        for arg in cmdline {
            cmdline_bytes.extend_from_slice(arg.as_bytes());
            cmdline_bytes.push(0);
        }
        fs::write(pid_dir.join("cmdline"), cmdline_bytes).expect("write fake cmdline");
        fs::write(
            pid_dir.join("stat"),
            format!("{pid} (fake process) S {ppid} 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0"),
        )
        .expect("write fake stat");
    }

    fn write_fake_fd(proc_root: &Path, pid: u32, fd: u32, target: &str, fdinfo: &str) {
        let pid_dir = proc_root.join(pid.to_string());
        symlink(target, pid_dir.join("fd").join(fd.to_string())).expect("create fake fd link");
        fs::write(pid_dir.join("fdinfo").join(fd.to_string()), fdinfo).expect("write fake fdinfo");
    }

    #[test]
    fn test_cur_max_pair_usage() {
        let p = CurMaxPair { cur: 10, max: 100 };
        assert_eq!(p.get_usage(), 90);
    }

    #[test]
    fn test_parse_ppid_from_stat_handles_comm_with_spaces() {
        let stat = "42 (dbus broker worker) S 7 0 0 0 0";
        assert_eq!(parse_ppid_from_stat(stat), Some(7));
    }

    #[test]
    fn test_collect_system_dbus_broker_stale_fds_from_proc() {
        let tempdir = tempfile::tempdir().expect("create tempdir");
        let proc_root = tempdir.path();

        write_fake_proc_process(
            proc_root,
            10,
            1,
            &[
                "/usr/bin/dbus-broker-launch",
                "--scope",
                "system",
                "--audit",
            ],
        );
        write_fake_proc_process(proc_root, 11, 10, &["dbus-broker", "--log", "10"]);
        write_fake_proc_process(
            proc_root,
            20,
            1,
            &["/usr/bin/dbus-broker-launch", "--scope", "user"],
        );
        write_fake_proc_process(proc_root, 21, 20, &["dbus-broker", "--log", "10"]);

        write_fake_fd(
            proc_root,
            11,
            0,
            "anon_inode:[pidfd]",
            "Pid:\t-1\nNSpid:\t-1\n",
        );
        write_fake_fd(
            proc_root,
            11,
            1,
            "anon_inode:[pidfd]",
            "Pid:\t123\nNSpid:\t123\n",
        );
        write_fake_fd(proc_root, 11, 2, "socket:[123]", "scm_fds: 0\n");
        write_fake_fd(
            proc_root,
            21,
            0,
            "anon_inode:[pidfd]",
            "Pid:\t-1\nNSpid:\t-1\n",
        );

        assert_eq!(
            collect_system_dbus_broker_stale_fds_from_proc(proc_root).expect("collect stale fds"),
            Some(1)
        );
    }

    #[test]
    fn test_combine_with_peer_option_summing() {
        let mut cg = DBusBrokerCGroupAccounting {
            name: "cg1".to_string(),
            name_objects: Some(5),
            match_bytes: None,
            matches: Some(3),
            reply_objects: None,
            incoming_bytes: Some(10),
            incoming_fds: None,
            outgoing_bytes: Some(7),
            outgoing_fds: Some(2),
            activation_request_bytes: None,
            activation_request_fds: Some(1),
        };

        let peer = DBusBrokerPeerAccounting {
            id: ":1.1".to_string(),
            well_known_name: Some("com.example".to_string()),
            unix_user_id: Some(1000),
            process_id: Some(1234),
            unix_group_ids: Some(vec![1000]),
            name_objects: Some(2),
            match_bytes: Some(4),
            matches: None,
            reply_objects: Some(1),
            incoming_bytes: None,
            incoming_fds: Some(5),
            outgoing_bytes: Some(3),
            outgoing_fds: None,
            activation_request_bytes: Some(8),
            activation_request_fds: None,
        };

        cg.combine_with_peer(&peer);

        assert_eq!(cg.name_objects, Some(7));
        assert_eq!(cg.match_bytes, Some(4));
        assert_eq!(cg.matches, Some(3));
        assert_eq!(cg.reply_objects, Some(1));
        assert_eq!(cg.incoming_bytes, Some(10));
        assert_eq!(cg.incoming_fds, Some(5));
        assert_eq!(cg.outgoing_bytes, Some(10));
        assert_eq!(cg.outgoing_fds, Some(2));
        assert_eq!(cg.activation_request_bytes, Some(8));
        assert_eq!(cg.activation_request_fds, Some(1));
    }

    #[test]
    fn test_parse_user_accounting_gating_and_parse() {
        // When user_stats=false, should return None
        let mut cfg = crate::config::Config::default();
        cfg.dbus_stats.user_stats = false;
        let empty_val = Value::Array(Array::from(Vec::<Value>::new()));
        let empty_owned = OwnedValue::try_from(empty_val).expect("owned value conversion");
        assert!(parse_user_accounting(&cfg, &empty_owned).is_none());

        // When user_stats=true, empty array should return Some(empty map)
        cfg.dbus_stats.user_stats = true;
        let empty_val = Value::Array(Array::from(Vec::<Value>::new()));
        let owned = OwnedValue::try_from(empty_val).expect("should convert empty array");
        let parsed = parse_user_accounting(&cfg, &owned).expect("should parse empty");
        assert_eq!(parsed.len(), 0);

        // Non-array input should return None
        let non_array = OwnedValue::try_from(Value::U32(0)).expect("should convert u32 value");
        assert!(parse_user_accounting(&cfg, &non_array).is_none());
    }

    #[test]
    fn test_parse_user_struct_invalid_returns_none() {
        // Build an invalid structure (wrong field types/order) to ensure None is returned
        let invalid = Value::Structure(Structure::from((
            Value::Str(Str::from_static("not_uid")),
            Value::U32(10),
            Value::U32(20),
        )));
        assert!(parse_user_struct(&invalid).is_none());
    }

    #[test]
    fn test_user_username_fallback() {
        // Use a likely-nonexistent uid to force fallback to stringified uid
        let mut user = DBusBrokerUserAccounting::new(999_999);
        user.bytes = Some(CurMaxPair { cur: 5, max: 10 });
        // If users crate can’t resolve uid, it should fallback to uid string
        assert_eq!(&user.username, "999999");
    }
}