From 013f02c21522534b2a0095b97bf85240bbfd2425 Mon Sep 17 00:00:00 2001 From: Soma <0421.soma@gmail.com> Date: Sat, 29 Aug 2026 21:30:03 +0900 Subject: [PATCH 1/3] fix(state-node): pair sync operations with their own DAG nodes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A sync stamps each operation with the timestamp of the DAG node it produced, because the receiver recomputes node CIDs from it. Both sides of that pairing were guesses based on timestamp proximity: - `get_operations` matched an operation to "the first node within ±1s", which is the genesis for every operation written in the same second. Two versions therefore shipped under one timestamp, collapsed onto a single CID at the receiver, and the newer one was lost. - `since_version` resolved to "the last operation at or before this node's timestamp", indistinguishable between versions in that same second, so an incremental catch-up could re-send or skip a version. Both are now resolved by position: `linear_history` is ordered genesis → head and the operation list is ordered by timestamp, so entry N in one is entry N in the other. Re-delivery is also no longer reported as a failure. A pull-based sync re-sends what it has, so most operations in a steady-state cluster are already held; committing one again rebuilds the same node CID, which the DAG reports as a cycle. Deployed node1 logged that 1048 times in 25 minutes while pinning a core at ~100%. Operations we already hold are now skipped by operation id. A replica left behind by the old pairing converges from an ordinary sync, so no store needs to be cleared to recover. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012zvxAj1nZ6H1TDnkRtxnaQ --- .../src/infrastructure/crdt_repository.rs | 347 ++++++++++++++---- 1 file changed, 273 insertions(+), 74 deletions(-) diff --git a/monas-state-node/src/infrastructure/crdt_repository.rs b/monas-state-node/src/infrastructure/crdt_repository.rs index d5f8e99..7849c7b 100644 --- a/monas-state-node/src/infrastructure/crdt_repository.rs +++ b/monas-state-node/src/infrastructure/crdt_repository.rs @@ -369,95 +369,68 @@ impl ContentRepository for CrslCrdtRepository { .get_operations_with_index(&genesis) .map_err(|e| anyhow::anyhow!("Failed to get operations: {}", e))?; - // Filter by since_version if provided - // Find the index of the operation corresponding to since_version + // The linear history is ordered genesis → head, and `indexed_ops` is + // ordered by timestamp, so position N in one is position N in the + // other. Every lookup below uses that correspondence rather than + // comparing timestamps: two versions written inside the same second + // are indistinguishable by timestamp, and guessing between them + // silently dropped or re-sent versions. + let history = repo + .linear_history(&genesis) + .map_err(|e| anyhow::anyhow!("Failed to get history: {}", e))?; + + // Filter by since_version if provided: return only what came after it. let since_index = if let Some(since) = since_version { let since_cid = Self::parse_cid(since)?; - // If since_version is the genesis CID, skip the first operation (Create) - if since_cid == genesis { - Some(1) // Skip index 1 (the Create operation) - } else { - // Find the operation index by matching the DAG node timestamp - match repo.dag.get_node(&since_cid) { - Ok(Some(since_node)) => { - let since_ts = since_node.timestamp(); - // DAG node timestamps may be in seconds while operation timestamps are in nanoseconds - // Convert to nanoseconds if the timestamp appears to be in seconds - // Use the end of the second to include all operations within that second - let since_ts_nanos = if since_ts < 1_000_000_000_000 { - // Likely in seconds, convert to nanoseconds (end of that second) - since_ts * 1_000_000_000 + 999_999_999 - } else { - since_ts - }; - // Find the operation with the closest timestamp <= since_ts - indexed_ops - .iter() - .filter(|(_, op)| op.timestamp <= since_ts_nanos) - .map(|(idx, _)| *idx) - .max() - } - Ok(None) => None, - Err(_) => None, - } - } + // 1-based, matching `indexed_ops`. An unknown version yields None, + // which sends the full history — the safe direction, since the + // receiver can discard what it already has but cannot invent what + // it never received. + history + .iter() + .position(|cid| *cid == since_cid) + .map(|pos| pos + 1) } else { None }; - // Get all DAG nodes for this genesis to find node timestamps - let history = repo - .linear_history(&genesis) - .map_err(|e| anyhow::anyhow!("Failed to get history: {}", e))?; - - // Build a map of operation timestamp -> DAG node timestamp - // This is needed because DAG node timestamps may differ from operation timestamps - let mut node_timestamps: Vec<(u64, u64)> = Vec::new(); - for node_cid in &history { - if let Ok(Some(node)) = repo.dag.get_node(node_cid) { - node_timestamps.push((node.timestamp(), node.timestamp())); - } - } + // Pair each operation with the DAG node it produced. + // + // The receiver recomputes a node's CID from `node_timestamp`, so an + // operation carrying the wrong one is re-derived as a different node — + // or, when two operations carry the same one, collapses onto a single + // CID and the newer version is lost. + // + // `linear_history` is ordered genesis → head and `indexed_ops` is + // ordered by timestamp, so the two line up position by position. That + // is the only reliable correspondence: matching by timestamp proximity + // returned the first node within ±1s, which is the genesis for every + // operation written in the same second. + let node_timestamps: Vec = history + .iter() + .filter_map(|cid| repo.dag.get_node(cid).ok().flatten()) + .map(|node| node.timestamp()) + .collect(); let mut operations = Vec::new(); for (idx, op) in indexed_ops { - // Skip operations at or before the since_version index + // `idx` is 1-based over the full operation list, which is exactly + // the position of this operation's node in the linear history. + let node_timestamp = node_timestamps + .get(idx - 1) + .copied() + .unwrap_or(op.timestamp); + + // Skip operations at or before the since_version index. This runs + // after the lookup above so that skipping never shifts the + // remaining operations onto the wrong nodes. if let Some(since_idx) = since_index { if idx <= since_idx { continue; } } - // Find the corresponding DAG node timestamp - // For Create operations, use the genesis node timestamp - // For other operations, find the node with matching or closest timestamp - let node_timestamp = if matches!(op.kind, OperationType::Create(_)) { - // For Create operation, get the genesis node timestamp - repo.dag - .get_node(&genesis) - .ok() - .flatten() - .map(|n| n.timestamp()) - .unwrap_or(op.timestamp) - } else { - // For Update/Delete/Merge, find the node in history that corresponds to this operation - // The node timestamp should be close to the operation timestamp - history - .iter() - .filter_map(|cid| repo.dag.get_node(cid).ok().flatten()) - .find(|node| { - // Node timestamp should be within a reasonable range of op timestamp - // or we just find the closest one - let node_ts = node.timestamp(); - // Allow some tolerance for timestamp matching - node_ts >= op.timestamp.saturating_sub(1_000_000_000) - && node_ts <= op.timestamp.saturating_add(1_000_000_000) - }) - .map(|n| n.timestamp()) - .unwrap_or(op.timestamp) - }; - // Serialize the operation using serde_json for network transfer let serialized = serde_json::to_vec(&op) .map_err(|e| anyhow::anyhow!("Failed to serialize operation: {}", e))?; @@ -488,11 +461,23 @@ impl ContentRepository for CrslCrdtRepository { // Set node_timestamp for import mode to ensure CID consistency across replicas op.node_timestamp = Some(serialized_op.node_timestamp); + // A pull-based sync re-sends everything it has, so most operations + // in a steady-state cluster are ones we already hold. Committing + // one again rebuilds the same node CID, which the DAG reports as a + // cycle — an error for what is the expected case. Skip the ones we + // know: the operation id is assigned by the author and travels + // with it, so it identifies the operation across replicas. + let already_have = matches!(repo.state.get_operation(&op.id), Ok(Some(_))); + if already_have { + applied += 1; + continue; + } + // Apply the operation match repo.commit_operation(op) { Ok(_) => applied += 1, Err(e) => { - // Log but continue - operation might be duplicate or conflict + // Log but continue - operation might be a genuine conflict. tracing::warn!("Failed to apply operation: {}", e); } } @@ -652,6 +637,220 @@ mod tests { use super::*; use tempfile::tempdir; + /// Builds a content network with history and returns (repo, genesis_cid). + async fn creator_with_three_versions() -> (CrslCrdtRepository, tempfile::TempDir, String) { + let tmp = tempdir().unwrap(); + let repo = CrslCrdtRepository::open(tmp.path().join("crdt")).unwrap(); + let created = repo.create_content(b"v1", "author-a", None).await.unwrap(); + repo.update_content(&created.genesis_cid, b"v2", "author-a", None) + .await + .unwrap(); + repo.update_content(&created.genesis_cid, b"v3", "author-a", None) + .await + .unwrap(); + (repo, tmp, created.genesis_cid) + } + + /// A sync must reproduce the sender's history exactly. + /// + /// `get_operations` stamps each operation with the timestamp of the DAG + /// node it belongs to, because the receiver recomputes node CIDs from it. + /// Matching an operation to its node by "any node within ±1s" returns the + /// *first* such node — the genesis — for every operation written in the + /// same second, so v2 and v3 were both re-derived under the genesis + /// timestamp, collapsed onto one CID, and the newest version was silently + /// lost. Every content network created and edited in one session hits + /// this, which is why nodes disagreed about `latest_version`. + #[tokio::test] + async fn syncing_preserves_every_version() { + let (creator, _creator_tmp, genesis_cid) = creator_with_three_versions().await; + + let ops = creator.get_operations(&genesis_cid, None).await.unwrap(); + assert_eq!(ops.len(), 3, "create + 2 updates"); + + // Each operation must carry its OWN node timestamp; sharing one means + // the receiver cannot tell the versions apart. + let stamps: std::collections::HashSet = ops.iter().map(|o| o.node_timestamp).collect(); + assert_eq!( + stamps.len(), + ops.len(), + "each operation must map to a distinct DAG node timestamp" + ); + + let receiver_tmp = tempdir().unwrap(); + let receiver = CrslCrdtRepository::open(receiver_tmp.path().join("crdt")).unwrap(); + assert_eq!(receiver.apply_operations(&ops).await.unwrap(), ops.len()); + + assert_eq!( + receiver.get_latest(&genesis_cid).await.unwrap().unwrap(), + b"v3".to_vec(), + "the receiver must end up on the creator's latest version" + ); + assert_eq!( + receiver.get_history(&genesis_cid).await.unwrap().len(), + creator.get_history(&genesis_cid).await.unwrap().len(), + "the receiver must hold the same number of versions as the creator" + ); + } + + /// Reproduces the production log flood: every periodic sync re-sends the + /// same operations, and a node that already holds them logged + /// "Failed to apply operation: graph error: cycle detected in graph" for + /// each one (1048 times in 25 minutes on node1). Re-delivery is the normal + /// steady state of a pull-based sync, so it must be a quiet no-op. + #[tokio::test] + async fn reapplying_known_operations_is_a_quiet_no_op() { + let (creator, _creator_tmp, genesis_cid) = creator_with_three_versions().await; + let ops = creator.get_operations(&genesis_cid, None).await.unwrap(); + + let receiver_tmp = tempdir().unwrap(); + let receiver = CrslCrdtRepository::open(receiver_tmp.path().join("crdt")).unwrap(); + receiver.apply_operations(&ops).await.unwrap(); + + // The next periodic sync hands us the very same operations again. + let reapplied = receiver.apply_operations(&ops).await.unwrap(); + + assert_eq!( + receiver.get_latest(&genesis_cid).await.unwrap().unwrap(), + b"v3".to_vec(), + "re-syncing known operations must not change the content" + ); + assert_eq!( + receiver.get_history(&genesis_cid).await.unwrap().len(), + 3, + "re-syncing must not duplicate or drop versions" + ); + assert_eq!( + reapplied, + ops.len(), + "operations we already hold count as applied, not as failures" + ); + } + + /// Incremental sync: a node that already holds part of the history asks + /// for "everything after version X". The operations it gets back must + /// still carry their own node timestamps — the `since_version` filter + /// must not shift them onto the wrong nodes. + #[tokio::test] + async fn incremental_sync_lands_on_the_same_history() { + let (creator, _creator_tmp, genesis_cid) = creator_with_three_versions().await; + + // Receiver catches up to v1 only. + let receiver_tmp = tempdir().unwrap(); + let receiver = CrslCrdtRepository::open(receiver_tmp.path().join("crdt")).unwrap(); + let all = creator.get_operations(&genesis_cid, None).await.unwrap(); + receiver.apply_operations(&all[..1]).await.unwrap(); + assert_eq!( + receiver.get_latest(&genesis_cid).await.unwrap().unwrap(), + b"v1".to_vec() + ); + + // Now it asks only for what came after the genesis version. + let tail = creator + .get_operations(&genesis_cid, Some(&genesis_cid)) + .await + .unwrap(); + assert_eq!(tail.len(), 2, "the two updates, not the create"); + receiver.apply_operations(&tail).await.unwrap(); + + assert_eq!( + receiver.get_latest(&genesis_cid).await.unwrap().unwrap(), + b"v3".to_vec(), + "an incremental catch-up must reach the creator's latest version" + ); + assert_eq!( + receiver.get_history(&genesis_cid).await.unwrap(), + creator.get_history(&genesis_cid).await.unwrap(), + "incremental and full sync must produce the same history" + ); + } + + /// `since_version` must mean "everything strictly after this version". + /// + /// Resolving it by "the last operation whose timestamp is <= the node's" + /// is the same guess that broke node timestamps: versions written inside + /// one second are indistinguishable that way, so a catch-up from v2 could + /// return v2 again (re-delivering work) or skip v3 (losing it). + #[tokio::test] + async fn since_version_returns_exactly_the_newer_operations() { + let (creator, _creator_tmp, genesis_cid) = creator_with_three_versions().await; + + let history = creator.get_history(&genesis_cid).await.unwrap(); + assert_eq!(history.len(), 3, "genesis + 2 updates"); + + // Asking from the middle version must yield only the last update. + let tail = creator + .get_operations(&genesis_cid, Some(&history[1])) + .await + .unwrap(); + assert_eq!( + tail.len(), + 1, + "since=v2 must return only the operation that produced v3" + ); + + // And that one operation must carry v3's node timestamp, so a receiver + // rebuilds the same node. + let receiver_tmp = tempdir().unwrap(); + let receiver = CrslCrdtRepository::open(receiver_tmp.path().join("crdt")).unwrap(); + let head = creator.get_operations(&genesis_cid, None).await.unwrap(); + receiver.apply_operations(&head[..2]).await.unwrap(); + receiver.apply_operations(&tail).await.unwrap(); + assert_eq!( + receiver.get_history(&genesis_cid).await.unwrap(), + history, + "catching up from the middle must rebuild the identical history" + ); + + // Asking from the newest version must yield nothing. + let none = creator + .get_operations(&genesis_cid, Some(&history[2])) + .await + .unwrap(); + assert!( + none.is_empty(), + "since=latest must return nothing, got {} operation(s)", + none.len() + ); + } + + /// A node that already synced under the old (broken) pairing holds a + /// truncated history. Once the sender is fixed, the operations it sends + /// carry correct node timestamps — so the stale receiver must be able to + /// converge on the full history without being wiped first. + /// + /// This decides whether the deployed cluster needs its CRDT store cleared + /// or heals on its own after a redeploy. + #[tokio::test] + async fn a_receiver_with_a_truncated_history_catches_up() { + let (creator, _creator_tmp, genesis_cid) = creator_with_three_versions().await; + let ops = creator.get_operations(&genesis_cid, None).await.unwrap(); + + // Simulate the damaged state: the receiver got the create and one + // update, and never received the newest version. + let receiver_tmp = tempdir().unwrap(); + let receiver = CrslCrdtRepository::open(receiver_tmp.path().join("crdt")).unwrap(); + receiver.apply_operations(&ops[..2]).await.unwrap(); + assert_eq!( + receiver.get_history(&genesis_cid).await.unwrap().len(), + 2, + "precondition: the receiver is behind" + ); + + // The next periodic sync delivers the full, correctly-stamped set. + receiver.apply_operations(&ops).await.unwrap(); + + assert_eq!( + receiver.get_history(&genesis_cid).await.unwrap(), + creator.get_history(&genesis_cid).await.unwrap(), + "a stale replica must converge from an ordinary sync" + ); + assert_eq!( + receiver.get_latest(&genesis_cid).await.unwrap().unwrap(), + b"v3".to_vec() + ); + } + #[tokio::test] async fn test_prepare_create_operations_is_deterministic_across_repos() { // Creator prepares operations without persisting to its own store. From 7c64deafebace6a95bccf49fca88f81181d19670 Mon Sep 17 00:00:00 2001 From: Soma <0421.soma@gmail.com> Date: Sat, 29 Aug 2026 22:05:53 +0900 Subject: [PATCH 2/3] test(state-node): cover repeated operation delivery at cluster scale The single re-delivery case did not show the cost the deployed cluster paid: node1 logged 1048 apply failures in 25 minutes, each a full commit attempt that walks the DAG before failing, while pinning a core at ~100%. This drives 20 sync rounds from 2 providers and asserts every round accounts for all operations, so a regression shows up as a shortfall rather than as log noise nobody reads. Verified discriminating: removing the dedup check fails it (1 of 3 applied on the second round). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012zvxAj1nZ6H1TDnkRtxnaQ --- .../src/infrastructure/crdt_repository.rs | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/monas-state-node/src/infrastructure/crdt_repository.rs b/monas-state-node/src/infrastructure/crdt_repository.rs index 7849c7b..a318613 100644 --- a/monas-state-node/src/infrastructure/crdt_repository.rs +++ b/monas-state-node/src/infrastructure/crdt_repository.rs @@ -851,6 +851,40 @@ mod tests { ); } + /// The production failure mode, at the scale it actually happened. + /// + /// A member is asked to re-apply the same operations on every sync round + /// and from every provider. node1 logged 1048 such failures in 25 minutes + /// while pinning a core; each one is a full commit attempt that walks the + /// DAG before failing. Repeated delivery must stay cheap and quiet. + #[tokio::test] + async fn repeated_delivery_never_errors() { + let (creator, _creator_tmp, genesis_cid) = creator_with_three_versions().await; + let ops = creator.get_operations(&genesis_cid, None).await.unwrap(); + + let receiver_tmp = tempdir().unwrap(); + let receiver = CrslCrdtRepository::open(receiver_tmp.path().join("crdt")).unwrap(); + + // 20 sync rounds × 2 providers, the shape of a steady-state cluster. + for _ in 0..20 { + for _ in 0..2 { + let applied = receiver.apply_operations(&ops).await.unwrap(); + assert_eq!( + applied, + ops.len(), + "every round must account for all operations; a shortfall is \ + an operation that failed to apply" + ); + } + } + + assert_eq!( + receiver.get_history(&genesis_cid).await.unwrap(), + creator.get_history(&genesis_cid).await.unwrap(), + "40 redeliveries must leave the history identical to the sender's" + ); + } + #[tokio::test] async fn test_prepare_create_operations_is_deterministic_across_repos() { // Creator prepares operations without persisting to its own store. From 73771f9322b9b616043907bf3bfa7f6fe66ab1af Mon Sep 17 00:00:00 2001 From: Soma <0421.soma@gmail.com> Date: Sat, 29 Aug 2026 23:16:47 +0900 Subject: [PATCH 3/3] fix(state-node): heal wrong-CID replicas and refuse unpairable histories MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review found two ways the previous commit fell short. **Damaged replicas did not heal.** A receiver that committed a node under a wrong CID — the actual damage the old ±1s pairing did — has the operation id recorded in state, so skipping on the id alone left the wrong node in place forever. The claim that no store needs clearing was therefore wrong for the case that matters most. Skipping now also requires the node itself to be present under the CID the stamp implies, which is recomputable from the operation. A replica damaged this way converges from an ordinary sync; `a_replica_that_committed_a_wrong_cid_ heals_from_an_ordinary_sync` fails without the check. **The positional pairing had no guard.** It holds only while the history is a line: `get_operations_with_index` returns every operation for the genesis, while `linear_history` picks one child per branch. Once the DAG forks the lists differ, and an index pairs an operation with an unrelated node — the same wrong-stamp corruption arriving through another door. A node that fails to load was also silently dropped, shifting every pair after it. Both now fail the call: a refused sync round is retried, a silently mis-stamped one is not. Also documents what `apply_operations` counts, since re-delivery makes the return value a coverage measure rather than a write count. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012zvxAj1nZ6H1TDnkRtxnaQ --- .../src/infrastructure/crdt_repository.rs | 248 +++++++++++++++++- 1 file changed, 238 insertions(+), 10 deletions(-) diff --git a/monas-state-node/src/infrastructure/crdt_repository.rs b/monas-state-node/src/infrastructure/crdt_repository.rs index a318613..9126408 100644 --- a/monas-state-node/src/infrastructure/crdt_repository.rs +++ b/monas-state-node/src/infrastructure/crdt_repository.rs @@ -79,6 +79,53 @@ impl CrslCrdtRepository { Ok(()) } + /// Whether the node this operation produces is already in the DAG. + /// + /// The node CID is a function of the payload, parents, genesis and the + /// stamped timestamp, so it can be recomputed from the operation alone — + /// which is what lets a receiver tell "I already have this" apart from "I + /// have a differently-stamped version of this". Metadata comes from the + /// parent, mirroring how crsl-lib resolves it when committing. + fn node_is_present(repo: &ContentRepo, op: &Operation) -> bool { + use crsl_lib::dasl::node::Node; + + let Some(timestamp) = op.node_timestamp else { + return false; + }; + + let node = match (&op.kind, op.parents.first()) { + (OperationType::Create(payload), _) => { + Node::::new_genesis( + payload.clone(), + timestamp, + ContentMetadata::default(), + ) + } + (OperationType::Update(payload), Some(parent)) + | (OperationType::Merge(payload), Some(parent)) => { + let Ok(Some(parent_node)) = repo.dag.get_node(parent) else { + return false; + }; + Node::new_child( + payload.clone(), + op.parents.clone(), + op.genesis, + timestamp, + parent_node.metadata().clone(), + ) + } + // Delete carries no payload of its own and parentless updates are + // resolved at commit time; neither can be reconstructed here, so + // fall back to committing and letting the DAG decide. + _ => return false, + }; + + let Ok(cid) = node.content_id() else { + return false; + }; + matches!(repo.dag.get_node(&cid), Ok(Some(_))) + } + /// Generate a placeholder CID from content data. /// This is used as a seed for Create operations. fn generate_placeholder_cid(data: &[u8]) -> Cid { @@ -407,20 +454,49 @@ impl ContentRepository for CrslCrdtRepository { // is the only reliable correspondence: matching by timestamp proximity // returned the first node within ±1s, which is the genesis for every // operation written in the same second. - let node_timestamps: Vec = history + // A node we cannot read must not be skipped: dropping one shifts every + // pair after it by one, which is worse than the tail simply running + // short. + let node_timestamps = history .iter() - .filter_map(|cid| repo.dag.get_node(cid).ok().flatten()) - .map(|node| node.timestamp()) - .collect(); + .map(|cid| { + repo.dag + .get_node(cid) + .ok() + .flatten() + .map(|node| node.timestamp()) + .ok_or_else(|| { + anyhow::anyhow!("Missing DAG node {cid} in the history of {genesis_cid}") + }) + }) + .collect::>>()?; + + // The pairing below is positional, and that only holds while the + // history is a line: `get_operations_with_index` returns every + // operation for this genesis, while `linear_history` picks a single + // child at each branch. Once the DAG forks, the lists differ and an + // index no longer identifies the node an operation produced. + // + // Refuse rather than guess. Handing an operation a timestamp from an + // unrelated node is what corrupted replicas in the first place, and a + // failed sync round is retried; a silently mis-stamped one is not. + if node_timestamps.len() != indexed_ops.len() { + return Err(anyhow::anyhow!( + "Cannot pair operations with their DAG nodes for {genesis_cid}: \ + {} operations over a {}-node linear history. The history is not \ + linear, so no positional pairing is trustworthy.", + indexed_ops.len(), + node_timestamps.len() + )); + } let mut operations = Vec::new(); for (idx, op) in indexed_ops { // `idx` is 1-based over the full operation list, which is exactly // the position of this operation's node in the linear history. - let node_timestamp = node_timestamps - .get(idx - 1) - .copied() - .unwrap_or(op.timestamp); + // Guarded above: the two lists are the same length, so every + // 1-based operation index addresses its own node. + let node_timestamp = node_timestamps[idx - 1]; // Skip operations at or before the since_version index. This runs // after the lookup above so that skipping never shifts the @@ -447,6 +523,10 @@ impl ContentRepository for CrslCrdtRepository { Ok(operations) } + /// Applies operations, returning how many of them the store now accounts + /// for — including ones it already held. This is a measure of coverage, + /// not of new writes: a caller uses it to tell "the batch landed" from + /// "some operations failed", and re-delivery is the normal case. async fn apply_operations(&self, operations: &[SerializedOperation]) -> Result { let mut applied = 0; @@ -467,8 +547,14 @@ impl ContentRepository for CrslCrdtRepository { // cycle — an error for what is the expected case. Skip the ones we // know: the operation id is assigned by the author and travels // with it, so it identifies the operation across replicas. - let already_have = matches!(repo.state.get_operation(&op.id), Ok(Some(_))); - if already_have { + // Knowing the operation is not the same as holding the node it + // produced. A replica damaged by the old pairing has the operation + // id recorded but committed the node under a wrong CID, so skipping + // on the id alone would leave it broken forever. Only skip when the + // operation is known AND its node is present under the CID this + // stamp implies. + let known = matches!(repo.state.get_operation(&op.id), Ok(Some(_))); + if known && Self::node_is_present(&repo, &op) { applied += 1; continue; } @@ -885,6 +971,148 @@ mod tests { ); } + /// The position pairing only holds while the history is linear. + /// + /// `get_operations_with_index` returns every operation for the genesis; + /// `linear_history` walks one path and picks a single child at each + /// branch. The moment the DAG forks the two lists differ in length and + /// order, and pairing by index hands an operation a timestamp belonging + /// to some unrelated node — the same "wrong stamp, wrong CID" failure + /// this module was changed to fix, arriving through another door. + /// + /// A node lookup that fails mid-history is worse still: `filter_map` + /// drops it, so every pair after that point shifts by one. + /// + /// We cannot honestly stamp operations we cannot place, so refuse rather + /// than guess. + #[tokio::test] + async fn refuses_to_stamp_operations_it_cannot_place() { + let (creator, _tmp, genesis_cid) = creator_with_three_versions().await; + + // Sanity: the linear case still works. + assert_eq!( + creator + .get_operations(&genesis_cid, None) + .await + .unwrap() + .len(), + 3 + ); + + let genesis = CrslCrdtRepository::parse_cid(&genesis_cid).unwrap(); + let (ops_len, history_len) = { + let repo = creator.repo.lock(); + ( + repo.get_operations_with_index(&genesis).unwrap().len(), + repo.linear_history(&genesis).unwrap().len(), + ) + }; + assert_eq!( + ops_len, history_len, + "precondition: a linear history pairs one-to-one" + ); + + // Now fork it. Two replicas edit independently and then exchange + // operations, which is ordinary behaviour for a multi-writer network. + let other_tmp = tempdir().unwrap(); + let other = CrslCrdtRepository::open(other_tmp.path().join("crdt")).unwrap(); + other + .apply_operations(&creator.get_operations(&genesis_cid, None).await.unwrap()) + .await + .unwrap(); + + creator + .update_content(&genesis_cid, b"from-creator", "author-a", None) + .await + .unwrap(); + other + .update_content(&genesis_cid, b"from-other", "author-b", None) + .await + .unwrap(); + + // Creator learns about the divergent branch. + creator + .apply_operations(&other.get_operations(&genesis_cid, None).await.unwrap()) + .await + .unwrap(); + + let (ops_len, history_len) = { + let repo = creator.repo.lock(); + ( + repo.get_operations_with_index(&genesis).unwrap().len(), + repo.linear_history(&genesis).unwrap().len(), + ) + }; + + if ops_len == history_len { + // Auto-merge collapsed the fork back into a line; the invariant + // still holds and there is nothing to refuse. + return; + } + + // The lists disagree, so no index pairing is trustworthy. Stamping + // anyway is what corrupts replicas, so the call must fail loudly. + let err = creator.get_operations(&genesis_cid, None).await.expect_err( + "with {ops_len} operations over a {history_len}-node history, \ + pairing by index is a guess and must be refused", + ); + let msg = err.to_string(); + assert!( + msg.contains("history") || msg.contains("operations"), + "the error should say the pairing broke, got: {msg}" + ); + } + + /// A replica that committed a WRONG node CID under the old pairing must + /// heal once the sender is fixed. + /// + /// This is the case the recovery claim turns on, and it is not the same as + /// a truncated prefix. Under the old code a receiver could commit v2 + /// stamped with the genesis timestamp: a node at the wrong CID, with the + /// operation id already recorded in state. Skipping on the operation id + /// alone would leave that wrong node in place forever, so the skip also + /// requires the node itself to be present. + #[tokio::test] + async fn a_replica_that_committed_a_wrong_cid_heals_from_an_ordinary_sync() { + let (creator, _tmp, genesis_cid) = creator_with_three_versions().await; + let correct = creator.get_operations(&genesis_cid, None).await.unwrap(); + + // Reproduce the damage: same operations, but the updates carry the + // genesis timestamp, which is exactly what the ±1s match produced. + let genesis_ts = correct[0].node_timestamp; + let damaged: Vec = correct + .iter() + .map(|op| SerializedOperation { + node_timestamp: genesis_ts, + ..op.clone() + }) + .collect(); + + let victim_tmp = tempdir().unwrap(); + let victim = CrslCrdtRepository::open(victim_tmp.path().join("crdt")).unwrap(); + victim.apply_operations(&damaged).await.unwrap(); + + let damaged_history = victim.get_history(&genesis_cid).await.unwrap(); + let healthy_history = creator.get_history(&genesis_cid).await.unwrap(); + assert_ne!( + damaged_history, healthy_history, + "precondition: the victim's history must actually be damaged" + ); + + // The sender is fixed now and re-sends correctly stamped operations, + // repeatedly, as a periodic sync would. + for _ in 0..5 { + victim.apply_operations(&correct).await.unwrap(); + } + + let after = victim.get_history(&genesis_cid).await.unwrap(); + assert_eq!( + after, healthy_history, + "a replica damaged by the old pairing must converge once the sender \ + is fixed; if this fails, such replicas need their store cleared" + ); + } + #[tokio::test] async fn test_prepare_create_operations_is_deterministic_across_repos() { // Creator prepares operations without persisting to its own store.