laser_wire/
change.rs

1use serde::{Deserialize, Serialize};
2
3/// One materialized-view advancement on the change feed: after committing a
4/// projector batch for a binding that opted in (`ProjectionBinding.notify`),
5/// the managed backend publishes one record naming the index, the partition, the offset
6/// window the batch covered, and how many rows landed. A consumer awaits the
7/// record then queries, instead of sleeping and retrying: change notification
8/// as records on a topic, consumed by offset, never a server-push watch (the
9/// substrate is a log). v1 scope is materialized-view advancement. Key-value
10/// and run-registry change records are candidate extensions of the same shape,
11/// not implied.
12#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
13pub struct ChangeRecord {
14    pub v: u32,
15    /// The materialized index the batch advanced.
16    pub index: String,
17    /// The source partition the batch covered.
18    pub partition_id: u32,
19    /// First source offset in the committed batch.
20    pub from_offset: u64,
21    /// Last source offset in the committed batch (the new watermark).
22    pub to_offset: u64,
23    /// Rows the batch landed in the view.
24    pub rows: u32,
25    /// The stream whose source the batch read, when the deployment publishes
26    /// change records per stream.
27    #[serde(default, skip_serializing_if = "Option::is_none")]
28    pub stream: Option<String>,
29}
30
31#[cfg(all(test, feature = "cbor"))]
32mod tests {
33    use super::*;
34    use crate::codes::CHANGE_OP_VERSION;
35    use crate::framing::{decode_named, encode_named};
36
37    #[test]
38    fn given_a_change_record_when_round_tripped_then_should_decode_unchanged() {
39        let record = ChangeRecord {
40            v: CHANGE_OP_VERSION,
41            index: "readings_v1".to_owned(),
42            partition_id: 3,
43            from_offset: 100,
44            to_offset: 141,
45            rows: 42,
46            stream: None,
47        };
48        let bytes = encode_named(&record).expect("encodes");
49        let back: ChangeRecord = decode_named(&bytes).expect("decodes");
50        assert_eq!(back, record);
51    }
52}
laser_wire/change.rs

laser_wire/
change.rs
