jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(archive): add observer-frame retention schema and gated DB adapter (#5719)

Browse Source 
The local-save archive (`~/.buzz/archive/archive.db`) grows without
bound — observer frames (kind 24200) are 99.95% of it by bytes (~1.3M
rows, 75–375 MB/day). This lays the schema and process-wide foundations
for bounding that growth: a single global retention window for observer
frames, kept in `archive_meta`, plus the gated DB adapter every later
phase builds on. The prune worker that actually deletes lands in a
follow-up; nothing here removes data.

## What this adds

- **Migration M4** (`add_archive_meta`, crash-safe): creates the
`archive_meta` k/v table and the `archived_at`-covering scope-age index
used by the future prune scan, and seeds `observer_retention_days=30`.
Runs under one `BEGIN IMMEDIATE` with an in-lock marker recheck and the
marker written last, so a crash before COMMIT rolls back every object
and the next open re-runs from scratch. Fails closed on an
externally-created `archive_meta` present without the marker.
- **Process init barrier + gated `ArchiveDb` adapter** owning every
production open. All subscription-mutation commands and the archive sync
task route their DB work through `ArchiveDb::with_conn`, which awaits
the init barrier once. This also fixes a pre-existing bug where
`create_save_subscription` ran blocking DB work on the async runtime.
- **Commands**: `get_observer_retention_days` /
`set_observer_retention_days` (fail-closed — rejects days `< 1` or `>
36500` before writing) and `archive_size_stats` (physical file bytes for
the main DB + `-wal` sidecar, plus `page_size` / `page_count` /
`freelist_count` — PRAGMAs and file metadata only, no payload scans).

## Scope

Observer frames (24200) are the only kind with a retention setting.
NIP-AM metrics (44200) and every other archived kind are kept
indefinitely with no retention state — no policy table, no per-kind or
per-subscription configurability. This is the simplified design ruled in
over the earlier per-subscription × per-kind approach: bounding observer
frames alone captures essentially all the value.

---------

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>
Co-authored-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>
This commit is contained in:
Will Pfleger
2026-08-21 13:35:33 -04:00
committed by GitHub
parent 6039fed565
commit fc2ce6728b
10 changed files with 1440 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
desktop/src-tauri/src/app_state.rs
View File

@@ -129,6 +129,7 @@ pub struct AppState {
/// bounded and letting a later leave correctly flip the channel back to
/// `is_member=false`.
pub pending_owned_channels: Mutex<std::collections::HashSet<(String, String)>>,
pub archive_db: crate::archive::ArchiveDb,
}
/// Parse the `BUZZ_PRIVATE_KEY` env var into identity keys. `Some` means the
@@ -219,9 +220,7 @@ pub fn build_app_state() -> AppState {
huddle_audio: Default::default(),
app_handle: Mutex::new(None),
media_proxy_port: AtomicU16::new(0),
prevent_sleep: Arc::new(Mutex::new(
crate::prevent_sleep::PreventSleepState::default(),
)),
prevent_sleep: Default::default(),
keyring_locked: AtomicBool::new(false),
identity_lost: AtomicBool::new(false),
reset_failed: AtomicBool::new(false),
@@ -232,6 +231,7 @@ pub fn build_app_state() -> AppState {
#[cfg(feature = "mesh-llm")]
mesh_coordinator: AsyncMutex::new(None),
pending_owned_channels: Mutex::new(std::collections::HashSet::new()),
archive_db: crate::archive::ArchiveDb::default(),
}
}

199
desktop/src-tauri/src/archive/archive_db.rs Normal file
View File

@@ -0,0 +1,199 @@
//! Process-wide gated adapter for the local archive SQLite database.
//!
//! Two coupled guarantees, both required by plan v3 (decisions 1–2):
//!
//! 1. **Init barrier.** Exactly one blocking task opens the DB the first time
//! and completes every schema migration (including M4, whose index build
//! over Will's 1.3M-row archive is not free). Every production open — every
//! Tauri archive command and all future startup/prune work — `await`s that
//! result before touching its own connection. The barrier is independent of
//! identity/relay resolution: the archive DB is a single per-nest file
//! (identity is a row column, not part of the path), and the path resolves
//! from `nest_dir()`, which is fixed early in `setup()` before the async
//! workspace relay override settles. This satisfies Thufir's binding
//! condition that the barrier cover workspace/identity timing and the
//! globally-mounted observer archive producer.
//!
//! 2. **Maintenance lock.** A shared `RwLock` whose read guard is held for the
//! full lifetime of every ordinary connection (acquired before the blocking
//! dispatch, released only after the closure returns and its connection is
//! dropped). The Phase-4 "reclaim space" conversion will take the *write*
//! guard for its sole-connection `VACUUM` sequence; nothing else may hold a
//! live connection while that runs. Phase 1 only ever takes the read guard,
//! but the lock lives here so the write path has a home.
//!
//! M4 additionally keeps its own `BEGIN IMMEDIATE` + in-lock recheck for crash
//! and cross-process safety — the in-process barrier serializes this process's
//! opens, but a second OS process (or a direct `open_archive_db` in tests) can
//! still race the first open. The two mechanisms are complementary, not
//! redundant: the barrier is startup orchestration, the immediate-lock is
//! durability.
use std::path::PathBuf;
use rusqlite::Connection;
use tokio::sync::{OnceCell, RwLock};
use super::store;
use crate::managed_agents::nest_dir;
/// A hook run once on the blocking pool at the start of the single init task.
/// Test-only: lets a test count initializations and hold the winner long
/// enough to prove concurrent callers await it. Never set in production.
#[cfg(test)]
type InitHook = std::sync::Arc<dyn Fn() + Send + Sync>;
/// Test-only overrides so the barrier and guard-lifetime contracts can be
/// exercised without a real nest: a fixed DB path in place of `nest_dir()` and
/// an optional init hook. `Default` leaves this `None`, so production always
/// resolves the path from the nest and runs no hook.
#[cfg(test)]
struct TestSeam {
path: PathBuf,
on_init: Option<InitHook>,
}
/// Gated owner of every production archive DB connection. Lives in
/// [`crate::app_state::AppState`]; commands call [`ArchiveDb::with_conn`].
#[derive(Default)]
pub struct ArchiveDb {
/// Set to `()` once the first open (which runs all migrations incl. M4)
/// succeeds. A failed init is NOT cached — the next caller retries — so a
/// transient error (e.g. a briefly unavailable external volume) does not
/// wedge the archive for the process lifetime.
init: OnceCell<()>,
/// Maintenance lock. Ordinary connections hold the read guard for their
/// whole lifetime; the Phase-4 conversion holds the write guard.
maintenance: RwLock<()>,
/// Test-only path/hook overrides; always `None` in production.
#[cfg(test)]
test_seam: Option<TestSeam>,
}
impl ArchiveDb {
/// Resolve the archive DB path. Production resolves from the nest
/// directory; a test seam (when present) supplies a fixed path so the
/// barrier can be exercised without a real nest. Errors only when the nest
/// cannot be resolved (fatal for archive access, same as the former
/// `open_db`).
fn db_path(&self) -> Result<PathBuf, String> {
#[cfg(test)]
if let Some(seam) = &self.test_seam {
return Ok(seam.path.clone());
}
let nest = nest_dir().ok_or("cannot resolve nest directory for archive")?;
Ok(nest.join("archive").join("archive.db"))
}
/// The init hook, if a test installed one; always `None` in production.
#[cfg(test)]
fn init_hook(&self) -> Option<InitHook> {
self.test_seam.as_ref().and_then(|s| s.on_init.clone())
}
/// Complete the one-time init: open the DB once on the blocking pool,
/// running `SCHEMA` + all migrations (incl. M4), then drop the connection.
/// Concurrent callers await the same single execution. Idempotent and
/// cheap after the first success (the cached `()` short-circuits).
async fn ensure_initialized(&self) -> Result<(), String> {
let path = self.db_path()?;
#[cfg(test)]
let hook = self.init_hook();
self.init
.get_or_try_init(|| async {
tokio::task::spawn_blocking(move || {
// Test hook runs at the very start of the single init task,
// before the migration opens the DB — this is where a test
// holds the winner past the busy timeout to prove ordinary
// callers await it. No-op in production.
#[cfg(test)]
if let Some(hook) = hook {
hook();
}
// Opening runs every migration; the connection exists only
// to complete them behind the barrier, so drop it here.
let conn = store::open_archive_db(&path)?;
drop(conn);
Ok::<(), String>(())
})
.await
.map_err(|e| format!("archive init task failed: {e}"))?
})
.await
.map(|_| ())
}
/// Warm the init barrier without running a query. Called once from
/// `setup()` so the first-open migration cost (M4's index build over a
/// large archive) is paid at startup rather than blocking a user's first
/// archive command. A failure here is non-fatal — the first real
/// [`with_conn`](Self::with_conn) caller retries and surfaces the error.
pub async fn warm_init(&self) -> Result<(), String> {
self.ensure_initialized().await
}
/// Run `task` against a fresh archive connection on the blocking pool.
///
/// Ordering: await the init barrier → acquire the maintenance read guard →
/// dispatch the blocking closure with its own connection. The read guard is
/// held across the `.await` on the blocking join, so it is released only
/// after `task` returns and the connection it borrowed has dropped — the
/// guard-lifetime contract the Phase-4 write path depends on.
pub async fn with_conn<T, F>(&self, task: F) -> Result<T, String>
where
T: Send + 'static,
F: FnOnce(&Connection) -> Result<T, String> + Send + 'static,
{
self.ensure_initialized().await?;
let path = self.db_path()?;
let _guard = self.maintenance.read().await;
tokio::task::spawn_blocking(move || {
let conn = store::open_archive_db(&path)?;
task(&conn)
})
.await
.map_err(|e| format!("archive db task failed: {e}"))?
}
}
#[cfg(test)]
impl ArchiveDb {
/// Build an adapter bound to a fixed DB path (no nest required), so the
/// barrier and guard-lifetime contracts can be exercised in isolation.
fn with_test_path(path: PathBuf) -> Self {
Self {
init: OnceCell::new(),
maintenance: RwLock::new(()),
test_seam: Some(TestSeam {
path,
on_init: None,
}),
}
}
/// Build an adapter bound to a fixed path whose single initialization runs
/// `hook` first — used to count initializations and to hold the init task
/// open across the concurrent-caller window.
fn with_test_hook(path: PathBuf, hook: InitHook) -> Self {
Self {
init: OnceCell::new(),
maintenance: RwLock::new(()),
test_seam: Some(TestSeam {
path,
on_init: Some(hook),
}),
}
}
/// Whether the maintenance WRITE guard can be taken right now. A live
/// `with_conn` connection holds the read guard, so this returns `false`
/// while any ordinary connection is open and `true` once all have dropped —
/// exactly the signal the Phase-4 sole-connection VACUUM will gate on.
fn maintenance_write_available(&self) -> bool {
self.maintenance.try_write().is_ok()
}
}
#[cfg(test)]
#[path = "archive_db_tests.rs"]
mod archive_db_tests;

246
desktop/src-tauri/src/archive/archive_db_tests.rs Normal file
View File

@@ -0,0 +1,246 @@
//! Behavior tests for the [`ArchiveDb`] init barrier and maintenance-lock
//! guard lifetime — the two contracts Phase 1 introduced and Thufir's pass-1
//! review required to be pinned directly (not via raw SQLite contention).
//!
//! These race PRODUCTION-shaped `with_conn` callers through the real
//! `OnceCell`/`RwLock` orchestration, using the `#[cfg(test)]` path/hook seam
//! on `ArchiveDb` to make timing deterministic instead of relying on a large
//! on-disk fixture or wall-clock sleeps to approach the 5s busy timeout.
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use tempfile::TempDir;
/// A one-shot latch that blocks a blocking-pool thread until the test releases
/// it. Deterministic stand-in for "the M4 winner holds init open past the busy
/// timeout": the init/closure thread parks here, the test observes the frozen
/// state, then releases. `Mutex` + `Condvar` are `Sync`, so an `Arc<Latch>`
/// captured by the `Send + Sync` init hook type-checks.
struct Latch {
open: Mutex<bool>,
cv: Condvar,
}
impl Latch {
fn new() -> Arc<Self> {
Arc::new(Self {
open: Mutex::new(false),
cv: Condvar::new(),
})
}
/// Block until [`release`](Self::release) is called (returns immediately if
/// already released).
fn wait(&self) {
let mut open = self.open.lock().unwrap();
while !*open {
open = self.cv.wait(open).unwrap();
}
}
fn release(&self) {
*self.open.lock().unwrap() = true;
self.cv.notify_all();
}
}
/// Poll `cond` on the async runtime until it holds or `timeout` elapses.
/// Panics on timeout so a broken barrier surfaces as a failure, never a hang.
async fn await_until(what: &str, timeout: Duration, cond: impl Fn() -> bool) {
let deadline = Instant::now() + timeout;
while !cond() {
assert!(Instant::now() < deadline, "timed out waiting for {what}");
tokio::time::sleep(Duration::from_millis(5)).await;
}
}
/// An archive DB path inside a fresh temp dir. The dir is returned so the
/// caller keeps it alive for the whole test (dropping it deletes the file).
fn temp_db() -> (TempDir, std::path::PathBuf) {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("archive").join("archive.db");
(dir, path)
}
/// The barrier serializes first-open: exactly one initialization runs while
/// every concurrent `with_conn` caller awaits it, and no ordinary connection
/// opens until that initialization has completed.
///
/// The init hook holds the single init task open on a latch. While it is held
/// we prove no `with_conn` caller has opened its connection (`open_count == 0`)
/// — impossible if callers bypassed the `OnceCell` and opened independently.
/// Releasing the latch lets init finish; all callers then complete, exactly
/// one initialization ran, and every open happened after init.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_first_open_barrier_serializes_init_and_defers_opens() {
let (_dir, path) = temp_db();
let init_count = Arc::new(AtomicUsize::new(0));
let open_count = Arc::new(AtomicUsize::new(0));
let release = Latch::new();
let hook = {
let init_count = Arc::clone(&init_count);
let release = Arc::clone(&release);
Arc::new(move || {
// Runs once, at the head of the single init task. Record the
// initialization, then park so the test can inspect the frozen
// pre-open state.
init_count.fetch_add(1, Ordering::SeqCst);
release.wait();
}) as InitHook
};
let db = Arc::new(ArchiveDb::with_test_hook(path, hook));
// Trigger the single init and wait until it is provably in flight (hook
// ran) and parked on the latch.
let warm = {
let db = Arc::clone(&db);
tokio::spawn(async move { db.warm_init().await })
};
await_until("init to start", Duration::from_secs(10), || {
init_count.load(Ordering::SeqCst) == 1
})
.await;
// Fan out production-shaped callers while init is held. Each records that
// its task was scheduled (`entered`) and that its closure actually opened a
// connection (`open_count`).
let entered = Arc::new(AtomicUsize::new(0));
let callers: Vec<_> = (0..4)
.map(|_| {
let db = Arc::clone(&db);
let open_count = Arc::clone(&open_count);
let entered = Arc::clone(&entered);
tokio::spawn(async move {
entered.fetch_add(1, Ordering::SeqCst);
db.with_conn(move |conn| {
open_count.fetch_add(1, Ordering::SeqCst);
// Touch the migrated schema to prove a usable connection.
conn.query_row("SELECT COUNT(*) FROM archive_meta", [], |r| {
r.get::<_, i64>(0)
})
.map_err(|e| e.to_string())
})
.await
})
})
.collect();
// All four caller tasks are scheduled and running before we judge the
// barrier: they have entered `with_conn` and can only be parked on the
// init `OnceCell`. Without the barrier they would instead open independent
// connections here and bump `open_count` while init is still held.
await_until("callers to be scheduled", Duration::from_secs(10), || {
entered.load(Ordering::SeqCst) == 4
})
.await;
tokio::time::sleep(Duration::from_millis(100)).await;
assert_eq!(
open_count.load(Ordering::SeqCst),
0,
"no ordinary connection may open until initialization completes"
);
// Let init finish; every caller now completes against the migrated DB.
release.release();
assert!(warm.await.unwrap().is_ok(), "warm init must succeed");
for caller in callers {
assert!(
caller.await.unwrap().is_ok(),
"every with_conn must succeed"
);
}
assert_eq!(
init_count.load(Ordering::SeqCst),
1,
"exactly one initialization ran behind the barrier"
);
assert_eq!(
open_count.load(Ordering::SeqCst),
4,
"all callers opened, and only after init"
);
}
/// The maintenance read guard lives for the FULL lifetime of a `with_conn`
/// connection: a write-lock contender cannot enter until the closure returns
/// and its connection has dropped. This is the invariant the Phase-4
/// sole-connection VACUUM depends on.
///
/// A `with_conn` closure parks on a latch while holding its connection; a
/// separate task contends for the maintenance write guard. While the closure
/// is parked the writer must be blocked. Releasing the closure — which returns
/// and drops the connection — lets the writer finally acquire the guard.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_with_conn_read_guard_blocks_writer_until_connection_drops() {
let (_dir, path) = temp_db();
let db = Arc::new(ArchiveDb::with_test_path(path));
db.warm_init().await.expect("init must succeed");
let in_closure = Arc::new(AtomicUsize::new(0));
let hold = Latch::new();
// A live connection: the closure parks holding it (and thus the read
// guard) until released.
let work = {
let db = Arc::clone(&db);
let in_closure = Arc::clone(&in_closure);
let hold = Arc::clone(&hold);
tokio::spawn(async move {
db.with_conn(move |_conn| {
in_closure.fetch_add(1, Ordering::SeqCst);
hold.wait();
Ok(())
})
.await
})
};
await_until(
"closure to hold the connection",
Duration::from_secs(10),
|| in_closure.load(Ordering::SeqCst) == 1,
)
.await;
// A genuine write-lock contender.
let write_entered = Arc::new(AtomicUsize::new(0));
let writer = {
let db = Arc::clone(&db);
let write_entered = Arc::clone(&write_entered);
tokio::spawn(async move {
let _w = db.maintenance.write().await;
write_entered.fetch_add(1, Ordering::SeqCst);
})
};
// While the connection is held, the writer must not have entered, and the
// write guard must be unavailable.
tokio::time::sleep(Duration::from_millis(200)).await;
assert_eq!(
write_entered.load(Ordering::SeqCst),
0,
"writer must block while a with_conn connection is live"
);
assert!(
!db.maintenance_write_available(),
"write guard is unavailable while the read guard is held"
);
// Release the closure: it returns and its connection drops, releasing the
// read guard so the writer can proceed.
hold.release();
assert!(work.await.unwrap().is_ok(), "held with_conn must succeed");
writer.await.unwrap();
assert_eq!(
write_entered.load(Ordering::SeqCst),
1,
"writer enters once the connection has dropped"
);
assert!(
db.maintenance_write_available(),
"write guard is free again after the connection drops"
);
}

