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 plane 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}
26
27#[cfg(all(test, feature = "cbor"))]
28mod tests {
29    use super::*;
30    use crate::codes::CHANGE_OP_VERSION;
31    use crate::framing::{decode_named, encode_named};
32
33    #[test]
34    fn given_a_change_record_when_round_tripped_then_should_decode_unchanged() {
35        let record = ChangeRecord {
36            v: CHANGE_OP_VERSION,
37            index: "orders_v1".to_owned(),
38            partition_id: 3,
39            from_offset: 100,
40            to_offset: 141,
41            rows: 42,
42        };
43        let bytes = encode_named(&record).expect("encodes");
44        let back: ChangeRecord = decode_named(&bytes).expect("decodes");
45        assert_eq!(back, record);
46    }
47}
laser_wire/change.rs

laser_wire/
change.rs
