From 88c391fba442d822cc40428f0746d3bd22e46596 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:25:10 +0100 Subject: [PATCH 1/4] fix(store): bound resumable assistant authority attachment work --- crates/fractal-api-core/src/api.rs | 6 + .../store/assistant_grant_attachment_tests.rs | 413 ++++++++++++++++++ .../src/store/assistant_grant_authority.rs | 158 +++++-- .../store/assistant_grant_authority_bench.rs | 2 + .../store/assistant_grant_authority_tests.rs | 5 + .../assistant_grant_receipt_cost_tests.rs | 2 + docs/st3/external-assistants.md | 26 ++ 7 files changed, 571 insertions(+), 41 deletions(-) create mode 100644 crates/smallclaims/src/store/assistant_grant_attachment_tests.rs diff --git a/crates/fractal-api-core/src/api.rs b/crates/fractal-api-core/src/api.rs index 43a6e0d9304..1b9594980c3 100644 --- a/crates/fractal-api-core/src/api.rs +++ b/crates/fractal-api-core/src/api.rs @@ -9612,6 +9612,12 @@ fn assistant_grant_subject_maintenance_checks( SELECT pending.grant_subject, 'retracted authority effective-state recomputation' FROM assistant_grant_recompute work JOIN assistant_grant_pending pending ON pending.claim_id=work.claim_id + UNION + SELECT grant_subject, 'authenticated grant attachment in progress' + FROM assistant_grant_attachments + UNION + SELECT grant_subject, 'attachment conflict-witness cleanup' + FROM assistant_grant_attachment_witnesses ORDER BY 1,2", )?; let rows = statement.query_map([], |row| { diff --git a/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs b/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs new file mode 100644 index 00000000000..9decd0c6250 --- /dev/null +++ b/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs @@ -0,0 +1,413 @@ +// Included by assistant_grant_authority_tests; uses its real local/replication fixtures. + +#[derive(Debug, Default)] +struct AttachmentSliceWork { + candidates: usize, + completions: usize, + witness_cleanups: usize, + pair_inserts: usize, + pair_updates: usize, + pair_deletes: usize, +} + +fn attachment_install_counter(connection: &Connection) { + connection.execute_batch( + "CREATE TEMP TABLE attachment_slice_work ( + candidates INTEGER NOT NULL, completions INTEGER NOT NULL, + pair_inserts INTEGER NOT NULL, pair_updates INTEGER NOT NULL, + pair_deletes INTEGER NOT NULL, witness_cleanups INTEGER NOT NULL); + INSERT INTO attachment_slice_work VALUES(0,0,0,0,0,0); + CREATE TEMP TRIGGER attachment_candidate_work + AFTER UPDATE OF after_claim_id ON main.assistant_grant_attachments + WHEN NEW.after_claim_id<>OLD.after_claim_id BEGIN + SELECT CASE WHEN NEW.after_claim_id<=OLD.after_claim_id + THEN RAISE(ABORT,'attachment candidate seek moved backwards') END; + UPDATE attachment_slice_work SET candidates=candidates+1; + END; + CREATE TEMP TRIGGER attachment_completion_work + AFTER DELETE ON main.assistant_grant_attachments BEGIN + UPDATE attachment_slice_work SET completions=completions+1; + END; + CREATE TEMP TRIGGER attachment_witness_cleanup_work + AFTER DELETE ON main.assistant_grant_attachment_witnesses BEGIN + UPDATE attachment_slice_work SET witness_cleanups=witness_cleanups+1; + END; + CREATE TEMP TRIGGER attachment_pair_insert_work + AFTER INSERT ON main.assistant_grant_authority + WHEN NEW.claim_id<>NEW.base_claim_id BEGIN + UPDATE attachment_slice_work SET pair_inserts=pair_inserts+1; + END; + CREATE TEMP TRIGGER attachment_pair_update_work + AFTER UPDATE ON main.assistant_grant_authority + WHEN NEW.claim_id<>NEW.base_claim_id BEGIN + UPDATE attachment_slice_work SET pair_updates=pair_updates+1; + END; + CREATE TEMP TRIGGER attachment_pair_delete_work + AFTER DELETE ON main.assistant_grant_authority + WHEN OLD.claim_id<>OLD.base_claim_id BEGIN + UPDATE attachment_slice_work SET pair_deletes=pair_deletes+1; + END;", + ).unwrap(); +} + +fn attachment_tasks(connection: &Connection) -> usize { + connection.query_row("SELECT COUNT(*) FROM assistant_grant_attachments", [], |row| row.get(0)).unwrap() +} + +fn attachment_connection_slice(connection: &mut Connection, budget: Duration) -> (bool, AttachmentSliceWork) { + // Enqueue INSERTs belong to authentication, not candidate work. Self-pairs + // and effective-row writes materialize the same candidate, so do not count + // them twice. Retiring a saved conflict witness is a separate shared unit. + connection.execute( + "UPDATE attachment_slice_work SET candidates=0,completions=0, + pair_inserts=0,pair_updates=0,pair_deletes=0,witness_cleanups=0", [], + ).unwrap(); + let transaction = connection.transaction().unwrap(); + let done = authority::backfill_slice(&transaction, budget).unwrap(); + let work = transaction.query_row( + "SELECT candidates,completions,pair_inserts,pair_updates,pair_deletes,witness_cleanups FROM attachment_slice_work", + [], |row| Ok(AttachmentSliceWork { + candidates: row.get(0)?, completions: row.get(1)?, + pair_inserts: row.get(2)?, pair_updates: row.get(3)?, pair_deletes: row.get(4)?, + witness_cleanups: row.get(5)?, + }), + ).unwrap(); + let limit = if budget.is_zero() { 1 } else { authority::CLAIMS_PER_SLICE }; + assert!(work.candidates+work.completions+work.witness_cleanups<=limit, + "attachment slice exceeded {limit} units: {work:?}"); + assert!(work.pair_inserts+work.pair_updates+work.pair_deletes<=limit, + "attachment pair mutation fanout exceeded {limit}: {work:?}"); + transaction.commit().unwrap(); + (done, work) +} + +fn attachment_slice(store: &Store, budget: Duration) -> (bool, AttachmentSliceWork) { + attachment_connection_slice(&mut store.connection.write(), budget) +} + +fn attachment_assert_fenced(store: &Store, grants: &[ClaimRecord], old_witnesses: usize) { + let connection = store.readers.get(); + let tasks = attachment_tasks(&connection); + let witnesses: usize = connection.query_row( + "SELECT COUNT(*) FROM assistant_grant_attachment_witnesses", [], |row| row.get(0), + ).unwrap(); + if tasks+witnesses==0 { return; } + assert!(authority::subject_ready(&connection, SUBJECT).is_err()); + for grant in grants { + assert!(!authority::contains(&connection, &grant.id).unwrap(), + "queued attachments exposed {}", grant.id); + } + let retained: usize = connection.query_row( + "SELECT COUNT(*) FROM assistant_grant_effective WHERE grant_subject=?1", [SUBJECT], |row| row.get(0), + ).unwrap(); + assert!(retained>=old_witnesses, "attachment fencing discarded an established conflict witness"); + if tasks>0 && old_witnesses>0 { + let candidates = authority::candidates( + &connection, "assistant/chatgpt", OWNER, Some(9999), None, None, + ).unwrap(); + for grant in grants { + assert!(candidates.iter().any(|candidate| candidate.id==grant.id), + "partial policy/revoke attachment lost baseline conflict witness {}", grant.id); + } + } +} + +fn attachment_append_group(store: &Store, bases: bool, count: usize) -> Vec { + (0..count).map(|index| { + let mut value = fields(); + value["expires_at_unix_ms"] = json!(10000+index); + let kind = if bases { GRANTED } else if index%2==0 { POLICY } else { REVOKED }; + append(store, SUBJECT, kind, value) + }).collect() +} + +fn attachment_budget_product(remote: bool, bases_first: bool, budget: Duration) { + const BASES: usize = 9; + const TRANSITIONS: usize = 19; + let (issuer, key) = source(); + let target = node("attachment-reader", Some(key.public())); + let store = if remote { &target } else { &issuer }; + let first = attachment_append_group(&issuer, bases_first, if bases_first { BASES } else { TRANSITIONS }); + drain_authority(&issuer); + if remote { receive(&target, &export(&issuer, &target)); } + let old_witnesses = if bases_first { BASES } else { 0 }; + attachment_install_counter(&store.connection.write()); + let second = attachment_append_group(&issuer, !bases_first, if bases_first { TRANSITIONS } else { BASES }); + if remote { + let exchange = export(&issuer, &target); + // Admit raw claims before their proofs. No signed maintenance work is + // hidden inside receive(), which intentionally drains its caller's work. + receive_pre_projection_unsigned_history(&target, &exchange); + let connection = target.connection.write(); + for signature in signatures(&exchange).signatures { + insert_raw_signature(&connection, &signature); + } + } + let grants = if bases_first { &first } else { &second }; + if !remote { + assert!(attachment_tasks(&store.readers.get())>0, "fanout was performed synchronously"); + attachment_assert_fenced(store, grants, old_witnesses); + } + let mut slices = 0; + let mut pairs = 0; + let mut candidates = 0; + let mut completions = 0; + // Includes authentication/cursor bookkeeping as well as every possible pair + // and partition completion; ZERO must still terminate without wall-clock aid. + let finite_bound = 32*(BASES+TRANSITIONS+BASES*TRANSITIONS+1); + loop { + attachment_assert_fenced(store, grants, old_witnesses); + let (done, work) = attachment_slice(store, budget); + slices += 1; + pairs += work.pair_inserts; + candidates += work.candidates; + completions += work.completions; + assert!(slices<=finite_bound, "one-unit attachment maintenance failed to finish"); + if done { break; } + } + assert_eq!(pairs, BASES*TRANSITIONS, "every compatible base/transition pair must attach exactly once"); + assert!(candidates>=BASES*TRANSITIONS, "pair writes escaped candidate accounting"); + assert!(completions>0, "partition completion must consume a unit"); + let per_slice = if budget.is_zero() { 1 } else { authority::CLAIMS_PER_SLICE }; + assert!(slices>=(BASES*TRANSITIONS).div_ceil(per_slice), "Cartesian attachment work escaped the slice budget"); + let connection = store.readers.get(); + assert_eq!(attachment_tasks(&connection), 0); + authority::subject_ready(&connection, SUBJECT).unwrap(); + let actual: usize = connection.query_row( + "SELECT COUNT(*) FROM assistant_grant_authority WHERE claim_id<>base_claim_id", [], |row| row.get(0), + ).unwrap(); + assert_eq!(actual, BASES*TRANSITIONS); + drop(connection); + assert_reference_and_rebuild(store); +} + +#[test] +fn local_attachment_cartesian_product_shares_the_full_slice_budget_in_both_directions() { + for bases_first in [true, false] { + attachment_budget_product(false, bases_first, Duration::from_secs(60)); + } +} + +#[test] +fn verified_attachment_cartesian_product_shares_the_full_slice_budget_in_both_directions() { + for bases_first in [true, false] { + attachment_budget_product(true, bases_first, Duration::from_secs(60)); + } +} + +#[test] +fn local_attachment_zero_duration_is_one_unit_and_finishes_in_both_directions() { + for bases_first in [true, false] { + attachment_budget_product(false, bases_first, Duration::ZERO); + } +} + +#[test] +fn verified_attachment_zero_duration_is_one_unit_and_finishes_in_both_directions() { + for bases_first in [true, false] { + attachment_budget_product(true, bases_first, Duration::ZERO); + } +} + +#[test] +fn attachment_cursor_and_conflict_witness_survive_file_reopen_and_zero_duration_completion() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("attachment-resume.sqlite3"); + let grants; + let saved_cursor; + let expected; + { + let store = Store::open(&path, "attachment-issuer", Arc::new(Plain)).unwrap(); + grants = attachment_append_group(&store, true, 11); + drain_authority(&store); + attachment_install_counter(&store.connection.write()); + attachment_append_group(&store, false, 13); + expected = reference(&store); + // Stop after actual pair work, not merely an authentication checkpoint. + for attempt in 0..100 { + let (_, work) = attachment_slice(&store, Duration::ZERO); + if work.candidates>0 { break; } + assert!(attempt<99, "attachment cursor never advanced"); + } + attachment_assert_fenced(&store, &grants, grants.len()); + let connection = store.readers.get(); + saved_cursor = connection.query_row( + "SELECT claim_id,candidate_kind,after_claim_id,upper_claim_id + FROM assistant_grant_attachments WHERE after_claim_id<>'' ORDER BY claim_id,candidate_kind LIMIT 1", + [], |row| Ok((row.get::<_,String>(0)?,row.get::<_,String>(1)?, + row.get::<_,String>(2)?,row.get::<_,String>(3)?)), + ).unwrap(); + assert!(attachment_tasks(&connection)>0); + } + // Open the durable database directly: Store startup is allowed to advance + // maintenance and would otherwise conceal whether the saved seek survived. + { + let mut connection = Connection::open(&path).unwrap(); + let durable: (String,String) = connection.query_row( + "SELECT after_claim_id,upper_claim_id FROM assistant_grant_attachments + WHERE claim_id=?1 AND candidate_kind=?2", params![saved_cursor.0,saved_cursor.1], + |row| Ok((row.get(0)?,row.get(1)?)), + ).unwrap(); + assert_eq!(durable, (saved_cursor.2.clone(),saved_cursor.3.clone())); + assert!(authority::subject_ready(&connection, SUBJECT).is_err()); + for grant in &grants { assert!(!authority::contains(&connection, &grant.id).unwrap()); } + attachment_install_counter(&connection); + let mut complete = false; + for _ in 0..2048 { + let (done, _) = attachment_connection_slice(&mut connection, Duration::ZERO); + if done { complete=true; break; } + } + assert!(complete, "durable one-unit attachment continuation did not terminate"); + assert_eq!(attachment_tasks(&connection), 0); + let witnesses: usize = connection.query_row( + "SELECT COUNT(*) FROM assistant_grant_attachment_witnesses", [], |row| row.get(0), + ).unwrap(); + assert_eq!(witnesses, 0); + authority::subject_ready(&connection, SUBJECT).unwrap(); + let actual: Vec<(String,String,bool)> = connection.prepare( + "SELECT claim_id,expires_at,revoked FROM assistant_grant_effective ORDER BY claim_id", + ).unwrap().query_map([], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?))) + .unwrap().collect::>().unwrap(); + assert_eq!(actual, expected, "resumed attachment state disagrees with raw history"); + } + let reopened = Store::open(&path, "attachment-issuer", Arc::new(Plain)).unwrap(); + assert_reference_and_rebuild(&reopened); + authority::subject_ready(&reopened.readers.get(), SUBJECT).unwrap(); + let resumed = snapshot(&reopened.readers.get()); + reset_projection(&reopened.connection.write()); + drain_authority(&reopened); + assert_eq!(snapshot(&reopened.readers.get()), resumed, + "restart continuation differs from a fresh authority backfill"); +} + +#[test] +fn invalid_attachment_candidates_each_consume_a_zero_duration_unit() { + let store = node("attachment-invalid", None); + for index in 0..23 { + let mut incompatible = fields(); + incompatible["profile"] = json!("service"); + incompatible["expires_at_unix_ms"] = json!(10000 + index); + append(&store, SUBJECT, GRANTED, incompatible); + } + drain_authority(&store); + let baseline = snapshot(&store.readers.get()); + attachment_install_counter(&store.connection.write()); + append(&store, SUBJECT, POLICY, fields()); + let mut candidates = 0; + let mut completions = 0; + let mut done = false; + for _ in 0..128 { + let (complete, work) = attachment_slice(&store, Duration::ZERO); + assert_eq!(work.pair_inserts, 0, "incompatible profile attached"); + candidates += work.candidates; + completions += work.completions; + if complete { done=true; break; } + } + assert!(done); + assert_eq!(candidates, 23, "invalid candidates escaped shared work accounting"); + assert_eq!(completions, 1, "empty partition completion must be its own unit"); + authority::subject_ready(&store.readers.get(), SUBJECT).unwrap(); + assert_eq!(snapshot(&store.readers.get()), baseline, + "an incompatible policy changed a base's effective projection"); + let mut connection = store.connection.write(); + let transaction = connection.transaction().unwrap(); + authority::rebuild_subject(&transaction, SUBJECT).unwrap(); + assert_eq!(snapshot(&transaction), baseline); + transaction.commit().unwrap(); +} + +#[test] +fn late_opposite_claim_attaches_after_an_older_task_captured_its_high_water() { + let store = node("attachment-late", None); + attachment_append_group(&store, true, 7); + drain_authority(&store); + let policy = append(&store, SUBJECT, POLICY, fields()); + let high_water: String = store.readers.get().query_row( + "SELECT upper_claim_id FROM assistant_grant_attachments WHERE claim_id=?1 AND candidate_kind=?2", + params![policy.id,GRANTED], |row| row.get(0), + ).unwrap(); + let mut late_fields = fields(); + late_fields["expires_at_unix_ms"] = json!(30000); + let late = append(&store, SUBJECT, GRANTED, late_fields); + let unchanged: String = store.readers.get().query_row( + "SELECT upper_claim_id FROM assistant_grant_attachments WHERE claim_id=?1 AND candidate_kind=?2", + params![policy.id,GRANTED], |row| row.get(0), + ).unwrap(); + assert_eq!(unchanged, high_water, "late arrival extended an existing fixed-high-water task"); + attachment_install_counter(&store.connection.write()); + let mut done = false; + for _ in 0..128 { + if attachment_slice(&store, Duration::ZERO).0 { done=true; break; } + } + assert!(done); + let attached: bool = store.readers.get().query_row( + "SELECT EXISTS(SELECT 1 FROM assistant_grant_authority WHERE claim_id=?1 AND base_claim_id=?2)", + params![policy.id,late.id], |row| row.get(0), + ).unwrap(); + assert!(attached, "late base was missed by both sides' attachment tasks"); + assert_reference_and_rebuild(&store); +} + +#[test] +fn trimming_an_attachment_owner_and_opposite_mid_task_does_not_restore_stale_pairs() { + let store = node("attachment-trim", None); + let grants = attachment_append_group(&store, true, 9); + drain_authority(&store); + attachment_install_counter(&store.connection.write()); + let transitions = attachment_append_group(&store, false, 5); + for _ in 0..100 { + if attachment_slice(&store, Duration::ZERO).1.candidates>0 { break; } + } + // Raw trim is a real demotion path: retire an owning transition and an + // opposite base while other tasks still hold their fixed candidate bounds. + { + let mut connection = store.connection.write(); + let transaction = connection.transaction().unwrap(); + for claim in [&transitions[0],&grants[0]] { + transaction.execute("DELETE FROM claims WHERE id=?1", [&claim.id]).unwrap(); + authority::refresh_claim(&transaction, &claim.id).unwrap(); + let own_tasks: usize = transaction.query_row( + "SELECT COUNT(*) FROM assistant_grant_attachments WHERE claim_id=?1", [&claim.id], |row| row.get(0), + ).unwrap(); + assert_eq!(own_tasks, 0, "demotion did not cancel the claim's own attachment tasks"); + } + transaction.commit().unwrap(); + } + let mut done = false; + for _ in 0..1024 { + if attachment_slice(&store, Duration::ZERO).0 { done=true; break; } + } + assert!(done, "stale endpoint work did not terminate"); + let stale: usize = store.readers.get().query_row( + "SELECT COUNT(*) FROM assistant_grant_authority + WHERE claim_id=?1 OR base_claim_id=?2", params![transitions[0].id,grants[0].id], |row| row.get(0), + ).unwrap(); + assert_eq!(stale, 0, "attachment execution resurrected a demoted endpoint"); + authority::subject_ready(&store.readers.get(), SUBJECT).unwrap(); + assert_reference_and_rebuild(&store); +} + +#[test] +fn first_local_base_without_opposite_history_is_immediately_usable_without_attachment_work() { + let store = node("attachment-immediate", None); + drain_authority(&store); + attachment_install_counter(&store.connection.write()); + let grant = append(&store, SUBJECT, GRANTED, fields()); + let connection = store.readers.get(); + assert_eq!(attachment_tasks(&connection), 0); + authority::subject_ready(&connection, SUBJECT).unwrap(); + assert!(authority::contains(&connection, &grant.id).unwrap()); + let witnesses: usize = connection.query_row( + "SELECT COUNT(*) FROM assistant_grant_attachment_witnesses", [], |row| row.get(0), + ).unwrap(); + assert_eq!(witnesses, 0); + drop(connection); + let connection = store.connection.write(); + let charged: usize = connection.query_row( + "SELECT candidates+completions+pair_inserts+pair_updates+pair_deletes+witness_cleanups + FROM attachment_slice_work", [], |row| row.get(0), + ).unwrap(); + assert_eq!(charged, 0, "constant self-materialization was mistaken for pair attachment"); + drop(connection); + assert_reference_and_rebuild(&store); +} diff --git a/crates/smallclaims/src/store/assistant_grant_authority.rs b/crates/smallclaims/src/store/assistant_grant_authority.rs index 4c1df9f6e0c..26a3d05b7a6 100644 --- a/crates/smallclaims/src/store/assistant_grant_authority.rs +++ b/crates/smallclaims/src/store/assistant_grant_authority.rs @@ -153,6 +153,23 @@ pub fn install(connection: &Connection, local_origin: &str) -> Result<()> { CREATE INDEX IF NOT EXISTS assistant_grant_retractions_writer_key_index ON assistant_grant_retractions(writer,member_key,lower_sequence,upper_sequence); CREATE TABLE IF NOT EXISTS assistant_grant_recompute (claim_id TEXT PRIMARY KEY); + CREATE TABLE IF NOT EXISTS assistant_grant_attachments ( + claim_id TEXT NOT NULL, candidate_kind TEXT NOT NULL, + grant_subject TEXT NOT NULL, origin TEXT NOT NULL, actor TEXT NOT NULL, + after_claim_id TEXT NOT NULL, upper_claim_id TEXT NOT NULL, + PRIMARY KEY(claim_id,candidate_kind)); + CREATE INDEX IF NOT EXISTS assistant_grant_attachments_subject_index + ON assistant_grant_attachments(grant_subject); + CREATE INDEX IF NOT EXISTS assistant_grant_pending_attachment_index + ON assistant_grant_pending(grant_subject,authenticated,origin,actor,kind,claim_id); + CREATE TABLE IF NOT EXISTS assistant_grant_attachment_witnesses ( + base_claim_id TEXT PRIMARY KEY, grant_subject TEXT NOT NULL, + principal TEXT NOT NULL, owner TEXT NOT NULL, profile TEXT NOT NULL, + expires_at TEXT NOT NULL, revoked INTEGER NOT NULL, grant_record TEXT NOT NULL); + CREATE INDEX IF NOT EXISTS assistant_grant_attachment_witnesses_subject_index + ON assistant_grant_attachment_witnesses(grant_subject); + CREATE INDEX IF NOT EXISTS assistant_grant_attachment_witnesses_active_index + ON assistant_grant_attachment_witnesses(principal,owner,revoked,expires_at,profile); CREATE INDEX IF NOT EXISTS assistant_grant_authority_base_index ON assistant_grant_authority(base_claim_id,kind,store_index DESC); CREATE TABLE IF NOT EXISTS assistant_grant_membership_state ( @@ -363,6 +380,8 @@ pub fn maintenance_pending(connection: &Connection) -> Result Result?4) AND (?5 IS NULL OR profile<>?5) + UNION + SELECT witness.grant_record FROM assistant_grant_attachment_witnesses AS witness + JOIN assistant_grant_pending AS pending ON pending.claim_id=witness.base_claim_id + WHERE witness.principal=?1 AND witness.owner=?2 + AND (?3 IS NULL OR (witness.revoked=0 AND witness.expires_at>?3)) + AND (?4 IS NULL OR witness.base_claim_id<>?4) AND (?5 IS NULL OR witness.profile<>?5) + AND pending.authenticated=1 + AND EXISTS(SELECT 1 FROM assistant_grant_attachments WHERE grant_subject=witness.grant_subject)"; /// Authenticated effective snapshots for conflict admission, not bearer access. /// A pending revoke must not erase an opposite-profile conflict witness before /// it authenticates. Bearer and transition readers independently fence subjects. @@ -532,49 +561,86 @@ fn recompute_base(connection: &Connection, base_id: &str) -> Result<()> { transition_unchecked(connection,&grant,REVOKED)?.is_some(),record])?; Ok(()) } +// Authentication creates constant-sized state; opposite-side history is queued. fn attach_claim(connection: &Connection, claim: &ClaimRecord) -> Result<()> { - let mut changed = false; - if claim.kind == GRANTED { - // A malformed base stays pending until a valid issuer policy repairs it. - // Otherwise a newer malformed copy could mask a valid bearer lookup. - if structurally_valid(claim,claim) { - changed |= insert_authority(connection,claim,claim)?; - } - // Only this base's issuer prefix can attach to it. In particular, an - // authenticated wrong-actor history cannot enlarge later admission work. - let mut statement = connection.prepare_cached( - "SELECT record FROM assistant_grant_pending WHERE grant_subject=?1 - AND authenticated=1 AND origin=?2 AND actor=?3 AND kind<>?4")?; - let records = statement.query_map(params![claim.subject,claim.origin,claim.actor,GRANTED], |row| row.get::<_,String>(0))?; - for record in records { - let transition: ClaimRecord = serde_json::from_str(&record?)?; - if transition_valid(claim,&transition) { - if transition.kind == POLICY { - changed |= insert_authority(connection,claim,claim)?; - } - changed |= insert_authority(connection,&transition,claim)?; - } - } - if changed { recompute_base(connection,&claim.id)?; } - } else { - let mut statement = connection.prepare_cached( - "SELECT record FROM assistant_grant_pending WHERE grant_subject=?1 - AND authenticated=1 AND kind=?2 AND origin=?3 AND actor=?4")?; - let records = statement.query_map(params![claim.subject,GRANTED,claim.origin,claim.actor], |row| row.get::<_,String>(0))?; - for record in records { - let grant: ClaimRecord = serde_json::from_str(&record?)?; - let mut changed = false; - if transition_valid(&grant,claim) { - if claim.kind == POLICY { - changed |= insert_authority(connection,&grant,&grant)?; - } - changed |= insert_authority(connection,claim,&grant)?; - if changed { recompute_base(connection,&grant.id)?; } - } - } + if claim.kind==GRANTED && structurally_valid(claim,claim) { + if insert_authority(connection,claim,claim)? { recompute_base(connection,&claim.id)?; } + } + let kinds: &[&str] = if claim.kind==GRANTED { &[POLICY,REVOKED] } else { &[GRANTED] }; + for kind in kinds { + connection.execute( + "INSERT OR IGNORE INTO assistant_grant_attachments + (claim_id,candidate_kind,grant_subject,origin,actor,after_claim_id,upper_claim_id) + SELECT ?1,?2,?3,?4,?5,'',claim_id FROM assistant_grant_pending + INDEXED BY assistant_grant_pending_attachment_index + WHERE grant_subject=?3 AND authenticated=1 AND origin=?4 AND actor=?5 AND kind=?2 + ORDER BY claim_id DESC LIMIT 1", + params![claim.id,kind,claim.subject,claim.origin,claim.actor], + )?; } Ok(()) } + +fn attachment_one(connection: &Connection) -> Result { + let task: Option<(String,String,String,String,String,String,String)> = connection.query_row( + "SELECT claim_id,candidate_kind,grant_subject,origin,actor,after_claim_id,upper_claim_id + FROM assistant_grant_attachments ORDER BY claim_id,candidate_kind LIMIT 1", [], + |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?,row.get(5)?,row.get(6)?)), + ).optional()?; + let Some((id,kind,subject,origin,actor,after,upper)) = task else { return Ok(false); }; + let record: Option = connection.query_row( + "SELECT record FROM assistant_grant_pending WHERE claim_id=?1 AND authenticated=1", + [&id], |row| row.get(0), + ).optional()?; + let candidate: Option<(String,String)> = if record.is_some() { + connection.query_row( + "SELECT claim_id,record FROM assistant_grant_pending + INDEXED BY assistant_grant_pending_attachment_index + WHERE grant_subject=?1 AND authenticated=1 AND origin=?2 AND actor=?3 AND kind=?4 + AND claim_id>?5 AND claim_id<=?6 ORDER BY claim_id LIMIT 1", + params![subject,origin,actor,kind,after,upper], + |row| Ok((row.get(0)?,row.get(1)?)), + ).optional()? + } else { None }; + let Some((candidate_id,candidate)) = candidate else { + connection.execute( + "DELETE FROM assistant_grant_attachments WHERE claim_id=?1 AND candidate_kind=?2", params![id,kind], + )?; + return Ok(true); + }; + let claim: ClaimRecord = serde_json::from_str(&record.expect("authenticated task owner"))?; + let candidate: ClaimRecord = serde_json::from_str(&candidate)?; + let (base,transition) = if claim.kind==GRANTED { (&claim,&candidate) } else { (&candidate,&claim) }; + // Both endpoint rows were re-read as authenticated in this transaction. + // Demotion precedes attachment and cancels the owner's tasks. + if transition_valid(base,transition) { + connection.execute( + "INSERT OR IGNORE INTO assistant_grant_attachment_witnesses + SELECT claim_id,grant_subject,principal,owner,profile,expires_at,revoked,grant_record + FROM assistant_grant_effective WHERE claim_id=?1", [&base.id], + )?; + let mut changed = false; + if transition.kind==POLICY { changed |= insert_authority(connection,base,base)?; } + changed |= insert_authority(connection,transition,base)?; + if changed { recompute_base(connection,&base.id)?; } + } + // Invalid candidates advance and consume exactly the same unit. + connection.execute( + "UPDATE assistant_grant_attachments SET after_claim_id=?3 WHERE claim_id=?1 AND candidate_kind=?2", + params![id,kind,candidate_id], + )?; + Ok(true) +} + +fn attachment_witness_one(connection: &Connection) -> Result { + let base: Option = connection.query_row( + "SELECT base_claim_id FROM assistant_grant_attachment_witnesses ORDER BY base_claim_id LIMIT 1", + [], |row| row.get(0), + ).optional()?; + let Some(base) = base else { return Ok(false); }; + connection.execute("DELETE FROM assistant_grant_attachment_witnesses WHERE base_claim_id=?1", [base])?; + Ok(true) +} fn local_origin(connection: &Connection) -> Result { Ok(connection.query_row("SELECT value FROM meta WHERE key='assistant_grant_authority_local_origin'", [], |row| row.get(0))?) } @@ -602,6 +668,7 @@ fn refresh_claim_with_authentication(connection: &Connection, id: &str, "UPDATE assistant_grant_pending SET authenticated=0,auth_member_key=NULL, auth_sequence=NULL,auth_envelope_hash=NULL,auth_record_position=NULL WHERE claim_id=?1", [id], )?; + connection.execute("DELETE FROM assistant_grant_attachments WHERE claim_id=?1", [id])?; connection.execute( "INSERT OR IGNORE INTO assistant_grant_recompute SELECT claim_id FROM assistant_grant_pending WHERE claim_id=?1", [id], @@ -676,6 +743,8 @@ fn refresh_claim_with_authentication(connection: &Connection, id: &str, // A still-pending unauthenticated transition continues to fence bearer use. fn rebuild_authenticated_subject(connection: &Connection, claim: &ClaimRecord) -> Result<()> { // Fanout is durable work, not a subject-history replay in this claim's unit. + // At most two kind partitions belong to an owner; cancellation is constant. + connection.execute("DELETE FROM assistant_grant_attachments WHERE claim_id=?1", [&claim.id])?; connection.execute( "INSERT OR IGNORE INTO assistant_grant_recompute(claim_id) VALUES(?1)", [&claim.id])?; @@ -1222,6 +1291,11 @@ fn backfill_slice_with_limit(connection: &Connection, budget: Duration, mut rema if remaining==0 || started.elapsed()>=budget { return Ok(false); } continue; } + if membership_current && (attachment_one(connection)? || attachment_witness_one(connection)?) { + remaining -= 1; + if remaining==0 || started.elapsed()>=budget { return Ok(false); } + continue; + } break; } // A completed version only drains the old-build dirty queue. Readiness also @@ -1253,6 +1327,8 @@ fn backfill_slice_with_limit(connection: &Connection, budget: Duration, mut rema OR EXISTS(SELECT 1 FROM assistant_grant_retractions) OR EXISTS(SELECT 1 FROM assistant_grant_additions) OR EXISTS(SELECT 1 FROM assistant_grant_recompute) + OR EXISTS(SELECT 1 FROM assistant_grant_attachments) + OR EXISTS(SELECT 1 FROM assistant_grant_attachment_witnesses) OR EXISTS(SELECT 1 FROM assistant_grant_proof_invalidations) OR EXISTS(SELECT 1 FROM assistant_grant_proof_rechecks)", [], |row| row.get(0))?; if dirty { return Ok(false); } diff --git a/crates/smallclaims/src/store/assistant_grant_authority_bench.rs b/crates/smallclaims/src/store/assistant_grant_authority_bench.rs index 88656f8c1c9..f09e0184eca 100644 --- a/crates/smallclaims/src/store/assistant_grant_authority_bench.rs +++ b/crates/smallclaims/src/store/assistant_grant_authority_bench.rs @@ -382,6 +382,8 @@ fn measure_backfill(config: Config) { DELETE FROM assistant_grant_authority; DELETE FROM assistant_grant_effective; DELETE FROM assistant_grant_pending; + DELETE FROM assistant_grant_attachments; + DELETE FROM assistant_grant_attachment_witnesses; DELETE FROM assistant_grant_dirty; DELETE FROM assistant_grant_auth_cursor; DELETE FROM assistant_grant_additions; diff --git a/crates/smallclaims/src/store/assistant_grant_authority_tests.rs b/crates/smallclaims/src/store/assistant_grant_authority_tests.rs index 716a856f15f..5560b0a794c 100644 --- a/crates/smallclaims/src/store/assistant_grant_authority_tests.rs +++ b/crates/smallclaims/src/store/assistant_grant_authority_tests.rs @@ -189,6 +189,7 @@ fn malformed_base_cannot_mask_a_valid_credential_but_valid_policy_can_repair_it( let mut renewal = fields(); renewal["expires_at_unix_ms"] = json!(20000); append(&store, SUBJECT, POLICY, renewal); + drain_authority(&store); assert!(authority::contains(&store.readers.get(), &pending.id).unwrap()); assert_eq!(authority::by_credential(&store.readers.get(), &digest).unwrap().unwrap().id, pending.id); assert_reference_and_rebuild(&store); @@ -461,6 +462,8 @@ fn reset_projection(connection: &Connection) { "DELETE FROM assistant_grant_effective; DELETE FROM assistant_grant_authority; DELETE FROM assistant_grant_pending; + DELETE FROM assistant_grant_attachments; + DELETE FROM assistant_grant_attachment_witnesses; DELETE FROM assistant_grant_membership_windows; DELETE FROM assistant_grant_dirty; DELETE FROM assistant_grant_auth_cursor; @@ -686,6 +689,7 @@ fn committed_backfill_slice_resumes_after_reopen_and_is_idempotent() { changed["expires_at_unix_ms"] = json!(20000); append(&store, SUBJECT, POLICY, changed); append(&store, SUBJECT, REVOKED, fields()); + drain_authority(&store); let mut connection = store.connection.write(); expected = snapshot(&connection); assert_eq!(expected.len(), 70); @@ -1416,6 +1420,7 @@ fn dirty_work_before_saved_round_robin_cursor_is_serviced_in_the_same_slice() { } include!("assistant_grant_receipt_cost_tests.rs"); +include!("assistant_grant_attachment_tests.rs"); #[path = "assistant_grant_membership_property_tests.rs"] mod membership_property_tests; diff --git a/crates/smallclaims/src/store/assistant_grant_receipt_cost_tests.rs b/crates/smallclaims/src/store/assistant_grant_receipt_cost_tests.rs index 809f0d79d44..187f1db48b7 100644 --- a/crates/smallclaims/src/store/assistant_grant_receipt_cost_tests.rs +++ b/crates/smallclaims/src/store/assistant_grant_receipt_cost_tests.rs @@ -46,6 +46,8 @@ fn receipt_projection_work_state(connection: &Connection) -> Vec Date: Tue, 10 Nov 2026 18:32:41 +0100 Subject: [PATCH 2/4] fix(store): fairly drain completed attachment witnesses --- .../store/assistant_grant_attachment_tests.rs | 60 +++++++++++++++++++ .../src/store/assistant_grant_authority.rs | 42 +++++++++++-- docs/st3/external-assistants.md | 5 ++ 3 files changed, 101 insertions(+), 6 deletions(-) diff --git a/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs b/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs index 9decd0c6250..548574a5205 100644 --- a/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs +++ b/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs @@ -411,3 +411,63 @@ fn first_local_base_without_opposite_history_is_immediately_usable_without_attac drop(connection); assert_reference_and_rebuild(&store); } + +#[test] +fn completed_subject_cleanup_lifts_fence_despite_unrelated_continuous_attachment_arrivals() { + let store = node("attachment-cleanup-fairness", None); + let grant = append(&store, SUBJECT, GRANTED, fields()); + drain_authority(&store); + attachment_install_counter(&store.connection.write()); + let mut policy = fields(); + policy["expires_at_unix_ms"] = json!(20000); + append(&store, SUBJECT, POLICY, policy); + for attempt in 0..32 { + if attachment_tasks(&store.readers.get())==0 { break; } + attachment_slice(&store, Duration::ZERO); + assert!(attempt<31, "subject A's attachment task did not finish"); + } + // Stop precisely after task completion, before the separately charged + // witness cleanup. A is still fenced and has one eligible witness. + assert!(authority::subject_ready(&store.readers.get(), SUBJECT).is_err()); + let witnesses: usize = store.readers.get().query_row( + "SELECT COUNT(*) FROM assistant_grant_attachment_witnesses WHERE grant_subject=?1", + [SUBJECT], |row| row.get(0), + ).unwrap(); + assert_eq!(witnesses, 1); + + let other = "custom/client/assistant-v1-fedcba9876543210fedcba98"; + let other_fields = || { + let mut value = fields(); + value["grant_id"] = json!("assistant-grant/fedcba9876543210fedcba98"); + value["session_actor"] = json!("assistant/chatgpt/session/assistant-grant/fedcba9876543210fedcba98"); + value + }; + for index in 0..40 { + let mut value = other_fields(); + value["expires_at_unix_ms"] = json!(30000+index); + append(&store, other, GRANTED, value); + } + for index in 0..40 { + let mut value = other_fields(); + value["expires_at_unix_ms"] = json!(40000+index); + append(&store, other, POLICY, value); + } + let mut lifted = false; + for index in 0..4 { + // Keep growing B's 40-by-40 backlog while A waits for cleanup. + let mut value = other_fields(); + value["expires_at_unix_ms"] = json!(50000+index); + append(&store, other, POLICY, value); + let (_, work) = attachment_slice(&store, Duration::ZERO); + assert!(work.candidates+work.completions+work.witness_cleanups<=1); + assert!(attachment_tasks(&store.readers.get())>0, + "regression requires B's unrelated backlog to remain"); + assert!(authority::subject_ready(&store.readers.get(), other).is_err()); + if authority::subject_ready(&store.readers.get(), SUBJECT).is_ok() { + assert!(authority::contains(&store.readers.get(), &grant.id).unwrap()); + lifted = true; + break; + } + } + assert!(lifted, "unrelated continuous arrivals starved completed-subject cleanup"); +} diff --git a/crates/smallclaims/src/store/assistant_grant_authority.rs b/crates/smallclaims/src/store/assistant_grant_authority.rs index 26a3d05b7a6..8adbf536be4 100644 --- a/crates/smallclaims/src/store/assistant_grant_authority.rs +++ b/crates/smallclaims/src/store/assistant_grant_authority.rs @@ -633,12 +633,28 @@ fn attachment_one(connection: &Connection) -> Result { } fn attachment_witness_one(connection: &Connection) -> Result { - let base: Option = connection.query_row( - "SELECT base_claim_id FROM assistant_grant_attachment_witnesses ORDER BY base_claim_id LIMIT 1", - [], |row| row.get(0), - ).optional()?; + let after = fleet_meta(connection,"assistant_grant_attachment_witness_cursor")?.unwrap_or_default(); + let seek = |after: &str| -> Result> { + Ok(connection.query_row( + "SELECT witness.base_claim_id FROM assistant_grant_attachment_witnesses witness + WHERE witness.base_claim_id>?1 AND NOT EXISTS( + SELECT 1 FROM assistant_grant_attachments tasks + WHERE tasks.grant_subject=witness.grant_subject) + ORDER BY witness.base_claim_id LIMIT 1", + [after], |row| row.get(0), + ).optional()?) + }; + let base = match seek(&after)? { + Some(base) => Some(base), + None if !after.is_empty() => seek("")?, + None => None, + }; let Some(base) = base else { return Ok(false); }; - connection.execute("DELETE FROM assistant_grant_attachment_witnesses WHERE base_claim_id=?1", [base])?; + connection.execute("DELETE FROM assistant_grant_attachment_witnesses WHERE base_claim_id=?1", [&base])?; + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_attachment_witness_cursor',?1)", + [&base], + )?; Ok(true) } fn local_origin(connection: &Connection) -> Result { @@ -1187,6 +1203,11 @@ fn backfill_slice_with_limit(connection: &Connection, budget: Duration, mut rema let dirty_limit = remaining.div_ceil(2); let mut dirty_visits = 0; let mut dirty_done = false; + // Completed subjects get a reserved cleanup share, independent of unrelated + // attachment backlogs. The persistent witness cursor survives one-unit slices. + let cleanup_limit = remaining.div_ceil(2); + let mut cleanup_visits = 0; + let mut cleanup_done = false; let mut dirty_after = fleet_meta(connection,"assistant_grant_dirty_cursor")?.unwrap_or_default(); // Include membership-driven retries in the same fixed item budget. Each // empty writer task also consumes a unit, bounding sparse writer queues. @@ -1198,6 +1219,15 @@ fn backfill_slice_with_limit(connection: &Connection, budget: Duration, mut rema if remaining==0 || started.elapsed()>=budget { return Ok(false); } continue; } + if membership_current && !cleanup_done && cleanup_visits=budget { return Ok(false); } + continue; + } + cleanup_done = true; + } if !dirty_done && dirty_visits = connection.query_row( "SELECT claim_id FROM assistant_grant_dirty WHERE claim_id>?1 ORDER BY claim_id LIMIT 1", @@ -1291,7 +1321,7 @@ fn backfill_slice_with_limit(connection: &Connection, budget: Duration, mut rema if remaining==0 || started.elapsed()>=budget { return Ok(false); } continue; } - if membership_current && (attachment_one(connection)? || attachment_witness_one(connection)?) { + if membership_current && attachment_one(connection)? { remaining -= 1; if remaining==0 || started.elapsed()>=budget { return Ok(false); } continue; diff --git a/docs/st3/external-assistants.md b/docs/st3/external-assistants.md index 4a5c6e54ba8..e3b4f0a8d9c 100644 --- a/docs/st3/external-assistants.md +++ b/docs/st3/external-assistants.md @@ -283,6 +283,11 @@ usable. Late opposite claims create their own tasks, including pairs beyond an older task's high-water mark. The subject remains fenced until its attachment tasks and witness cleanup finish. +Witness cleanup receives a reserved half-slice share after demotion, including +one-unit zero-duration slices. An indexed subject-task probe makes a witness +eligible as soon as its own subject has no attachment tasks; unrelated attachment +backlogs do not delay it. A persistent round-robin witness cursor prevents newly +eligible lower-ID arrivals from repeatedly overtaking a waiting subject. Before changing an effective base, maintenance retains its previous effective record as a conflict witness, so partially attached restrictive transitions cannot hide an established cross-subject conflict. Demotion cancels the demoted claim's From 4d26734bdb493f37454cb5b4332defda91a07688 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:39:21 +0100 Subject: [PATCH 3/4] fix(store): fairly budget every authority maintenance queue --- .../store/assistant_grant_attachment_tests.rs | 92 +++++- .../src/store/assistant_grant_authority.rs | 304 +++++++++--------- .../store/assistant_grant_authority_tests.rs | 117 +++++-- docs/st3/external-assistants.md | 25 +- 4 files changed, 343 insertions(+), 195 deletions(-) diff --git a/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs b/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs index 548574a5205..ebef25aa38f 100644 --- a/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs +++ b/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs @@ -453,7 +453,7 @@ fn completed_subject_cleanup_lifts_fence_despite_unrelated_continuous_attachment append(&store, other, POLICY, value); } let mut lifted = false; - for index in 0..4 { + for index in 0..8 { // Keep growing B's 40-by-40 backlog while A waits for cleanup. let mut value = other_fields(); value["expires_at_unix_ms"] = json!(50000+index); @@ -471,3 +471,93 @@ fn completed_subject_cleanup_lifts_fence_despite_unrelated_continuous_attachment } assert!(lifted, "unrelated continuous arrivals starved completed-subject cleanup"); } + +#[test] +fn ineligible_witness_prefix_consumes_bounded_visits_in_zero_and_full_slices() { + const PREFIX: usize = authority::CLAIMS_PER_SLICE * 3 + 1; + for budget in [Duration::ZERO, Duration::from_secs(60)] { + let store = node("attachment-witness-prefix", None); + let grant = append(&store, SUBJECT, GRANTED, fields()); + drain_authority(&store); + let mut connection = store.connection.write(); + let expected = snapshot(&connection); + // Clone a real witness payload only to exercise durable cleanup seeks. + // Synthetic IDs never enter authenticated pending/authority rows, and + // the unfinished subject's tasks intentionally have missing owners. + for index in 0..PREFIX { + connection.execute( + "INSERT INTO assistant_grant_attachment_witnesses + SELECT ?1,'prefix/unfinished',principal,owner,profile,expires_at,revoked,grant_record + FROM assistant_grant_effective WHERE claim_id=?2", + params![format!("prefix/{index:08}"),grant.id], + ).unwrap(); + } + connection.execute( + "INSERT INTO assistant_grant_attachment_witnesses + SELECT 'tail/eligible','tail/finished',principal,owner,profile,expires_at,revoked,grant_record + FROM assistant_grant_effective WHERE claim_id=?1", [&grant.id], + ).unwrap(); + // More tasks than all visits needed below: attachment turns cannot + // accidentally make the prefix eligible before the tail is reached. + for index in 0..(PREFIX * 4) { + connection.execute( + "INSERT INTO assistant_grant_attachments + VALUES(?1,?2,'prefix/unfinished','missing','missing','','')", + params![format!("prefix-task/{index:08}"),GRANTED], + ).unwrap(); + } + connection.execute_batch( + "DELETE FROM meta WHERE key='assistant_grant_attachment_witness_cursor'; + CREATE TEMP TABLE witness_visits (id INTEGER PRIMARY KEY, base_claim_id TEXT NOT NULL); + CREATE TEMP TRIGGER witness_visit_counter AFTER INSERT ON main.meta + WHEN NEW.key='assistant_grant_attachment_witness_cursor' BEGIN + INSERT INTO witness_visits(base_claim_id) VALUES(NEW.value); + END;", + ).unwrap(); + let limit = if budget.is_zero() { 1 } else { authority::CLAIMS_PER_SLICE }; + let mut visited = 0; + let mut finished = false; + for _ in 0..(8 * (PREFIX + 1)) { + connection.execute("DELETE FROM witness_visits", []).unwrap(); + let transaction = connection.transaction().unwrap(); + assert!(!authority::backfill_slice(&transaction, budget).unwrap()); + let visits: Vec = transaction.prepare( + "SELECT base_claim_id FROM witness_visits ORDER BY id", + ).unwrap().query_map([], |row| row.get(0)).unwrap() + .collect::>().unwrap(); + assert!(visits.len()<=limit, + "ineligible prefix escaped the {limit}-visit slice budget: {visits:?}"); + for base in visits { + if visited0, "ineligible-prefix fixture ran out of tasks"); + let tail_left: bool = transaction.query_row( + "SELECT EXISTS(SELECT 1 FROM assistant_grant_attachment_witnesses + WHERE base_claim_id='tail/eligible')", [], |row| row.get(0), + ).unwrap(); + assert_eq!(tail_left, visited<=PREFIX, + "eligible tail was reached without charging every preceding witness"); + assert_eq!(snapshot(&transaction), expected, "cleanup fixtures changed authority"); + transaction.commit().unwrap(); + if !tail_left { + finished = true; + break; + } + } + assert!(finished, "bounded cleanup never reached the eligible tail"); + } +} diff --git a/crates/smallclaims/src/store/assistant_grant_authority.rs b/crates/smallclaims/src/store/assistant_grant_authority.rs index 8adbf536be4..c322fea7f33 100644 --- a/crates/smallclaims/src/store/assistant_grant_authority.rs +++ b/crates/smallclaims/src/store/assistant_grant_authority.rs @@ -634,23 +634,26 @@ fn attachment_one(connection: &Connection) -> Result { fn attachment_witness_one(connection: &Connection) -> Result { let after = fleet_meta(connection,"assistant_grant_attachment_witness_cursor")?.unwrap_or_default(); - let seek = |after: &str| -> Result> { + let seek = |after: &str| -> Result> { Ok(connection.query_row( - "SELECT witness.base_claim_id FROM assistant_grant_attachment_witnesses witness - WHERE witness.base_claim_id>?1 AND NOT EXISTS( - SELECT 1 FROM assistant_grant_attachments tasks - WHERE tasks.grant_subject=witness.grant_subject) - ORDER BY witness.base_claim_id LIMIT 1", - [after], |row| row.get(0), + "SELECT base_claim_id,grant_subject FROM assistant_grant_attachment_witnesses + WHERE base_claim_id>?1 ORDER BY base_claim_id LIMIT 1", + [after], |row| Ok((row.get(0)?,row.get(1)?)), ).optional()?) }; - let base = match seek(&after)? { - Some(base) => Some(base), + let candidate = match seek(&after)? { + Some(candidate) => Some(candidate), None if !after.is_empty() => seek("")?, None => None, }; - let Some(base) = base else { return Ok(false); }; - connection.execute("DELETE FROM assistant_grant_attachment_witnesses WHERE base_claim_id=?1", [&base])?; + let Some((base,subject)) = candidate else { return Ok(false); }; + let pending: bool = connection.query_row( + "SELECT EXISTS(SELECT 1 FROM assistant_grant_attachments WHERE grant_subject=?1)", + [&subject], |row| row.get(0), + )?; + if !pending { + connection.execute("DELETE FROM assistant_grant_attachment_witnesses WHERE base_claim_id=?1", [&base])?; + } connection.execute( "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_attachment_witness_cursor',?1)", [&base], @@ -1194,163 +1197,146 @@ fn refresh_recheck(connection: &Connection, claim: &str) -> Result { Ok(!unfinished) } +fn dirty_one(connection: &Connection, membership_current: bool) -> Result { + let after = fleet_meta(connection,"assistant_grant_dirty_cursor")?.unwrap_or_default(); + let seek = |after: &str| -> Result> { + Ok(connection.query_row( + "SELECT claim_id FROM assistant_grant_dirty WHERE claim_id>?1 ORDER BY claim_id LIMIT 1", + [after], |row| row.get(0), + ).optional()?) + }; + let claim = match seek(&after)? { + Some(claim) => Some(claim), + None if !after.is_empty() => seek("")?, + None => None, + }; + let Some(claim) = claim else { return Ok(false); }; + let recheck_owned: bool = connection.query_row( + "SELECT EXISTS(SELECT 1 FROM assistant_grant_proof_rechecks WHERE claim_id=?1)", + [&claim], |row| row.get(0), + )?; + if recheck_owned { + if membership_current { refresh_recheck(connection,&claim)?; } + } else { + refresh_claim_inner(connection,&claim,membership_current)?; + } + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_dirty_cursor',?1)", [&claim], + )?; + Ok(true) +} + +fn recheck_one(connection: &Connection) -> Result { + let claim: Option = connection.query_row( + "SELECT claim_id FROM assistant_grant_proof_rechecks ORDER BY claim_id LIMIT 1", + [], |row| row.get(0), + ).optional()?; + let Some(claim) = claim else { return Ok(false); }; + refresh_recheck(connection,&claim)?; + Ok(true) +} + +fn retraction_one(connection: &Connection) -> Result { + let task: Option<(i64,String,String,u64,u64,i64,String)> = connection.query_row( + "SELECT id,writer,member_key,lower_sequence,upper_sequence,sequence,claim_id + FROM assistant_grant_retractions ORDER BY id LIMIT 1", [], + |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?,row.get(5)?,row.get(6)?)), + ).optional()?; + let Some((task,writer,key,lower,upper,sequence,id)) = task else { return Ok(false); }; + if trim_restored_retraction(connection,task,&writer,&key,lower,upper,sequence,&id)? { + return Ok(true); + } + let pending: Option<(u64,String)> = connection.query_row( + "SELECT auth_sequence,claim_id FROM assistant_grant_pending INDEXED BY assistant_grant_pending_evidence_index + WHERE origin=?1 AND auth_member_key=?2 AND auth_sequence BETWEEN ?3 AND ?4 + AND (auth_sequence,claim_id)>(?5,?6) + ORDER BY auth_sequence,claim_id LIMIT 1", + params![writer,key,lower,upper,sequence,id], |row| Ok((row.get(0)?,row.get(1)?)), + ).optional()?; + if let Some((sequence,id)) = pending { + connection.execute("DELETE FROM assistant_grant_auth_cursor WHERE claim_id=?1", [&id])?; + connection.execute("INSERT OR IGNORE INTO assistant_grant_proof_rechecks VALUES(?1)", [&id])?; + connection.execute( + "UPDATE assistant_grant_retractions SET sequence=?2,claim_id=?3 WHERE id=?1", + params![task,sequence,id], + )?; + } else { + connection.execute("DELETE FROM assistant_grant_retractions WHERE id=?1", [task])?; + } + Ok(true) +} + +fn history_one(connection: &Connection, membership_current: bool) -> Result { + // A completed version never reopens historical scanning. + if fleet_meta(connection,"assistant_grant_authority_version")?.as_deref()==Some(VERSION) { + return Ok(false); + } + for kind in KINDS { + let key = format!("assistant_grant_authority_cursor/{kind}"); + let cursor: u64 = connection.query_row( + "SELECT CAST(value AS INTEGER) FROM meta WHERE key=?1", + [&key], |row| row.get(0), + ).optional()?.unwrap_or(0); + let claim: Option<(String,u64)> = connection.query_row( + "SELECT id,store_index FROM claims INDEXED BY claims_kind_index + WHERE kind=?1 AND store_index>?2 ORDER BY store_index LIMIT 1", + params![kind,cursor], |row| Ok((row.get(0)?,row.get(1)?)), + ).optional()?; + if let Some((id,index)) = claim { + refresh_claim_inner(connection,&id,membership_current)?; + connection.execute( + /* sqlite-large-value: allow-for-now lv-h3f49c0e579cd owner=smallclaims::store::assistant_grant_authority TODO=bound-or-reference-values-and-reclamation */ + "INSERT OR REPLACE INTO meta(key,value) VALUES(?1,?2)", + params![key,index.to_string()], + )?; + return Ok(true); + } + } + Ok(false) +} + +const MAINTENANCE_QUEUES: usize = 8; + +fn maintenance_one(connection: &Connection, turn: usize, membership_current: bool) -> Result { + match turn { + 0 => dirty_one(connection,membership_current), + 1 => if membership_current { recheck_one(connection) } else { Ok(false) }, + 2 => proof_invalidation_one(connection), + 3 => retraction_one(connection), + 4 => if membership_current { addition_one(connection) } else { Ok(false) }, + 5 => if membership_current { attachment_one(connection) } else { Ok(false) }, + 6 => if membership_current { attachment_witness_one(connection) } else { Ok(false) }, + 7 => history_one(connection,membership_current), + _ => unreachable!("maintenance turn is reduced modulo queue count"), + } +} + fn backfill_slice_with_limit(connection: &Connection, budget: Duration, mut remaining: usize, membership_current: bool) -> Result { let started = Instant::now(); if remaining==0 { return Ok(false); } - // Give ordinary dirty authentication the first half of each transaction. - // Persist its round-robin position so a long key search cannot monopolize - // this share and an addition backlog cannot fence unrelated dirty subjects. - let dirty_limit = remaining.div_ceil(2); - let mut dirty_visits = 0; - let mut dirty_done = false; - // Completed subjects get a reserved cleanup share, independent of unrelated - // attachment backlogs. The persistent witness cursor survives one-unit slices. - let cleanup_limit = remaining.div_ceil(2); - let mut cleanup_visits = 0; - let mut cleanup_done = false; - let mut dirty_after = fleet_meta(connection,"assistant_grant_dirty_cursor")?.unwrap_or_default(); - // Include membership-driven retries in the same fixed item budget. Each - // empty writer task also consumes a unit, bounding sparse writer queues. + let mut turn = fleet_meta(connection,"assistant_grant_maintenance_turn")? + .and_then(|value| value.parse::().ok()).unwrap_or(0)%MAINTENANCE_QUEUES; while remaining>0 { - // Finish demotion before a later authentication can restore the row. - // Each authority-pair deletion/effective recompute costs one unit. - if recompute_one(connection)? { - remaining -= 1; - if remaining==0 || started.elapsed()>=budget { return Ok(false); } - continue; - } - if membership_current && !cleanup_done && cleanup_visits=budget { return Ok(false); } - continue; + // Demotion only shrinks authority. Finish its bounded units before any + // positive work; all other queues share one persistent round-robin turn. + let mut visited = recompute_one(connection)?; + if !visited { + for _ in 0..MAINTENANCE_QUEUES { + visited = maintenance_one(connection,turn,membership_current)?; + turn = (turn+1)%MAINTENANCE_QUEUES; + if visited { break; } } - cleanup_done = true; - } - if !dirty_done && dirty_visits = connection.query_row( - "SELECT claim_id FROM assistant_grant_dirty WHERE claim_id>?1 ORDER BY claim_id LIMIT 1", - [&dirty_after], |row| row.get(0), - ).optional()?; - if let Some(claim) = dirty { - let recheck_owned: bool = connection.query_row( - "SELECT EXISTS(SELECT 1 FROM assistant_grant_proof_rechecks WHERE claim_id=?1)", - [&claim], |row| row.get(0), - )?; - // Dispatch to the owning verifier here, rather than spending a - // skip unit forever when every slice has only one unit/time. - let unfinished = if recheck_owned { - if !membership_current { return Ok(false); } - !refresh_recheck(connection,&claim)? - } else { - refresh_claim_inner(connection,&claim,membership_current)?; - false - }; + if visited { connection.execute( - "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_dirty_cursor',?1)", [&claim], + "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_maintenance_turn',?1)", + [turn as i64], )?; - dirty_after = claim; - dirty_visits += 1; - remaining -= 1; - if unfinished || remaining==0 || started.elapsed()>=budget { return Ok(false); } - continue; - } - connection.execute("DELETE FROM meta WHERE key='assistant_grant_dirty_cursor'", [])?; - if dirty_visits==0 && !dirty_after.is_empty() { - // A late insert/evidence arrival may sort before the saved - // cursor. Wrap now, not in a later otherwise empty slice. - dirty_after.clear(); - continue; - } - dirty_done = true; - } - let recheck: Option = connection.query_row( - "SELECT claim_id FROM assistant_grant_proof_rechecks ORDER BY claim_id LIMIT 1", - [], |row| row.get(0), - ).optional()?; - if let Some(claim) = recheck { - if !membership_current { return Ok(false); } - // A single owning verifier advances and discharges this obligation, - // regardless of which dispatch queue selected the claim. - let complete = refresh_recheck(connection,&claim)?; - remaining -= 1; - if !complete { return Ok(false); } - if remaining==0 || started.elapsed()>=budget { return Ok(false); } - continue; - } - if proof_invalidation_one(connection)? { - remaining -= 1; - if remaining==0 || started.elapsed()>=budget { return Ok(false); } - continue; - } - let retraction: Option<(i64,String,String,u64,u64,i64,String)> = connection.query_row( - "SELECT id,writer,member_key,lower_sequence,upper_sequence,sequence,claim_id - FROM assistant_grant_retractions ORDER BY id LIMIT 1", [], - |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?,row.get(5)?,row.get(6)?)), - ).optional()?; - if let Some((task,writer,key,lower,upper,sequence,id)) = retraction { - if trim_restored_retraction(connection,task,&writer,&key,lower,upper,sequence,&id)? { - remaining -= 1; - if remaining==0 || started.elapsed()>=budget { return Ok(false); } - continue; - } - let pending: Option<(u64,String)> = connection.query_row( - "SELECT auth_sequence,claim_id FROM assistant_grant_pending INDEXED BY assistant_grant_pending_evidence_index - WHERE origin=?1 AND auth_member_key=?2 AND auth_sequence BETWEEN ?3 AND ?4 - AND (auth_sequence,claim_id)>(?5,?6) - ORDER BY auth_sequence,claim_id LIMIT 1", - params![writer,key,lower,upper,sequence,id], |row| Ok((row.get(0)?,row.get(1)?)), - ).optional()?; - if let Some((sequence,id)) = pending { - connection.execute("DELETE FROM assistant_grant_auth_cursor WHERE claim_id=?1", [&id])?; - connection.execute("INSERT OR IGNORE INTO assistant_grant_proof_rechecks VALUES(?1)", [&id])?; - connection.execute( - "UPDATE assistant_grant_retractions SET sequence=?2,claim_id=?3 WHERE id=?1", - params![task,sequence,id], - )?; - } else { - connection.execute("DELETE FROM assistant_grant_retractions WHERE id=?1", [task])?; - } - remaining -= 1; - if remaining==0 || started.elapsed()>=budget { return Ok(false); } - continue; - } - if membership_current && addition_one(connection)? { - remaining -= 1; - if remaining==0 || started.elapsed()>=budget { return Ok(false); } - continue; - } - if membership_current && attachment_one(connection)? { - remaining -= 1; - if remaining==0 || started.elapsed()>=budget { return Ok(false); } - continue; - } - break; - } - // A completed version only drains the old-build dirty queue. Readiness also - // includes that queue, but must never cause a completed history rescan. - if fleet_meta(connection,"assistant_grant_authority_version")?.as_deref() != Some(VERSION) { - for kind in KINDS { - let key = format!("assistant_grant_authority_cursor/{kind}"); - loop { - let cursor: u64 = connection.query_row("SELECT CAST(value AS INTEGER) FROM meta WHERE key=?1",[&key],|row|row.get(0)).optional()?.unwrap_or(0); - let rows = { - let mut statement = connection.prepare_cached("SELECT id,store_index FROM claims INDEXED BY claims_kind_index WHERE kind=?1 AND store_index>?2 ORDER BY store_index LIMIT 64")?; - statement.query_map(params![kind,cursor],|row|Ok((row.get::<_,String>(0)?,row.get::<_,u64>(1)?)))?.collect::>>()? - }; - if rows.is_empty() { break; } - for (id,index) in rows { - refresh_claim_inner(connection,&id,membership_current)?; - remaining -= 1; - connection.execute(/* sqlite-large-value: allow-for-now lv-h3f49c0e579cd owner=smallclaims::store::assistant_grant_authority TODO=bound-or-reference-values-and-reclamation */ "INSERT OR REPLACE INTO meta(key,value) VALUES(?1,?2)",params![key,index.to_string()])?; - if connection.query_row("SELECT EXISTS(SELECT 1 FROM assistant_grant_recompute)", [], |row| row.get::<_,bool>(0))? { - return Ok(false); - } - if remaining==0 || started.elapsed()>=budget { return Ok(false); } - } } } + if !visited { break; } + remaining -= 1; + // Persisted turn already points past this unit, even for ZERO slices. + if remaining==0 || started.elapsed()>=budget { return Ok(false); } } let dirty: bool = connection.query_row( "SELECT EXISTS(SELECT 1 FROM assistant_grant_dirty) diff --git a/crates/smallclaims/src/store/assistant_grant_authority_tests.rs b/crates/smallclaims/src/store/assistant_grant_authority_tests.rs index 5560b0a794c..afc7b0f22ff 100644 --- a/crates/smallclaims/src/store/assistant_grant_authority_tests.rs +++ b/crates/smallclaims/src/store/assistant_grant_authority_tests.rs @@ -716,7 +716,7 @@ fn committed_backfill_slice_resumes_after_reopen_and_is_idempotent() { transaction.commit().unwrap(); slices += 1; if complete { break; } - assert!(slices < 200, "backfill failed to make durable forward progress"); + assert!(slices < 8*(70*3+3), "backfill failed to make durable forward progress"); } assert!(slices > 1); assert_eq!(snapshot(&connection), expected); @@ -840,31 +840,75 @@ fn duplicate_verified_signature_does_not_refresh_or_dirty_claims() { } #[test] -fn dirty_claim_flood_reserves_half_the_slice_for_other_maintenance() { +fn every_nonempty_maintenance_queue_receives_a_unit_within_one_zero_slice_round() { + // History is the eighth turn, empty after a completed backfill. Exercise + // every starting turn, including that empty slot, with all seven durable + // maintenance queues continuously occupied. + const Q: usize = 8; let store = node("issuer", None); - for index in 0..(authority::CLAIMS_PER_SLICE * 3 + 7) { - let mut invalid = fields(); - invalid["issuer"] = json!(format!("invalid/{index}")); - append(&store,SUBJECT,GRANTED,invalid); - } + let grant = append(&store, SUBJECT, GRANTED, fields()); drain_authority(&store); let mut connection = store.connection.write(); let markers = backfill_markers(&connection); - connection.execute("INSERT INTO assistant_grant_dirty SELECT id FROM claims WHERE kind=?1", [GRANTED]).unwrap(); - let mut remaining = authority::CLAIMS_PER_SLICE * 3 + 7; - loop { - let transaction = connection.transaction().unwrap(); - let done = authority::backfill_slice(&transaction,Duration::from_secs(60)).unwrap(); - let after: usize = transaction.query_row("SELECT COUNT(*) FROM assistant_grant_dirty", [], |row| row.get(0)).unwrap(); - // Dirty authentication owns half the 64-unit slice; the other half - // remains available to proof withdrawals and membership additions. - assert_eq!(remaining-after, remaining.min(authority::CLAIMS_PER_SLICE.div_ceil(2))); - assert_eq!(backfill_markers(&transaction),markers,"completed history was rescanned"); - transaction.commit().unwrap(); - remaining = after; - if done { break; } + let expected = snapshot(&connection); + let queues = [ + "assistant_grant_dirty", "assistant_grant_proof_rechecks", + "assistant_grant_proof_invalidations", "assistant_grant_retractions", + "assistant_grant_additions", "assistant_grant_attachments", + "assistant_grant_attachment_witnesses", + ]; + for start in 0..Q { + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_maintenance_turn',?1)", + [start.to_string()], + ).unwrap(); + let mut visited = [false; 7]; + for _ in 0..Q { + // Missing claims/proof identities are real stale-work retirement + // cases. None of these tasks supplies authentication or inserts a + // forged authenticated pending row. + connection.execute_batch( + "INSERT OR IGNORE INTO assistant_grant_dirty VALUES('fairness/dirty'); + INSERT OR IGNORE INTO assistant_grant_proof_rechecks VALUES('fairness/recheck'); + INSERT OR IGNORE INTO assistant_grant_proof_invalidations + (id,writer,sequence,envelope_hash,member_key) + VALUES(1,'fairness/missing-writer',0,'missing','missing'); + INSERT OR IGNORE INTO assistant_grant_retractions + (id,writer,member_key,lower_sequence,upper_sequence) + VALUES(1,'fairness/missing-writer','missing',0,0); + INSERT OR IGNORE INTO assistant_grant_additions + (id,writer,member_key,lower_sequence,upper_sequence) + VALUES(1,'fairness/missing-writer','missing',0,0); + INSERT OR IGNORE INTO assistant_grant_attachments + VALUES('fairness/attachment','custom.client.assistant.v1.granted', + 'fairness/unfinished','missing','missing','','');", + ).unwrap(); + connection.execute( + "INSERT OR IGNORE INTO assistant_grant_attachment_witnesses + SELECT claim_id,grant_subject,principal,owner,profile,expires_at,revoked,grant_record + FROM assistant_grant_effective WHERE claim_id=?1", [&grant.id], + ).unwrap(); + let transaction = connection.transaction().unwrap(); + assert!(!authority::backfill_slice(&transaction, Duration::ZERO).unwrap()); + let mut units = 0; + for (index, queue) in queues.iter().enumerate() { + let left: usize = transaction.query_row( + &format!("SELECT COUNT(*) FROM {queue}"), [], |row| row.get(0), + ).unwrap(); + assert!(left<=1, "unexpected work in {queue}"); + if left==0 { + visited[index] = true; + units += 1; + } + } + assert_eq!(units, 1, "ZERO must perform exactly one occupied visit"); + assert_eq!(backfill_markers(&transaction), markers, "completed history was rescanned"); + assert_eq!(snapshot(&transaction), expected, "stale queue tasks changed authority"); + transaction.commit().unwrap(); + } + assert!(visited.iter().all(|seen| *seen), + "starting turn {start} starved a queue within {Q} ZERO slices: {visited:?}"); } - assert_eq!(remaining,0); } fn authentication_key_cost(connection: &Connection, signature: &ReplicaEnvelopeSignature) -> (usize,i32,i32) { @@ -947,8 +991,12 @@ fn eligible_evidence_beyond_first_key_page_resumes_and_cryptographically_verifie assert_eq!(cost.2,0); let mut previous_key = String::new(); for page in 0..4 { + // This test measures one authentication page, not a full fair round. + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_maintenance_turn','0')", [], + ).unwrap(); let transaction = connection.transaction().unwrap(); - let done = authority::backfill_slice(&transaction,Duration::from_secs(60)).unwrap(); + let done = authority::backfill_slice(&transaction,Duration::ZERO).unwrap(); assert_eq!(backfill_markers(&transaction),markers); if page<3 { assert!(!done,"invalid earlier evidence authenticated the claim"); @@ -964,11 +1012,14 @@ fn eligible_evidence_beyond_first_key_page_resumes_and_cryptographically_verifie "SELECT authenticated FROM assistant_grant_pending WHERE claim_id=?1", [&grant.id], |row| row.get(0)).unwrap(); assert!(!admitted); } else { - assert!(done); + assert!(!done, "the final occupied ZERO visit still consumes its slice"); assert!(authority::contains(&transaction,&grant.id).unwrap()); } transaction.commit().unwrap(); } + let transaction = connection.transaction().unwrap(); + assert!(authority::backfill_slice(&transaction, Duration::from_secs(60)).unwrap()); + transaction.commit().unwrap(); } #[test] @@ -1039,8 +1090,11 @@ fn expired_membership_windows_consume_bounded_pages_and_future_windows_stay_out_ "future-only windows entered the candidate seek"); let mut previous = (-1,String::new()); for page in 0..4 { + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_maintenance_turn','0')", [], + ).unwrap(); let transaction = connection.transaction().unwrap(); - let done = authority::backfill_slice(&transaction,Duration::from_secs(60)).unwrap(); + let done = authority::backfill_slice(&transaction,Duration::ZERO).unwrap(); if page<3 { assert!(!done,"expired candidates must consume their own page budget"); let cursor: (i64,String) = transaction.query_row( @@ -1050,11 +1104,14 @@ fn expired_membership_windows_consume_bounded_pages_and_future_windows_stay_out_ previous = cursor; assert!(!authority::contains(&transaction,&grant.id).unwrap()); } else { - assert!(done); + assert!(!done, "the final occupied ZERO visit still consumes its slice"); assert!(authority::contains(&transaction,&grant.id).unwrap()); } transaction.commit().unwrap(); } + let transaction = connection.transaction().unwrap(); + assert!(authority::backfill_slice(&transaction, Duration::from_secs(60)).unwrap()); + transaction.commit().unwrap(); // The signature remains cryptographically valid, but an expired window // must never authenticate it during a subsequent proof recheck. connection.execute( @@ -1131,8 +1188,11 @@ fn late_signature_restarts_an_in_progress_key_cursor_before_its_old_position() { params![signature.writer,format!("~{index:08}"),member_start]).unwrap(); } connection.execute("INSERT INTO assistant_grant_dirty VALUES(?1)", [&grant.id]).unwrap(); + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_maintenance_turn','0')", [], + ).unwrap(); let transaction = connection.transaction().unwrap(); - assert!(!authority::backfill_slice(&transaction,Duration::from_secs(60)).unwrap()); + assert!(!authority::backfill_slice(&transaction,Duration::ZERO).unwrap()); let cursor: String = transaction.query_row( "SELECT member_key FROM assistant_grant_auth_cursor WHERE claim_id=?1", [&grant.id], |row| row.get(0)).unwrap(); @@ -1180,8 +1240,11 @@ fn envelope_record_arriving_before_auth_cursor_restarts_preexisting_signature_se params![signature.writer,format!("~{index:08}")]).unwrap(); } connection.execute("INSERT OR IGNORE INTO assistant_grant_dirty VALUES(?1)", [&grant.id]).unwrap(); + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_maintenance_turn','0')", [], + ).unwrap(); let transaction = connection.transaction().unwrap(); - assert!(!authority::backfill_slice(&transaction,Duration::from_secs(60)).unwrap()); + assert!(!authority::backfill_slice(&transaction,Duration::ZERO).unwrap()); let position: i64 = transaction.query_row( "SELECT position FROM assistant_grant_auth_cursor WHERE claim_id=?1", [&grant.id], |row| row.get(0)).unwrap(); diff --git a/docs/st3/external-assistants.md b/docs/st3/external-assistants.md index e3b4f0a8d9c..ec1af2b0feb 100644 --- a/docs/st3/external-assistants.md +++ b/docs/st3/external-assistants.md @@ -253,9 +253,16 @@ work index, maintained atomically with unauthenticated rows and cursor/recheck obligations. Each task then seeks that claim's own receipts in the added range. One claim selection, receipt attempt or claim completion consumes a budget unit; unrelated history is never enumerated. Total work is proportional to needing -grant rows and their own matching receipts, not all writer receipts. Ordinary -dirty authentication receives the first half of each transaction with a durable -round-robin cursor, so an addition backlog cannot keep dirty subjects fenced. +grant rows and their own matching receipts, not all writer receipts. +After demotion, maintenance uses one durable round-robin queue turn, in order: +dirty authentication, proof rechecks, proof invalidation, membership retraction, +coverage addition, positive attachment, conflict-witness cleanup, initial history +backfill. Each nonempty turn performs one charged unit and advances the persisted +turn before a slice can end, including `Duration::ZERO`. Empty queues cost a +bounded indexed probe each; dispatch visits at most eight queues per unit. +Thus, absent demotion work, each runnable queue receives a unit within eight +one-unit slices even under continuous arrivals to other queues. There are no +per-queue slice reservations. Dirty authentication retains its own durable cursor. Late dirty work sorting before the saved cursor wraps within the same slice, without waiting for a later maintenance wakeup. Repeated cursor or recheck writes retain one work-set entry even under an outer SQL upsert's conflict policy. @@ -283,11 +290,13 @@ usable. Late opposite claims create their own tasks, including pairs beyond an older task's high-water mark. The subject remains fenced until its attachment tasks and witness cleanup finish. -Witness cleanup receives a reserved half-slice share after demotion, including -one-unit zero-duration slices. An indexed subject-task probe makes a witness -eligible as soon as its own subject has no attachment tasks; unrelated attachment -backlogs do not delay it. A persistent round-robin witness cursor prevents newly -eligible lower-ID arrivals from repeatedly overtaking a waiting subject. +Cleanup seeks exactly one witness by its persistent primary-key cursor before +probing that subject's attachment index. It advances and charges the visit whether +the witness is deleted or skipped because its own subject still has tasks. +Unrelated attachment backlogs do not prevent eligible cleanup; a long ineligible +prefix consumes separate bounded visits instead of an uncharged eligibility scan. +The witness cursor wraps and prevents lower-ID arrivals repeatedly overtaking +waiting witnesses. Before changing an effective base, maintenance retains its previous effective record as a conflict witness, so partially attached restrictive transitions cannot hide an established cross-subject conflict. Demotion cancels the demoted claim's From 22b947e4909b5efb1ce1802ce60ec57a61fd18bc Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:47:04 +0100 Subject: [PATCH 4/4] fix(store): order maintenance owners by durable admission sequence --- .../store/assistant_grant_attachment_tests.rs | 85 +++++++++- .../src/store/assistant_grant_authority.rs | 159 +++++++++++++----- .../store/assistant_grant_authority_tests.rs | 79 ++++++++- .../assistant_grant_receipt_cost_tests.rs | 3 + docs/st3/external-assistants.md | 29 ++-- 5 files changed, 295 insertions(+), 60 deletions(-) diff --git a/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs b/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs index ebef25aa38f..f0d97a290c2 100644 --- a/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs +++ b/crates/smallclaims/src/store/assistant_grant_attachment_tests.rs @@ -507,11 +507,13 @@ fn ineligible_witness_prefix_consumes_bounded_visits_in_zero_and_full_slices() { ).unwrap(); } connection.execute_batch( - "DELETE FROM meta WHERE key='assistant_grant_attachment_witness_cursor'; + "DELETE FROM meta WHERE key IN ('assistant_grant_owner_cursor/witness','assistant_grant_owner_upper/witness'); CREATE TEMP TABLE witness_visits (id INTEGER PRIMARY KEY, base_claim_id TEXT NOT NULL); CREATE TEMP TRIGGER witness_visit_counter AFTER INSERT ON main.meta - WHEN NEW.key='assistant_grant_attachment_witness_cursor' BEGIN - INSERT INTO witness_visits(base_claim_id) VALUES(NEW.value); + WHEN NEW.key='assistant_grant_owner_cursor/witness' BEGIN + INSERT INTO witness_visits(base_claim_id) + SELECT owner FROM assistant_grant_queue_owners + WHERE queue='witness' AND sequence=CAST(NEW.value AS INTEGER); END;", ).unwrap(); let limit = if budget.is_zero() { 1 } else { authority::CLAIMS_PER_SLICE }; @@ -561,3 +563,80 @@ fn ineligible_witness_prefix_consumes_bounded_visits_in_zero_and_full_slices() { assert!(finished, "bounded cleanup never reached the eligible tail"); } } + +#[test] +fn attachment_owner_turn_survives_reopen_and_lower_and_between_id_arrivals() { + const BOUND: usize = 256; + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("attachment-owner-fairness.sqlite3"); + let grant; + let policy; + { + let store = Store::open(&path, "attachment-owner-fairness", Arc::new(Plain)).unwrap(); + grant = append(&store, SUBJECT, GRANTED, fields()); + drain_authority(&store); + // Only queue fixtures have controlled IDs. The victim and its base are + // real local appends, with their ordinary content-derived claim IDs. + store.connection.write().execute( + "INSERT INTO assistant_grant_attachments + VALUES('!m/initial',?1,'attacker/subject','missing','missing','','')", + [GRANTED], + ).unwrap(); + let mut value = fields(); + value["expires_at_unix_ms"] = json!(20000); + policy = append(&store, SUBJECT, POLICY, value); + assert!(authority::subject_ready(&store.readers.get(), SUBJECT).is_err()); + } + let mut connection = Connection::open(&path).unwrap(); + attachment_install_counter(&connection); + let mut reopened = false; + let mut lifted = false; + for slice in 0..BOUND { + // New owners arrive both below the already visited attacker and in the + // hash interval before the victim. They must join a later arrival round, + // rather than continually extending the victim's current round. + for prefix in ["!a/lower", "!t/between"] { + let id = format!("{prefix}/{slice:08}"); + assert!(id.as_str()0, + "victim must finish without draining continuously arriving unrelated owners"); + if !reopened && work.candidates>0 { + let after: String = connection.query_row( + "SELECT after_claim_id FROM assistant_grant_attachments + WHERE claim_id=?1 AND candidate_kind=?2", + params![policy.id,GRANTED], |row| row.get(0), + ).unwrap(); + assert_eq!(after, grant.id); + assert!(authority::subject_ready(&connection, SUBJECT).is_err()); + // Reopening Store would itself perform maintenance. A raw reopen + // isolates the durable owner turn/frontier and the unfinished page. + drop(connection); + connection = Connection::open(&path).unwrap(); + attachment_install_counter(&connection); + reopened = true; + } + if authority::subject_ready(&connection, SUBJECT).is_ok() { + assert!(authority::contains(&connection, &grant.id).unwrap()); + assert!(connection.query_row( + "SELECT EXISTS(SELECT 1 FROM assistant_grant_authority + WHERE claim_id=?1 AND base_claim_id=?2)", + params![policy.id,grant.id], |row| row.get::<_,bool>(0), + ).unwrap(), "the victim's actual policy attachment was skipped"); + lifted = true; + break; + } + } + assert!(reopened, "regression never exercised an unfinished victim across reopen"); + assert!(lifted, "lower and between-ID arrivals starved the attachment victim beyond {BOUND} ZERO slices"); + assert!(!connection.query_row( + "SELECT EXISTS(SELECT 1 FROM assistant_grant_attachments WHERE claim_id='!m/initial')", + [], |row| row.get::<_,bool>(0), + ).unwrap(), "stale attacker owners must retire normally"); +} diff --git a/crates/smallclaims/src/store/assistant_grant_authority.rs b/crates/smallclaims/src/store/assistant_grant_authority.rs index c322fea7f33..cfb38d0243d 100644 --- a/crates/smallclaims/src/store/assistant_grant_authority.rs +++ b/crates/smallclaims/src/store/assistant_grant_authority.rs @@ -210,6 +210,7 @@ pub fn install(connection: &Connection, local_origin: &str) -> Result<()> { install_auth_cursor(connection)?; install_proof_invalidation_triggers(connection)?; install_auth_work(connection)?; + install_queue_owners(connection)?; connection.execute(/* sqlite-large-value: allow-for-now lv-h3f117dab1bdd owner=smallclaims::store::assistant_grant_authority TODO=bound-or-reference-values-and-reclamation */ "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_authority_local_origin',?1)", [local_origin])?; connection.execute_batch("RELEASE assistant_grant_upgrade")?; Ok(()) @@ -232,6 +233,106 @@ fn install_auth_cursor(connection: &Connection) -> Result<()> { Ok(()) } +// Owner admission order is local maintenance state, never an attacker-controlled +// content hash. Queue rows keep their existing schemas and inner page cursors. +fn install_queue_owners(connection: &Connection) -> Result<()> { + connection.execute_batch( + "CREATE TABLE IF NOT EXISTS assistant_grant_queue_sequences ( + queue TEXT PRIMARY KEY, sequence INTEGER NOT NULL); + CREATE TABLE IF NOT EXISTS assistant_grant_queue_owners ( + queue TEXT NOT NULL, owner TEXT NOT NULL, subowner TEXT NOT NULL, + sequence INTEGER NOT NULL, PRIMARY KEY(queue,owner,subowner), + UNIQUE(queue,sequence));" + )?; + let installed = fleet_meta(connection,"assistant_grant_queue_owners_installed")?.is_some(); + for (queue,table,owner,subowner) in [ + ("dirty","assistant_grant_dirty","claim_id","''"), + ("recheck","assistant_grant_proof_rechecks","claim_id","''"), + ("invalidation","assistant_grant_proof_invalidations","id","''"), + ("retraction","assistant_grant_retractions","id","''"), + ("addition","assistant_grant_additions","id","''"), + ("attachment","assistant_grant_attachments","claim_id","candidate_kind"), + ("witness","assistant_grant_attachment_witnesses","base_claim_id","''"), + ] { + connection.execute( + "INSERT OR IGNORE INTO assistant_grant_queue_sequences VALUES(?1,0)", [queue], + )?; + let new_subowner = if subowner=="''" { "''".to_owned() } else { format!("NEW.{subowner}") }; + let old_subowner = if subowner=="''" { "''".to_owned() } else { format!("OLD.{subowner}") }; + connection.execute_batch(&format!( + "CREATE TRIGGER IF NOT EXISTS {table}_owner_insert AFTER INSERT ON {table} + WHEN NOT EXISTS(SELECT 1 FROM assistant_grant_queue_owners + WHERE queue='{queue}' AND owner=CAST(NEW.{owner} AS TEXT) AND subowner={new_subowner}) + BEGIN + UPDATE assistant_grant_queue_sequences SET sequence=sequence+1 WHERE queue='{queue}'; + INSERT INTO assistant_grant_queue_owners + SELECT '{queue}',CAST(NEW.{owner} AS TEXT),{new_subowner},sequence + FROM assistant_grant_queue_sequences WHERE queue='{queue}'; + END; + CREATE TRIGGER IF NOT EXISTS {table}_owner_delete AFTER DELETE ON {table} + BEGIN + DELETE FROM assistant_grant_queue_owners WHERE queue='{queue}' + AND owner=CAST(OLD.{owner} AS TEXT) AND subowner={old_subowner}; + END;" + ))?; + if !installed { + // Only the existing work set is indexed once on upgrade; raw claim + // history and already-completed authority are never replayed. + connection.execute_batch(&format!( + "INSERT INTO assistant_grant_queue_owners + SELECT '{queue}',CAST({owner} AS TEXT),{subowner}, + ROW_NUMBER() OVER(ORDER BY {owner},{subowner}) + FROM {table}; + UPDATE assistant_grant_queue_sequences SET sequence=COALESCE( + (SELECT MAX(sequence) FROM assistant_grant_queue_owners WHERE queue='{queue}'),0) + WHERE queue='{queue}';" + ))?; + } + } + connection.execute( + "DELETE FROM meta WHERE key IN ('assistant_grant_dirty_cursor','assistant_grant_attachment_witness_cursor')", + [], + )?; + connection.execute( + "INSERT OR IGNORE INTO meta(key,value) VALUES('assistant_grant_queue_owners_installed','1')", [], + )?; + Ok(()) +} + +fn queue_owner(connection: &Connection, queue: &str) -> Result> { + let cursor_key = format!("assistant_grant_owner_cursor/{queue}"); + let upper_key = format!("assistant_grant_owner_upper/{queue}"); + let after = fleet_meta(connection,&cursor_key)?.and_then(|value| value.parse::().ok()).unwrap_or(0); + let upper = fleet_meta(connection,&upper_key)?.and_then(|value| value.parse::().ok()).unwrap_or(0); + let seek = |after: i64, upper: i64| -> Result> { + Ok(connection.query_row( + "SELECT owner,subowner,sequence FROM assistant_grant_queue_owners + WHERE queue=?1 AND sequence>?2 AND sequence<=?3 ORDER BY sequence LIMIT 1", + params![queue,after,upper], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?)), + ).optional()?) + }; + let candidate = match seek(after,upper)? { + Some(candidate) => Some(candidate), + None => { + let upper: i64 = connection.query_row( + "SELECT sequence FROM assistant_grant_queue_sequences WHERE queue=?1", + [queue], |row| row.get(0), + )?; + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES(?1,?2)", params![upper_key,upper], + )?; + seek(0,upper)? + } + }; + let Some((owner,subowner,sequence)) = candidate else { return Ok(None); }; + // The fixed round frontier excludes all later arrivals, including arrivals + // above the current cursor. An unfinished owner resumes next round. + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES(?1,?2)", params![cursor_key,sequence], + )?; + Ok(Some((owner,subowner))) +} + fn install_auth_work(connection: &Connection) -> Result<()> { let old_additions: bool = connection.query_row( "SELECT NOT EXISTS(SELECT 1 FROM pragma_table_info('assistant_grant_additions') WHERE name='claim_id')", @@ -582,9 +683,10 @@ fn attach_claim(connection: &Connection, claim: &ClaimRecord) -> Result<()> { } fn attachment_one(connection: &Connection) -> Result { + let Some((owner,partition)) = queue_owner(connection,"attachment")? else { return Ok(false); }; let task: Option<(String,String,String,String,String,String,String)> = connection.query_row( "SELECT claim_id,candidate_kind,grant_subject,origin,actor,after_claim_id,upper_claim_id - FROM assistant_grant_attachments ORDER BY claim_id,candidate_kind LIMIT 1", [], + FROM assistant_grant_attachments WHERE claim_id=?1 AND candidate_kind=?2", params![owner,partition], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?,row.get(5)?,row.get(6)?)), ).optional()?; let Some((id,kind,subject,origin,actor,after,upper)) = task else { return Ok(false); }; @@ -633,20 +735,11 @@ fn attachment_one(connection: &Connection) -> Result { } fn attachment_witness_one(connection: &Connection) -> Result { - let after = fleet_meta(connection,"assistant_grant_attachment_witness_cursor")?.unwrap_or_default(); - let seek = |after: &str| -> Result> { - Ok(connection.query_row( - "SELECT base_claim_id,grant_subject FROM assistant_grant_attachment_witnesses - WHERE base_claim_id>?1 ORDER BY base_claim_id LIMIT 1", - [after], |row| Ok((row.get(0)?,row.get(1)?)), - ).optional()?) - }; - let candidate = match seek(&after)? { - Some(candidate) => Some(candidate), - None if !after.is_empty() => seek("")?, - None => None, - }; - let Some((base,subject)) = candidate else { return Ok(false); }; + let Some((base,_)) = queue_owner(connection,"witness")? else { return Ok(false); }; + let subject: String = connection.query_row( + "SELECT grant_subject FROM assistant_grant_attachment_witnesses WHERE base_claim_id=?1", + [&base], |row| row.get(0), + )?; let pending: bool = connection.query_row( "SELECT EXISTS(SELECT 1 FROM assistant_grant_attachments WHERE grant_subject=?1)", [&subject], |row| row.get(0), @@ -654,10 +747,6 @@ fn attachment_witness_one(connection: &Connection) -> Result { if !pending { connection.execute("DELETE FROM assistant_grant_attachment_witnesses WHERE base_claim_id=?1", [&base])?; } - connection.execute( - "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_attachment_witness_cursor',?1)", - [&base], - )?; Ok(true) } fn local_origin(connection: &Connection) -> Result { @@ -1049,9 +1138,10 @@ pub fn backfill_slice(connection: &Connection, budget: Duration) -> Result // history-sized dependent set. Indexed dispatch and real authentication consume // separate shared budget units. Only the verifier may discharge the latter. fn proof_invalidation_one(connection: &Connection) -> Result { + let Some((owner,_)) = queue_owner(connection,"invalidation")? else { return Ok(false); }; let task: Option<(i64,String,u64,String,String,String)> = connection.query_row( "SELECT id,writer,sequence,envelope_hash,member_key,claim_id - FROM assistant_grant_proof_invalidations ORDER BY id LIMIT 1", [], + FROM assistant_grant_proof_invalidations WHERE id=?1", [&owner], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?,row.get(5)?)), ).optional()?; let Some((task,writer,sequence,hash,key,cursor)) = task else { return Ok(false); }; @@ -1110,9 +1200,10 @@ fn trim_restored_retraction(connection: &Connection, task: i64, writer: &str, ke // Each addition seeks only indexed grant work, then that grant's own receipts. // A claim selection, receipt verification or completed claim costs one unit. fn addition_one(connection: &Connection) -> Result { + let Some((owner,_)) = queue_owner(connection,"addition")? else { return Ok(false); }; let task: Option<(i64,String,String,u64,u64,i64,String,i64,String,String)> = connection.query_row( "SELECT id,writer,member_key,lower_sequence,upper_sequence,sequence,envelope_hash,position,claim_id,active_claim - FROM assistant_grant_additions ORDER BY id LIMIT 1", [], + FROM assistant_grant_additions WHERE id=?1", [&owner], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?,row.get(5)?,row.get(6)?,row.get(7)?,row.get(8)?,row.get(9)?)), ).optional()?; let Some((task,writer,key,lower,upper,sequence,hash,position,after,claim)) = task else { return Ok(false); }; @@ -1198,19 +1289,7 @@ fn refresh_recheck(connection: &Connection, claim: &str) -> Result { } fn dirty_one(connection: &Connection, membership_current: bool) -> Result { - let after = fleet_meta(connection,"assistant_grant_dirty_cursor")?.unwrap_or_default(); - let seek = |after: &str| -> Result> { - Ok(connection.query_row( - "SELECT claim_id FROM assistant_grant_dirty WHERE claim_id>?1 ORDER BY claim_id LIMIT 1", - [after], |row| row.get(0), - ).optional()?) - }; - let claim = match seek(&after)? { - Some(claim) => Some(claim), - None if !after.is_empty() => seek("")?, - None => None, - }; - let Some(claim) = claim else { return Ok(false); }; + let Some((claim,_)) = queue_owner(connection,"dirty")? else { return Ok(false); }; let recheck_owned: bool = connection.query_row( "SELECT EXISTS(SELECT 1 FROM assistant_grant_proof_rechecks WHERE claim_id=?1)", [&claim], |row| row.get(0), @@ -1220,26 +1299,20 @@ fn dirty_one(connection: &Connection, membership_current: bool) -> Result } else { refresh_claim_inner(connection,&claim,membership_current)?; } - connection.execute( - "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_dirty_cursor',?1)", [&claim], - )?; Ok(true) } fn recheck_one(connection: &Connection) -> Result { - let claim: Option = connection.query_row( - "SELECT claim_id FROM assistant_grant_proof_rechecks ORDER BY claim_id LIMIT 1", - [], |row| row.get(0), - ).optional()?; - let Some(claim) = claim else { return Ok(false); }; + let Some((claim,_)) = queue_owner(connection,"recheck")? else { return Ok(false); }; refresh_recheck(connection,&claim)?; Ok(true) } fn retraction_one(connection: &Connection) -> Result { + let Some((owner,_)) = queue_owner(connection,"retraction")? else { return Ok(false); }; let task: Option<(i64,String,String,u64,u64,i64,String)> = connection.query_row( "SELECT id,writer,member_key,lower_sequence,upper_sequence,sequence,claim_id - FROM assistant_grant_retractions ORDER BY id LIMIT 1", [], + FROM assistant_grant_retractions WHERE id=?1", [&owner], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?,row.get(5)?,row.get(6)?)), ).optional()?; let Some((task,writer,key,lower,upper,sequence,id)) = task else { return Ok(false); }; diff --git a/crates/smallclaims/src/store/assistant_grant_authority_tests.rs b/crates/smallclaims/src/store/assistant_grant_authority_tests.rs index afc7b0f22ff..e068f69231b 100644 --- a/crates/smallclaims/src/store/assistant_grant_authority_tests.rs +++ b/crates/smallclaims/src/store/assistant_grant_authority_tests.rs @@ -1471,17 +1471,88 @@ fn dirty_work_before_saved_round_robin_cursor_is_serviced_in_the_same_slice() { let grant = append(&store,SUBJECT,GRANTED,fields()); drain_authority(&store); let mut connection = store.connection.write(); - connection.execute( - "INSERT OR REPLACE INTO meta(key,value) VALUES('assistant_grant_dirty_cursor',?1)", - [format!("{}z",grant.id)], - ).unwrap(); connection.execute("INSERT INTO assistant_grant_dirty VALUES(?1)", [&grant.id]).unwrap(); + let sequence: i64 = connection.query_row( + "SELECT sequence FROM assistant_grant_queue_owners WHERE queue='dirty' AND owner=?1", + [&grant.id], |row| row.get(0), + ).unwrap(); + for key in ["assistant_grant_owner_cursor/dirty", "assistant_grant_owner_upper/dirty"] { + connection.execute( + "INSERT OR REPLACE INTO meta(key,value) VALUES(?1,?2)", + params![key,sequence+1], + ).unwrap(); + } let transaction = connection.transaction().unwrap(); assert!(authority::backfill_slice(&transaction,Duration::from_secs(60)).unwrap()); assert!(authority::contains(&transaction,&grant.id).unwrap()); transaction.commit().unwrap(); } +#[test] +fn authenticated_proof_recheck_owner_finishes_despite_lower_and_between_id_arrivals() { + const BOUND: usize = 64; + let (source, key) = source(); + let target = node("proof-owner-fairness", Some(key.public())); + let grant = append(&source, SUBJECT, GRANTED, fields()); + receive(&target, &export(&source, &target)); + let mut connection = target.connection.write(); + assert!(authority::contains(&connection, &grant.id).unwrap()); + let proof: (String,u64,String,i64) = connection.query_row( + "SELECT auth_member_key,auth_sequence,auth_envelope_hash,auth_record_position + FROM assistant_grant_pending WHERE claim_id=?1 AND authenticated=1", + [&grant.id], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?)), + ).unwrap(); + // The victim keeps its real signed claim and evidence. Controlled IDs exist + // only as stale queue owners whose missing claims should be retired. + connection.execute( + "INSERT INTO assistant_grant_proof_rechecks VALUES('!m/initial')", [], + ).unwrap(); + connection.execute( + "INSERT INTO assistant_grant_proof_rechecks VALUES(?1)", [&grant.id], + ).unwrap(); + assert!(authority::subject_ready(&connection, SUBJECT).is_err()); + let mut lifted = false; + for slice in 0..BOUND { + for prefix in ["!a/lower", "!t/between"] { + let id = format!("{prefix}/{slice:08}"); + assert!(id.as_str()?1)", [&grant.id], |row| row.get::<_,bool>(0), + ).unwrap(), "the victim must finish while unrelated rechecks remain"); + let ready = authority::subject_ready(&transaction, SUBJECT).is_ok(); + transaction.commit().unwrap(); + if ready { + assert!(authority::contains(&connection, &grant.id).unwrap()); + let rechecked: (String,u64,String,i64) = connection.query_row( + "SELECT auth_member_key,auth_sequence,auth_envelope_hash,auth_record_position + FROM assistant_grant_pending WHERE claim_id=?1 AND authenticated=1", + [&grant.id], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?)), + ).unwrap(); + assert_eq!(rechecked, proof, "recheck lost the victim's genuine authentication evidence"); + assert!(!connection.query_row( + "SELECT EXISTS(SELECT 1 FROM assistant_grant_proof_rechecks WHERE claim_id=?1)", + [&grant.id], |row| row.get::<_,bool>(0), + ).unwrap(), "the subject fence lifted before its proof obligation retired"); + lifted = true; + break; + } + } + assert!(lifted, "lower and between-ID arrivals starved the proof victim beyond {BOUND} ZERO slices"); + assert!(!connection.query_row( + "SELECT EXISTS(SELECT 1 FROM assistant_grant_proof_rechecks WHERE claim_id='!m/initial')", + [], |row| row.get::<_,bool>(0), + ).unwrap(), "stale missing-claim rechecks must retire normally"); +} + include!("assistant_grant_receipt_cost_tests.rs"); include!("assistant_grant_attachment_tests.rs"); diff --git a/crates/smallclaims/src/store/assistant_grant_receipt_cost_tests.rs b/crates/smallclaims/src/store/assistant_grant_receipt_cost_tests.rs index 187f1db48b7..4cfbd637e40 100644 --- a/crates/smallclaims/src/store/assistant_grant_receipt_cost_tests.rs +++ b/crates/smallclaims/src/store/assistant_grant_receipt_cost_tests.rs @@ -48,6 +48,9 @@ fn receipt_projection_work_state(connection: &Connection) -> Vec