mirror of
https://github.com/block/buzz.git
synced 2026-10-31 06:21:34 +00:00
feat(archive): prune observer frames past the retention window
Add the opportunistic prune worker for the local archive DB: it deletes kind-24200 observer-frame scope rows whose archived_at is older than observer_retention_days, cascading orphan GC to both derived indexes, and keeps every other kind forever. Triggers are startup + post-commit, detached from the archive write and collapsed by an in-process single-flight flag plus a >=24h archive_meta.prune_last_success_at gate into at most one scan per day. Candidate scope keys are materialized and fixed per batch transaction; the candidate-scoped GC cascades observer_channel_index (which the whole-scan GC does not cover) and agent_metric_index. The observer-index insert becomes a conditional INSERT ... SELECT so a stale backfill callback cannot resurrect an index row for a pruned event. The prune-candidate SQL constant is colocated with the worker so the EXPLAIN-shape test pins the shipped query. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
@@ -30,6 +30,7 @@
|
||||
//! durability.
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
use rusqlite::Connection;
|
||||
use tokio::sync::{OnceCell, RwLock};
|
||||
@@ -65,6 +66,11 @@ pub struct ArchiveDb {
|
||||
/// Maintenance lock. Ordinary connections hold the read guard for their
|
||||
/// whole lifetime; the Phase-4 conversion holds the write guard.
|
||||
maintenance: RwLock<()>,
|
||||
/// In-process single-flight flag for the opportunistic prune worker
|
||||
/// (observer-frame retention). A `compare_exchange` admits exactly one
|
||||
/// prune at a time in this process; a trigger arriving while a prune runs
|
||||
/// is dropped, not queued. See [`super::prune`].
|
||||
prune_running: AtomicBool,
|
||||
/// Test-only path/hook overrides; always `None` in production.
|
||||
#[cfg(test)]
|
||||
test_seam: Option<TestSeam>,
|
||||
@@ -154,6 +160,27 @@ impl ArchiveDb {
|
||||
.await
|
||||
.map_err(|e| format!("archive db task failed: {e}"))?
|
||||
}
|
||||
|
||||
/// Try to claim the prune single-flight flag. Returns `Some(guard)` for the
|
||||
/// one caller that transitions the flag `false -> true`; the guard clears it
|
||||
/// on drop (including on early return or panic). A caller that arrives while
|
||||
/// a prune is already running gets `None` and skips its pass.
|
||||
pub(super) fn try_prune_guard(&self) -> Option<PruneGuard<'_>> {
|
||||
self.prune_running
|
||||
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
||||
.is_ok()
|
||||
.then_some(PruneGuard(&self.prune_running))
|
||||
}
|
||||
}
|
||||
|
||||
/// RAII guard that clears the prune single-flight flag when a prune finishes.
|
||||
/// Held across the blocking prune dispatch.
|
||||
pub(super) struct PruneGuard<'a>(&'a AtomicBool);
|
||||
|
||||
impl Drop for PruneGuard<'_> {
|
||||
fn drop(&mut self) {
|
||||
self.0.store(false, Ordering::Release);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -164,6 +191,7 @@ impl ArchiveDb {
|
||||
Self {
|
||||
init: OnceCell::new(),
|
||||
maintenance: RwLock::new(()),
|
||||
prune_running: AtomicBool::new(false),
|
||||
test_seam: Some(TestSeam {
|
||||
path,
|
||||
on_init: None,
|
||||
@@ -178,6 +206,7 @@ impl ArchiveDb {
|
||||
Self {
|
||||
init: OnceCell::new(),
|
||||
maintenance: RwLock::new(()),
|
||||
prune_running: AtomicBool::new(false),
|
||||
test_seam: Some(TestSeam {
|
||||
path,
|
||||
on_init: Some(hook),
|
||||
|
||||
@@ -21,6 +21,7 @@ mod agent_usage;
|
||||
mod archive_db;
|
||||
mod metric_store;
|
||||
mod pipeline;
|
||||
mod prune;
|
||||
pub mod retention;
|
||||
pub mod store;
|
||||
mod store_migrations;
|
||||
@@ -186,7 +187,7 @@ pub(crate) async fn archive_candidates(
|
||||
};
|
||||
let commit_identity_pk = identity_pk.clone();
|
||||
let commit_relay_url = relay_url.clone();
|
||||
state
|
||||
let result = state
|
||||
.archive_db
|
||||
.with_conn(move |conn| {
|
||||
commit_archive(
|
||||
@@ -200,7 +201,19 @@ pub(crate) async fn archive_candidates(
|
||||
conn,
|
||||
)
|
||||
})
|
||||
.await
|
||||
.await?;
|
||||
|
||||
// Post-commit prune trigger, detached from this write. The prune's own
|
||||
// single-flight flag and ≥24h gate collapse the frequent per-batch triggers
|
||||
// into at most one real prune per day; a prune failure is swallowed inside
|
||||
// the spawned task and never affects this archive result. Requires the app
|
||||
// handle (absent in headless tests that drive `archive_candidates`
|
||||
// directly), so a missing handle simply skips the trigger.
|
||||
if let Some(app) = state.app_handle.lock().ok().and_then(|g| g.clone()) {
|
||||
prune::trigger_prune(app);
|
||||
}
|
||||
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
/// Validate an ephemeral observer frame (kind 24200) against ALL local rules.
|
||||
|
||||
310
desktop/src-tauri/src/archive/prune.rs
Normal file
310
desktop/src-tauri/src/archive/prune.rs
Normal file
@@ -0,0 +1,310 @@
|
||||
//! Observer-frame retention prune worker (v4).
|
||||
//!
|
||||
//! Deletes archived kind-24200 observer-frame rows older than the configured
|
||||
//! window (`observer_retention_days`, from `archive_meta`). Every other
|
||||
//! archived kind — NIP-AM metrics (44200) and any custom subscription kind — is
|
||||
//! kept indefinitely and never touched here (Will's v4 ruling, 2026-08-19).
|
||||
//!
|
||||
//! # 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:
|
||||
//!
|
||||
//! 1. **In-process single-flight** ([`super::ArchiveDb::try_prune_guard`]): a
|
||||
//! `compare_exchange` on an `AtomicBool` admits exactly one prune at a time
|
||||
//! in this process; a trigger that arrives while a prune is running is
|
||||
//! dropped, not queued. This is a single-process desktop app, so an
|
||||
//! in-process flag is sufficient — no lease or fenced token (the v3 design's
|
||||
//! fenced prune lease was cut in v4).
|
||||
//! 2. **≥24h timestamp gate** (`archive_meta.prune_last_success_at`): a prune
|
||||
//! that finds the last success within 24h returns immediately. The timestamp
|
||||
//! is written after every clean pass (even one that deleted nothing), so a
|
||||
//! burst of post-commit triggers costs one meta read apiece, not a scan.
|
||||
//!
|
||||
//! The gate is database-global, so a due pass does global work: it enumerates a
|
||||
//! fixed worklist of every archived `(identity_pubkey, relay_url)` partition and
|
||||
//! prunes each under the same cutoff, stamping success only after the whole
|
||||
//! worklist completes. It never scopes to the identity/relay that happened to be
|
||||
//! active when the trigger fired, so an inactive historical partition is not
|
||||
//! starved and a partition failure leaves the gate unstamped for a retry.
|
||||
//!
|
||||
//! A prune failure is fully detached from the archive commit that triggered it:
|
||||
//! [`trigger_prune`] spawns onto the async runtime and swallows the result, so a
|
||||
//! prune error can never fail a user's archive write.
|
||||
//!
|
||||
//! # Delete discipline
|
||||
//!
|
||||
//! Candidate scope-row primary keys are materialized inside each batch
|
||||
//! transaction and fixed for that transaction (Thufir's binding condition #4):
|
||||
//! the batch deletes exactly those rows, then runs a **candidate-scoped** orphan
|
||||
//! GC over only the event ids it just unscoped — never a widened whole-table
|
||||
//! anti-join (`store::gc_orphaned_events` stays a rare repair pass, not the
|
||||
//! normal path) and never a post-delete re-scan to discover new orphans. The GC
|
||||
//! cascades BOTH derived indexes, `observer_channel_index` (the one the
|
||||
//! whole-scan GC does not cover) and `agent_metric_index`, so no index row
|
||||
//! outlives the canonical event it was derived from.
|
||||
|
||||
use rusqlite::{params, Connection};
|
||||
use tauri::{AppHandle, Manager};
|
||||
|
||||
use crate::app_state::AppState;
|
||||
|
||||
use super::retention;
|
||||
|
||||
/// Minimum spacing between successful prune passes. Frequent triggers below
|
||||
/// this interval no-op against the `prune_last_success_at` gate.
|
||||
const PRUNE_MIN_INTERVAL_SECS: i64 = 24 * 60 * 60;
|
||||
|
||||
/// Seconds per day, for the `archived_at` cutoff arithmetic. `days` is bounded
|
||||
/// to `retention::MAX_RETENTION_DAYS` (36 500), so `days * SECS_PER_DAY` cannot
|
||||
/// overflow `i64`.
|
||||
const SECS_PER_DAY: i64 = 24 * 60 * 60;
|
||||
|
||||
/// Scope-row primary keys deleted per batch transaction. Large enough that the
|
||||
/// first prune over a multi-million-row archive completes in a bounded number
|
||||
/// of transactions, small enough that each transaction's write lock is brief.
|
||||
const PRUNE_BATCH_SIZE: i64 = 2_000;
|
||||
|
||||
/// The prune-candidate scan: the single global observer window. Filters scope
|
||||
/// rows on `(identity_pubkey, relay_url, archived_at < cutoff)` and joins
|
||||
/// `archived_events` for `kind = 24200`, selecting the scope-row primary key so
|
||||
/// the batch `DELETE` can materialize its candidates. It constrains neither
|
||||
/// `scope_type` nor `scope_value` — the reason the M4 scope-age index leads
|
||||
/// with `archived_at` immediately after the identity/relay equality keys.
|
||||
///
|
||||
/// This is the production query. The EXPLAIN-shape test in `prune_tests.rs`
|
||||
/// runs `EXPLAIN QUERY PLAN` against this exact constant so the covering-index
|
||||
/// access path and the shipped query can never drift (Thufir's carry-forward
|
||||
/// note — the constant previously lived only in the test file).
|
||||
pub(super) const PRUNE_CANDIDATE_SQL: &str = "
|
||||
SELECT s.id, s.scope_type, s.scope_value
|
||||
FROM archived_event_scopes s
|
||||
JOIN archived_events e
|
||||
ON e.identity_pubkey = s.identity_pubkey
|
||||
AND e.relay_url = s.relay_url
|
||||
AND e.id = s.id
|
||||
WHERE s.identity_pubkey = ?1
|
||||
AND s.relay_url = ?2
|
||||
AND s.archived_at < ?3
|
||||
AND e.kind = 24200
|
||||
LIMIT ?4
|
||||
";
|
||||
|
||||
/// The fixed prune worklist: every archived `(identity_pubkey, relay_url)`
|
||||
/// partition holding scope rows. Enumerated once at the start of a due pass so
|
||||
/// 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.
|
||||
pub(super) const PRUNE_PARTITIONS_SQL: &str = "
|
||||
SELECT DISTINCT identity_pubkey, relay_url FROM archived_event_scopes
|
||||
";
|
||||
|
||||
/// 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.
|
||||
pub fn trigger_prune(app: AppHandle) {
|
||||
tauri::async_runtime::spawn(async move {
|
||||
let state = app.state::<AppState>();
|
||||
maybe_prune(&state).await;
|
||||
});
|
||||
}
|
||||
|
||||
/// Run one prune pass if the single-flight flag admits it. The pass is archive
|
||||
/// global — it prunes every stored partition, not just the active
|
||||
/// identity/relay — so it dispatches the gate check + worklist prune to the
|
||||
/// blocking pool via the gated archive adapter without resolving the active
|
||||
/// identity. Errors are logged, not returned: every caller is fire-and-forget.
|
||||
pub async fn maybe_prune(state: &AppState) {
|
||||
// Single-flight: admit exactly one prune at a time. A trigger arriving
|
||||
// while a prune runs is dropped — the next post-commit trigger will pick up
|
||||
// any rows this pass could not reach.
|
||||
let Some(_guard) = state.archive_db.try_prune_guard() else {
|
||||
return;
|
||||
};
|
||||
|
||||
let now = super::now_secs();
|
||||
|
||||
if let Err(error) = state
|
||||
.archive_db
|
||||
.with_conn(move |conn| run_prune_if_due(conn, now))
|
||||
.await
|
||||
{
|
||||
eprintln!("buzz-desktop: archive prune failed: {error}");
|
||||
}
|
||||
}
|
||||
|
||||
/// Enforce the ≥24h gate, then prune every partition and stamp the global
|
||||
/// success timestamp.
|
||||
///
|
||||
/// The gate (`archive_meta.prune_last_success_at`) is database-global, so the
|
||||
/// work it gates must be too: a due pass enumerates a fixed worklist of every
|
||||
/// archived `(identity_pubkey, relay_url)` partition ([`PRUNE_PARTITIONS_SQL`])
|
||||
/// and prunes each under the same cutoff. The timestamp is stamped only after
|
||||
/// the entire worklist succeeds — including a pass that deleted nothing — so a
|
||||
/// burst of triggers is bounded to a single scan per day, and a mid-worklist
|
||||
/// failure leaves the gate unstamped for the idempotent pass to retry.
|
||||
fn run_prune_if_due(conn: &Connection, now: i64) -> Result<(), String> {
|
||||
if let Some(last) = retention::get_prune_last_success_at(conn)? {
|
||||
if now - last < PRUNE_MIN_INTERVAL_SECS {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
let days = retention::get_observer_retention_days(conn)?;
|
||||
let cutoff = now - days * SECS_PER_DAY;
|
||||
|
||||
let mut stmt = conn
|
||||
.prepare(PRUNE_PARTITIONS_SQL)
|
||||
.map_err(|e| format!("prune: prepare partition scan: {e}"))?;
|
||||
let partitions: Vec<(String, String)> = stmt
|
||||
.query_map([], |row| {
|
||||
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
|
||||
})
|
||||
.map_err(|e| format!("prune: query partitions: {e}"))?
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| format!("prune: read partition row: {e}"))?;
|
||||
drop(stmt);
|
||||
|
||||
for (identity_pk, relay_url) in &partitions {
|
||||
prune_observer_frames(conn, identity_pk, relay_url, cutoff)?;
|
||||
}
|
||||
|
||||
retention::set_prune_last_success_at(conn, now)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Delete every observer-frame (kind 24200) scope row whose `archived_at` is
|
||||
/// before `cutoff`, in bounded batches, cascading the orphan GC to both derived
|
||||
/// indexes for the events those deletes leave unscoped.
|
||||
///
|
||||
/// Each batch is one transaction: materialize the candidate scope keys, delete
|
||||
/// them, then GC only those candidate events. The candidate set is fixed per
|
||||
/// transaction (binding condition #4). The loop ends when a batch returns fewer
|
||||
/// candidates than the batch size — i.e. the last, partial batch.
|
||||
pub(super) fn prune_observer_frames(
|
||||
conn: &Connection,
|
||||
identity_pk: &str,
|
||||
relay_url: &str,
|
||||
cutoff: i64,
|
||||
) -> Result<usize, String> {
|
||||
let mut total_events_deleted = 0usize;
|
||||
|
||||
loop {
|
||||
let tx = conn
|
||||
.unchecked_transaction()
|
||||
.map_err(|e| format!("prune: begin batch transaction: {e}"))?;
|
||||
|
||||
// Materialize the candidate scope rows in-tx and fix them for this
|
||||
// transaction — (id, scope_type, scope_value); identity/relay are the
|
||||
// fixed ?1/?2 params.
|
||||
let mut stmt = tx
|
||||
.prepare(PRUNE_CANDIDATE_SQL)
|
||||
.map_err(|e| format!("prune: prepare candidate scan: {e}"))?;
|
||||
let candidates: Vec<(String, String, String)> = stmt
|
||||
.query_map(
|
||||
params![identity_pk, relay_url, cutoff, PRUNE_BATCH_SIZE],
|
||||
|row| {
|
||||
Ok((
|
||||
row.get::<_, String>(0)?,
|
||||
row.get::<_, String>(1)?,
|
||||
row.get::<_, String>(2)?,
|
||||
))
|
||||
},
|
||||
)
|
||||
.map_err(|e| format!("prune: query candidates: {e}"))?
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| format!("prune: read candidate row: {e}"))?;
|
||||
drop(stmt);
|
||||
|
||||
if candidates.is_empty() {
|
||||
// Nothing left in range; the empty read did no writes, so rolling
|
||||
// back is the honest close (commit would be equally harmless).
|
||||
drop(candidates);
|
||||
let _ = tx.rollback();
|
||||
break;
|
||||
}
|
||||
let batch_len = candidates.len();
|
||||
|
||||
// 1. Delete the exact candidate scope rows by full primary key.
|
||||
{
|
||||
let mut del_scope = tx
|
||||
.prepare(
|
||||
"DELETE FROM archived_event_scopes
|
||||
WHERE identity_pubkey = ?1 AND relay_url = ?2
|
||||
AND id = ?3 AND scope_type = ?4 AND scope_value = ?5",
|
||||
)
|
||||
.map_err(|e| format!("prune: prepare scope delete: {e}"))?;
|
||||
for (id, scope_type, scope_value) in &candidates {
|
||||
del_scope
|
||||
.execute(params![identity_pk, relay_url, id, scope_type, scope_value])
|
||||
.map_err(|e| format!("prune: delete scope row: {e}"))?;
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Candidate-scoped orphan GC. Only the distinct event ids just
|
||||
// unscoped are checked — no widened anti-join, no post-delete
|
||||
// re-scan for other orphans. An event still holding another scope
|
||||
// row survives; one left with zero scope rows is deleted, and its
|
||||
// derived-index rows in BOTH indexes are cascaded away in the same
|
||||
// transaction so neither can outlive the canonical event.
|
||||
{
|
||||
let mut del_event = tx
|
||||
.prepare(
|
||||
"DELETE FROM archived_events
|
||||
WHERE identity_pubkey = ?1 AND relay_url = ?2 AND id = ?3
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM archived_event_scopes
|
||||
WHERE identity_pubkey = ?1 AND relay_url = ?2 AND id = ?3
|
||||
)",
|
||||
)
|
||||
.map_err(|e| format!("prune: prepare event GC: {e}"))?;
|
||||
let mut del_observer = tx
|
||||
.prepare(
|
||||
"DELETE FROM observer_channel_index
|
||||
WHERE identity_pubkey = ?1 AND relay_url = ?2 AND id = ?3",
|
||||
)
|
||||
.map_err(|e| format!("prune: prepare observer-index cascade: {e}"))?;
|
||||
let mut del_metric = tx
|
||||
.prepare(
|
||||
"DELETE FROM agent_metric_index
|
||||
WHERE identity_pubkey = ?1 AND relay_url = ?2 AND id = ?3",
|
||||
)
|
||||
.map_err(|e| format!("prune: prepare metric-index cascade: {e}"))?;
|
||||
|
||||
let mut seen = std::collections::HashSet::new();
|
||||
for (id, _, _) in &candidates {
|
||||
if !seen.insert(id.as_str()) {
|
||||
continue;
|
||||
}
|
||||
let orphaned = del_event
|
||||
.execute(params![identity_pk, relay_url, id])
|
||||
.map_err(|e| format!("prune: gc orphaned event: {e}"))?;
|
||||
if orphaned > 0 {
|
||||
total_events_deleted += orphaned;
|
||||
del_observer
|
||||
.execute(params![identity_pk, relay_url, id])
|
||||
.map_err(|e| format!("prune: cascade observer index: {e}"))?;
|
||||
del_metric
|
||||
.execute(params![identity_pk, relay_url, id])
|
||||
.map_err(|e| format!("prune: cascade metric index: {e}"))?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
tx.commit()
|
||||
.map_err(|e| format!("prune: commit batch: {e}"))?;
|
||||
|
||||
if (batch_len as i64) < PRUNE_BATCH_SIZE {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(total_events_deleted)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "prune_tests.rs"]
|
||||
mod prune_tests;
|
||||
465
desktop/src-tauri/src/archive/prune_tests.rs
Normal file
465
desktop/src-tauri/src/archive/prune_tests.rs
Normal file
@@ -0,0 +1,465 @@
|
||||
//! Behavior tests for the observer-retention prune worker.
|
||||
//!
|
||||
//! `#[path]`-included from `prune.rs`. `super::*` brings the prune API into
|
||||
//! scope; `super::super::{store, retention}` reach the neighbouring store
|
||||
//! mutators and the retention accessors. These drive the synchronous prune core
|
||||
//! (`prune_observer_frames`, `run_prune_if_due`) directly against a temp-file DB
|
||||
//! — the async single-flight/trigger wiring needs a Tauri `AppState` and is
|
||||
//! exercised by the live lane, not here.
|
||||
|
||||
use super::super::{retention, store};
|
||||
use super::*;
|
||||
use tempfile::NamedTempFile;
|
||||
|
||||
const ID: &str = "idpk";
|
||||
const RELAY: &str = "wss://r";
|
||||
const OWNER: &str = "owner_p";
|
||||
|
||||
/// Open a fresh archive DB (full schema + every migration incl. M4).
|
||||
fn fresh(db: &NamedTempFile) -> Connection {
|
||||
store::open_archive_db(db.path()).expect("open_archive_db must succeed")
|
||||
}
|
||||
|
||||
/// Insert an observer frame (kind 24200) with a matching owner_p scope row at a
|
||||
/// given `archived_at`, plus its `observer_channel_index` status row — the full
|
||||
/// shape the ingest path writes for one frame.
|
||||
fn seed_observer(conn: &Connection, id: &str, archived_at: i64) {
|
||||
store::upsert_archived_event(conn, ID, RELAY, id, 24200, "agent", 1000, "{}", archived_at)
|
||||
.unwrap();
|
||||
store::upsert_event_scope(conn, ID, RELAY, id, "owner_p", OWNER, archived_at).unwrap();
|
||||
store::upsert_observer_channel_index(conn, ID, RELAY, id, Some("ch"), 1000).unwrap();
|
||||
}
|
||||
|
||||
fn count(conn: &Connection, table: &str) -> i64 {
|
||||
conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0))
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn event_exists(conn: &Connection, id: &str) -> bool {
|
||||
conn.query_row(
|
||||
"SELECT COUNT(*) FROM archived_events WHERE id = ?1",
|
||||
params![id],
|
||||
|r| r.get::<_, i64>(0),
|
||||
)
|
||||
.unwrap()
|
||||
> 0
|
||||
}
|
||||
|
||||
// ── Cutoff correctness ─────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn prune_deletes_only_frames_older_than_cutoff() {
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
seed_observer(&conn, "old", 100);
|
||||
seed_observer(&conn, "recent", 1_000);
|
||||
|
||||
// Cutoff 500: "old" (100 < 500) goes, "recent" (1000) stays.
|
||||
let deleted = prune_observer_frames(&conn, ID, RELAY, 500).unwrap();
|
||||
assert_eq!(deleted, 1, "only the pre-cutoff frame's event is deleted");
|
||||
assert!(!event_exists(&conn, "old"));
|
||||
assert!(event_exists(&conn, "recent"));
|
||||
// Derived index rows cascaded for the deleted event only.
|
||||
assert_eq!(count(&conn, "observer_channel_index"), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prune_uses_archived_at_not_event_created_at() {
|
||||
// A late-arriving OLD event (small created_at, but archived_at is recent)
|
||||
// must be KEPT — age is "kept locally for N days", the scope-row archived_at.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
// created_at is fixed at 1000 in seed_observer; vary archived_at.
|
||||
seed_observer(&conn, "late-old", 9_000); // archived recently
|
||||
seed_observer(&conn, "genuinely-old", 100); // archived long ago
|
||||
|
||||
let deleted = prune_observer_frames(&conn, ID, RELAY, 5_000).unwrap();
|
||||
assert_eq!(deleted, 1);
|
||||
assert!(
|
||||
event_exists(&conn, "late-old"),
|
||||
"a late-arriving old event archived recently must be kept (archived_at basis)"
|
||||
);
|
||||
assert!(!event_exists(&conn, "genuinely-old"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prune_keeps_future_dated_frames() {
|
||||
// A future-dated archived_at is never before any realistic cutoff, so it is
|
||||
// never pruned — the archived_at basis is immune to a bogus-future stamp.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
seed_observer(&conn, "future", i64::MAX / 2);
|
||||
|
||||
let deleted = prune_observer_frames(&conn, ID, RELAY, 10_000).unwrap();
|
||||
assert_eq!(deleted, 0);
|
||||
assert!(event_exists(&conn, "future"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prune_never_touches_non_observer_kinds() {
|
||||
// A kind-44200 metric row with an ancient archived_at must survive: the
|
||||
// candidate scan filters `e.kind = 24200`, so metrics are kept forever.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
store::upsert_archived_event(&conn, ID, RELAY, "metric", 44200, "agent", 1, "{}", 1).unwrap();
|
||||
store::upsert_event_scope(&conn, ID, RELAY, "metric", "owner_p", OWNER, 1).unwrap();
|
||||
seed_observer(&conn, "obs", 1);
|
||||
|
||||
let deleted = prune_observer_frames(&conn, ID, RELAY, 5_000).unwrap();
|
||||
assert_eq!(deleted, 1, "only the observer frame is deleted");
|
||||
assert!(event_exists(&conn, "metric"), "kind 44200 is kept forever");
|
||||
assert!(!event_exists(&conn, "obs"));
|
||||
}
|
||||
|
||||
// ── Multi-scope events ─────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn prune_keeps_event_still_held_by_another_scope() {
|
||||
// An observer frame re-scoped later under a second scope: the OLD owner_p
|
||||
// scope row (archived long ago) is a prune candidate, but the RECENT
|
||||
// channel_h scope row is not. Deleting only the old scope row must NOT
|
||||
// delete the event — it still has a within-window scope proof.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
store::upsert_archived_event(&conn, ID, RELAY, "multi", 24200, "agent", 1000, "{}", 100)
|
||||
.unwrap();
|
||||
store::upsert_event_scope(&conn, ID, RELAY, "multi", "owner_p", OWNER, 100).unwrap();
|
||||
store::upsert_event_scope(&conn, ID, RELAY, "multi", "channel_h", "chan", 1000).unwrap();
|
||||
|
||||
let deleted = prune_observer_frames(&conn, ID, RELAY, 500).unwrap();
|
||||
assert_eq!(
|
||||
deleted, 0,
|
||||
"event retains a within-window channel_h scope, so no event delete"
|
||||
);
|
||||
assert!(event_exists(&conn, "multi"));
|
||||
// The pre-cutoff owner_p scope row was still removed; the recent one survives.
|
||||
assert_eq!(count(&conn, "archived_event_scopes"), 1);
|
||||
}
|
||||
|
||||
// ── Orphan cascade ───────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn prune_cascades_both_derived_indexes() {
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
// Observer frame with an observer_channel_index row.
|
||||
seed_observer(&conn, "obs", 100);
|
||||
// A kind-24200 frame that ALSO has a stale agent_metric_index row planted on
|
||||
// its id (defensive: the cascade must clear both indexes for a pruned id).
|
||||
let metric_row = super::super::metric_store::AgentMetricIndexRow::from_payload(
|
||||
"{}", "obs", "agent", 1000, 100,
|
||||
);
|
||||
super::super::metric_store::insert_metric_index_row(&conn, ID, RELAY, &metric_row).unwrap();
|
||||
assert_eq!(count(&conn, "agent_metric_index"), 1);
|
||||
|
||||
prune_observer_frames(&conn, ID, RELAY, 500).unwrap();
|
||||
assert_eq!(
|
||||
count(&conn, "observer_channel_index"),
|
||||
0,
|
||||
"observer_channel_index must be cascaded (the whole-scan GC does not cover it)"
|
||||
);
|
||||
assert_eq!(
|
||||
count(&conn, "agent_metric_index"),
|
||||
0,
|
||||
"agent_metric_index must be cascaded in the same transaction"
|
||||
);
|
||||
}
|
||||
|
||||
// ── Batching ───────────────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn prune_clears_more_than_one_batch() {
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
let n = super::PRUNE_BATCH_SIZE + 250; // spills into a second batch
|
||||
for i in 0..n {
|
||||
seed_observer(&conn, &format!("e{i}"), 100);
|
||||
}
|
||||
let deleted = prune_observer_frames(&conn, ID, RELAY, 500).unwrap();
|
||||
assert_eq!(
|
||||
deleted as i64, n,
|
||||
"every old frame across batches is deleted"
|
||||
);
|
||||
assert_eq!(count(&conn, "archived_events"), 0);
|
||||
assert_eq!(count(&conn, "archived_event_scopes"), 0);
|
||||
assert_eq!(count(&conn, "observer_channel_index"), 0);
|
||||
}
|
||||
|
||||
// ── 24h gate ─────────────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn run_prune_if_due_skips_within_24h_window() {
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
seed_observer(&conn, "old", 100);
|
||||
retention::set_observer_retention_days(&conn, 1).unwrap();
|
||||
|
||||
let now = 10_000_000;
|
||||
// Last success 1 hour ago → gated, no prune.
|
||||
retention::set_prune_last_success_at(&conn, now - 3_600).unwrap();
|
||||
run_prune_if_due(&conn, now).unwrap();
|
||||
assert!(
|
||||
event_exists(&conn, "old"),
|
||||
"prune within 24h must be skipped"
|
||||
);
|
||||
assert_eq!(
|
||||
retention::get_prune_last_success_at(&conn).unwrap(),
|
||||
Some(now - 3_600),
|
||||
"the gated pass must not advance the timestamp"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn run_prune_if_due_runs_after_24h_and_stamps_timestamp() {
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
seed_observer(&conn, "old", 100);
|
||||
retention::set_observer_retention_days(&conn, 1).unwrap();
|
||||
|
||||
let now = 10_000_000;
|
||||
// Last success 25 hours ago → eligible.
|
||||
retention::set_prune_last_success_at(&conn, now - 25 * 3_600).unwrap();
|
||||
run_prune_if_due(&conn, now).unwrap();
|
||||
assert!(
|
||||
!event_exists(&conn, "old"),
|
||||
"eligible prune deletes stale frames"
|
||||
);
|
||||
assert_eq!(
|
||||
retention::get_prune_last_success_at(&conn).unwrap(),
|
||||
Some(now),
|
||||
"a clean pass stamps the success timestamp"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn run_prune_if_due_runs_on_never_pruned_db() {
|
||||
// No prune_last_success_at row → treated as "never pruned" → eligible.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
seed_observer(&conn, "old", 100);
|
||||
retention::set_observer_retention_days(&conn, 1).unwrap();
|
||||
assert_eq!(retention::get_prune_last_success_at(&conn).unwrap(), None);
|
||||
|
||||
run_prune_if_due(&conn, 10_000_000).unwrap();
|
||||
assert!(!event_exists(&conn, "old"));
|
||||
assert!(retention::get_prune_last_success_at(&conn)
|
||||
.unwrap()
|
||||
.is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn run_prune_if_due_stamps_even_when_nothing_deleted() {
|
||||
// A clean pass that deletes nothing still advances the gate, so a burst of
|
||||
// post-commit triggers is bounded to one scan per day.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
seed_observer(&conn, "recent", 9_999_999);
|
||||
retention::set_observer_retention_days(&conn, 1).unwrap();
|
||||
|
||||
let now = 10_000_000;
|
||||
run_prune_if_due(&conn, now).unwrap();
|
||||
assert!(
|
||||
event_exists(&conn, "recent"),
|
||||
"nothing is old enough to prune"
|
||||
);
|
||||
assert_eq!(
|
||||
retention::get_prune_last_success_at(&conn).unwrap(),
|
||||
Some(now),
|
||||
"an empty-but-clean pass still stamps the gate"
|
||||
);
|
||||
}
|
||||
|
||||
// ── Global gate covers every partition (not just the active one) ────────────────
|
||||
|
||||
/// Seed an observer frame in an explicit `(identity, relay)` partition. The
|
||||
/// partition-scoped analogue of `seed_observer`, used to prove one due pass
|
||||
/// prunes every stored partition — not just the active identity/relay.
|
||||
fn seed_observer_in(conn: &Connection, id_pk: &str, relay: &str, id: &str, archived_at: i64) {
|
||||
store::upsert_archived_event(
|
||||
conn,
|
||||
id_pk,
|
||||
relay,
|
||||
id,
|
||||
24200,
|
||||
"agent",
|
||||
1000,
|
||||
"{}",
|
||||
archived_at,
|
||||
)
|
||||
.unwrap();
|
||||
store::upsert_event_scope(conn, id_pk, relay, id, "owner_p", OWNER, archived_at).unwrap();
|
||||
store::upsert_observer_channel_index(conn, id_pk, relay, id, Some("ch"), 1000).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn run_prune_if_due_prunes_every_partition_before_stamping() {
|
||||
// The gate is database-global, so one due pass must prune ALL stored
|
||||
// partitions — including an inactive/historical one that no active sync is
|
||||
// ever scoped to — before stamping success. Two partitions, both with an
|
||||
// expired frame; neither is "the active pair" the pass resolves (it no
|
||||
// longer resolves one).
|
||||
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();
|
||||
|
||||
run_prune_if_due(&conn, 10_000_000).unwrap();
|
||||
assert!(
|
||||
!event_exists(&conn, "old_a"),
|
||||
"partition A's expired frame must be pruned"
|
||||
);
|
||||
assert!(
|
||||
!event_exists(&conn, "old_b"),
|
||||
"partition B's expired frame must be pruned in the same pass"
|
||||
);
|
||||
assert_eq!(count(&conn, "archived_events"), 0);
|
||||
assert_eq!(count(&conn, "observer_channel_index"), 0);
|
||||
assert!(retention::get_prune_last_success_at(&conn)
|
||||
.unwrap()
|
||||
.is_some());
|
||||
}
|
||||
|
||||
#[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.
|
||||
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();
|
||||
|
||||
let result = run_prune_if_due(&conn, 10_000_000);
|
||||
assert!(
|
||||
result.is_err(),
|
||||
"a failing partition must surface the error"
|
||||
);
|
||||
assert_eq!(
|
||||
retention::get_prune_last_success_at(&conn).unwrap(),
|
||||
None,
|
||||
"a partial worklist must not stamp success — the pass retries"
|
||||
);
|
||||
}
|
||||
|
||||
// ── Stale-backfill orphan race (conditional observer-index insert) ──────────────
|
||||
|
||||
#[test]
|
||||
fn observer_index_insert_is_skipped_when_event_is_gone() {
|
||||
// The stale-backfill race: the TS driver snapshots an unindexed id, the
|
||||
// frame is pruned, then the driver calls back to index it. The conditional
|
||||
// INSERT ... SELECT must NOT resurrect an index row for the absent event.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
// No archived_events row for "gone" — simulate it already pruned away.
|
||||
store::upsert_observer_channel_index(&conn, ID, RELAY, "gone", Some("ch"), 1000).unwrap();
|
||||
assert_eq!(
|
||||
count(&conn, "observer_channel_index"),
|
||||
0,
|
||||
"no index row may be created for an event that no longer exists"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn observer_index_insert_succeeds_when_event_present() {
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
store::upsert_archived_event(&conn, ID, RELAY, "here", 24200, "agent", 1000, "{}", 100)
|
||||
.unwrap();
|
||||
store::upsert_observer_channel_index(&conn, ID, RELAY, "here", Some("ch"), 1000).unwrap();
|
||||
assert_eq!(count(&conn, "observer_channel_index"), 1);
|
||||
}
|
||||
|
||||
// ── First-archive ON CONFLICT DO NOTHING pin ────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn observer_index_insert_is_idempotent_on_conflict() {
|
||||
// Re-indexing an already-indexed frame is a no-op (first write wins), even
|
||||
// now that the insert is gated on the canonical event.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
store::upsert_archived_event(&conn, ID, RELAY, "e", 24200, "agent", 1000, "{}", 100).unwrap();
|
||||
store::upsert_observer_channel_index(&conn, ID, RELAY, "e", Some("first"), 1000).unwrap();
|
||||
store::upsert_observer_channel_index(&conn, ID, RELAY, "e", Some("second"), 2000).unwrap();
|
||||
let ch: Option<String> = conn
|
||||
.query_row(
|
||||
"SELECT channel_id FROM observer_channel_index WHERE id = 'e'",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
ch.as_deref(),
|
||||
Some("first"),
|
||||
"first write wins on PK conflict"
|
||||
);
|
||||
}
|
||||
|
||||
// ── Batch atomicity (forced error between delete stages) ────────────────────────
|
||||
|
||||
#[test]
|
||||
fn prune_batch_rolls_back_fully_when_a_cascade_fails() {
|
||||
// The delete stages run in one transaction: scope delete → event GC →
|
||||
// observer-index cascade → metric-index cascade. If a later stage errors,
|
||||
// the whole batch must roll back so no event is left with orphaned index
|
||||
// rows (or, conversely, an index row orphaned by a half-applied delete).
|
||||
// Force the failure by removing agent_metric_index, so the metric cascade —
|
||||
// the last stage — errors after the earlier deletes have run in-tx.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
seed_observer(&conn, "old", 100);
|
||||
conn.execute_batch("DROP TABLE agent_metric_index").unwrap();
|
||||
|
||||
let result = prune_observer_frames(&conn, ID, RELAY, 500);
|
||||
assert!(result.is_err(), "the failed metric cascade must surface");
|
||||
// Every earlier delete in the batch is rolled back: the frame, its scope
|
||||
// row, and its observer index row all survive intact.
|
||||
assert!(
|
||||
event_exists(&conn, "old"),
|
||||
"a mid-batch cascade failure must roll back the event delete"
|
||||
);
|
||||
assert_eq!(count(&conn, "archived_event_scopes"), 1);
|
||||
assert_eq!(count(&conn, "observer_channel_index"), 1);
|
||||
}
|
||||
|
||||
// ── EXPLAIN-shape against the real worker SQL ───────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn prune_candidate_scan_seeks_archived_at_through_the_index() {
|
||||
// Pinned against the SHIPPED PRUNE_CANDIDATE_SQL constant (the same one the
|
||||
// worker prepares), so the covering-index access path and the production
|
||||
// query cannot drift.
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db);
|
||||
|
||||
let plan: Vec<String> = conn
|
||||
.prepare(&format!("EXPLAIN QUERY PLAN {PRUNE_CANDIDATE_SQL}"))
|
||||
.unwrap()
|
||||
.query_map(params![ID, RELAY, 0_i64, PRUNE_BATCH_SIZE], |r| {
|
||||
r.get::<_, String>(3)
|
||||
})
|
||||
.unwrap()
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.unwrap();
|
||||
|
||||
let scope_step = plan
|
||||
.iter()
|
||||
.find(|d| d.contains(retention::SCOPE_AGE_INDEX_NAME))
|
||||
.unwrap_or_else(|| {
|
||||
panic!(
|
||||
"prune scan must use {}; plan was {plan:?}",
|
||||
retention::SCOPE_AGE_INDEX_NAME
|
||||
)
|
||||
});
|
||||
assert!(
|
||||
scope_step.contains("archived_at<?"),
|
||||
"scope step must range-seek archived_at, got: {scope_step}"
|
||||
);
|
||||
assert!(
|
||||
!plan.iter().any(|d| d.contains("USE TEMP B-TREE")),
|
||||
"age-first index must remove the ORDER-BY temp b-tree; plan was {plan:?}"
|
||||
);
|
||||
}
|
||||
@@ -31,6 +31,13 @@ pub const DEFAULT_OBSERVER_RETENTION_DAYS: i64 = 30;
|
||||
/// admitting any realistic user choice.
|
||||
pub const MAX_RETENTION_DAYS: i64 = 36_500;
|
||||
|
||||
/// `archive_meta` key holding the unix-seconds timestamp of the last successful
|
||||
/// prune pass, stored as its decimal text. The prune worker reads this to
|
||||
/// enforce its ≥24h gate and writes it after a clean pass (see
|
||||
/// [`super::prune`]). Kept here beside the retention window because
|
||||
/// `retention.rs` owns every `archive_meta` accessor.
|
||||
pub const PRUNE_LAST_SUCCESS_AT_KEY: &str = "prune_last_success_at";
|
||||
|
||||
// ── Schema (created by migration M4, not the base SCHEMA) ─────────────────────
|
||||
|
||||
/// `archive_meta`. Created inside M4 under `BEGIN IMMEDIATE` (see
|
||||
@@ -148,6 +155,42 @@ pub fn set_observer_retention_days(conn: &Connection, days: i64) -> Result<(), S
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ── Prune timestamp (archive_meta accessor) ─────────────────────────────────────
|
||||
|
||||
/// Read the unix-seconds timestamp of the last successful prune pass, or `None`
|
||||
/// when no prune has ever completed (the seeded DB has no such row). A `None`
|
||||
/// means "never pruned", which the ≥24h gate treats as immediately eligible.
|
||||
pub fn get_prune_last_success_at(conn: &Connection) -> Result<Option<i64>, String> {
|
||||
let raw: Option<String> = conn
|
||||
.query_row(
|
||||
"SELECT value FROM archive_meta WHERE key = ?1",
|
||||
params![PRUNE_LAST_SUCCESS_AT_KEY],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.optional()
|
||||
.map_err(|e| format!("read prune last success at: {e}"))?;
|
||||
|
||||
match raw {
|
||||
Some(s) => s
|
||||
.parse::<i64>()
|
||||
.map(Some)
|
||||
.map_err(|e| format!("prune last success at not an integer ({s:?}): {e}")),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
/// Record `at` (unix seconds) as the timestamp of the most recent successful
|
||||
/// prune pass. A single idempotent upsert, atomic under autocommit.
|
||||
pub fn set_prune_last_success_at(conn: &Connection, at: i64) -> Result<(), String> {
|
||||
conn.execute(
|
||||
"INSERT INTO archive_meta (key, value) VALUES (?1, ?2)
|
||||
ON CONFLICT (key) DO UPDATE SET value = excluded.value",
|
||||
params![PRUNE_LAST_SUCCESS_AT_KEY, at.to_string()],
|
||||
)
|
||||
.map_err(|e| format!("set prune last success at: {e}"))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ── Size accounting ─────────────────────────────────────────────────────────────
|
||||
|
||||
/// Collect physical (file) and logical (page) size figures for the archive DB.
|
||||
|
||||
@@ -383,58 +383,3 @@ fn test_m4_two_conn_first_open_race_neither_times_out_and_marks_once() {
|
||||
DEFAULT_OBSERVER_RETENTION_DAYS
|
||||
);
|
||||
}
|
||||
|
||||
// ── Prune-candidate access path (index shape) ──────────────────────────────────
|
||||
|
||||
/// The Phase-2 prune-candidate scan (per `PLANS/ARCHIVE_RETENTION_PLAN.md`
|
||||
/// line 40): the single global observer window filters `identity + relay +
|
||||
/// archived_at < cutoff`, joins `archived_events` for `kind = 24200`, and
|
||||
/// selects the scope-row PK so the bounded `DELETE` can materialize candidates.
|
||||
/// The scope side constrains neither `scope_type` nor `scope_value`.
|
||||
const PRUNE_CANDIDATE_SQL: &str = "
|
||||
SELECT s.id, s.scope_type, s.scope_value
|
||||
FROM archived_event_scopes s
|
||||
JOIN archived_events e
|
||||
ON e.identity_pubkey = s.identity_pubkey
|
||||
AND e.relay_url = s.relay_url
|
||||
AND e.id = s.id
|
||||
WHERE s.identity_pubkey = ?1
|
||||
AND s.relay_url = ?2
|
||||
AND s.archived_at < ?3
|
||||
AND e.kind = 24200
|
||||
LIMIT 1000
|
||||
";
|
||||
|
||||
/// The scope-age index must let the real Phase-2 prune query range-seek
|
||||
/// `archived_at` directly rather than fall back to a bare identity/relay seek
|
||||
/// plus a temp b-tree. The index is this PR's deliverable, so its access path
|
||||
/// is pinned here even though the prune worker lands in Phase 2.
|
||||
#[test]
|
||||
fn test_prune_candidate_scan_seeks_archived_at_through_the_index() {
|
||||
let db = NamedTempFile::new().unwrap();
|
||||
let conn = fresh(&db); // runs M4, creating the scope-age index
|
||||
|
||||
let plan: Vec<String> = conn
|
||||
.prepare(&format!("EXPLAIN QUERY PLAN {PRUNE_CANDIDATE_SQL}"))
|
||||
.unwrap()
|
||||
.query_map(params![ID, RELAY, 0_i64], |r| r.get::<_, String>(3))
|
||||
.unwrap()
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.unwrap();
|
||||
|
||||
let scope_step = plan
|
||||
.iter()
|
||||
.find(|d| d.contains(SCOPE_AGE_INDEX_NAME))
|
||||
.unwrap_or_else(|| panic!("prune scan must use {SCOPE_AGE_INDEX_NAME}; plan was {plan:?}"));
|
||||
// The planner range-seeks archived_at through the index (equality on the
|
||||
// two leading identity/relay keys, then the age bound) — not a bare
|
||||
// identity/relay seek that would leave age to a scan / temp b-tree.
|
||||
assert!(
|
||||
scope_step.contains("archived_at<?"),
|
||||
"scope step must range-seek archived_at, got: {scope_step}"
|
||||
);
|
||||
assert!(
|
||||
!plan.iter().any(|d| d.contains("USE TEMP B-TREE")),
|
||||
"age-first index must remove the ORDER-BY temp b-tree; plan was {plan:?}"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -697,9 +697,21 @@ pub fn read_archived_events(
|
||||
/// Will's (a) ruling (2026-07-08), while ensuring the row exists so a re-run
|
||||
/// is a no-op.
|
||||
///
|
||||
/// **Conditional on the canonical event.** The insert is an `INSERT ... SELECT`
|
||||
/// gated on the matching `archived_events` row still existing, so a derived
|
||||
/// index row can never be created for an event that is absent — the
|
||||
/// stale-backfill orphan race. The TS backfill driver reads a snapshot of
|
||||
/// unindexed rows, decrypts each asynchronously, then calls back here; a prune
|
||||
/// pass (or a concurrent GC) can delete the underlying observer frame in that
|
||||
/// window, and an unconditional insert would resurrect an index row pointing at
|
||||
/// a deleted event — an orphan the prune's candidate-scoped GC already ran past.
|
||||
/// Gating the insert on `EXISTS(archived_events)` closes that window at the
|
||||
/// write itself rather than relying on a later repair pass.
|
||||
///
|
||||
/// Idempotent: if the (identity, relay, id) PK already exists the row is left
|
||||
/// unchanged (INSERT OR IGNORE). Called both at ingest time (new frames) and
|
||||
/// from the one-shot backfill for existing archived rows.
|
||||
/// unchanged (`ON CONFLICT DO NOTHING`). Called both at ingest time (new
|
||||
/// frames, where the canonical row was just inserted in the same transaction)
|
||||
/// and from the one-shot backfill for existing archived rows.
|
||||
pub fn upsert_observer_channel_index(
|
||||
conn: &Connection,
|
||||
identity_pubkey: &str,
|
||||
@@ -709,9 +721,14 @@ pub fn upsert_observer_channel_index(
|
||||
created_at: i64,
|
||||
) -> Result<(), String> {
|
||||
conn.execute(
|
||||
"INSERT OR IGNORE INTO observer_channel_index
|
||||
"INSERT INTO observer_channel_index
|
||||
(identity_pubkey, relay_url, id, channel_id, created_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5)",
|
||||
SELECT ?1, ?2, ?3, ?4, ?5
|
||||
WHERE EXISTS (
|
||||
SELECT 1 FROM archived_events
|
||||
WHERE identity_pubkey = ?1 AND relay_url = ?2 AND id = ?3
|
||||
)
|
||||
ON CONFLICT (identity_pubkey, relay_url, id) DO NOTHING",
|
||||
params![identity_pubkey, relay_url, event_id, channel_id, created_at],
|
||||
)
|
||||
.map_err(|e| format!("failed to upsert observer_channel_index: {e}"))?;
|
||||
|
||||
@@ -744,6 +744,9 @@ fn test_upsert_observer_channel_index_non_null_is_idempotent() {
|
||||
let pk = "owner";
|
||||
let relay = "wss://r";
|
||||
|
||||
// The insert is gated on the canonical event existing (stale-backfill
|
||||
// orphan race fix), so seed it first.
|
||||
upsert_archived_event(&conn, pk, relay, "ev1", 24200, "agent", 1000, "{}", 0).unwrap();
|
||||
upsert_observer_channel_index(&conn, pk, relay, "ev1", Some("ch-abc"), 1000).unwrap();
|
||||
// Second call with same PK → INSERT OR IGNORE, no error, row unchanged.
|
||||
upsert_observer_channel_index(&conn, pk, relay, "ev1", Some("ch-abc"), 2000).unwrap();
|
||||
@@ -775,6 +778,8 @@ fn test_upsert_observer_channel_index_null_channel_id_is_idempotent() {
|
||||
let relay = "wss://r";
|
||||
|
||||
// Write a NULL channel_id status row (unscoped / decrypt-failed frame).
|
||||
// Gated on the canonical event existing, so seed it first.
|
||||
upsert_archived_event(&conn, pk, relay, "ev-null", 24200, "agent", 500, "{}", 0).unwrap();
|
||||
upsert_observer_channel_index(&conn, pk, relay, "ev-null", None, 500).unwrap();
|
||||
// Re-run → no-op.
|
||||
upsert_observer_channel_index(&conn, pk, relay, "ev-null", None, 600).unwrap();
|
||||
|
||||
@@ -594,6 +594,15 @@ pub async fn start_archive_sync(
|
||||
run_sync(&io, reload, events, cancel).await;
|
||||
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(())
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user