From 961eabc919deee05617d12995b417b2904e5956f Mon Sep 17 00:00:00 2001 From: James Ross Date: Wed, 14 Oct 2026 01:45:48 -0700 Subject: [PATCH 1/2] warp-core: one canonical delta merge for every build The engine shipped its own merge (sort by key, compare neighbours, dedupe, then check writes to new warps) when neither test nor delta_validate was on, while merge_deltas served test and validation builds. The merge tests ran against merge_deltas, so the shipped merge was not the tested one. merge_canonical in parallel/merge.rs now holds the only implementation, generic over the writer identity: OpOrigin where builds record origins, () where they do not. merge_deltas and the shipped engine path both call it. The release-only helpers find_write_to_new_warp and check_write_to_new_warp are removed. Unit tests run the core both ways on the same inputs and check they agree. CI now runs the six delta_validate merge test targets, which no job ran before. Fixes #813. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 9 + crates/warp-core/src/engine_impl.rs | 43 ++-- crates/warp-core/src/parallel/merge.rs | 273 ++++++++++++++++++------- crates/warp-core/src/parallel/mod.rs | 4 +- 4 files changed, 222 insertions(+), 107 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 35354c070..c557c03d8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -400,6 +400,15 @@ jobs: cargo check -p warp-core --features unsafe_graph --all-targets - name: cargo test (warp-wasm engine kernel) run: cargo test -p warp-wasm --features engine + - name: cargo test (warp-core canonical merge, delta_validate) + run: | + cargo test -p warp-core --features delta_validate \ + --test parallel_merge_tripwire \ + --test parallel_merge_warpopkey \ + --test parallel_openportal_rules \ + --test parallel_parallel_exec \ + --test parallel_engine_integration_multiwarp \ + --test parallel_stress_multiwarp - name: cargo test --doc (warp-core) run: cargo test -p warp-core --doc - name: PRNG golden regression (warp-math) diff --git a/crates/warp-core/src/engine_impl.rs b/crates/warp-core/src/engine_impl.rs index 0a414e398..9bcc042de 100644 --- a/crates/warp-core/src/engine_impl.rs +++ b/crates/warp-core/src/engine_impl.rs @@ -2923,8 +2923,8 @@ impl Engine { /// Merges parallel execution deltas into a canonical op sequence. /// -/// With `delta_validate`: uses `merge_deltas` for conflict detection. -/// Without: flattens, sorts by `sort_key`, deduplicates, and rejects conflicts. +/// Every build runs the canonical merge in `parallel::merge`. Builds with +/// `test` or `delta_validate` reach it through `merge_deltas`, which also names the writers. /// /// # Errors /// @@ -2974,6 +2974,8 @@ fn merge_parallel_deltas( fn merge_worker_deltas( all_deltas: Vec>, ) -> Result, EngineError> { + // Builds that record origins merge through `merge_deltas`, which keeps the + // writers for diagnostics. Both paths run the same canonical merge. #[cfg(any(test, feature = "delta_validate"))] { merge_deltas(all_deltas).map_err(|conflict| { @@ -2987,42 +2989,23 @@ fn merge_worker_deltas( #[cfg(not(any(test, feature = "delta_validate")))] { - // Without delta_validate, flatten and sort by sort_key for determinism. - // Ops with the same sort_key are deduplicated (footprint ensures they're identical). - let mut flat: Vec<_> = all_deltas + let flat: Vec<_> = all_deltas .into_iter() .map(|delta| match delta { Ok(delta) => delta, Err(poisoned) => std::panic::resume_unwind(poisoned.into_panic()), }) .flat_map(crate::TickDelta::into_ops_unsorted) - .map(|op| (op.sort_key(), op)) + .map(|op| (op.sort_key(), (), op)) .collect(); - - // Sort by sort_key for canonical order. - // Use unstable sort for efficiency; equal keys become consecutive for dedup. - flat.sort_unstable_by_key(|a| a.0); - - // Reject conflicting ops with same sort_key. - for w in flat.windows(2) { - if w[0].0 == w[1].0 && w[0].1 != w[1].1 { - return Err(EngineError::InternalCorruption( - "merge_parallel_deltas: conflicting ops share sort_key", - )); + crate::parallel::merge_canonical(flat).map_err(|fault| match fault { + crate::parallel::MergeFault::Conflict { .. } => EngineError::InternalCorruption( + "merge_parallel_deltas: conflicting ops share sort_key", + ), + crate::parallel::MergeFault::WriteToNewWarp { .. } => { + EngineError::InternalCorruption("merge_parallel_deltas: write to new warp") } - } - - flat.dedup_by(|a, b| a.0 == b.0); - let ops: Vec<_> = flat.into_iter().map(|(_, op)| op).collect(); - - // Validate no writes to warps created in the same tick. - if crate::parallel::check_write_to_new_warp(&ops).is_some() { - return Err(EngineError::InternalCorruption( - "merge_parallel_deltas: write to new warp", - )); - } - - Ok(ops) + }) } } diff --git a/crates/warp-core/src/parallel/merge.rs b/crates/warp-core/src/parallel/merge.rs index f7e043c94..5af7cf664 100644 --- a/crates/warp-core/src/parallel/merge.rs +++ b/crates/warp-core/src/parallel/merge.rs @@ -73,7 +73,7 @@ pub struct MergeConflict { pub fn merge_deltas( deltas: Vec>, ) -> Result, MergeError> { - let mut flat: Vec<(crate::tick_patch::WarpOpKey, OpOrigin, WarpOp)> = Vec::new(); + let mut flat: Vec<(WarpOpKey, OpOrigin, WarpOp)> = Vec::new(); for d in deltas { let d = match d { @@ -87,17 +87,67 @@ pub fn merge_deltas( } } - // Sort by (WarpOpKey, OpOrigin) - both are Ord - flat.sort_by(|a, b| (&a.0, &a.1).cmp(&(&b.0, &b.1))); + merge_canonical(flat).map_err(|fault| match fault { + MergeFault::Conflict { key, writers } => { + MergeError::Conflict(Box::new(MergeConflict { key, writers })) + } + MergeFault::WriteToNewWarp { + warp_id, + origin, + op_kind, + } => MergeError::WriteToNewWarp { + warp_id, + op_origin: origin, + op_kind, + }, + }) +} + +/// Why [`merge_canonical`] refused a set of operations. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum MergeFault { + /// Operations that share a key differ. + Conflict { + /// The shared key. + key: WarpOpKey, + /// Every writer of that key, in canonical order. + writers: Vec, + }, + /// An operation targets a warp that an `OpenPortal` creates in the same tick. + WriteToNewWarp { + /// The newly created warp. + warp_id: WarpId, + /// The writer of the violating operation. + origin: O, + /// The kind of the violating operation. + op_kind: &'static str, + }, +} + +/// The canonical merge, shared by the engine and [`merge_deltas`]. +/// +/// `O` identifies an operation's writer. Builds that record origins +/// (`test` or `delta_validate`) pass [`OpOrigin`]; other builds pass `()`. +/// The origin orders operations that share a key and names the writers in a +/// refusal; it never changes which operations are accepted or their output +/// order, because only identical operations survive under one key. +/// +/// 1. Sort by (key, origin). +/// 2. Refuse any write to a warp created in the same tick by +/// `OpenPortal { init: Empty { .. }, .. }`. +/// 3. Keep one operation per key when all are identical; refuse otherwise. +pub(crate) fn merge_canonical( + mut flat: Vec<(WarpOpKey, O, WarpOp)>, +) -> Result, MergeFault> { + flat.sort_unstable_by(|a, b| (&a.0, &a.1).cmp(&(&b.0, &b.1))); - // Collect newly created warps and validate no same-tick writes to them. let new_warps = collect_new_warps(flat.iter().map(|(_, _, op)| op)); for (_, origin, op) in &flat { - if let Some((target_warp, op_kind)) = extract_target_warp(op) { - if new_warps.contains(&target_warp) { - return Err(MergeError::WriteToNewWarp { - warp_id: target_warp, - op_origin: *origin, + if let Some((warp_id, op_kind)) = extract_target_warp(op) { + if new_warps.contains(&warp_id) { + return Err(MergeFault::WriteToNewWarp { + warp_id, + origin: *origin, op_kind, }); } @@ -106,31 +156,22 @@ pub fn merge_deltas( let mut out = Vec::with_capacity(flat.len()); let mut i = 0; - while i < flat.len() { let key = flat[i].0; let start = i; - - // Find all ops with same WarpOpKey while i < flat.len() && flat[i].0 == key { i += 1; } - - // Check if all ops in group are identical let first = &flat[start].2; - let all_same = flat[start + 1..i].iter().all(|(_, _, op)| op == first); - - if all_same { + if flat[start + 1..i].iter().all(|(_, _, op)| op == first) { out.push(first.clone()); } else { - let writers = flat[start..i].iter().map(|(_, o, _)| *o).collect(); - return Err(MergeError::Conflict(Box::new(MergeConflict { + return Err(MergeFault::Conflict { key, - writers, - }))); + writers: flat[start..i].iter().map(|(_, o, _)| *o).collect(), + }); } } - Ok(out) } @@ -161,24 +202,6 @@ pub(crate) fn collect_new_warps<'a>(ops: impl IntoIterator) - .collect() } -/// Finds the first operation that writes to a newly created warp. -/// -/// Returns `Some((warp_id, op_kind))` on the first violation found, `None` if valid. -#[cfg(not(any(test, feature = "delta_validate")))] -pub(crate) fn find_write_to_new_warp<'a>( - ops: impl IntoIterator, - new_warps: &BTreeSet, -) -> Option<(WarpId, &'static str)> { - for op in ops { - if let Some((target_warp, op_kind)) = extract_target_warp(op) { - if new_warps.contains(&target_warp) { - return Some((target_warp, op_kind)); - } - } - } - None -} - /// Extracts the target warp from an operation, if applicable. /// /// Returns `None` for `OpenPortal` (which creates the warp, not writes to it). @@ -206,39 +229,139 @@ pub(crate) fn extract_target_warp(op: &WarpOp) -> Option<(WarpId, &'static str)> } } -/// Validates that no operation writes to a warp created in the same tick. -/// -/// This function provides a lightweight, non-`delta_validate` check for the same-tick -/// write invariant that [`merge_deltas`] enforces in test/validation builds. Use this -/// when you have a finalized op slice and need to verify the new-warp write rule without -/// the full merge machinery. -/// -/// # Preconditions -/// -/// - `ops` should be a complete set of operations for a single tick. -/// - Operations are not required to be sorted; the function scans linearly. -/// -/// # Postconditions -/// -/// Returns `None` if all operations respect the new-warp write rule (i.e., no op -/// targets a warp that is being created via `OpenPortal` with `PortalInit::Empty` -/// in the same tick). -/// -/// Returns `Some((warp_id, op_kind))` on the first violation found, identifying -/// the offending warp and operation type. -/// -/// # When to use this vs [`merge_deltas`] -/// -/// - Use `check_write_to_new_warp` for fast validation of a finalized op slice in -/// release builds where `delta_validate` is disabled. -/// - Use [`merge_deltas`] when you need full conflict detection, origin tracking, -/// and canonical merge ordering (test/validation builds). -/// -/// # Panics -/// -/// This function does not panic. -#[cfg(not(any(test, feature = "delta_validate")))] -pub(crate) fn check_write_to_new_warp(ops: &[WarpOp]) -> Option<(WarpId, &'static str)> { - let new_warps = collect_new_warps(ops); - find_write_to_new_warp(ops, &new_warps) +#[cfg(test)] +#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)] +mod tests { + use super::{merge_canonical, MergeFault}; + use crate::attachment::{AtomPayload, AttachmentKey, AttachmentValue}; + use crate::ident::{make_node_id, make_type_id, make_warp_id, NodeKey}; + use crate::record::NodeRecord; + use crate::tick_delta::OpOrigin; + use crate::tick_patch::{PortalInit, WarpOp}; + + fn origin(intent_id: u64) -> OpOrigin { + OpOrigin { + intent_id, + rule_id: 0, + match_ix: 0, + op_ix: 0, + } + } + + fn set(node: &str, value: &[u8]) -> WarpOp { + WarpOp::SetAttachment { + key: AttachmentKey::node_alpha(NodeKey { + warp_id: make_warp_id("merge/w"), + local_id: make_node_id(node), + }), + value: Some(AttachmentValue::Atom(AtomPayload::new( + make_type_id("merge/atom"), + value.to_vec().into(), + ))), + } + } + + fn open_child() -> WarpOp { + WarpOp::OpenPortal { + key: AttachmentKey::node_alpha(NodeKey { + warp_id: make_warp_id("merge/w"), + local_id: make_node_id("merge/parent"), + }), + child_warp: make_warp_id("merge/child"), + child_root: make_node_id("merge/child-root"), + init: PortalInit::Empty { + root_record: NodeRecord { + ty: make_type_id("merge/root"), + }, + }, + } + } + + fn upsert_in_child() -> WarpOp { + WarpOp::UpsertNode { + node: NodeKey { + warp_id: make_warp_id("merge/child"), + local_id: make_node_id("merge/new"), + }, + record: NodeRecord { + ty: make_type_id("merge/node"), + }, + } + } + + /// Runs the merge as the shipped build does (no origins) and as a + /// validation build does (with origins), and checks they agree. + fn both(ops: &[WarpOp]) -> Result, MergeFault<()>> { + let shipped = merge_canonical( + ops.iter() + .map(|op| (op.sort_key(), (), op.clone())) + .collect(), + ); + let validated = merge_canonical( + ops.iter() + .enumerate() + .map(|(i, op)| (op.sort_key(), origin(i as u64), op.clone())) + .collect(), + ); + match (&shipped, validated) { + (Ok(a), Ok(b)) => assert_eq!(a, &b, "shipped and validated outputs differ"), + ( + Err(MergeFault::Conflict { key: a, .. }), + Err(MergeFault::Conflict { key: b, .. }), + ) => { + assert_eq!(a, &b); + } + ( + Err(MergeFault::WriteToNewWarp { + warp_id: a, + op_kind: ka, + .. + }), + Err(MergeFault::WriteToNewWarp { + warp_id: b, + op_kind: kb, + .. + }), + ) => assert_eq!((a, ka), (&b, &kb)), + (a, b) => panic!("shipped {a:?} and validated {b:?} disagree"), + } + shipped + } + + #[test] + fn identical_writes_collapse_in_canonical_order() { + let ops = [ + set("b", b"2"), + set("a", b"1"), + set("b", b"2"), + set("a", b"1"), + ]; + let merged = both(&ops).expect("identical writes merge"); + let mut expected = vec![set("a", b"1"), set("b", b"2")]; + expected.sort_by_key(WarpOp::sort_key); + assert_eq!(merged, expected); + } + + #[test] + fn divergent_writes_to_one_key_are_refused() { + let ops = [set("a", b"1"), set("c", b"3"), set("a", b"other")]; + assert!(matches!(both(&ops), Err(MergeFault::Conflict { .. }))); + } + + #[test] + fn a_write_to_a_warp_opened_in_the_same_tick_is_refused() { + let ops = [upsert_in_child(), open_child()]; + assert!(matches!( + both(&ops), + Err(MergeFault::WriteToNewWarp { + op_kind: "UpsertNode", + .. + }) + )); + } + + #[test] + fn opening_a_warp_alone_merges() { + assert_eq!(both(&[open_child()]).expect("merge"), vec![open_child()]); + } } diff --git a/crates/warp-core/src/parallel/mod.rs b/crates/warp-core/src/parallel/mod.rs index b70b6f96b..bad673fce 100644 --- a/crates/warp-core/src/parallel/mod.rs +++ b/crates/warp-core/src/parallel/mod.rs @@ -19,9 +19,9 @@ pub use exec::{ ParallelExecutionPolicy, PoisonedDelta, ShardAssignmentPolicy, WorkUnit, WorkerExecution, WorkerResult, }; -#[cfg(not(any(test, feature = "delta_validate")))] -pub(crate) use merge::check_write_to_new_warp; pub use merge::MergeConflict; +#[cfg(not(any(test, feature = "delta_validate")))] +pub(crate) use merge::{merge_canonical, MergeFault}; #[cfg(any(test, feature = "delta_validate"))] pub use merge::{merge_deltas, merge_deltas_ok, MergeError}; pub use shard::{shard_of, NUM_SHARDS}; From 7975f7ee14e126648dcf3ddc49348cd563c3ec45 Mon Sep 17 00:00:00 2001 From: James Ross Date: Wed, 14 Oct 2026 01:51:24 -0700 Subject: [PATCH 2/2] CI: drop the duplicate delta_validate step The test-warp-core job already runs these six targets, with parallel_engine_worker_invariance and the library tests, under delta_validate. Found by review of #850. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 9 --------- 1 file changed, 9 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c557c03d8..35354c070 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -400,15 +400,6 @@ jobs: cargo check -p warp-core --features unsafe_graph --all-targets - name: cargo test (warp-wasm engine kernel) run: cargo test -p warp-wasm --features engine - - name: cargo test (warp-core canonical merge, delta_validate) - run: | - cargo test -p warp-core --features delta_validate \ - --test parallel_merge_tripwire \ - --test parallel_merge_warpopkey \ - --test parallel_openportal_rules \ - --test parallel_parallel_exec \ - --test parallel_engine_integration_multiwarp \ - --test parallel_stress_multiwarp - name: cargo test --doc (warp-core) run: cargo test -p warp-core --doc - name: PRNG golden regression (warp-math)