atomic_lib 0.41.0-beta.2

Library for creating, storing, querying, validating and converting Atomic Data.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
//! Range-based set reconciliation (RBSR) core.
//!
//! The flat whole-drive hash (`engine::compute_drive_hash`) can only answer
//! "are these two drives identical?" — on any difference it forces exchanging
//! the entire version-vector set to find the delta (O(drive)). RBSR instead
//! finds *where* two sorted `(subject → version vector)` sets differ by
//! recursively comparing **range fingerprints**, transferring only the items in
//! the ranges that actually differ. Cost tracks the divergence, not the drive.
//! See `planning/drive-reconciliation.md` (Phase 2).
//!
//! This module is the transport-free, storage-free algorithm core: fingerprint
//! primitives + the recursive reconcile driver, driven through callbacks that
//! stand in for "ask the remote node". Wiring it onto the WS/Iroh transports
//! and backing the range queries with an incrementally-maintained tree (so a
//! range fingerprint is O(log n) rather than O(range)) are later steps; the
//! algorithm and its convergence are pinned here first.
//!
//! **Fingerprint = XOR of per-item SHA-256 hashes.** XOR is an order-independent,
//! incremental monoid: a range's fingerprint is the XOR of its items' hashes in
//! any order, matching items cancel out, and an item can be added/removed by
//! XORing its hash in/out. Like all RBSR, this is probabilistic — two different
//! item sets could XOR to the same fingerprint with probability ~2⁻²⁵⁶, which
//! we treat as never.

use std::collections::BTreeMap;

use sha2::{Digest, Sha256};

/// A per-item or per-range fingerprint.
pub type Fingerprint = [u8; 32];

/// The empty-range fingerprint (identity for XOR).
pub const EMPTY: Fingerprint = [0u8; 32];

/// One reconciliation item: a subject and its version vector (peer → counter).
pub type Item = (String, BTreeMap<String, i32>);

/// Canonical per-item fingerprint: SHA-256 of `"{subject}={peer:counter,…}"`
/// with the `(peer, counter)` pairs sorted by peer (a `BTreeMap` iterates
/// sorted). Self-contained — it depends only on this item, not on any global
/// peer index — so two nodes fingerprint the same item identically. Same hash
/// family as the Phase 1 drive hash (SHA-256), reproducible byte-for-byte in
/// TypeScript when this is wired to the browser transport.
pub fn item_fingerprint(subject: &str, vv: &BTreeMap<String, i32>) -> Fingerprint {
    let pairs = vv
        .iter()
        .map(|(peer, counter)| format!("{peer}:{counter}"))
        .collect::<Vec<_>>()
        .join(",");

    let mut hasher = Sha256::new();
    hasher.update(subject.as_bytes());
    hasher.update(b"=");
    hasher.update(pairs.as_bytes());
    hasher.finalize().into()
}

fn xor_into(acc: &mut Fingerprint, x: &Fingerprint) {
    for (a, b) in acc.iter_mut().zip(x.iter()) {
        *a ^= *b;
    }
}

/// Fingerprint of the items whose subject falls in `[lo, hi)` (half-open;
/// `hi == None` means "unbounded above"). `items` MUST be sorted by subject.
/// O(range) — the incremental tree that makes this O(log n) is a later step.
pub fn range_fingerprint(items: &[Item], lo: &str, hi: Option<&str>) -> Fingerprint {
    let mut fp = EMPTY;
    for (subject, vv) in items_in_range(items, lo, hi) {
        let item = item_fingerprint(subject, vv);
        xor_into(&mut fp, &item);
    }
    fp
}

fn items_in_range<'a>(
    items: &'a [Item],
    lo: &str,
    hi: Option<&str>,
) -> impl Iterator<Item = &'a Item> {
    let lo = lo.to_string();
    let hi = hi.map(|h| h.to_string());
    items.iter().filter(move |(s, _)| {
        s.as_str() >= lo.as_str() && hi.as_deref().map(|h| s.as_str() < h).unwrap_or(true)
    })
}

/// The subjects that differ between the local and remote sets.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct Diff {
    /// Present locally, absent remotely.
    pub only_local: Vec<String>,
    /// Present remotely, absent locally.
    pub only_remote: Vec<String>,
    /// Present on both, but with different version vectors.
    pub differ: Vec<String>,
}

