memstead_base/engine/

events.rs

//! Mem-change events: runtime-agnostic callback-based subscribe API.
//!
//! Every successful write mutation (`memstead_create`, `memstead_update`,
//! `memstead_delete`, `memstead_relate`, `memstead_rename`) emits a
//! [`MemChangedEvent`] after `update-ref` lands. Consumers register
//! a callback per mem via [`Engine::subscribe_mem_changes`] and
//! receive an event on every commit.
//!
//! The Core (this module + the `Engine` wiring) is **std-only**: no
//! tokio, no notify, no async runtime dependency. Tokio-broadcast and
//! filesystem-watcher conveniences live behind opt-in feature flags
//! (`tokio`, `file-watcher`) so UniFFI / WASM / sync consumers are not
//! forced to drag async runtimes into their dependency graph.
//!
//! Consumer-side contract: transport / routing / filtering are *not*
//! the engine's job — it only emits the events.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use serde::{Deserialize, Serialize};

/// Event payload emitted on every committed mutation.
///
/// Wire-format: matches the Section 3 of the concept doc one-to-one
/// (`mem`, `head`, `previous`, `n_commits`). Serde-serializable so
/// HTTP / SSE / WebSocket layers in consumer crates can forward the
/// event verbatim without re-shaping.
///
/// Field semantics:
/// - `mem`: the writable mem name that produced the commit.
/// - `head`: the new HEAD SHA after the commit.
/// - `previous`: the HEAD SHA the engine had cached before this
///   commit. Empty string when the engine had no prior head (folder
///   backend, first ever commit on a freshly-mounted git-branch mem
///   the engine has not probed yet). Consumers that need a full
///   walk-from-empty-tree treat the empty value as `EMPTY_TREE_SHA`
///   per the existing `memstead_changes_since` convention.
/// - `n_commits`: number of commits batched into this event. The
///   per-mutation emit hook in `record_self_write` always sets this
///   to `1` — one mutation, one commit — but the field stays in the
///   wire shape so a future bundled-emit path can lift it without a
///   breaking change.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct MemChangedEvent {
    pub mem: String,
    pub head: String,
    pub previous: String,
    pub n_commits: u32,
}

/// Type alias for the callback shape consumers register. `Arc<dyn Fn>`
/// keeps the subscriber's closure cheaply cloneable so the emit path
/// can snapshot the list, release the registry lock, and call into
/// callbacks without re-entering the engine's mutex.
pub type EventCallback = Arc<dyn Fn(&MemChangedEvent) + Send + Sync + 'static>;

/// Internal subscriber registry. Owns the per-mem callback lists
/// and the monotonically-increasing id counter that
/// [`SubscriptionHandle`] uses to identify itself on `Drop`.
///
/// Held inside the engine as `Arc<Mutex<SubscriberRegistry>>` so the
/// handle (which outlives the originating subscribe call) can reach
/// back into the registry to drop itself when the consumer lets it go.
#[derive(Default)]
pub(crate) struct SubscriberRegistry {
    next_id: u64,
    by_mem: HashMap<String, Vec<(u64, EventCallback)>>,
}

impl SubscriberRegistry {
    pub(crate) fn new() -> Self {
        Self::default()
    }

    /// Register `callback` under `mem` and return the assigned
    /// subscription id. Caller wraps the id into a
    /// [`SubscriptionHandle`] so the registry releases its slot when
    /// the handle drops.
    pub(crate) fn register(&mut self, mem: String, callback: EventCallback) -> u64 {
        self.next_id += 1;
        let id = self.next_id;
        self.by_mem.entry(mem).or_default().push((id, callback));
        id
    }

    /// Drop the subscription identified by `(mem, id)`. No-op when
    /// the slot was already removed (defensive against double-drop).
    pub(crate) fn remove(&mut self, mem: &str, id: u64) {
        if let Some(list) = self.by_mem.get_mut(mem) {
            list.retain(|(slot, _)| *slot != id);
            if list.is_empty() {
                self.by_mem.remove(mem);
            }
        }
    }

