cratestack_axum/rpc/
model_feed.rs

1//! A `@@subscribe` model feed as a [`SubscriptionStream`]: what the
2//! generated `SubscriptionDispatcher` hands every binding — SSE,
3//! WebTransport, WebSocket (`docs/design/subscriptions-websocket.md` §6.1).
4//!
5//! The events come off the subscription backplane ([`model_events`]), so a
6//! write committed on any replica reaches a subscriber on this one (§4). Each
7//! event is routed per subscriber ([`route_event`]) before it is sent: it
8//! passes the subscriber's filter and read policy, or is dropped, or, for an
9//! update that moved the row out of both, becomes a filter-exit event (§5.3).
10//! The feed never ends on its own, so every ending is an error the binding
11//! sends as its terminal frame:
12//!
13//! - this subscription fell behind its buffer: `unavailable` /
14//!   `"subscription lagged"`, the frame the SSE binding has always sent;
15//! - the backplane may have lost events (its listener reconnected), or went
16//!   away: `unavailable` / `"subscription interrupted"` (§4.5);
17//! - routing failed: the policy could not be decided for one event, so that
18//!   error, rather than an event nobody checked or a silent gap.
19
20use std::future::Future;
21use std::pin::Pin;
22
23use cratestack_core::{
24    BackplaneMessage, BackplaneReceiver, CratestackError, CratestackEventEnvelope, ModelEvent,
25    SubscriptionStream,
26};
27use futures_util::Stream;
28use futures_util::stream::{self, StreamExt};
29use serde::Serialize;
30use serde::de::DeserializeOwned;
31
32/// The error a feed ends with when its buffer overflowed.
33pub(crate) fn lagged() -> CratestackError {
34    CratestackError::Unavailable("subscription lagged".to_owned())
35}
36
37fn interrupted() -> CratestackError {
38    CratestackError::Unavailable("subscription interrupted".to_owned())
39}
40
41/// A model event as it comes off the backplane: the event a subscriber may be
42/// sent, and, for an update on an `@@subscribe` model, the row before it. The
43/// pre-image stays server-side: it is only ever the `data` of a filter-exit
44/// event, after the subscriber's own read check ([`route_event`]).
45pub struct FeedEvent<T> {
46    pub event: ModelEvent<T>,
47    pub before: Option<T>,
48}
49
50impl<T: DeserializeOwned> TryFrom<CratestackEventEnvelope> for FeedEvent<T> {
51    type Error = CratestackError;
52
53    fn try_from(mut envelope: CratestackEventEnvelope) -> Result<Self, Self::Error> {
54        let before = envelope
55            .before
56            .take()
57            .map(serde_json::from_value)
58            .transpose()
59            .map_err(|error| {
60                CratestackError::Codec(format!("failed to decode event pre-image: {error}"))
61            })?;
62        Ok(Self {
63            event: envelope.try_into()?,
64            before,
65        })
66    }
67}
68
69/// The model events published on a backplane channel, decoded. A payload
70/// that does not decode (a replica on another schema version mid-deploy)
71/// ends the feed: the client refetches rather than miss an event.
72pub fn model_events<T>(
73    receiver: BackplaneReceiver,
74) -> impl Stream<Item = Result<FeedEvent<T>, CratestackError>> + Send + 'static
75where
76    T: DeserializeOwned + Send + 'static,
77{
78    receiver.map(|message| match message {
79        BackplaneMessage::Payload(json) => serde_json::from_str::<CratestackEventEnvelope>(&json)
80            .map_err(|error| CratestackError::Codec(format!("unreadable model event: {error}")))
81            .and_then(FeedEvent::try_from),
82        BackplaneMessage::Lagged => Err(lagged()),
83        BackplaneMessage::Resync => Err(interrupted()),
84    })
85}
86
87/// What one subscriber is sent for `feed`, if anything: the event itself when
88/// its row passes the subscriber's filter and read policy (`matches`,
89/// `readable`, both on a row); otherwise, for an update, a filter-exit event
90/// (`matches: false`, §5.3) if the row before the write had passed both, so
91/// the subscriber's copy is not left stale.
92///
93/// A filter-exit event's `data` is the pre-image and never the post-image: the
94/// subscriber already holds that row (it passed the same checks, as a delete
95/// event's row does), while the post-image may hold values it may not read,
96/// which is why the row left. A row that moved out of the read policy is
97/// therefore indistinguishable, to the subscriber, from one that left its
98/// filter. An update with no pre-image (a bulk update) yields no exit event.
99pub async fn route_event<T, M, R, Fut>(
100    feed: FeedEvent<T>,
101    wanted: bool,
102    matches: M,
103    readable: R,
104) -> Result<Option<ModelEvent<T>>, CratestackError>
105where
106    M: Fn(&T) -> bool,
107    R: Fn(&T) -> Fut,
108    Fut: Future<Output = Result<bool, CratestackError>>,
109{
110    let FeedEvent { mut event, before } = feed;
111    if !wanted {
112        return Ok(None);
113    }
114    // The filter first: it needs no I/O.
115    if matches(&event.data) && readable(&event.data).await? {
116        return Ok(Some(event));
117    }
118    let Some(before) = before else {
119        return Ok(None);
120    };
121    if !(matches(&before) && readable(&before).await?) {
122        return Ok(None);
123    }
124    event.data = before;
125    event.matches = false;
126    Ok(Some(event))
127}
128
129/// `items` routed by `route` and encoded to JSON values, ending with the error
130/// that stopped it (module docs). `route` returns what to send for an item, or
131/// nothing to drop it.
132pub fn model_feed<I, O, S, F, Fut>(items: S, route: F) -> SubscriptionStream<serde_json::Value>
133where
134    O: Serialize + Send + 'static,
135    I: Send + 'static,
136    S: Stream<Item = Result<I, CratestackError>> + Send + 'static,
137    F: FnMut(I) -> Fut + Send + 'static,
138    Fut: Future<Output = Result<Option<O>, CratestackError>> + Send + 'static,
139{
140    let state: (Pin<Box<S>>, F, bool) = (Box::pin(items), route, false);
141    Box::pin(stream::unfold(
142        state,
143        |(mut items, mut route, ended)| async move {
144            if ended {
145                return None;
146            }
147            loop {
148                let item = match items.next().await {
149                    Some(Ok(item)) => item,
150                    Some(Err(error)) => return Some((Err(error), (items, route, true))),
151                    None => return Some((Err(interrupted()), (items, route, true))),
152                };
153                let verdict = match route(item).await {
154                    Ok(Some(event)) => serde_json::to_value(&event).map_err(|error| {
155                        CratestackError::Codec(format!("failed to encode a model event: {error}"))
156                    }),
157                    Ok(None) => continue,
158                    Err(error) => {
159                        tracing::warn!(
160                            target: "cratestack",
161                            %error,
162                            "subscription read policy undecided; ending the stream",
163                        );
164                        Err(error)
165                    }
166                };
167                let ended = verdict.is_err();
168                return Some((verdict, (items, route, ended)));
169            }
170        },
171    ))
172}
173
174#[cfg(test)]
175mod tests;
cratestack_axum/rpc/model_feed.rs

cratestack_axum/rpc/
model_feed.rs