impl Diff {
    pub fn is_empty(&self) -> bool {
        self.only_local.is_empty() && self.only_remote.is_empty() && self.differ.is_empty()
    }
}

/// What the remote node answers when asked about a subject range.
pub trait RemoteRange {
    /// The remote's fingerprint for `[lo, hi)`.
    fn fingerprint(&mut self, lo: &str, hi: Option<&str>) -> Fingerprint;
    /// The remote's items in `[lo, hi)` — requested only for a small range.
    fn items(&mut self, lo: &str, hi: Option<&str>) -> Vec<Item>;
}

/// Reconcile the local set against a remote (accessed via `remote`), returning
/// the subjects that differ. `split` is the branching factor for a mismatched
/// range; `leaf` is the item count at or below which a range's items are
/// fetched and diffed directly instead of split further.
pub fn reconcile(local: &[Item], remote: &mut impl RemoteRange, split: usize, leaf: usize) -> Diff {
    let split = split.max(2);
    let leaf = leaf.max(1);
    let mut diff = Diff::default();
    reconcile_range(local, "", None, remote, split, leaf, &mut diff);
    diff
}

#[allow(clippy::too_many_arguments)]
fn reconcile_range(
    local: &[Item],
    lo: &str,
    hi: Option<&str>,
    remote: &mut impl RemoteRange,
    split: usize,
    leaf: usize,
    out: &mut Diff,
) {
    let local_fp = range_fingerprint(local, lo, hi);
    let remote_fp = remote.fingerprint(lo, hi);

    if local_fp == remote_fp {
        // This whole range matches — prune it, transferring nothing.
        return;
    }

    let local_slice: Vec<&Item> = items_in_range(local, lo, hi).collect();

    if local_slice.len() <= leaf {
        // Small enough (or empty locally): fetch the remote's items for this
        // range and diff directly. An empty local slice against a non-empty
        // remote range is the "we have nothing here" bootstrap — everything the
        // remote has is `only_remote`, which is correct.
        let remote_slice = remote.items(lo, hi);
        diff_slices(&local_slice, &remote_slice, out);
        return;
    }

    // Split the local slice into `split` roughly-equal chunks and recurse into
    // each, using the local keys as boundaries. Ranges that match on both sides
    // prune immediately at the top of the recursion.
    let chunk = local_slice.len().div_ceil(split);
    let mut idx = 0;
    while idx < local_slice.len() {
        // The first chunk starts at the range's own lower bound, NOT at the
        // first local key. Starting at the local key leaves `[lo, first_local)`
        // covered by no child range at all, so a subject the remote has and we
        // lack — one that sorts before everything we hold here — is silently
        // dropped from the diff and never syncs. The gap only opens when a
        // range splits, and only for remote-only subjects at its low end, which
        // is why it surfaced as an intermittent failure rather than an obvious
        // one.
        let chunk_lo = if idx == 0 {
            lo
        } else {
            local_slice[idx].0.as_str()
        };
        let next = (idx + chunk).min(local_slice.len());
        // The chunk's upper bound is the next chunk's first key (open above for
        // the final chunk), so [chunk_lo, chunk_hi) tiles the parent range.
        let chunk_hi = if next < local_slice.len() {
            Some(local_slice[next].0.as_str())
        } else {
            hi
        };
        reconcile_range(local, chunk_lo, chunk_hi, remote, split, leaf, out);
        idx = next;
    }
}