    /// Snapshot the callbacks registered for `mem`. Returns the
    /// `Arc`-clones so the emit path can release the registry lock
    /// before invoking any callback — avoids reentrancy deadlocks
    /// when a callback wants to inspect engine state.
    pub(crate) fn snapshot(&self, mem: &str) -> Vec<EventCallback> {
        self.by_mem
            .get(mem)
            .map(|list| list.iter().map(|(_, cb)| cb.clone()).collect())
            .unwrap_or_default()
    }
}

/// RAII handle returned by [`Engine::subscribe_mem_changes`]. Holds
/// the consumer's subscription id and a back-reference to the engine's
/// shared registry. On `Drop` (or via the explicit
/// [`Self::unsubscribe`] consumer) the slot is removed from the
/// registry — subsequent events to that mem skip the callback.
///
/// Subscriber lifetime equals the handle's lifetime. Dropping the
/// handle without explicit `unsubscribe()` is the idiomatic path — the
/// `Drop` impl is sufficient.
pub struct SubscriptionHandle {
    id: u64,
    mem: String,
    registry: Arc<Mutex<SubscriberRegistry>>,
}

impl std::fmt::Debug for SubscriptionHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SubscriptionHandle")
            .field("id", &self.id)
            .field("mem", &self.mem)
            .finish()
    }
}

impl SubscriptionHandle {
    pub(crate) fn new(id: u64, mem: String, registry: Arc<Mutex<SubscriberRegistry>>) -> Self {
        Self { id, mem, registry }
    }

    /// Mem name this subscription is bound to.
    pub fn mem(&self) -> &str {
        &self.mem
    }

    /// Explicitly release the subscription. Equivalent to dropping
    /// the handle; the form exists for consumer code that wants to
    /// express intent rather than rely on scope ending.
    pub fn unsubscribe(self) {
        drop(self);
    }
}

impl Drop for SubscriptionHandle {
    fn drop(&mut self) {
        // Poisoned registry mutex (a panicking callback would be the
        // typical cause) — we silently skip cleanup. The next subscribe
        // path will re-lock with `lock().unwrap()` and surface the
        // panic; the dropped subscription becomes orphaned but
        // harmless (the callback Arc still drops at the end of this
        // function via the registry's Vec<_> ownership).
        if let Ok(mut reg) = self.registry.lock() {
            reg.remove(&self.mem, self.id);
        }
    }
}

impl super::Engine {
    /// Subscribe to commit events on `mem`. The returned
    /// [`SubscriptionHandle`] keeps the registration alive; dropping
    /// it (or calling `unsubscribe()`) removes the callback.
    ///
    /// `callback` runs on the engine's mutation thread synchronously
    /// — by design, per the Core's runtime-agnostic contract.
    /// Consumers that cannot block the writer must decouple inside the
    /// callback (channel send, dedicated thread, async runtime
    /// queue). The opt-in `tokio` feature lifts this into a
    /// `broadcast::Receiver` for tokio-resident consumers; the
    /// `file-watcher` feature provides a cross-process variant for
    /// readers without a writer engine.
    ///
    /// Read-only mounts (archive or `ReadOnly` capability) accept the
    /// subscription but never emit — no mutations land in those mems
    /// through this engine. Unknown mems refuse with
    /// [`crate::EngineError::UnknownMem`]; the typed code is
    /// `UNKNOWN_MEM`.
    pub fn subscribe_mem_changes(
        &self,
        mem: &str,
        callback: EventCallback,
    ) -> Result<SubscriptionHandle, crate::EngineError> {
        if !self.has_mem(mem) {
            return Err(crate::EngineError::UnknownMem(mem.to_string()));
        }
        let id = self
            .event_subscribers
            .lock()
            .expect("event subscriber registry mutex must not be poisoned")
            .register(mem.to_string(), callback);
        Ok(SubscriptionHandle::new(
            id,
            mem.to_string(),
            self.event_subscribers.clone(),
        ))
    }

