diff --git a/monas-state-node/src/infrastructure/crdt_repository.rs b/monas-state-node/src/infrastructure/crdt_repository.rs index d5f8e99..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 { @@ -369,95 +416,97 @@ 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. + // 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() + .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 { - // 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. + // 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 + // 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))?; @@ -474,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; @@ -488,11 +541,29 @@ 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. + // 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; + } + // 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 +723,396 @@ 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() + ); + } + + /// 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" + ); + } + + /// 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.