/// Merge-walk two subject-sorted slices, classifying each subject.
fn diff_slices(local: &[&Item], remote: &[Item], out: &mut Diff) {
    let mut i = 0;
    let mut j = 0;
    while i < local.len() && j < remote.len() {
        let (ls, lvv) = local[i];
        let (rs, rvv) = &remote[j];
        match ls.as_str().cmp(rs.as_str()) {
            std::cmp::Ordering::Less => {
                out.only_local.push(ls.clone());
                i += 1;
            }
            std::cmp::Ordering::Greater => {
                out.only_remote.push(rs.clone());
                j += 1;
            }
            std::cmp::Ordering::Equal => {
                if lvv != rvv {
                    out.differ.push(ls.clone());
                }
                i += 1;
                j += 1;
            }
        }
    }
    for (ls, _) in &local[i..] {
        out.only_local.push(ls.clone());
    }
    for (rs, _) in &remote[j..] {
        out.only_remote.push(rs.clone());
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn vv(pairs: &[(&str, i32)]) -> BTreeMap<String, i32> {
        pairs.iter().map(|(p, c)| (p.to_string(), *c)).collect()
    }

    fn item(subject: &str, pairs: &[(&str, i32)]) -> Item {
        (subject.to_string(), vv(pairs))
    }

    /// A remote node backed by an in-memory item set — stands in for the wire.
    /// Counts fingerprint queries so tests can assert the reconcile is
    /// logarithmic (doesn't degrade to scanning every item).
    struct MemRemote {
        items: Vec<Item>,
        fp_calls: usize,
    }

    impl MemRemote {
        fn new(mut items: Vec<Item>) -> Self {
            items.sort_by(|a, b| a.0.cmp(&b.0));
            Self { items, fp_calls: 0 }
        }
    }

    impl RemoteRange for MemRemote {
        fn fingerprint(&mut self, lo: &str, hi: Option<&str>) -> Fingerprint {
            self.fp_calls += 1;
            range_fingerprint(&self.items, lo, hi)
        }
        fn items(&mut self, lo: &str, hi: Option<&str>) -> Vec<Item> {
            items_in_range(&self.items, lo, hi).cloned().collect()
        }
    }

    fn sorted(mut items: Vec<Item>) -> Vec<Item> {
        items.sort_by(|a, b| a.0.cmp(&b.0));
        items
    }

    /// A subject the remote has and we lack, sorting *before* everything we
    /// hold, must still be found.
    ///
    /// This is the low-end gap: when a range splits, its child ranges have to
    /// tile `[lo, hi)` exactly. Anchoring the first child at the first local
    /// key instead leaves `[lo, first_local)` unvisited, so a remote-only
    /// subject down there is dropped from the diff and never syncs. Enough
    /// local items are used here to force a split — under the leaf size the
    /// range is compared directly and the gap cannot open.
    #[test]
    fn a_remote_only_subject_below_every_local_key_is_still_found() {
        let local = sorted(vec![
            item("b", &[("p1", 1)]),
            item("c", &[("p1", 1)]),
            item("d", &[("p1", 1)]),
            item("e", &[("p1", 1)]),
            item("f", &[("p1", 1)]),
        ]);
        let mut remote = MemRemote::new({
            let mut items = local.clone();
            items.push(item("a", &[("p1", 1)]));
            items
        });

        let diff = reconcile(&local, &mut remote, 4, 2);

        assert_eq!(
            diff.only_remote,
            vec!["a".to_string()],
            "the remote-only subject sorting below every local key must be \
             reported, or it never syncs"
        );
        assert!(diff.only_local.is_empty());
        assert!(diff.differ.is_empty());
    }

    /// The same gap, carried down the leftmost spine.
    ///
    /// Only a range whose `lo` sits strictly below its first local key can have
    /// this gap, and after tiling correctly that is exactly the chain of
    /// leftmost children. A set large enough to recurse several levels proves
    /// the unbounded lower bound is propagated the whole way down, not just
    /// handled once at the root.
    #[test]
    fn a_remote_only_subject_below_everything_is_found_through_deep_recursion() {
        let local = sorted(
            (0..24)
                .map(|n| item(&format!("k{n:02}"), &[("p1", 1)]))
                .collect(),
        );
        let mut remote = MemRemote::new({
            let mut items = local.clone();
            items.push(item("a", &[("p1", 1)]));
            items
        });

        let diff = reconcile(&local, &mut remote, 4, 2);

        assert_eq!(diff.only_remote, vec!["a".to_string()]);
        assert!(diff.only_local.is_empty());
        assert!(diff.differ.is_empty());
    }

    #[test]
    fn identical_sets_reconcile_with_a_single_root_comparison() {
        let items = sorted(vec![
            item("a", &[("p1", 1)]),
            item("b", &[("p1", 2)]),
            item("c", &[("p2", 3)]),
        ]);
        let mut remote = MemRemote::new(items.clone());
        let diff = reconcile(&items, &mut remote, 4, 2);
        assert!(
            diff.is_empty(),
            "identical sets must produce no diff: {diff:?}"
        );
        assert_eq!(
            remote.fp_calls, 1,
            "a matching root fingerprint must end the reconcile in one comparison"
        );
    }

    #[test]
    fn detects_a_single_changed_version_vector() {
        let local = sorted(vec![item("a", &[("p1", 1)]), item("b", &[("p1", 2)])]);
        // Same subjects; b's counter advanced remotely.
        let mut remote = MemRemote::new(vec![item("a", &[("p1", 1)]), item("b", &[("p1", 5)])]);
        let diff = reconcile(&local, &mut remote, 4, 2);
        assert_eq!(diff.differ, vec!["b".to_string()]);
        assert!(diff.only_local.is_empty() && diff.only_remote.is_empty());
    }

    #[test]
    fn detects_local_only_and_remote_only_subjects() {
        let local = sorted(vec![item("a", &[("p1", 1)]), item("local", &[("p1", 1)])]);
        let mut remote =
            MemRemote::new(vec![item("a", &[("p1", 1)]), item("remote", &[("p1", 1)])]);
        let diff = reconcile(&local, &mut remote, 4, 2);
        assert_eq!(diff.only_local, vec!["local".to_string()]);
        assert_eq!(diff.only_remote, vec!["remote".to_string()]);
        assert!(diff.differ.is_empty());
    }

    #[test]
    fn empty_local_pulls_every_remote_subject() {
        let local: Vec<Item> = vec![];
        let mut remote = MemRemote::new(vec![item("a", &[("p1", 1)]), item("b", &[("p1", 2)])]);
        let diff = reconcile(&local, &mut remote, 4, 2);
        assert_eq!(diff.only_remote, vec!["a".to_string(), "b".to_string()]);
        assert!(diff.only_local.is_empty() && diff.differ.is_empty());
    }

    #[test]
    fn finds_one_change_in_a_large_set_without_scanning_everything() {
        // 256 subjects; exactly one differs. The whole point of RBSR: the
        // number of fingerprint comparisons must be logarithmic in the set
        // size, NOT linear (which is what the flat hash forces).
        let n = 256usize;
        let mut base: Vec<Item> = (0..n)
            .map(|i| item(&format!("subject-{i:04}"), &[("p1", i as i32)]))
            .collect();
        base = sorted(base);

        let local = base.clone();
        let mut remote_items = base.clone();
        // Advance one subject's counter on the remote side.
        let target = "subject-0123".to_string();
        for (s, v) in remote_items.iter_mut() {
            if *s == target {
                v.insert("p1".to_string(), 9999);
            }
        }
        let mut remote = MemRemote::new(remote_items);

        let diff = reconcile(&local, &mut remote, 4, 4);
        assert_eq!(diff.differ, vec![target]);
        assert!(diff.only_local.is_empty() && diff.only_remote.is_empty());
        // With split=4 over 256 items, a full scan would be ~O(256) fingerprint
        // calls; a logarithmic descent is a few dozen. Assert well under linear.
        assert!(
            remote.fp_calls < 40,
            "expected a logarithmic number of fingerprint comparisons, got {} for {n} items",
            remote.fp_calls
        );
    }

    /// Golden cross-implementation vector for `item_fingerprint`. The TS client
    /// asserts the SAME hex for the SAME item (`canonical-drive-hash.test.ts`).
    /// The reconcile only converges if both sides fingerprint an item
    /// identically, so a drift in the string format or hash fails one of the
    /// two golden tests instead of silently mis-diffing.
    #[test]
    fn item_fingerprint_matches_golden_vector() {
        // "s=p1:1,p2:2" → SHA-256.
        assert_eq!(
            hex::encode(item_fingerprint("s", &vv(&[("p1", 1), ("p2", 2)]))),
            "8b6067440e370aeaf5e85936d9d67477224a664f8b2811a2008309b590edd5d8",
        );
    }

    #[test]
    fn item_fingerprint_is_deterministic_and_order_independent_in_vv() {
        // Two VV maps built in different insertion orders must fingerprint the
        // same (BTreeMap canonicalizes), and different content must differ.
        let a = item_fingerprint("s", &vv(&[("p1", 1), ("p2", 2)]));
        let b = item_fingerprint("s", &vv(&[("p2", 2), ("p1", 1)]));
        assert_eq!(a, b);
        let c = item_fingerprint("s", &vv(&[("p1", 1), ("p2", 3)]));
        assert_ne!(a, c);
    }
}