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(()) }