demo_feed_publisher/
cadence.rs

1use std::time::Duration;
2
3use liminal_sdk::{OBSERVABILITY_CHANNEL, PushWriter, SdkError};
4
5use crate::authority::{AuthorityError, Cut, FeedAuthority};
6use crate::envelope::{
7    ComponentId, ContractId, EnvelopeCodec, EnvelopeError, EnvelopeHeader, FrameKind,
8};
9use crate::graph::{GraphError, GraphState};
10
11/// Demo-only content pacing: four updates per second keeps motion legible without
12/// making this wall clock a protocol, authority, reconnect, retry, or delivery timer.
13pub const TICK_INTERVAL: Duration = Duration::from_millis(250);
14
15/// Replace the whole graph after twenty deltas: at four updates per second a new
16/// generation baseline bounds passive demo resynchronization to roughly five seconds.
17pub const SNAPSHOT_PERIOD: usize = 20;
18
19#[derive(Clone, Copy, Debug, PartialEq, Eq)]
20pub enum PublishOutcome {
21    /// The opaque publish frame was written; this is not a delivery acknowledgement.
22    Written,
23}
24
25#[derive(Debug, thiserror::Error)]
26pub enum PublishError {
27    #[error("SDK opaque publish failed: {0}")]
28    Sdk(#[from] SdkError),
29}
30
31pub trait PublishSink {
32    fn publish(&mut self, channel: &str, payload: Vec<u8>) -> Result<PublishOutcome, PublishError>;
33}
34
35impl PublishSink for PushWriter {
36    fn publish(&mut self, channel: &str, payload: Vec<u8>) -> Result<PublishOutcome, PublishError> {
37        Self::publish(self, channel, payload)?;
38        Ok(PublishOutcome::Written)
39    }
40}
41
42#[derive(Debug, thiserror::Error)]
43pub enum CadenceError {
44    #[error("demo feed channel cannot be empty")]
45    EmptyChannel,
46    #[error("demo feed must not publish to reserved observability channel {0:?}")]
47    ReservedChannel(&'static str),
48    #[error("snapshot period must be greater than zero")]
49    ZeroSnapshotPeriod,
50    #[error("initial snapshot has already been emitted")]
51    AlreadyStarted,
52    #[error("initial snapshot must be emitted before a delta")]
53    NotStarted,
54    #[error(transparent)]
55    Authority(#[from] AuthorityError),
56    #[error(transparent)]
57    Envelope(#[from] EnvelopeError),
58    #[error(transparent)]
59    Graph(#[from] GraphError),
60    #[error(transparent)]
61    Publish(#[from] PublishError),
62    #[error("encoded envelope failed its own round-trip invariant")]
63    EncodedFrameMismatch,
64}
65
66/// Deterministic cut and snapshot/delta policy; it owns no wall clock or reconnect policy.
67/// A standalone authority-snapshot request is the named post-demo resync mechanism.
68pub struct CadenceEngine<C, G, S> {
69    channel: String,
70    component_id: ComponentId,
71    contract: ContractId,
72    authority: FeedAuthority,
73    codec: C,
74    graph: G,
75    sink: S,
76    snapshot_period: usize,
77    deltas_since_snapshot: usize,
78    started: bool,
79}
80
81impl<C, G, S> CadenceEngine<C, G, S>
82where
83    C: EnvelopeCodec,
84    G: GraphState,
85    S: PublishSink,
86{
87    pub fn new(
88        channel: String,
89        component_id: ComponentId,
90        authority: FeedAuthority,
91        codec: C,
92        graph: G,
93        sink: S,
94        snapshot_period: usize,
95    ) -> Result<Self, CadenceError> {
96        if channel.is_empty() {
97            return Err(CadenceError::EmptyChannel);
98        }
99        if channel == OBSERVABILITY_CHANNEL {
100            return Err(CadenceError::ReservedChannel(OBSERVABILITY_CHANNEL));
101        }
102        if snapshot_period == 0 {
103            return Err(CadenceError::ZeroSnapshotPeriod);
104        }
105        let contract = ContractId::new("frame", "graph-view", 1)?;
106        Ok(Self {
107            channel,
108            component_id,
109            contract,
110            authority,
111            codec,
112            graph,
113            sink,
114            snapshot_period,
115            deltas_since_snapshot: 0,
116            started: false,
117        })
118    }
119
120    pub fn emit_initial_snapshot(&mut self) -> Result<PublishOutcome, CadenceError> {
121        if self.started {
122            return Err(CadenceError::AlreadyStarted);
123        }
124        let state = self.graph.snapshot_bytes()?;
125        let cut = self.authority.mint_baseline(&self.channel)?;
126        let outcome = self.publish(FrameKind::Snapshot, cut, &state)?;
127        self.started = true;
128        Ok(outcome)
129    }
130
131    pub fn emit_tick(&mut self) -> Result<PublishOutcome, CadenceError> {
132        if !self.started {
133            return Err(CadenceError::NotStarted);
134        }
135        let delta = self.graph.advance_delta_bytes()?;
136        let cut = self.authority.mint_delta(&self.channel)?;
137        let delta_outcome = self.publish(FrameKind::Delta, cut, &delta)?;
138        self.deltas_since_snapshot += 1;
139        if self.deltas_since_snapshot == self.snapshot_period {
140            let snapshot = self.graph.refresh_snapshot_bytes()?;
141            self.authority.advance_generation()?;
142            let baseline = self.authority.mint_baseline(&self.channel)?;
143            let snapshot_outcome = self.publish(FrameKind::Snapshot, baseline, &snapshot)?;
144            self.deltas_since_snapshot = 0;
145            Ok(snapshot_outcome)
146        } else {
147            Ok(delta_outcome)
148        }
149    }
150
151    fn publish(
152        &mut self,
153        kind: FrameKind,
154        cut: Cut,
155        state: &[u8],
156    ) -> Result<PublishOutcome, CadenceError> {
157        let header = EnvelopeHeader {
158            component_id: self.component_id.clone(),
159            contract_id: self.contract.clone(),
160            generation: cut.generation().get(),
161            kind,
162            seq: cut.seq(),
163        };
164        let payload = self.codec.encode(&header, state)?;
165        let decoded = self.codec.decode(&payload)?;
166        if decoded.header != header || decoded.state != state {
167            return Err(CadenceError::EncodedFrameMismatch);
168        }
169        Ok(self.sink.publish(&self.channel, payload)?)
170    }
171}
172
173#[cfg(test)]
174mod tests {
175    use std::cell::Cell;
176    use std::rc::Rc;
177
178    use super::{
179        CadenceEngine, CadenceError, PublishError, PublishOutcome, PublishSink, SNAPSHOT_PERIOD,
180    };
181    use crate::authority::{FeedAuthority, GenerationStore, GenerationStoreError};
182    use crate::envelope::{ComponentId, EnvelopeCodec, FrameEnvelopeCodec, FrameKind};
183    use crate::graph::GraphViewState;
184
185    #[derive(Clone, Default)]
186    struct MemoryGenerationStore(Rc<Cell<Option<u64>>>);
187
188    impl GenerationStore for MemoryGenerationStore {
189        fn load(&mut self) -> Result<Option<u64>, GenerationStoreError> {
190            Ok(self.0.get())
191        }
192
193        fn persist(&mut self, generation: u64) -> Result<(), GenerationStoreError> {
194            self.0.set(Some(generation));
195            Ok(())
196        }
197    }
198
199    #[derive(Default)]
200    struct FakeSink {
201        frames: Vec<Vec<u8>>,
202        channels: Vec<String>,
203        fail_at: Option<usize>,
204    }
205
206    impl PublishSink for FakeSink {
207        fn publish(
208            &mut self,
209            channel: &str,
210            payload: Vec<u8>,
211        ) -> Result<PublishOutcome, PublishError> {
212            if self.fail_at == Some(self.frames.len()) {
213                return Err(PublishError::Sdk(liminal_sdk::SdkError::Connection {
214                    description: "injected sink failure".to_owned(),
215                }));
216            }
217            self.channels.push(channel.to_owned());
218            self.frames.push(payload);
219            Ok(PublishOutcome::Written)
220        }
221    }
222
223    fn engine(
224        sink: FakeSink,
225        period: usize,
226    ) -> Result<
227        CadenceEngine<FrameEnvelopeCodec, GraphViewState, FakeSink>,
228        Box<dyn std::error::Error>,
229    > {
230        Ok(CadenceEngine::new(
231            "frame.demo.graph-view".to_owned(),
232            ComponentId::new("graph-view-demo")?,
233            FeedAuthority::start(MemoryGenerationStore::default())?,
234            FrameEnvelopeCodec,
235            GraphViewState::new()?,
236            sink,
237            period,
238        )?)
239    }
240
241    #[test]
242    fn baseline_snapshot_precedes_monotonic_deltas() -> Result<(), Box<dyn std::error::Error>> {
243        let mut cadence = engine(FakeSink::default(), SNAPSHOT_PERIOD)?;
244        cadence.emit_initial_snapshot()?;
245        cadence.emit_tick()?;
246        let first = FrameEnvelopeCodec.decode(&cadence.sink.frames[0])?;
247        let second = FrameEnvelopeCodec.decode(&cadence.sink.frames[1])?;
248        assert_eq!(first.header.kind, FrameKind::Snapshot);
249        assert_eq!(second.header.kind, FrameKind::Delta);
250        assert_eq!((first.header.generation, first.header.seq), (1, 0));
251        assert_eq!((second.header.generation, second.header.seq), (1, 1));
252        assert!(
253            cadence
254                .sink
255                .channels
256                .iter()
257                .all(|channel| channel == "frame.demo.graph-view")
258        );
259        Ok(())
260    }
261
262    #[test]
263    fn periodic_snapshot_bumps_generation_and_resets_sequence()
264    -> Result<(), Box<dyn std::error::Error>> {
265        let mut cadence = engine(FakeSink::default(), SNAPSHOT_PERIOD)?;
266        cadence.emit_initial_snapshot()?;
267        for _ in 0..SNAPSHOT_PERIOD {
268            cadence.emit_tick()?;
269        }
270        let headers: Result<Vec<_>, _> = cadence
271            .sink
272            .frames
273            .iter()
274            .map(|frame| {
275                FrameEnvelopeCodec
276                    .decode(frame)
277                    .map(|decoded| decoded.header)
278            })
279            .collect();
280        let headers = headers?;
281        assert_eq!(headers.len(), SNAPSHOT_PERIOD + 2);
282        assert_eq!(
283            headers.first().map(|header| header.kind),
284            Some(FrameKind::Snapshot)
285        );
286        assert_eq!(
287            headers.last().map(|header| header.kind),
288            Some(FrameKind::Snapshot)
289        );
290        assert_eq!(
291            headers
292                .first()
293                .map(|header| (header.generation, header.seq)),
294            Some((1, 0))
295        );
296        assert_eq!(
297            headers.last().map(|header| (header.generation, header.seq)),
298            Some((2, 0))
299        );
300        assert!(
301            headers[1..=SNAPSHOT_PERIOD]
302                .iter()
303                .enumerate()
304                .all(|(index, header)| {
305                    header.kind == FrameKind::Delta
306                        && header.generation == 1
307                        && usize::try_from(header.seq) == Ok(index + 1)
308                })
309        );
310        let cuts: Vec<_> = headers
311            .iter()
312            .map(|header| (header.generation, header.seq))
313            .collect();
314        assert!(cuts.windows(2).all(|pair| pair[0] < pair[1]));
315        Ok(())
316    }
317
318    #[test]
319    fn sink_failure_is_typed_and_its_cut_is_not_reused() -> Result<(), Box<dyn std::error::Error>> {
320        let sink = FakeSink {
321            frames: Vec::new(),
322            channels: Vec::new(),
323            fail_at: Some(1),
324        };
325        let mut cadence = engine(sink, SNAPSHOT_PERIOD)?;
326        cadence.emit_initial_snapshot()?;
327        let error = cadence.emit_tick();
328        assert!(matches!(error, Err(CadenceError::Publish(_))));
329        cadence.sink.fail_at = None;
330        cadence.emit_tick()?;
331        let after_gap = FrameEnvelopeCodec.decode(&cadence.sink.frames[1])?;
332        assert_eq!((after_gap.header.generation, after_gap.header.seq), (1, 2));
333        Ok(())
334    }
335
336    #[test]
337    fn reserved_observability_channel_is_refused() -> Result<(), Box<dyn std::error::Error>> {
338        let result = CadenceEngine::new(
339            liminal_sdk::OBSERVABILITY_CHANNEL.to_owned(),
340            ComponentId::new("graph-view-demo")?,
341            FeedAuthority::start(MemoryGenerationStore::default())?,
342            FrameEnvelopeCodec,
343            GraphViewState::new()?,
344            FakeSink::default(),
345            SNAPSHOT_PERIOD,
346        );
347        assert!(matches!(result, Err(CadenceError::ReservedChannel(_))));
348        Ok(())
349    }
350}
demo_feed_publisher/cadence.rs

demo_feed_publisher/
cadence.rs
