cratestack_axum/rpc/
subscription_bridge.rs

1//! Bridges a `CratestackEventBus`-style push callback into a bounded,
2//! backpressure-aware `Stream` for the SSE encoder ([`super::sse`]). See
3//! `docs/design/rpc-transport.md` §3.4/§3.4a: "bounded per-subscription
4//! send buffer; on overflow, emit Error{unavailable} ... and end the
5//! stream."
6//!
7//! `CratestackEventBus::subscribe` handlers must never block or fail the
8//! emitting transaction just because one particular SSE client is slow
9//! to drain, so the push side here is always non-blocking (`try_send`)
10//! and infallible from the bus's point of view — overflow is signaled to
11//! the *consumer* by closing the channel, never back to the caller of
12//! `emit()`. See [`super::sse`]'s module doc for why "the stream ends"
13//! is therefore an unambiguous overflow signal, never confused with an
14//! ordinary client disconnect (which just drops the whole future
15//! instead of closing this channel).
16
17use std::sync::{Arc, Mutex};
18
19use futures_util::Stream;
20use futures_util::stream;
21use tokio::sync::mpsc;
22
23/// One SSE subscription's outbox capacity before it's considered lagged.
24/// A slow consumer past this many buffered, undelivered events gets
25/// disconnected with `Error{unavailable}` (see [`super::sse`]) rather
26/// than growing memory unboundedly.
27const SUBSCRIPTION_BUFFER_CAPACITY: usize = 64;
28
29/// Handed to one or more `CratestackEventBus::subscribe` callbacks registered
30/// for the same logical subscription (e.g. one per `@@emit`ted operation
31/// on a model). Every clone shares the same underlying sender slot, so
32/// the *first* overflow observed by any of them permanently closes the
33/// channel — subsequent pushes from any clone become silent no-ops.
34pub struct SubscriptionPush<T> {
35    slot: Arc<Mutex<Option<mpsc::Sender<T>>>>,
36}
37
38impl<T> Clone for SubscriptionPush<T> {
39    fn clone(&self) -> Self {
40        Self {
41            slot: Arc::clone(&self.slot),
42        }
43    }
44}
45
46impl<T: Send + 'static> SubscriptionPush<T> {
47    /// Non-blocking push. Never fails from the caller's perspective —
48    /// overflow just closes the slot so this and every other clone
49    /// becomes a no-op from then on; the *consumer* observes that as the
50    /// stream ending (see [`guarded_receiver_stream`]).
51    pub fn push(&self, value: T) {
52        let sender = {
53            let guard = self.slot.lock().expect("subscription sender slot poisoned");
54            guard.clone()
55        };
56        let Some(sender) = sender else {
57            return;
58        };
59        if sender.try_send(value).is_err() {
60            // Either full (backpressure) or the receiver already
61            // dropped (client disconnected without ever overflowing —
62            // harmless to also close here, `guarded_receiver_stream`'s
63            // future was already being torn down for that case anyway).
64            *self.slot.lock().expect("subscription sender slot poisoned") = None;
65        }
66    }
67}
68
69/// Builds a fresh bounded channel plus the [`SubscriptionPush`] handle
70/// callers clone into every `CratestackEventBus::subscribe` closure that
71/// should feed it.
72pub fn subscription_channel<T: Send + 'static>() -> (SubscriptionPush<T>, mpsc::Receiver<T>) {
73    let (tx, rx) = mpsc::channel(SUBSCRIPTION_BUFFER_CAPACITY);
74    (
75        SubscriptionPush {
76            slot: Arc::new(Mutex::new(Some(tx))),
77        },
78        rx,
79    )
80}
81
82/// Wraps a raw `mpsc::Receiver` into a `Stream`, keeping `guard` alive
83/// for exactly as long as the stream is — dropped together whether the
84/// stream ends normally (overflow, see [`SubscriptionPush`]) or is
85/// cancelled mid-poll (an ordinary client disconnect just drops this
86/// whole future). This is what lets a `cratestack_core::SubscriptionGuard`
87/// passed as `guard` unsubscribe cleanly in either case without the
88/// caller needing to distinguish which one happened.
89pub fn guarded_receiver_stream<T, G>(
90    rx: mpsc::Receiver<T>,
91    guard: G,
92) -> impl Stream<Item = T> + Send + 'static
93where
94    T: Send + 'static,
95    G: Send + 'static,
96{
97    stream::unfold((rx, guard), |(mut rx, guard)| async move {
98        let item = rx.recv().await?;
99        Some((item, (rx, guard)))
100    })
101}
102
103#[cfg(test)]
104mod tests;
cratestack_axum/rpc/subscription_bridge.rs

cratestack_axum/rpc/
subscription_bridge.rs