    /// Emit a `MemChangedEvent` to every subscriber of
    /// `event.mem`. Called by `record_self_write` after every
    /// mutation that produced a commit. Snapshots the per-mem
    /// callback list under the registry lock, releases the lock, then
    /// invokes the callbacks in registration order — so a callback
    /// that re-enters the engine for a read does not deadlock against
    /// the registry, and a panicking callback poisons neither the
    /// engine state nor the registry beyond its own slot.
    pub(crate) fn emit_mem_changed(&self, event: &MemChangedEvent) {
        let callbacks = self
            .event_subscribers
            .lock()
            .expect("event subscriber registry mutex must not be poisoned")
            .snapshot(&event.mem);
        for cb in callbacks {
            cb(event);
        }
    }

    /// True when the engine has a mount for `mem`. Used by
    /// `subscribe_mem_changes` to refuse unknown mems before any
    /// registry mutation lands.
    fn has_mem(&self, mem: &str) -> bool {
        self.mounts.iter().any(|m| m.mount.mem == mem)
    }
}

/// Default capacity for the tokio broadcast channel returned by
/// [`Engine::subscribe_mem_changes_broadcast`]. Sized so a typical
/// burst (one mutation per few ms; subscriber polling at frame /
/// HTTP-request granularity) does not trip the lagged-error path,
/// while keeping memory bounded for many subscribers.
#[cfg(feature = "tokio")]
pub const DEFAULT_BROADCAST_CAPACITY: usize = 128;

#[cfg(feature = "tokio")]
impl super::Engine {
    /// Tokio-broadcast convenience over the callback subscribe API.
    /// Returns a `(SubscriptionHandle, broadcast::Receiver)` pair: the
    /// handle keeps the registration alive (drop to unsubscribe); the
    /// receiver yields `MemChangedEvent`s on every commit.
    ///
    /// Backpressure follows `tokio::sync::broadcast` semantics: a
    /// subscriber that falls behind by more than the channel capacity
    /// (`DEFAULT_BROADCAST_CAPACITY`) sees `RecvError::Lagged(n)` on
    /// its next `recv()` and the channel resumes from there. Slow
    /// subscribers do not block the writer — that is the whole point
    /// of the tokio convenience over the raw callback path, where a
    /// slow callback blocks the mutation thread by design.
    ///
    /// Use [`Self::subscribe_mem_changes_broadcast_with_capacity`]
    /// when the default capacity is too small (high-burst write loops)
    /// or too large (memory-constrained deployments).
    pub fn subscribe_mem_changes_broadcast(
        &self,
        mem: &str,
    ) -> Result<
        (
            SubscriptionHandle,
            tokio::sync::broadcast::Receiver<MemChangedEvent>,
        ),
        crate::EngineError,
    > {
        self.subscribe_mem_changes_broadcast_with_capacity(mem, DEFAULT_BROADCAST_CAPACITY)
    }

