laser_wire/snapshot.rs
1use crate::agent::ConversationId;
2use serde::{Deserialize, Serialize};
3use std::collections::BTreeMap;
4
5/// A fold snapshot: the folded `state` plus the per-partition offsets it folded
6/// through. `as_of` maps a partition id to the last offset folded into `state`
7/// (inclusive), matching the cursor's per-partition offsets, so a multi-partition
8/// fold resumes correctly. Resume seeds the cursor at `offset + 1` per partition,
9/// because the cursor takes the next offset to read (exclusive) while the
10/// snapshot records the last offset folded (inclusive).
11#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
12pub struct FoldSnapshot {
13 /// The fold this snapshots (the conversation or journal partition key).
14 pub conversation: ConversationId,
15 /// Per-partition last folded offset, inclusive. A windowless fold over one
16 /// partition is a single entry.
17 pub as_of: BTreeMap<u32, u64>,
18 /// The opaque folded state. The codec is the producer's choice, the wire
19 /// crate never inspects it.
20 #[serde(with = "crate::encoding::bin_bytes")]
21 pub state: Vec<u8>,
22}
23
24impl FoldSnapshot {
25 /// The offset a partition's resume should start reading from: one past the
26 /// last folded offset, or `0` for a partition the snapshot did not cover.
27 pub fn resume_offset(&self, partition: u32) -> u64 {
28 self.as_of
29 .get(&partition)
30 .map_or(0, |offset| offset.saturating_add(1))
31 }
32}
33
34#[cfg(test)]
35mod tests {
36 use super::*;
37
38 #[test]
39 fn given_a_snapshot_when_asked_for_resume_then_should_return_one_past_the_folded_offset() {
40 let snapshot = FoldSnapshot {
41 conversation: ConversationId::from_u128(1),
42 as_of: BTreeMap::from([(0, 41), (1, 9)]),
43 state: vec![1, 2, 3],
44 };
45 assert_eq!(snapshot.resume_offset(0), 42);
46 assert_eq!(snapshot.resume_offset(1), 10);
47 // A partition the snapshot never folded resumes from zero.
48 assert_eq!(snapshot.resume_offset(2), 0);
49 }
50}