stream/
items.rs

1//! What travels on the stream: [`LaneItem`]s down each sink's lane,
2//! [`Request`]s up the request lane, and the [`Stream`] port that hands out
3//! their ends and owns the watermark.
4//!
5//! ```
6//! use kernel::Position;
7//! use stream::{Batch, LaneItem};
8//!
9//! let live = LaneItem::Batch(Batch { position: Some(Position(7)), changes: 3, envelopes: Vec::new().into() });
10//! let snapshot = LaneItem::Batch(Batch { position: None, changes: 0, envelopes: Vec::new().into() });
11//! assert_eq!(live.position(), Some(Position(7)));
12//! assert_eq!(snapshot.position(), None, "snapshot rows never move the watermark");
13//! ```
14
15use std::sync::Arc;
16
17use kernel::{Envelope, IndexName, Position, SinkName};
18use serde::{Deserialize, Serialize};
19
20/// The unit a lane carries down to one sink engine.
21#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
22#[serde(rename_all = "snake_case")]
23pub enum LaneItem {
24    /// The envelopes one ingest commit built, in build order.
25    Batch(Batch),
26    /// The snapshot the lane requested for `indexes` is fully published; the
27    /// sink engine may now record them as seeded.
28    SnapshotComplete { indexes: Vec<IndexName> },
29}
30
31/// The envelopes one ingest commit built, in build order, with the position of
32/// the last change the batch covers. A live batch with no envelopes (every
33/// change resolved to nothing this sink indexes) still carries its position,
34/// so acknowledging it moves the watermark. A snapshot batch carries none.
35///
36/// The envelopes are shared, not owned: one build is published to every lane
37/// and held for redelivery without copying the documents, so a batch is
38/// immutable once built — a sink stamps nothing into it.
39#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
40pub struct Batch {
41    /// The position of the last change in the batch; `None` for snapshot rows.
42    #[serde(default, skip_serializing_if = "Option::is_none")]
43    pub position: Option<Position>,
44    /// How many source changes the batch was built from, for progress
45    /// accounting; `0` for a snapshot batch.
46    #[serde(default)]
47    pub changes: usize,
48    /// The documents, in build order.
49    pub envelopes: Arc<[Envelope]>,
50}
51
52impl LaneItem {
53    /// The position acknowledging this item confirms, if any.
54    pub fn position(&self) -> Option<Position> {
55        match self {
56            LaneItem::Batch(batch) => batch.position,
57            LaneItem::SnapshotComplete { .. } => None,
58        }
59    }
60}
61
62/// What a sink engine sends up the request lane.
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
64#[serde(rename_all = "snake_case")]
65pub enum Request {
66    /// Snapshot `indexes` into `sink`'s lane, ending with a
67    /// [`LaneItem::SnapshotComplete`]. Acknowledged by the ingest engine only
68    /// after that marker is published, so a crash mid-snapshot redelivers it.
69    Backfill {
70        sink: SinkName,
71        indexes: Vec<IndexName>,
72    },
73}
74
75/// The two ends of one sink's lane.
76#[derive(Debug)]
77pub struct Lane {
78    /// Where the ingest engine publishes.
79    pub producer: Box<dyn crate::Producer<LaneItem>>,
80    /// Where that sink's engine receives.
81    pub consumer: Box<dyn crate::Consumer<LaneItem>>,
82}
83
84/// The two ends of the request lane.
85#[derive(Debug)]
86pub struct RequestLane {
87    /// Where a sink engine publishes its requests.
88    pub producer: Box<dyn crate::Producer<Request>>,
89    /// Where the ingest engine receives them.
90    pub consumer: Box<dyn crate::Consumer<Request>>,
91}
92
93/// The stream port: one lane per sink downward, one request lane upward, and
94/// the watermark that ties every lane's acknowledgements together.
95///
96/// `lane` and `requests` are idempotent: calling them again returns fresh ends
97/// over the *same* lane, which is how the daemon re-attaches a restarted engine
98/// (the previous consumer's unacknowledged item is redelivered to the new one).
99pub trait Stream: std::fmt::Debug + Send + Sync {
100    /// The lane for `sink`. An unknown sink is an error: the set of lanes is
101    /// fixed when the stream is built.
102    fn lane(&self, sink: &SinkName) -> crate::Result<Lane>;
103
104    /// The request lane.
105    fn requests(&self) -> crate::Result<RequestLane>;
106
107    /// The lowest position every lane has acknowledged: what the source may
108    /// confirm. `None` until every lane has acknowledged a positioned batch.
109    fn watermark(&self) -> Option<Position>;
110
111    /// Whether every lane is drained: nothing queued and nothing delivered but
112    /// unacknowledged. The daemon waits for this before it stops the sink
113    /// engines after the source stream ended.
114    fn is_idle(&self) -> bool;
115}
stream/items.rs

stream/
items.rs
