exfiltrate 0.4.0

// SPDX-License-Identifier: MIT OR Apache-2.0
//! Server-pushed events, so a client can wait for something instead of polling.
//!
//! # The shape this replaces
//!
//! Without a push path, "wait until X, then inspect" — the most common
//! debugging shape there is — can only be written as
//! `while ! exfiltrate state | grep -q ready; do sleep 1; done`. That is slow,
//! it is racy, and it misses transient state entirely: whatever happened between
//! two polls did not happen as far as the poller is concerned.
//!
//! # Cost at the emit site
//!
//! Emit sites end up on hot paths — a frame presented, a request completed — so
//! [`emit`](crate::emit) must be cheap when nobody is listening. It checks a
//! single relaxed atomic before doing anything else, including before formatting
//! its payload, so an unsubscribed topic costs one load and a branch.
//!
//! # Drop policy
//!
//! Each subscriber has a bounded queue. When it fills, the *oldest* event is
//! dropped and a per-subscriber counter increments, which rides along on the
//! next delivered event as [`Event::dropped`](exfiltrate_internal::rpc::Event::dropped).
//! Dropping the oldest keeps the consumer current rather than making it replay a
//! backlog, and reporting the count is what stops a filtered view from passing
//! itself off as a complete one.

use exfiltrate_internal::command::Response;
use exfiltrate_internal::rpc::Event;
use std::collections::VecDeque;
use std::sync::LazyLock;
use std::sync::atomic::{AtomicUsize, Ordering};
use wasm_lite_std::Mutex;

/// How many subscriptions exist across all connections.
///
/// Read on every [`crate::emit`] call, so it is deliberately the cheapest thing
/// that can answer "is anyone listening".
static SUBSCRIPTION_COUNT: AtomicUsize = AtomicUsize::new(0);

/// One connection's view of the event system.
#[derive(Debug)]
struct Subscriber {
    id: u64,
    topics: Vec<String>,
    queue: VecDeque<Event>,
    capacity: usize,
    dropped: u64,
    next_seq: u64,
}

static SUBSCRIBERS: LazyLock<Mutex<Vec<Subscriber>>> = LazyLock::new(|| Mutex::new(Vec::new()));
static NEXT_ID: AtomicUsize = AtomicUsize::new(0);

/// Registers a connection as an event consumer and returns its handle.
pub(crate) fn attach(capacity: usize) -> u64 {
    let id = NEXT_ID.fetch_add(1, Ordering::Relaxed) as u64;
    SUBSCRIBERS.with_mut_sync(|subscribers| {
        subscribers.push(Subscriber {
            id,
            topics: Vec::new(),
            queue: VecDeque::new(),
            capacity: capacity.max(1),
            dropped: 0,
            next_seq: 0,
        })
    });
    id
}

/// Removes a connection and everything it was subscribed to.
pub(crate) fn detach(id: u64) {
    SUBSCRIBERS.with_mut_sync(|subscribers| {
        if let Some(index) = subscribers.iter().position(|s| s.id == id) {
            let removed = subscribers.remove(index);
            SUBSCRIPTION_COUNT.fetch_sub(removed.topics.len(), Ordering::Relaxed);
        }
    });
}

/// Adds a topic pattern to a connection's subscriptions.
///
/// Returns an error for a pattern this connection already holds, so a client
/// that subscribes twice is told rather than silently double-counted.
pub(crate) fn subscribe(id: u64, topic: &str) -> Result<(), String> {
    if topic.is_empty() {
        return Err("a subscription topic must not be empty".to_string());
    }
    SUBSCRIBERS.with_mut_sync(|subscribers| {
        let Some(subscriber) = subscribers.iter_mut().find(|s| s.id == id) else {
            return Err("this connection is not attached to the event system".to_string());
        };
        if subscriber.topics.iter().any(|held| held == topic) {
            return Err(format!("already subscribed to {topic:?}"));
        }
        subscriber.topics.push(topic.to_string());
        SUBSCRIPTION_COUNT.fetch_add(1, Ordering::Relaxed);
        Ok(())
    })
}

/// Removes a topic pattern from a connection's subscriptions.
pub(crate) fn unsubscribe(id: u64, topic: &str) -> Result<(), String> {
    SUBSCRIBERS.with_mut_sync(|subscribers| {
        let Some(subscriber) = subscribers.iter_mut().find(|s| s.id == id) else {
            return Err("this connection is not attached to the event system".to_string());
        };
        let Some(index) = subscriber.topics.iter().position(|held| held == topic) else {
            return Err(format!("not subscribed to {topic:?}"));
        };
        subscriber.topics.remove(index);
        SUBSCRIPTION_COUNT.fetch_sub(1, Ordering::Relaxed);
        Ok(())
    })
}

/// Whether any connection is subscribed to anything at all.
///
/// This is the fast path [`crate::emit`] takes before building a payload.
pub(crate) fn anyone_listening() -> bool {
    SUBSCRIPTION_COUNT.load(Ordering::Relaxed) > 0
}

