From 7c0674f4cf749e4cd76cd7b791245d42aa571773 Mon Sep 17 00:00:00 2001 From: malibio Date: Fri, 25 Sep 2026 19:26:39 +0200 Subject: [PATCH 1/3] Report store version conflicts as typed values, not parsed messages (closes #3032) move_children_to_parent_in_tx and delete_subtree_atomic now return Ok(Err(VersionConflict { node_id, expected, actual })) instead of formatting the conflict into an anyhow message the service had to string-match. move_children_to_parent now reports the real expected and persisted versions (previously 0/0), and a zero-row gated DELETE on a child still at its expected version (no parent edge) is a plain error rather than a misreported conflict. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/core/src/db/mod.rs | 2 +- packages/core/src/db/sqlite_store/mod.rs | 84 +++++++++++++-- packages/core/src/db/sqlite_store/nodes.rs | 102 ++++++++++++------ .../core/src/services/node_service/crud.rs | 15 +-- .../src/services/node_service/hierarchy.rs | 22 ++-- .../core/src/services/node_service/mod.rs | 59 +++++++--- 6 files changed, 199 insertions(+), 85 deletions(-) diff --git a/packages/core/src/db/mod.rs b/packages/core/src/db/mod.rs index 07f1d4356..fbc95647c 100644 --- a/packages/core/src/db/mod.rs +++ b/packages/core/src/db/mod.rs @@ -16,5 +16,5 @@ pub(crate) use sqlite_store::collection_not_root; pub(crate) use sqlite_store::tx::Tx; pub use sqlite_store::{ ensure_sqlite_vec_registered, BulkNodeRow, RelationshipRecord, ResolvedEntity, SqliteStore, - StoreChange, StoreOperation, + StoreChange, StoreOperation, VersionConflict, }; diff --git a/packages/core/src/db/sqlite_store/mod.rs b/packages/core/src/db/sqlite_store/mod.rs index 8c57cf546..5287bd5f2 100644 --- a/packages/core/src/db/sqlite_store/mod.rs +++ b/packages/core/src/db/sqlite_store/mod.rs @@ -87,6 +87,33 @@ impl RelationshipRecord { } } +/// An optimistic-concurrency mismatch observed by a store write: `node_id` +/// was expected at `expected` but is persisted at `actual`. +/// +/// Store methods return it as `Ok(Err(VersionConflict))` rather than as an +/// `anyhow` error (ADR-069 §2a: a version mismatch is an expected outcome, not +/// a transaction failure), so the service maps it to +/// `NodeServiceError::VersionConflict` by type — never by parsing message +/// text — and carries the real versions through to the caller. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct VersionConflict { + pub node_id: String, + pub expected: i64, + pub actual: i64, +} + +impl std::fmt::Display for VersionConflict { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "version conflict on node '{}': expected {}, actual {}", + self.node_id, self.expected, self.actual + ) + } +} + +impl std::error::Error for VersionConflict {} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum StoreOperation { Created, @@ -663,7 +690,11 @@ mod tests { .iter() .map(|(id, ver)| (id.as_str(), *ver)) .collect(); - SqliteStore::move_children_to_parent_in_tx(tx, &new_parent_id, &refs).await + // A conflict must fail the closure, not return `Ok`, or + // `with_transaction` would commit the edges already swapped. + SqliteStore::move_children_to_parent_in_tx(tx, &new_parent_id, &refs) + .await? + .map_err(anyhow::Error::new) }) }) .await @@ -2225,6 +2256,36 @@ mod tests { Ok(()) } + /// A gated DELETE that removes nothing is only a version conflict when the + /// persisted version actually differs. A child at its expected version with + /// no `has_child` edge (a root) must fail as a plain error, not surface as a + /// conflict carrying equal expected and actual versions. + #[tokio::test] + async fn test_move_children_to_parent_store_root_child_is_not_a_conflict() -> Result<()> { + let (store, _temp) = create_test_store().await?; + + let new_parent = Node::new("text".to_string(), "New Parent".to_string(), json!({})); + let new_parent_id = new_parent.id.clone(); + store.create_node(new_parent, None, None).await?; + + let root = Node::new("text".to_string(), "Root".to_string(), json!({})); + let root_id = root.id.clone(); + store.create_node(root, None, None).await?; + let root_version = store.get_node(&root_id).await?.unwrap().version; + + let err = + move_children_to_parent(&store, &new_parent_id, &[(root_id.as_str(), root_version)]) + .await + .expect_err("a root child has no edge to replace"); + assert!( + err.downcast_ref::().is_none(), + "a missing parent edge is not a version conflict, got: {err:#}" + ); + assert_eq!(store.get_parent_id(&root_id).await?, None); + + Ok(()) + } + // C3c: store-layer atomicity — exercises the in-transaction version check // that the service-level pre-validation cannot catch (concurrent version bump). #[tokio::test] @@ -2264,15 +2325,18 @@ mod tests { ) .await; - assert!( - result.is_err(), - "stale version should cause store-level failure" - ); - let err_msg = result.unwrap_err().to_string(); - assert!( - err_msg.contains("VERSION_CONFLICT"), - "error should contain VERSION_CONFLICT, got: {}", - err_msg + let err = result.expect_err("stale version should cause store-level failure"); + let conflict = err + .downcast_ref::() + .unwrap_or_else(|| panic!("expected a typed VersionConflict, got: {err:#}")); + let child2_actual = store.get_node(&child2_id).await?.unwrap().version; + assert_eq!( + conflict, + &VersionConflict { + node_id: child2_id.clone(), + expected: stale_version, + actual: child2_actual, + } ); // ALL-OR-NOTHING: child1 must NOT have moved (transaction was rolled back). diff --git a/packages/core/src/db/sqlite_store/nodes.rs b/packages/core/src/db/sqlite_store/nodes.rs index f31088db5..3bec3c019 100644 --- a/packages/core/src/db/sqlite_store/nodes.rs +++ b/packages/core/src/db/sqlite_store/nodes.rs @@ -1335,18 +1335,20 @@ impl SqliteStore { /// drifting from — a second walk. /// /// Returns `(existed, deleted_nodes)` where `deleted_nodes` contains the target and all - /// descendants that were deleted (empty vec when target didn't exist). + /// descendants that were deleted (empty vec when target didn't exist), or + /// `Ok(Err(VersionConflict))` when the target's persisted version is not + /// `expected_version` — nothing is deleted in that case. pub async fn delete_subtree_atomic( &self, node_id: &str, expected_version: i64, subtree_ids: &[String], source: Option, - ) -> Result<(bool, Vec)> { + ) -> Result), VersionConflict>> { // Collect the target + all descendants before mutating. let target = match self.get_node(node_id).await? { Some(n) => n, - None => return Ok((false, vec![])), + None => return Ok(Ok((false, vec![]))), }; // Only the target can be a schema node (descendants are its description @@ -1356,12 +1358,11 @@ impl SqliteStore { // OCC check on target before entering the transaction. if target.version != expected_version { - return Err(anyhow::anyhow!( - "version_conflict:{}:{}:{}", - node_id, - expected_version, - target.version - )); + return Ok(Err(VersionConflict { + node_id: node_id.to_string(), + expected: expected_version, + actual: target.version, + })); } let all_ids: Vec = subtree_ids.to_vec(); @@ -1412,16 +1413,16 @@ impl SqliteStore { Some(row) => row.get(0)?, None => { // Node disappeared between pre-check and transaction — idempotent. - return Ok((false, vec![])); + return Ok(Ok((false, vec![]))); } }; if actual_version != expected_version { - return Err(anyhow::anyhow!( - "version_conflict:{}:{}:{}", - node_id, - expected_version, - actual_version - )); + // Dropping `tx` uncommitted rolls it back; nothing was written yet. + return Ok(Err(VersionConflict { + node_id: node_id.to_string(), + expected: expected_version, + actual: actual_version, + })); } // Chunked, but every chunk executes against the same `tx` opened above — @@ -1460,7 +1461,7 @@ impl SqliteStore { }); } - Ok((true, nodes_to_delete)) + Ok(Ok((true, nodes_to_delete))) } /// Delete all descendants of `parent_id` (the full child subtree) without touching the parent. @@ -2833,11 +2834,15 @@ impl SqliteStore { /// Re-parent an ordered set of existing children to `new_parent_id` inside /// `tx`. Validates each child's version using `SELECT changes()` after a - /// version-gated DELETE — any mismatch returns an error, so the caller's - /// transaction rolls back every edge moved so far (all-or-nothing OCC). + /// version-gated DELETE. /// /// Returns the assigned fractional order for each child, preserving input - /// array order as sibling order under the new parent. + /// array order as sibling order under the new parent, or + /// `Ok(Err(VersionConflict))` carrying the first mismatched child's + /// expected and persisted versions. On a conflict, edges for earlier + /// children have already been swapped in `tx`: the caller must turn the + /// conflict into an `Err` from its transaction closure so every one of + /// them rolls back (all-or-nothing OCC). /// /// `_in_tx` only (ADR-069 §1a): the service composes the version bumps and /// event emission into the same transaction, so edges and bumps commit @@ -2846,9 +2851,9 @@ impl SqliteStore { tx: &Tx<'_>, new_parent_id: &str, children: &[(&str, i64)], - ) -> Result> { + ) -> Result, VersionConflict>> { if children.is_empty() { - return Ok(Vec::new()); + return Ok(Ok(Vec::new())); } let now = Utc::now().to_rfc3339(); @@ -2902,15 +2907,41 @@ impl SqliteStore { .query("SELECT changes()", libsql::params![]) .await .context("Failed to query changes()")?; - if let Some(row) = changes_rows.next().await? { - let affected: i64 = row.get(0)?; - if affected == 0 { - return Err(anyhow::anyhow!( - "VERSION_CONFLICT: node '{}' version mismatch (expected {})", - child_id, - expected_version - )); + let affected: i64 = match changes_rows.next().await? { + Some(row) => row.get(0)?, + None => return Err(anyhow::anyhow!("SELECT changes() returned no row")), + }; + if affected == 0 { + // Nothing was deleted: either the version moved on, or the + // child has no has_child edge to replace. Read the persisted + // version in this same transaction to tell them apart. + let mut version_rows = db + .query( + "SELECT version FROM node WHERE id = ?1", + libsql::params![child_id.clone()], + ) + .await + .context("Failed to read child version after a gated DELETE")?; + let actual: i64 = match version_rows.next().await? { + Some(row) => row.get(0)?, + None => { + return Err(anyhow::anyhow!( + "Node not found while moving children: {}", + child_id + )) + } + }; + if actual != *expected_version { + return Ok(Err(VersionConflict { + node_id: child_id, + expected: *expected_version, + actual, + })); } + return Err(anyhow::anyhow!( + "Node '{}' has no parent edge to replace", + child_id + )); } // Insert new has_child edge under new_parent_id. @@ -2931,7 +2962,7 @@ impl SqliteStore { .context("Failed to insert new has_child edge")?; } - Ok(orders) + Ok(Ok(orders)) } pub async fn get_schema(&self, node_type: &str) -> Result> { @@ -5341,7 +5372,8 @@ mod large_subtree_chunking_tests { // subtree nodes" without deleting anything. let (existed, deleted_nodes) = store .delete_subtree_atomic(&root.id, root.version, &subtree_ids, None) - .await?; + .await? + .expect("target is at root.version"); assert!(existed); assert_eq!(deleted_nodes.len(), DESCENDANT_COUNT + 1); @@ -5368,7 +5400,8 @@ mod large_subtree_chunking_tests { // call's chunks committed together as one unit. let (existed_again, _) = store .delete_subtree_atomic(&root.id, root.version, &subtree_ids, None) - .await?; + .await? + .expect("target is at root.version"); assert!(!existed_again); Ok(()) @@ -5442,7 +5475,8 @@ mod large_subtree_chunking_tests { let (existed, deleted_nodes) = store .delete_subtree_atomic(&root.id, root.version, &subtree_ids, None) - .await?; + .await? + .expect("target is at root.version"); assert!(existed); assert_eq!(deleted_nodes.len(), MULTI_CHUNK_COUNT + 1); diff --git a/packages/core/src/services/node_service/crud.rs b/packages/core/src/services/node_service/crud.rs index 4651c5c5e..f9bbbc717 100644 --- a/packages/core/src/services/node_service/crud.rs +++ b/packages/core/src/services/node_service/crud.rs @@ -1935,19 +1935,8 @@ impl NodeService { self.client_id.clone(), ) .await - .map_err(|e| { - let msg = e.to_string(); - // Parse version_conflict sentinel: "version_conflict:::" - if let Some(rest) = msg.strip_prefix("version_conflict:") { - let parts: Vec<&str> = rest.splitn(3, ':').collect(); - if parts.len() == 3 { - let exp: i64 = parts[1].parse().unwrap_or(expected_version); - let act: i64 = parts[2].parse().unwrap_or(0); - return NodeServiceError::version_conflict(node_id, exp, act); - } - } - NodeServiceError::query_failed(msg) - })?; + .map_err(|e| NodeServiceError::query_failed(e.to_string()))? + .map_err(|c| NodeServiceError::version_conflict(c.node_id, c.expected, c.actual))?; if !existed { return Ok(crate::models::DeleteResult { diff --git a/packages/core/src/services/node_service/hierarchy.rs b/packages/core/src/services/node_service/hierarchy.rs index df393fd50..449f54528 100644 --- a/packages/core/src/services/node_service/hierarchy.rs +++ b/packages/core/src/services/node_service/hierarchy.rs @@ -713,9 +713,9 @@ impl NodeService { ))); } - // Root nodes have no has_child edge — the in-transaction DELETE would - // return 0 changes and be misidentified as a version conflict. Reject - // root nodes explicitly so callers get a clear InvalidParent error. + // Root nodes have no has_child edge to replace, so the in-transaction + // swap would fail with a generic store error. Reject them here so + // callers get a clear hierarchy-violation error before any write. let Some(former_parent) = self.get_parent(node_id).await?.map(|p| p.id) else { return Err(NodeServiceError::hierarchy_violation(format!( "Root node '{}' cannot be batch-moved (no parent edge to replace)", @@ -758,17 +758,11 @@ impl NodeService { &children_with_versions, ) .await - .map_err(|e| { - let msg = e.to_string(); - // The store embeds the node ID in the error string: - // "VERSION_CONFLICT: node '' ...". Parse it out so - // the caller gets an actionable conflict message. - if let Some(rest) = msg.strip_prefix("VERSION_CONFLICT: node '") { - let node_id = rest.split('\'').next().unwrap_or("unknown"); - NodeServiceError::version_conflict(node_id, 0, 0) - } else { - NodeServiceError::query_failed(msg) - } + .map_err(|e| NodeServiceError::query_failed(e.to_string()))? + // Returning the conflict as `Err` from this closure is what + // rolls back the edges already swapped for earlier children. + .map_err(|c| { + NodeServiceError::version_conflict(c.node_id, c.expected, c.actual) })?; let mut updated = Vec::with_capacity(nodes.len()); diff --git a/packages/core/src/services/node_service/mod.rs b/packages/core/src/services/node_service/mod.rs index 05b105fbd..28bcc4a01 100644 --- a/packages/core/src/services/node_service/mod.rs +++ b/packages/core/src/services/node_service/mod.rs @@ -5576,13 +5576,19 @@ mod tests { .move_children_to_parent(&new_parent_id, &children) .await; - assert!(result.is_err(), "stale version should cause failure"); - let err = result.unwrap_err(); - assert!( - matches!(err, NodeServiceError::VersionConflict { .. }), - "expected VersionConflict, got {:?}", - err - ); + let child2_version = service.get_node(&child2_id).await.unwrap().unwrap().version; + match result { + Err(NodeServiceError::VersionConflict { + node_id, + expected_version, + actual_version, + }) => { + assert_eq!(node_id, child2_id); + assert_eq!(expected_version, stale_version); + assert_eq!(actual_version, child2_version); + } + other => panic!("expected VersionConflict, got {other:?}"), + } // ALL-OR-NOTHING: child1 must still be under old_parent, not new_parent let old_children = service.get_children(&old_parent_id).await.unwrap(); @@ -5649,7 +5655,8 @@ mod tests { &[(child_id.as_str(), version_before)], ) .await - .map_err(|e| NodeServiceError::query_failed(e.to_string()))?; + .map_err(|e| NodeServiceError::query_failed(e.to_string()))? + .expect("the edge swap runs at the child's current version"); // A concurrent writer bumped the child after the edge swap. service @@ -5746,6 +5753,24 @@ mod tests { } } } + if let Err(err) = &moved { + match err { + NodeServiceError::VersionConflict { + node_id, + expected_version, + actual_version, + } => { + assert_eq!(node_id, &child_id); + assert_eq!(*expected_version, version); + assert_eq!( + *actual_version, + version + 1, + "the conflict must carry the concurrent writer's version (iteration {i})" + ); + } + other => panic!("expected VersionConflict (iteration {i}), got {other:?}"), + } + } if moved.is_ok() { assert_eq!(parent.as_deref(), Some(new_parent_id.as_str())); assert_eq!(relationship_events, 1); @@ -6007,12 +6032,20 @@ mod tests { .unwrap(); // Delete with stale version → VersionConflict, nothing deleted + let root_version = service.get_node(&root_id).await.unwrap().unwrap().version; let result = service.delete_node(&root_id, 999).await; - assert!( - matches!(result, Err(NodeServiceError::VersionConflict { .. })), - "expected VersionConflict, got {:?}", - result - ); + match result { + Err(NodeServiceError::VersionConflict { + node_id, + expected_version, + actual_version, + }) => { + assert_eq!(node_id, root_id); + assert_eq!(expected_version, 999); + assert_eq!(actual_version, root_version); + } + other => panic!("expected VersionConflict, got {other:?}"), + } // Subtree intact assert!(service.get_node(&root_id).await.unwrap().is_some()); From 69a005332b6d04da737933d187784346be7bf549 Mon Sep 17 00:00:00 2001 From: malibio Date: Fri, 25 Sep 2026 19:36:04 +0200 Subject: [PATCH 2/3] Point the rename_schema_field allowlist entry at validate_schema_field_rename The own-fields read moved into a validation helper when rename_schema_field was hardened, so the extends-chain-blindness guard no longer matched its allowlist entry and failed on main. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/core/tests/schema_chain_blindness_guard_test.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/core/tests/schema_chain_blindness_guard_test.rs b/packages/core/tests/schema_chain_blindness_guard_test.rs index 68b8c298e..bd33bf49b 100644 --- a/packages/core/tests/schema_chain_blindness_guard_test.rs +++ b/packages/core/tests/schema_chain_blindness_guard_test.rs @@ -197,7 +197,7 @@ const ALLOWLIST: &[Allowed] = &[ }, Allowed { file: "services/node_service/schema.rs", - function: "rename_schema_field", + function: "validate_schema_field_rename", field_kind: "fields", why: "Two own-only reads, both intentional. (1) The source field \ being renamed must be OWN-declared on type_id — you can only \ @@ -217,7 +217,7 @@ const ALLOWLIST: &[Allowed] = &[ file: "services/node_service/schema.rs", function: "update_schema_field_friendly_name", field_kind: "fields", - why: "Same reasoning as rename_schema_field's source-field check: a \ + why: "Same reasoning as validate_schema_field_rename's source-field check: a \ friendly_name relabel is a modification of a field definition, \ only valid for a field type_id itself declares.", }, From 667be804b1b7b1930752d8a7a1b529d3cac2f834 Mon Sep 17 00:00:00 2001 From: malibio Date: Fri, 25 Sep 2026 19:56:09 +0200 Subject: [PATCH 3/3] Address review: map VersionConflict via From, pin root-child error - impl From for NodeServiceError; both service call sites now propagate it with ? instead of repeating the mapping closure - root-child store test asserts the missing-parent-edge error, not only that the error is not a conflict Skipped: a child deleted between pre-validation and the transaction now surfaces as QueryFailed rather than a (fabricated 0/0) conflict. Reporting it as a conflict would need an invented actual version; the window is narrow and the batch still rolls back. Revisit if the frontend needs a typed not-found for batch moves. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/core/src/db/sqlite_store/mod.rs | 4 ++++ packages/core/src/services/error.rs | 6 ++++++ packages/core/src/services/node_service/crud.rs | 3 +-- packages/core/src/services/node_service/hierarchy.rs | 10 ++++------ 4 files changed, 15 insertions(+), 8 deletions(-) diff --git a/packages/core/src/db/sqlite_store/mod.rs b/packages/core/src/db/sqlite_store/mod.rs index 5287bd5f2..748f06ece 100644 --- a/packages/core/src/db/sqlite_store/mod.rs +++ b/packages/core/src/db/sqlite_store/mod.rs @@ -2281,6 +2281,10 @@ mod tests { err.downcast_ref::().is_none(), "a missing parent edge is not a version conflict, got: {err:#}" ); + assert!( + err.to_string().contains("has no parent edge to replace"), + "expected the missing-parent-edge error, got: {err:#}" + ); assert_eq!(store.get_parent_id(&root_id).await?, None); Ok(()) diff --git a/packages/core/src/services/error.rs b/packages/core/src/services/error.rs index c400e3018..357747db2 100644 --- a/packages/core/src/services/error.rs +++ b/packages/core/src/services/error.rs @@ -173,6 +173,12 @@ pub enum NodeServiceError { }, } +impl From for NodeServiceError { + fn from(c: crate::db::VersionConflict) -> Self { + Self::version_conflict(c.node_id, c.expected, c.actual) + } +} + impl NodeServiceError { /// Create a node not found error pub fn node_not_found(id: impl Into) -> Self { diff --git a/packages/core/src/services/node_service/crud.rs b/packages/core/src/services/node_service/crud.rs index f9bbbc717..c7ccaee13 100644 --- a/packages/core/src/services/node_service/crud.rs +++ b/packages/core/src/services/node_service/crud.rs @@ -1935,8 +1935,7 @@ impl NodeService { self.client_id.clone(), ) .await - .map_err(|e| NodeServiceError::query_failed(e.to_string()))? - .map_err(|c| NodeServiceError::version_conflict(c.node_id, c.expected, c.actual))?; + .map_err(|e| NodeServiceError::query_failed(e.to_string()))??; if !existed { return Ok(crate::models::DeleteResult { diff --git a/packages/core/src/services/node_service/hierarchy.rs b/packages/core/src/services/node_service/hierarchy.rs index 449f54528..8b809cb2b 100644 --- a/packages/core/src/services/node_service/hierarchy.rs +++ b/packages/core/src/services/node_service/hierarchy.rs @@ -752,18 +752,16 @@ impl NodeService { .iter() .map(|(id, ver)| (id.as_str(), *ver)) .collect(); + // The second `?` turns a `VersionConflict` into this + // closure's `Err`, which is what rolls back the edges + // already swapped for earlier children. let orders = crate::db::SqliteStore::move_children_to_parent_in_tx( tx.store_tx(), &new_parent_id, &children_with_versions, ) .await - .map_err(|e| NodeServiceError::query_failed(e.to_string()))? - // Returning the conflict as `Err` from this closure is what - // rolls back the edges already swapped for earlier children. - .map_err(|c| { - NodeServiceError::version_conflict(c.node_id, c.expected, c.actual) - })?; + .map_err(|e| NodeServiceError::query_failed(e.to_string()))??; let mut updated = Vec::with_capacity(nodes.len()); for ((node, order), former_parent) in