From 747f320c8dbfc888ed6b1d837151bb8af8ee9cf3 Mon Sep 17 00:00:00 2001 From: James Ross Date: Tue, 20 Oct 2026 11:30:22 -0700 Subject: [PATCH 1/6] Commit path: hash the patch digest once; sort slots and ops once Profiling a 10,000-write engine tick showed most of its commit time in three places, none of them rule work: - The patch digest fed BLAKE3 one field at a time, about eight calls per op. It now gathers the same bytes and hashes them in one call, so BLAKE3 can hash several chunks at once. The digest is unchanged; the golden vectors GV-001 to GV-004 pin it. - Reservation built the tick's in and out slots as BTreeSets, which the patch then sorted and deduplicated again. They are now plain vectors, sorted and deduplicated once by the patch. - The patch canonicalized ops through a BTreeMap. A stable sort followed by a last-wins pass over equal keys gives the same order and the same survivors. A new test compares it with the map it replaced over 400 ops that repeat keys. Carried onto main from f118d7ef on claude/stack-envelope, as an isolated candidate. The conflict was only lazy-blockers context; main's eager blocked_by is kept unchanged. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/engine_impl.rs | 47 ++++++------ crates/warp-core/src/tick_patch.rs | 110 +++++++++++++++++++++++----- 2 files changed, 115 insertions(+), 42 deletions(-) diff --git a/crates/warp-core/src/engine_impl.rs b/crates/warp-core/src/engine_impl.rs index 53717d85..d739be2f 100644 --- a/crates/warp-core/src/engine_impl.rs +++ b/crates/warp-core/src/engine_impl.rs @@ -538,8 +538,9 @@ pub struct Engine { struct ReserveOutcome { receipt: TickReceipt, reserved: Vec, - in_slots: std::collections::BTreeSet, - out_slots: std::collections::BTreeSet, + /// Unsorted, with repeats: the patch sorts and dedups them once. + in_slots: Vec, + out_slots: Vec, } enum SavedField { @@ -2163,8 +2164,8 @@ impl Engine { policy_id, rule_pack_id, TickCommitStatus::Committed, - in_slots.into_iter().collect(), - out_slots.into_iter().collect(), + in_slots, + out_slots, ops, ); let patch_digest = patch.digest(); @@ -2289,8 +2290,8 @@ impl Engine { ) -> Result { // Reserve phase: enforce independence against active frontier. let mut receipt_entries: Vec = Vec::with_capacity(drained.len()); - let mut in_slots: std::collections::BTreeSet = std::collections::BTreeSet::new(); - let mut out_slots: std::collections::BTreeSet = std::collections::BTreeSet::new(); + let mut in_slots: Vec = Vec::new(); + let mut out_slots: Vec = Vec::new(); let mut blocked_by: Vec> = Vec::with_capacity(drained.len()); let mut reserved: Vec = Vec::new(); let mut reserved_entry_indices: Vec = Vec::new(); @@ -3432,52 +3433,52 @@ pub fn scope_hash(rule_id: &Hash, scope: &NodeKey) -> Hash { /// for call-site compatibility but is no longer used (the `warp_id` is embedded /// in the footprint keys). fn extend_slots_from_footprint( - in_slots: &mut std::collections::BTreeSet, - out_slots: &mut std::collections::BTreeSet, + in_slots: &mut Vec, + out_slots: &mut Vec, _warp_id: &WarpId, // Kept for API compat; warp_id is now in the footprint keys fp: &crate::footprint::Footprint, ) { // Nodes (warp-scoped NodeKey) for key in fp.n_read.iter() { - in_slots.insert(SlotId::Node(*key)); + in_slots.push(SlotId::Node(*key)); } for key in fp.n_write.iter() { - in_slots.insert(SlotId::Node(*key)); - out_slots.insert(SlotId::Node(*key)); + in_slots.push(SlotId::Node(*key)); + out_slots.push(SlotId::Node(*key)); } // Edges (warp-scoped EdgeKey) for key in fp.e_read.iter() { - in_slots.insert(SlotId::Edge(*key)); + in_slots.push(SlotId::Edge(*key)); } for key in fp.e_write.iter() { - in_slots.insert(SlotId::Edge(*key)); - out_slots.insert(SlotId::Edge(*key)); + in_slots.push(SlotId::Edge(*key)); + out_slots.push(SlotId::Edge(*key)); } for key in fp.r_read.iter() { - in_slots.insert(SlotId::OutgoingEdges(*key)); + in_slots.push(SlotId::OutgoingEdges(*key)); } for key in fp.r_write.iter() { - in_slots.insert(SlotId::OutgoingEdges(*key)); - out_slots.insert(SlotId::OutgoingEdges(*key)); + in_slots.push(SlotId::OutgoingEdges(*key)); + out_slots.push(SlotId::OutgoingEdges(*key)); } // Attachments (already warp-scoped via AttachmentKey) for key in fp.a_read.iter() { - in_slots.insert(SlotId::Attachment(*key)); + in_slots.push(SlotId::Attachment(*key)); } for key in fp.a_write.iter() { - in_slots.insert(SlotId::Attachment(*key)); - out_slots.insert(SlotId::Attachment(*key)); + in_slots.push(SlotId::Attachment(*key)); + out_slots.push(SlotId::Attachment(*key)); } // Ports (warp-scoped: (WarpId, PortKey)) for port_key in fp.b_in.keys() { - in_slots.insert(SlotId::Port(*port_key)); + in_slots.push(SlotId::Port(*port_key)); } for port_key in fp.b_out.keys() { - in_slots.insert(SlotId::Port(*port_key)); - out_slots.insert(SlotId::Port(*port_key)); + in_slots.push(SlotId::Port(*port_key)); + out_slots.push(SlotId::Port(*port_key)); } } diff --git a/crates/warp-core/src/tick_patch.rs b/crates/warp-core/src/tick_patch.rs index d0f09e0c..5ad8da6c 100644 --- a/crates/warp-core/src/tick_patch.rs +++ b/crates/warp-core/src/tick_patch.rs @@ -12,7 +12,6 @@ //! Value versioning is recovered by interpretation along a worldline payload //! `P = (μ0, …, μn-1)` via `ValueVersionId := (slot_id, tick_index)`. -use blake3::Hasher; use thiserror::Error; use crate::attachment::{AtomPayload, AttachmentKey, AttachmentOwner, AttachmentValue}; @@ -404,14 +403,22 @@ impl WarpTickPatchV1 { out_slots.sort(); out_slots.dedup(); let ops = { - let mut op_map: std::collections::BTreeMap = - std::collections::BTreeMap::new(); - for op in ops { - // Last-wins: if multiple ops share the same canonical sort key, keep the - // last op provided by the caller and drop earlier duplicates. - op_map.insert(op.sort_key(), op); - } - op_map.into_values().collect::>() + let mut keyed: Vec<(WarpOpKey, WarpOp)> = + ops.into_iter().map(|op| (op.sort_key(), op)).collect(); + // A stable sort keeps ops that share a key in the caller's order. + keyed.sort_by_key(|(key, _)| *key); + let mut canonical: Vec = Vec::with_capacity(keyed.len()); + let mut previous: Option = None; + for (key, op) in keyed { + match canonical.last_mut() { + // Last-wins: if multiple ops share the same canonical sort key, keep the + // last op provided by the caller and drop earlier duplicates. + Some(last) if previous == Some(key) => *last = op, + _ => canonical.push(op), + } + previous = Some(key); + } + canonical }; let digest = compute_patch_digest_v2( policy_id, @@ -1047,7 +1054,12 @@ fn compute_patch_digest_v2( out_slots: &[SlotId], ops: &[WarpOp], ) -> ContentHash { - let mut h = Hasher::new(); + // The encoding is gathered first and hashed in one call. Fed a few bytes + // at a time, BLAKE3 compresses one block per call; given the whole + // input, it can hash several chunks at once. The digest is the same. + let mut h = DigestBytes(Vec::with_capacity( + 64 + 72 * (in_slots.len() + out_slots.len()) + 160 * ops.len(), + )); h.update(domain::PATCH_DIGEST_V1); // Patch format version. h.update(&2u16.to_le_bytes()); @@ -1058,10 +1070,19 @@ fn compute_patch_digest_v2( encode_slots(&mut h, in_slots); encode_slots(&mut h, out_slots); encode_ops(&mut h, ops); - h.finalize().into() + blake3::hash(&h.0).into() +} + +/// The patch digest's input, gathered before it is hashed. +struct DigestBytes(Vec); + +impl DigestBytes { + fn update(&mut self, bytes: &[u8]) { + self.0.extend_from_slice(bytes); + } } -fn encode_slots(h: &mut Hasher, slots: &[SlotId]) { +fn encode_slots(h: &mut DigestBytes, slots: &[SlotId]) { h.update(&(slots.len() as u64).to_le_bytes()); for slot in slots { match slot { @@ -1100,7 +1121,7 @@ fn encode_slots(h: &mut Hasher, slots: &[SlotId]) { /// stable, versioned encoding for hashing (`patch_digest`). They are /// intentionally distinct from `WarpOp::sort_key`’s `kind` values, which exist /// only to define deterministic replay ordering. -fn encode_ops(h: &mut Hasher, ops: &[WarpOp]) { +fn encode_ops(h: &mut DigestBytes, ops: &[WarpOp]) { h.update(&(ops.len() as u64).to_le_bytes()); for op in ops { match op { @@ -1164,7 +1185,7 @@ fn encode_ops(h: &mut Hasher, ops: &[WarpOp]) { } } -fn encode_portal_init(h: &mut Hasher, init: &PortalInit) { +fn encode_portal_init(h: &mut DigestBytes, init: &PortalInit) { match init { PortalInit::RequireExisting => { h.update(&[0u8]); @@ -1176,7 +1197,7 @@ fn encode_portal_init(h: &mut Hasher, init: &PortalInit) { } } -fn encode_attachment_key_opt(h: &mut Hasher, key: Option<&AttachmentKey>) { +fn encode_attachment_key_opt(h: &mut DigestBytes, key: Option<&AttachmentKey>) { match key { None => { h.update(&[0u8]); @@ -1188,7 +1209,7 @@ fn encode_attachment_key_opt(h: &mut Hasher, key: Option<&AttachmentKey>) { } } -fn encode_attachment_key(h: &mut Hasher, key: &AttachmentKey) { +fn encode_attachment_key(h: &mut DigestBytes, key: &AttachmentKey) { let (owner_tag, plane_tag) = key.tag(); h.update(&[owner_tag]); h.update(&[plane_tag]); @@ -1204,7 +1225,7 @@ fn encode_attachment_key(h: &mut Hasher, key: &AttachmentKey) { } } -fn encode_attachment_value_opt(h: &mut Hasher, value: Option<&AttachmentValue>) { +fn encode_attachment_value_opt(h: &mut DigestBytes, value: Option<&AttachmentValue>) { match value { None => { h.update(&[0u8]); @@ -1216,7 +1237,7 @@ fn encode_attachment_value_opt(h: &mut Hasher, value: Option<&AttachmentValue>) } } -fn encode_attachment_value(h: &mut Hasher, value: &AttachmentValue) { +fn encode_attachment_value(h: &mut DigestBytes, value: &AttachmentValue) { match value { AttachmentValue::Atom(atom) => { h.update(&[1u8]); @@ -1229,7 +1250,7 @@ fn encode_attachment_value(h: &mut Hasher, value: &AttachmentValue) { } } -fn encode_atom_payload(h: &mut Hasher, atom: &AtomPayload) { +fn encode_atom_payload(h: &mut DigestBytes, atom: &AtomPayload) { h.update(&(atom.type_id).0); h.update(&(atom.bytes.len() as u64).to_le_bytes()); h.update(&atom.bytes); @@ -2607,6 +2628,57 @@ mod tests { }, )); } + #[test] + fn op_canonicalization_matches_the_last_wins_map_it_replaced() { + let (_, first, second) = reversible_fixture(); + let nodes: Vec = (0..5) + .map(|i| NodeKey { + warp_id: first.warp_id, + local_id: make_node_id(&format!("canonical-{i}")), + }) + .chain([first, second]) + .collect(); + let values: [&'static [u8]; 3] = [b"a", b"b", b"c"]; + // A fixed linear congruential sequence, so the ops repeat keys in a + // varied but reproducible order. + let mut x: u64 = 0x9e37_79b9_7f4a_7c15; + let mut ops = Vec::new(); + for _ in 0..400 { + x = x + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + let node = nodes[((x >> 33) % nodes.len() as u64) as usize]; + let value = values[((x >> 20) % values.len() as u64) as usize]; + ops.push(match (x >> 40) % 3 { + 0 => set_attachment(node, Some(reversible_atom(value))), + 1 => set_attachment(node, None), + _ => WarpOp::UpsertNode { + node, + record: NodeRecord { + ty: make_type_id(std::str::from_utf8(value).unwrap_or("x")), + }, + }, + }); + } + let mut reference: std::collections::BTreeMap = + std::collections::BTreeMap::new(); + for op in ops.clone() { + reference.insert(op.sort_key(), op); + } + let reference: Vec = reference.into_values().collect(); + + let patch = WarpTickPatchV1::new( + 0, + [0u8; 32], + TickCommitStatus::Committed, + Vec::new(), + Vec::new(), + ops, + ); + + assert!(reference.len() < 400, "the fixture must repeat keys"); + assert_eq!(patch.ops(), reference.as_slice()); + } } #[cfg(test)] From 1f347bfe7617b447a63d6246484829bff88b6f07 Mon Sep 17 00:00:00 2001 From: James Ross Date: Wed, 21 Oct 2026 07:15:52 -0700 Subject: [PATCH 2/6] Pin patch bytes and digests to the map-based implementation Add baseline-derived compatibility vectors for WarpTickPatchV1 (crates/warp-core/src/tick_patch/compat_vectors.rs). Four vectors hand `WarpTickPatchV1::new` unsorted inputs with repeats: - every-kind: every WarpOp variant and each sub-variant the encoding distinguishes (OpenPortal Empty/RequireExisting, UpsertWarpInstance with and without parent, SetAttachment clear/atom/empty atom/Descend), every SlotId kind, all four attachment owner/plane pairs, a 1,100-byte payload (over one BLAKE3 chunk on its own), Committed status. - last-wins: a repeated key for every op variant, with differing values where the op carries one; same edge id from another source and different kinds on one target, which must both survive; Aborted status. - slots-only: repeated slots of every kind, no ops, policy u32::MAX. - generated: 400 ops and 300 slots from a fixed LCG over a small key space. Each vector pins the patch digest, the digest-input length, the full digest-input bytes in hex (all but `generated`), and the canonical order of in_slots, out_slots and ops as indices into the inputs (ops: the last equal input op, i.e. the last-wins survivor). How the expected values were captured. A detached scratch worktree at main 0faaf58a (/Users/j/git/echo-claude-cpvec, created and removed under the .git/ lock) received the same fixture, appended as a test-only module, and a capture test that printed each value: 1. Unmodified main: printed digests and canonical orders. 2. Main with only `use blake3::Hasher` replaced by a scratch shim that forwards every update to blake3::Hasher unchanged and records the bytes: printed the input bytes, their length and BLAKE3 of the recorded bytes. Digests and orders were identical to run 1, and BLAKE3 of the recorded bytes equalled each digest. The pinned literals were generated from run 2's output file by script, not typed. Logs: scratchpad runners/logs/cpvec-capture-pure.log and cpvec-capture-shim.log. The tests on this branch: the vectors match bytes, length, digest and order; `validate_digest` accepts each patch; a coverage test fails to compile on a new op, slot, owner or plane variant and fails if a covered kind leaves the vectors; and the replaced canonicalizers (BTreeMap last-wins for ops, BTreeSet for slots) are kept as an oracle over every vector, beside the existing randomized `op_canonicalization_matches_the_last_wins_map_it_replaced`. To expose the bytes, `compute_patch_digest_v2` now hashes the output of `patch_digest_input_v2`, which holds the unchanged encoding. Consumers of the unsorted slot vectors. Reservation's `in_slots` and `out_slots` (now Vec with repeats, formerly BTreeSet) are built only in `reserve_for_receipt`, carried only in `ReserveOutcome`, destructured only in `commit_with_receipt`, and passed, unread, straight into `WarpTickPatchV1::new`, which sorts and dedups them before anything else. Nothing between (rewrites digest, rule execution, materialization, touched_diff) reads them, and an error or panic drops them. Every other reader (application trie advance, worldline slicing, provenance codec, settlement, observation, coordinator, trusted runtime host, echo_operation) reads `patch.in_slots()`/`out_slots()`, the canonical lists. So no consumer can observe a repeat. The new engine witness `overlapping_footprints_commit_the_same_sorted_slot_sets` commits rewrites whose footprints overlap across rewrites (shared reads) and within one footprint (read and written), and checks the patch carries exactly the sorted, repeat-free union the BTreeSet collection held. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/engine_impl.rs | 3 + .../src/engine_impl/overlap_slot_tests.rs | 160 ++++ crates/warp-core/src/tick_patch.rs | 25 +- .../src/tick_patch/compat_vectors.rs | 830 ++++++++++++++++++ 4 files changed, 1017 insertions(+), 1 deletion(-) create mode 100644 crates/warp-core/src/engine_impl/overlap_slot_tests.rs create mode 100644 crates/warp-core/src/tick_patch/compat_vectors.rs diff --git a/crates/warp-core/src/engine_impl.rs b/crates/warp-core/src/engine_impl.rs index d739be2f..f88e1bf8 100644 --- a/crates/warp-core/src/engine_impl.rs +++ b/crates/warp-core/src/engine_impl.rs @@ -52,6 +52,9 @@ use crate::warp_state::{WarpInstance, WarpState}; use crate::worldline_state::WorldlineState; use crate::ContractInverseHandler; +#[cfg(test)] +#[path = "engine_impl/overlap_slot_tests.rs"] +mod overlap_slot_tests; #[cfg(test)] #[path = "engine_impl/range_tests.rs"] mod range_tests; diff --git a/crates/warp-core/src/engine_impl/overlap_slot_tests.rs b/crates/warp-core/src/engine_impl/overlap_slot_tests.rs new file mode 100644 index 00000000..df828881 --- /dev/null +++ b/crates/warp-core/src/engine_impl/overlap_slot_tests.rs @@ -0,0 +1,160 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +#![allow(clippy::expect_used)] // Test assertions may abort the witness. +//! Reservation collects a tick's slots as vectors with repeats. This witness +//! commits rewrites whose footprints overlap, both across rewrites (shared +//! reads) and within one footprint (a slot that is read and written), and +//! checks that the committed patch carries exactly the sorted, repeat-free +//! union that the former `BTreeSet` collection produced. +use std::collections::BTreeSet; + +use super::*; +use crate::{AtomPayload, EdgeId, EdgeKey, Footprint, OutgoingEdgeRange, PatternGraph, TickDelta}; + +const RULE: &str = "test/overlap-slots"; +const OWN: [&str; 3] = [ + "test/overlap-own-0", + "test/overlap-own-1", + "test/overlap-own-2", +]; + +fn shared_node() -> NodeId { + make_node_id("test/overlap-shared") +} + +fn shared_edge() -> EdgeId { + make_edge_id("test/overlap-shared-edge") +} + +fn shared_range(warp_id: WarpId) -> OutgoingEdgeRange { + OutgoingEdgeRange { + source: NodeKey { + warp_id, + local_id: shared_node(), + }, + edge_type: make_type_id("test/overlap-link"), + } +} + +fn port_of(scope: &NodeId) -> u64 { + u64::from(scope.0[0]) +} + +fn footprint(view: GraphView<'_>, scope: &NodeId) -> Footprint { + let warp_id = view.warp_id(); + let shared = NodeKey { + warp_id, + local_id: shared_node(), + }; + let own = NodeKey { + warp_id, + local_id: *scope, + }; + let mut footprint = Footprint::default(); + // Read by every rewrite. + footprint.n_read.insert(shared); + footprint.a_read.insert(AttachmentKey::node_alpha(shared)); + footprint.e_read.insert(EdgeKey { + warp_id, + local_id: shared_edge(), + }); + footprint.r_read.insert(shared_range(warp_id)); + // Read and written by this rewrite alone. + footprint.n_read.insert(own); + footprint.n_write.insert(own); + footprint.a_read.insert(AttachmentKey::node_alpha(own)); + footprint.a_write.insert(AttachmentKey::node_alpha(own)); + footprint.b_in.insert(warp_id, port_of(scope)); + footprint.b_out.insert(warp_id, port_of(scope)); + footprint +} + +fn executor(view: GraphView<'_>, scope: &NodeId, delta: &mut TickDelta) { + delta.push(WarpOp::SetAttachment { + key: AttachmentKey::node_alpha(NodeKey { + warp_id: view.warp_id(), + local_id: *scope, + }), + value: Some(AttachmentValue::Atom(AtomPayload::new( + make_type_id("test/overlap-value"), + bytes::Bytes::from_static(b"written"), + ))), + }); +} + +#[test] +fn overlapping_footprints_commit_the_same_sorted_slot_sets() { + let mut store = GraphStore::default(); + let ty = make_type_id("test/overlap-node"); + store.insert_node(shared_node(), NodeRecord { ty }); + let own: Vec = OWN.iter().map(|label| make_node_id(label)).collect(); + for node in &own { + store.insert_node(*node, NodeRecord { ty }); + } + let ports: BTreeSet = own.iter().map(port_of).collect(); + assert_eq!(ports.len(), own.len(), "each rewrite needs its own port"); + let mut engine = Engine::new(store, shared_node()); + engine + .register_rule(RewriteRule { + id: [7; 32], + name: RULE, + left: PatternGraph { nodes: vec![] }, + matcher: |_, _| true, + executor: RuleExecutor::legacy(executor), + compute_footprint: footprint, + factor_mask: 0, + conflict_policy: ConflictPolicy::Abort, + join_fn: None, + }) + .expect("rule registers"); + let tx = engine.begin(); + for node in &own { + assert!(matches!( + engine.apply(tx, RULE, node), + Ok(ApplyResult::Applied) + )); + } + let (_, receipt, patch) = engine.commit_with_receipt(tx).expect("tick commits"); + assert!(receipt + .entries() + .iter() + .all(|entry| entry.disposition == TickReceiptDisposition::Applied)); + + // What the former `BTreeSet` collection held: the union, once each. + let warp_id = engine.current_root.warp_id; + let shared = NodeKey { + warp_id, + local_id: shared_node(), + }; + let mut expected_in: BTreeSet = BTreeSet::from([ + SlotId::Node(shared), + SlotId::Attachment(AttachmentKey::node_alpha(shared)), + SlotId::Edge(EdgeKey { + warp_id, + local_id: shared_edge(), + }), + SlotId::OutgoingEdges(shared_range(warp_id)), + ]); + let mut expected_out: BTreeSet = BTreeSet::new(); + for node in &own { + let key = NodeKey { + warp_id, + local_id: *node, + }; + for slot in [ + SlotId::Node(key), + SlotId::Attachment(AttachmentKey::node_alpha(key)), + SlotId::Port((warp_id, port_of(node))), + ] { + expected_in.insert(slot); + expected_out.insert(slot); + } + } + let expected_in: Vec = expected_in.into_iter().collect(); + let expected_out: Vec = expected_out.into_iter().collect(); + assert_eq!(patch.in_slots(), expected_in.as_slice()); + assert_eq!(patch.out_slots(), expected_out.as_slice()); + assert!(patch.in_slots().windows(2).all(|w| w[0] < w[1])); + assert!(patch.out_slots().windows(2).all(|w| w[0] < w[1])); + assert_eq!(patch.ops().len(), own.len()); +} diff --git a/crates/warp-core/src/tick_patch.rs b/crates/warp-core/src/tick_patch.rs index 5ad8da6c..1678dfef 100644 --- a/crates/warp-core/src/tick_patch.rs +++ b/crates/warp-core/src/tick_patch.rs @@ -1057,6 +1057,26 @@ fn compute_patch_digest_v2( // The encoding is gathered first and hashed in one call. Fed a few bytes // at a time, BLAKE3 compresses one block per call; given the whole // input, it can hash several chunks at once. The digest is the same. + let input = patch_digest_input_v2( + policy_id, + rule_pack_id, + commit_status, + in_slots, + out_slots, + ops, + ); + blake3::hash(&input).into() +} + +/// The exact bytes `patch_digest` hashes: the patch's canonical encoding. +fn patch_digest_input_v2( + policy_id: u32, + rule_pack_id: &ContentHash, + commit_status: TickCommitStatus, + in_slots: &[SlotId], + out_slots: &[SlotId], + ops: &[WarpOp], +) -> Vec { let mut h = DigestBytes(Vec::with_capacity( 64 + 72 * (in_slots.len() + out_slots.len()) + 160 * ops.len(), )); @@ -1070,7 +1090,7 @@ fn compute_patch_digest_v2( encode_slots(&mut h, in_slots); encode_slots(&mut h, out_slots); encode_ops(&mut h, ops); - blake3::hash(&h.0).into() + h.0 } /// The patch digest's input, gathered before it is hashed. @@ -3180,3 +3200,6 @@ mod touched_diff_tests { } } } + +#[cfg(test)] +mod compat_vectors; diff --git a/crates/warp-core/src/tick_patch/compat_vectors.rs b/crates/warp-core/src/tick_patch/compat_vectors.rs new file mode 100644 index 00000000..3ddbea98 --- /dev/null +++ b/crates/warp-core/src/tick_patch/compat_vectors.rs @@ -0,0 +1,830 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +//! Compatibility vectors for the patch digest and patch canonicalization. +//! +//! Each vector hands [`WarpTickPatchV1::new`] unsorted inputs with repeats. +//! The pinned canonical order, digest-input bytes and digests were produced +//! by the implementation that canonicalized ops through a `BTreeMap`, +//! collected slots into `BTreeSet`s and fed `blake3::Hasher` field by field +//! (main at 0faaf58a). The vectors cover every op variant, every slot kind, +//! every attachment owner and plane, non-empty and empty payloads, a payload +//! longer than one BLAKE3 chunk, and repeated keys of every op variant. +use super::*; + +use bytes::Bytes; + +use crate::attachment::AttachmentPlane; +use crate::footprint::OutgoingEdgeRange; +use crate::ident::{make_edge_id, make_node_id, make_type_id, make_warp_id, TypeId}; + +// ---- fixture: identical in the committed test and the baseline capture ---- + +/// One patch's inputs, as a caller hands them to [`WarpTickPatchV1::new`]: +/// unsorted, with repeats. +struct Vector { + name: &'static str, + policy_id: u32, + rule_pack_id: ContentHash, + status: TickCommitStatus, + in_slots: Vec, + out_slots: Vec, + ops: Vec, +} + +fn root() -> WarpId { + make_warp_id("cv/root") +} + +fn warp(i: u32) -> WarpId { + make_warp_id(&format!("cv/warp/{i}")) +} + +fn node(i: u32) -> NodeKey { + NodeKey { + warp_id: root(), + local_id: make_node_id(&format!("cv/node/{i}")), + } +} + +fn edge(i: u32) -> EdgeKey { + EdgeKey { + warp_id: root(), + local_id: make_edge_id(&format!("cv/edge/{i}")), + } +} + +fn ty(label: &str) -> TypeId { + make_type_id(&format!("cv/type/{label}")) +} + +fn node_key(i: u32, plane: AttachmentPlane) -> AttachmentKey { + AttachmentKey { + owner: AttachmentOwner::Node(node(i)), + plane, + } +} + +fn edge_key(i: u32, plane: AttachmentPlane) -> AttachmentKey { + AttachmentKey { + owner: AttachmentOwner::Edge(edge(i)), + plane, + } +} + +/// An atom whose payload is `len` bytes of a fixed pattern. +fn atom(label: &str, len: usize, seed: u8) -> AttachmentValue { + let bytes: Vec = (0..=u8::MAX) + .cycle() + .take(len) + .map(|i| i.wrapping_mul(31).wrapping_add(seed)) + .collect(); + AttachmentValue::Atom(AtomPayload::new(ty(label), Bytes::from(bytes))) +} + +fn range(i: u32, label: &str) -> SlotId { + SlotId::OutgoingEdges(OutgoingEdgeRange { + source: node(i), + edge_type: ty(label), + }) +} + +fn edge_record(id: u32, from: u32, to: u32, label: &str) -> EdgeRecord { + EdgeRecord { + id: edge(id).local_id, + from: node(from).local_id, + to: node(to).local_id, + ty: ty(label), + } +} + +fn set(key: AttachmentKey, value: Option) -> WarpOp { + WarpOp::SetAttachment { key, value } +} + +fn upsert_node(i: u32, label: &str) -> WarpOp { + WarpOp::UpsertNode { + node: node(i), + record: NodeRecord { ty: ty(label) }, + } +} + +fn upsert_edge(id: u32, from: u32, to: u32, label: &str) -> WarpOp { + WarpOp::UpsertEdge { + warp_id: root(), + record: edge_record(id, from, to, label), + } +} + +fn delete_edge(from: u32, id: u32) -> WarpOp { + WarpOp::DeleteEdge { + warp_id: root(), + from: node(from).local_id, + edge_id: edge(id).local_id, + } +} + +fn open_portal(key: AttachmentKey, child: u32, child_root: &str, init: PortalInit) -> WarpOp { + WarpOp::OpenPortal { + key, + child_warp: warp(child), + child_root: make_node_id(&format!("cv/child-root/{child_root}")), + init, + } +} + +fn upsert_instance(i: u32, root_node: &str, parent: Option) -> WarpOp { + WarpOp::UpsertWarpInstance { + instance: WarpInstance { + warp_id: warp(i), + root_node: make_node_id(&format!("cv/child-root/{root_node}")), + parent, + }, + } +} + +fn every_kind() -> Vector { + use AttachmentPlane::{Alpha, Beta}; + let in_slots = vec![ + range(0, "link"), + SlotId::Port((root(), 7)), + SlotId::Attachment(edge_key(0, Beta)), + SlotId::Node(node(1)), + SlotId::Attachment(node_key(0, Alpha)), + SlotId::Edge(edge(1)), + SlotId::Attachment(node_key(1, Beta)), + SlotId::Attachment(edge_key(1, Alpha)), + SlotId::Port((warp(1), u64::MAX)), + SlotId::Node(node(0)), + SlotId::Edge(edge(0)), + SlotId::Node(node(1)), + SlotId::Port((root(), 7)), + range(1, "other"), + range(0, "link"), + ]; + let out_slots = vec![ + SlotId::Attachment(edge_key(0, Beta)), + SlotId::Node(node(0)), + range(0, "link"), + SlotId::Port((root(), 7)), + SlotId::Edge(edge(1)), + SlotId::Attachment(node_key(0, Alpha)), + SlotId::Node(node(0)), + ]; + let ops = vec![ + // A payload over one BLAKE3 chunk (1 KiB) on its own. + set(edge_key(0, Beta), Some(atom("blob", 1100, 3))), + upsert_edge(1, 0, 1, "link"), + WarpOp::DeleteNode { node: node(2) }, + open_portal( + node_key(0, Beta), + 1, + "a", + PortalInit::Empty { + root_record: NodeRecord { ty: ty("child") }, + }, + ), + set(node_key(1, Alpha), Some(AttachmentValue::Descend(warp(1)))), + upsert_instance(2, "b", Some(node_key(3, Alpha))), + delete_edge(0, 2), + upsert_node(1, "body"), + open_portal(edge_key(1, Alpha), 3, "c", PortalInit::RequireExisting), + set(node_key(0, Alpha), None), + WarpOp::DeleteWarpInstance { warp_id: warp(4) }, + upsert_instance(5, "d", None), + set(edge_key(1, Beta), Some(atom("small", 5, 9))), + set(node_key(2, Alpha), Some(atom("empty", 0, 0))), + ]; + Vector { + name: "every-kind", + policy_id: 0x0102_0304, + rule_pack_id: *blake3::hash(b"cv/rule-pack/every-kind").as_bytes(), + status: TickCommitStatus::Committed, + in_slots, + out_slots, + ops, + } +} + +fn last_wins() -> Vector { + use AttachmentPlane::{Alpha, Beta}; + let ops = vec![ + upsert_node(1, "first"), + set(node_key(0, Alpha), Some(atom("x", 3, 1))), + upsert_edge(1, 0, 1, "link"), + open_portal( + node_key(0, Beta), + 1, + "a", + PortalInit::Empty { + root_record: NodeRecord { ty: ty("child") }, + }, + ), + upsert_instance(2, "a", None), + WarpOp::DeleteNode { node: node(2) }, + delete_edge(0, 2), + WarpOp::DeleteWarpInstance { warp_id: warp(4) }, + // Same keys again, with different values where the op carries one. + upsert_node(1, "second"), + set(node_key(0, Alpha), None), + upsert_edge(1, 0, 2, "link-2"), + open_portal(node_key(0, Beta), 3, "b", PortalInit::RequireExisting), + upsert_instance(2, "b", Some(edge_key(0, Beta))), + WarpOp::DeleteNode { node: node(2) }, + delete_edge(0, 2), + WarpOp::DeleteWarpInstance { warp_id: warp(4) }, + // Same edge id from another source: a different key, so both stay. + upsert_edge(1, 3, 2, "link-3"), + // Different kinds on one target: different keys, so all stay. + WarpOp::DeleteNode { node: node(1) }, + set(node_key(0, Beta), Some(atom("beside-portal", 2, 4))), + // Third writes; the last one of each key wins. + set(node_key(0, Alpha), Some(atom("y", 40, 2))), + upsert_node(1, "third"), + upsert_edge(1, 0, 3, "link-4"), + set(edge_key(0, Beta), Some(atom("edge", 9, 5))), + set(edge_key(0, Beta), Some(AttachmentValue::Descend(warp(6)))), + ]; + let in_slots = vec![ + SlotId::Node(node(1)), + SlotId::Attachment(node_key(0, Alpha)), + SlotId::Node(node(1)), + range(0, "link"), + SlotId::Attachment(node_key(0, Alpha)), + SlotId::Edge(edge(1)), + range(0, "link"), + SlotId::Node(node(1)), + ]; + let out_slots = vec![ + SlotId::Attachment(node_key(0, Alpha)), + SlotId::Node(node(1)), + SlotId::Attachment(node_key(0, Alpha)), + SlotId::Node(node(1)), + ]; + Vector { + name: "last-wins", + policy_id: 7, + rule_pack_id: *blake3::hash(b"cv/rule-pack/last-wins").as_bytes(), + status: TickCommitStatus::Aborted, + in_slots, + out_slots, + ops, + } +} + +fn slots_only() -> Vector { + use AttachmentPlane::{Alpha, Beta}; + let pool = [ + SlotId::Port((warp(2), 0)), + SlotId::Attachment(edge_key(3, Alpha)), + range(2, "z"), + SlotId::Edge(edge(4)), + SlotId::Attachment(node_key(4, Beta)), + SlotId::Node(node(5)), + range(2, "a"), + SlotId::Port((warp(2), 1)), + ]; + let in_slots = (0..24).map(|i| pool[(i * 5 + 3) % pool.len()]).collect(); + let out_slots = (0..9).map(|i| pool[(i * 3 + 1) % pool.len()]).collect(); + Vector { + name: "slots-only", + policy_id: u32::MAX, + rule_pack_id: [0xa5; 32], + status: TickCommitStatus::Committed, + in_slots, + out_slots, + ops: Vec::new(), + } +} + +fn small(r: u64) -> u32 { + u32::try_from(r).unwrap_or(0) +} + +/// 400 ops over a small key space, so keys repeat in a varied order, with +/// 300 slots drawn from a small pool. +fn generated() -> Vector { + use AttachmentPlane::{Alpha, Beta}; + let mut x: u64 = 0x2545_f491_4f6c_dd1d; + let mut next = move || { + x = x + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + x >> 24 + }; + let mut ops = Vec::new(); + for _ in 0..400 { + let r = next(); + let a = small(r % 6); + let b = small((r >> 8) % 4); + let plane = if (r >> 12) % 2 == 0 { Alpha } else { Beta }; + let len = usize::try_from((r >> 16) % 70).unwrap_or(0); + let seed = (r >> 24).to_le_bytes()[0]; + ops.push(match (r >> 32) % 9 { + 0 => set(node_key(a, plane), Some(atom("g", len, seed))), + 1 => set(edge_key(b, plane), Some(atom("h", len, seed))), + 2 => set(node_key(a, plane), None), + 3 => upsert_node(a, if seed % 2 == 0 { "even" } else { "odd" }), + 4 => WarpOp::DeleteNode { node: node(a) }, + 5 => upsert_edge(b, a % 2, a, "g"), + 6 => delete_edge(a % 2, b), + 7 => set(node_key(a, plane), Some(AttachmentValue::Descend(warp(b)))), + _ => upsert_instance(b, if seed % 2 == 0 { "e" } else { "o" }, None), + }); + } + let mut slot = || { + let r = next(); + let a = small(r % 6); + match (r >> 8) % 5 { + 0 => SlotId::Node(node(a)), + 1 => SlotId::Edge(edge(a % 4)), + 2 => SlotId::Attachment(node_key(a, Alpha)), + 3 => SlotId::Port((root(), u64::from(a))), + _ => range(a, "g"), + } + }; + let in_slots = (0..200).map(|_| slot()).collect(); + let out_slots = (0..100).map(|_| slot()).collect(); + Vector { + name: "generated", + policy_id: 42, + rule_pack_id: *blake3::hash(b"cv/rule-pack/generated").as_bytes(), + status: TickCommitStatus::Committed, + in_slots, + out_slots, + ops, + } +} + +fn vectors() -> [Vector; 4] { + [every_kind(), last_wins(), slots_only(), generated()] +} + +fn build(v: &Vector) -> WarpTickPatchV1 { + WarpTickPatchV1::new( + v.policy_id, + v.rule_pack_id, + v.status, + v.in_slots.clone(), + v.out_slots.clone(), + v.ops.clone(), + ) +} + +/// Where each canonical op came from: the index of the last input op equal +/// to it. Last-wins keeps the last op of each key, so this names the survivor. +fn op_sources(v: &Vector, patch: &WarpTickPatchV1) -> Vec { + patch + .ops() + .iter() + .map(|op| { + v.ops + .iter() + .rposition(|input| input == op) + .unwrap_or(usize::MAX) + }) + .collect() +} + +/// Each canonical slot as the index of its first occurrence in the input. +fn slot_sources(input: &[SlotId], canonical: &[SlotId]) -> Vec { + canonical + .iter() + .map(|slot| input.iter().position(|s| s == slot).unwrap_or(usize::MAX)) + .collect() +} + +// ---- end fixture ---- + +// ---- pinned baseline values ---- +// +// Produced by the map-based implementation this module's parent replaced: +// main at 0faaf58a, which canonicalized ops through a `BTreeMap` and fed +// `blake3::Hasher` field by field. See the commit that added this module for +// how each value was captured. + +struct Expected { + name: &'static str, + digest: &'static str, + input_len: usize, + /// The full digest input, for the vectors small enough to pin in hex. + input: Option<&'static str>, + in_slots: &'static [usize], + out_slots: &'static [usize], + ops: &'static [usize], +} + +const EXPECTED: [Expected; 4] = [ + Expected { + name: "every-kind", + digest: "595c11495cbda213c2ae7d66b118b0c29fb5dfea7b1f54b82bb2baae55361f8d", + input_len: 3838, + input: Some( + "6563686f3a70617463685f6469676573743a76310002000403020118bed28c92e7d37b8bcdb1fe22fc110d0f\ + cf9f1b1d6740b29a299292db61e32d010c00000000000000017cfd0ecd52438094cebc57213a8088cdd025cc\ + 04e5ebe04786094e87b8e5a1e8c7031e935da5d0d07397f788c983158b593e4888a29eecb18e9f636b603a61\ + a4017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc\ + 3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109a027cfd0ecd52438094cebc57213a8088cdd025cc04e5\ + ebe04786094e87b8e5a1e8802dbb84cf8ee62b5f6d2363258f3c3a9d694218006bd7edfad04595c22cf7c302\ + 7cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8dd323370f109585c7159764b\ + 9101af942710b689e5740356bbf81ddeff2c847c0301027cfd0ecd52438094cebc57213a8088cdd025cc04e5\ + ebe04786094e87b8e5a1e8c7031e935da5d0d07397f788c983158b593e4888a29eecb18e9f636b603a61a403\ + 01017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc\ + 3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109a0302027cfd0ecd52438094cebc57213a8088cdd025cc\ + 04e5ebe04786094e87b8e5a1e8802dbb84cf8ee62b5f6d2363258f3c3a9d694218006bd7edfad04595c22cf7\ + c30302017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8dd323370f109585c\ + 7159764b9101af942710b689e5740356bbf81ddeff2c847c0402795db50eb71e7272a99f80b971caa9854542\ + ea3cc809cb60a83c13f1172984ffffffffffffffff047cfd0ecd52438094cebc57213a8088cdd025cc04e5eb\ + e04786094e87b8e5a1e80700000000000000057cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786\ + 094e87b8e5a1e8c7031e935da5d0d07397f788c983158b593e4888a29eecb18e9f636b603a61a49bc196fc65\ + cd931dd1e2ac56e25b459cdf6134d24b31312fbb3d09040f57bfd1057cfd0ecd52438094cebc57213a8088cd\ + d025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0bcbb\ + 848d109aa68820c27384e15676f3c6959602423ed2da55c424c4750d1cf69c285c9105430600000000000000\ + 017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc3e\ + a15ae1a85936508ad2e399c25bf4b0bcbb848d109a027cfd0ecd52438094cebc57213a8088cdd025cc04e5eb\ + e04786094e87b8e5a1e8dd323370f109585c7159764b9101af942710b689e5740356bbf81ddeff2c847c0301\ + 017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc3e\ + a15ae1a85936508ad2e399c25bf4b0bcbb848d109a0302027cfd0ecd52438094cebc57213a8088cdd025cc04\ + e5ebe04786094e87b8e5a1e8802dbb84cf8ee62b5f6d2363258f3c3a9d694218006bd7edfad04595c22cf7c3\ + 047cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e80700000000000000057cfd\ + 0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc3ea15ae1\ + a85936508ad2e399c25bf4b0bcbb848d109aa68820c27384e15676f3c6959602423ed2da55c424c4750d1cf6\ + 9c285c9105430e000000000000000801027cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e\ + 87b8e5a1e8cb34d310ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109a02795db50eb71e\ + 7272a99f80b971caa9854542ea3cc809cb60a83c13f1172984d72f7063a2bf3406562fac2af6ba1c1497d3e4\ + 839aa95f780f1890045b5af08e01b05a0e3c4f2ba9eae3701b9844368830e329c827811fc86d77937e902753\ + 539e0802017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8dd323370f10958\ + 5c7159764b9101af942710b689e5740356bbf81ddeff2c847c33f6a8278dd1a5c9f38cb39a5abd058f837b33\ + 6a6b9236bfa9266ee05e84ff3343a59cc753d34e675afafe6fcb3c9e95262d65754302c9ec96a6e5141de16c\ + 380001118d5a736470be5afee95736817acdc9688fa0c0947b5337c078d39f1048e3e3798076ba5e6ebd35b3\ + 2006ed8784867df776ac6ca6ec55ec466a2d201bb0f45200013ecd35811337548b187bc717ef73da35105f72\ + 8d8c64fc5007b319addcb54875dc47c911f83a8b37f7cc551706fa1fb4000eb7d1719fa7b76c2f7cb672300c\ + 280101017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e89af1c9b95e3bf77c\ + 38b0e531f9a7190cac7fab0edc0ee751ed9e92903280010f028518043b559aa886db195aa75323be9ba10d99\ + e67b26fff0f0a92c0f6bfc115a067cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5\ + a1e8cb34d310ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109a924b6a204cb6db8d9410\ + 9014181a37206fcbe20ef07779fdbaecc1f32be8584e047cfd0ecd52438094cebc57213a8088cdd025cc04e5\ + ebe04786094e87b8e5a1e88c2876d3f6c70d034b3780594ca324ca1e722e2ec662e2298285c4815024ddad03\ + 7cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8c7031e935da5d0d07397f788\ + c983158b593e4888a29eecb18e9f636b603a61a4422a5d807a4046a1b3b79c5f121794de54d24ef6ed34d86d\ + 393ddf8ece658d63057cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8cb34d3\ + 10ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109add323370f109585c7159764b9101af\ + 942710b689e5740356bbf81ddeff2c847cc7031e935da5d0d07397f788c983158b593e4888a29eecb18e9f63\ + 6b603a61a4a68820c27384e15676f3c6959602423ed2da55c424c4750d1cf69c285c9105430701017cfd0ecd\ + 52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e88c2876d3f6c70d034b3780594ca324ca\ + 1e722e2ec662e2298285c4815024ddad0101624ff5959be8abdad1f66c3cd0556ec652d007ba6e5902044222\ + ef0d7cafe17e00000000000000000701017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e\ + 87b8e5a1e8c7031e935da5d0d07397f788c983158b593e4888a29eecb18e9f636b603a61a4010202795db50e\ + b71e7272a99f80b971caa9854542ea3cc809cb60a83c13f11729840701017cfd0ecd52438094cebc57213a80\ + 88cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0\ + bcbb848d109a000702027cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8802d\ + bb84cf8ee62b5f6d2363258f3c3a9d694218006bd7edfad04595c22cf7c30101593b294b89a9d82d3266c4f5\ + 42d1bb3b68aed67782409154eb168c2796bee8594c04000000000000032241607f9ebddcfb1a39587796b5d4\ + f31231506f8eadcceb0a29486786a5c4e30221405f7e9dbcdbfa1938577695b4d3f211304f6e8daccbea0928\ + 476685a4c3e201203f5e7d9cbbdaf91837567594b3d2f1102f4e6d8cabcae90827466584a3c2e1001f3e5d7c\ + 9bbad9f81736557493b2d1f00f2e4d6c8baac9e80726456483a2c1e0ff1e3d5c7b9ab9d8f71635547392b1d0\ + ef0e2d4c6b8aa9c8e70625446382a1c0dffe1d3c5b7a99b8d7f61534537291b0cfee0d2c4b6a89a8c7e60524\ + 436281a0bfdefd1c3b5a7998b7d6f51433527190afceed0c2b4a6988a7c6e504234261809fbeddfc1b3a5978\ + 97b6d5f4133251708faecdec0b2a496887a6c5e4032241607f9ebddcfb1a39587796b5d4f31231506f8eadcc\ + eb0a29486786a5c4e30221405f7e9dbcdbfa1938577695b4d3f211304f6e8daccbea0928476685a4c3e20120\ + 3f5e7d9cbbdaf91837567594b3d2f1102f4e6d8cabcae90827466584a3c2e1001f3e5d7c9bbad9f817365574\ + 93b2d1f00f2e4d6c8baac9e80726456483a2c1e0ff1e3d5c7b9ab9d8f71635547392b1d0ef0e2d4c6b8aa9c8\ + e70625446382a1c0dffe1d3c5b7a99b8d7f61534537291b0cfee0d2c4b6a89a8c7e60524436281a0bfdefd1c\ + 3b5a7998b7d6f51433527190afceed0c2b4a6988a7c6e504234261809fbeddfc1b3a597897b6d5f413325170\ + 8faecdec0b2a496887a6c5e4032241607f9ebddcfb1a39587796b5d4f31231506f8eadcceb0a29486786a5c4\ + e30221405f7e9dbcdbfa1938577695b4d3f211304f6e8daccbea0928476685a4c3e201203f5e7d9cbbdaf918\ + 37567594b3d2f1102f4e6d8cabcae90827466584a3c2e1001f3e5d7c9bbad9f81736557493b2d1f00f2e4d6c\ + 8baac9e80726456483a2c1e0ff1e3d5c7b9ab9d8f71635547392b1d0ef0e2d4c6b8aa9c8e70625446382a1c0\ + dffe1d3c5b7a99b8d7f61534537291b0cfee0d2c4b6a89a8c7e60524436281a0bfdefd1c3b5a7998b7d6f514\ + 33527190afceed0c2b4a6988a7c6e504234261809fbeddfc1b3a597897b6d5f4133251708faecdec0b2a4968\ + 87a6c5e4032241607f9ebddcfb1a39587796b5d4f31231506f8eadcceb0a29486786a5c4e30221405f7e9dbc\ + dbfa1938577695b4d3f211304f6e8daccbea0928476685a4c3e201203f5e7d9cbbdaf91837567594b3d2f110\ + 2f4e6d8cabcae90827466584a3c2e1001f3e5d7c9bbad9f81736557493b2d1f00f2e4d6c8baac9e807264564\ + 83a2c1e0ff1e3d5c7b9ab9d8f71635547392b1d0ef0e2d4c6b8aa9c8e70625446382a1c0dffe1d3c5b7a99b8\ + d7f61534537291b0cfee0d2c4b6a89a8c7e60524436281a0bfdefd1c3b5a7998b7d6f51433527190afceed0c\ + 2b4a6988a7c6e504234261809fbeddfc1b3a597897b6d5f4133251708faecdec0b2a496887a6c5e403224160\ + 7f9ebddcfb1a39587796b5d4f31231506f8eadcceb0a29486786a5c4e30221405f7e9dbcdbfa1938577695b4\ + d3f211304f6e8daccbea0928476685a4c3e201203f5e7d9cbbdaf9180702027cfd0ecd52438094cebc57213a\ + 8088cdd025cc04e5ebe04786094e87b8e5a1e8dd323370f109585c7159764b9101af942710b689e5740356bb\ + f81ddeff2c847c0101d7c9c1633359aa670e100631e5a07c776ba83fd72f0c927bba00824cad4271c8050000\ + 00000000000928476685", + ), + in_slots: &[3, 9, 10, 5, 6, 4, 2, 7, 8, 1, 13, 0], + out_slots: &[1, 4, 5, 0, 3, 2], + ops: &[3, 8, 11, 5, 10, 6, 2, 7, 1, 13, 4, 9, 0, 12], + }, + Expected { + name: "last-wins", + digest: "a7e9f7a84bbc9ec49bf1934d21bf4b7a76e473a69e6fb778189e6f5d9ccf46ac", + input_len: 1814, + input: Some( + "6563686f3a70617463685f6469676573743a7631000200070000002df57616db48b7d383847627da0a203c5a\ + 00ffdc3e20348c4225a19fb9af806d020400000000000000017cfd0ecd52438094cebc57213a8088cdd025cc\ + 04e5ebe04786094e87b8e5a1e8c7031e935da5d0d07397f788c983158b593e4888a29eecb18e9f636b603a61\ + a4027cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8dd323370f109585c7159\ + 764b9101af942710b689e5740356bbf81ddeff2c847c0301017cfd0ecd52438094cebc57213a8088cdd025cc\ + 04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0bcbb848d10\ + 9a057cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc\ + 3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109aa68820c27384e15676f3c6959602423ed2da55c424c4\ + 750d1cf69c285c9105430200000000000000017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786\ + 094e87b8e5a1e8c7031e935da5d0d07397f788c983158b593e4888a29eecb18e9f636b603a61a40301017cfd\ + 0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc3ea15ae1\ + a85936508ad2e399c25bf4b0bcbb848d109a0c000000000000000801027cfd0ecd52438094cebc57213a8088\ + cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0bc\ + bb848d109a33f6a8278dd1a5c9f38cb39a5abd058f837b336a6b9236bfa9266ee05e84ff33dc47c911f83a8b\ + 37f7cc551706fa1fb4000eb7d1719fa7b76c2f7cb672300c2800013ecd35811337548b187bc717ef73da3510\ + 5f728d8c64fc5007b319addcb54875dc47c911f83a8b37f7cc551706fa1fb4000eb7d1719fa7b76c2f7cb672\ + 300c280102027cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8802dbb84cf8e\ + e62b5f6d2363258f3c3a9d694218006bd7edfad04595c22cf7c3028518043b559aa886db195aa75323be9ba1\ + 0d99e67b26fff0f0a92c0f6bfc115a067cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87\ + b8e5a1e8cb34d310ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109a924b6a204cb6db8d\ + 94109014181a37206fcbe20ef07779fdbaecc1f32be8584e047cfd0ecd52438094cebc57213a8088cdd025cc\ + 04e5ebe04786094e87b8e5a1e88c2876d3f6c70d034b3780594ca324ca1e722e2ec662e2298285c4815024dd\ + ad047cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8c7031e935da5d0d07397\ + f788c983158b593e4888a29eecb18e9f636b603a61a4037cfd0ecd52438094cebc57213a8088cdd025cc04e5\ + ebe04786094e87b8e5a1e8c7031e935da5d0d07397f788c983158b593e4888a29eecb18e9f636b603a61a484\ + e39c444a71ee7de44499c33a9fd70922dbf78a4ddf1b9a60962baf16294b08057cfd0ecd52438094cebc5721\ + 3a8088cdd025cc04e5ebe04786094e87b8e5a1e89af1c9b95e3bf77c38b0e531f9a7190cac7fab0edc0ee751\ + ed9e92903280010fdd323370f109585c7159764b9101af942710b689e5740356bbf81ddeff2c847c8c2876d3\ + f6c70d034b3780594ca324ca1e722e2ec662e2298285c4815024ddad56b636c1be8ec2bad5b43286e049e828\ + a7789830285e523b20b86e708776b5fc057cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e\ + 87b8e5a1e8cb34d310ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109add323370f10958\ + 5c7159764b9101af942710b689e5740356bbf81ddeff2c847c9af1c9b95e3bf77c38b0e531f9a7190cac7fab\ + 0edc0ee751ed9e92903280010f2cf6c61b974de434a9bed84b98f7e162e29e82927b69ed655586fff83f762e\ + e20701017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8cb34d310ccbe0e32\ + 3abc3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109a0101c1dca8679dc258dcba6c4b8c1cbcbd53f46f\ + 79eee298e628d4ecc6c90585593928000000000000000221405f7e9dbcdbfa1938577695b4d3f211304f6e8d\ + accbea0928476685a4c3e201203f5e7d9cbb0701027cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe0\ + 4786094e87b8e5a1e8cb34d310ccbe0e323abc3ea15ae1a85936508ad2e399c25bf4b0bcbb848d109a0101fa\ + 2467432f8991507761023bda45f1c15f564adf00d49301c3303899eda23a6b02000000000000000423070202\ + 7cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8802dbb84cf8ee62b5f6d2363\ + 258f3c3a9d694218006bd7edfad04595c22cf7c30102d226b66ead06217b86091a64eb029ac191cdaa714723\ + acbf7ea35ffafd383e6a", + ), + in_slots: &[0, 5, 1, 3], + out_slots: &[1, 0], + ops: &[11, 12, 15, 14, 13, 17, 20, 16, 21, 19, 18, 23], + }, + Expected { + name: "slots-only", + digest: "fac1598ca120c8a9c82b54b5179105449635e067d22bb74df8fc5749fde081fb", + input_len: 1164, + input: Some( + "6563686f3a70617463685f6469676573743a7631000200ffffffffa5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5\ + a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5010800000000000000017cfd0ecd52438094cebc57213a8088cdd025cc\ + 04e5ebe04786094e87b8e5a1e86643743695fe0625bf8fe1066f0251bdc4782dad4fd175230232e04edf780f\ + 28027cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8c90cf1688c4ff9294b79\ + 7047040280dd6e64591f564d4cbfab330ce5e205880b0301027cfd0ecd52438094cebc57213a8088cdd025cc\ + 04e5ebe04786094e87b8e5a1e8541697e512bf77289a7db12aac5e98afe865acea3681eb1d095abdafc81397\ + ef0302017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e82b0c4993d5accd7d\ + 703f954a4787ebde7f163a922bfb3bf7d32a1798252d6d18043ecd35811337548b187bc717ef73da35105f72\ + 8d8c64fc5007b319addcb548750000000000000000043ecd35811337548b187bc717ef73da35105f728d8c64\ + fc5007b319addcb548750100000000000000057cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786\ + 094e87b8e5a1e88c2876d3f6c70d034b3780594ca324ca1e722e2ec662e2298285c4815024ddad4181a4d86f\ + 31a3bdd2a3bf966b723a108c86bf457af711a8bcc5fb1ca1d8ac25057cfd0ecd52438094cebc57213a8088cd\ + d025cc04e5ebe04786094e87b8e5a1e88c2876d3f6c70d034b3780594ca324ca1e722e2ec662e2298285c481\ + 5024ddad72a4950faeb19e049ed341b5f7309174cc14fa6f0d3ce65f181de1e3a5e016c00800000000000000\ + 017cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e86643743695fe0625bf8fe1\ + 066f0251bdc4782dad4fd175230232e04edf780f28027cfd0ecd52438094cebc57213a8088cdd025cc04e5eb\ + e04786094e87b8e5a1e8c90cf1688c4ff9294b797047040280dd6e64591f564d4cbfab330ce5e205880b0301\ + 027cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e8541697e512bf77289a7db1\ + 2aac5e98afe865acea3681eb1d095abdafc81397ef0302017cfd0ecd52438094cebc57213a8088cdd025cc04\ + e5ebe04786094e87b8e5a1e82b0c4993d5accd7d703f954a4787ebde7f163a922bfb3bf7d32a1798252d6d18\ + 043ecd35811337548b187bc717ef73da35105f728d8c64fc5007b319addcb548750000000000000000043ecd\ + 35811337548b187bc717ef73da35105f728d8c64fc5007b319addcb548750100000000000000057cfd0ecd52\ + 438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e88c2876d3f6c70d034b3780594ca324ca1e\ + 722e2ec662e2298285c4815024ddad4181a4d86f31a3bdd2a3bf966b723a108c86bf457af711a8bcc5fb1ca1\ + d8ac25057cfd0ecd52438094cebc57213a8088cdd025cc04e5ebe04786094e87b8e5a1e88c2876d3f6c70d03\ + 4b3780594ca324ca1e722e2ec662e2298285c4815024ddad72a4950faeb19e049ed341b5f7309174cc14fa6f\ + 0d3ce65f181de1e3a5e016c00000000000000000", + ), + in_slots: &[2, 0, 5, 6, 1, 4, 3, 7], + out_slots: &[4, 6, 1, 0, 5, 2, 3, 7], + ops: &[], + }, + Expected { + name: "generated", + digest: "875f62830875abe16e5e21d1ea01c3e28610d12607762f1da6239736ca0b1119", + input_len: 9579, + input: None, + in_slots: &[26, 17, 6, 10, 45, 65, 4, 7, 19, 39, 22, 15, 61, 52, 0, 76, 5, 98, 1, 12, 8, 25, 43, 16, 11, 42, 2, 36], + out_slots: &[6, 31, 10, 26, 4, 19, 16, 34, 74, 7, 9, 57, 39, 25, 0, 65, 38, 47, 51, 36, 11, 1, 5, 67, 3, 93, 63], + ops: &[206, 386, 369, 391, 264, 290, 336, 308, 378, 249, 381, 343, 379, 397, 370, 322, 344, 237, 396, 394, 393, 384, 380, 305, 329, 289, 376, 363, 285, 390, 226, 348, 389, 399, 383, 382, 398, 392, 350, 262, 373, 385, 365, 328, 252, 395, 387, 297, 375, 307, 303, 337], + }, +]; + +fn digest_input(patch: &WarpTickPatchV1) -> Vec { + patch_digest_input_v2( + patch.policy_id(), + &patch.rule_pack_id(), + patch.commit_status(), + patch.in_slots(), + patch.out_slots(), + patch.ops(), + ) +} + +#[test] +fn patch_vectors_match_the_baseline_bytes_and_digests() { + for (v, expected) in vectors().iter().zip(EXPECTED.iter()) { + assert_eq!(v.name, expected.name); + let patch = build(v); + let input = digest_input(&patch); + assert_eq!(input.len(), expected.input_len, "{}: input length", v.name); + if let Some(hex) = expected.input { + assert_eq!(hex::encode(&input), hex, "{}: digest input bytes", v.name); + } + assert_eq!( + hex::encode(patch.digest()), + expected.digest, + "{}: patch digest", + v.name + ); + assert_eq!(blake3::hash(&input).as_bytes(), &patch.digest()); + assert!(patch.validate_digest().is_ok()); + assert_eq!( + slot_sources(&v.in_slots, patch.in_slots()), + expected.in_slots, + "{}: canonical in_slots", + v.name + ); + assert_eq!( + slot_sources(&v.out_slots, patch.out_slots()), + expected.out_slots, + "{}: canonical out_slots", + v.name + ); + assert_eq!( + op_sources(v, &patch), + expected.ops, + "{}: canonical ops", + v.name + ); + } +} + +/// The map and set canonicalizers this module's parent replaced, kept as an +/// independent oracle over every vector. +#[test] +fn patch_vectors_match_the_map_and_set_canonicalizers() { + for v in vectors() { + let patch = build(&v); + let mut map = std::collections::BTreeMap::new(); + for op in v.ops.iter().cloned() { + map.insert(op.sort_key(), op); + } + let ops: Vec = map.into_values().collect(); + let in_slots: Vec = v + .in_slots + .iter() + .copied() + .collect::>() + .into_iter() + .collect(); + let out_slots: Vec = v + .out_slots + .iter() + .copied() + .collect::>() + .into_iter() + .collect(); + assert_eq!(patch.ops(), ops.as_slice(), "{}: ops", v.name); + assert_eq!( + patch.in_slots(), + in_slots.as_slice(), + "{}: in_slots", + v.name + ); + assert_eq!( + patch.out_slots(), + out_slots.as_slice(), + "{}: out_slots", + v.name + ); + } +} + +fn op_kind(op: &WarpOp) -> &'static str { + match op { + WarpOp::OpenPortal { + init: PortalInit::Empty { .. }, + .. + } => "open-portal/empty", + WarpOp::OpenPortal { + init: PortalInit::RequireExisting, + .. + } => "open-portal/require-existing", + WarpOp::UpsertWarpInstance { instance } => match instance.parent { + None => "upsert-instance/no-parent", + Some(_) => "upsert-instance/parent", + }, + WarpOp::DeleteWarpInstance { .. } => "delete-instance", + WarpOp::UpsertNode { .. } => "upsert-node", + WarpOp::DeleteNode { .. } => "delete-node", + WarpOp::UpsertEdge { .. } => "upsert-edge", + WarpOp::DeleteEdge { .. } => "delete-edge", + WarpOp::SetAttachment { value: None, .. } => "set-attachment/clear", + WarpOp::SetAttachment { + value: Some(AttachmentValue::Atom(atom)), + .. + } => { + if atom.bytes.is_empty() { + "set-attachment/atom-empty" + } else { + "set-attachment/atom" + } + } + WarpOp::SetAttachment { + value: Some(AttachmentValue::Descend(_)), + .. + } => "set-attachment/descend", + } +} + +fn slot_kind(slot: &SlotId) -> &'static str { + match slot { + SlotId::Node(_) => "node", + SlotId::Edge(_) => "edge", + SlotId::Attachment(key) => key_kind(key), + SlotId::Port(_) => "port", + SlotId::OutgoingEdges(_) => "outgoing-edges", + } +} + +fn key_kind(key: &AttachmentKey) -> &'static str { + match (key.owner, key.plane) { + (AttachmentOwner::Node(_), AttachmentPlane::Alpha) => "attachment/node-alpha", + (AttachmentOwner::Node(_), AttachmentPlane::Beta) => "attachment/node-beta", + (AttachmentOwner::Edge(_), AttachmentPlane::Alpha) => "attachment/edge-alpha", + (AttachmentOwner::Edge(_), AttachmentPlane::Beta) => "attachment/edge-beta", + } +} + +/// The vectors exercise every op variant (and each sub-variant the encoding +/// distinguishes), every slot kind and every attachment owner and plane. A +/// new variant fails to compile in the matches above until it is covered. +#[test] +fn patch_vectors_cover_every_op_and_slot_kind() { + use std::collections::BTreeSet; + let all = vectors(); + let ops: BTreeSet<&str> = all.iter().flat_map(|v| v.ops.iter().map(op_kind)).collect(); + assert_eq!( + ops, + BTreeSet::from([ + "open-portal/empty", + "open-portal/require-existing", + "upsert-instance/no-parent", + "upsert-instance/parent", + "delete-instance", + "upsert-node", + "delete-node", + "upsert-edge", + "delete-edge", + "set-attachment/clear", + "set-attachment/atom", + "set-attachment/atom-empty", + "set-attachment/descend", + ]) + ); + let every_slot = BTreeSet::from([ + "node", + "edge", + "attachment/node-alpha", + "attachment/node-beta", + "attachment/edge-alpha", + "attachment/edge-beta", + "port", + "outgoing-edges", + ]); + let every = &all[0]; + let in_kinds: BTreeSet<&str> = every.in_slots.iter().map(slot_kind).collect(); + assert_eq!(in_kinds, every_slot, "every-kind in_slots"); + let out_kinds: BTreeSet<&str> = every.out_slots.iter().map(slot_kind).collect(); + for kind in ["node", "edge", "port", "outgoing-edges"] { + assert!( + out_kinds.contains(kind), + "every-kind out_slots lacks {kind}" + ); + } + // Each op variant repeats a key in the last-wins vector. + let last_wins = &all[1]; + let mut seen = BTreeSet::new(); + let mut repeated = BTreeSet::new(); + for op in &last_wins.ops { + if !seen.insert(op.sort_key()) { + repeated.insert(op_kind(op).split('/').next().unwrap_or_default()); + } + } + assert_eq!( + repeated, + BTreeSet::from([ + "open-portal", + "upsert-instance", + "delete-instance", + "upsert-node", + "delete-node", + "upsert-edge", + "delete-edge", + "set-attachment", + ]) + ); +} From 60bcf0951bc59a908777d96da5b00215e0d6ac46 Mon Sep 17 00:00:00 2001 From: James Ross Date: Wed, 21 Oct 2026 07:17:01 -0700 Subject: [PATCH 3/6] Bound the commit's slot lists when footprints overlap Collecting a tick's slots as plain vectors with repeats (747f320c) made peak memory grow with the sum of all footprint claims instead of the distinct slots, and the patch kept the full repeat-sized capacity after dedup, so a committed patch retained it too. Measured with a new peak-live-bytes witness, crates/echo-alloc-witness/tests/commit_overlap_peak.rs (its own counting allocator, which tracks live bytes and their peak; no global elsewhere). 4,096 rewrites, each reading the same 64 nodes and their attachments and writing its own attachment, against a control tick with no shared reads. Peak live bytes above the commit's start, from the logs (scratchpad runners/logs/peak3-*.log): revision control peak overlap peak overlap net live main 0faaf58a 15,604,360 58,538,568 -56,143,480 747f320c 15,357,224 166,091,952 52,043,144 this commit 15,357,224 60,447,920 -56,143,480 So 747f320c raised the overlapping tick's peak 2.8x over main and kept about 108 MB live after the commit. Now: - `push_slot` compacts a full slot list (stable sort and dedup) before it grows, and doubles only when that frees less than half. The list holds at most about four times the distinct slots, and each compaction is paid for by at least half a list of pushes. - `WarpTickPatchV1::new` shrinks its slot lists to fit after dedup, as the former BTreeSet-to-Vec collection did implicitly. The overlapping peak is now 3.3% above main's. The witness asserts the overlapping peak stays under 6 times the control's: it fails on 747f320c (166,091,952 vs 15,357,224; log peak3-red-747f320c.log) and passes on main (peak3-main.log) and this commit (peak3-candidate.log). Total bytes allocated during the overlapping commit remain higher than main (423,807,536 vs 324,837,616), from the sort buffers and the gathered digest input; they are freed within the commit. Speed on the non-overlapping 10,000-write tick (study bench scheduler_workers, run in a scratch worktree, 31 rounds, alternating sources; logs bench-747-{1,2}.log, bench-cand-{1,2}.log): commit p50 with 1 worker 21.044 and 21.263 ms at 747f320c, 21.184 and 21.956 ms with this change; with 18 workers 21.833 and 22.108 ms against 22.327 and 23.032 ms. That is up to about 3% slower than 747f320c on this bench. Slot order, patch bytes and digests are unchanged: the compatibility vectors and the overlapping-footprint engine witness pass. Co-Authored-By: Claude Opus 5.5 --- .../tests/commit_overlap_peak.rs | 260 ++++++++++++++++++ crates/warp-core/src/engine_impl.rs | 52 +++- crates/warp-core/src/tick_patch.rs | 4 + 3 files changed, 301 insertions(+), 15 deletions(-) create mode 100644 crates/echo-alloc-witness/tests/commit_overlap_peak.rs diff --git a/crates/echo-alloc-witness/tests/commit_overlap_peak.rs b/crates/echo-alloc-witness/tests/commit_overlap_peak.rs new file mode 100644 index 00000000..dbfd1110 --- /dev/null +++ b/crates/echo-alloc-witness/tests/commit_overlap_peak.rs @@ -0,0 +1,260 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +//! Peak temporary memory of one commit whose rewrites claim heavily +//! overlapping footprints. +//! +//! Reservation collects the tick's `in_slots` and `out_slots` as plain +//! vectors that keep repeats until the patch sorts and dedups them, and the +//! patch digest gathers its whole input before hashing it. Both add +//! temporary memory. Here every rewrite reads the same `SHARED` nodes and +//! their attachments, so the repeats are as dense as footprints allow (reads +//! may overlap; writes may not). +//! +//! The allocator tracks live bytes (allocated minus freed) during the +//! commit and records their peak above the level when the commit began. +//! This file holds one test and its own allocator, so that nothing else in +//! the process allocates while the measurement runs. +#![allow( + unsafe_code, + missing_docs, + clippy::expect_used, + clippy::unwrap_used, + clippy::panic, + clippy::print_stderr +)] + +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicBool, AtomicIsize, AtomicUsize, Ordering}; + +use bytes::Bytes; +use echo_dry_tests::build_motion_demo_engine; +use warp_core::{ + make_node_id, make_type_id, AtomPayload, AttachmentKey, AttachmentValue, ConflictPolicy, + Engine, ExecutionGraphView, Footprint, GraphView, NodeId, NodeKey, NodeRecord, PatternGraph, + RewriteRule, RuleExecutor, TickDelta, WarpOp, +}; + +/// Tracks live bytes, and their peak, while `ACTIVE` is set. +struct PeakAllocator; + +static ACTIVE: AtomicBool = AtomicBool::new(false); +static LIVE: AtomicIsize = AtomicIsize::new(0); +static PEAK: AtomicIsize = AtomicIsize::new(0); +static TOTAL: AtomicUsize = AtomicUsize::new(0); + +fn grew(bytes: usize) { + let bytes = isize::try_from(bytes).unwrap_or(isize::MAX); + let live = LIVE.fetch_add(bytes, Ordering::SeqCst) + bytes; + PEAK.fetch_max(live, Ordering::SeqCst); +} + +fn shrank(bytes: usize) { + LIVE.fetch_sub( + isize::try_from(bytes).unwrap_or(isize::MAX), + Ordering::SeqCst, + ); +} + +// SAFETY: every call is forwarded unchanged to the system allocator; the +// counters do not affect what is allocated or freed. +unsafe impl GlobalAlloc for PeakAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + if ACTIVE.load(Ordering::SeqCst) { + TOTAL.fetch_add(layout.size(), Ordering::SeqCst); + grew(layout.size()); + } + // SAFETY: `layout` is the caller's layout, passed through. + unsafe { System.alloc(layout) } + } + + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + if ACTIVE.load(Ordering::SeqCst) { + shrank(layout.size()); + } + // SAFETY: `ptr` and `layout` are the caller's, passed through. + unsafe { System.dealloc(ptr, layout) } + } + + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { + if ACTIVE.load(Ordering::SeqCst) { + TOTAL.fetch_add(new_size, Ordering::SeqCst); + // A moving realloc holds both blocks for a moment; count the + // new block before the old one is freed. + grew(new_size); + shrank(layout.size()); + } + // SAFETY: the caller's arguments, passed through. + unsafe { System.realloc(ptr, layout, new_size) } + } +} + +#[global_allocator] +static ALLOCATOR: PeakAllocator = PeakAllocator; + +/// Nodes every rewrite reads, with their attachments, in the overlapping +/// tick. The control tick reads none of them. +const SHARED: usize = 64; +/// Rewrites in the measured tick, one per own node. +const REWRITES: usize = 4_096; +/// How far the overlapping tick's peak may rise above the control's. +/// +/// The overlapping tick claims 131 read slots per rewrite instead of three, +/// so the receipt and scheduler legitimately need more room: the +/// `BTreeSet`-based collection peaked at about 3.8 times the control. Slot +/// lists that keep every repeat and hand their full capacity to the patch +/// peaked at about 11 times. Compacting the lists keeps it near 4. +const BUDGET_FACTOR: isize = 6; + +fn shared(i: usize) -> NodeId { + make_node_id(&format!("commit-overlap-peak/shared/{i}")) +} + +fn own(i: usize) -> NodeId { + make_node_id(&format!("commit-overlap-peak/own/{i}")) +} + +fn matcher(_view: GraphView<'_>, _n: &NodeId) -> bool { + true +} + +fn executor(view: &mut ExecutionGraphView<'_, '_>, n: &NodeId, delta: &mut TickDelta) { + // The patch is a state diff, so each tick must change the value: + // alternate between two payloads. + let next: &'static [u8] = match view.node_attachment(n) { + Some(AttachmentValue::Atom(atom)) if atom.bytes.as_ref() == b"even" => b"odd", + _ => b"even", + }; + delta.emit(WarpOp::SetAttachment { + key: AttachmentKey::node_alpha(NodeKey { + warp_id: view.warp_id(), + local_id: *n, + }), + value: Some(AttachmentValue::Atom(AtomPayload::new( + make_type_id("commit-overlap-peak/value"), + Bytes::from_static(next), + ))), + }); +} + +/// Reads the first `S` shared nodes and their attachments, and reads and +/// writes the scope's own attachment. +fn footprint(view: GraphView<'_>, n: &NodeId) -> Footprint { + let warp_id = view.warp_id(); + let mut footprint = Footprint::default(); + for i in 0..S { + let node = NodeKey { + warp_id, + local_id: shared(i), + }; + footprint.n_read.insert(node); + footprint.a_read.insert(AttachmentKey::node_alpha(node)); + } + let own = NodeKey { + warp_id, + local_id: *n, + }; + footprint.n_read.insert(own); + footprint.a_read.insert(AttachmentKey::node_alpha(own)); + footprint.a_write.insert(AttachmentKey::node_alpha(own)); + footprint +} + +fn rule(name: &'static str) -> RewriteRule { + RewriteRule { + id: blake3::hash(name.as_bytes()).into(), + name, + left: PatternGraph { nodes: vec![] }, + matcher, + executor: RuleExecutor::observed(executor), + compute_footprint: footprint::, + factor_mask: 0, + conflict_policy: ConflictPolicy::Abort, + join_fn: None, + } +} + +fn engine(rule: RewriteRule) -> Engine { + let mut engine = build_motion_demo_engine(); + engine.register_rule(rule).expect("rule registers"); + let ty = make_type_id("commit-overlap-peak/entity"); + for i in 0..SHARED { + engine + .insert_node(shared(i), NodeRecord { ty }) + .expect("shared node inserts"); + } + for i in 0..REWRITES { + engine + .insert_node(own(i), NodeRecord { ty }) + .expect("own node inserts"); + } + engine +} + +struct Measured { + peak: isize, + net: isize, + total: usize, +} + +/// Peak live bytes above the commit's starting level, after a warm-up tick. +fn measure_commit(rule: RewriteRule) -> Measured { + let name = rule.name; + let mut engine = engine(rule); + let mut measured = None; + for measuring in [false, true] { + let tx = engine.begin(); + for i in 0..REWRITES { + engine.apply(tx, name, &own(i)).expect("rule applies"); + } + LIVE.store(0, Ordering::SeqCst); + PEAK.store(0, Ordering::SeqCst); + TOTAL.store(0, Ordering::SeqCst); + ACTIVE.store(measuring, Ordering::SeqCst); + let (snapshot, receipt, patch) = engine.commit_with_receipt(tx).expect("tick commits"); + ACTIVE.store(false, Ordering::SeqCst); + if measuring { + let applied = receipt + .entries() + .iter() + .filter(|entry| entry.disposition == warp_core::TickReceiptDisposition::Applied) + .count(); + assert_eq!(applied, REWRITES, "every rewrite must be accepted"); + assert_eq!(patch.ops().len(), REWRITES, "one op per rewrite"); + measured = Some(Measured { + peak: PEAK.load(Ordering::SeqCst), + net: LIVE.load(Ordering::SeqCst), + total: TOTAL.load(Ordering::SeqCst), + }); + } + std::hint::black_box((snapshot, receipt, patch)); + } + measured.expect("the measured commit ran") +} + +#[test] +fn overlapping_footprints_do_not_inflate_commit_peak_memory() { + if warp_core::DELTA_VALIDATE { + assert!( + std::env::var_os("ECHO_ALLOC_WITNESS_REQUIRED").is_none(), + "the allocation witness must run on an ordinary build here" + ); + eprintln!("not measured: warp-core is built with delta_validate"); + return; + } + let control = measure_commit(rule::<0>("commit-overlap-peak/control")); + let overlap = measure_commit(rule::("commit-overlap-peak/overlap")); + for (label, m) in [("control", &control), ("overlap", &overlap)] { + eprintln!( + "PEAK commit_overlap {label} rewrites={REWRITES} peak_live_bytes={} \ + net_live_bytes={} total_alloc_bytes={}", + m.peak, m.net, m.total + ); + } + assert!(control.peak > 0, "the allocator measured nothing"); + assert!( + overlap.peak < control.peak * BUDGET_FACTOR, + "overlapping footprints raised the commit's peak to {} bytes from {}", + overlap.peak, + control.peak + ); +} diff --git a/crates/warp-core/src/engine_impl.rs b/crates/warp-core/src/engine_impl.rs index f88e1bf8..889490fc 100644 --- a/crates/warp-core/src/engine_impl.rs +++ b/crates/warp-core/src/engine_impl.rs @@ -3443,48 +3443,70 @@ fn extend_slots_from_footprint( ) { // Nodes (warp-scoped NodeKey) for key in fp.n_read.iter() { - in_slots.push(SlotId::Node(*key)); + push_slot(in_slots, SlotId::Node(*key)); } for key in fp.n_write.iter() { - in_slots.push(SlotId::Node(*key)); - out_slots.push(SlotId::Node(*key)); + push_slot(in_slots, SlotId::Node(*key)); + push_slot(out_slots, SlotId::Node(*key)); } // Edges (warp-scoped EdgeKey) for key in fp.e_read.iter() { - in_slots.push(SlotId::Edge(*key)); + push_slot(in_slots, SlotId::Edge(*key)); } for key in fp.e_write.iter() { - in_slots.push(SlotId::Edge(*key)); - out_slots.push(SlotId::Edge(*key)); + push_slot(in_slots, SlotId::Edge(*key)); + push_slot(out_slots, SlotId::Edge(*key)); } for key in fp.r_read.iter() { - in_slots.push(SlotId::OutgoingEdges(*key)); + push_slot(in_slots, SlotId::OutgoingEdges(*key)); } for key in fp.r_write.iter() { - in_slots.push(SlotId::OutgoingEdges(*key)); - out_slots.push(SlotId::OutgoingEdges(*key)); + push_slot(in_slots, SlotId::OutgoingEdges(*key)); + push_slot(out_slots, SlotId::OutgoingEdges(*key)); } // Attachments (already warp-scoped via AttachmentKey) for key in fp.a_read.iter() { - in_slots.push(SlotId::Attachment(*key)); + push_slot(in_slots, SlotId::Attachment(*key)); } for key in fp.a_write.iter() { - in_slots.push(SlotId::Attachment(*key)); - out_slots.push(SlotId::Attachment(*key)); + push_slot(in_slots, SlotId::Attachment(*key)); + push_slot(out_slots, SlotId::Attachment(*key)); } // Ports (warp-scoped: (WarpId, PortKey)) for port_key in fp.b_in.keys() { - in_slots.push(SlotId::Port(*port_key)); + push_slot(in_slots, SlotId::Port(*port_key)); } for port_key in fp.b_out.keys() { - in_slots.push(SlotId::Port(*port_key)); - out_slots.push(SlotId::Port(*port_key)); + push_slot(in_slots, SlotId::Port(*port_key)); + push_slot(out_slots, SlotId::Port(*port_key)); } } +/// Below this many slots, a full vector simply grows. +const SLOT_COMPACTION_MIN: usize = 256; + +/// Adds `slot` to a tick's unsorted slot list. +/// +/// The list keeps repeats, because the patch sorts and dedups it once. When +/// the list is full, it first drops its repeats; if that frees less than +/// half of it, it doubles. So it holds at most about four times the tick's +/// distinct slots however much the footprints overlap, and each compaction +/// is paid for by at least half a list of pushes. The stable sort keeps the +/// already sorted prefix cheap to sort again. +fn push_slot(slots: &mut Vec, slot: SlotId) { + if slots.len() == slots.capacity() && slots.len() >= SLOT_COMPACTION_MIN { + slots.sort(); + slots.dedup(); + if slots.len() > slots.capacity() / 2 { + slots.reserve(slots.len()); + } + } + slots.push(slot); +} + fn runtime_ingress_kind_node_id(intent_kind: &IntentKind) -> NodeId { NodeId(*intent_kind.as_hash()) } diff --git a/crates/warp-core/src/tick_patch.rs b/crates/warp-core/src/tick_patch.rs index 1678dfef..51509dfd 100644 --- a/crates/warp-core/src/tick_patch.rs +++ b/crates/warp-core/src/tick_patch.rs @@ -402,6 +402,10 @@ impl WarpTickPatchV1 { in_slots.dedup(); out_slots.sort(); out_slots.dedup(); + // The engine hands over lists that held repeats; the patch outlives + // the tick, so it keeps only the room its slots need. + in_slots.shrink_to_fit(); + out_slots.shrink_to_fit(); let ops = { let mut keyed: Vec<(WarpOpKey, WarpOp)> = ops.into_iter().map(|op| (op.sort_key(), op)).collect(); From ae196933ffb2cad9b1b2b0126903700002cf6f18 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 22 Oct 2026 15:59:37 -0700 Subject: [PATCH 4/6] Read the tick's slots back from the scheduler's conflict sets Reservation already records every admitted footprint, once per key and in order, in the scheduler's conflict sets. The engine collected the same keys a second time into in_slots and out_slots. Read the slots from the conflict sets instead: each kind's read and written sets are merged in SlotId order, so the lists come out sorted and free of repeats with no further sort. This replaces the compacting slot vectors of 60bcf095, which bounded the peak-memory defect of 747f320c at some speed cost; the collection that held repeats is gone. Output ports gain their own set (ports_written), because the combined port set cannot tell b_in from b_out. The conflict sets are now taken at the end of the reserve phase instead of at finalize_tx. A commit whose rule panics leaves its transaction open, and the old sets then outlived it: committing the transaction again rejected any rewrite that overlapped the failed attempt, and reported InternalCorruption("scheduler rejected rewrite but no blockers were found"). The new witness a_failed_commit_leaves_no_reservations_behind fails that way on 60bcf095 and passes here; it also checks that the retry's slots exclude the failed attempt's. Co-Authored-By: Claude Opus 5.5 --- .../tests/commit_overlap_peak.rs | 19 +-- crates/warp-core/src/engine_impl.rs | 91 +------------- .../src/engine_impl/overlap_slot_tests.rs | 113 +++++++++++++++++- crates/warp-core/src/scheduler.rs | 108 +++++++++++++++++ crates/warp-core/src/tick_patch.rs | 4 +- 5 files changed, 231 insertions(+), 104 deletions(-) diff --git a/crates/echo-alloc-witness/tests/commit_overlap_peak.rs b/crates/echo-alloc-witness/tests/commit_overlap_peak.rs index dbfd1110..9422ecfa 100644 --- a/crates/echo-alloc-witness/tests/commit_overlap_peak.rs +++ b/crates/echo-alloc-witness/tests/commit_overlap_peak.rs @@ -3,12 +3,13 @@ //! Peak temporary memory of one commit whose rewrites claim heavily //! overlapping footprints. //! -//! Reservation collects the tick's `in_slots` and `out_slots` as plain -//! vectors that keep repeats until the patch sorts and dedups them, and the -//! patch digest gathers its whole input before hashing it. Both add -//! temporary memory. Here every rewrite reads the same `SHARED` nodes and -//! their attachments, so the repeats are as dense as footprints allow (reads -//! may overlap; writes may not). +//! Reservation reads the tick's `in_slots` and `out_slots` back from the +//! scheduler's conflict sets, and the patch digest gathers its whole input +//! before hashing it. Both add temporary memory. Here every rewrite reads +//! the same `SHARED` nodes and their attachments, so the footprints overlap +//! as densely as they are allowed to (reads may overlap; writes may not). +//! A slot collection that kept one entry per footprint claim would grow with +//! that overlap; this test bounds the peak. //! //! The allocator tracks live bytes (allocated minus freed) during the //! commit and records their peak above the level when the commit began. @@ -100,9 +101,9 @@ const REWRITES: usize = 4_096; /// /// The overlapping tick claims 131 read slots per rewrite instead of three, /// so the receipt and scheduler legitimately need more room: the -/// `BTreeSet`-based collection peaked at about 3.8 times the control. Slot -/// lists that keep every repeat and hand their full capacity to the patch -/// peaked at about 11 times. Compacting the lists keeps it near 4. +/// former second `BTreeSet` collection peaked at about 3.8 times the +/// control. Slot lists that kept every repeat and handed their full capacity +/// to the patch peaked at about 11 times. const BUDGET_FACTOR: isize = 6; fn shared(i: usize) -> NodeId { diff --git a/crates/warp-core/src/engine_impl.rs b/crates/warp-core/src/engine_impl.rs index 889490fc..510bd00c 100644 --- a/crates/warp-core/src/engine_impl.rs +++ b/crates/warp-core/src/engine_impl.rs @@ -541,7 +541,7 @@ pub struct Engine { struct ReserveOutcome { receipt: TickReceipt, reserved: Vec, - /// Unsorted, with repeats: the patch sorts and dedups them once. + /// Sorted, each slot once: read back from the scheduler's conflict sets. in_slots: Vec, out_slots: Vec, } @@ -2293,8 +2293,6 @@ impl Engine { ) -> Result { // Reserve phase: enforce independence against active frontier. let mut receipt_entries: Vec = Vec::with_capacity(drained.len()); - let mut in_slots: Vec = Vec::new(); - let mut out_slots: Vec = Vec::new(); let mut blocked_by: Vec> = Vec::with_capacity(drained.len()); let mut reserved: Vec = Vec::new(); let mut reserved_entry_indices: Vec = Vec::new(); @@ -2340,17 +2338,14 @@ impl Engine { }, }); if accepted { - extend_slots_from_footprint( - &mut in_slots, - &mut out_slots, - &rewrite.scope.warp_id, - &rewrite.footprint, - ); reserved.push(rewrite); reserved_entry_indices.push(entry_idx_u32); } blocked_by.push(blockers); } + // The scheduler's conflict sets already hold the union of the + // reserved footprints, once each; the tick's slots are read from them. + let (in_slots, out_slots) = self.scheduler.take_reserved_slots(tx); Ok(ReserveOutcome { receipt: TickReceipt::new(tx, receipt_entries, blocked_by), @@ -3429,84 +3424,6 @@ pub fn scope_hash(rule_id: &Hash, scope: &NodeKey) -> Hash { hasher.finalize().into() } -/// Extends the slot sets with resources from a warp-scoped footprint. -/// -/// Footprint sets are now warp-scoped: they contain full `NodeKey`, `EdgeKey`, -/// and `WarpScopedPortKey` values directly. The `_warp_id` parameter is kept -/// for call-site compatibility but is no longer used (the `warp_id` is embedded -/// in the footprint keys). -fn extend_slots_from_footprint( - in_slots: &mut Vec, - out_slots: &mut Vec, - _warp_id: &WarpId, // Kept for API compat; warp_id is now in the footprint keys - fp: &crate::footprint::Footprint, -) { - // Nodes (warp-scoped NodeKey) - for key in fp.n_read.iter() { - push_slot(in_slots, SlotId::Node(*key)); - } - for key in fp.n_write.iter() { - push_slot(in_slots, SlotId::Node(*key)); - push_slot(out_slots, SlotId::Node(*key)); - } - - // Edges (warp-scoped EdgeKey) - for key in fp.e_read.iter() { - push_slot(in_slots, SlotId::Edge(*key)); - } - for key in fp.e_write.iter() { - push_slot(in_slots, SlotId::Edge(*key)); - push_slot(out_slots, SlotId::Edge(*key)); - } - for key in fp.r_read.iter() { - push_slot(in_slots, SlotId::OutgoingEdges(*key)); - } - for key in fp.r_write.iter() { - push_slot(in_slots, SlotId::OutgoingEdges(*key)); - push_slot(out_slots, SlotId::OutgoingEdges(*key)); - } - - // Attachments (already warp-scoped via AttachmentKey) - for key in fp.a_read.iter() { - push_slot(in_slots, SlotId::Attachment(*key)); - } - for key in fp.a_write.iter() { - push_slot(in_slots, SlotId::Attachment(*key)); - push_slot(out_slots, SlotId::Attachment(*key)); - } - - // Ports (warp-scoped: (WarpId, PortKey)) - for port_key in fp.b_in.keys() { - push_slot(in_slots, SlotId::Port(*port_key)); - } - for port_key in fp.b_out.keys() { - push_slot(in_slots, SlotId::Port(*port_key)); - push_slot(out_slots, SlotId::Port(*port_key)); - } -} - -/// Below this many slots, a full vector simply grows. -const SLOT_COMPACTION_MIN: usize = 256; - -/// Adds `slot` to a tick's unsorted slot list. -/// -/// The list keeps repeats, because the patch sorts and dedups it once. When -/// the list is full, it first drops its repeats; if that frees less than -/// half of it, it doubles. So it holds at most about four times the tick's -/// distinct slots however much the footprints overlap, and each compaction -/// is paid for by at least half a list of pushes. The stable sort keeps the -/// already sorted prefix cheap to sort again. -fn push_slot(slots: &mut Vec, slot: SlotId) { - if slots.len() == slots.capacity() && slots.len() >= SLOT_COMPACTION_MIN { - slots.sort(); - slots.dedup(); - if slots.len() > slots.capacity() / 2 { - slots.reserve(slots.len()); - } - } - slots.push(slot); -} - fn runtime_ingress_kind_node_id(intent_kind: &IntentKind) -> NodeId { NodeId(*intent_kind.as_hash()) } diff --git a/crates/warp-core/src/engine_impl/overlap_slot_tests.rs b/crates/warp-core/src/engine_impl/overlap_slot_tests.rs index df828881..a03bcbd9 100644 --- a/crates/warp-core/src/engine_impl/overlap_slot_tests.rs +++ b/crates/warp-core/src/engine_impl/overlap_slot_tests.rs @@ -1,11 +1,12 @@ // SPDX-License-Identifier: Apache-2.0 // © James Ross Ω FLYING•ROBOTS -#![allow(clippy::expect_used)] // Test assertions may abort the witness. -//! Reservation collects a tick's slots as vectors with repeats. This witness -//! commits rewrites whose footprints overlap, both across rewrites (shared -//! reads) and within one footprint (a slot that is read and written), and -//! checks that the committed patch carries exactly the sorted, repeat-free -//! union that the former `BTreeSet` collection produced. +#![allow(clippy::expect_used, clippy::panic)] // Test assertions may abort the witness; one rule panics on purpose. +//! Reservation reads a tick's slots back from the scheduler's conflict sets. +//! This witness commits rewrites whose footprints overlap, both across +//! rewrites (shared reads) and within one footprint (a slot that is read and +//! written), and checks that the committed patch carries exactly the sorted, +//! repeat-free union that the former per-footprint `BTreeSet` collection +//! produced. use std::collections::BTreeSet; use super::*; @@ -158,3 +159,103 @@ fn overlapping_footprints_commit_the_same_sorted_slot_sets() { assert!(patch.out_slots().windows(2).all(|w| w[0] < w[1])); assert_eq!(patch.ops().len(), own.len()); } + +const FAILING_RULE: &str = "test/overlap-fails"; +const RETRY_RULE: &str = "test/overlap-retry"; + +fn retry_target() -> NodeId { + make_node_id("test/overlap-retry-target") +} + +fn failed_only() -> NodeId { + make_node_id("test/overlap-failed-only") +} + +fn writes(view: GraphView<'_>, nodes: &[NodeId]) -> Footprint { + let mut footprint = Footprint::default(); + for node in nodes { + let key = NodeKey { + warp_id: view.warp_id(), + local_id: *node, + }; + footprint.n_read.insert(key); + footprint.n_write.insert(key); + footprint.a_read.insert(AttachmentKey::node_alpha(key)); + footprint.a_write.insert(AttachmentKey::node_alpha(key)); + } + footprint +} + +fn failing_rule() -> RewriteRule { + RewriteRule { + id: [8; 32], + name: FAILING_RULE, + left: PatternGraph { nodes: vec![] }, + matcher: |_, _| true, + executor: RuleExecutor::legacy(|_, _, _| panic!("rule fails on purpose")), + compute_footprint: |view, _| writes(view, &[retry_target(), failed_only()]), + factor_mask: 0, + conflict_policy: ConflictPolicy::Abort, + join_fn: None, + } +} + +fn retry_rule() -> RewriteRule { + RewriteRule { + id: [9; 32], + name: RETRY_RULE, + left: PatternGraph { nodes: vec![] }, + matcher: |_, _| true, + executor: RuleExecutor::legacy(executor), + compute_footprint: |view, _| writes(view, &[retry_target()]), + factor_mask: 0, + conflict_policy: ConflictPolicy::Abort, + join_fn: None, + } +} + +/// A commit whose rule panics leaves its transaction open. Committing that +/// transaction again must judge and record only the new attempt: the failed +/// attempt's reservations are neither conflicts nor slots of the new tick. +#[test] +fn a_failed_commit_leaves_no_reservations_behind() { + let mut store = GraphStore::default(); + let ty = make_type_id("test/overlap-node"); + for node in [retry_target(), failed_only()] { + store.insert_node(node, NodeRecord { ty }); + } + let mut engine = Engine::new(store, retry_target()); + engine.register_rule(failing_rule()).expect("rule registers"); + engine.register_rule(retry_rule()).expect("rule registers"); + let tx = engine.begin(); + assert!(matches!( + engine.apply(tx, FAILING_RULE, &retry_target()), + Ok(ApplyResult::Applied) + )); + let failed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + engine.commit_with_receipt(tx) + })); + assert!(failed.is_err(), "the failing rule must panic out of commit"); + + assert!(matches!( + engine.apply(tx, RETRY_RULE, &retry_target()), + Ok(ApplyResult::Applied) + )); + let (_, receipt, patch) = engine + .commit_with_receipt(tx) + .expect("the retry is judged on its own reservations"); + assert!(receipt + .entries() + .iter() + .all(|entry| entry.disposition == TickReceiptDisposition::Applied)); + let key = NodeKey { + warp_id: engine.current_root.warp_id, + local_id: retry_target(), + }; + let expected = [ + SlotId::Node(key), + SlotId::Attachment(AttachmentKey::node_alpha(key)), + ]; + assert_eq!(patch.in_slots(), expected.as_slice()); + assert_eq!(patch.out_slots(), expected.as_slice()); +} diff --git a/crates/warp-core/src/scheduler.rs b/crates/warp-core/src/scheduler.rs index 41724ab7..4e2b4e4f 100644 --- a/crates/warp-core/src/scheduler.rs +++ b/crates/warp-core/src/scheduler.rs @@ -17,6 +17,7 @@ use std::sync::Arc; use crate::footprint::{Footprint, OutgoingEdgeRange, WarpScopedPortKey}; use crate::ident::{CompactRuleId, EdgeKey, Hash, NodeKey}; use crate::telemetry::TelemetrySink; +use crate::tick_patch::SlotId; use crate::tick_delta::OpOrigin; use crate::tx::TxId; @@ -50,6 +51,9 @@ pub(crate) struct ActiveFootprints { attachments_read: GenSet, /// Boundary ports touched - warp-scoped (both `b_in` and `b_out`, since any intersection conflicts) ports: GenSet, + /// Boundary ports written (`b_out`), kept apart from `ports` only so + /// that the tick's written slots can be read back from these sets. + ports_written: GenSet, } impl ActiveFootprints { @@ -64,6 +68,81 @@ impl ActiveFootprints { attachments_written: GenSet::new(), attachments_read: GenSet::new(), ports: GenSet::new(), + ports_written: GenSet::new(), + } + } + + /// The tick's slots: everything read or written (`in_slots`) and + /// everything written (`out_slots`), each sorted and free of repeats. + /// + /// These sets already hold exactly the union of the reserved footprints, + /// once each, so this replaces collecting the same keys a second time. + /// [`SlotId`] orders by kind (node, edge, attachment, port, outgoing + /// edges) and then by key, and each set iterates its keys in order, so + /// visiting the kinds in that order produces sorted lists directly. + fn slots(&self) -> (Vec, Vec) { + let mut in_slots = Vec::new(); + union_into( + &mut in_slots, + &self.nodes_read, + &self.nodes_written, + SlotId::Node, + ); + union_into( + &mut in_slots, + &self.edges_read, + &self.edges_written, + SlotId::Edge, + ); + union_into( + &mut in_slots, + &self.attachments_read, + &self.attachments_written, + SlotId::Attachment, + ); + in_slots.extend(self.ports.keys().map(SlotId::Port)); + union_into( + &mut in_slots, + &self.ranges_read, + &self.ranges_written, + SlotId::OutgoingEdges, + ); + + let mut out_slots = Vec::new(); + out_slots.extend(self.nodes_written.keys().map(SlotId::Node)); + out_slots.extend(self.edges_written.keys().map(SlotId::Edge)); + out_slots.extend(self.attachments_written.keys().map(SlotId::Attachment)); + out_slots.extend(self.ports_written.keys().map(SlotId::Port)); + out_slots.extend(self.ranges_written.keys().map(SlotId::OutgoingEdges)); + (in_slots, out_slots) + } +} + +/// Appends the sorted union of two sorted key sets, each key once. +fn union_into( + out: &mut Vec, + read: &GenSet, + written: &GenSet, + slot: fn(K) -> SlotId, +) { + let mut read = read.keys().peekable(); + let mut written = written.keys().peekable(); + loop { + let next = match (read.peek(), written.peek()) { + (Some(r), Some(w)) => match r.cmp(w) { + Ordering::Less => read.next(), + Ordering::Greater => written.next(), + Ordering::Equal => { + written.next(); + read.next() + } + }, + (Some(_), None) => read.next(), + (None, Some(_)) => written.next(), + (None, None) => break, + }; + if let Some(key) = next { + out.push(slot(key)); } } } @@ -280,9 +359,23 @@ impl RadixScheduler { } for port_key in pr.footprint.b_out.keys() { active.ports.mark(*port_key); + active.ports_written.mark(*port_key); } } + /// Ends the reservation phase for `tx` and returns its slots (see + /// [`ActiveFootprints::slots`]). + /// + /// The conflict sets are dropped here rather than at [`Self::finalize_tx`], + /// so they hold only this phase's reservations: if the commit then fails + /// and the transaction commits again, its slots and its conflict checks + /// start clean instead of including the failed attempt. + pub(crate) fn take_reserved_slots(&mut self, tx: TxId) -> (Vec, Vec) { + self.active + .remove(&tx) + .map_or_else(|| (Vec::new(), Vec::new()), |active| active.slots()) + } + /// Finalizes accounting for `tx`: clears internal state. pub(crate) fn finalize_tx(&mut self, tx: TxId) { self.active.remove(&tx); @@ -540,6 +633,14 @@ impl GenSet { pub fn mark(&mut self, key: K) { self.seen.insert(key, self.gen); } + + /// Keys marked in the current generation, in ascending order. + pub fn keys(&self) -> impl Iterator + '_ { + self.seen + .iter() + .filter(|(_, &g)| g == self.gen) + .map(|(&key, _)| key) + } } // ============================================================================ @@ -640,6 +741,13 @@ impl DeterministicScheduler { accepted } + /// Ends the reservation phase for `tx` and returns its slots. + pub(crate) fn take_reserved_slots(&mut self, tx: TxId) -> (Vec, Vec) { + match &mut self.inner { + SchedulerImpl::Radix(s) => s.take_reserved_slots(tx), + } + } + pub(crate) fn finalize_tx(&mut self, tx: TxId) { // Emit summary telemetry before clearing state if let Some((reserved, conflict)) = self.counters.remove(&tx) { diff --git a/crates/warp-core/src/tick_patch.rs b/crates/warp-core/src/tick_patch.rs index 51509dfd..a717ea67 100644 --- a/crates/warp-core/src/tick_patch.rs +++ b/crates/warp-core/src/tick_patch.rs @@ -402,8 +402,8 @@ impl WarpTickPatchV1 { in_slots.dedup(); out_slots.sort(); out_slots.dedup(); - // The engine hands over lists that held repeats; the patch outlives - // the tick, so it keeps only the room its slots need. + // Callers may hand over lists with repeats or spare room; the patch + // outlives the tick, so it keeps only the room its slots need. in_slots.shrink_to_fit(); out_slots.shrink_to_fit(); let ops = { From ca8c3a4826e1649f8fc71ec4fc8882b077ff784b Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 22 Oct 2026 16:02:47 -0700 Subject: [PATCH 5/6] Format, and record the failed-commit retry fix rustfmt orders the new SlotId import and wraps one test line. The CHANGELOG records the behaviour change of ae196933: a transaction committed again after a panicked commit is judged on its own reservations. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 8 ++++++++ crates/warp-core/src/engine_impl/overlap_slot_tests.rs | 4 +++- crates/warp-core/src/scheduler.rs | 2 +- 3 files changed, 12 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 928e9360..7279e4c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2056,6 +2056,14 @@ Applied, Rejected, Obstructed}` with receipt evidence and typed contract ### Fixed +- Committing a transaction again after a commit whose rule panicked no + longer fails with `InternalCorruption("scheduler rejected rewrite but no + blockers were found")`. The scheduler kept the failed attempt's + reservations until the transaction finished, so they rejected any new + rewrite that overlapped them. Reservations now end with the commit's + reserve phase, and the tick's `in_slots` and `out_slots` are read from + them, so neither includes the failed attempt. + - Completed shadow application-root benchmark rounds are invalid when their recorded incremental/rebuild or live/replay parity is missing or failed, including builds without `delta_validate`. diff --git a/crates/warp-core/src/engine_impl/overlap_slot_tests.rs b/crates/warp-core/src/engine_impl/overlap_slot_tests.rs index a03bcbd9..bddd52bc 100644 --- a/crates/warp-core/src/engine_impl/overlap_slot_tests.rs +++ b/crates/warp-core/src/engine_impl/overlap_slot_tests.rs @@ -225,7 +225,9 @@ fn a_failed_commit_leaves_no_reservations_behind() { store.insert_node(node, NodeRecord { ty }); } let mut engine = Engine::new(store, retry_target()); - engine.register_rule(failing_rule()).expect("rule registers"); + engine + .register_rule(failing_rule()) + .expect("rule registers"); engine.register_rule(retry_rule()).expect("rule registers"); let tx = engine.begin(); assert!(matches!( diff --git a/crates/warp-core/src/scheduler.rs b/crates/warp-core/src/scheduler.rs index 4e2b4e4f..a7427fbf 100644 --- a/crates/warp-core/src/scheduler.rs +++ b/crates/warp-core/src/scheduler.rs @@ -17,8 +17,8 @@ use std::sync::Arc; use crate::footprint::{Footprint, OutgoingEdgeRange, WarpScopedPortKey}; use crate::ident::{CompactRuleId, EdgeKey, Hash, NodeKey}; use crate::telemetry::TelemetrySink; -use crate::tick_patch::SlotId; use crate::tick_delta::OpOrigin; +use crate::tick_patch::SlotId; use crate::tx::TxId; #[cfg(test)] From 9b4466889b665bf29db3b23c6462bf153b9dbb6f Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 22 Oct 2026 17:14:53 -0700 Subject: [PATCH 6/6] Close a transaction whose commit fails Reusing a transaction after a failed commit is unsupported, so it now fails closed. commit_with_receipt runs the commit under catch_unwind; if it returns an error or a rule panics, the transaction leaves live_txs and the scheduler drops its reservations and pending queue before the error returns or the panic resumes. Applying to the transaction or committing it again then returns EngineError::UnknownTx, whose documentation already covers a closed transaction. This replaces ae196933's behaviour, which let the same transaction commit again with fresh reservations. On main, that path failed with InternalCorruption whenever a new rewrite overlapped the failed attempt. The witness a_failed_commit_closes_its_transaction fails on ca8c3a48's engine (applying to the failed transaction succeeds there) and passes here; it also checks that a new transaction's slots exclude the failed attempt's. Only the panic path is witnessed: commit's error exits before execution end in debug_assert!(false), so a test build panics on them instead. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 15 +++---- crates/warp-core/src/engine_impl.rs | 39 ++++++++++++++++++- .../src/engine_impl/overlap_slot_tests.rs | 27 +++++++++---- crates/warp-core/src/scheduler.rs | 6 +-- 4 files changed, 66 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7279e4c4..f6a0c27a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2056,13 +2056,14 @@ Applied, Rejected, Obstructed}` with receipt evidence and typed contract ### Fixed -- Committing a transaction again after a commit whose rule panicked no - longer fails with `InternalCorruption("scheduler rejected rewrite but no - blockers were found")`. The scheduler kept the failed attempt's - reservations until the transaction finished, so they rejected any new - rewrite that overlapped them. Reservations now end with the commit's - reserve phase, and the tick's `in_slots` and `out_slots` are read from - them, so neither includes the failed attempt. +- A commit that fails, by returning an error or because a rule panics, now + closes its transaction; applying to it or committing it again returns + `EngineError::UnknownTx`. Before, the transaction stayed open with the + failed attempt's reservations still held, so committing it again rejected + any rewrite that overlapped them and failed with + `InternalCorruption("scheduler rejected rewrite but no blockers were + found")`. Reusing a transaction after a failed commit is unsupported, so + it is now refused. - Completed shadow application-root benchmark rounds are invalid when their recorded incremental/rebuild or live/replay parity is missing or failed, including diff --git a/crates/warp-core/src/engine_impl.rs b/crates/warp-core/src/engine_impl.rs index 510bd00c..b9101d08 100644 --- a/crates/warp-core/src/engine_impl.rs +++ b/crates/warp-core/src/engine_impl.rs @@ -2062,15 +2062,24 @@ impl Engine { /// This method also produces a delta tick patch (Paper III): a replayable boundary artifact /// whose digest is committed into the v2 commit hash. /// + /// A commit that fails closes its transaction. Whether it returns an error + /// or a rule panics out of it, `tx` is no longer live afterwards: applying + /// to it or committing it again returns [`EngineError::UnknownTx`], and new + /// work needs a new transaction. Committing a transaction again after a + /// failed attempt is unsupported, so it is refused rather than judged + /// against what the failed attempt left behind. + /// /// # Errors - /// - Returns [`EngineError::UnknownTx`] if `tx` does not refer to a live transaction. + /// - Returns [`EngineError::UnknownTx`] if `tx` does not refer to a live + /// transaction, including one whose earlier commit failed. /// - Returns [`EngineError::InternalCorruption`] if internal rule tables are /// corrupted (e.g., a reserved rewrite references a missing rule). /// /// # Panics /// /// Panics if delta validation is enabled and the `SnapshotAccumulator` produces - /// a different `state_root` than the legacy computation. + /// a different `state_root` than the legacy computation, and propagates a + /// panic raised by a rule. Either way the transaction is closed first. pub fn commit_with_receipt( &mut self, tx: TxId, @@ -2078,6 +2087,32 @@ impl Engine { if tx.value() == 0 || !self.live_txs.contains(&tx.value()) { return Err(EngineError::UnknownTx); } + let outcome = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| self.commit_live_tx(tx))); + match outcome { + Ok(Ok(committed)) => Ok(committed), + Ok(Err(err)) => { + self.close_failed_tx(tx); + Err(err) + } + Err(panic) => { + self.close_failed_tx(tx); + std::panic::resume_unwind(panic) + } + } + } + + /// Closes a transaction whose commit failed, releasing its reservations. + fn close_failed_tx(&mut self, tx: TxId) { + self.live_txs.remove(&tx.value()); + self.scheduler.finalize_tx(tx); + } + + /// The body of [`Self::commit_with_receipt`] for a transaction known to be live. + fn commit_live_tx( + &mut self, + tx: TxId, + ) -> Result<(Snapshot, TickReceipt, WarpTickPatchV1), EngineError> { self.last_execution_footprints = std::sync::Arc::default(); self.last_aborted_execution_footprints = std::sync::Arc::default(); let policy_id = self.policy_id; diff --git a/crates/warp-core/src/engine_impl/overlap_slot_tests.rs b/crates/warp-core/src/engine_impl/overlap_slot_tests.rs index bddd52bc..c759abef 100644 --- a/crates/warp-core/src/engine_impl/overlap_slot_tests.rs +++ b/crates/warp-core/src/engine_impl/overlap_slot_tests.rs @@ -214,11 +214,12 @@ fn retry_rule() -> RewriteRule { } } -/// A commit whose rule panics leaves its transaction open. Committing that -/// transaction again must judge and record only the new attempt: the failed -/// attempt's reservations are neither conflicts nor slots of the new tick. +/// A commit whose rule panics closes its transaction. Applying to it or +/// committing it again is refused with `UnknownTx`, and a new transaction is +/// judged only on its own reservations: the failed attempt leaves neither +/// conflicts nor slots behind. #[test] -fn a_failed_commit_leaves_no_reservations_behind() { +fn a_failed_commit_closes_its_transaction() { let mut store = GraphStore::default(); let ty = make_type_id("test/overlap-node"); for node in [retry_target(), failed_only()] { @@ -229,23 +230,33 @@ fn a_failed_commit_leaves_no_reservations_behind() { .register_rule(failing_rule()) .expect("rule registers"); engine.register_rule(retry_rule()).expect("rule registers"); - let tx = engine.begin(); + let failed_tx = engine.begin(); assert!(matches!( - engine.apply(tx, FAILING_RULE, &retry_target()), + engine.apply(failed_tx, FAILING_RULE, &retry_target()), Ok(ApplyResult::Applied) )); let failed = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - engine.commit_with_receipt(tx) + engine.commit_with_receipt(failed_tx) })); assert!(failed.is_err(), "the failing rule must panic out of commit"); + assert!(matches!( + engine.apply(failed_tx, RETRY_RULE, &retry_target()), + Err(EngineError::UnknownTx) + )); + assert!(matches!( + engine.commit_with_receipt(failed_tx), + Err(EngineError::UnknownTx) + )); + + let tx = engine.begin(); assert!(matches!( engine.apply(tx, RETRY_RULE, &retry_target()), Ok(ApplyResult::Applied) )); let (_, receipt, patch) = engine .commit_with_receipt(tx) - .expect("the retry is judged on its own reservations"); + .expect("a new transaction is judged on its own reservations"); assert!(receipt .entries() .iter() diff --git a/crates/warp-core/src/scheduler.rs b/crates/warp-core/src/scheduler.rs index a7427fbf..1a77e6b8 100644 --- a/crates/warp-core/src/scheduler.rs +++ b/crates/warp-core/src/scheduler.rs @@ -366,10 +366,8 @@ impl RadixScheduler { /// Ends the reservation phase for `tx` and returns its slots (see /// [`ActiveFootprints::slots`]). /// - /// The conflict sets are dropped here rather than at [`Self::finalize_tx`], - /// so they hold only this phase's reservations: if the commit then fails - /// and the transaction commits again, its slots and its conflict checks - /// start clean instead of including the failed attempt. + /// Nothing reads the conflict sets after the reservation phase, so they + /// are taken here rather than kept until [`Self::finalize_tx`]. pub(crate) fn take_reserved_slots(&mut self, tx: TxId) -> (Vec, Vec) { self.active .remove(&tx)