test(zset): split-tick rollback, integrate and step_continue costs; docs and evidence for the review of #355 (improvise-bw4p.5)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze

crates/zset/tests/cost.rs

@@ -101,6 +101,7 @@ fn clones_of_one_step(n: u32, mode: Mode) -> usize {
     let dl = c.delay(&sa);
     let j = c.join_indexed(&sa, &sb, |k, x, y| (k.0, *x, *y));
     let g = c.aggregate(&sa, |_, vals| Some(vals.len() as i64));
+    let _ig = c.integrate(&sa);
     let sinks = (c.output(&d), c.output(&dl), c.output(&j), c.output(&g));
     let rows =
         |lo: u32, hi: u32| -> ZSet<E> { (lo..hi).map(|i| ((Counted::new(i), 0), 1)).collect() };
@@ -141,3 +142,32 @@ fn a_transaction_costs_the_keys_a_step_touches_and_not_the_size_of_the_circuit()
         assert_eq!(small, large, "[cost] the overhead grew with the state");
     }
 }
+
+/// Clones made by one 1-key `step_continue` over an integral of `n` keys.
+fn clones_of_one_continue(n: u32, txn: bool) -> usize {
+    let mut c = Circuit::new();
+    let (input, s) = c.input::<Counted>();
+    let _i = c.integrate(&s);
+    input.push((0..n).map(|k| (Counted::new(k), 1)).collect());
+    c.step();
+    let before = clones();
+    if txn {
+        c.begin();
+    }
+    input.push((n..n + 1).map(|k| (Counted::new(k), 1)).collect());
+    c.step_continue();
+    if txn {
+        c.commit();
+    }
+    clones() - before
+}
+
+/// A limit, pinned so the docs cannot drift from it: a continued piece clones the edge's
+/// whole value once per transaction, which is the size of the state.
+#[test]
+fn a_split_tick_in_a_transaction_costs_the_size_of_the_edge() {
+    let extra = |n| clones_of_one_continue(n, true) - clones_of_one_continue(n, false);
+    let (small, large) = (extra(10), extra(300));
+    assert!(large >= 300, "[split] the edge's value is cloned: {large}");
+    assert!(large > small, "[split] and it grows with the state");
+}

crates/zset/tests/rollback.rs

@@ -277,3 +277,33 @@ fn a_rollback_with_nothing_open_is_refused_loudly() {
 fn a_commit_with_nothing_open_is_refused_loudly() {
     Circuit::new().commit();
 }
+
+/// A tick in two pieces, rolled back: the edge's whole value comes back, not the last piece.
+#[test]
+fn a_rollback_across_a_split_tick_restores_the_whole_value() {
+    let (mut x, mut y) = (rig(), rig());
+    let first: Z<P> = [((1, 1), 2), ((2, 1), 1)].into_iter().collect();
+    x.tick(&first, &Z::new());
+    y.tick(&first, &Z::new());
+    // The tick is already split when the transaction opens, so every edge holds a `whole`.
+    for r in [&mut x, &mut y] {
+        r.a.push([((3, 1), 1)].into_iter().collect());
+        r.c.step_continue();
+    }
+    x.c.begin();
+    for k in [4u8, 5] {
+        x.a.push([((k, 1), 1)].into_iter().collect());
+        x.c.step_continue();
+    }
+    x.c.rollback();
+    assert_eq!((x.view)(), (y.view)(), "[split] state after the rollback");
+    assert_eq!(
+        (x.drain)(),
+        (y.drain)(),
+        "[split] output after the rollback"
+    );
+    let next: Z<P> = [((5, 1), 1)].into_iter().collect();
+    x.tick(&next, &Z::new());
+    y.tick(&next, &Z::new());
+    assert_eq!((x.view)(), (y.view)(), "[split] the tick after");
+}