/// Queues an event for every subscriber whose pattern matches.
///
/// Returns how many subscribers took it.
pub(crate) fn publish(topic: &str, payload: Response) -> usize {
    SUBSCRIBERS.with_mut_sync(|subscribers| {
        let mut delivered = 0;
        for subscriber in subscribers.iter_mut() {
            if !subscriber
                .topics
                .iter()
                .any(|pattern| exfiltrate_internal::topic::matches(pattern, topic))
            {
                continue;
            }
            while subscriber.queue.len() >= subscriber.capacity {
                subscriber.queue.pop_front();
                subscriber.dropped += 1;
            }
            let seq = subscriber.next_seq;
            subscriber.next_seq += 1;
            subscriber.queue.push_back(Event {
                topic: topic.to_string(),
                seq,
                payload: payload.clone(),
                dropped: subscriber.dropped,
            });
            delivered += 1;
        }
        delivered
    })
}

/// Takes everything queued for a connection.
pub(crate) fn drain(id: u64) -> Vec<Event> {
    SUBSCRIBERS.with_mut_sync(
        |subscribers| match subscribers.iter_mut().find(|s| s.id == id) {
            Some(subscriber) => subscriber.queue.drain(..).collect(),
            None => Vec::new(),
        },
    )
}

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

    /// Serialises these tests against each other.
    ///
    /// The subscriber table and the subscription counter are process-global and
    /// libtest runs tests on parallel threads, so without this a subscriber in
    /// one test sees another test's `publish` and the counter arithmetic below
    /// races. Isolating by topic name is not enough on its own: the whole point
    /// of some of these cases is a subscription that matches everything.
    static TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    /// Takes the serialising lock for the duration of one test.
    ///
    /// Separate from [`Attached`] because a test may attach more than one
    /// connection and the lock is not reentrant.
    ///
    /// A panicking test poisons the lock; taking the inner guard anyway means
    /// one failure does not cascade into every other case.
    fn serialised() -> std::sync::MutexGuard<'static, ()> {
        TEST_LOCK.lock().unwrap_or_else(|error| error.into_inner())
    }

    /// Each test attaches its own connection and detaches at the end, because
    /// the subscriber table is process-global.
    struct Attached(u64);

    impl Attached {
        fn new(capacity: usize) -> Attached {
            Attached(attach(capacity))
        }
    }

    impl Drop for Attached {
        fn drop(&mut self) {
            detach(self.0);
        }
    }

    #[test]
    fn nothing_is_delivered_to_a_connection_that_did_not_subscribe() {
        let _serialised = serialised();
        let connection = Attached::new(8);
        publish("render.frame", Response::String("x".into()));
        assert!(drain(connection.0).is_empty());
    }

    #[test]
    fn a_prefix_subscription_takes_the_subtree_and_nothing_else() {
        let _serialised = serialised();
        let connection = Attached::new(8);
        subscribe(connection.0, "render.*").unwrap();
        publish("render.frame", Response::String("a".into()));
        publish("renderer.stats", Response::String("b".into()));
        publish("net.request", Response::String("c".into()));

        let events = drain(connection.0);
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].topic, "render.frame");
        assert_eq!(events[0].seq, 0);
    }

    #[test]
    fn a_full_queue_drops_the_oldest_and_says_how_many() {
        let _serialised = serialised();
        let connection = Attached::new(2);
        subscribe(connection.0, "*").unwrap();
        for index in 0..5 {
            publish("t", Response::String(index.to_string()));
        }
        let events = drain(connection.0);
        assert_eq!(events.len(), 2);
        // The two newest survived...
        assert_eq!(events[0].payload, Response::String("3".into()));
        assert_eq!(events[1].payload, Response::String("4".into()));
        // ...and the consumer is told it missed three.
        assert_eq!(events[1].dropped, 3);
    }

    #[test]
    fn sequence_numbers_count_every_event_including_dropped_ones() {
        let _serialised = serialised();
        let connection = Attached::new(1);
        subscribe(connection.0, "*").unwrap();
        publish("t", Response::default());
        publish("t", Response::default());
        let events = drain(connection.0);
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].seq, 1, "the surviving event is the second one");
    }

    #[test]
    fn the_listening_flag_tracks_subscribe_unsubscribe_and_detach() {
        let _serialised = serialised();
        let connection = Attached::new(4);
        let base = SUBSCRIPTION_COUNT.load(Ordering::Relaxed);
        {
            subscribe(connection.0, "a").unwrap();
            subscribe(connection.0, "b").unwrap();
            assert!(anyone_listening());
            assert_eq!(SUBSCRIPTION_COUNT.load(Ordering::Relaxed), base + 2);
            unsubscribe(connection.0, "a").unwrap();
            assert_eq!(SUBSCRIPTION_COUNT.load(Ordering::Relaxed), base + 1);
        }
        // Dropping the connection detaches it, taking its remaining subscription.
        drop(connection);
        assert_eq!(SUBSCRIPTION_COUNT.load(Ordering::Relaxed), base);
    }

    #[test]
    fn subscribing_twice_and_unsubscribing_from_nothing_are_both_reported() {
        let _serialised = serialised();
        let connection = Attached::new(4);
        subscribe(connection.0, "dup").unwrap();
        assert!(
            subscribe(connection.0, "dup")
                .unwrap_err()
                .contains("already")
        );
        assert!(
            unsubscribe(connection.0, "never")
                .unwrap_err()
                .contains("not subscribed")
        );
    }

    #[test]
    fn two_connections_get_independent_copies() {
        let _serialised = serialised();
        let first = Attached::new(8);
        let second = Attached::new(8);
        subscribe(first.0, "*").unwrap();
        subscribe(second.0, "*").unwrap();
        assert_eq!(publish("t", Response::String("shared".into())), 2);
        assert_eq!(drain(first.0).len(), 1);
        assert_eq!(drain(second.0).len(), 1);
    }
}