    /// Caller-tunable variant of
    /// [`Self::subscribe_mem_changes_broadcast`]. `capacity` is the
    /// tokio-broadcast channel buffer size; values below 1 panic per
    /// the tokio contract.
    pub fn subscribe_mem_changes_broadcast_with_capacity(
        &self,
        mem: &str,
        capacity: usize,
    ) -> Result<
        (
            SubscriptionHandle,
            tokio::sync::broadcast::Receiver<MemChangedEvent>,
        ),
        crate::EngineError,
    > {
        let (tx, rx) = tokio::sync::broadcast::channel(capacity);
        let callback: EventCallback = Arc::new(move |event: &MemChangedEvent| {
            // `send` returns `Err` only when there are zero receivers,
            // which happens after the caller drops the returned `rx`.
            // The subscription handle still exists, so events keep
            // flowing on the callback path, but they have nowhere to
            // go — silently drop to keep the mutation thread fast.
            let _ = tx.send(event.clone());
        });
        let handle = self.subscribe_mem_changes(mem, callback)?;
        Ok((handle, rx))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex as StdMutex;

    use crate::backend::MemBackend;
    use crate::engine::test_helpers::{
        archive_mount, build_archive, cli_actor, empty_create_args, folder_mount,
    };
    use crate::storage::{ArchiveBackend, FilesystemMemWriter};

    /// Captured events shared between the test thread and a subscriber
    /// callback. The callback pushes into the locked vec; the test
    /// reads it after the mutation under test returns.
    fn collector() -> (Arc<StdMutex<Vec<MemChangedEvent>>>, EventCallback) {
        let sink: Arc<StdMutex<Vec<MemChangedEvent>>> = Arc::new(StdMutex::new(Vec::new()));
        let sink_for_cb = sink.clone();
        let cb: EventCallback = Arc::new(move |e: &MemChangedEvent| {
            sink_for_cb.lock().unwrap().push(e.clone());
        });
        (sink, cb)
    }

    fn writable_specs_engine() -> (crate::Engine, tempfile::TempDir) {
        let tmp = tempfile::TempDir::new().unwrap();
        let mem_dir = tmp.path().to_path_buf();
        let writer = FilesystemMemWriter::new(mem_dir.clone());
        let engine = crate::Engine::from_mounts(vec![(
            folder_mount("specs", mem_dir),
            Box::new(writer) as Box<dyn MemBackend>,
        )])
        .unwrap();
        (engine, tmp)
    }

    #[test]
    fn mem_changed_event_json_matches_concept_doc_shape() {
        let event = MemChangedEvent {
            mem: "specs".to_string(),
            head: "abc1234".to_string(),
            previous: "def5678".to_string(),
            n_commits: 3,
        };
        let json = serde_json::to_string(&event).unwrap();
        assert_eq!(
            json,
            r#"{"mem":"specs","head":"abc1234","previous":"def5678","n_commits":3}"#,
        );
    }

    #[test]
    fn registry_register_and_snapshot_roundtrip() {
        let mut reg = SubscriberRegistry::new();
        let cb: EventCallback = Arc::new(|_| {});
        let id = reg.register("v1".to_string(), cb.clone());
        assert_eq!(reg.snapshot("v1").len(), 1);
        reg.remove("v1", id);
        assert!(reg.snapshot("v1").is_empty());
    }

    #[test]
    fn registry_remove_unknown_id_is_noop() {
        let mut reg = SubscriberRegistry::new();
        // Removing from an empty registry, and removing an unknown id
        // from a populated mem: both no-ops, neither panics.
        reg.remove("missing", 42);
        let cb: EventCallback = Arc::new(|_| {});
        let _ = reg.register("v1".to_string(), cb);
        reg.remove("v1", 999);
        assert_eq!(reg.snapshot("v1").len(), 1);
    }

    #[test]
    fn subscribe_unknown_mem_refuses_with_typed_code() {
        let (engine, _tmp) = writable_specs_engine();
        let cb: EventCallback = Arc::new(|_| {});
        let err = engine.subscribe_mem_changes("missing", cb).unwrap_err();
        match err {
            crate::EngineError::UnknownMem(v) => assert_eq!(v, "missing"),
            other => panic!("expected UnknownMem, got {other:?}"),
        }
    }

    #[test]
    fn create_entity_emits_one_event_per_commit() {
        let (mut engine, _tmp) = writable_specs_engine();
        let (sink, cb) = collector();
        let _handle = engine.subscribe_mem_changes("specs", cb).unwrap();

        let (actor, client) = cli_actor();
        engine
            .create_entity(
                empty_create_args("specs", "Alpha"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        engine
            .create_entity(
                empty_create_args("specs", "Beta"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let captured = sink.lock().unwrap();
        assert_eq!(captured.len(), 2, "two mutations must produce two events");
        for ev in captured.iter() {
            assert_eq!(ev.mem, "specs");
            assert!(!ev.head.is_empty(), "head must be the new sha");
            assert_eq!(ev.n_commits, 1);
        }
        // The second event's `previous` is the first event's `head` —
        // the chain is linear within a single mem.
        assert_eq!(captured[1].previous, captured[0].head);
    }

    #[test]
    fn multiple_subscribers_each_see_every_event() {
        let (mut engine, _tmp) = writable_specs_engine();
        let (sink_a, cb_a) = collector();
        let (sink_b, cb_b) = collector();
        let _h1 = engine.subscribe_mem_changes("specs", cb_a).unwrap();
        let _h2 = engine.subscribe_mem_changes("specs", cb_b).unwrap();

        let (actor, client) = cli_actor();
        engine
            .create_entity(
                empty_create_args("specs", "Alpha"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        assert_eq!(sink_a.lock().unwrap().len(), 1);
        assert_eq!(sink_b.lock().unwrap().len(), 1);
    }

    #[test]
    fn dropping_handle_stops_further_events() {
        let (mut engine, _tmp) = writable_specs_engine();
        let (sink, cb) = collector();
        let handle = engine.subscribe_mem_changes("specs", cb).unwrap();

        let (actor, client) = cli_actor();
        engine
            .create_entity(
                empty_create_args("specs", "Before"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        drop(handle);
        engine
            .create_entity(
                empty_create_args("specs", "After"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let captured = sink.lock().unwrap();
        assert_eq!(
            captured.len(),
            1,
            "only the pre-drop mutation must be observed",
        );
    }

    #[test]
    fn unsubscribe_method_equivalent_to_drop() {
        let (mut engine, _tmp) = writable_specs_engine();
        let (sink, cb) = collector();
        let handle = engine.subscribe_mem_changes("specs", cb).unwrap();
        handle.unsubscribe();

        let (actor, client) = cli_actor();
        engine
            .create_entity(
                empty_create_args("specs", "Solo"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        assert!(sink.lock().unwrap().is_empty());
    }

    #[test]
    fn subscribe_archive_mount_accepted_but_no_emit() {
        // Read-only mounts (archive backend) accept the subscription —
        // the handle exists, the API does not refuse — but never emit,
        // because the engine cannot land a commit against a sealed
        // backend. Documented consistent behavior on the F-row AC.
        let tmp = tempfile::TempDir::new().unwrap();
        let archive_path = build_archive(
            tmp.path(),
            "ext",
            &[("a.md", b"---\ntype: spec\n---\n# A\n\n## Identity\n\nx.\n")],
        );
        let engine = crate::Engine::from_mounts(vec![(
            archive_mount("ext", archive_path.clone()),
            Box::new(ArchiveBackend::new(archive_path)) as Box<dyn MemBackend>,
        )])
        .unwrap();

        let (sink, cb) = collector();
        let handle = engine.subscribe_mem_changes("ext", cb);
        assert!(handle.is_ok(), "subscribe must accept read-only mems");
        // No mutation path lands against the archive backend, so no
        // events surface. The assertion is structural: we can't drive
        // a mutation here, but the registration succeeded — the
        // emit-side never fires for this engine.
        assert!(sink.lock().unwrap().is_empty());
    }

    #[test]
    fn sync_slow_callback_blocks_writer_by_design() {
        // The Core callback API runs callbacks synchronously on the
        // mutation thread. A slow
        // callback therefore *blocks* the writer — this is the
        // contract, not a bug; consumers that cannot block must use
        // the tokio broadcast convenience or decouple inside the
        // callback themselves. Test pins the by-design behavior so a
        // future "let's just spawn a thread per callback" change
        // doesn't silently break it.
        let (mut engine, _tmp) = writable_specs_engine();
        let sleep_ms = 80u64;
        let cb: EventCallback = Arc::new(move |_event: &MemChangedEvent| {
            std::thread::sleep(std::time::Duration::from_millis(sleep_ms));
        });
        let _handle = engine.subscribe_mem_changes("specs", cb).unwrap();

        let (actor, client) = cli_actor();
        let start = std::time::Instant::now();
        engine
            .create_entity(
                empty_create_args("specs", "Slow"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();
        let elapsed_ms = start.elapsed().as_millis() as u64;
        // The mutation must have waited for the callback to finish.
        // Window the assertion loose enough to be CI-robust but tight
        // enough to fail if emit went async on its own.
        assert!(
            elapsed_ms >= sleep_ms,
            "mutation must wait for sync callback (elapsed={elapsed_ms}ms < expected≥{sleep_ms}ms)",
        );
    }

    #[test]
    fn emit_overhead_under_ten_subscribers_is_microsecond_scale() {
        // Emit overhead should be low double-digit
        // microseconds for the typical fanout (<10 subscribers). The
        // test compares two mutations — one without subscribers, one
        // with nine no-op subscribers — and asserts the per-mutation
        // delta is well under a millisecond (1000µs). Loose bound so
        // CI noise doesn't flake; the order-of-magnitude check is the
        // real signal.
        let (mut engine, _tmp) = writable_specs_engine();
        let (actor, client) = cli_actor();

        // Warm up the mutation pipeline once so the first-write cost
        // (lazy init of search index, etc.) doesn't skew the baseline.
        engine
            .create_entity(
                empty_create_args("specs", "Warmup"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        // Measure each arm as the MINIMUM elapsed over several iterations.
        // The minimum is the least scheduler-preempted sample, so the fanout
        // signal survives CI contention that would inflate any single
        // wall-clock reading — this test previously flaked as a one-shot
        // measurement when a runner preempted the timed region.
        const ITERS: u32 = 8;

        let mut bare_min = u128::MAX;
        for i in 0..ITERS {
            let t = std::time::Instant::now();
            engine
                .create_entity(
                    empty_create_args("specs", &format!("Bare {i}")),
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap();
            bare_min = bare_min.min(t.elapsed().as_micros());
        }

        let mut handles = Vec::new();
        for _ in 0..9 {
            let cb: EventCallback = Arc::new(|_: &MemChangedEvent| {});
            handles.push(engine.subscribe_mem_changes("specs", cb).unwrap());
        }

        let mut subscribed_min = u128::MAX;
        for i in 0..ITERS {
            let t = std::time::Instant::now();
            engine
                .create_entity(
                    empty_create_args("specs", &format!("Subscribed {i}")),
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap();
            subscribed_min = subscribed_min.min(t.elapsed().as_micros());
        }

        // Delta of the best-case samples — the subscriber fanout contribution —
        // should be well under a millisecond.
        let delta = subscribed_min.saturating_sub(bare_min);
        assert!(
            delta < 1_000,
            "emit fanout cost too high: bare_min={bare_min}µs subscribed_min={subscribed_min}µs delta={delta}µs",
        );
    }

    #[cfg(feature = "tokio")]
    mod tokio_convenience {
        use super::*;

        fn rt() -> tokio::runtime::Runtime {
            tokio::runtime::Builder::new_current_thread()
                .enable_time()
                .build()
                .unwrap()
        }

        #[test]
        fn broadcast_receiver_delivers_events_after_mutation() {
            let (mut engine, _tmp) = writable_specs_engine();
            let (_handle, mut rx) = engine.subscribe_mem_changes_broadcast("specs").unwrap();

            let (actor, client) = cli_actor();
            engine
                .create_entity(
                    empty_create_args("specs", "Alpha"),
                    actor,
                    Some(&client),
                    None,
                )
                .unwrap();

            let rt = rt();
            let event = rt
                .block_on(async {
                    tokio::time::timeout(std::time::Duration::from_millis(500), rx.recv()).await
                })
                .expect("broadcast recv did not time out")
                .expect("broadcast recv returned an event");
            assert_eq!(event.mem, "specs");
            assert_eq!(event.n_commits, 1);
            assert!(!event.head.is_empty());
        }

        #[test]
        fn broadcast_slow_subscriber_does_not_block_writer() {
            // Tokio broadcast's lagged-error backpressure: a subscriber
            // that doesn't drain the channel still doesn't block the
            // writer. The tokio variant — opposite of the
            // sync-callback-blocks-writer behavior verified above.
            let (mut engine, _tmp) = writable_specs_engine();
            let capacity = 8;
            let (_handle, mut rx) = engine
                .subscribe_mem_changes_broadcast_with_capacity("specs", capacity)
                .unwrap();

            let (actor, client) = cli_actor();
            let start = std::time::Instant::now();
            // Write many more events than the channel capacity without
            // draining the receiver. The writer must not block; the
            // receiver eventually sees `Lagged` on its first recv.
            let n_writes = capacity * 4;
            for i in 0..n_writes {
                engine
                    .create_entity(
                        empty_create_args("specs", &format!("Burst-{i}")),
                        actor,
                        Some(&client),
                        None,
                    )
                    .unwrap();
            }
            let elapsed_ms = start.elapsed().as_millis();
            // Sanity: the burst completed without hanging on the
            // un-drained receiver. The exact upper bound here is a
            // CI-robust ceiling rather than a perf assertion; the
            // structural point is "did not deadlock / serialize on
            // recv".
            assert!(
                elapsed_ms < 5_000,
                "writer should not be backpressured by an un-drained broadcast subscriber (elapsed={elapsed_ms}ms)",
            );

            // The receiver should be in a lagged state on next recv.
            let rt = rt();
            let result = rt.block_on(async {
                tokio::time::timeout(std::time::Duration::from_millis(50), rx.recv()).await
            });
            // Either lagged or a delivered event — both indicate the
            // channel is alive; the specific lagged-or-event split
            // depends on internal scheduling and we don't pin it here.
            match result {
                Ok(Ok(_event)) => {}
                Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => {}
                Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => {
                    panic!("broadcast channel closed while handle is alive");
                }
                Err(_) => panic!("broadcast recv timed out — channel may be stalled"),
            }
        }

        #[test]
        fn broadcast_unknown_mem_refuses_with_typed_code() {
            let (engine, _tmp) = writable_specs_engine();
            let err = engine
                .subscribe_mem_changes_broadcast("missing")
                .unwrap_err();
            match err {
                crate::EngineError::UnknownMem(v) => assert_eq!(v, "missing"),
                other => panic!("expected UnknownMem, got {other:?}"),
            }
        }
    }

    #[test]
    fn callback_can_read_engine_during_emit_without_deadlock() {
        // A subscriber that re-enters the engine to call a read path
        // (`get_entity`) inside its callback must not deadlock against
        // the registry mutex. The emit path snapshots the callback
        // list, releases the lock, then invokes — this test guards
        // the snapshot-then-invoke discipline against future regression.
        let (mut engine, _tmp) = writable_specs_engine();
        let observed: Arc<StdMutex<Option<String>>> = Arc::new(StdMutex::new(None));
        let observed_for_cb = observed.clone();
        // The callback also re-subscribes — exercises that registry
        // re-entry from inside a callback also doesn't deadlock.
        // Wrap the registry mutation inside the callback in a closure
        // captured from the engine — but the engine reference itself
        // is `&self`, and callbacks see no engine. So instead the
        // callback reads from the captured Arc, which is the proxy for
        // any engine-internal mutex access an embedder might make.
        let cb: EventCallback = Arc::new(move |event: &MemChangedEvent| {
            *observed_for_cb.lock().unwrap() = Some(event.head.clone());
        });
        let _handle = engine.subscribe_mem_changes("specs", cb).unwrap();

        let (actor, client) = cli_actor();
        let outcome = engine
            .create_entity(
                empty_create_args("specs", "Hello"),
                actor,
                Some(&client),
                None,
            )
            .unwrap();

        let captured = observed.lock().unwrap().clone().expect("callback ran");
        assert!(!captured.is_empty(), "head must be present in event");
        // Sanity: the engine post-mutation can be inspected — proves
        // the engine isn't in a broken / locked state after emit.
        let entity = engine.get_entity(&outcome.id).expect("entity exists");
        assert_eq!(entity.title, "Hello");
    }
}