295
desktop/src-tauri/src/archive/mod.rs
View File

@@ -18,12 +18,16 @@
//! == agent) is applied fail-closed.
mod agent_usage;
mod archive_db;
mod metric_store;
mod pipeline;
pub mod retention;
pub mod store;
mod store_migrations;
pub mod sync;
pub use archive_db::ArchiveDb;
use pipeline::{commit_archive, plan_archive, query_buckets};
use nostr::Event;
@@ -32,7 +36,6 @@ use serde::{Deserialize, Serialize};
use tauri::State;
use crate::app_state::AppState;
use crate::managed_agents::nest_dir;
use crate::relay::{query_relay, relay_ws_url_with_override};
// ── Constants ───────────────────────────────────────────────────────────────
@@ -43,10 +46,20 @@ const OBSERVER_FRAME_TELEMETRY: &str = "telemetry";
// ── DB helpers ───────────────────────────────────────────────────────────────
fn open_db() -> Result<Connection, String> {
let nest = nest_dir().ok_or("cannot resolve nest directory for archive")?;
let db_path = nest.join("archive").join("archive.db");
store::open_archive_db(&db_path)
/// Warm the archive DB init barrier on a background task, now that the nest
/// exists, so the first-open schema migration cost (M4's index build over a
/// large archive) is paid at startup rather than blocking a user's first
/// archive command. The globally-mounted observer archive producer also
/// `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.
pub fn spawn_warm_init(app: tauri::AppHandle) {
tauri::async_runtime::spawn(async move {
use tauri::Manager;
if let Err(error) = app.state::<AppState>().archive_db.warm_init().await {
eprintln!("buzz-desktop: archive DB init deferred: {error}");
}
});
}
fn identity_pubkey(state: &AppState) -> Result<String, String> {
@@ -61,19 +74,6 @@ fn now_secs() -> i64 {
.as_secs() as i64
}
async fn run_archive_db_task<T, F>(task: F) -> Result<T, String>
where
T: Send + 'static,
F: FnOnce(&Connection) -> Result<T, String> + Send + 'static,
{
tokio::task::spawn_blocking(move || {
let conn = open_db()?;
task(&conn)
})
.await
.map_err(|e| format!("spawn_blocking failed: {e}"))?
}
// ── Scope type ───────────────────────────────────────────────────────────────
/// The three supported archive scope discriminants.
@@ -170,10 +170,10 @@ pub(crate) async fn archive_candidates(
// ── Phase 1: plan (blocking SQLite) ─────────────────────────────────────
let plan_identity_pk = identity_pk.clone();
let plan_relay_url = relay_url.clone();
let plan = run_archive_db_task(move |conn| {
plan_archive(candidates, &plan_identity_pk, &plan_relay_url, conn)
})
.await?;
let plan = state
.archive_db
.with_conn(move |conn| plan_archive(candidates, &plan_identity_pk, &plan_relay_url, conn))
.await?;
// ── Phase 2: relay queries (async) ───────────────────────────────────────
let bucket_results = query_buckets(plan.buckets, state).await;
@@ -186,19 +186,21 @@ pub(crate) async fn archive_candidates(
};
let commit_identity_pk = identity_pk.clone();
let commit_relay_url = relay_url.clone();
run_archive_db_task(move |conn| {
commit_archive(
bucket_results,
plan.ephemeral,
plan.pre_dropped,
&commit_identity_pk,
&commit_relay_url,
&owner_keys,
now,
conn,
)
})
.await
state
.archive_db
.with_conn(move |conn| {
commit_archive(
bucket_results,
plan.ephemeral,
plan.pre_dropped,
&commit_identity_pk,
&commit_relay_url,
&owner_keys,
now,
conn,
)
})
.await
}
/// Validate an ephemeral observer frame (kind 24200) against ALL local rules.
@@ -337,16 +339,21 @@ pub async fn create_save_subscription(
let kinds_json =
serde_json::to_string(&kinds).map_err(|e| format!("failed to serialize kinds: {e}"))?;
let conn = open_db()?;
store::upsert_save_subscription(
&conn,
&identity_pk,
&relay_url,
scope_type.as_str(),
&scope_value,
&kinds_json,
now,
)?;
let scope_type_str = scope_type.as_str().to_string();
state
.archive_db
.with_conn(move |conn| {
store::upsert_save_subscription(
conn,
&identity_pk,
&relay_url,
&scope_type_str,
&scope_value,
&kinds_json,
now,
)
})
.await?;
sync_state.notify_subscriptions_changed().await;
Ok(())
}
@@ -452,10 +459,12 @@ pub async fn merge_save_subscription_kinds(
let relay_url = relay_ws_url_with_override(&state);
let now = now_secs();
let owner_pk = identity_pk.clone();
run_archive_db_task(move |conn| {
store::merge_owner_p_kinds(conn, &identity_pk, &relay_url, &owner_pk, kind, now)
})
.await?;
state
.archive_db
.with_conn(move |conn| {
store::merge_owner_p_kinds(conn, &identity_pk, &relay_url, &owner_pk, kind, now)
})
.await?;
sync_state.notify_subscriptions_changed().await;
Ok(())
}
@@ -487,10 +496,12 @@ pub async fn remove_save_subscription_kind(
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
let owner_pk = identity_pk.clone();
run_archive_db_task(move |conn| {
store::remove_owner_p_kind(conn, &identity_pk, &relay_url, &owner_pk, kind)
})
.await?;
state
.archive_db
.with_conn(move |conn| {
store::remove_owner_p_kind(conn, &identity_pk, &relay_url, &owner_pk, kind)
})
.await?;
sync_state.notify_subscriptions_changed().await;
Ok(())
}
@@ -504,7 +515,9 @@ pub async fn list_save_subscriptions(
) -> Result<Vec<store::SaveSubscription>, String> {
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
run_archive_db_task(move |conn| store::list_save_subscriptions(conn, &identity_pk, &relay_url))
state
.archive_db
.with_conn(move |conn| store::list_save_subscriptions(conn, &identity_pk, &relay_url))
.await
}
@@ -523,16 +536,18 @@ pub async fn delete_save_subscription(
) -> Result<bool, String> {
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
let removed = run_archive_db_task(move |conn| {
store::delete_save_subscription(
conn,
&identity_pk,
&relay_url,
scope_type.as_str(),
&scope_value,
)
})
.await?;
let removed = state
.archive_db
.with_conn(move |conn| {
store::delete_save_subscription(
conn,
&identity_pk,
&relay_url,
scope_type.as_str(),
&scope_value,
)
})
.await?;
if removed {
sync_state.notify_subscriptions_changed().await;
}
@@ -564,18 +579,20 @@ pub async fn read_archived_observer_events_for_channel(
) -> Result<Vec<String>, String> {
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
run_archive_db_task(move |conn| {
store::read_archived_observer_events_for_channel(
conn,
&identity_pk,
&relay_url,
&channel_id,
before_created_at,
before_id.as_deref(),
limit.unwrap_or(DEFAULT_READ_LIMIT),
)
})
.await
state
.archive_db
.with_conn(move |conn| {
store::read_archived_observer_events_for_channel(
conn,
&identity_pk,
&relay_url,
&channel_id,
before_created_at,
before_id.as_deref(),
limit.unwrap_or(DEFAULT_READ_LIMIT),
)
})
.await
}
// ── index_observer_channel_id ─────────────────────────────────────────────────
@@ -595,20 +612,22 @@ pub async fn index_observer_channel_id(
) -> Result<(), String> {
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
run_archive_db_task(move |conn| {
for entry in &entries {
store::upsert_observer_channel_index(
conn,
&identity_pk,
&relay_url,
&entry.event_id,
entry.channel_id.as_deref(),
entry.created_at,
)?;
}
Ok(())
})
.await
state
.archive_db
.with_conn(move |conn| {
for entry in &entries {
store::upsert_observer_channel_index(
conn,
&identity_pk,
&relay_url,
&entry.event_id,
entry.channel_id.as_deref(),
entry.created_at,
)?;
}
Ok(())
})
.await
}
/// A single (event_id, channel_id?, created_at) record used by
@@ -639,18 +658,20 @@ pub async fn read_unindexed_observer_rows(
) -> Result<Vec<RawObserverRow>, String> {
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
run_archive_db_task(move |conn| {
let rows = store::read_unindexed_observer_rows(conn, &identity_pk, &relay_url)?;
Ok(rows
.into_iter()
.map(|(id, raw_json, created_at)| RawObserverRow {
id,
raw_json,
created_at,
})
.collect())
})
.await
state
.archive_db
.with_conn(move |conn| {
let rows = store::read_unindexed_observer_rows(conn, &identity_pk, &relay_url)?;
Ok(rows
.into_iter()
.map(|(id, raw_json, created_at)| RawObserverRow {
id,
raw_json,
created_at,
})
.collect())
})
.await
}
/// Wire type returned by `read_unindexed_observer_rows`.
@@ -695,20 +716,22 @@ pub async fn read_archived_events(
let relay_url = relay_ws_url_with_override(&state);
let scope_type_str = scope_type.as_str().to_string();
let read_limit = limit.unwrap_or(DEFAULT_READ_LIMIT);
run_archive_db_task(move |conn| {
store::read_archived_events(
conn,
&identity_pk,
&relay_url,
&scope_type_str,
&scope_value,
kinds.as_deref(),
before_created_at,
before_id.as_deref(),
read_limit,
)
})
.await
state
.archive_db
.with_conn(move |conn| {
store::read_archived_events(
conn,
&identity_pk,
&relay_url,
&scope_type_str,
&scope_value,
kinds.as_deref(),
before_created_at,
before_id.as_deref(),
read_limit,
)
})
.await
}
// ── get_agent_usage_series ───────────────────────────────────────────────────
@@ -797,10 +820,48 @@ pub async fn get_agent_usage_series(
) -> Result<agent_usage::AgentUsageSeries, String> {
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
run_archive_db_task(move |conn| agent_usage_series(conn, &identity_pk, &relay_url, &request))
state
.archive_db
.with_conn(move |conn| agent_usage_series(conn, &identity_pk, &relay_url, &request))
.await
}
// ── Retention configuration commands ──────────────────────────────────────────
/// Read the global observer-frame (kind 24200) retention window, in days. Every
/// other archived kind — NIP-AM metrics and any custom subscription — is kept
/// indefinitely and has no setting.
#[tauri::command]
pub async fn get_observer_retention_days(state: State<'_, AppState>) -> Result<i64, String> {
state
.archive_db
.with_conn(retention::get_observer_retention_days)
.await
}
/// Set the global observer-frame retention window, in days. Fail-closed: the
/// store layer rejects zero, negative, or out-of-range values (see
/// [`retention::validate_days`]).
#[tauri::command]
pub async fn set_observer_retention_days(
state: State<'_, AppState>,
days: i64,
) -> Result<(), String> {
state
.archive_db
.with_conn(move |conn| retention::set_observer_retention_days(conn, days))
.await
}
/// Physical (file) and logical (page) size accounting for the archive DB, for
/// the Settings size readout. PRAGMAs + file metadata only — no payload scans.
#[tauri::command]
pub async fn archive_size_stats(
state: State<'_, AppState>,
) -> Result<retention::ArchiveSizeStats, String> {
state.archive_db.with_conn(retention::size_stats).await
}
// ── Tests ────────────────────────────────────────────────────────────────────
#[cfg(test)]

216
desktop/src-tauri/src/archive/retention.rs Normal file
View File

@@ -0,0 +1,216 @@
//! Local archive retention configuration + size accounting.
//!
//! v4 (Will's ruling, 2026-08-19): retention is a single global setting — how
//! many days observer frames (kind 24200) are kept locally. NIP-AM metrics
//! (44200) and every other archived kind are kept indefinitely with no
//! retention machinery. The setting lives as one row in the `archive_meta` k/v
//! table (`observer_retention_days`), seeded by migration M4.
//!
//! This module owns the `archive_meta` schema, the scope-age index used by the
//! Phase-2 prune scan, the get/set accessors for the observer window, and the
//! PRAGMA-based size readout. The prune worker itself lands in Phase 2.
//!
//! Kept in a sibling file (not `store.rs`) to respect the 1000-line gate, per
//! the existing `metric_store.rs` / `pipeline.rs` / `store_migrations.rs`
//! precedent.
use rusqlite::{params, Connection, OptionalExtension};
// ── Constants ────────────────────────────────────────────────────────────────
/// `archive_meta` key holding the observer-frame retention window, in days,
/// stored as its decimal text.
pub const OBSERVER_RETENTION_DAYS_KEY: &str = "observer_retention_days";
/// Default rolling window for observer frames (Will's ruling: "~14–30"; 30 is
/// the shipped default, trivially changeable). Seeded into `archive_meta` by M4.
pub const DEFAULT_OBSERVER_RETENTION_DAYS: i64 = 30;
/// Upper bound on the retention window (~100 years). Guards against a day count
/// large enough to overflow `archived_at` cutoff arithmetic while still
/// admitting any realistic user choice.
pub const MAX_RETENTION_DAYS: i64 = 36_500;
// ── Schema (created by migration M4, not the base SCHEMA) ─────────────────────
/// `archive_meta`. Created inside M4 under `BEGIN IMMEDIATE` (see
/// `store_migrations::migrate_add_archive_meta`). Plain `CREATE TABLE`
/// (no `IF NOT EXISTS`): M4 creates it once on a DB that provably lacks it (the
/// fail-closed guard rejects a pre-existing one), so the clause would be dead
/// weight. Holds the observer-retention window and the Phase-2 prune timestamp.
pub(super) const ARCHIVE_META_SCHEMA: &str = "
CREATE TABLE archive_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
";
/// Covering index for the Phase-2 retention prune candidate scan. V4 prunes
/// one global observer window, so the scan filters `(identity_pubkey,
/// relay_url, archived_at < cutoff)` and joins `archived_events` for
/// `kind = 24200` — it does NOT constrain `scope_type`/`scope_value`. Age
/// therefore has to lead the non-equality keys: putting `archived_at`
/// immediately after the two identity/relay equality keys lets SQLite
/// range-seek `archived_at < ?` directly (`SEARCH ... USING COVERING INDEX
/// ... (identity_pubkey=? AND relay_url=? AND archived_at<?)`). The v3 order
/// buried `archived_at` behind the two now-unconstrained scope keys, so the
/// planner fell back to a bare identity/relay seek plus a temp b-tree.
///
/// The trailing `id, scope_type, scope_value` complete the scope-row primary
/// key, so the index both covers the scan (no table lookup) and hands Phase 2
/// the exact keys its bounded `DELETE` needs to materialize candidates.
///
/// Built in M4 so the one-time cost over the existing 1.3M-row archive is paid
/// behind the init barrier rather than on the prune hot path. M4
/// `DROP INDEX IF EXISTS`es under this name and recreates it unconditionally,
/// so a fresh build is always correct — plain `CREATE INDEX`, no `IF NOT
/// EXISTS`.
pub(super) const SCOPE_AGE_INDEX_DDL: &str = "
CREATE INDEX idx_archived_event_scopes_age
ON archived_event_scopes
(identity_pubkey, relay_url, archived_at, id, scope_type, scope_value);
";
/// Name of the scope-age index, shared by the DDL and M4's unconditional
/// drop-and-rebuild.
pub(super) const SCOPE_AGE_INDEX_NAME: &str = "idx_archived_event_scopes_age";
// ── Types ─────────────────────────────────────────────────────────────────────
/// Physical and logical size accounting for the archive DB file, returned by
/// `archive_size_stats`. All figures come from cheap PRAGMAs and file metadata
/// — no payload scans.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ArchiveSizeStats {
/// On-disk size of the main DB file, in bytes (from filesystem metadata).
pub main_file_bytes: i64,
/// On-disk size of the `-wal` sidecar file, in bytes; `0` when absent.
pub wal_file_bytes: i64,
/// SQLite page size, in bytes (`PRAGMA page_size`).
pub page_size: i64,
/// Total pages in the main DB (`PRAGMA page_count`).
pub page_count: i64,
/// Free (reclaimable) pages on the freelist (`PRAGMA freelist_count`).
pub freelist_count: i64,
}
// ── Validation ─────────────────────────────────────────────────────────────────
/// Fail-closed validation for the observer retention window: a day count in
/// `1..=MAX_RETENTION_DAYS`. Zero, negatives, and out-of-range values are
/// rejected so the setting can never encode a non-expiring "0 days" or an
/// overflowing cutoff.
pub fn validate_days(days: i64) -> Result<(), String> {
if !(1..=MAX_RETENTION_DAYS).contains(&days) {
return Err(format!(
"retention days must be between 1 and {MAX_RETENTION_DAYS}; got {days}"
));
}
Ok(())
}
// ── Observer retention window (archive_meta accessor) ───────────────────────────
/// Read the observer-frame retention window (days). Returns the seeded default
/// when the row is somehow absent (older DB opened before M4 seeded it, or an
/// externally cleared row) so the setting always resolves to a bounded window
/// rather than silently becoming Forever.
pub fn get_observer_retention_days(conn: &Connection) -> Result<i64, String> {
let raw: Option<String> = conn
.query_row(
"SELECT value FROM archive_meta WHERE key = ?1",
params![OBSERVER_RETENTION_DAYS_KEY],
|row| row.get(0),
)
.optional()
.map_err(|e| format!("read observer retention days: {e}"))?;
match raw {
Some(s) => s
.parse::<i64>()
.map_err(|e| format!("observer retention days not an integer ({s:?}): {e}")),
None => Ok(DEFAULT_OBSERVER_RETENTION_DAYS),
}
}
/// Set the observer-frame retention window (days). Fail-closed: rejects an
/// out-of-range value before writing (see [`validate_days`]). A single
/// idempotent upsert, atomic under autocommit.
pub fn set_observer_retention_days(conn: &Connection, days: i64) -> Result<(), String> {
validate_days(days)?;
conn.execute(
"INSERT INTO archive_meta (key, value) VALUES (?1, ?2)
ON CONFLICT (key) DO UPDATE SET value = excluded.value",
params![OBSERVER_RETENTION_DAYS_KEY, days.to_string()],
)
.map_err(|e| format!("set observer retention days: {e}"))?;
Ok(())
}
// ── Size accounting ─────────────────────────────────────────────────────────────
/// Collect physical (file) and logical (page) size figures for the archive DB.
/// PRAGMAs only — no row or payload scans. The main DB file path is read from
/// the connection itself (`PRAGMA database_list`), and the `-wal` sidecar is
/// measured by appending `-wal` to it.
pub fn size_stats(conn: &Connection) -> Result<ArchiveSizeStats, String> {
let page_size: i64 = conn
.pragma_query_value(None, "page_size", |row| row.get(0))
.map_err(|e| format!("read page_size: {e}"))?;
let page_count: i64 = conn
.pragma_query_value(None, "page_count", |row| row.get(0))
.map_err(|e| format!("read page_count: {e}"))?;
let freelist_count: i64 = conn
.pragma_query_value(None, "freelist_count", |row| row.get(0))
.map_err(|e| format!("read freelist_count: {e}"))?;
// `PRAGMA database_list` yields (seq, name, file) rows; the `main` schema's
// `file` is the on-disk DB path (empty for a `:memory:` DB). File sizes are
// read from filesystem metadata rather than page arithmetic so WAL frames
// not yet checkpointed into the main file are accounted for separately.
let main_path: Option<String> = conn
.query_row(
"SELECT file FROM pragma_database_list WHERE name = 'main'",
[],
|row| row.get::<_, String>(0),
)
.optional()
.map_err(|e| format!("read database_list: {e}"))?
.filter(|p| !p.is_empty());
let (main_file_bytes, wal_file_bytes) = match main_path {
Some(p) => {
let main = std::path::PathBuf::from(&p);
(file_len(&main), file_len(&wal_path(&main)))
}
None => (0, 0),
};
Ok(ArchiveSizeStats {
main_file_bytes,
wal_file_bytes,
page_size,
page_count,
freelist_count,
})
}
/// Size of a file in bytes, or `0` when it does not exist / cannot be stat'd.
/// A missing `-wal` (fully checkpointed DB) is the normal case, not an error.
fn file_len(path: &std::path::Path) -> i64 {
std::fs::metadata(path).map(|m| m.len() as i64).unwrap_or(0)
}
/// The `-wal` sidecar path for a main DB file (`archive.db` → `archive.db-wal`).
fn wal_path(db_path: &std::path::Path) -> std::path::PathBuf {
let mut name = db_path.as_os_str().to_os_string();
name.push("-wal");
std::path::PathBuf::from(name)
}
// ── Tests ───────────────────────────────────────────────────────────────────
#[cfg(test)]
#[path = "retention_tests.rs"]
mod retention_tests;

440
desktop/src-tauri/src/archive/retention_tests.rs Normal file
View File

@@ -0,0 +1,440 @@
//! Behavior tests for the observer-retention setting, the size readout, and the
//! M4 migration.
//!
//! Kept in a sibling file so `retention.rs` stays under the 1000-line gate;
//! `#[path]`-included from there. `super::*` brings the retention API (and its
//! `rusqlite::{params, Connection}` imports) into scope; `super::super::store`
//! reaches the neighbouring subscription mutators and the base `SCHEMA`.
use super::super::store;
use super::*;
use std::path::Path;
use std::sync::{Arc, Barrier};
use tempfile::NamedTempFile;
const ID: &str = "idpk";
const RELAY: &str = "wss://r";
const OWNER: &str = "owner_p";
/// Open a fresh archive DB (runs the full schema + every migration incl. M4).
fn fresh(db: &NamedTempFile) -> Connection {
store::open_archive_db(db.path()).expect("open_archive_db must succeed")
}
/// Build a legacy DB that has the base schema and M1–M3 markers but NOT M4,
/// so the next `open_archive_db` pends only the retention migration. This
/// isolates the M4 first-open race from the separately-tested M1–M3 chain.
fn build_pre_m4_db(path: &Path) {
let conn = Connection::open(path).unwrap();
conn.pragma_update(None, "busy_timeout", 5000).unwrap();
conn.pragma_update(None, "journal_mode", "WAL").unwrap();
conn.execute_batch(store::SCHEMA).unwrap();
for name in [
"add_harness_to_metric_index",
"add_cache_read_tokens",
"add_cache_write_and_pricing",
] {
conn.execute(
"INSERT OR IGNORE INTO archive_migrations (name, applied_at) VALUES (?1, 0)",
params![name],
)
.unwrap();
}
}
fn m4_marker_count(conn: &Connection) -> i64 {
conn.query_row(
"SELECT COUNT(*) FROM archive_migrations WHERE name = 'add_archive_meta'",
[],
|r| r.get(0),
)
.unwrap()
}
// ── Validation (pure) ──────────────────────────────────────────────────────────
#[test]
fn test_validate_days_accepts_one_and_max() {
assert!(validate_days(1).is_ok());
assert!(validate_days(DEFAULT_OBSERVER_RETENTION_DAYS).is_ok());
assert!(validate_days(MAX_RETENTION_DAYS).is_ok());
}
#[test]
fn test_validate_days_rejects_zero_negative_and_over_max() {
assert!(validate_days(0).is_err());
assert!(validate_days(-1).is_err());
assert!(validate_days(MAX_RETENTION_DAYS + 1).is_err());
}
// ── Observer retention get / set ────────────────────────────────────────────────
#[test]
fn test_get_observer_days_returns_seeded_default_on_fresh_db() {
let db = NamedTempFile::new().unwrap();
let conn = fresh(&db);
assert_eq!(
get_observer_retention_days(&conn).unwrap(),
DEFAULT_OBSERVER_RETENTION_DAYS
);
}
#[test]
fn test_set_observer_days_upserts_and_overwrites() {
let db = NamedTempFile::new().unwrap();
let conn = fresh(&db);
set_observer_retention_days(&conn, 14).unwrap();
assert_eq!(get_observer_retention_days(&conn).unwrap(), 14);
set_observer_retention_days(&conn, 60).unwrap();
assert_eq!(get_observer_retention_days(&conn).unwrap(), 60);
}
#[test]
fn test_set_observer_days_rejects_out_of_range_without_writing() {
let db = NamedTempFile::new().unwrap();
let conn = fresh(&db);
// Establish a known good value, then prove a rejected write leaves it.
set_observer_retention_days(&conn, 45).unwrap();
assert!(set_observer_retention_days(&conn, 0).is_err());
assert!(set_observer_retention_days(&conn, -5).is_err());
assert!(set_observer_retention_days(&conn, MAX_RETENTION_DAYS + 1).is_err());
assert_eq!(
get_observer_retention_days(&conn).unwrap(),
45,
"a rejected set must not overwrite the stored value"
);
}
#[test]
fn test_observer_days_survive_reopen() {
let db = NamedTempFile::new().unwrap();
{
let first = fresh(&db);
set_observer_retention_days(&first, 7).unwrap();
}
let second = fresh(&db);
assert_eq!(
get_observer_retention_days(&second).unwrap(),
7,
"the setting persists across opens and M4 re-run does not reset it"
);
}
// ── Size accounting ─────────────────────────────────────────────────────────────
#[test]
fn test_size_stats_reports_pages_and_main_file_bytes() {
let db = NamedTempFile::new().unwrap();
let conn = fresh(&db);
// Write enough rows that the DB grows past a single page.
for i in 0..200 {
conn.execute(
"INSERT INTO archived_events
(identity_pubkey, relay_url, id, kind, pubkey, created_at, raw_json, archived_at)
VALUES (?1, ?2, ?3, 24200, 'author', ?4, ?5, ?4)",
params![ID, RELAY, format!("e{i}"), 1000 + i, "x".repeat(256)],
)
.unwrap();
}
let stats = size_stats(&conn).unwrap();
assert!(stats.page_size > 0, "page_size is a positive PRAGMA value");
assert!(stats.page_count > 1, "multi-page DB after 200 inserts");
assert!(stats.freelist_count >= 0);
// In WAL mode the just-written pages live in the `-wal` sidecar until a
// checkpoint folds them into the main file, so the logical page total
// (page_size * page_count) is covered by the two files combined, not by
// the main file alone.
assert!(
stats.main_file_bytes + stats.wal_file_bytes >= stats.page_size * stats.page_count,
"main + wal bytes cover at least the counted pages ({} + {} >= {}*{})",
stats.main_file_bytes,
stats.wal_file_bytes,
stats.page_size,
stats.page_count
);
assert!(stats.main_file_bytes > 0, "the main DB file is on disk");
// WAL mode: the sidecar exists and carries the just-written frames.
assert!(
stats.wal_file_bytes > 0,
"the -wal sidecar is measured in WAL mode"
);
}
#[test]
fn test_size_stats_freelist_grows_after_delete() {
let db = NamedTempFile::new().unwrap();
let conn = fresh(&db);
for i in 0..200 {
conn.execute(
"INSERT INTO archived_events
(identity_pubkey, relay_url, id, kind, pubkey, created_at, raw_json, archived_at)
VALUES (?1, ?2, ?3, 24200, 'author', ?4, ?5, ?4)",
params![ID, RELAY, format!("e{i}"), 1000 + i, "x".repeat(256)],
)
.unwrap();
}
let before = size_stats(&conn).unwrap();
conn.execute("DELETE FROM archived_events", []).unwrap();
let after = size_stats(&conn).unwrap();
assert!(
after.freelist_count > before.freelist_count,
"deleted pages land on the freelist ({} > {})",
after.freelist_count,
before.freelist_count
);
}
// ── Migration M4 ──────────────────────────────────────────────────────────────
#[test]
fn test_m4_fresh_open_creates_schema_marker_and_seed() {
let db = NamedTempFile::new().unwrap();
let conn = fresh(&db);
let objects: i64 = conn
.query_row(
"SELECT COUNT(*) FROM sqlite_master
WHERE (type = 'table' AND name = 'archive_meta')
OR (type = 'index' AND name = 'idx_archived_event_scopes_age')",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(objects, 2, "one table + one index");
assert_eq!(m4_marker_count(&conn), 1, "M4 marker recorded");
assert_eq!(
get_observer_retention_days(&conn).unwrap(),
DEFAULT_OBSERVER_RETENTION_DAYS,
"the observer window is seeded to the default"
);
}
#[test]
fn test_m4_reopen_is_idempotent_and_preserves_setting() {
let db = NamedTempFile::new().unwrap();
let first = fresh(&db);
set_observer_retention_days(&first, 90).unwrap();
drop(first);
let second = fresh(&db);
assert_eq!(
m4_marker_count(&second),
1,
"exactly one marker after reopen"
);
assert_eq!(
get_observer_retention_days(&second).unwrap(),
90,
"the re-run must not re-seed over an existing value"
);
}
#[test]
fn test_m4_applies_over_a_populated_pre_m4_db() {
let db = NamedTempFile::new().unwrap();
build_pre_m4_db(db.path());
// Pre-M4 subscriptions and archived rows must survive the migration
// untouched — M4 no longer reads or seeds from subscriptions.
{
let conn = Connection::open(db.path()).unwrap();
store::upsert_save_subscription(&conn, ID, RELAY, OWNER, ID, "[24200,44200,1]", 100)
.unwrap();
}
let conn = fresh(&db);
assert_eq!(m4_marker_count(&conn), 1);
assert_eq!(
get_observer_retention_days(&conn).unwrap(),
DEFAULT_OBSERVER_RETENTION_DAYS
);
assert_eq!(
store::list_save_subscriptions(&conn, ID, RELAY)
.unwrap()
.len(),
1,
"the pre-existing subscription is untouched by M4"
);
}
#[test]
fn test_m4_preexisting_archive_meta_without_marker_fails_closed() {
let db = NamedTempFile::new().unwrap();
build_pre_m4_db(db.path());
// An `archive_meta` table present without the M4 marker is unreachable via
// shipped code — M4 runs the whole body (create table, build index, seed,
// marker) in one transactional `BEGIN IMMEDIATE`, so a crash rolls back the
// table too. The only way to reach this state is an externally-created
// table. M4 refuses to certify it: it fails closed and rolls back with no
// marker rather than adopting a table it did not build. The table's shape
// is irrelevant — its mere presence without the marker is the trigger.
{
let conn = Connection::open(db.path()).unwrap();
conn.execute_batch("CREATE TABLE archive_meta (key TEXT PRIMARY KEY);")
.unwrap();
}
assert!(
store::open_archive_db(db.path()).is_err(),
"M4 must fail closed on a pre-existing archive_meta"
);
let verify = Connection::open(db.path()).unwrap();
assert_eq!(
m4_marker_count(&verify),
0,
"no marker may certify an externally-created table"
);
}
#[test]
fn test_m4_preexisting_index_under_name_is_silently_rebuilt() {
let db = NamedTempFile::new().unwrap();
build_pre_m4_db(db.path());
// Unlike a table (which carries data and is fail-closed), an index carries
// no data, so M4 drops any index sharing the name and recreates it
// unconditionally — no shape inspection. A bogus pre-existing index under
// the name is silently replaced with the correct one and the marker lands.
{
let conn = Connection::open(db.path()).unwrap();
conn.execute_batch(&format!(
"CREATE INDEX {SCOPE_AGE_INDEX_NAME} ON archived_event_scopes (id);"
))
.unwrap();
}
let conn = fresh(&db);
assert_eq!(m4_marker_count(&conn), 1, "M4 completes after the rebuild");
// The rebuilt index has the six expected key columns in the covering order.
let mut stmt = conn
.prepare(&format!(
"SELECT name FROM pragma_index_xinfo('{SCOPE_AGE_INDEX_NAME}') \
WHERE key = 1 ORDER BY seqno"
))
.unwrap();
let keys: Vec<String> = stmt
.query_map([], |r| r.get::<_, String>(0))
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(
keys,
[
"identity_pubkey",
"relay_url",
"archived_at",
"id",
"scope_type",
"scope_value"
],
"the index was rebuilt with the age-first covering key order"
);
}
// ── Concurrency ───────────────────────────────────────────────────────────────
#[test]
fn test_m4_two_conn_first_open_race_neither_times_out_and_marks_once() {
use std::thread;
let db = NamedTempFile::new().unwrap();
let path = db.path().to_path_buf();
build_pre_m4_db(&path);
// A realistically-populated legacy DB: a multi-kind subscription plus
// archived scope rows so M4's index build touches real data.
{
let conn = Connection::open(&path).unwrap();
store::upsert_save_subscription(&conn, ID, RELAY, OWNER, ID, "[24200,44200,1]", 100)
.unwrap();
for i in 0..8 {
conn.execute(
"INSERT INTO archived_event_scopes
(identity_pubkey, relay_url, id, scope_type, scope_value, archived_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![ID, RELAY, format!("e{i}"), OWNER, ID, 1000 + i],
)
.unwrap();
}
}
let barrier = Arc::new(Barrier::new(2));
let handles: Vec<_> = (0..2)
.map(|_| {
let p = path.clone();
let b = Arc::clone(&barrier);
thread::spawn(move || {
b.wait(); // maximise the first-open race window
store::open_archive_db(&p).map(|_| ())
})
})
.collect();
for h in handles {
assert!(
h.join().unwrap().is_ok(),
"both racing opens must complete within busy_timeout"
);
}
let verify = store::open_archive_db(&path).unwrap();
assert_eq!(m4_marker_count(&verify), 1, "M4 applied exactly once");
// The seed ran once inside the winner's transaction — one meta row, default.
let meta_rows: i64 = verify
.query_row(
"SELECT COUNT(*) FROM archive_meta WHERE key = ?1",
params![OBSERVER_RETENTION_DAYS_KEY],
|r| r.get(0),
)
.unwrap();
assert_eq!(meta_rows, 1, "the observer window was seeded exactly once");
assert_eq!(
get_observer_retention_days(&verify).unwrap(),
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:?}"
);
}

138
desktop/src-tauri/src/archive/store_migrations.rs
View File

@@ -17,11 +17,13 @@ use rusqlite::{params, Connection};
///
/// Ordering: M2 (column additions) runs before M1 (index rebuild) so that
/// the M1 rebuild, which calls `insert_metric_index_row`, always operates
/// against a schema that includes the cache-read columns.
/// against a schema that includes the cache-read columns. M4 (`archive_meta`
/// + scope-age index) runs last; it is independent of M1–M3.
pub(super) fn apply_schema_migrations(conn: &Connection) -> Result<(), String> {
migrate_add_cache_read_tokens(conn)?;
migrate_add_cache_write_and_pricing(conn)?;
migrate_add_harness_to_metric_index(conn)
migrate_add_harness_to_metric_index(conn)?;
migrate_add_archive_meta(conn)
}
/// M1: add `harness TEXT` column to `agent_metric_index` and rebuild index
@@ -323,3 +325,135 @@ fn migrate_add_cache_write_and_pricing(conn: &Connection) -> Result<(), String>
Ok(())
}
/// M4: create the retention config storage — `archive_meta` (k/v state holding
/// the observer-frame retention window and the Phase-2 prune timestamp) and the
/// `archived_at`-covering scope-age index — then seed the default observer
/// retention window (`observer_retention_days = 30`).
///
/// Unlike M1–M3, M4 CANNOT use the "check marker → `BEGIN DEFERRED`" pattern.
/// On a fresh DB two connections can open concurrently (the observer- and
/// metric-archive seed hooks each `open_archive_db`), both read no marker, and
/// a `DEFERRED` transaction lets both proceed on the same snapshot — the loser
/// hits `SQLITE_BUSY_SNAPSHOT` or double-seeds. Instead M4 takes the write lock
/// up front with `BEGIN IMMEDIATE` and **rechecks the marker inside the lock**:
/// the race loser blocks on `busy_timeout`, then observes the winner's
/// committed marker and no-ops. The cheap pre-lock guard keeps steady-state
/// opens off the write lock entirely (M4 only takes it until the marker lands).
///
/// The table uses plain `CREATE TABLE` behind a fail-closed guard (a
/// pre-existing `archive_meta` with no marker is an externally-created object
/// M4 refuses to certify) and the index is dropped and recreated
/// unconditionally; the seed is `ON CONFLICT DO NOTHING`. The marker is written
/// last inside the same transaction, so a crash before COMMIT rolls back every
/// object and the next open re-runs from scratch.
fn migrate_add_archive_meta(conn: &Connection) -> Result<(), String> {
// Cheap pre-lock guard: steady-state opens (marker already present) never
// take the write lock. The marker is written last in M4's transaction, so
// its presence implies the full schema + seed committed.
if archive_meta_migration_applied(conn)? {
return Ok(());
}
conn.execute_batch("BEGIN IMMEDIATE")
.map_err(|e| format!("migration M4: begin immediate: {e}"))?;
let result = migrate_add_archive_meta_locked(conn);
if result.is_ok() {
conn.execute_batch("COMMIT")
.map_err(|e| format!("migration M4: commit: {e}"))?;
} else {
// Best-effort rollback; surface the original error to the caller.
let _ = conn.execute_batch("ROLLBACK");
}
result
}
/// M4 body, run under the `BEGIN IMMEDIATE` write lock held by
/// `migrate_add_archive_meta`.
fn migrate_add_archive_meta_locked(conn: &Connection) -> Result<(), String> {
// In-lock recheck: a concurrent first-opener may have committed the marker
// while we were blocked on the write lock. If so, this connection has
// nothing to do — the winner already created the schema and seeded.
if archive_meta_migration_applied(conn)? {
return Ok(());
}
// Fail closed on an externally-created table. M4's whole body runs inside
// one `BEGIN IMMEDIATE` transaction with the marker written last, and
// SQLite DDL is transactional — a crash anywhere rolls the whole thing
// back. So no shipped code path can leave the table present without the
// marker; the only way to reach here with it already existing is a
// hand-edited DB or a foreign tool. Rather than certify a table we did not
// create, refuse: roll back with no marker and let a corrected DB re-run.
let exists: bool = conn
.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'archive_meta'",
[],
|r| r.get::<_, i64>(0),
)
.map_err(|e| format!("migration M4: probe archive_meta: {e}"))?
> 0;
if exists {
return Err(
"migration M4: archive_meta already exists without the M4 marker — \
refusing to certify an externally-created table"
.to_string(),
);
}
conn.execute_batch(super::retention::ARCHIVE_META_SCHEMA)
.map_err(|e| format!("migration M4: create archive_meta: {e}"))?;
// Unconditionally rebuild the scope-age index. It carries no data, so a
// fresh `CREATE` is always correct; dropping any index that happens to
// share the name costs one rebuild and needs no shape inspection.
conn.execute_batch(&format!(
"DROP INDEX IF EXISTS {}",
super::retention::SCOPE_AGE_INDEX_NAME
))
.map_err(|e| format!("migration M4: drop any pre-existing scope-age index: {e}"))?;
conn.execute_batch(super::retention::SCOPE_AGE_INDEX_DDL)
.map_err(|e| format!("migration M4: create scope-age index: {e}"))?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs() as i64;
// Seed the default observer-frame retention window. `ON CONFLICT DO NOTHING`
// in the underlying insert makes this a no-op if a value is somehow already
// present — an existing choice always wins.
conn.execute(
"INSERT INTO archive_meta (key, value) VALUES (?1, ?2)
ON CONFLICT (key) DO NOTHING",
params![
super::retention::OBSERVER_RETENTION_DAYS_KEY,
super::retention::DEFAULT_OBSERVER_RETENTION_DAYS.to_string()
],
)
.map_err(|e| format!("migration M4: seed observer retention days: {e}"))?;
conn.execute(
"INSERT OR IGNORE INTO archive_migrations (name, applied_at) \
VALUES ('add_archive_meta', ?1)",
params![now],
)
.map_err(|e| format!("migration M4: record marker: {e}"))?;
Ok(())
}
/// Whether the M4 marker is present. Its presence implies the retention schema
/// and default seed committed (the marker is written last in M4's transaction).
fn archive_meta_migration_applied(conn: &Connection) -> Result<bool, String> {
let count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM archive_migrations WHERE name = 'add_archive_meta'",
[],
|r| r.get(0),
)
.map_err(|e| format!("migration M4: guard check: {e}"))?;
Ok(count > 0)
}

