From 5d803541bb9646c48bfef26a402c9dbb4802ac6c Mon Sep 17 00:00:00 2001 From: Duncan Date: Fri, 21 Aug 2026 15:40:41 -0400 Subject: [PATCH] fix(archive): trigger prune from unconditional startup, not sync start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prune worker's startup trigger hung off start_archive_sync, but the renderer skips start_archive_sync while observer reconciliation is not ready (useArchiveSync returns early), so an archive whose reconciliation never succeeds and takes no new commits was never pruned — violating the plan's startup + post-commit trigger contract. Move the startup trigger into spawn_warm_init, which runs on every application startup after the init barrier warms, and drop the now-redundant sync-start hook. Also make the partition worklist deterministic (ORDER BY identity/relay) so a mid-worklist failure leaves a well-defined pruned prefix, and rework the failure regression to fail only the later partition, assert the earlier committed with the gate unstamped, then clear the fault and prove the retry completes the remainder and stamps success. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- desktop/src-tauri/src/archive/mod.rs | 12 ++++- desktop/src-tauri/src/archive/prune.rs | 20 ++++++--- desktop/src-tauri/src/archive/prune_tests.rs | 46 ++++++++++++++++---- desktop/src-tauri/src/archive/sync.rs | 10 +---- 4 files changed, 63 insertions(+), 25 deletions(-) diff --git a/desktop/src-tauri/src/archive/mod.rs b/desktop/src-tauri/src/archive/mod.rs index 9982a69576..b7e552f403 100644 --- a/desktop/src-tauri/src/archive/mod.rs +++ b/desktop/src-tauri/src/archive/mod.rs @@ -54,12 +54,22 @@ const OBSERVER_FRAME_TELEMETRY: &str = "telemetry"; /// `await`s this barrier before its first write, so warming it early avoids a /// stall on the first observer frame. Non-fatal: the first real archive command /// retries and surfaces any error. +/// +/// This is also the archive's unconditional-startup prune trigger. After +/// warming init, it runs one opportunistic global prune pass, so a launch whose +/// archive sync never starts (reconciliation never succeeds) and which takes no +/// new archive commits still prunes stale observer frames once the ≥24h gate +/// allows — the plan's "startup" trigger. `maybe_prune` rides the same gated +/// adapter (its own init `await`), and its single-flight flag + gate collapse +/// this pass with the post-commit trigger into at most one real prune per day. pub fn spawn_warm_init(app: tauri::AppHandle) { tauri::async_runtime::spawn(async move { use tauri::Manager; - if let Err(error) = app.state::().archive_db.warm_init().await { + let state = app.state::(); + if let Err(error) = state.archive_db.warm_init().await { eprintln!("buzz-desktop: archive DB init deferred: {error}"); } + prune::maybe_prune(&state).await; }); } diff --git a/desktop/src-tauri/src/archive/prune.rs b/desktop/src-tauri/src/archive/prune.rs index ea5d8bf4cf..90b88a8079 100644 --- a/desktop/src-tauri/src/archive/prune.rs +++ b/desktop/src-tauri/src/archive/prune.rs @@ -8,9 +8,9 @@ //! # Scheduling //! //! No cron. The worker is triggered opportunistically — once shortly after -//! archive sync starts, and again after each archive commit — and gated two -//! ways so those frequent triggers collapse into at most one real prune per -//! day: +//! application startup (after the archive init barrier warms), and again after +//! each archive commit — and gated two ways so those frequent triggers collapse +//! into at most one real prune per day: //! //! 1. **In-process single-flight** ([`super::ArchiveDb::try_prune_guard`]): a //! `compare_exchange` on an `AtomicBool` admits exactly one prune at a time @@ -97,14 +97,20 @@ LIMIT ?4 /// the global gate covers the whole archive rather than just the partition that /// happened to be active when the trigger fired. A partition with no kind-24200 /// scopes in range simply prunes nothing. +/// +/// Ordered so the worklist is deterministic: a pass that fails partway leaves a +/// well-defined prefix pruned, and the idempotent retry resumes the remainder +/// in the same order (already-pruned partitions re-run as no-ops). pub(super) const PRUNE_PARTITIONS_SQL: &str = " SELECT DISTINCT identity_pubkey, relay_url FROM archived_event_scopes +ORDER BY identity_pubkey, relay_url "; -/// Spawn a detached prune pass keyed off the app handle. Called post-commit and -/// at archive-sync startup. The prune runs on the async runtime and its result -/// is swallowed, so a prune error never propagates into the archive write that -/// triggered it. +/// Spawn a detached prune pass keyed off the app handle. Called post-commit +/// from the archive write path (the unconditional-startup trigger calls +/// [`maybe_prune`] directly, since it already holds the [`AppState`]). The prune +/// runs on the async runtime and its result is swallowed, so a prune error never +/// propagates into the archive write that triggered it. pub fn trigger_prune(app: AppHandle) { tauri::async_runtime::spawn(async move { let state = app.state::(); diff --git a/desktop/src-tauri/src/archive/prune_tests.rs b/desktop/src-tauri/src/archive/prune_tests.rs index 23f1a25117..79f3a301e3 100644 --- a/desktop/src-tauri/src/archive/prune_tests.rs +++ b/desktop/src-tauri/src/archive/prune_tests.rs @@ -321,28 +321,58 @@ fn run_prune_if_due_prunes_every_partition_before_stamping() { } #[test] -fn run_prune_if_due_leaves_gate_unstamped_when_a_partition_fails() { - // A mid-worklist failure must leave the gate unstamped so the idempotent - // pass retries the remainder. Force the failure by dropping the metric index - // (the last cascade stage), so whichever partition is pruned first errors - // before the worklist completes. +fn run_prune_if_due_leaves_gate_unstamped_then_retry_completes_remainder() { + // A mid-worklist failure must commit the prefix, leave the gate unstamped, + // and let the idempotent retry finish the remainder. The worklist is + // ordered by (identity_pubkey, relay_url), so "idA" prunes before "idB". + // A BEFORE DELETE trigger that aborts only on partition idB's events fails + // the LATER partition — after idA has already committed its own batch. let db = NamedTempFile::new().unwrap(); let conn = fresh(&db); seed_observer_in(&conn, "idA", "wss://a", "old_a", 100); seed_observer_in(&conn, "idB", "wss://b", "old_b", 100); retention::set_observer_retention_days(&conn, 1).unwrap(); - conn.execute_batch("DROP TABLE agent_metric_index").unwrap(); + conn.execute_batch( + "CREATE TRIGGER fail_idB BEFORE DELETE ON archived_events + WHEN OLD.identity_pubkey = 'idB' + BEGIN SELECT RAISE(ABORT, 'forced idB failure'); END", + ) + .unwrap(); - let result = run_prune_if_due(&conn, 10_000_000); + let now = 10_000_000; + let result = run_prune_if_due(&conn, now); assert!( result.is_err(), - "a failing partition must surface the error" + "the later partition's forced failure must surface" + ); + assert!( + !event_exists(&conn, "old_a"), + "partition idA was pruned and committed before idB failed" + ); + assert!( + event_exists(&conn, "old_b"), + "partition idB's batch rolled back on the forced failure" ); assert_eq!( retention::get_prune_last_success_at(&conn).unwrap(), None, "a partial worklist must not stamp success — the pass retries" ); + + // Clear the fault and rerun: idA is already pruned (a no-op), idB now + // prunes, and the complete worklist stamps success. + conn.execute_batch("DROP TRIGGER fail_idB").unwrap(); + run_prune_if_due(&conn, now).unwrap(); + assert!( + !event_exists(&conn, "old_b"), + "the retry prunes the remainder" + ); + assert_eq!(count(&conn, "archived_events"), 0); + assert_eq!( + retention::get_prune_last_success_at(&conn).unwrap(), + Some(now), + "the completed worklist stamps success" + ); } // ── Stale-backfill orphan race (conditional observer-index insert) ────────────── diff --git a/desktop/src-tauri/src/archive/sync.rs b/desktop/src-tauri/src/archive/sync.rs index 6fbc8856e9..225b15d534 100644 --- a/desktop/src-tauri/src/archive/sync.rs +++ b/desktop/src-tauri/src/archive/sync.rs @@ -587,7 +587,7 @@ pub async fn start_archive_sync( .await; let io = AppIo { - app: app.clone(), + app, session: Arc::clone(&session), }; tauri::async_runtime::spawn(async move { @@ -595,14 +595,6 @@ pub async fn start_archive_sync( session.set_subscriptions(Vec::new()).await; }); - // Startup prune trigger: one opportunistic pass shortly after sync begins, - // so a launch that never archives anything (idle agents) still prunes stale - // observer frames once its ≥24h gate allows. The pass is archive-global — it - // enumerates every stored partition, not the identity/relay this sync is - // for — so hooking it here rather than at unconditional startup costs only - // latency to the first pass, never coverage. Detached and gated exactly like - // the post-commit trigger; a prune failure never affects sync startup. - super::prune::trigger_prune(app); Ok(()) }