session model test

1 files changed, 328 insertions, 0 deletions

diff --git a/src/document/sync/model.rs b/src/document/sync/model.rs
new file mode 100644
index 0000000..ad8c6c3
--- /dev/null
+++ b/src/document/sync/model.rs
@@ -0,0 +1,328 @@
+//! A model of two neighbours' sync sessions over a channel that loses, duplicates and
+//! reorders, with writes on both sides: the session as `swarm::sync` drives it (pushes
+//! and messages taken as delivered where level, answers awaited otherwise, reply
+//! timeouts, a STATUS now and then that makes the one behind pull), without the swarm.
+//! Seeded (`testrand::for_seeds`): a failure names its seed.
+
+use std::cmp::Reverse;
+use std::collections::{BTreeMap, BinaryHeap, HashMap};
+use std::time::Duration;
+
+use automerge::{ROOT, ReadDoc, transaction::Transactable};
+use tokio::time::Instant;
+
+use super::{Received, SyncSession, decode};
+use crate::backoff::Backoff;
+use crate::document::testing::try_change;
+use crate::document::{ChangeHash, Document};
+use crate::swarm::throughput::SyncBytes;
+use crate::testrand::{Rng, for_seeds};
+use crate::transport::SyncPayload;
+
+/// One step of model time.
+const TICK: Duration = Duration::from_millis(10);
+/// Writes happen in the first ticks, over a lossy channel; then it turns fair.
+const WRITING: u32 = 300;
+/// How long after that it may take to be quiet (nothing in the channel, nothing to
+/// say, nothing awaited), in ticks: a generous bound for the reply timeouts' backoff.
+const QUIET_WITHIN: u32 = 20_000;
+/// How often each side says its heads (the keepalive STATUS), in ticks.
+const STATUS_EVERY: u32 = 50;
+
+#[derive(Clone)]
+enum Envelope {
+    Message {
+        payload: bytes::Bytes,
+        delivered: bool,
+    },
+    Changes {
+        heads: Vec<ChangeHash>,
+        changes: Vec<Vec<u8>>,
+    },
+    Status {
+        heads: Vec<ChangeHash>,
+    },
+}
+
+struct Side {
+    doc: Document,
+    session: SyncSession,
+    /// Who brought each change first: `true` if the other side (else written here).
+    from_peer: HashMap<ChangeHash, bool>,
+}
+
+struct Channel {
+    queue: BinaryHeap<Reverse<(u32, u64, usize)>>,
+    envelopes: HashMap<u64, Envelope>,
+    seq: u64,
+    loss: f64,
+    dup: f64,
+}
+
+impl Channel {
+    fn send(&mut self, rng: &mut Rng, now: u32, to: usize, envelope: Envelope) {
+        let copies = if rng.chance(self.loss) {
+            0
+        } else if rng.chance(self.dup) {
+            2
+        } else {
+            1
+        };
+        for _ in 0..copies {
+            self.seq += 1;
+            // 1 to 8 ticks: later messages overtake earlier ones.
+            let at = now + rng.between(1, 8) as u32;
+            self.envelopes.insert(self.seq, envelope.clone());
+            self.queue.push(Reverse((at, self.seq, to)));
+        }
+    }
+
+    fn due(&mut self, now: u32) -> Vec<(usize, Envelope)> {
+        let mut due = Vec::new();
+        while let Some(Reverse((at, seq, to))) = self.queue.peek().copied() {
+            if at > now {
+                break;
+            }
+            self.queue.pop();
+            if let Some(envelope) = self.envelopes.remove(&seq) {
+                due.push((to, envelope));
+            }
+        }
+        due
+    }
+}
+
+/// Whether `envelope` carries the whole document saved (Automerge's chunk of type 0, a
+/// document, after the magic bytes and the checksum), not changes.
+fn saved_document(envelope: &Envelope) -> bool {
+    let Envelope::Message { payload, .. } = envelope else {
+        return false;
+    };
+    let message = decode(&SyncPayload(payload.clone())).expect("decodes");
+    message
+        .0
+        .changes
+        .iter()
+        .any(|chunk| chunk.get(8) == Some(&0))
+}
+
+/// What a document holds, by key: what must be equal on both sides at the end, and
+/// across runs that differ only in how changes travelled.
+fn contents(doc: &Document) -> BTreeMap<String, String> {
+    doc.keys(ROOT)
+        .map(|key| {
+            let value = doc
+                .get(ROOT, &key)
+                .ok()
+                .flatten()
+                .map(|(value, _)| value.to_string())
+                .unwrap_or_default();
+            (key, value)
+        })
+        .collect()
+}
+
+/// Runs the model from `seed` (the writes, and apart from them the channel); pushes
+/// allowed below `push_below` bytes (0: never). Returns both documents' contents, once
+/// quiet.
+fn run(seed: u64, push_below: usize) -> (BTreeMap<String, String>, BTreeMap<String, String>) {
+    let mut writes = Rng::new(seed);
+    let mut net = Rng::new(seed ^ 0x5eed);
+    let base = Instant::now();
+    let at = |tick: u32| base + TICK * tick;
+    let bytes = SyncBytes::default();
+    let mut sides = [0u8, 1].map(|which| Side {
+        doc: Document::new().with_actor(automerge::ActorId::from([which + 1; 16])),
+        session: SyncSession::new(Backoff::new(
+            Duration::from_secs(1),
+            Duration::from_secs(30),
+        )),
+        from_peer: HashMap::new(),
+    });
+    let mut channel = Channel {
+        queue: BinaryHeap::new(),
+        envelopes: HashMap::new(),
+        seq: 0,
+        loss: 0.2,
+        dup: 0.1,
+    };
+
+    let mut tick = 0u32;
+    let mut quiet_for = 0u32;
+    loop {
+        tick += 1;
+        assert!(
+            tick < WRITING + QUIET_WITHIN,
+            "seed {seed}: not quiet {QUIET_WITHIN} ticks after the writes"
+        );
+        if tick == WRITING {
+            channel.loss = 0.0;
+            channel.dup = 0.0;
+        }
+        let now = at(tick);
+        let mut busy = false;
+
+        // Deliveries, as `swarm::sync::handle_sync` makes them.
+        for (to, envelope) in channel.due(tick) {
+            let side = &mut sides[to];
+            let message = match envelope {
+                Envelope::Status { heads } => {
+                    // Behind the peer: pull (`relay::pull_if_behind`).
+                    if heads
+                        .iter()
+                        .any(|head| side.doc.get_change_by_hash(head).is_none())
+                    {
+                        side.session.pull_from(heads);
+                    } else if side.session.reply.is_some() && heads == side.doc.get_heads() {
+                        side.session.reply_settled();
+                    }
+                    continue;
+                }
+                Envelope::Changes { heads, changes } => {
+                    (Received::pushed(heads, changes).expect("decodes"), true)
+                }
+                Envelope::Message { payload, delivered } => {
+                    (decode(&SyncPayload(payload)).expect("decodes"), delivered)
+                }
+            };
+            busy = true;
+            let (message, delivered) = message;
+            let before = side.doc.get_heads();
+            side.session
+                .receive(&mut side.doc, message)
+                .expect("applies");
+            side.session.reply_settled();
+            side.session.resend_if_needed(&side.doc, "peer", now);
+            let after = side.doc.get_heads();
+            if delivered {
+                side.session.answered_as_delivered(&side.doc, &after);
+            }
+            if after != before {
+                for change in side.doc.get_changes_meta(&before) {
+                    side.from_peer.entry(change.hash).or_insert(true);
+                }
+                side.session.came_from_peer(&side.doc, &before);
+            }
+        }
+
+        // Writes.
+        if tick < WRITING {
+            for (which, side) in sides.iter_mut().enumerate() {
+                if writes.chance(0.05) {
+                    let key = format!("{which}-{tick}");
+                    let value = writes.next_u64() as i64;
+                    let before = side.doc.get_heads();
+                    try_change(&mut side.doc, |doc| doc.put(ROOT, key, value)).unwrap();
+                    for change in side.doc.get_changes_meta(&before) {
+                        side.from_peer.insert(change.hash, false);
+                    }
+                }
+            }
+        }
+
+        // Sends, as `swarm::sync::deliver_sync_to_peer` makes them.
+        for which in 0..2 {
+            let to = 1 - which;
+            let side = &mut sides[which];
+            side.session.take_overdue_reply(now);
+            let level = side.session.reply.is_none() && side.session.have_all_theirs(&side.doc);
+            let pushed = (level && push_below > 0)
+                .then(|| side.session.push(&side.doc, push_below))
+                .flatten();
+            let envelope =
+                match pushed {
+                    Some((heads, changes)) => Some(Envelope::Changes { heads, changes }),
+                    None => side.session.generate(&side.doc, now, bytes).map(|payload| {
+                        Envelope::Message {
+                            payload: payload.0,
+                            delivered: level,
+                        }
+                    }),
+                };
+            let Some(envelope) = envelope else {
+                continue;
+            };
+            busy = true;
+            if level {
+                let heads = side.doc.get_heads();
+                side.session.assume_delivered(&heads);
+            }
+            // What it carries for the first time (what Automerge chose to send): none of
+            // it may have come from the peer. Unless it is the whole document saved, which
+            // Automerge sends a peer that lacks more than a third of it: the peer's own
+            // changes go along then, by design, and count as sent.
+            let carried = side
+                .session
+                .sending
+                .as_ref()
+                .map_or_else(Vec::new, |sending| sending.sent.clone());
+            let carried = if saved_document(&envelope) {
+                Vec::new()
+            } else {
+                carried
+            };
+            side.session.send_succeeded(side.session.id());
+            // No echo: nothing goes back to the side it came from.
+            for hash in carried {
+                assert_ne!(
+                    side.from_peer.get(&hash),
+                    Some(&true),
+                    "seed {seed}: a change went back to the side it came from"
+                );
+            }
+            channel.send(&mut net, tick, to, envelope);
+        }
+
+        // The keepalive STATUS.
+        if tick % STATUS_EVERY == 0 {
+            for which in 0..2 {
+                let heads = sides[which].doc.get_heads();
+                channel.send(&mut net, tick, 1 - which, Envelope::Status { heads });
+            }
+        }
+
+        let awaiting = sides.iter().any(|side| side.session.reply.is_some());
+        let level = sides[0].doc.get_heads() == sides[1].doc.get_heads();
+        if tick > WRITING && !busy && !awaiting && level {
+            quiet_for += 1;
+            // Quiet through two keepalives: nobody had anything to say.
+            if quiet_for > 2 * STATUS_EVERY {
+                break;
+            }
+        } else {
+            quiet_for = 0;
+        }
+    }
+
+    // Nothing awaited, nothing in the transport's hands, no offer out.
+    for side in &sides {
+        assert!(
+            side.session.reply.is_none(),
+            "seed {seed}: an answer still awaited"
+        );
+        assert!(
+            side.session.sending.is_none(),
+            "seed {seed}: a send not finished"
+        );
+        assert!(
+            !side.session.delta_offered,
+            "seed {seed}: an offer still out"
+        );
+    }
+    let [a, b] = sides;
+    (contents(&a.doc), contents(&b.doc))
+}
+
+/// Both sides end with the same document, without a change going back to where it came
+/// from, with nothing awaited; with pushes as without.
+#[test]
+fn sessions_converge_over_a_lossy_channel() {
+    for_seeds(8, |rng| {
+        let seed = rng.next_u64();
+        let (a, b) = run(seed, 1024);
+        assert_eq!(a, b, "seed {seed}: the sides differ");
+        assert!(!a.is_empty(), "seed {seed}: nothing was written");
+        let (unpushed, _) = run(seed, 0);
+        assert_eq!(a, unpushed, "seed {seed}: pushes made a difference");
+    });
+}