mold_server/
events.rs

1//! Server-wide lifecycle event broadcast.
2//!
3//! One `tokio::sync::broadcast` channel fans out [`mold_core::ServerEvent`]s
4//! (job lifecycle + gallery mutations) to every subscriber of
5//! `GET /api/events`, so a client can watch the whole server over a single
6//! SSE connection instead of one per job. Unlike [`crate::resources`] there
7//! is no "latest" cache — these are deltas, not snapshots; new subscribers
8//! bootstrap from `GET /api/queue` + `GET /api/gallery`.
9
10use mold_core::ServerEvent;
11use std::sync::Arc;
12use tokio::sync::broadcast;
13
14/// Events are bursty (a batch submit queues several jobs at once), so the
15/// buffer is larger than the resources channel's. Lagged receivers skip the
16/// gap and recover via the REST endpoints.
17const BROADCAST_BUFFER: usize = 64;
18
19pub struct EventBroadcaster {
20    tx: broadcast::Sender<ServerEvent>,
21}
22
23impl EventBroadcaster {
24    pub fn new() -> Arc<Self> {
25        let (tx, _rx) = broadcast::channel(BROADCAST_BUFFER);
26        Arc::new(Self { tx })
27    }
28
29    /// Publish an event. Synchronous — safe to call from `spawn_blocking`
30    /// contexts (the queue worker's save path). No-subscriber send errors
31    /// are deliberately ignored.
32    pub fn publish(&self, event: ServerEvent) {
33        let _ = self.tx.send(event);
34    }
35
36    pub fn subscribe(&self) -> broadcast::Receiver<ServerEvent> {
37        self.tx.subscribe()
38    }
39}
40
41#[cfg(test)]
42mod tests {
43    use super::*;
44
45    #[tokio::test]
46    async fn publish_reaches_all_subscribers() {
47        let events = EventBroadcaster::new();
48        let mut a = events.subscribe();
49        let mut b = events.subscribe();
50
51        events.publish(ServerEvent::JobEnded { id: "j1".into() });
52
53        for rx in [&mut a, &mut b] {
54            match rx.recv().await.unwrap() {
55                ServerEvent::JobEnded { id } => assert_eq!(id, "j1"),
56                other => panic!("unexpected event: {other:?}"),
57            }
58        }
59    }
60
61    #[tokio::test]
62    async fn publish_without_subscribers_does_not_panic() {
63        let events = EventBroadcaster::new();
64        events.publish(ServerEvent::JobQueued {
65            id: "j1".into(),
66            model: "flux-dev:q4".into(),
67        });
68        // A subscriber attached afterwards sees nothing (no replay cache).
69        let mut rx = events.subscribe();
70        assert!(matches!(
71            rx.try_recv(),
72            Err(broadcast::error::TryRecvError::Empty)
73        ));
74    }
75}
mold_server/events.rs

mold_server/
events.rs
