diff --git a/crates/fractal-github/src/idle_holds.rs b/crates/fractal-github/src/idle_holds.rs index 7af86694938..94b477c6dbe 100644 --- a/crates/fractal-github/src/idle_holds.rs +++ b/crates/fractal-github/src/idle_holds.rs @@ -30,9 +30,12 @@ const QUEUE_EXIT_STATE_SQL: &str = "SELECT json_extract(body, '$.fields'), accep ORDER BY length(accepted_at_unix_ms) DESC,accepted_at_unix_ms DESC,store_index DESC LIMIT ?2"; -/// When a thread's `merge-queue-left` events were written: one subject per exit. +/// One canonical first observation per exit: another poller's observation is not a new exit. const QUEUE_EXIT_EVENT_SQL: &str = "SELECT accepted_at_unix_ms FROM claims INDEXED BY claims_subject_kind_accepted_index - WHERE subject>=?1 AND subject=?1 AND subject Result> { let upper = format!("{subject}0"); - let mut events = connection.prepare_cached(/* sqlite-large-value: allow-for-now lv-h13f5ed40c752 owner=fractal_github::idle_holds TODO=trace-statement-and-value-representation */ QUEUE_EXIT_EVENT_SQL)?; + let mut events = connection.prepare_cached(/* sqlite-large-value: allow-for-now lv-haa1264e7b371 owner=fractal_github::idle_holds TODO=trace-statement-and-value-representation */ &smallclaims::store::canonical::canonical_sql(QUEUE_EXIT_EVENT_SQL))?; let times = events .query_map(params![subject, upper], |row| row.get::<_, String>(0))? .collect::>>()?; @@ -409,6 +412,28 @@ mod tests { "at": "2026-10-30T12:00:00.000Z"})); assert_eq!(store.thread_left_merge_queue(&thread, 0).unwrap(), Some(event), "the newest exit event"); assert_eq!(store.thread_left_merge_queue(&thread, event).unwrap(), None); + // A second observation can arrive by replication despite local Once admission. + // Simulate that raw history with the append primitive, preserving the old payload. + { + let mut connection = store.connection.write(); + let transaction = connection.transaction().unwrap(); + smallclaims::store::append_claim_tx( + &transaction, &store.origin, + "github-event/github/acme/garden/pull-request/12/merge-queue-left/2", + "github.event", None, + &json!({"fields": { + "version": 1, "item": item, "event": "merge-queue-left", + "observer": "observer/another-poller", "at": "2026-11-11T10:00:00Z", + "head_sha": "another-head", "detail": {} + }}), + &[], None, + ).unwrap(); + transaction.commit().unwrap(); + } + assert_eq!(store.thread_left_merge_queue(&thread, 0).unwrap(), Some(event), + "another observation must not move the canonical exit"); + assert_eq!(store.thread_left_merge_queue(&thread, event).unwrap(), None, + "another observation must not wake an idle holder again"); append(&store, item, "vcs.item.state", compact(json!({"state": "closed", "merged": true}))); assert_eq!(store.thread_left_merge_queue(&thread, 0).unwrap(), None, "a merged thread did not stall"); } diff --git a/crates/fractal-reconcile/src/reconcile.rs b/crates/fractal-reconcile/src/reconcile.rs index 52bf4ce18eb..5f4424625e2 100644 --- a/crates/fractal-reconcile/src/reconcile.rs +++ b/crates/fractal-reconcile/src/reconcile.rs @@ -19409,6 +19409,7 @@ fn mirror_observation( .admit(subject, fractal_schema::github_plugin::EVENT, &fields) .map_err(anyhow::Error::new) .and_then(|()| { + let key = github_events::observation_key(subject, &fields)?; store.append_claim(&ClaimInput { subject: subject.to_owned(), kind: fractal_schema::github_plugin::EVENT.into(), @@ -19416,7 +19417,7 @@ fn mirror_observation( fields, evidence: Vec::new(), expected_subject: None, - idempotency_key: Some(format!("github-event:{subject}")), + idempotency_key: Some(key), }).map_err(|error| anyhow::anyhow!(error.message)) }); if let Err(error) = written { @@ -43106,6 +43107,8 @@ subscription "issues" { observer "observer/repo"; on "issues"; to "agent/example } } let store = Arc::new(Store::open_memory("node").unwrap()); + const EVENT_FLEET: &str = "018f6f0d-4a5d-7b8c-9d0e-123456789abc"; + store.bind_fleet(EVENT_FLEET).unwrap(); apply_source( &store, r#"version 2 @@ -43163,6 +43166,36 @@ subscription "comments" { observer "observer/repo"; on "comments"; to "agent/exa poll(plugin.clone(), listing("open", "Orchid", &[1]), "c2").await; assert_eq!(delivered(), 1); assert_eq!(count("resource.observed"), snapshots + 1, "a comment writes a snapshot"); + // An older poller raced this look, before receiving its event. Its global operation + // and node-local payload must not cause another listing row or subscription delivery. + let old = Store::open_memory("old-poller").unwrap(); + old.bind_fleet(EVENT_FLEET).unwrap(); + let subject = "github-event/repo/issue/1/comment/1"; + old.append_claim(&ClaimInput { + subject: subject.into(), kind: fractal_schema::github_plugin::EVENT.into(), + actor: None, + fields: BTreeMap::from([ + ("version".into(), serde_json::json!(1)), + ("item".into(), serde_json::json!(item)), + ("event".into(), serde_json::json!("comment")), + ("observer".into(), serde_json::json!("observer/old")), + ("at".into(), serde_json::json!("2026-11-11T10:00:01Z")), + ("detail".into(), serde_json::json!({"author": "fern"})), + ]), + evidence: Vec::new(), expected_subject: None, + idempotency_key: Some(format!("github-event:{subject}")), + }).unwrap(); + let exchange = old.export_replication_exchange_answering( + EVENT_FLEET, &store.replication_inventory().unwrap(), + &store.replication_signature_requests().unwrap(), + ).unwrap(); + store.receive_replication_exchange(&old.origin, EVENT_FLEET, &exchange).unwrap(); + store.validate_replication_backlog().unwrap(); + store.project_replication_backlog().unwrap(); + assert_eq!(store.claims_for(subject, None).unwrap().len(), 2); + assert_eq!(store.github_events("resource/repo", Some(1), 10).unwrap().len(), 1); + assert_eq!(store.idempotency_conflicts(10).unwrap().0, 0); + assert_eq!(delivered(), 1, "event provenance does not deliver mail"); poll(plugin.clone(), listing("open", "Orchid", &[1]), "c3").await; assert_eq!(delivered(), 1, "a known comment is not delivered again"); diff --git a/crates/fractal-reconcile/src/reconcile/github_events.rs b/crates/fractal-reconcile/src/reconcile/github_events.rs index 4ff55a13aa3..390027c9355 100644 --- a/crates/fractal-reconcile/src/reconcile/github_events.rs +++ b/crates/fractal-reconcile/src/reconcile/github_events.rs @@ -5,6 +5,17 @@ //! mirror is new, wakes nobody. use serde_json::{Map, Value, json}; +use std::collections::BTreeMap; + +/// The operation names this observation, not the global event subject. Keep every payload +/// field in the fingerprint: even the same observer on two pollers can see different facts. +pub(super) fn observation_key(subject: &str, fields: &BTreeMap) -> anyhow::Result { + let fields = fields.iter() + .map(|(name, value)| (name, smallclaims::hash::canonical_json_value(value))) + .collect::>(); + let digest = smallclaims::hash::canonical_hash(&("github-event-observation.v1", subject, fields))?; + Ok(format!("github-event-observation:v1:{digest}")) +} /// One thing that happened to one item. #[derive(Clone, Debug, PartialEq)] @@ -171,6 +182,98 @@ pub fn view(event: &Event, item: &str, subject: &str, at: &str) -> Value { mod tests { use super::*; + fn event_claim(observer: &str, at: &str) -> crate::model::ClaimInput { + let subject = "github-event/github/acme/garden/issue/7/closed"; + let fields = BTreeMap::from([ + ("version".into(), json!(1)), + ("item".into(), json!("resource/github/acme/garden/issue/7")), + ("event".into(), json!("closed")), + ("observer".into(), json!(observer)), + ("at".into(), json!(at)), + ("detail".into(), json!({"state": "closed"})), + ]); + let key = observation_key(subject, &fields).unwrap(); + crate::model::ClaimInput { + subject: subject.into(), kind: fractal_schema::github_plugin::EVENT.into(), + actor: None, fields, evidence: Vec::new(), expected_subject: None, + idempotency_key: Some(key), + } + } + + #[test] + fn observation_keys_cover_every_field_and_canonicalize_nested_objects() { + let input = event_claim("observer/a", "2026-11-11T10:00:00Z"); + let original = observation_key(&input.subject, &input.fields).unwrap(); + for (name, value) in [ + ("observer", json!("observer/b")), + ("at", json!("2026-11-11T10:00:01Z")), + ("head_sha", json!("abc")), + ("detail", json!({"state": "open"})), + ("event", json!("reopened")), + ("version", json!(2)), + ("item", json!("resource/github/acme/garden/issue/8")), + ] { + let mut fields = input.fields.clone(); + fields.insert(name.into(), value); + assert_ne!(observation_key(&input.subject, &fields).unwrap(), original, "{name}"); + } + assert_ne!(observation_key("github-event/github/acme/garden/issue/8/closed", &input.fields).unwrap(), original); + let mut fields = input.fields; + fields.insert("detail".into(), serde_json::from_str(r#"{"b":{"y":2,"x":1},"a":0}"#).unwrap()); + let first = observation_key(&input.subject, &fields).unwrap(); + fields.insert("detail".into(), serde_json::from_str(r#"{"a":0,"b":{"x":1,"y":2}}"#).unwrap()); + assert_eq!(observation_key(&input.subject, &fields).unwrap(), first); + } + + /// Independent writers race before either sees the other's event. Old global keys and + /// new observation keys must fold by subject, regardless of replication arrival order. + #[test] + fn concurrent_event_observations_and_legacy_claims_fold_once_without_conflicts() { + use crate::store::Store; + const FLEET: &str = "018f6f0d-4a5d-7b8c-9d0e-123456789abc"; + for (other_observer, legacy) in [("observer/b", false), ("observer/a", false), ("observer/b", true)] { + let a = Store::open_memory("alder").unwrap(); + let b = Store::open_memory("birch").unwrap(); + let forward = Store::open_memory("cedar").unwrap(); + let reverse = Store::open_memory("dogwood").unwrap(); + for store in [&a, &b, &forward, &reverse] { + store.bind_fleet(FLEET).unwrap(); + } + let input = event_claim("observer/a", "2026-11-11T10:00:00Z"); + let first = a.append_claim(&input).unwrap(); + assert_eq!(a.append_claim(&input).unwrap().id, first.id, "exact retry"); + let mut other = event_claim(other_observer, "2026-11-11T10:00:01Z"); + if legacy { + other.idempotency_key = Some(format!("github-event:{}", other.subject)); + } + let second = b.append_claim(&other).unwrap(); + assert_ne!(first.body["_operation"]["id"], second.body["_operation"]["id"]); + for (target, sources) in [(&forward, [&a, &b]), (&reverse, [&b, &a])] { + for source in sources { + let exchange = source.export_replication_exchange_answering( + FLEET, &target.replication_inventory().unwrap(), + &target.replication_signature_requests().unwrap(), + ).unwrap(); + target.receive_replication_exchange(&source.origin, FLEET, &exchange).unwrap(); + target.validate_replication_backlog().unwrap(); + target.project_replication_backlog().unwrap(); + } + assert_eq!(target.idempotency_conflicts(10).unwrap().0, 0); + assert_eq!(target.claims_for(&input.subject, None).unwrap().len(), 2, "provenance survives"); + let events = target.github_events("resource/github/acme/garden", Some(7), 10).unwrap(); + assert_eq!(events.len(), 1, "one event, not one row per observation"); + assert_eq!(events, target.github_events("resource/github/acme/garden", None, 1).unwrap()); + target.replay_replication_graph().unwrap(); + assert_eq!(events, target.github_events("resource/github/acme/garden", Some(7), 10).unwrap()); + } + assert_eq!( + forward.github_events("resource/github/acme/garden", None, 10).unwrap(), + reverse.github_events("resource/github/acme/garden", None, 10).unwrap(), + "canonical fold is independent of arrival", + ); + } + } + fn facts(value: Value) -> Map { value.as_object().unwrap().clone() } diff --git a/crates/fractal-store/src/store/resources.rs b/crates/fractal-store/src/store/resources.rs index ae1589ab076..6570447513b 100644 --- a/crates/fractal-store/src/store/resources.rs +++ b/crates/fractal-store/src/store/resources.rs @@ -541,10 +541,15 @@ impl Store { }; let upper = prefix_successor(&prefix).unwrap_or_default(); let connection = self.readers.get(); - let mut statement = connection.prepare_cached( + // Old global operations and new observation operations share the event subject. + // Select its canonical first observation before applying the listing limit. + let mut statement = connection.prepare_cached(/* sqlite-large-value: allow-for-now lv-h05417f7bc8ab owner=fractal_store::store::resources TODO=trace-statement-and-value-representation */ &canonical_sql( "SELECT subject, json_extract(body, '$.fields'), accepted_at_unix_ms FROM claims - WHERE subject>=?1 AND subject=?1 AND subject(0)?, row.get::<_, Option>(1)?, row.get::<_, String>(2)?)) @@ -570,7 +575,9 @@ impl Store { }))) }) .collect::>(); - events.sort_by_key(|(at, _)| std::cmp::Reverse(*at)); + events.sort_by(|(left_at, left), (right_at, right)| { + right_at.cmp(left_at).then_with(|| left["subject"].as_str().cmp(&right["subject"].as_str())) + }); events.truncate(limit); Ok(events.into_iter().map(|(_, event)| event).collect()) } diff --git a/docs/st3/large-value-sites.json b/docs/st3/large-value-sites.json index 6adaa294d2e..b24e4b97283 100644 --- a/docs/st3/large-value-sites.json +++ b/docs/st3/large-value-sites.json @@ -911,6 +911,7 @@ {"id": "lv-h05344f98e859", "path": "crates/fractal-evidence/src/store/owned_sets_tests.rs", "kind": "sql-template", "fingerprint": "213cec2e91e4c3178d12187013806b91a5557b7cc08e46780ec767bbcb655dd2", "owner": "fractal_evidence::store::owned_sets_tests", "todo": "chunk-or-reference-exact-wire-values", "disposition": "allow-for-now", "line_hint": 341, "excerpt": "UPDATE replica_records SET state='invalid' WHERE record_ref=?1"}, {"id": "lv-h054022c6a729", "path": "crates/fractal-store/src/store/resources_preparation.rs", "kind": "sql-template", "fingerprint": "6b6cfe0900359f6fd853b388da06606454a9a018f7d87d9eb528bec593052305", "owner": "fractal_store::store::resources_preparation", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 689, "excerpt": "DELETE FROM local_resource_preparation_deferred WHERE subject=?1"}, {"id": "lv-h054145fcbc3a", "path": "crates/fractal-daemon/src/store_tests/mission_eval_reuse.rs", "kind": "sql-template", "fingerprint": "49547af58a17d2594c4b7be8b97c583f8cbee6e0790ecf150ca50b5cfaaa530e", "owner": "fractal_daemon::store_tests::mission_eval_reuse", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 113, "excerpt": "UPDATE mission_revisions SET body='not-json'"}, + {"id": "lv-h05417f7bc8ab", "path": "crates/fractal-store/src/store/resources.rs", "kind": "unresolved-sql-sink", "fingerprint": "be5d1046d4bfd24f1626324f04fd298809d07ce0df697c6424ab8d70a2ee1c0b", "owner": "fractal_store::store::resources", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 546, "excerpt": "& canonical_sql ( \"SELECT subject, json_extract(body, '$.fields'), accepted_at_unix_ms FROM claims WHERE subject>=?1 AND subject=0 AND revision<9223372036854775807 \\ AND {config} AND journal_rows>=0 AND journal_rows<{} AND journal_by"}, @@ -3685,6 +3685,7 @@ {"id": "lv-ha9bd7a5e5f62", "path": "crates/fractal-missions/src/store/lanes.rs", "kind": "unresolved-sql-sink", "fingerprint": "6d62d0ae7a75c1da176f8801ec7af20309d56086a7a92d835da541a1cb9fc5a6", "owner": "fractal_missions::store::lanes", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 115, "excerpt": "& canonical_sql ( \"SELECT claims.id, claims.kind, claims.actor, claims.body, claims.accepted_at_unix_ms FROM claims JOIN batches ON batches.id=claims.batch_id WHERE claims.subject=?1 AND claims.kind I"}, {"id": "lv-ha9be26003c2f", "path": "crates/smallclaims/src/store/principals/migration.rs", "kind": "sql-template", "fingerprint": "358a7729655d9e0a0414d87b93e0299bfd0dc10768f189880862690e1a7df7ff", "owner": "smallclaims::store::principals::migration", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 528, "excerpt": "DROP TABLE {table};"}, {"id": "lv-ha9cdb844c613", "path": "crates/smallclaims/src/store/checkpoint_trim.rs", "kind": "sql-template", "fingerprint": "f83a7eeeeca74669b860ce5f1f9abb5829c39eeb987bd407c137c05393608aa9", "owner": "smallclaims::store::checkpoint_trim", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 543, "excerpt": "UPDATE projection_digest_generation SET value=value+1"}, + {"id": "lv-haa1264e7b371", "path": "crates/fractal-github/src/idle_holds.rs", "kind": "unresolved-sql-sink", "fingerprint": "9ed9319e27cf3c4023c6957e8298938778d78861b3f5b11301fe8d9bcad416ee", "owner": "fractal_github::idle_holds", "todo": "trace-statement-and-value-representation", "disposition": "allow-for-now", "line_hint": 204, "excerpt": "& smallclaims : : store : : canonical : : canonical_sql ( QUEUE_EXIT_EVENT_SQL )"}, {"id": "lv-haa329ca6b535", "path": "crates/fractal-store/src/store/authored_pull_requests_tests_shared_missions.rs", "kind": "sql-template", "fingerprint": "7ff0c3e15c3c4bd40499f52045073953856d1730001620964d3a301afd6208df", "owner": "fractal_store::store::authored_pull_requests_tests_shared_missions", "todo": "prepare-body-and-predecessor-values", "disposition": "allow-for-now", "line_hint": 14, "excerpt": "INSERT INTO claims(id,batch_id,subject,kind,origin,body,predecessors,accepted_at_unix_ms) VALUES(?1,?1,?2,?3,'node',?4,'[]',?5)"}, {"id": "lv-haa37d14feaa5", "path": "crates/fractal-missions/src/store/missions.rs", "kind": "sql-template", "fingerprint": "5d3baa4cb1a6ebdc4527226258f80f2f07dfd1bd8aa4ff43204d68c70a101ee7", "owner": "fractal_missions::store::missions", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 3319, "excerpt": "INSERT INTO idempotency(operation_id, response) VALUES (?1, ?2)"}, {"id": "lv-haa3f0bf32513", "path": "crates/fractal-store/src/store.rs", "kind": "sql-template", "fingerprint": "5bf75ea5c9c1387a53434c800b650e325f67d72956553b497fbc7a546bf3cda5", "owner": "fractal_store::store", "todo": "bound-or-reference-values-and-reclamation", "disposition": "allow-for-now", "line_hint": 11628, "excerpt": "UPDATE planning_sessions SET created_at_unix_ms=?2,updated_at_unix_ms=?2 WHERE id=?1"}, diff --git a/docs/st3/large-values.md b/docs/st3/large-values.md index b8c33e81cd3..c943700757e 100644 --- a/docs/st3/large-values.md +++ b/docs/st3/large-values.md @@ -94,6 +94,20 @@ rationale in their place. It checks neither the linked review's existence nor its approval. Owner/TODO strings alone are not cost, correctness, or storage approval; do not replace a TODO with a permanent exemption. +### GitHub event observation fold inventory + +`fractal_store::store::resources` owns `lv-h05417f7bc8ab`: the event-list reader binds +the repository's subject prefix bounds and selects the canonical first observation per event +subject. It reads existing subject, fields JSON and accepted-time TEXT values, retaining the +current inline representation rather than adding storage. The eventual reference representation +must preserve whole canonical fields and subject identity before listing limits are applied. +`fractal_github::idle_holds` owns `lv-haa1264e7b371`, replacing `lv-h13f5ed40c752`: +its queue-exit reader binds the thread's exit-subject bounds and reads one canonical accepted-time +TEXT value per event subject. The canonical selector also reads claim/batch identity and position +metadata. Both are dynamic SQL readers, not size proofs. Their maintainers retain the +`trace-statement-and-value-representation` task; representation changes must not count repeated +observations as new events or move a queue-exit time. + ### Client operation admission inventory (PR #5921) The `fractal_store::pairing` maintainer owns `lv-h37e16ce1bdcb`, diff --git a/docs/st3/plugins.md b/docs/st3/plugins.md index 6ce6c4dc9ff..b3301eaf4a1 100644 --- a/docs/st3/plugins.md +++ b/docs/st3/plugins.md @@ -133,6 +133,23 @@ own key table, so a key another poller already wrote is never reused. - **`github.event` v1**, once per event subject: `version`, `item`, `event`, `observer`, `at`, `head_sha`, and a small `detail`. Bound: 1 KiB. +The event subject is global; its operation identifies one observation, not the event itself. +The idempotency key is `github-event-observation:v1:DIGEST`, where `DIGEST` is +`smallclaims::hash::canonical_hash` of `("github-event-observation.v1", subject, fields)`, +with every field value recursively canonicalized. This includes `observer`, `at`, `head_sha` +when present, and all of `detail`: even two pollers of the same observer can differ. Exact +retries share a key; distinct observations keep their provenance without an operation conflict. +Generic claim request hashing is unchanged. + +Local admission remains once per subject. Concurrent pollers can nevertheless replicate several +observations of the same event. Graph event listings and the idle hold's queue-exit reader select +the canonical first claim per subject, including older claims with the global +`github-event:SUBJECT` operation key and node-local fields. No history is rewritten. Subscriptions +and watches still deliver from item facts, semantic event keys and existing delivery receipts, +not these observation operations, so another event observation is not another delivery. +New and old operation keys cannot collide; two unmodified old pollers can still conflict with +each other and need the separate operation-conflict recovery path. + Readers on this build take an item's newest state. A `vcs.item.state` accepted no earlier than the item's newest `resource.observed` lays its compact facts over the observed ones; facts it does not carry, such as the title or recent comments, stay. This holds for the local resource projection diff --git a/release-notes/github-event-observation-keys.json b/release-notes/github-event-observation-keys.json new file mode 100644 index 00000000000..d290bfcb50e --- /dev/null +++ b/release-notes/github-event-observation-keys.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "GitHub pollers key event operations by their complete canonical observation, preventing concurrent observers from conflicting while preserving event-level deduplication", + "replay": {"status": "none", "detail": "No replay or history rewrite. Graph event listings and idle queue-exit reads fold existing and new observations by event subject."}, + "schema": {"status": "none", "detail": "The github.event v1 subject and fields, including observer, at, head_sha and detail, are unchanged. No tables or generic claim request hashes change."}, + "rules": {"status": "none", "detail": "Checkpoint rules and digests are unchanged."}, + "compatibility": {"status": "changed", "detail": "New producers use github-event-observation:v1 plus a fingerprint of the subject and all canonical fields instead of github-event:SUBJECT. Exact retries retain their operation; distinct observations do not collide. Old global-key claims remain readable and fold together with new claims as one event. Subscription and watch delivery identities are unchanged. Two unmodified old producers may still conflict with each other; this change does not repair their historical operation conflicts."}, + "downtime": {"status": "none", "detail": "No coordinated shutdown or migration is required."}, + "manual": {"status": "none", "detail": "No manual step for adopting observation keys; pre-existing operation conflicts still require the separate recovery path."}, + "recovery": {"status": "none", "detail": "Older builds can read the unchanged claim fields. Rolling back a producer restores its old global-key collision risk; rolling back graph listing and idle readers also restores their per-observation listing and timestamp behavior."} +}