diff --git a/crates/tracedecay-global-db/src/lib.rs b/crates/tracedecay-global-db/src/lib.rs index b0106dcfc6e..6f21dec711b 100644 --- a/crates/tracedecay-global-db/src/lib.rs +++ b/crates/tracedecay-global-db/src/lib.rs @@ -85,9 +85,10 @@ mod stack_delivery; mod store_registration; pub use schema_stages::ensure_registered_schema; pub use stack_delivery::{ - GitHubStackDeliveryRecordV1, GitHubStackDeliveryStateV1, GitHubStackSignalAppendOutcomeV1, - GitHubStackSignalRecordV1, MAX_GITHUB_STACK_ACTIVE_PENDING_V1, - MAX_GITHUB_STACK_DELIVERY_BATCH_V1, + GITHUB_STACK_TERMINAL_RETENTION_DAYS_V1, GitHubStackDeliveryRecordV1, + GitHubStackDeliveryStateV1, GitHubStackSignalAppendOutcomeV1, GitHubStackSignalRecordV1, + GitHubStackTerminalRetirementPageV1, MAX_GITHUB_STACK_ACTIVE_PENDING_V1, + MAX_GITHUB_STACK_DELIVERY_BATCH_V1, MAX_GITHUB_STACK_RETIRE_BATCH_V1, }; /// Installs the canonical registered global/session schema installer into the diff --git a/crates/tracedecay-global-db/src/stack_delivery.rs b/crates/tracedecay-global-db/src/stack_delivery.rs index 633af133335..028c5652d5b 100644 --- a/crates/tracedecay-global-db/src/stack_delivery.rs +++ b/crates/tracedecay-global-db/src/stack_delivery.rs @@ -8,6 +8,7 @@ use std::collections::BTreeSet; use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params, params_from_iter}; use crate::{RegisteredGlobalDb, registered::JobExit}; @@ -22,6 +23,23 @@ pub const MAX_GITHUB_STACK_DELIVERY_BATCH_V1: usize = 256; /// before they reach the registered store. const MAX_GITHUB_STACK_ID_BYTES: usize = 512; const MAX_GITHUB_STACK_SIGNAL_JSON_BYTES: usize = 1_048_576; +/// The supported inspection/replay window is measured from signal issuance. +/// Older terminal-only records can be compacted without losing native receipts. +pub const GITHUB_STACK_TERMINAL_RETENTION_DAYS_V1: i64 = 14; +/// Each maintenance transaction holds no more than 16 1-MiB signal payloads. +pub const MAX_GITHUB_STACK_RETIRE_BATCH_V1: usize = 16; +const RETIRED_PAYLOAD_KEY: &str = "__tracedecay_retired_stack_sha256_v1"; +/// Bound raw candidates examined even when the entire old prefix is deferred. +pub(crate) const MAX_GITHUB_STACK_RETIRE_SCAN_V1: usize = 64; + +/// One atomic retirement slice. An empty retirement can still advance the +/// durable scan frontier; callers must continue until the scan is exhausted. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct GitHubStackTerminalRetirementPageV1 { + pub retired: usize, + pub inspected: usize, + pub exhausted: bool, +} /// Neutral, serialized representation of an immutable stack signal. /// @@ -131,7 +149,7 @@ pub(crate) async fn ensure_github_stack_delivery_schema( watermark_id TEXT NOT NULL, observed_at_micros INTEGER NOT NULL CHECK(observed_at_micros > 0), signal_json TEXT NOT NULL - CHECK(json_valid(signal_json) AND length(signal_json) <= 1048576), + CHECK(json_valid(signal_json) AND octet_length(signal_json) <= 1048576), PRIMARY KEY(project_id, signal_id) ) STRICT; CREATE TABLE IF NOT EXISTS github_stack_delivery_recipients ( @@ -146,6 +164,12 @@ pub(crate) async fn ensure_github_stack_delivery_schema( REFERENCES github_stack_delivery_signals(project_id, signal_id) ON DELETE RESTRICT ) STRICT; + CREATE TABLE IF NOT EXISTS github_stack_delivery_retirement_frontier ( + singleton INTEGER PRIMARY KEY CHECK(singleton = 1), + observed_at_micros INTEGER NOT NULL CHECK(observed_at_micros > 0), + project_id TEXT NOT NULL, + signal_id TEXT NOT NULL + ) STRICT; CREATE INDEX IF NOT EXISTS idx_github_stack_delivery_pending_recipient ON github_stack_delivery_recipients( project_id, recipient, signal_id @@ -154,8 +178,59 @@ pub(crate) async fn ensure_github_stack_delivery_schema( ON github_stack_delivery_recipients( project_id, recipient, signal_id ) WHERE state = 'deferred'; - CREATE TRIGGER IF NOT EXISTS github_stack_delivery_signals_immutable_update + CREATE INDEX IF NOT EXISTS idx_github_stack_delivery_unretired_cursor + ON github_stack_delivery_signals( + observed_at_micros, project_id, signal_id + ) + WHERE json_type(signal_json, '$.__tracedecay_retired_stack_sha256_v1') IS NULL; + -- Existing registered stores retain the original character-count + -- CHECK. CREATE TABLE IF NOT EXISTS cannot change its constraint. + CREATE TRIGGER IF NOT EXISTS github_stack_delivery_signals_byte_limit + BEFORE INSERT ON github_stack_delivery_signals + WHEN octet_length(NEW.signal_json) > 1048576 + BEGIN + SELECT RAISE(ABORT, 'GitHub stack signal JSON exceeds 1 MiB'); + END; + -- Raw SQL must not mint an already-retired identity. Only the + -- guarded UPDATE may retire a settled signal after the window. + -- json_each sees escaped and duplicate top-level member names. + CREATE TRIGGER IF NOT EXISTS github_stack_delivery_signals_insert_unretired + BEFORE INSERT ON github_stack_delivery_signals + WHEN json_valid(NEW.signal_json) AND EXISTS ( + SELECT 1 FROM json_each(NEW.signal_json) + WHERE key = '__tracedecay_retired_stack_sha256_v1' + ) + BEGIN + SELECT RAISE(ABORT, 'GitHub stack signal uses a reserved retirement marker'); + END; + DROP TRIGGER IF EXISTS github_stack_delivery_signals_immutable_update; + CREATE TRIGGER github_stack_delivery_signals_immutable_update BEFORE UPDATE ON github_stack_delivery_signals + WHEN OLD.project_id != NEW.project_id + OR OLD.signal_id != NEW.signal_id + OR OLD.scope_digest != NEW.scope_digest + OR OLD.repository_id != NEW.repository_id + OR OLD.watermark_id != NEW.watermark_id + OR OLD.observed_at_micros != NEW.observed_at_micros + -- The issued inspection window is enforced at the registered + -- storage boundary, including direct SQL updates. + OR OLD.observed_at_micros > + (CAST(strftime('%s', 'now') AS INTEGER) - {retention_days} * 86400) + * 1000000 + OR json_type(OLD.signal_json, '$.__tracedecay_retired_stack_sha256_v1') IS NOT NULL + OR EXISTS ( + SELECT 1 FROM json_each(NEW.signal_json) LIMIT 1 OFFSET 1 + ) + OR json_type(NEW.signal_json, '$.__tracedecay_retired_stack_sha256_v1') IS NOT 'text' + OR length(json_extract(NEW.signal_json, '$.__tracedecay_retired_stack_sha256_v1')) != 64 + OR json_extract(NEW.signal_json, '$.__tracedecay_retired_stack_sha256_v1') + GLOB '*[^0-9a-f]*' + OR EXISTS ( + SELECT 1 FROM github_stack_delivery_recipients AS r + WHERE r.project_id = OLD.project_id + AND r.signal_id = OLD.signal_id + AND r.state IN ('pending', 'deferred') + ) BEGIN SELECT RAISE(ABORT, 'GitHub stack delivery signals are immutable'); END; @@ -182,13 +257,22 @@ pub(crate) async fn ensure_github_stack_delivery_schema( BEGIN SELECT RAISE(ABORT, 'invalid GitHub stack delivery state transition'); END; - CREATE TRIGGER IF NOT EXISTS github_stack_delivery_recipients_no_delete + DROP TRIGGER IF EXISTS github_stack_delivery_recipients_no_delete; + CREATE TRIGGER github_stack_delivery_recipients_no_delete BEFORE DELETE ON github_stack_delivery_recipients + WHEN OLD.state IN ('pending', 'deferred') + OR NOT EXISTS ( + SELECT 1 FROM github_stack_delivery_signals AS s + WHERE s.project_id = OLD.project_id + AND s.signal_id = OLD.signal_id + AND json_type(s.signal_json, '$.__tracedecay_retired_stack_sha256_v1') = 'text' + ) BEGIN - SELECT RAISE(ABORT, 'GitHub stack delivery recipient rows are retained'); + SELECT RAISE(ABORT, 'GitHub stack delivery recipient is not retired'); END; {promote}", - promote = promote_deferred_sql("") + promote = promote_deferred_sql(""), + retention_days = GITHUB_STACK_TERMINAL_RETENTION_DAYS_V1 ); connection.execute_batch(&schema).await.map_err(|error| { crate::global_db_operation_error("initialize GitHub stack delivery schema", error) @@ -222,11 +306,68 @@ fn validate_signal(record: &GitHubStackSignalRecordV1) -> Result<(), String> { if record.signal_json.len() > MAX_GITHUB_STACK_SIGNAL_JSON_BYTES { return Err("GitHub stack signal JSON exceeds 1 MiB".to_owned()); } - serde_json::from_str::(&record.signal_json) + let payload: serde_json::Value = serde_json::from_str(&record.signal_json) .map_err(|error| format!("GitHub stack signal JSON is invalid: {error}"))?; + if payload + .as_object() + .is_some_and(|object| object.contains_key(RETIRED_PAYLOAD_KEY)) + { + return Err("GitHub stack signal uses a reserved retirement marker".to_owned()); + } Ok(()) } +/// A compacted payload is only an identity fingerprint, never an expandable +/// native signal. Keep the signal key to reject late replays without retaining +/// a second copy of the native preview/terminal receipt. +/// Recognize any valid JSON representation of the retired identity. A +/// noncanonical spacing choice must never turn a retired marker back into an +/// expandable native signal; corrupt markers fail closed instead. +fn retired_payload_digest(json: &str) -> Result, String> { + // JSON member names may contain Unicode escapes. A raw substring check + // would miss a retired key accepted by SQLite's json_type() guard and + // could expose its fingerprint as if it were a live signal. + let value: serde_json::Value = serde_json::from_str(json) + .map_err(|error| format!("decode GitHub stack retirement marker: {error}"))?; + let Some(object) = value.as_object() else { + return Ok(None); + }; + let Some(marker) = object.get(RETIRED_PAYLOAD_KEY) else { + return Ok(None); + }; + let digest = marker + .as_str() + .ok_or_else(|| "GitHub stack retirement marker is not a string".to_owned())?; + if object.len() != 1 + || digest.len() != 64 + || !digest + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + return Err("GitHub stack retirement marker is malformed".to_owned()); + } + Ok(Some(digest.to_owned())) +} + +fn matches_replayed_signal( + existing: &GitHubStackSignalRecordV1, + offered: &GitHubStackSignalRecordV1, +) -> bool { + if existing == offered { + return true; + } + let Ok(Some(retired_hash)) = retired_payload_digest(&existing.signal_json) else { + return false; + }; + existing.project_id == offered.project_id + && existing.signal_id == offered.signal_id + && existing.scope_digest == offered.scope_digest + && existing.repository_id == offered.repository_id + && existing.watermark_id == offered.watermark_id + && existing.observed_at_micros == offered.observed_at_micros + && retired_hash == hex::encode(Sha256::digest(offered.signal_json.as_bytes())) +} + fn validate_recipient(recipient: &str) -> Result<(), String> { validate_text(recipient, "recipient") } @@ -442,7 +583,7 @@ impl RegisteredGlobalDb { if let Some(existing) = lookup_signal(&transaction, &record.project_id, &record.signal_id).await? { - if existing != record { + if !matches_replayed_signal(&existing, &record) { return Err("GitHub stack signal identity conflict".to_owned().into()); } let (pending_count, deferred_count) = @@ -740,7 +881,245 @@ impl RegisteredGlobalDb { .read_snapshot() .await .map_err(|error| format!("begin GitHub stack signal lookup: {error}"))?; - lookup_signal(&snapshot, project_id, signal_id).await + let Some(signal) = lookup_signal(&snapshot, project_id, signal_id).await? else { + return Ok(None); + }; + if retired_payload_digest(&signal.signal_json)?.is_some() { + return Ok(None); + } + Ok(Some(signal)) + } + + /// Scan a bounded ordered window of old signal identities, and compact + /// only terminal rows. The frontier is committed alongside the mutations + /// so a restart cannot repeatedly rescan a blocked pending/deferred prefix. + /// + /// At the end of the keyspace the frontier wraps: an earlier pending + /// signal which later settles will become eligible on a subsequent pass. + /// Neither the scan nor the eligible payload count can grow with backlog. + #[tracing::instrument( + name = "global_db.stack_delivery.retire_terminal", + level = "trace", + skip_all + )] + pub async fn retire_github_stack_terminal_payloads_before( + &self, + cutoff_micros: i64, + ) -> Result { + // Match SQLite's conservative whole-second retirement-age guard so + // an eligible application page cannot fail at the fractional boundary. + let oldest_permitted_micros = tracedecay_contracts::now_micros() + .0 + .saturating_sub(GITHUB_STACK_TERMINAL_RETENTION_DAYS_V1 * 86_400 * 1_000_000) + .div_euclid(1_000_000) + .saturating_mul(1_000_000); + let cutoff_micros = cutoff_micros.min(oldest_permitted_micros); + if cutoff_micros <= 0 { + return Ok(GitHubStackTerminalRetirementPageV1 { + retired: 0, + inspected: 0, + exhausted: true, + }); + } + self.transact_or( + |error| format!("retire terminal GitHub stack payloads: {error}"), + async move |transaction| { + let (after_time, after_project, after_signal) = { + let mut rows = transaction + .query( + "SELECT observed_at_micros, project_id, signal_id + FROM github_stack_delivery_retirement_frontier + WHERE singleton = 1", + (), + ) + .await + .map_err(|error| format!("read stack retirement frontier: {error}"))?; + match rows + .next() + .await + .map_err(|error| format!("read stack retirement frontier row: {error}"))? + { + Some(row) => ( + row.get::(0) + .map_err(|error| format!("decode stack frontier time: {error}"))?, + row.get::(1).map_err(|error| { + format!("decode stack frontier project: {error}") + })?, + row.get::(2).map_err(|error| { + format!("decode stack frontier signal: {error}") + })?, + ), + None => (0, String::new(), String::new()), + } + }; + // Do not select signal_json or filter out pending recipients + // here: both would make LIMIT a bound on returned rows rather + // than on the work examined inside this writer transaction. + let mut rows = transaction + .query( + "SELECT s.observed_at_micros, s.project_id, s.signal_id + FROM github_stack_delivery_signals AS s + WHERE s.observed_at_micros < ?1 + AND json_type(s.signal_json, + '$.__tracedecay_retired_stack_sha256_v1') IS NULL + AND (s.observed_at_micros, s.project_id, s.signal_id) + > (?2, ?3, ?4) + ORDER BY s.observed_at_micros, s.project_id, s.signal_id + LIMIT ?5", + params![ + cutoff_micros, + after_time, + after_project.as_str(), + after_signal.as_str(), + MAX_GITHUB_STACK_RETIRE_SCAN_V1 as i64, + ], + ) + .await + .map_err(|error| format!("scan bounded GitHub stack identities: {error}"))?; + let mut keys = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|error| format!("read bounded GitHub stack identity: {error}"))? + { + keys.push(( + row.get::(0) + .map_err(|error| format!("decode stack candidate time: {error}"))?, + row.get::(1) + .map_err(|error| format!("decode stack candidate project: {error}"))?, + row.get::(2) + .map_err(|error| format!("decode stack candidate signal: {error}"))?, + )); + } + drop(rows); + + let mut inspected = 0usize; + let mut retired = 0usize; + for (observed_at, project_id, signal_id) in &keys { + if retired == MAX_GITHUB_STACK_RETIRE_BATCH_V1 { + break; + } + inspected += 1; + let mut pending = transaction + .query( + "SELECT EXISTS ( + SELECT 1 FROM github_stack_delivery_recipients + WHERE project_id = ?1 AND signal_id = ?2 + AND state IN ('pending', 'deferred') + )", + params![project_id.as_str(), signal_id.as_str()], + ) + .await + .map_err(|error| format!("test stack terminal eligibility: {error}"))?; + let blocked = pending + .next() + .await + .map_err(|error| format!("read stack terminal eligibility: {error}"))? + .ok_or_else(|| "stack eligibility query returned no row".to_owned())? + .get::(0) + .map_err(|error| format!("decode stack terminal eligibility: {error}"))? + != 0; + drop(pending); + if blocked { + continue; + } + // Load at most 16 eligible payloads, each already bounded + // by the store's 1 MiB CHECK. Ineligible rows cost only an + // indexed identity and recipient-status probe. + let mut payload_rows = transaction + .query( + "SELECT signal_json FROM github_stack_delivery_signals + WHERE project_id = ?1 AND signal_id = ?2 + AND observed_at_micros = ?3 + AND json_type(signal_json, + '$.__tracedecay_retired_stack_sha256_v1') IS NULL", + params![project_id.as_str(), signal_id.as_str(), *observed_at], + ) + .await + .map_err(|error| format!("read eligible stack payload: {error}"))?; + let payload = payload_rows + .next() + .await + .map_err(|error| format!("read eligible stack payload row: {error}"))? + .ok_or_else(|| "stack signal changed during retirement".to_owned())? + .get::(0) + .map_err(|error| format!("decode eligible stack payload: {error}"))?; + drop(payload_rows); + let digest = hex::encode(Sha256::digest(payload.as_bytes())); + let compact = + format!(r#"{{"__tracedecay_retired_stack_sha256_v1":"{digest}"}}"#); + let changed = transaction + .execute( + "UPDATE github_stack_delivery_signals SET signal_json = ?3 + WHERE project_id = ?1 AND signal_id = ?2 + AND signal_json = ?4 AND observed_at_micros < ?5 + AND NOT EXISTS ( + SELECT 1 FROM github_stack_delivery_recipients + WHERE project_id = ?1 AND signal_id = ?2 + AND state IN ('pending', 'deferred') + )", + params![ + project_id.as_str(), + signal_id.as_str(), + compact.as_str(), + payload.as_str(), + cutoff_micros, + ], + ) + .await + .map_err(|error| format!("compact GitHub stack signal: {error}"))?; + if changed != 1 { + return Err( + "terminal GitHub stack signal changed during retirement".to_owned() + ); + } + transaction + .execute( + "DELETE FROM github_stack_delivery_recipients + WHERE project_id = ?1 AND signal_id = ?2 + AND state IN ('settled', 'authorization_lost')", + params![project_id.as_str(), signal_id.as_str()], + ) + .await + .map_err(|error| format!("retire GitHub stack recipient rows: {error}"))?; + retired += 1; + } + let exhausted = + inspected == keys.len() && keys.len() < MAX_GITHUB_STACK_RETIRE_SCAN_V1; + if exhausted { + // A wrap revisits old blocked rows that have since settled. + // Empty and completed passes must not leave a stale cursor. + transaction + .execute( + "DELETE FROM github_stack_delivery_retirement_frontier + WHERE singleton = 1", + (), + ) + .await + .map_err(|error| format!("wrap stack retirement frontier: {error}"))?; + } else if let Some((time, project_id, signal_id)) = keys.get(inspected - 1) { + transaction + .execute( + "INSERT INTO github_stack_delivery_retirement_frontier + (singleton, observed_at_micros, project_id, signal_id) + VALUES (1, ?1, ?2, ?3) + ON CONFLICT(singleton) DO UPDATE SET + observed_at_micros = excluded.observed_at_micros, + project_id = excluded.project_id, + signal_id = excluded.signal_id", + params![*time, project_id.as_str(), signal_id.as_str()], + ) + .await + .map_err(|error| format!("advance stack retirement frontier: {error}"))?; + } + Ok(GitHubStackTerminalRetirementPageV1 { + retired, + inspected, + exhausted, + }) + }, + ) + .await } /// Returns one recipient's durable binding state. diff --git a/crates/tracedecay-global-db/src/stack_delivery_tests.rs b/crates/tracedecay-global-db/src/stack_delivery_tests.rs index fdf79bad25a..d2864047252 100644 --- a/crates/tracedecay-global-db/src/stack_delivery_tests.rs +++ b/crates/tracedecay-global-db/src/stack_delivery_tests.rs @@ -1,3 +1,7 @@ +use sha2::Digest; +use tracedecay_runtime_core::db::engine::params; + +use crate::tests::harness::writer_telemetry; use crate::{ GitHubStackDeliveryStateV1, GitHubStackSignalAppendOutcomeV1, GitHubStackSignalRecordV1, MAX_GITHUB_STACK_ACTIVE_PENDING_V1, tests::harness::RegisteredGlobalDbHarness, @@ -31,6 +35,150 @@ async fn state( .expect("read recipient state") } +#[asupersync::test] +async fn registered_stack_signal_limit_counts_utf8_bytes_not_characters() { + let harness = RegisteredGlobalDbHarness::open("github-stack-utf8-byte-limit").await; + // SQLite length(TEXT) counts characters; the on-disk and Rust limits + // count UTF-8 bytes. This valid JSON fits the former but not the latter. + let mut oversized = signal("signal.github-stack-utf8-oversize", 10); + oversized.signal_json = format!(r#"{{"payload":"{}"}}"#, "é".repeat(600_000)); + assert!(oversized.signal_json.chars().count() < 1_048_576); + assert!(oversized.signal_json.len() > 1_048_576); + + let writer = harness + .registered + .writer_connection() + .expect("registered stack byte-limit writer"); + let refusal = writer + .execute( + "INSERT INTO github_stack_delivery_signals + (project_id, signal_id, scope_digest, repository_id, + watermark_id, observed_at_micros, signal_json) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![ + oversized.project_id.as_str(), + oversized.signal_id.as_str(), + oversized.scope_digest.as_str(), + oversized.repository_id.as_str(), + oversized.watermark_id.as_str(), + oversized.observed_at_micros, + oversized.signal_json.as_str(), + ], + ) + .await + .expect_err("raw registered SQL must reject a payload over the UTF-8 byte limit") + .to_string(); + assert!( + refusal.contains("GitHub stack signal JSON exceeds 1 MiB"), + "{refusal}" + ); + assert!( + harness + .registered + .github_stack_signal(PROJECT_ID, &oversized.signal_id) + .await + .expect("failed raw insert must leave no signal") + .is_none() + ); + + // The application boundary must make the same refusal rather than + // relying on SQLite to accept a second, larger copy of the history. + assert!( + harness + .registered + .append_github_stack_signal(oversized, vec![RECIPIENT.to_owned()]) + .await + .is_err() + ); + let mut admitted = signal("signal.github-stack-utf8-within-limit", 11); + admitted.signal_json = format!(r#"{{"payload":"{}"}}"#, "é".repeat(400_000)); + harness + .registered + .append_github_stack_signal(admitted.clone(), vec![RECIPIENT.to_owned()]) + .await + .expect("non-ASCII bytes within the 1 MiB limit remain admissible"); + assert_eq!( + harness + .registered + .github_stack_signal(PROJECT_ID, &admitted.signal_id) + .await + .expect("read admitted non-ASCII payload"), + Some(admitted) + ); +} + +#[asupersync::test] +async fn registered_sql_cannot_forge_a_retired_signal_identity() { + let harness = RegisteredGlobalDbHarness::open("github-stack-no-forged-retirement").await; + let record = signal("signal.github-stack-forged-retirement", 10); + let digest = "a".repeat(64); + // The registered database must enforce the same reserved-key policy as + // append_github_stack_signal, including null and escaped/duplicate keys. + // An INSERT-only forgery would reserve this identity permanently without + // a delivery recipient, bypassing the required settlement and age checks. + for forged in [ + format!(r#"{{"__tracedecay_retired_stack_sha256_v1":"{digest}"}}"#), + r#"{"__tracedecay_retired_stack_sha256_v1":null}"#.to_owned(), + format!(r#"{{"__trac\u0065decay_retired_stack_sha256_v1":"{digest}"}}"#), + r#"{"payload":true,"__tracedecay_retired_stack_sha256_v1":null}"#.to_owned(), + format!( + r#"{{"__tracedecay_retired_stack_sha256_v1":null,"__tracedecay_retired_stack_sha256_v1":"{digest}"}}"# + ), + ] { + let writer = harness + .registered + .writer_connection() + .expect("registered signal insert writer"); + let refusal = writer + .execute( + "INSERT INTO github_stack_delivery_signals + (project_id, signal_id, scope_digest, repository_id, + watermark_id, observed_at_micros, signal_json) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![ + record.project_id.as_str(), + record.signal_id.as_str(), + record.scope_digest.as_str(), + record.repository_id.as_str(), + record.watermark_id.as_str(), + record.observed_at_micros, + forged.as_str(), + ], + ) + .await + .expect_err("registered INSERT must refuse forged retirement") + .to_string(); + assert!( + refusal.contains("GitHub stack signal uses a reserved retirement marker"), + "{refusal}" + ); + assert!( + harness + .registered + .github_stack_signal(PROJECT_ID, &record.signal_id) + .await + .expect("refused forgery must not reserve a signal identity") + .is_none() + ); + } + + let mut valid = signal("signal.github-stack-after-forgery", 11); + valid.signal_json = r#"{"payload":{"not_retired":true}}"#.to_owned(); + harness + .registered + .append_github_stack_signal(valid.clone(), vec![RECIPIENT.to_owned()]) + .await + .expect("normal signal admission must remain available"); + assert_eq!( + harness + .registered + .github_stack_signal(PROJECT_ID, &valid.signal_id) + .await + .expect("read admitted live signal"), + Some(valid) + ); +} + #[asupersync::test] async fn pending_delivery_survives_restart_and_settles_only_after_acknowledgement() { let harness = RegisteredGlobalDbHarness::open("github-stack-delivery-restart").await; @@ -206,3 +354,676 @@ async fn settling_a_recipient_promotes_only_its_oldest_deferred_row() { "another recipient's backlog is untouched" ); } + +async fn stored_signal_payload_bytes(harness: &RegisteredGlobalDbHarness, signal_id: &str) -> i64 { + let snapshot = harness + .registered + .read_snapshot() + .await + .expect("read stored signal"); + let mut rows = snapshot + .query( + "SELECT length(signal_json) FROM github_stack_delivery_signals + WHERE project_id = ?1 AND signal_id = ?2", + params![PROJECT_ID, signal_id], + ) + .await + .expect("query stored payload size"); + rows.next() + .await + .expect("read stored payload size") + .expect("signal identity remains after retirement") + .get::(0) + .expect("decode stored payload bytes") +} + +#[asupersync::test] +async fn noncanonical_retirement_marker_cannot_reopen_an_expired_signal() { + let harness = RegisteredGlobalDbHarness::open("github-stack-retirement-marker-shape").await; + let record = signal("signal.github-stack-retirement-whitespace", 10); + harness + .registered + .append_github_stack_signal(record.clone(), vec![RECIPIENT.to_owned()]) + .await + .expect("append a real signal"); + harness + .registered + .acknowledge_github_stack_host_delivery(PROJECT_ID, &record.signal_id, RECIPIENT) + .await + .expect("settle the recipient"); + + // The installed SQL trigger accepts equivalent JSON with whitespace. + // The old byte-prefix decoder treated it as an unretired signal. + let digest = hex::encode(sha2::Sha256::digest(record.signal_json.as_bytes())); + let whitespace_marker = + format!("{{\n \"__tracedecay_retired_stack_sha256_v1\": \"{digest}\"\n}}"); + let writer = harness + .registered + .writer_connection() + .expect("registered retirement-marker writer"); + writer + .execute( + "UPDATE github_stack_delivery_signals SET signal_json = ?3 + WHERE project_id = ?1 AND signal_id = ?2", + params![ + PROJECT_ID, + record.signal_id.as_str(), + whitespace_marker.as_str() + ], + ) + .await + .expect("the terminal-only marker is allowed by SQLite"); + assert!( + harness + .registered + .github_stack_signal(PROJECT_ID, &record.signal_id) + .await + .expect("a retired marker must never expand as a signal") + .is_none(), + "whitespace in a legitimate retirement marker must not expose it" + ); + assert!(matches!( + harness + .registered + .append_github_stack_signal(record.clone(), vec![RECIPIENT.to_owned()]) + .await + .expect("the identical replay remains recognized"), + GitHubStackSignalAppendOutcomeV1::Replayed { .. } + )); + let mut conflict = record.clone(); + conflict.signal_json = r#"{"different":true}"#.to_owned(); + assert!( + harness + .registered + .append_github_stack_signal(conflict, vec![RECIPIENT.to_owned()]) + .await + .expect_err("changed bytes must conflict with retained fingerprint") + .contains("identity conflict") + ); + + // SQLite decodes Unicode-escaped object keys as the same retirement + // field, even though its literal byte spelling differs from the key. + let escaped = signal("signal.github-stack-escaped-retirement-key", 12); + harness + .registered + .append_github_stack_signal(escaped.clone(), vec![RECIPIENT.to_owned()]) + .await + .expect("append the Unicode-key fixture"); + harness + .registered + .acknowledge_github_stack_host_delivery(PROJECT_ID, &escaped.signal_id, RECIPIENT) + .await + .expect("settle Unicode-key fixture"); + let escaped_marker = format!(r#"{{"__trac\u0065decay_retired_stack_sha256_v1":"{digest}"}}"#); + let writer = harness + .registered + .writer_connection() + .expect("registered Unicode-marker writer"); + writer + .execute( + "UPDATE github_stack_delivery_signals SET signal_json = ?3 + WHERE project_id = ?1 AND signal_id = ?2", + params![ + PROJECT_ID, + escaped.signal_id.as_str(), + escaped_marker.as_str() + ], + ) + .await + .expect("SQLite accepts the escaped retirement key"); + assert!( + harness + .registered + .github_stack_signal(PROJECT_ID, &escaped.signal_id) + .await + .expect("escaped-key marker must not expand") + .is_none(), + "escaped JSON keys must retain the same retired identity" + ); + + let malformed = signal("signal.github-stack-malformed-marker", 11); + harness + .registered + .append_github_stack_signal(malformed.clone(), vec![RECIPIENT.to_owned()]) + .await + .expect("append a second signal"); + harness + .registered + .acknowledge_github_stack_host_delivery(PROJECT_ID, &malformed.signal_id, RECIPIENT) + .await + .expect("settle second recipient"); + // SQL and the reader must agree: a retirement marker has one member. + // A second field or a duplicate key is ambiguous, even if its hash is valid. + for ambiguous_marker in [ + format!(r#"{{"__tracedecay_retired_stack_sha256_v1":"{digest}","other":"unexpected"}}"#), + format!( + r#"{{"__tracedecay_retired_stack_sha256_v1":"{digest}","__tracedecay_retired_stack_sha256_v1":"{digest}"}}"# + ), + ] { + let rejection = harness + .registered + .writer_connection() + .expect("registered ambiguous-marker writer") + .execute( + "UPDATE github_stack_delivery_signals SET signal_json = ?3 + WHERE project_id = ?1 AND signal_id = ?2", + params![ + PROJECT_ID, + malformed.signal_id.as_str(), + ambiguous_marker.as_str() + ], + ) + .await + .expect_err("retirement must reject extra and duplicated JSON members") + .to_string(); + assert!( + rejection.contains("GitHub stack delivery signals are immutable"), + "{rejection}" + ); + assert_eq!( + harness + .registered + .github_stack_signal(PROJECT_ID, &malformed.signal_id) + .await + .expect("refused ambiguous retirement keeps the original"), + Some(malformed.clone()) + ); + } + let invalid_marker = format!( + r#"{{"__tracedecay_retired_stack_sha256_v1":"{}"}}"#, + "g".repeat(64) + ); + let rejection = harness + .registered + .writer_connection() + .expect("registered malformed-marker writer") + .execute( + "UPDATE github_stack_delivery_signals SET signal_json = ?3 + WHERE project_id = ?1 AND signal_id = ?2", + params![ + PROJECT_ID, + malformed.signal_id.as_str(), + invalid_marker.as_str() + ], + ) + .await + .expect_err("the SQL guard must reject a non-hexadecimal fingerprint") + .to_string(); + assert!( + rejection.contains("GitHub stack delivery signals are immutable"), + "{rejection}" + ); + assert_eq!( + harness + .registered + .github_stack_signal(PROJECT_ID, &malformed.signal_id) + .await + .expect("refused invalid marker cannot alter the original"), + Some(malformed) + ); +} + +#[asupersync::test] +async fn terminal_compaction_preserves_pending_peers_and_refuses_replay_collision() { + let harness = RegisteredGlobalDbHarness::open("github-stack-terminal-compaction").await; + let mut record = signal("signal.github-stack-old", 10); + record.signal_json = format!(r#"{{"large":"{}"}}"#, "x".repeat(80_000)); + let recipients = vec![RECIPIENT.to_owned(), "actor.other".to_owned()]; + harness + .registered + .append_github_stack_signal(record.clone(), recipients.clone()) + .await + .expect("durably append old signal"); + let original_size = stored_signal_payload_bytes(&harness, &record.signal_id).await; + + harness + .registered + .acknowledge_github_stack_host_delivery(PROJECT_ID, &record.signal_id, RECIPIENT) + .await + .expect("settle first recipient"); + assert_eq!( + harness + .registered + .retire_github_stack_terminal_payloads_before(11) + .await + .expect("attempt cleanup with pending peer") + .retired, + 0, + "a pending peer retains its exact payload" + ); + assert_eq!( + harness + .registered + .github_stack_signal(PROJECT_ID, &record.signal_id) + .await + .expect("read live signal"), + Some(record.clone()) + ); + harness + .registered + .record_github_stack_authorization_loss(PROJECT_ID, &record.signal_id, "actor.other") + .await + .expect("record permanently lost authorization"); + assert_eq!( + harness + .registered + .retire_github_stack_terminal_payloads_before(10) + .await + .expect("respect inspection window") + .retired, + 0, + "the cutoff excludes the issue timestamp itself" + ); + assert_eq!( + harness + .registered + .retire_github_stack_terminal_payloads_before(11) + .await + .expect("retire fully terminal signal") + .retired, + 1 + ); + assert!( + stored_signal_payload_bytes(&harness, &record.signal_id).await < original_size / 100, + "retire the large transport payload rather than moving it to another log" + ); + assert!( + harness + .registered + .github_stack_signal(PROJECT_ID, &record.signal_id) + .await + .expect("old expansion must expire") + .is_none(), + "a compact fingerprint must never be returned as expandable signal JSON" + ); + assert_eq!( + state(&harness, &record.signal_id, RECIPIENT).await, + None, + "retired recipient bookkeeping is no longer part of the active index" + ); + assert!(matches!( + harness + .registered + .append_github_stack_signal(record.clone(), recipients.clone()) + .await + .expect("identical replay remains recognized"), + GitHubStackSignalAppendOutcomeV1::Replayed { .. } + )); + let mut conflict = record.clone(); + conflict.signal_json = r#"{"different":true}"#.to_owned(); + assert!( + harness + .registered + .append_github_stack_signal(conflict, recipients.clone()) + .await + .is_err(), + "retired signal identities reject mismatched payloads" + ); + + let restarted = harness.restart().await; + assert!(matches!( + restarted + .registered + .append_github_stack_signal(record, recipients) + .await + .expect("the small replay fingerprint survives restart"), + GitHubStackSignalAppendOutcomeV1::Replayed { .. } + )); + assert_eq!( + restarted + .registered + .pending_github_stack_deliveries(PROJECT_ID, SCOPE_DIGEST, None) + .await + .expect("retired data cannot redeliver"), + vec![] + ); +} + +#[asupersync::test] +async fn compaction_is_bounded_and_raw_sql_cannot_delete_unsettled_delivery() { + let harness = RegisteredGlobalDbHarness::open("github-stack-bounded-retirement").await; + let count = crate::MAX_GITHUB_STACK_RETIRE_BATCH_V1 + 3; + for index in 0..count { + let record = signal( + &format!("signal.github-stack-batch.{index}"), + index as i64 + 1, + ); + harness + .registered + .append_github_stack_signal(record.clone(), vec![RECIPIENT.to_owned()]) + .await + .expect("append batch signal"); + if index + 1 != count { + harness + .registered + .acknowledge_github_stack_host_delivery(PROJECT_ID, &record.signal_id, RECIPIENT) + .await + .expect("settle eligible batch signal"); + } + } + let live = format!("signal.github-stack-batch.{}", count - 1); + let pending_deletion = harness + .registered + .writer_connection() + .expect("registered test writer") + .execute( + "DELETE FROM github_stack_delivery_recipients + WHERE project_id = ?1 AND signal_id = ?2", + params![PROJECT_ID, live.as_str()], + ) + .await; + assert!( + pending_deletion + .expect_err("the SQL guard protects active pending state") + .to_string() + .contains("GitHub stack delivery recipient is not retired"), + "the trigger must reject deletion of an unsettled recipient" + ); + + let first = harness + .registered + .retire_github_stack_terminal_payloads_before(1000) + .await + .expect("bounded first page") + .retired; + assert_eq!(first, crate::MAX_GITHUB_STACK_RETIRE_BATCH_V1); + let second = harness + .registered + .retire_github_stack_terminal_payloads_before(1000) + .await + .expect("next page") + .retired; + assert_eq!(second, 2, "pending signal is ineligible across pages"); + assert_eq!( + state(&harness, &live, RECIPIENT).await, + Some(GitHubStackDeliveryStateV1::Pending) + ); + assert_eq!( + harness + .registered + .retire_github_stack_terminal_payloads_before(1000) + .await + .expect("fully drained next page") + .retired, + 0 + ); +} + +#[asupersync::test] +async fn retention_frontier_bounds_writer_work_and_resumes_past_deferred_backlog() { + const BLOCKED: usize = 4_096; + let harness = RegisteredGlobalDbHarness::open("github-stack-retirement-frontier").await; + let writer = harness + .registered + .writer_connection() + .expect("registered retention fixture writer"); + // One recipient's first 256 signals are pending and all later ones are + // deferred. The terminal signal is beyond this old, ineligible prefix. + writer + .execute( + "WITH RECURSIVE seq(n) AS ( + SELECT 0 UNION ALL SELECT n + 1 FROM seq WHERE n < 4095 + ) + INSERT INTO github_stack_delivery_signals + (project_id, signal_id, scope_digest, repository_id, + watermark_id, observed_at_micros, signal_json) + SELECT ?1, printf('signal.github-stack-prefix.%05d', n), ?2, + 'repository.github-stack-delivery', 'watermark.github-stack-delivery', + n + 1, json_object('n', n) FROM seq", + params![PROJECT_ID, SCOPE_DIGEST], + ) + .await + .expect("seed a large genuine signal table"); + writer + .execute( + "INSERT INTO github_stack_delivery_recipients + (project_id, signal_id, recipient, state) + SELECT project_id, signal_id, ?1, + CASE WHEN observed_at_micros <= 256 THEN 'pending' + ELSE 'deferred' END + FROM github_stack_delivery_signals WHERE project_id = ?2", + params![RECIPIENT, PROJECT_ID], + ) + .await + .expect("seed a saturated recipient backlog"); + let late = signal("signal.github-stack-eligible-tail", 10_000); + // The old recipient is saturated. Give the eligible tail an independently + // admitted recipient so it can actually settle before retirement. + let tail_recipient = "actor.terminal-tail"; + harness + .registered + .append_github_stack_signal(late.clone(), vec![tail_recipient.to_owned()]) + .await + .expect("append terminal tail beyond the backlog"); + harness + .registered + .acknowledge_github_stack_host_delivery(PROJECT_ID, &late.signal_id, tail_recipient) + .await + .expect("settle eligible tail"); + + let before = writer_telemetry(&harness.registered); + let first = harness + .registered + .retire_github_stack_terminal_payloads_before(10_001) + .await + .expect("scan one bounded pending window"); + let after = writer_telemetry(&harness.registered); + assert_eq!( + after.transactions.commands - before.transactions.commands, + 1 + ); + let steps = after.sqlite_vm.vm_steps - before.sqlite_vm.vm_steps; + assert_eq!(first.retired, 0); + assert_eq!( + first.inspected, + crate::stack_delivery::MAX_GITHUB_STACK_RETIRE_SCAN_V1 + ); + assert!( + !first.exhausted, + "the deferred prefix cannot starve the tail" + ); + assert!( + (1..30_000).contains(&steps), + "one idle writer slice must not examine the 4096-row deferred prefix: {steps} VM steps" + ); + assert_eq!( + harness + .registered + .github_stack_signal(PROJECT_ID, &late.signal_id) + .await + .unwrap(), + Some(late.clone()), + "the first bounded window must not skip ahead and compact the tail" + ); + + // The scan cursor is durable, not an in-process iterator. A restart must + // resume past exactly the first 64 inspected rows and reach the tail. + let restarted = harness.restart().await; + let mut retired = 0; + for _ in 0..(BLOCKED / crate::stack_delivery::MAX_GITHUB_STACK_RETIRE_SCAN_V1 + 3) { + let page = restarted + .registered + .retire_github_stack_terminal_payloads_before(10_001) + .await + .expect("continue bounded retirement after restart"); + assert!(page.inspected <= crate::stack_delivery::MAX_GITHUB_STACK_RETIRE_SCAN_V1); + assert!(page.retired <= crate::MAX_GITHUB_STACK_RETIRE_BATCH_V1); + retired += page.retired; + if retired != 0 { + break; + } + } + assert_eq!(retired, 1, "the terminal tail must eventually be reached"); + assert!( + restarted + .registered + .github_stack_signal(PROJECT_ID, &late.signal_id) + .await + .unwrap() + .is_none() + ); + + // On wrap, earlier signals that have since settled are reconsidered. + let formerly_pending = "signal.github-stack-prefix.00000"; + restarted + .registered + .record_github_stack_authorization_loss(PROJECT_ID, formerly_pending, RECIPIENT) + .await + .expect("permanently settle an earlier signal"); + let mut saw_earlier = false; + for _ in 0..(BLOCKED / crate::stack_delivery::MAX_GITHUB_STACK_RETIRE_SCAN_V1 + 3) { + let page = restarted + .registered + .retire_github_stack_terminal_payloads_before(10_001) + .await + .expect("revisit old newly settled signals"); + assert!(page.inspected <= crate::stack_delivery::MAX_GITHUB_STACK_RETIRE_SCAN_V1); + if restarted + .registered + .github_stack_signal(PROJECT_ID, formerly_pending) + .await + .unwrap() + .is_none() + { + saw_earlier = true; + break; + } + } + assert!( + saw_earlier, + "wrap must not abandon later-settling old signals" + ); +} + +#[asupersync::test] +async fn terminal_retirement_cannot_be_accelerated_past_the_issued_window() { + let harness = RegisteredGlobalDbHarness::open("github-stack-terminal-freshness").await; + let mut fresh = signal( + "signal.github-stack-fresh", + tracedecay_contracts::now_micros().0, + ); + fresh.signal_json = r#"{"recent":true}"#.to_owned(); + harness + .registered + .append_github_stack_signal(fresh.clone(), vec![RECIPIENT.to_owned()]) + .await + .expect("append current signal"); + harness + .registered + .acknowledge_github_stack_host_delivery(PROJECT_ID, &fresh.signal_id, RECIPIENT) + .await + .expect("settle current signal"); + assert_eq!( + harness + .registered + .retire_github_stack_terminal_payloads_before(i64::MAX) + .await + .expect("future cutoff is conservatively clamped") + .retired, + 0 + ); + // A direct registered writer must not retire fresh transport evidence, + // even with a syntactically valid marker for its real original bytes. + let digest = hex::encode(sha2::Sha256::digest(fresh.signal_json.as_bytes())); + let marker = format!(r#"{{"__tracedecay_retired_stack_sha256_v1":"{digest}"}}"#); + let writer = harness + .registered + .writer_connection() + .expect("registered fresh-retirement writer"); + let refusal = writer + .execute( + "UPDATE github_stack_delivery_signals SET signal_json = ?3 + WHERE project_id = ?1 AND signal_id = ?2", + params![PROJECT_ID, fresh.signal_id.as_str(), marker.as_str()], + ) + .await + .expect_err("registered SQL must preserve the inspection window") + .to_string(); + assert!( + refusal.contains("GitHub stack delivery signals are immutable"), + "{refusal}" + ); + assert_eq!( + harness + .registered + .github_stack_signal(PROJECT_ID, &fresh.signal_id) + .await + .expect("recent expanded signal remains inspectable"), + Some(fresh) + ); +} + +#[asupersync::test] +async fn failed_terminal_recipient_retirement_rolls_back_payload_compaction() { + let harness = RegisteredGlobalDbHarness::open("github-stack-retire-rollback").await; + let record = signal("signal.github-stack-rollback", 10); + harness + .registered + .append_github_stack_signal(record.clone(), vec![RECIPIENT.to_owned()]) + .await + .expect("persist signal"); + harness + .registered + .acknowledge_github_stack_host_delivery(PROJECT_ID, &record.signal_id, RECIPIENT) + .await + .expect("settle signal before retention"); + + harness + .registered + .writer_connection() + .expect("registered test writer") + .execute_batch( + "CREATE TRIGGER github_stack_retirement_failure + BEFORE DELETE ON github_stack_delivery_recipients + BEGIN SELECT RAISE(ABORT, 'simulate interrupted retirement'); END;", + ) + .await + .expect("inject failure after payload update"); + let problem = harness + .registered + .retire_github_stack_terminal_payloads_before(11) + .await + .expect_err("recipient deletion must fail inside the transaction"); + assert!( + problem.contains("simulate interrupted retirement"), + "{problem}" + ); + assert_eq!( + harness + .registered + .github_stack_signal(PROJECT_ID, &record.signal_id) + .await + .expect("failed retirement preserves the full payload"), + Some(record.clone()) + ); + assert_eq!( + state(&harness, &record.signal_id, RECIPIENT).await, + Some(GitHubStackDeliveryStateV1::Settled) + ); + + harness + .registered + .writer_connection() + .expect("registered test writer") + .execute_batch("DROP TRIGGER github_stack_retirement_failure") + .await + .expect("remove failure injection"); + assert_eq!( + harness + .registered + .retire_github_stack_terminal_payloads_before(11) + .await + .expect("retry after failure") + .retired, + 1 + ); + let restarted = harness.restart().await; + assert!( + restarted + .registered + .github_stack_signal(PROJECT_ID, &record.signal_id) + .await + .expect("retirement persisted across restart") + .is_none() + ); +} diff --git a/crates/tracedecay-maintenance/src/retention/registered_store.rs b/crates/tracedecay-maintenance/src/retention/registered_store.rs index 52236a295c4..67522cc5f1e 100644 --- a/crates/tracedecay-maintenance/src/retention/registered_store.rs +++ b/crates/tracedecay-maintenance/src/retention/registered_store.rs @@ -5,7 +5,10 @@ use tracedecay_global_db::observation::retention::{ ObservationRetentionConfig, ObservationRetentionReport, RetentionMode, }; use tracedecay_global_db::observation::{CanonicalBodyGcReport, MAX_CANONICAL_BODY_GC_ROWS}; -use tracedecay_global_db::{ObservabilityRetentionReceiptV1, RegisteredGlobalDb}; +use tracedecay_global_db::{ + GITHUB_STACK_TERMINAL_RETENTION_DAYS_V1, GitHubStackTerminalRetirementPageV1, + ObservabilityRetentionReceiptV1, RegisteredGlobalDb, +}; use tracedecay_lcm::retention::{ MAX_REPLAY_METADATA_COMPACTION_ROWS, ReplayMetadataCompactionReport, }; @@ -20,6 +23,7 @@ pub enum RegisteredStoreRetentionErrorV1 { PayloadGc { diagnostic: String }, CanonicalBodyGc { diagnostic: String }, ReplayMetadataCompaction { diagnostic: String }, + GitHubStackTerminal { diagnostic: String }, } impl RegisteredStoreRetentionErrorV1 { @@ -29,7 +33,8 @@ impl RegisteredStoreRetentionErrorV1 { Self::Observability { diagnostic } | Self::PayloadGc { diagnostic } | Self::CanonicalBodyGc { diagnostic } - | Self::ReplayMetadataCompaction { diagnostic } => diagnostic, + | Self::ReplayMetadataCompaction { diagnostic } + | Self::GitHubStackTerminal { diagnostic } => diagnostic, } } } @@ -48,6 +53,12 @@ pub struct RegisteredStoreRetentionReportV1 { pub canonical_body_gc: Result, /// Lossless repair of old provider replay strings duplicated by raw bodies. pub replay_metadata: Result, + /// Completed GitHub stack delivery payloads past their fixed inspection + /// window. Their window is a product invariant, not an operator policy, so + /// this runs for every store whenever store retention runs, independent of + /// automation jobs. + pub github_stack_terminal: + Result, } impl RegisteredStoreRetentionReportV1 { @@ -71,6 +82,7 @@ impl RegisteredStoreRetentionReportV1 { && payload_gc_succeeded && self.canonical_body_gc.is_ok() && self.replay_metadata.is_ok() + && self.github_stack_terminal.is_ok() } } @@ -117,6 +129,7 @@ pub async fn run_registered_store_retention( let payload_gc = run_payload_gc(database, now).await; let canonical_body_gc = run_canonical_body_gc(database).await; let replay_metadata = run_replay_metadata_compaction(database).await; + let github_stack_terminal = run_github_stack_terminal_retention(database, now).await; RegisteredStoreRetentionReportV1 { session_lcm, observations, @@ -124,7 +137,48 @@ pub async fn run_registered_store_retention( payload_gc, canonical_body_gc, replay_metadata, + github_stack_terminal, + } +} + +/// Retire completed stack delivery payloads from one registered store through +/// one page per admitted store unit. A page examines and compacts a bounded +/// number of payloads and commits its durable scan frontier atomically, so a +/// long pending prefix +/// is neither rescanned nor allowed to starve later terminal rows. +/// +/// This is fixed product housekeeping, so the daemon also runs it alone when +/// every optional age, deletion and compaction policy is disabled. +pub async fn run_github_stack_terminal_retention( + database: &RegisteredGlobalDb, + now: i64, +) -> Result { + let cutoff_micros = now + .saturating_sub(GITHUB_STACK_TERMINAL_RETENTION_DAYS_V1 * 86_400) + .saturating_mul(1_000_000); + // The daemon releases admission after this page. The store commits its + // frontier with the mutations; the shared continuation resumes that work + // under a fresh admission rather than looping while holding the gate. + let page = database + .retire_github_stack_terminal_payloads_before(cutoff_micros) + .await + .map_err(|diagnostic| { + tracing::warn!( + event = "retention_degraded", + pass = "github_stack_terminal", + failure = %diagnostic, + ); + RegisteredStoreRetentionErrorV1::GitHubStackTerminal { diagnostic } + })?; + if page.retired > 0 { + tracing::info!( + event = "retention_prune", + table = "github_stack_delivery_signals", + signals = page.retired, + window_days = GITHUB_STACK_TERMINAL_RETENTION_DAYS_V1, + ); } + Ok(page) } /// Resume bounded, lossless storage work without repeating the raw-drop policy. @@ -428,4 +482,94 @@ mod tests { ); assert!(!report.succeeded()); } + + /// Stack deliveries live in each registered session store, so the + /// store-wide pass must retire them there with every operator retention + /// policy disabled. Pending history and a still-open window stay intact. + #[asupersync::test] + async fn registered_retention_retires_only_expired_settled_stack_payloads() { + const NOW: i64 = 40 * 86_400; + const PROJECT: &str = "project.stack-retention"; + const RECIPIENT: &str = "actor.stack-retention"; + let harness = tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness::open( + "maintenance-registered-stack-terminal-retention", + ) + .await; + let record = |signal_id: &str, observed_at_micros: i64| { + tracedecay_global_db::GitHubStackSignalRecordV1 { + project_id: PROJECT.to_owned(), + signal_id: signal_id.to_owned(), + scope_digest: "sha256:stack-retention-scope".to_owned(), + repository_id: "repository.stack-retention".to_owned(), + watermark_id: "watermark.stack-retention".to_owned(), + observed_at_micros, + signal_json: format!(r#"{{"payload":"{signal_id}"}}"#), + } + }; + let in_window = (NOW - 86_400) * 1_000_000; + for (signal_id, observed_at, settle) in [ + ("signal.expired-settled", 10, true), + ("signal.expired-pending", 11, false), + ("signal.in-window-settled", in_window, true), + ] { + harness + .registered + .append_github_stack_signal( + record(signal_id, observed_at), + vec![RECIPIENT.to_owned()], + ) + .await + .expect("persist stack signal"); + if settle { + harness + .registered + .acknowledge_github_stack_host_delivery(PROJECT, signal_id, RECIPIENT) + .await + .expect("settle stack signal"); + } + } + let session_lcm = LcmRetentionConfig { + enabled: false, + ..LcmRetentionConfig::default() + }; + let observations = ObservationRetentionConfig { + enabled: false, + ..ObservationRetentionConfig::default() + }; + + let report = + run_registered_store_retention(&harness.registered, &session_lcm, &observations, NOW) + .await; + + assert_eq!( + report.github_stack_terminal, + Ok(GitHubStackTerminalRetirementPageV1 { + retired: 1, + inspected: 2, + exhausted: true, + }) + ); + assert!(report.succeeded()); + let lookup = |signal_id: &'static str| { + let database = harness.registered.clone(); + async move { + database + .github_stack_signal(PROJECT, signal_id) + .await + .expect("read stack signal") + } + }; + assert!( + lookup("signal.expired-settled").await.is_none(), + "an expired settled payload is compacted" + ); + assert!( + lookup("signal.expired-pending").await.is_some(), + "an undelivered signal is never retired by age" + ); + assert!( + lookup("signal.in-window-settled").await.is_some(), + "a settled signal inside the inspection window is kept" + ); + } } diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 9cc365460f0..cc9d41729ac 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -485,6 +485,9 @@ async fn run_registered_store_retention( session_storage_outcome .combine(canonical_body_outcome) .combine(replay_metadata_outcome) + .combine(record_terminal_housekeeping_outcome( + &report.github_stack_terminal, + )) } else { MaintenanceTickOutcome::Retry }; @@ -500,6 +503,43 @@ async fn run_registered_store_retention( outcome } +async fn run_registered_terminal_housekeeping( + database: &tracedecay_global_db::RegisteredGlobalDb, +) -> MaintenanceTickOutcome { + let now = match now_secs_i64() { + Ok(now) => now, + Err(failure) => { + log_daemon_event( + "retention_degraded", + &[ + ("pass", "terminal_housekeeping".to_owned()), + ("failure", failure.to_owned()), + ], + ); + return MaintenanceTickOutcome::Retry; + } + }; + let report = + tracedecay_maintenance::retention::registered_store::run_github_stack_terminal_retention( + database, now, + ) + .await; + record_terminal_housekeeping_outcome(&report) +} + +fn record_terminal_housekeeping_outcome( + report: &Result< + tracedecay_global_db::GitHubStackTerminalRetirementPageV1, + tracedecay_maintenance::retention::registered_store::RegisteredStoreRetentionErrorV1, + >, +) -> MaintenanceTickOutcome { + match report { + Ok(report) if report.exhausted => MaintenanceTickOutcome::Complete, + Ok(_) => MaintenanceTickOutcome::Continue(MaintenanceContinuation::StoreRetention), + Err(_) => MaintenanceTickOutcome::Retry, + } +} + fn record_session_lcm_outcome( outcome: &Option>, ) -> MaintenanceTickOutcome { @@ -579,9 +619,11 @@ async fn run_registered_storage_continuation( database, ) .await; + let terminal = run_registered_terminal_housekeeping(database).await; record_session_lcm_outcome(&session_lcm) .combine(record_canonical_body_gc_outcome(&canonical_body_gc)) .combine(record_replay_metadata_outcome(&replay_metadata)) + .combine(terminal) } fn record_canonical_body_gc_outcome( @@ -896,10 +938,11 @@ impl MaintenanceStoreWork { fn writer_scope(&self) -> WriterScope { match self { - // Session-wide administration has no independently proven canonical - // store gate. Keep its global probe nonqueued rather than blocking - // unrelated project opens behind a potentially long session pass. - Self::Session(_) => WriterScope::Daemon, + // Registered database paths are canonical and sit directly below + // their owner store root, the same identity used by logical writers. + Self::Session(path) => path + .parent() + .map_or(WriterScope::Daemon, store_maintenance_writer_scope), Self::Graph { data_root, .. } => store_maintenance_writer_scope(data_root), Self::Registered { store, .. } => store_maintenance_writer_scope(&store.data_root), } @@ -981,39 +1024,38 @@ impl MaintenanceCoordinator { tracing::trace_span!("daemon.maintenance.resident_memory_sampler"), ), ); - if !retention_maintenance_enabled(&retention) { - return coordinator; - } + let interval = Duration::from_secs(retention.interval_hours.max(1).saturating_mul(3_600)); // A sealed code generation supersedes its predecessor, whose sealed // artifact, read bundle, and segments only the retention tick // reclaims. Pull that tick forward on each publication instead of // letting a day of publications pile up behind the daily cadence. - let publication_owner = coordinator.clone(); - let mut publications = code_index_schedulers.subscribe_generation_publications(); - // Detached: it ends with the coordinator's cancellation or the - // publication channel. - drop(spawn_maintenance_task( - "publication_due_requests", - tracing::Instrument::instrument( - async move { - loop { - tokio::select! { - biased; - () = publication_owner.cancellation.cancelled() => break, - received = publications.recv() => match received { - Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { - publication_owner.request_due(); - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => break, - }, + if optional_retention_enabled(&retention) { + let publication_owner = coordinator.clone(); + let mut publications = code_index_schedulers.subscribe_generation_publications(); + // Detached: it ends with the coordinator's cancellation or the + // publication channel. + drop(spawn_maintenance_task( + "publication_due_requests", + tracing::Instrument::instrument( + async move { + loop { + tokio::select! { + biased; + () = publication_owner.cancellation.cancelled() => break, + received = publications.recv() => match received { + Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { + publication_owner.request_due(); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + }, + } } - } - }, - tracing::trace_span!("daemon.maintenance.publication_due_requests"), - ), - )); + }, + tracing::trace_span!("daemon.maintenance.publication_due_requests"), + ), + )); + } let task_owner = coordinator.clone(); - let interval = Duration::from_secs(retention.interval_hours.max(1).saturating_mul(3_600)); *coordinator.task.lock().await = spawn_maintenance_task( "retention_tick", tracing::Instrument::instrument( @@ -1179,7 +1221,7 @@ impl MaintenanceCoordinator { clippy::too_many_lines, reason = "A maintenance tick is one ordered pass over session, graph, and retention owners." )] - async fn run_tick( + pub(super) async fn run_tick( &self, profile_root: &Path, profile_database: &tracedecay_global_db::RegisteredGlobalDbLeaseV1, @@ -1199,6 +1241,7 @@ impl MaintenanceCoordinator { ); return MaintenanceTickOutcome::Retry; }; + let optional = optional_retention_enabled(retention); administration .store_telemetry_sampling() .begin_retention_tick_log_window(); @@ -1208,17 +1251,21 @@ impl MaintenanceCoordinator { .iter() .map(|database| database.db_path().to_path_buf()) .collect::>(); - let project_graphs = administration - .mounted_project_graphs() - .await - .iter() - .map(|graph| MountedGraphListing { - key: graph_work_key(graph), - database_path: graph.db().database_path().to_path_buf(), - project_root: graph.project_root().to_path_buf(), - data_root: graph.store_layout().data_root.clone(), - }) - .collect::>(); + let project_graphs = if optional { + administration + .mounted_project_graphs() + .await + .iter() + .map(|graph| MountedGraphListing { + key: graph_work_key(graph), + database_path: graph.db().database_path().to_path_buf(), + project_root: graph.project_root().to_path_buf(), + data_root: graph.store_layout().data_root.clone(), + }) + .collect::>() + } else { + Vec::new() + }; #[cfg(any(test, feature = "test-helpers"))] { let _ = crate::test_support::hold_after_maintenance_listing_for_test().await; @@ -1228,22 +1275,25 @@ impl MaintenanceCoordinator { let Some(tick_cpu) = background_cpu.try_acquire() else { return MaintenanceTickOutcome::Retry; }; - let (registered_projects, registry_listed) = - match registered_project_stores(profile_root, profile_database).await { - Ok(stores) => (stores, true), - Err(error) => { - log_daemon_event( - "retention_degraded", - &[ - ("pass", "code_generations".to_owned()), - ("failure", "project_registry_unavailable".to_owned()), - ("error", error.to_string()), - ], - ); - (Vec::new(), false) - } - }; - if continuation.is_none() && !self.cancellation.is_cancelled() { + let (registered_projects, registry_listed) = match if optional { + registered_project_stores(profile_root, profile_database).await + } else { + Ok(Vec::new()) + } { + Ok(stores) => (stores, true), + Err(error) => { + log_daemon_event( + "retention_degraded", + &[ + ("pass", "code_generations".to_owned()), + ("failure", "project_registry_unavailable".to_owned()), + ("error", error.to_string()), + ], + ); + (Vec::new(), false) + } + }; + if optional && continuation.is_none() && !self.cancellation.is_cancelled() { let handle_roots = std::iter::once(profile_root) .chain( registered_projects @@ -1296,7 +1346,20 @@ impl MaintenanceCoordinator { } work.sort_by(|left, right| left.0.cmp(&right.0)); let keys = work.iter().map(|(key, _)| key.clone()).collect::>(); - self.store_continuations.lock().await.retain_stores(&keys); + { + let mut continuations = self.store_continuations.lock().await; + continuations.retain_stores(&keys); + if !optional && continuation.is_none() { + // A fixed-only cycle visits every mounted shard through the + // same bounded window. Unvisited stores keep the short cadence. + for key in &keys { + continuations.record( + key, + MaintenanceTickOutcome::Continue(MaintenanceContinuation::StoreRetention), + ); + } + } + } self.writer_jobs.retain( &work .iter() @@ -1338,7 +1401,8 @@ impl MaintenanceCoordinator { maintenance_observations.retain_project_maintenance_state(&active_maintenance_projects); // A capacity retirement releasing stores skips this pass: sampling // leases every sampled store until it ends. - let telemetry_sampling = if continuation.is_none() + let telemetry_sampling = if optional + && continuation.is_none() && let Ok(_clients) = administration.maintenance_store_clients().try_lock() { maintenance_observations @@ -1351,7 +1415,8 @@ impl MaintenanceCoordinator { } else { StoreTelemetrySamplingOutcome::default() }; - if continuation.is_none() + if optional + && continuation.is_none() && let Some(registry) = administration.mounted_session_runtime_registry().await { record_lost_store_files(&maintenance_observations, ®istry); @@ -1410,7 +1475,9 @@ impl MaintenanceCoordinator { else { return MaintenanceTickOutcome::Complete; }; - if continuation.is_some() { + if !optional { + run_registered_terminal_housekeeping(&database).await + } else if continuation.is_some() { run_registered_storage_continuation(&database, retention).await } else { run_registered_store_retention(&database, retention).await @@ -1497,6 +1564,9 @@ impl MaintenanceCoordinator { if !registry_listed { outcome = MaintenanceTickOutcome::Retry; } + if !optional { + return outcome; + } let Some(_background_cpu) = background_cpu.try_acquire() else { return outcome.combine(MaintenanceTickOutcome::Retry); @@ -1816,7 +1886,11 @@ impl ResidentMemoryLogStateV1 { } } -pub(super) fn retention_maintenance_enabled( +/// Whether an operator-controlled age, deletion or compaction policy is on. +/// Soft budgets are Doctor findings and never count. When this is false the +/// daemon still runs the fixed terminal-delivery pass, but none of the full +/// tick's optional work. +pub(super) fn optional_retention_enabled( retention: &tracedecay_configuration::RetentionConfig, ) -> bool { retention.session_lcm.enabled @@ -1843,6 +1917,179 @@ mod tests { use tracedecay_runtime_core::cancellation::CancellationToken; use tracedecay_runtime_core::native_runtime; + #[test] + fn terminal_storage_failure_is_retry_and_rolls_back_progress() { + native_runtime::block_on(async { + let harness = tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness::open( + "terminal-rollback", + ) + .await; + harness + .registered + .append_github_stack_signal( + tracedecay_global_db::GitHubStackSignalRecordV1 { + project_id: "project.rollback".to_owned(), + signal_id: "signal.rollback".to_owned(), + scope_digest: "scope".to_owned(), + repository_id: "repo".to_owned(), + watermark_id: "watermark".to_owned(), + observed_at_micros: 100, + signal_json: "{}".to_owned(), + }, + vec!["actor.rollback".to_owned()], + ) + .await + .expect("terminal signal"); + harness + .registered + .acknowledge_github_stack_host_delivery( + "project.rollback", + "signal.rollback", + "actor.rollback", + ) + .await + .expect("settled"); + harness + .registered + .writer_connection() + .unwrap() + .execute_batch( + "CREATE TRIGGER refuse_terminal_delete + BEFORE DELETE ON github_stack_delivery_recipients + BEGIN SELECT RAISE(ABORT, 'retirement refused'); END;", + ) + .await + .unwrap(); + assert_eq!( + super::run_registered_terminal_housekeeping(&harness.registered).await, + MaintenanceTickOutcome::Retry + ); + assert!( + harness + .registered + .github_stack_signal("project.rollback", "signal.rollback") + .await + .unwrap() + .is_some(), + "payload update rolled back with the failed recipient deletion" + ); + let mut rows = harness + .registered + .read_connection() + .query( + "SELECT COUNT(*) FROM github_stack_delivery_retirement_frontier", + (), + ) + .await + .unwrap(); + assert_eq!( + rows.next().await.unwrap().unwrap().get::(0).unwrap(), + 0 + ); + drop(rows); + harness + .registered + .writer_connection() + .unwrap() + .execute_batch("DROP TRIGGER refuse_terminal_delete") + .await + .unwrap(); + assert_eq!( + super::run_registered_terminal_housekeeping(&harness.registered).await, + MaintenanceTickOutcome::Complete + ); + }) + .expect("native runtime"); + } + + #[test] + fn terminal_progress_continues_without_repeating_optional_age_work() { + native_runtime::block_on(async { + let harness = tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness::open( + "terminal-age-continuation", + ) + .await; + for ordinal in 0..1041 { + harness + .registered + .append_github_stack_signal( + tracedecay_global_db::GitHubStackSignalRecordV1 { + project_id: "project.age-continuation".to_owned(), + signal_id: format!("signal.{ordinal:05}"), + scope_digest: "scope".to_owned(), + repository_id: "repo".to_owned(), + watermark_id: "watermark".to_owned(), + observed_at_micros: 100 + ordinal, + signal_json: "{}".to_owned(), + }, + Vec::new(), + ) + .await + .expect("old terminal signal"); + } + let config = tracedecay_configuration::RetentionConfig::default(); + let first = super::run_registered_store_retention(&harness.registered, &config).await; + // Insert old optional detail after the first pass. A successful + // storage continuation must leave it for the next regular age pass. + harness + .registered + .writer_connection() + .expect("writer") + .execute_batch( + "CREATE TABLE age_pass_audit (runs INTEGER NOT NULL) STRICT; + INSERT INTO age_pass_audit VALUES (0); + CREATE TRIGGER count_age_work AFTER DELETE ON analytics_events + BEGIN UPDATE age_pass_audit SET runs = runs + 1; END;", + ) + .await + .expect("audit actual age deletions"); + harness + .registered + .append_observability_event(&tracedecay_global_db::AnalyticsEventInsert { + provider: "tracedecay-observability".to_owned(), + project_id: "scope:age-continuation".to_owned(), + session_id: None, + timestamp: 0, + event_kind: "retrieval.query.completed.v1".to_owned(), + hook_name: None, + tool_name: None, + tool_category: None, + skill_name: None, + hint_category: None, + hint_id: Some("age-continuation:marker".to_owned()), + outcome: Some("succeeded".to_owned()), + metadata_json: Some( + serde_json::json!({"retention_class":"optional_local_detail30d"}) + .to_string(), + ), + }) + .await + .expect("expired optional detail after first pass"); + let resumed = if first.continuation().is_some() { + super::run_registered_storage_continuation(&harness.registered, &config).await + } else { + super::run_registered_store_retention(&harness.registered, &config).await + }; + let mut rows = harness + .registered + .read_connection() + .query("SELECT runs FROM age_pass_audit", ()) + .await + .expect("age audit"); + assert_eq!( + rows.next().await.unwrap().unwrap().get::(0).unwrap(), + 0, + "successful terminal progress must not replay optional age deletion" + ); + assert_eq!( + first, + MaintenanceTickOutcome::Continue(MaintenanceContinuation::StoreRetention) + ); + assert!(resumed.succeeded()); + }) + .expect("native runtime"); + } + #[test] fn released_store_admits_ordinary_writer_without_another_retention_tick() { native_runtime::block_on(async { @@ -2497,7 +2744,7 @@ mod tests { } #[test] - fn soft_budget_alone_never_enables_destructive_maintenance() { + fn soft_budget_alone_never_enables_optional_retention() { let mut retention = tracedecay_configuration::RetentionConfig::default(); retention.session_lcm.enabled = false; retention.observation.enabled = false; @@ -2508,11 +2755,49 @@ mod tests { .insert("sessions.db".to_string(), 1); assert!( - !super::retention_maintenance_enabled(&retention), + !super::optional_retention_enabled(&retention), "soft budgets are Doctor findings, never a retention trigger" ); } + /// Disabled optional policies still start the fixed terminal-delivery + /// owner, and that owner joins at shutdown like the full retention loop. + #[test] + fn disabled_optional_policies_still_start_the_terminal_housekeeping_owner() { + tracedecay_runtime_core::native_runtime::block_on(async { + let temp = tempfile::TempDir::new().expect("temp root"); + let mut retention = tracedecay_configuration::RetentionConfig::default(); + retention.session_lcm.enabled = false; + retention.observation.enabled = false; + retention.orphan_store_gc_days = None; + retention.compaction = None; + retention + .store_soft_budgets_bytes + .insert("sessions.db".to_string(), 1); + assert!(!super::optional_retention_enabled(&retention)); + let coordinator = MaintenanceCoordinator::spawn( + temp.path().join("profile"), + crate::daemon::branch_admin::StoreAdministration::default(), + tracedecay_code_index_runtime::code_index_scheduler::CodeIndexSchedulerRegistryV1::new(1), + retention, + super::BranchStoreGcCadenceV1 { branch_gc_days: 30 }, + ) + .await; + assert!( + coordinator.task.lock().await.is_some(), + "fixed terminal housekeeping is admitted without optional policies" + ); + tracedecay_runtime_core::native_runtime::timeout( + Duration::from_secs(10), + coordinator.shutdown(), + ) + .await + .expect("shutdown joins the terminal housekeeping owner"); + assert!(coordinator.task.lock().await.is_none()); + }) + .expect("native runtime"); + } + #[test] fn resident_memory_log_reports_each_transition_once() { use super::{ResidentMemoryLogStateV1, ResidentMemoryLogTransitionV1}; @@ -3032,7 +3317,7 @@ mod tests { orphan_store_gc_days: Some(30), ..tracedecay_configuration::RetentionConfig::default() }; - assert!(super::retention_maintenance_enabled(&retention)); + assert!(super::optional_retention_enabled(&retention)); let coordinator = MaintenanceCoordinator::spawn( temp.path().join("profile"), crate::daemon::branch_admin::StoreAdministration::default(), diff --git a/crates/tracedecay/src/daemon/scheduler.rs b/crates/tracedecay/src/daemon/scheduler.rs index 3e664c16e74..006987f92bc 100644 --- a/crates/tracedecay/src/daemon/scheduler.rs +++ b/crates/tracedecay/src/daemon/scheduler.rs @@ -1490,7 +1490,9 @@ async fn scheduler_sleep(project_path: &Path, duration: Duration) -> bool { false } -async fn automation_scheduler_has_work_for_project(cg: &TraceDecay) -> Result { +pub(in crate::daemon) async fn automation_scheduler_has_work_for_project( + cg: &TraceDecay, +) -> Result { let configuration = effective_automation_config_for_project(cg).await?; automation_scheduler_has_work(cg, &configuration.settings).await } diff --git a/crates/tracedecay/src/daemon/tests.rs b/crates/tracedecay/src/daemon/tests.rs index 82422ff274c..79d51673df9 100644 --- a/crates/tracedecay/src/daemon/tests.rs +++ b/crates/tracedecay/src/daemon/tests.rs @@ -54,6 +54,8 @@ mod runtime_identity; mod scheduler_config; mod scheduler_shutdown; mod socket; +#[cfg(unix)] +mod terminal_housekeeping; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) enum ObservedMcpRoute { diff --git a/crates/tracedecay/src/daemon/tests/terminal_housekeeping.rs b/crates/tracedecay/src/daemon/tests/terminal_housekeeping.rs new file mode 100644 index 00000000000..5407b7d71a4 --- /dev/null +++ b/crates/tracedecay/src/daemon/tests/terminal_housekeeping.rs @@ -0,0 +1,490 @@ +//! Fixed terminal GitHub stack delivery housekeeping through the daemon's +//! maintenance owner, on a mounted project whose agent jobs and optional +//! retention policies are all disabled. + +use std::sync::Arc; +use std::time::Duration; + +use crate::daemon::branch_admin::{StoreWriterClass, store_writer_scope}; +use tracedecay_runtime_core::native_runtime::{self, NativeTask}; + +use tempfile::TempDir; +use tracedecay_automation_runtime::automation::config::AutomationConfigPatch; +use tracedecay_daemon_protocol::DaemonHandshake; +use tracedecay_global_db::RegisteredGlobalDb; +use tracedecay_maintenance::tick::{MaintenanceContinuation, MaintenanceTickOutcome}; + +use super::DaemonEngine; +use super::{ + apply_project_automation_patch_via_surface, enter_test_daemon_database_scope, + initialize_test_project, test_client_identity_for, test_daemon_engine_for_profile, + test_handshake_defaults, +}; +use crate::daemon::maintenance::{BranchStoreGcCadenceV1, MaintenanceCoordinator}; + +const STACK_PROJECT: &str = "project.terminal-housekeeping"; +const STACK_RECIPIENT: &str = "actor.terminal-housekeeping"; + +/// A mounted project with every agent job disabled, its daemon engine, and a +/// maintenance owner started with every optional retention policy disabled. +struct DisabledAutomationDaemon { + _dir: TempDir, + _database_scope: tracedecay_runtime_core::db::DaemonDatabaseScope, + engine: DaemonEngine, + server: Arc, + coordinator: MaintenanceCoordinator, + continuation: tokio::sync::Mutex>, +} + +impl DisabledAutomationDaemon { + async fn open() -> Self { + Self::open_with_projects(1).await + } + + async fn open_with_projects(project_count: usize) -> Self { + let dir = TempDir::new().expect("temp dir"); + let project = dir.path().canonicalize().expect("canonical temp dir"); + let client_identity = test_client_identity_for(project.join("profile")); + std::fs::create_dir_all(project.join("src")).expect("src dir"); + std::fs::write(project.join("src/main.rs"), "fn main() {}\n").expect("source file"); + initialize_test_project(&project, &client_identity).await; + let handshake = DaemonHandshake { + project_path: Some(project.clone()), + client_identity, + ..test_handshake_defaults() + }; + let mut extra_handshakes = Vec::new(); + for ordinal in 1..project_count { + let root = project.join(format!("project-{ordinal}")); + std::fs::create_dir_all(root.join("src")).expect("project source directory"); + std::fs::write(root.join("src/lib.rs"), "pub fn marker() {}\n") + .expect("project source"); + initialize_test_project(&root, &handshake.client_identity).await; + extra_handshakes.push(DaemonHandshake { + project_path: Some(root), + ..handshake.clone() + }); + } + let engine = test_daemon_engine_for_profile(&handshake.client_identity.profile_root); + let database_scope = enter_test_daemon_database_scope( + &handshake.client_identity.profile_root, + "terminal-housekeeping", + ); + let server = apply_project_automation_patch_via_surface( + &engine, + &handshake, + AutomationConfigPatch { + enabled: Some(false), + ..AutomationConfigPatch::default() + }, + ) + .await; + for extra in extra_handshakes { + apply_project_automation_patch_via_surface( + &engine, + &extra, + AutomationConfigPatch { + enabled: Some(false), + ..AutomationConfigPatch::default() + }, + ) + .await; + } + let graph = server.cg_snapshot().await; + assert!( + !crate::daemon::scheduler::automation_scheduler_has_work_for_project(graph.as_ref()) + .await + .expect("automation admission"), + "no agent job is configured, so no project automation scheduler is admitted" + ); + drop(graph); + + let retention = disabled_retention(); + let coordinator = MaintenanceCoordinator::spawn( + handshake.client_identity.profile_root.clone(), + engine.store_administration.clone(), + tracedecay_code_index_runtime::code_index_scheduler::CodeIndexSchedulerRegistryV1::new( + 1, + ), + retention, + BranchStoreGcCadenceV1 { branch_gc_days: 30 }, + ) + .await; + Self { + _dir: dir, + _database_scope: database_scope, + engine, + server, + coordinator, + continuation: tokio::sync::Mutex::new(None), + } + } + + fn shard(&self) -> tracedecay_global_db::RegisteredGlobalDbLeaseV1 { + self.server + .project_session_db() + .expect("project session shard") + } + + async fn tick(&self) -> MaintenanceTickOutcome { + let mut continuation = self.continuation.lock().await; + let outcome = terminal_tick( + &self.coordinator, + &self.engine.store_administration, + *continuation, + ) + .await; + *continuation = outcome.continuation(); + outcome + } + + async fn close(self) { + tracedecay_runtime_core::native_runtime::timeout( + std::time::Duration::from_secs(10), + self.coordinator.shutdown(), + ) + .await + .expect("shutdown joins the terminal housekeeping owner"); + } +} + +fn disabled_retention() -> tracedecay_configuration::RetentionConfig { + let mut retention = tracedecay_configuration::RetentionConfig::default(); + retention.session_lcm.enabled = false; + retention.observation.enabled = false; + retention.orphan_store_gc_days = None; + retention.compaction = None; + retention + .store_soft_budgets_bytes + .insert("sessions.db".to_owned(), 1); + retention +} + +async fn terminal_tick( + owner: &MaintenanceCoordinator, + administration: &crate::daemon::branch_admin::StoreAdministration, + continuation: Option, +) -> MaintenanceTickOutcome { + let profile = administration + .registered_profile_database() + .await + .expect("profile authority"); + owner + .run_tick( + profile.db_path().parent().expect("profile root"), + &profile, + administration, + &tracedecay_code_index_runtime::code_index_scheduler::CodeIndexSchedulerRegistryV1::new( + 1, + ), + &disabled_retention(), + BranchStoreGcCadenceV1 { branch_gc_days: 30 }, + continuation, + ) + .await +} + +async fn seed_stack_signal( + database: &RegisteredGlobalDb, + signal_id: &str, + observed_at_micros: i64, + settled: bool, +) { + database + .append_github_stack_signal( + tracedecay_global_db::GitHubStackSignalRecordV1 { + project_id: STACK_PROJECT.to_owned(), + signal_id: signal_id.to_owned(), + scope_digest: "sha256:terminal-housekeeping-scope".to_owned(), + repository_id: "repository.terminal-housekeeping".to_owned(), + watermark_id: "watermark.terminal-housekeeping".to_owned(), + observed_at_micros, + signal_json: format!(r#"{{"payload":"{signal_id}"}}"#), + }, + vec![STACK_RECIPIENT.to_owned()], + ) + .await + .expect("persist stack signal"); + if settled { + database + .acknowledge_github_stack_host_delivery(STACK_PROJECT, signal_id, STACK_RECIPIENT) + .await + .expect("settle stack signal"); + } +} + +async fn count_rows(database: &RegisteredGlobalDb, sql: &str) -> i64 { + let mut rows = database + .read_connection() + .query(sql, ()) + .await + .expect("count query"); + rows.next() + .await + .expect("count row") + .expect("count result") + .get::(0) + .expect("count value") +} + +async fn retired_signals(database: &RegisteredGlobalDb) -> i64 { + count_rows( + database, + "SELECT COUNT(*) FROM github_stack_delivery_signals + WHERE json_type(signal_json, '$.__tracedecay_retired_stack_sha256_v1') IS NOT NULL", + ) + .await +} + +async fn retirement_frontiers(database: &RegisteredGlobalDb) -> i64 { + count_rows( + database, + "SELECT COUNT(*) FROM github_stack_delivery_retirement_frontier", + ) + .await +} + +async fn payload_present(database: &RegisteredGlobalDb, signal_id: &str) -> bool { + database + .github_stack_signal(STACK_PROJECT, signal_id) + .await + .expect("read stack signal") + .is_some() +} + +/// A mounted project's stack history lives in its own session shard. With +/// every agent job and optional policy disabled the fixed pass retires it +/// there; the profile database cannot stand in, and none of the full tick's +/// other retention work runs. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn expired_settled_project_shard_signals_retire_with_all_automation_disabled() { + let daemon = DisabledAutomationDaemon::open().await; + let shard = daemon.shard(); + let profile = daemon + .engine + .store_administration + .registered_profile_database() + .await + .expect("profile database"); + assert_ne!(shard.db_path(), profile.db_path()); + + let recent = tracedecay_contracts::now_micros().0; + seed_stack_signal(&shard, "signal.expired-settled", 100, true).await; + seed_stack_signal(&shard, "signal.expired-pending", 101, false).await; + seed_stack_signal(&shard, "signal.recent-settled", recent, true).await; + // Optional-detail observability pruning belongs to the full tick, not to + // the fixed pass, so an expired row must survive it. + shard + .append_observability_event(&tracedecay_global_db::AnalyticsEventInsert { + provider: "tracedecay-observability".to_owned(), + project_id: "scope:terminal-housekeeping".to_owned(), + session_id: None, + timestamp: 0, + event_kind: "retrieval.query.completed.v1".to_owned(), + hook_name: None, + tool_name: None, + tool_category: None, + skill_name: None, + hint_category: None, + hint_id: Some("terminal-housekeeping:optional-marker".to_owned()), + outcome: Some("succeeded".to_owned()), + metadata_json: Some( + serde_json::json!({ "retention_class": "optional_local_detail30d" }).to_string(), + ), + }) + .await + .expect("seed an expired optional observability detail"); + + // Retiring only the profile shard is the wrong-shard cleanup. + let wrong_shard = + tracedecay_maintenance::retention::registered_store::run_github_stack_terminal_retention( + profile.as_ref(), + tracedecay_runtime_core::tracedecay::current_timestamp(), + ) + .await + .expect("profile-only retirement"); + assert_eq!(wrong_shard.retired, 0); + assert_eq!(retired_signals(&shard).await, 0); + + assert_eq!(daemon.tick().await, MaintenanceTickOutcome::Complete); + + assert_eq!(retired_signals(&shard).await, 1); + assert!( + !payload_present(&shard, "signal.expired-settled").await, + "the expired settled payload is compacted in the project shard" + ); + assert!(payload_present(&shard, "signal.expired-pending").await); + assert!(payload_present(&shard, "signal.recent-settled").await); + assert_eq!( + shard + .query_analytics_events(&tracedecay_global_db::AnalyticsEventQuery { + provider: Some("tracedecay-observability".to_owned()), + project_id: Some("scope:terminal-housekeeping".to_owned()), + limit: 10, + ..tracedecay_global_db::AnalyticsEventQuery::default() + }) + .await + .expect("query optional marker") + .len(), + 1, + "the fixed pass runs none of the full tick's other retention work" + ); + + drop(shard); + drop(profile); + daemon.close().await; +} + +/// Each admitted unit retires one page and releases admission. The same +/// durable frontier advances through continuation ticks and clears on drain. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn terminal_housekeeping_resumes_from_its_frontier_after_the_page_budget() { + // Preserve the 1041 settled + 3 pending journey across page admissions. + const FIRST_TICK: i64 = 64 * 16; + const TAIL: i64 = 17; + let daemon = DisabledAutomationDaemon::open().await; + let shard = daemon.shard(); + for pending in 0..3 { + seed_stack_signal( + &shard, + &format!("signal.pending-{pending}"), + 1 + pending, + false, + ) + .await; + } + for ordinal in 0..(FIRST_TICK + TAIL) { + seed_stack_signal( + &shard, + &format!("signal.settled-{ordinal:05}"), + 1_000 + ordinal, + true, + ) + .await; + } + + for _ in 0..64 { + assert_eq!( + daemon.tick().await, + MaintenanceTickOutcome::Continue(MaintenanceContinuation::StoreRetention) + ); + } + assert_eq!(retired_signals(&shard).await, FIRST_TICK); + assert_eq!( + retirement_frontiers(&shard).await, + 1, + "an unfinished pass keeps its durable frontier" + ); + + assert_eq!( + daemon.tick().await, + MaintenanceTickOutcome::Continue(MaintenanceContinuation::StoreRetention) + ); + assert_eq!(retired_signals(&shard).await, FIRST_TICK + 16); + assert_eq!(daemon.tick().await, MaintenanceTickOutcome::Complete); + assert_eq!(retired_signals(&shard).await, FIRST_TICK + TAIL); + assert_eq!( + retirement_frontiers(&shard).await, + 0, + "a drained pass clears the frontier so settled history is revisited" + ); + assert_eq!( + count_rows( + shard.as_ref(), + "SELECT COUNT(*) FROM github_stack_delivery_recipients WHERE state = 'pending'", + ) + .await, + 3, + "undelivered signals are never retired by age" + ); + + drop(shard); + daemon.close().await; +} + +/// Hold the real SQLite writer in A after it has a terminal backlog, so its +/// admitted maintenance is suspended inside storage while B opens a writer. +#[test] +fn terminal_backlog_does_not_block_an_unrelated_store_writer() { + native_runtime::block_on(async { + let daemon = DisabledAutomationDaemon::open().await; + let shard = daemon.shard(); + for ordinal in 0..33 { + seed_stack_signal(&shard, &format!("signal.{ordinal:03}"), 100 + ordinal, true).await; + } + let held = shard.begin_write_transaction().await.expect("hold A SQLite writer"); + let owner = daemon.coordinator.clone(); + let administration = daemon.engine.store_administration.clone(); + let tick = NativeTask::spawn(async move { + terminal_tick(&owner, &administration, None).await + }).expect("native terminal tick"); + let a_scope = store_writer_scope(shard.db_path().parent().expect("A store root"), StoreWriterClass::Content); + native_runtime::timeout(Duration::from_secs(10), async { + while daemon.engine.store_administration.try_with_writer_in(a_scope.clone(), || async {}).await.is_some() { + assert!(!tick.is_finished(), "tick must wait inside A storage"); + tokio::task::yield_now().await; + } + }).await.expect("maintenance admitted A"); + let other = tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness::open("terminal-writer-b").await; + let b_scope = store_writer_scope(other.registered.db_path().parent().expect("B store root"), StoreWriterClass::Content); + let progressed = native_runtime::timeout(Duration::from_secs(2), + daemon.engine.store_administration.with_writer_in(b_scope, || async { + other.registered.writer_connection().expect("B writer") + .execute_batch("CREATE TABLE unrelated_writer_progress (id INTEGER PRIMARY KEY) STRICT; + INSERT INTO unrelated_writer_progress VALUES (1);").await.expect("B writes"); + }) + ).await; + // Release A before asserting, so a failed interleaving proof still + // settles its admitted owner and releases every fixture lease. + held.rollback().await.expect("release A"); + let outcome = native_runtime::timeout(Duration::from_secs(10), tick.wait()).await.expect("terminal tick settled"); + assert!(matches!(outcome, native_runtime::NativeTaskOutcome::Completed(_))); + assert!(progressed.is_ok(), "B must write while A is suspended with a terminal backlog"); + assert_eq!(retired_signals(&shard).await, 16, "one admission runs one bounded terminal page"); + assert_eq!(retirement_frontiers(&shard).await, 1); + drop(shard); + daemon.close().await; + }).expect("native runtime"); +} + +/// The profile shard plus eight mounted projects exceeds one store window. +/// Unvisited shards must continue promptly without retaining their leases. +#[test] +fn terminal_housekeeping_bounds_a_mounted_store_window_and_reaches_its_tail() { + native_runtime::block_on(async { + let daemon = DisabledAutomationDaemon::open_with_projects(8).await; + let databases = daemon + .engine + .store_administration + .mounted_registered_session_databases() + .await; + assert_eq!( + databases.len(), + 9, + "eight projects and their profile shard are mounted" + ); + for database in &databases { + seed_stack_signal(database, "signal.window", 100, true).await; + } + assert_eq!( + daemon.tick().await, + MaintenanceTickOutcome::Continue(MaintenanceContinuation::StoreRetention) + ); + let mut retired = 0; + for database in &databases { + retired += retired_signals(database).await; + } + assert_eq!(retired, 8, "one tick admits at most eight store pages"); + assert_eq!(daemon.tick().await, MaintenanceTickOutcome::Complete); + for database in &databases { + assert_eq!( + retired_signals(database).await, + 1, + "every mounted shard is reached" + ); + } + drop(databases); + daemon.close().await; + }) + .expect("native runtime"); +}