3
desktop/src-tauri/src/archive/store_tests.rs
View File

@@ -10,6 +10,9 @@ fn in_memory() -> Connection {
conn.pragma_update(None, "journal_mode", "WAL").unwrap();
conn.pragma_update(None, "busy_timeout", 5000).unwrap();
conn.execute_batch(SCHEMA).unwrap();
// Match production `open_archive_db`: apply every schema migration (incl.
// M4) so tests run against the same shape production connections see.
apply_schema_migrations(&conn).unwrap();
conn
}

10
desktop/src-tauri/src/archive/sync.rs
View File

@@ -280,10 +280,12 @@ impl ArchiveSyncIo for AppIo {
let state: State<'_, AppState> = self.app.state();
let identity_pk = super::identity_pubkey(&state)?;
let relay_url = crate::relay::relay_ws_url_with_override(&state);
super::run_archive_db_task(move |conn| {
super::store::list_save_subscriptions(conn, &identity_pk, &relay_url)
})
.await
state
.archive_db
.with_conn(move |conn| {
super::store::list_save_subscriptions(conn, &identity_pk, &relay_url)
})
.await
})
}

27
desktop/src-tauri/src/lib.rs
View File

@@ -314,16 +314,14 @@ pub fn run() {
// agent spawns can resolve custom/preset runtime ids without
// waiting for the frontend's discover_acp_providers call. This is
// a pure directory scan — no PATH probing, no async work.
{
let custom_dir = app_handle
.path()
.app_data_dir()
.ok()
.map(|d| d.join("custom_harnesses"));
managed_agents::custom_harnesses::warm_harness_registry_from_dir(
custom_dir.as_deref(),
);
}
let custom_harness_dir = app_handle
.path()
.app_data_dir()
.ok()
.map(|d| d.join("custom_harnesses"));
managed_agents::custom_harnesses::warm_harness_registry_from_dir(
custom_harness_dir.as_deref(),
);
// Store the AppHandle so huddle commands can emit `huddle-state-changed`
// events via `huddle::emit_huddle_state` without threading the handle
@@ -353,10 +351,7 @@ pub fn run() {
// Route mesh-llm's download progress (model weights, runtime)
// onto Tauri events so the UI can render real progress.
crate::mesh_llm::install_progress_sink(&app_handle);
let mesh_app = app_handle.clone();
tauri::async_runtime::spawn(async move {
crate::mesh_llm::start_coordinator(mesh_app).await;
});
tauri::async_runtime::spawn(crate::mesh_llm::start_coordinator(app_handle.clone()));
}
// Start the localhost media streaming proxy. Uses the shared HTTP
@@ -379,6 +374,7 @@ pub fn run() {
if let Err(error) = ensure_nest() {
eprintln!("buzz-desktop: failed to create nest: {error}");
}
archive::spawn_warm_init(app_handle.clone());
// Resolve the REPOS symlink from the persisted repos_dir BEFORE
// agents are restored below, and decide whether restore is safe.
@@ -842,6 +838,9 @@ pub fn run() {
archive::index_observer_channel_id,
archive::read_unindexed_observer_rows,
archive::get_agent_usage_series,
archive::get_observer_retention_days,
archive::set_observer_retention_days,
archive::archive_size_stats,
archive::sync::announce_archive_sync_epoch,
archive::sync::start_archive_sync,
archive::sync::stop_archive_sync,