From a6315448677e36edce15c38443fcf73f41c2b40d Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Wed, 11 Nov 2026 01:36:21 +0100 Subject: [PATCH 1/3] fix(store): reconcile graph-neutral operation conflicts incrementally agent-identity: agent/compoundingtech/st-conversations agent-persona: generalist agent-supervisor: unavailable agent-tool: unknown agent-runtime: unknown agent-model: unknown --- crates/fractal-store/src/store.rs | 33 +++++--- .../src/store/message_send_tests.rs | 77 +++++++++++++++++++ crates/fractal-store/src/store/tests.rs | 6 +- docs/st3/large-value-sites.json | 3 + ...graph-neutral-operation-conflict-9101.json | 11 +++ 5 files changed, 117 insertions(+), 13 deletions(-) create mode 100644 release-notes/graph-neutral-operation-conflict-9101.json diff --git a/crates/fractal-store/src/store.rs b/crates/fractal-store/src/store.rs index ee2cdfae1e4..d355605fb0b 100644 --- a/crates/fractal-store/src/store.rs +++ b/crates/fractal-store/src/store.rs @@ -17558,11 +17558,10 @@ pub fn try_project_simple_replication_tx( } } } - // Event-only claims do not change the structural graph. Their operation - // registry can advance with the frontier. Message sends reconcile an ambiguous - // receipt from that operation's indexed history, using the full replay's reducer. - // Other kinds retain the full replay guard for ambiguous IDs. - let mut reconciled_message_operations = BTreeSet::new(); + // Event-only claims and message sends reconcile ambiguous receipts from that + // operation's indexed history, using the full replay's reducer. Structural + // kinds retain the full replay guard for ambiguous IDs. + let mut reconciled_operations = BTreeSet::new(); for claim in &claims { let Some((operation_id, request_digest)) = operation_parts(&claim.body) else { continue; @@ -17579,20 +17578,32 @@ pub fn try_project_simple_replication_tx( if repaired { continue; } - if claim.kind == "message.sent" && reconciled_message_operations.contains(operation_id) { + let graph_neutral = !replay_projects_kind(&claim.kind); + if (claim.kind == "message.sent" || graph_neutral) + && reconciled_operations.contains(operation_id) + { continue; } - if let Some((stored_digest, _, state)) = - operation_tx(transaction, operation_id).map_err(|error| { + let stored_operation = operation_tx(transaction, operation_id).map_err(|error| { + incremental_claim_error(internal(error), claim, "operation-registry") + })?; + // A checkpoint can leave only tombstones, with no operations row. Fold + // those too instead of registering just the incoming claim. + if graph_neutral && stored_operation.is_none() { + repair_operations_tx(transaction, &[operation_id.to_owned()]).map_err(|error| { incremental_claim_error(internal(error), claim, "operation-registry") - })? + })?; + reconciled_operations.insert(operation_id.to_owned()); + continue; + } + if let Some((stored_digest, _, state)) = stored_operation && (stored_digest != request_digest || state != "active") { - if claim.kind == "message.sent" { + if claim.kind == "message.sent" || graph_neutral { repair_operations_tx(transaction, &[operation_id.to_owned()]).map_err(|error| { incremental_claim_error(internal(error), claim, "operation-registry") })?; - reconciled_message_operations.insert(operation_id.to_owned()); + reconciled_operations.insert(operation_id.to_owned()); continue; } crate::profile::note("replay: operation-conflict"); diff --git a/crates/fractal-store/src/store/message_send_tests.rs b/crates/fractal-store/src/store/message_send_tests.rs index 3e48d6ed9d2..2c2839cdffd 100644 --- a/crates/fractal-store/src/store/message_send_tests.rs +++ b/crates/fractal-store/src/store/message_send_tests.rs @@ -163,3 +163,80 @@ fn message_operation_conflict_receipt_cost_depends_on_its_own_history() { ); } +#[test] +fn graph_neutral_operation_conflicts_match_replay_in_both_orders_and_after_checkpoint() { + const FLEET: &str = "018f6f0d-4a5d-7b8c-9d0e-123456789abc"; + let subject = "github-event/github/acme/garden/pull-request/7/checks/failed"; + let sources = [ + Store::open_memory("node-a").unwrap(), + Store::open_memory("node-b").unwrap(), + ]; + let claims = sources.iter().enumerate().map(|(index, source)| { + source.bind_fleet(FLEET).unwrap(); + source.append_claim(&ClaimInput { + subject: subject.into(), + kind: "github.event".into(), + actor: None, + fields: BTreeMap::from([ + ("version".into(), json!(1)), + ("item".into(), json!("resource/github/acme/garden/pull-request/7")), + ("event".into(), json!("checks-failed")), + ("observer".into(), json!(format!("observer/node-{}", index))), + ("at".into(), json!(format!("2026-11-11T00:00:0{index}Z"))), + ("detail".into(), json!({"check":format!("check-{index}")})), + ("head_sha".into(), json!(format!("{index:040x}"))), + ]), + evidence: Vec::new(), + expected_subject: None, + idempotency_key: Some(format!("github-event:{subject}")), + }).unwrap() + }).collect::>(); + let (operation, digest) = operation_parts(&claims[0].body).unwrap(); + let (other_operation, other_digest) = operation_parts(&claims[1].body).unwrap(); + assert_eq!(operation, other_operation); + assert_ne!(digest, other_digest); + assert!(!replay_projects_kind("github.event")); + assert!(!Store::simple_replication_kind("github.event")); + let exchanges = sources.iter().map(|source| { + source.export_replication_exchange(FLEET, &ReplicationInventory::default()).unwrap() + }).collect::>(); + for checkpointed in [false, true] { + for order in [[0, 1], [1, 0]] { + let store = Store::open_memory("target").unwrap(); + store.bind_fleet(FLEET).unwrap(); + store.project_replication_backlog().unwrap(); + for (position, index) in order.into_iter().enumerate() { + if checkpointed && position == 1 { + let previous = &claims[order[0]]; + let (_, previous_digest) = operation_parts(&previous.body).unwrap(); + let mut connection = store.connection.write(); + let tx = connection.transaction().unwrap(); + tx.execute(/* sqlite-large-value: allow-for-now lv-h0f43df342761 owner=fractal_store::store::message_send_tests TODO=stage-generation-and-keyed-reclamation */ "INSERT INTO checkpoint_claims(id,writer,sequence,envelope_hash,subject,kind, + actor,predecessors,operation_id,request_digest,accepted_at_unix_ms,checkpoint) + VALUES (?1,?2,1,'invented-envelope',?3,'github.event',NULL,'[]', + ?4,?5,0,'checkpoint/2026-11-11')", + params![previous.id, previous.origin, previous.subject, operation, previous_digest]).unwrap(); + tx.execute(/* sqlite-large-value: allow-for-now lv-h3c87df5ded70 owner=fractal_store::store::message_send_tests TODO=bound-or-reference-values-and-reclamation */ "DELETE FROM operations WHERE id=?1", [operation]).unwrap(); + tx.execute(/* sqlite-large-value: allow-for-now lv-hede6c7531978 owner=fractal_store::store::message_send_tests TODO=prepare-body-and-predecessor-values */ "DELETE FROM claims WHERE id=?1", [&previous.id]).unwrap(); + tx.commit().unwrap(); + assert!(operation_tx(&connection, operation).unwrap().is_none()); + } + let before = FULL_REPLAYS.with(std::cell::Cell::get); + store.receive_replication_exchange(&claims[index].origin, FLEET, &exchanges[index]).unwrap(); + store.project_replication_backlog().unwrap(); + assert_eq!(FULL_REPLAYS.with(std::cell::Cell::get), before, + "order={order:?} checkpointed={checkpointed} position={position}"); + let operations = operation_rows(&store.readers.get()); + let digests = store.replication_snapshot().unwrap().projection_digests.clone(); + assert!(digests.contains_key("claim_sources")); + assert!(digests.contains_key("custom_registrations")); + assert!(digests.contains_key("custom_sources")); + assert!(digests.contains_key("custom_dependencies")); + store.replay_replication_graph().unwrap(); + assert_eq!(operation_rows(&store.readers.get()), operations); + assert_eq!(store.replication_snapshot().unwrap().projection_digests, digests); + } + assert_eq!(operation_tx(&store.readers.get(), operation).unwrap().unwrap().2, "conflict"); + } + } +} diff --git a/crates/fractal-store/src/store/tests.rs b/crates/fractal-store/src/store/tests.rs index 83c2bace5eb..7597ae87a82 100644 --- a/crates/fractal-store/src/store/tests.rs +++ b/crates/fractal-store/src/store/tests.rs @@ -2328,8 +2328,10 @@ fn full_replay_logs_every_guard_and_incremental_error_once() { json!({"fields":{"state":"ready"}, "_operation":{"id":"op/malformed"}}), ), "operation-conflict" => ( - "harness.observed", - json!({"fields":{"state":"ready"}, "_operation":operation}), + "work.nudged", + json!({"fields":{"attempt":1,"agent":"agent/node.test", + "message":"message/invented-nudge","idle_since_unix_ms":"1", + "reason":"invented control","waits":[]}, "_operation":operation}), ), "incremental-error:internal" => ("intent.desired", json!({})), _ => ("harness.observed", json!({"fields":{"state":"ready"}})), diff --git a/docs/st3/large-value-sites.json b/docs/st3/large-value-sites.json index 6adaa294d2e..e1c69d94d8c 100644 --- a/docs/st3/large-value-sites.json +++ b/docs/st3/large-value-sites.json @@ -1078,6 +1078,7 @@ {"id": "lv-h0f139902fc2a", "path": "crates/fractal-seats/src/store/canonical_audit.rs", "kind": "sql-template", "fingerprint": "999b39ff0e02c6e5b0b89697bfc6fffc4afb7214067d2a76cbf41ff0d38658da", "owner": "fractal_seats::store::canonical_audit", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 1441, "excerpt": "UPDATE message_index SET closed=closed"}, {"id": "lv-h0f241e3a72c0", "path": "crates/smallclaims/src/sqlite/read_lifetime/tests.rs", "kind": "sql-template", "fingerprint": "13a97155956a550b048bef66d23d093cfba4a4785ee24a721b3c3e96511f5098", "owner": "smallclaims::sqlite::read_lifetime::tests", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 124, "excerpt": "UPDATE payload SET value=zeroblob(16384)"}, {"id": "lv-h0f3ce9d1ee52", "path": "crates/fractal-store/src/store/message_marks.rs", "kind": "unresolved-sql-sink", "fingerprint": "0ea9b7e6a7949b706e339f65714efccdcefa739776d197c0c281a953913120e5", "owner": "fractal_store::store::message_marks", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 350, "excerpt": "params ! [ message , serde_json : : to_string ( kinds ) ? , consumer ] , | row | row . get ( 0 )"}, + {"id": "lv-h0f43df342761", "path": "crates/fractal-store/src/store/message_send_tests.rs", "kind": "sql-template", "fingerprint": "2abb21ca1230e84ec0fd302ae658f78c7555d0b4aad5ae1a0bb076bc6638f020", "owner": "fractal_store::store::message_send_tests", "todo": "stage-generation-and-keyed-reclamation", "disposition": "allow-for-now", "line_hint": 58, "excerpt": "INSERT INTO checkpoint_claims(id,writer,sequence,envelope_hash,subject,kind, actor,predecessors,operation_id,request_digest,accepted_at_unix_ms,checkpoint) VALUES (?1,?2,1,'invented-envelope',?3,'gith"}, {"id": "lv-h0f4af853a427", "path": "crates/fractal-seats/src/store/reconcile_work_context/tests.rs", "kind": "unresolved-sql-sink", "fingerprint": "670380e6c1568143746eefd5717d30685a73d61436c5663c11ede8fb9d9aec6f", "owner": "fractal_seats::store::reconcile_work_context::tests", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 412, "excerpt": "& format ! ( \"EXPLAIN QUERY PLAN {query}\" )"}, {"id": "lv-h0f82774c88b3", "path": "crates/smallclaims/src/store/checkpoint_trim/pacing_tests.rs", "kind": "sql-template", "fingerprint": "11cda2afe8f046d8ab9a034d7553f23a4ee6c312f735c09f8875d271f7a57f59", "owner": "smallclaims::store::checkpoint_trim::pacing_tests", "todo": "chunk-or-reference-exact-wire-values", "disposition": "allow-for-now", "line_hint": 30, "excerpt": "INSERT INTO replica_envelopes VALUES('writer',10001,'hash')"}, {"id": "lv-h0f84e29accc7", "path": "crates/smallclaims/src/store/checkpoint_staging.rs", "kind": "sql-template", "fingerprint": "801d780479ea140546359d4a836ed44d0fcbc5a2667ea898c798df32fed9417a", "owner": "smallclaims::store::checkpoint_staging", "todo": "chunk-or-reference-exact-wire-values", "disposition": "allow-for-now", "line_hint": 643, "excerpt": "INSERT OR REPLACE INTO envelopes VALUES (?1, ?2, ?3, ?4, ?5, 1)"}, @@ -1823,6 +1824,7 @@ {"id": "lv-h3c6e65d180a0", "path": "crates/fractal-missions/src/store/missions.rs", "kind": "unresolved-sql-sink", "fingerprint": "afc23c7b6f902345ee28f20e775bcdbabac5db7207ef92eb337d90d186521297", "owner": "fractal_missions::store::missions", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 6245, "excerpt": "[ & subject ] , | row | row . get : : < _ , Option < u64 > > ( 0 ) ,"}, {"id": "lv-h3c7130318620", "path": "crates/fractal-missions/src/store/missions.rs", "kind": "sql-template", "fingerprint": "2a42e7db6035364dbfcae839df56720e6db79e733f4fdac5661f296b7a342e85", "owner": "fractal_missions::store::missions", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 4011, "excerpt": "UPDATE run_generations SET status=?2, updated_at_unix_ms=?3 WHERE id=?1"}, {"id": "lv-h3c86624acf5f", "path": "crates/smallclaims/src/store/retired.rs", "kind": "unresolved-sql-sink", "fingerprint": "14437138656c6d168b16514dd32fda0ff17204f834890b5f620e428af3cc2b11", "owner": "smallclaims::store::retired", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 140, "excerpt": "params ! [ writer , sequence , envelope_hash ] , | row | row . get ( 0 )"}, + {"id": "lv-h3c87df5ded70", "path": "crates/fractal-store/src/store/message_send_tests.rs", "kind": "sql-template", "fingerprint": "76ef26834ee7e7466075a10630f7c4d683b3ae12c37e1ad9451eacc53b7d029b", "owner": "fractal_store::store::message_send_tests", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 63, "excerpt": "DELETE FROM operations WHERE id=?1"}, {"id": "lv-h3c9d35a10089", "path": "crates/fractal-store/src/store.rs", "kind": "sql-template", "fingerprint": "192bc9a3c8768d5b2bc11ae2d6c5a403e389d9e42fccc9977a3330132232037a", "owner": "fractal_store::store", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 16846, "excerpt": "DELETE FROM mission_run_after WHERE run_id=?1"}, {"id": "lv-h3cb5f02ee7b8", "path": "crates/fractal-missions/src/store/missions.rs", "kind": "unresolved-sql-sink", "fingerprint": "156db974484d64cd6d253c16872df181d33b433c8fc4b66f873393b10c3dd28f", "owner": "fractal_missions::store::missions", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 3115, "excerpt": "& canonical_sql ( \"SELECT p.id FROM revision_proposals p JOIN mission_runs r ON r.id=p.run_id JOIN claims created ON created.subject='revision-proposal/' || p.id AND created.kind='revision-proposal.cr"}, {"id": "lv-h3cc529e49cd3", "path": "crates/fractal-store/src/store/retirements.rs", "kind": "sql-template", "fingerprint": "0092aa93d03350190dad3dc6757903831121c2d7bb4ddd3c84aea6da34dcf0b0", "owner": "fractal_store::store::retirements", "todo": "chunk-or-reference-variable-columns", "disposition": "allow-for-now", "line_hint": 5, "excerpt": "CREATE INDEX IF NOT EXISTS st3_replica_lifecycle_unavailable ON replica_records(subject_hint,kind_hint) WHERE state IN ('unknown','invalid'); DROP INDEX IF EXISTS st3_lifecycle_operation_index; CREATE"}, @@ -4848,6 +4850,7 @@ {"id": "lv-hedc149804801", "path": "crates/smallclaims/src/store/operation_claim_digests.rs", "kind": "unresolved-sql-sink", "fingerprint": "3f39022e29e62d73ec64379c1d1048906783cc860da616cd227639b577f3fa4c", "owner": "smallclaims::store::operation_claim_digests", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 161, "excerpt": "if remainder { REMAINDER_PAGE } else { NEXT_PAGE }"}, {"id": "lv-hedd9a4ccf337", "path": "crates/smallclaims/src/store/principals/migration.rs", "kind": "sql-template", "fingerprint": "483bd07eef1e384481eb64f289a83727049a93d74072fdece5ead6d31ffb51f3", "owner": "smallclaims::store::principals::migration", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 497, "excerpt": "UPDATE claim_signature_migration SET fault=NULL,legacy_fault=NULL WHERE id=1"}, {"id": "lv-hedda02447ecc", "path": "crates/smallclaims/src/store/projection_digest.rs", "kind": "unresolved-sql-sink", "fingerprint": "00ec713fc2be2060bab2be92d80982f4fe4ee4ee5c9689fe6c79bb13ac8bccb2", "owner": "smallclaims::store::projection_digest", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 89, "excerpt": "& format ! ( \"PRAGMA table_info({table})\" )"}, + {"id": "lv-hede6c7531978", "path": "crates/fractal-store/src/store/message_send_tests.rs", "kind": "sql-template", "fingerprint": "da375a6ef935682d924483cbd6c26b9c32a0a576b639560d3ec18290cfd5125b", "owner": "fractal_store::store::message_send_tests", "todo": "prepare-body-and-predecessor-values", "disposition": "allow-for-now", "line_hint": 64, "excerpt": "DELETE FROM claims WHERE id=?1"}, {"id": "lv-hedf22a898ad6", "path": "crates/fractal-store/src/store/tests.rs", "kind": "sql-template", "fingerprint": "570e2e138e8a0237449fbf8f1a93a63186d1035c78d768951caad0194fc1ffb3", "owner": "fractal_store::store::tests", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 12778, "excerpt": "DELETE FROM claim_positions WHERE claim_id=?1"}, {"id": "lv-hedf96a83a9a2", "path": "crates/smallclaims/src/store/repair_operation_tests.rs", "kind": "sql-template", "fingerprint": "da375a6ef935682d924483cbd6c26b9c32a0a576b639560d3ec18290cfd5125b", "owner": "smallclaims::store::repair_operation_tests", "todo": "prepare-body-and-predecessor-values", "disposition": "allow-for-now", "line_hint": 586, "excerpt": "DELETE FROM claims WHERE id=?1"}, {"id": "lv-hedfb698e9105", "path": "crates/fractal-store/src/store/attention_budget/tests.rs", "kind": "unresolved-sql-sink", "fingerprint": "4d9af0aa3b15ccf2041ce9a1015c22ccf0ede53bfe9f6b94c332affecb109dce", "owner": "fractal_store::store::attention_budget::tests", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 275, "excerpt": "SUBJECTS"}, diff --git a/release-notes/graph-neutral-operation-conflict-9101.json b/release-notes/graph-neutral-operation-conflict-9101.json new file mode 100644 index 00000000000..105b9d3f807 --- /dev/null +++ b/release-notes/graph-neutral-operation-conflict-9101.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Keep conflicting GitHub observation receipts incremental instead of replaying the whole graph", + "replay": {"status":"changed", "detail":"Graph-neutral claim kinds, including github.event, reconcile conflicting operation receipts through the existing indexed canonical reducer. Missing operation rows also fold checkpoint tombstones. Graph-projecting kinds retain their full-replay guard; message.sent behavior is unchanged. No migration replay is required."}, + "schema": {"status":"none", "detail":"No schema, index, trigger, claim or signature changes. Uses the existing operations row and reducer."}, + "rules": {"status":"none", "detail":"Canonical operation selection, conflict outcomes, checkpoint tombstones and repaired-claim exclusions retain the full replay's existing meaning."}, + "compatibility": {"status":"none", "detail":"No registry, wire or client changes. Signed claims and idempotency keys are unchanged."}, + "downtime": {"status":"none", "detail":"Normal binary restart only; no coordinated fleet interruption is required."}, + "manual": {"status":"none", "detail":"No live replay, database edit, heal or reset is requested."}, + "recovery": {"status":"none", "detail":"Older binaries read the same canonical operation rows and retain their former full-replay guard. No irreversible format change."} +} From 11bd608e83dc9aa7fe56132a247945696e20208b Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Wed, 11 Nov 2026 01:55:54 +0100 Subject: [PATCH 2/3] fix(store): replay operation conflicts with structural history agent-identity: agent/compoundingtech/st-conversations agent-persona: generalist agent-supervisor: unavailable agent-tool: unknown agent-runtime: unknown agent-model: unknown --- crates/fractal-store/src/store.rs | 46 +++++++++++----- .../src/store/identities_tests.rs | 52 +++++++++++++++++++ ...graph-neutral-operation-conflict-9101.json | 2 +- 3 files changed, 85 insertions(+), 15 deletions(-) diff --git a/crates/fractal-store/src/store.rs b/crates/fractal-store/src/store.rs index d355605fb0b..b72f7a265cb 100644 --- a/crates/fractal-store/src/store.rs +++ b/crates/fractal-store/src/store.rs @@ -17579,33 +17579,51 @@ pub fn try_project_simple_replication_tx( continue; } let graph_neutral = !replay_projects_kind(&claim.kind); - if (claim.kind == "message.sent" || graph_neutral) - && reconciled_operations.contains(operation_id) - { + if graph_neutral && reconciled_operations.contains(operation_id) { continue; } let stored_operation = operation_tx(transaction, operation_id).map_err(|error| { incremental_claim_error(internal(error), claim, "operation-registry") })?; - // A checkpoint can leave only tombstones, with no operations row. Fold - // those too instead of registering just the incoming claim. - if graph_neutral && stored_operation.is_none() { - repair_operations_tx(transaction, &[operation_id.to_owned()]).map_err(|error| { + let conflict = stored_operation.as_ref().is_some_and(|(digest, _, state)| { + digest.as_str() != request_digest || state != "active" + }); + // A checkpoint can leave only tombstones, with no operations row. Before + // folding either case, inspect only this operation's indexed history: + // a structural claim sharing the key needs its projections replayed too. + if conflict || (graph_neutral && stored_operation.is_none()) { + let neutral_history = graph_neutral && (|| -> rusqlite::Result { + let mut statement = transaction.prepare_cached( + "SELECT kind FROM claims INDEXED BY claims_operation_index + WHERE json_extract(body, '$._operation.id')=?1 + UNION ALL + SELECT kind FROM checkpoint_claims INDEXED BY checkpoint_claims_operation + WHERE operation_id=?1 AND operation_id IS NOT NULL", + )?; + let mut rows = statement.query([operation_id])?; + while let Some(row) = rows.next()? { + let kind = row.get_ref(0)?; + let kind = kind.as_str().map_err(|error| { + rusqlite::Error::FromSqlConversionFailure(0, kind.data_type(), Box::new(error)) + })?; + if replay_projects_kind(kind) { + return Ok(false); + } + } + Ok(true) + })().map_err(|error| { incremental_claim_error(internal(error), claim, "operation-registry") })?; - reconciled_operations.insert(operation_id.to_owned()); - continue; - } - if let Some((stored_digest, _, state)) = stored_operation - && (stored_digest != request_digest || state != "active") - { - if claim.kind == "message.sent" || graph_neutral { + if neutral_history { repair_operations_tx(transaction, &[operation_id.to_owned()]).map_err(|error| { incremental_claim_error(internal(error), claim, "operation-registry") })?; reconciled_operations.insert(operation_id.to_owned()); continue; } + let (stored_digest, state) = stored_operation.as_ref() + .map(|(digest, _, state)| (digest.as_str(), state.as_str())) + .unwrap_or(("", "missing")); crate::profile::note("replay: operation-conflict"); return Ok(IncrementalProjection::ReplayWithContext { reason: "operation-conflict", diff --git a/crates/fractal-store/src/store/identities_tests.rs b/crates/fractal-store/src/store/identities_tests.rs index 8688842f1e2..f3b2cb25f55 100644 --- a/crates/fractal-store/src/store/identities_tests.rs +++ b/crates/fractal-store/src/store/identities_tests.rs @@ -410,3 +410,55 @@ proptest! { } } } + +#[test] +fn mixed_structural_and_neutral_operation_history_requires_replay_in_both_orders() { + for neutral_first in [false, true] { + let source = node("node-a", None, None); + let store = node("node-b", None, None); + declare(&source, ORIGINAL); + sync(&source, &store); + store.resolve_identity(IdentityKind::Agent, ORIGINAL).unwrap(); + let receipt = source.rename_identity( + IdentityKind::Agent, &request(ORIGINAL, RENAMED, "mixed-operation"), &[], + ).unwrap(); + let rename = source.claim_by_id(&receipt.claim_id).unwrap().unwrap(); + let (operation, _) = operation_parts(&rename.body).unwrap(); + let neutral_body = json!({"fields": {}, "_operation": { + "id": operation, "request_digest": "invented-neutral-digest" + }}); + for neutral in [neutral_first, !neutral_first] { + { + let mut connection = store.connection.write(); + let tx = connection.transaction().unwrap(); + let (subject, kind, actor, body) = if neutral { + ("github-event/github/acme/garden/pull-request/7/checks/failed", + "github.event", None, &neutral_body) + } else { + (rename.subject.as_str(), rename.kind.as_str(), rename.actor.as_deref(), &rename.body) + }; + append_claim_record_tx(&tx, source.origin(), subject, kind, actor, body, &[], None).unwrap(); + tx.commit().unwrap(); + } + if neutral != neutral_first { + let mut connection = store.connection.write(); + let tx = connection.transaction().unwrap(); + let result = try_project_simple_replication_tx( + &tx, store.origin(), current_index_tx(&tx).unwrap(), + ).unwrap(); + assert!(matches!(result, IncrementalProjection::ReplayWithContext { + reason: "operation-conflict", .. + }), "neutral_first={neutral_first}"); + tx.rollback().unwrap(); + } + store.project_replication_backlog().unwrap(); + } + assert_eq!(operation_tx(&store.readers.get(), operation).unwrap().unwrap().2, "conflict"); + let error = store.resolve_identity(IdentityKind::Agent, ORIGINAL).unwrap_err(); + assert_eq!(error.code, "lifecycle-unavailable"); + assert_eq!(error.details["reason"], "conflict"); + let digests = store.replication_snapshot().unwrap().projection_digests.clone(); + store.replay_replication_graph().unwrap(); + assert_eq!(store.replication_snapshot().unwrap().projection_digests, digests); + } +} diff --git a/release-notes/graph-neutral-operation-conflict-9101.json b/release-notes/graph-neutral-operation-conflict-9101.json index 105b9d3f807..d7a80456ab7 100644 --- a/release-notes/graph-neutral-operation-conflict-9101.json +++ b/release-notes/graph-neutral-operation-conflict-9101.json @@ -1,7 +1,7 @@ { "version": 1, "summary": "Keep conflicting GitHub observation receipts incremental instead of replaying the whole graph", - "replay": {"status":"changed", "detail":"Graph-neutral claim kinds, including github.event, reconcile conflicting operation receipts through the existing indexed canonical reducer. Missing operation rows also fold checkpoint tombstones. Graph-projecting kinds retain their full-replay guard; message.sent behavior is unchanged. No migration replay is required."}, + "replay": {"status":"changed", "detail":"Graph-neutral claim kinds, including github.event and message.sent, reconcile operation receipts through the existing indexed canonical reducer only when every stored claim and checkpoint tombstone for that operation is graph-neutral. Missing operation rows also fold checkpoint tombstones. Mixed-kind histories involving a graph-projecting claim retain the full-replay guard so naming and other structural projections remain canonical. No migration replay is required."}, "schema": {"status":"none", "detail":"No schema, index, trigger, claim or signature changes. Uses the existing operations row and reducer."}, "rules": {"status":"none", "detail":"Canonical operation selection, conflict outcomes, checkpoint tombstones and repaired-claim exclusions retain the full replay's existing meaning."}, "compatibility": {"status":"none", "detail":"No registry, wire or client changes. Signed claims and idempotency keys are unchanged."}, From af65763595791f7b50e46fcfd67ef1858a49cb3b Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Wed, 11 Nov 2026 02:16:05 +0100 Subject: [PATCH 3/3] test(store): admit both conflict claims before replay parity checks agent-identity: unknown agent-persona: unknown agent-supervisor: unavailable agent-tool: unknown agent-runtime: unknown agent-model: unknown --- .../src/store/message_send_tests.rs | 41 ++++++++++++++----- ...graph-neutral-operation-conflict-9101.json | 2 +- 2 files changed, 32 insertions(+), 11 deletions(-) diff --git a/crates/fractal-store/src/store/message_send_tests.rs b/crates/fractal-store/src/store/message_send_tests.rs index 2c2839cdffd..58dbc72ef9d 100644 --- a/crates/fractal-store/src/store/message_send_tests.rs +++ b/crates/fractal-store/src/store/message_send_tests.rs @@ -200,11 +200,14 @@ fn graph_neutral_operation_conflicts_match_replay_in_both_orders_and_after_check let exchanges = sources.iter().map(|source| { source.export_replication_exchange(FLEET, &ReplicationInventory::default()).unwrap() }).collect::>(); + assert!(exchanges.iter().all(|exchange| !exchange.envelopes.is_empty())); for checkpointed in [false, true] { for order in [[0, 1], [1, 0]] { let store = Store::open_memory("target").unwrap(); store.bind_fleet(FLEET).unwrap(); store.project_replication_backlog().unwrap(); + let context = format!("order={order:?} checkpointed={checkpointed}"); + let mut admitted = [false; 2]; for (position, index) in order.into_iter().enumerate() { if checkpointed && position == 1 { let previous = &claims[order[0]]; @@ -223,20 +226,38 @@ fn graph_neutral_operation_conflicts_match_replay_in_both_orders_and_after_check } let before = FULL_REPLAYS.with(std::cell::Cell::get); store.receive_replication_exchange(&claims[index].origin, FLEET, &exchanges[index]).unwrap(); + // Receiving stores envelopes; admission materializes their claims. + store.validate_replication_backlog().unwrap(); + assert!(store.claim_by_id(&claims[index].id).unwrap().is_some(), + "{context} position={position}: incoming claim must be admitted"); + admitted[index] = true; + let prior_state = operation_tx(&store.readers.get(), operation).unwrap() + .map(|(_, _, state)| state); + assert_eq!(prior_state.as_deref(), + if position == 0 || checkpointed { None } else { Some("active") }, + "{context} position={position}: repair must still be pending"); store.project_replication_backlog().unwrap(); assert_eq!(FULL_REPLAYS.with(std::cell::Cell::get), before, "order={order:?} checkpointed={checkpointed} position={position}"); - let operations = operation_rows(&store.readers.get()); - let digests = store.replication_snapshot().unwrap().projection_digests.clone(); - assert!(digests.contains_key("claim_sources")); - assert!(digests.contains_key("custom_registrations")); - assert!(digests.contains_key("custom_sources")); - assert!(digests.contains_key("custom_dependencies")); - store.replay_replication_graph().unwrap(); - assert_eq!(operation_rows(&store.readers.get()), operations); - assert_eq!(store.replication_snapshot().unwrap().projection_digests, digests); + // An admitted digest mismatch (or tombstone-only history) must + // actually transition to conflict here, without a replay. + let state = operation_tx(&store.readers.get(), operation).unwrap() + .unwrap_or_else(|| panic!("{context} position={position}: missing repaired operation")).2; + assert_eq!(state, if position == 0 { "active" } else { "conflict" }, + "{context} position={position}: indexed operation repair must run"); } - assert_eq!(operation_tx(&store.readers.get(), operation).unwrap().unwrap().2, "conflict"); + // Never compare two empty projections: both claims were admitted, + // and the second incremental projection already produced conflict. + assert_eq!(admitted, [true, true], "{context}"); + let operations = operation_rows(&store.readers.get()); + let digests = store.replication_snapshot().unwrap().projection_digests.clone(); + assert!(digests.contains_key("claim_sources")); + assert!(digests.contains_key("custom_registrations")); + assert!(digests.contains_key("custom_sources")); + assert!(digests.contains_key("custom_dependencies")); + store.replay_replication_graph().unwrap(); + assert_eq!(operation_rows(&store.readers.get()), operations, "{context}"); + assert_eq!(store.replication_snapshot().unwrap().projection_digests, digests, "{context}"); } } } diff --git a/release-notes/graph-neutral-operation-conflict-9101.json b/release-notes/graph-neutral-operation-conflict-9101.json index d7a80456ab7..bc2752d6927 100644 --- a/release-notes/graph-neutral-operation-conflict-9101.json +++ b/release-notes/graph-neutral-operation-conflict-9101.json @@ -3,7 +3,7 @@ "summary": "Keep conflicting GitHub observation receipts incremental instead of replaying the whole graph", "replay": {"status":"changed", "detail":"Graph-neutral claim kinds, including github.event and message.sent, reconcile operation receipts through the existing indexed canonical reducer only when every stored claim and checkpoint tombstone for that operation is graph-neutral. Missing operation rows also fold checkpoint tombstones. Mixed-kind histories involving a graph-projecting claim retain the full-replay guard so naming and other structural projections remain canonical. No migration replay is required."}, "schema": {"status":"none", "detail":"No schema, index, trigger, claim or signature changes. Uses the existing operations row and reducer."}, - "rules": {"status":"none", "detail":"Canonical operation selection, conflict outcomes, checkpoint tombstones and repaired-claim exclusions retain the full replay's existing meaning."}, + "rules": {"status":"none", "detail":"Canonical operation selection, conflict outcomes, checkpoint tombstones and repaired-claim exclusions retain the full replay's existing meaning. Regression coverage admits received envelopes and asserts both claims arrived and the operation actually transitions to conflict before comparing full-replay digests."}, "compatibility": {"status":"none", "detail":"No registry, wire or client changes. Signed claims and idempotency keys are unchanged."}, "downtime": {"status":"none", "detail":"Normal binary restart only; no coordinated fleet interruption is required."}, "manual": {"status":"none", "detail":"No live replay, database edit, heal or reset is requested."},