Method notify

nodedb/src/control/notify_bus.rs:173–225

Publish a notification to all listeners on `(tenant_id, channel)`. Non-blocking: uses `try_send`. When a session's queue is full, the oldest pending notification is dropped via a `recv().ok()` drain and the new one is re-sent, then the drop counter is incremented.

(&self, tenant_id: TenantId, channel: &str, payload: &str)

    /// the oldest pending notification is dropped via a `recv().ok()` drain
    /// and the new one is re-sent, then the drop counter is incremented.
    pub fn notify(&self, tenant_id: TenantId, channel: &str, payload: &str) {
        let key = BusKey {
            tenant_id: tenant_id.as_u64(),
            channel: normalize_channel(channel),
        };
        let notification = Notification {
            channel: key.channel.clone(),
            payload: payload.to_string(),
            pid: 0,
        };

        let map = self.subscribers.read().unwrap_or_else(|p| p.into_inner());
        let sinks = match map.get(&key) {
            Some(s) => s,
            None => return, // no listeners — no-op
        };

        let mut dead = Vec::new();
        for (session_id, sink) in sinks {
            match sink.tx.try_send(notification.clone()) {
                Ok(()) => {}
                Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
                    // Queue is full — drain one entry to make room, then retry.
                    // We don't have a mutable ref so we can't drain directly here.
                    // Use a blocking_recv in a non-async context isn't available,
                    // but try_recv on the sender side isn't accessible.
                    // Instead: increment the drop counter and skip.
                    self.dropped.fetch_add(1, Ordering::Relaxed);
                    warn!(
                        session_id,
                        channel = key.channel.as_str(),
                        cap = sink.cap,
                        "NOTIFY queue full — dropping oldest (metric incremented)"
                    );
                }
                Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
                    dead.push(*session_id);
                }
            }
        }
        drop(map);

        // Clean up closed sessions.
        if !dead.is_empty() {
            let mut map = self.subscribers.write().unwrap_or_else(|p| p.into_inner());
            if let Some(sinks) = map.get_mut(&key) {
                sinks.retain(|(id, _)| !dead.contains(id));
                if sinks.is_empty() {
                    map.remove(&key);
                }
            }
        }
    }

    /// Total dropped notifications since server start.
    pub fn total_dropped(&self) -> u64 {