crabka_replicator/
offset_sync_store.rs

1//! In-memory source->target offset translation built from MM2 offset-sync records.
2//! Translation is "at-or-before": find the latest sync with upstream <= committed,
3//! then downstream + (committed - upstream). Never maps past un-replicated data.
4
5use std::collections::BTreeMap;
6
7use crate::mm2::OffsetSync;
8
9/// Accumulates [`OffsetSync`] records and translates committed source offsets to
10/// their corresponding target offsets.
11#[derive(Default)]
12pub struct OffsetSyncStore {
13    /// Outer key: (topic, partition). Inner key: upstream offset → downstream offset.
14    syncs: BTreeMap<(String, i32), BTreeMap<i64, i64>>,
15}
16
17impl OffsetSyncStore {
18    /// Ingest one offset-sync record, replacing any prior entry for the same
19    /// `(topic, partition, upstream)` triple.
20    pub fn ingest(&mut self, s: OffsetSync) {
21        self.syncs
22            .entry((s.topic, s.partition))
23            .or_default()
24            .insert(s.upstream, s.downstream);
25    }
26
27    /// Translate a committed source offset to its target offset.
28    ///
29    /// Returns `None` when:
30    /// - the `(topic, partition)` pair has no syncs, or
31    /// - `committed` is below every known upstream offset (un-replicated data).
32    #[must_use]
33    pub fn translate(&self, topic: &str, partition: i32, committed: i64) -> Option<i64> {
34        let m = self.syncs.get(&(topic.to_string(), partition))?;
35        let (&up, &down) = m.range(..=committed).next_back()?;
36        Some(down + (committed - up))
37    }
38}
39
40#[cfg(test)]
41mod tests {
42    use assert2::assert;
43
44    use super::*;
45    use crate::mm2::OffsetSync;
46
47    #[test]
48    fn translates_at_or_before_nearest_sync() {
49        let mut s = OffsetSyncStore::default();
50        s.ingest(OffsetSync {
51            topic: "orders".into(),
52            partition: 0,
53            upstream: 100,
54            downstream: 70,
55        });
56        s.ingest(OffsetSync {
57            topic: "orders".into(),
58            partition: 0,
59            upstream: 200,
60            downstream: 165,
61        });
62        assert!(s.translate("orders", 0, 250) == Some(215)); // 165 + (250-200)
63        assert!(s.translate("orders", 0, 150) == Some(120)); // 70 + (150-100)
64        assert!(s.translate("orders", 0, 50) == None); // below first sync
65        assert!(s.translate("orders", 9, 100) == None); // unknown partition
66    }
67}
crabka_replicator/offset_sync_store.rs

crabka_replicator/
offset_sync_store.rs
