From 3d3a3618df3451d7237f31fae250b6111134a70e Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 13:40:55 +0000 Subject: [PATCH 1/6] STA-5: Add Store::open_read_only and a shared scan lock A process that only reads the store (a desktop dashboard) can now open it with Store::open_read_only, which returns a ReadStore: a SQLite read-only (mode=ro, not immutable) connection with query_only on that never creates, migrates, backfills, reprices, chmods or checkpoints, and exposes only audited read APIs. A missing store or a schema older or newer than CURRENT_SCHEMA_VERSION is a typed ReadOnlyOpenError. Scanners now coordinate through an advisory lock in scan.lock beside the database (~/.statsai/scan.lock by default), built on std's flock/LockFileEx file locks. statsai scan and the scanning paths of statsai sync take it before opening the store, wait up to 5 seconds, then print a notice and exit 0. statsai daemon --watch takes it per background pass only, and keeps changed paths queued while another process holds it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G3EZeUTrrXQYg1NwFdgV3K --- Cargo.lock | 1 + README.md | 21 + crates/statsai-daemon/src/lib.rs | 10 +- crates/statsai-daemon/src/watch/mod.rs | 48 +- crates/statsai-daemon/src/watch/scan.rs | 73 ++- crates/statsai-daemon/src/watch/tests/scan.rs | 79 ++- crates/statsai-store/Cargo.toml | 1 + crates/statsai-store/src/lib.rs | 62 ++- crates/statsai-store/src/migrations/mod.rs | 5 +- crates/statsai-store/src/read_only.rs | 316 ++++++++++++ crates/statsai-store/src/rollups/mod.rs | 6 + crates/statsai-store/src/scan_lock.rs | 188 +++++++ crates/statsai-store/tests/read_only_store.rs | 477 ++++++++++++++++++ crates/statsai-store/tests/scan_lock.rs | 153 ++++++ crates/statsai/src/cli/daemon.rs | 18 +- crates/statsai/src/cli/dispatch.rs | 64 ++- crates/statsai/src/cli/tests/mod.rs | 1 + crates/statsai/src/cli/tests/scan_lock.rs | 65 +++ crates/statsai/src/lib.rs | 30 ++ crates/statsai/tests/scan_lock_cli.rs | 95 ++++ 20 files changed, 1667 insertions(+), 46 deletions(-) create mode 100644 crates/statsai-store/src/read_only.rs create mode 100644 crates/statsai-store/src/scan_lock.rs create mode 100644 crates/statsai-store/tests/read_only_store.rs create mode 100644 crates/statsai-store/tests/scan_lock.rs create mode 100644 crates/statsai/src/cli/tests/scan_lock.rs create mode 100644 crates/statsai/tests/scan_lock_cli.rs diff --git a/Cargo.lock b/Cargo.lock index 0b6d475..4528a73 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3032,6 +3032,7 @@ dependencies = [ "statsai-core", "statsai-pricing", "tempfile", + "thiserror 1.0.69", ] [[package]] diff --git a/README.md b/README.md index a988fab..a2a9483 100644 --- a/README.md +++ b/README.md @@ -432,6 +432,13 @@ statsai report cache --from 2026-05-01 --details Normal scans use a per-source file signature cache. `--no-cache` forces a one-off reread; `--replace` performs a destructive source rebuild. +One scan runs against a store at a time. `statsai scan`, `statsai sync` (other +than `--status`, `--verify`, and `--reset-remote`), and each pass of `statsai +daemon --watch` hold `scan.lock` beside the database (`~/.statsai/scan.lock` by +default). While another scan holds it, `scan` and `sync` wait up to 5 seconds, +then print `Another statsai scan is running (lock: …); skipping.` and exit +successfully; the daemon keeps changed files queued until the lock is free. + JSONL input is streamed with a 16 MiB per-record ceiling. Invalid or oversized records are counted and discarded through the next newline, after which parsing continues. @@ -506,6 +513,20 @@ parameters. Browser-originated requests are rejected. Sync writes must use `Content-Type: application/json` and stay below 8 MiB. +### Reading the store from another process + +An application that reads `~/.statsai/statsai.sqlite` while the CLI or daemon +writes it should open it with `statsai_store::Store::open_read_only`. That opens +SQLite read-only (`mode=ro`, not `immutable`, so WAL commits stay visible), +never creates, migrates, backfills, reprices, or changes the permissions of the +file, and returns a `ReadStore` with only the read APIs: session, daily, and +cache reports, quota and list queries, `data_version`, and +`with_read_snapshot`. A missing store, or one whose schema is older or newer +than the binary's, is a typed `ReadOnlyOpenError` the caller can act on. A +process that collects into the store should hold the scan lock, via +`statsai::try_acquire_scan_lock` or `statsai::acquire_scan_lock_with_timeout` +with `statsai::default_scan_lock_path()`. + ### Rust SDK `crates/statsai-sdk` exposes a facade for embedding collection and reporting in diff --git a/crates/statsai-daemon/src/lib.rs b/crates/statsai-daemon/src/lib.rs index f744f8c..52befbf 100644 --- a/crates/statsai-daemon/src/lib.rs +++ b/crates/statsai-daemon/src/lib.rs @@ -284,20 +284,28 @@ pub fn watch_and_serve( _store: Arc>, _device_id: &str, _auth_token: &str, + _scan_lock: Option<&std::path::Path>, ) -> Result<()> { anyhow::bail!( "daemon --watch requires the `watch` cargo feature (enable with --features watch)" ) } +/// Serves the loopback API and rescans sources as their files change. +/// +/// `scan_lock` is the lock file other scanners of this store take, normally +/// `statsai_store::scan_lock_path(store_path)`. Each rescan holds it and only +/// that long: while another process holds it the pass waits, and the changed +/// paths stay queued until the lock is free. `None` scans without it. #[cfg(feature = "watch")] pub fn watch_and_serve( addr: &str, store: Arc>, device_id: &str, auth_token: &str, + scan_lock: Option<&std::path::Path>, ) -> Result<()> { - watch::watch_and_serve(addr, store, device_id, auth_token) + watch::watch_and_serve(addr, store, device_id, auth_token, scan_lock) } fn resolve_loopback_addr(addr: &str) -> Result { diff --git a/crates/statsai-daemon/src/watch/mod.rs b/crates/statsai-daemon/src/watch/mod.rs index 7fccc09..3f4b6f6 100644 --- a/crates/statsai-daemon/src/watch/mod.rs +++ b/crates/statsai-daemon/src/watch/mod.rs @@ -32,12 +32,20 @@ const WATCH_SCAN_MAX_RETRY_DELAY: Duration = if cfg!(test) { } else { Duration::from_secs(5) }; +/// How long a pass waits before trying again while another process holds the +/// scan lock. +const WATCH_SCAN_LOCK_RETRY_DELAY: Duration = if cfg!(test) { + Duration::ZERO +} else { + Duration::from_secs(1) +}; pub fn watch_and_serve( addr: &str, store: Arc>, device_id: &str, auth_token: &str, + scan_lock: Option<&Path>, ) -> Result<()> { let bind_addr = super::resolve_loopback_addr(addr)?; let startup_executable = current_executable_stamp(); @@ -91,6 +99,7 @@ pub fn watch_and_serve( pending_changed_paths, startup_executable, bind_addr, + scan_lock, ) } #[cfg(not(target_os = "macos"))] @@ -114,6 +123,7 @@ pub fn watch_and_serve( pending_changed_paths, startup_executable, bind_addr, + scan_lock, ) } } @@ -146,11 +156,13 @@ fn run_watch_loop( pending_changed_paths: Arc>, startup_executable: Option, bind_addr: std::net::SocketAddr, + scan_lock: Option<&Path>, ) -> Result<()> { let background_store = { let store = super::lock_store(&store); store.reopen() }; + let worker_scan_lock = scan_lock.map(Path::to_path_buf); let (scan_signal_tx, scan_signal_rx) = mpsc::sync_channel(1); let worker_scan_signal_tx = scan_signal_tx.clone(); let pending_scan_paths = Arc::new(Mutex::new(PendingWatch::default())); @@ -162,6 +174,7 @@ fn run_watch_loop( .name("statsai-watch-scan".to_string()) .spawn(move || { let mut retry_delay = WATCH_SCAN_INITIAL_RETRY_DELAY; + let mut reported_scan_lock_wait = false; while scan_signal_rx.recv().is_ok() { let notice = worker_pending_scan_paths .lock() @@ -174,11 +187,12 @@ fn run_watch_loop( .read() .unwrap_or_else(|error| error.into_inner()) .clone(); - let scan_succeeded = process_background_scan( + let outcome = process_background_scan( &worker_pending_scan_paths, &worker_scan_signal_tx, notice, retry_delay, + worker_scan_lock.as_deref(), |notice| match background_store.as_ref() { Ok(store) => rescan_changed_sources( store, @@ -205,13 +219,31 @@ fn run_watch_loop( } }, ); - retry_delay = if scan_succeeded { - WATCH_SCAN_INITIAL_RETRY_DELAY - } else { - retry_delay - .saturating_mul(2) - .min(WATCH_SCAN_MAX_RETRY_DELAY) - }; + match outcome { + BackgroundScanOutcome::Completed => { + retry_delay = WATCH_SCAN_INITIAL_RETRY_DELAY; + reported_scan_lock_wait = false; + } + BackgroundScanOutcome::Failed => { + retry_delay = retry_delay + .saturating_mul(2) + .min(WATCH_SCAN_MAX_RETRY_DELAY); + reported_scan_lock_wait = false; + } + // Said once per wait: the pass retries every + // WATCH_SCAN_LOCK_RETRY_DELAY while a long scan runs. + BackgroundScanOutcome::Deferred => { + if !reported_scan_lock_wait { + if let Some(lock) = worker_scan_lock.as_deref() { + eprintln!( + "daemon: another statsai scan is running (lock: {}); watching resumes when it finishes", + lock.display() + ); + } + reported_scan_lock_wait = true; + } + } + } } }) .context("start background scan worker")?; diff --git a/crates/statsai-daemon/src/watch/scan.rs b/crates/statsai-daemon/src/watch/scan.rs index d0aa4b4..47ca922 100644 --- a/crates/statsai-daemon/src/watch/scan.rs +++ b/crates/statsai-daemon/src/watch/scan.rs @@ -13,12 +13,12 @@ use statsai_core::{ UsageSummary, ACTIVITY_PARSER_REVISION, }; use statsai_store::{ - find_existing_provider_account, reconcile_verified_source_state, upsert_provider_account, - verified_source_observation_hash, ActivityPersistMode, ActivityScanCursor, ScanFileReplacement, - ScanFileStateEntry, Store, UpsertProviderAccountInput, + find_existing_provider_account, reconcile_verified_source_state, try_acquire_scan_lock, + upsert_provider_account, verified_source_observation_hash, ActivityPersistMode, + ActivityScanCursor, ScanFileReplacement, ScanFileStateEntry, Store, UpsertProviderAccountInput, }; use std::collections::{HashMap, HashSet}; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::mpsc; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -43,27 +43,68 @@ pub(super) fn enqueue_watch_notice( let _ = signal.try_send(()); } +/// What became of one pass of the background scan worker. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum BackgroundScanOutcome { + /// The pass scanned, or had nothing to scan. + Completed, + /// The scan failed; its paths were queued again after `retry_delay`. + Failed, + /// Another process held the scan lock; the paths were queued again after + /// [`super::WATCH_SCAN_LOCK_RETRY_DELAY`] without scanning. + Deferred, +} + +/// Runs one pass of the background scan worker under the scan lock. +/// +/// The lock is taken for the scan alone, never while the worker waits for +/// file events or for a retry, so `statsai scan` gets its turn between passes. pub(super) fn process_background_scan( pending: &Arc>, signal: &mpsc::SyncSender<()>, notice: WatchNotice, retry_delay: Duration, + scan_lock: Option<&Path>, scan: impl FnOnce(&WatchNotice) -> Result<()>, -) -> bool { +) -> BackgroundScanOutcome { if notice.paths().is_empty() && !notice.rescan_all() { - return true; + return BackgroundScanOutcome::Completed; } - if let Err(error) = scan(¬ice) { - eprintln!("daemon: background scan failed and will be retried: {error:#}"); - std::thread::sleep(retry_delay); - pending - .lock() - .unwrap_or_else(|error| error.into_inner()) - .restore_failed(notice); - let _ = signal.try_send(()); - return false; + let result = match scan_lock.map(try_acquire_scan_lock).transpose() { + // Another process is scanning: keep the paths and look again shortly. + Ok(Some(None)) => { + requeue_background_scan(pending, signal, notice, super::WATCH_SCAN_LOCK_RETRY_DELAY); + return BackgroundScanOutcome::Deferred; + } + // Held (or not configured) for exactly the length of the scan. + Ok(guard) => { + let _guard = guard.flatten(); + scan(¬ice) + } + Err(error) => Err(error), + }; + match result { + Ok(()) => BackgroundScanOutcome::Completed, + Err(error) => { + eprintln!("daemon: background scan failed and will be retried: {error:#}"); + requeue_background_scan(pending, signal, notice, retry_delay); + BackgroundScanOutcome::Failed + } } - true +} + +fn requeue_background_scan( + pending: &Arc>, + signal: &mpsc::SyncSender<()>, + notice: WatchNotice, + delay: Duration, +) { + std::thread::sleep(delay); + pending + .lock() + .unwrap_or_else(|error| error.into_inner()) + .restore_failed(notice); + let _ = signal.try_send(()); } pub(super) fn rescan_changed_sources( diff --git a/crates/statsai-daemon/src/watch/tests/scan.rs b/crates/statsai-daemon/src/watch/tests/scan.rs index 1daaeb4..b37af9e 100644 --- a/crates/statsai-daemon/src/watch/tests/scan.rs +++ b/crates/statsai-daemon/src/watch/tests/scan.rs @@ -32,15 +32,16 @@ fn failed_background_scan_is_requeued_for_retry() { let (signal_tx, signal_rx) = mpsc::sync_channel(1); let changed = PathBuf::from("/tmp/statsai-scan-retry"); - let scan_succeeded = process_background_scan( + let outcome = process_background_scan( &pending, &signal_tx, WatchNotice::Paths(vec![changed.clone()]), Duration::ZERO, + None, |_| anyhow::bail!("database is locked"), ); - assert!(!scan_succeeded); + assert_eq!(outcome, BackgroundScanOutcome::Failed); signal_rx.try_recv().expect("retry wakeup"); assert_eq!( pending.lock().expect("pending scan paths").paths(), @@ -48,6 +49,80 @@ fn failed_background_scan_is_requeued_for_retry() { ); } +#[test] +fn a_background_scan_waits_while_another_scanner_holds_the_scan_lock() { + let directory = tempfile::tempdir().expect("tempdir"); + let lock = directory.path().join(statsai_store::SCAN_LOCK_FILE_NAME); + let pending = Arc::new(Mutex::new(PendingWatch::default())); + let (signal_tx, signal_rx) = mpsc::sync_channel(1); + let changed = PathBuf::from("/tmp/statsai-scan-locked"); + let other_scanner = statsai_store::try_acquire_scan_lock(&lock) + .expect("acquire") + .expect("uncontended lock"); + + let outcome = process_background_scan( + &pending, + &signal_tx, + WatchNotice::Paths(vec![changed.clone()]), + Duration::ZERO, + Some(&lock), + |_| panic!("scanned while another scanner held the lock"), + ); + + assert_eq!(outcome, BackgroundScanOutcome::Deferred); + signal_rx.try_recv().expect("retry wakeup"); + let notice = pending.lock().expect("pending scan paths").take(); + assert_eq!(notice, WatchNotice::Paths(vec![changed.clone()])); + + drop(other_scanner); + let outcome = process_background_scan( + &pending, + &signal_tx, + notice, + Duration::ZERO, + Some(&lock), + |notice| { + assert_eq!(notice.paths(), std::slice::from_ref(&changed)); + assert!( + statsai_store::try_acquire_scan_lock(&lock)?.is_none(), + "the pass must hold the scan lock while it scans" + ); + Ok(()) + }, + ); + assert_eq!(outcome, BackgroundScanOutcome::Completed); + assert!( + statsai_store::try_acquire_scan_lock(&lock) + .expect("try") + .is_some(), + "the pass must release the scan lock when it finishes" + ); +} + +#[test] +fn a_failed_background_scan_releases_the_scan_lock_before_its_retry() { + let directory = tempfile::tempdir().expect("tempdir"); + let lock = directory.path().join(statsai_store::SCAN_LOCK_FILE_NAME); + let pending = Arc::new(Mutex::new(PendingWatch::default())); + let (signal_tx, signal_rx) = mpsc::sync_channel(1); + + let outcome = process_background_scan( + &pending, + &signal_tx, + WatchNotice::RescanAll, + Duration::ZERO, + Some(&lock), + |_| anyhow::bail!("database is locked"), + ); + + assert_eq!(outcome, BackgroundScanOutcome::Failed); + signal_rx.try_recv().expect("retry wakeup"); + assert!(pending.lock().expect("pending scan paths").rescan_all()); + assert!(statsai_store::try_acquire_scan_lock(&lock) + .expect("try") + .is_some()); +} + pub(super) struct TestAdapter { pub(super) provider: &'static str, pub(super) verified_observation: VerifiedSourceObservation, diff --git a/crates/statsai-store/Cargo.toml b/crates/statsai-store/Cargo.toml index 80f6c92..f91163c 100644 --- a/crates/statsai-store/Cargo.toml +++ b/crates/statsai-store/Cargo.toml @@ -21,6 +21,7 @@ serde.workspace = true serde_json.workspace = true sha2.workspace = true tempfile.workspace = true +thiserror.workspace = true [target.'cfg(target_os = "macos")'.dependencies] libc.workspace = true diff --git a/crates/statsai-store/src/lib.rs b/crates/statsai-store/src/lib.rs index fdb4aac..41f76fc 100644 --- a/crates/statsai-store/src/lib.rs +++ b/crates/statsai-store/src/lib.rs @@ -12,7 +12,9 @@ mod migrations; mod pricing; mod privacy; mod quota; +mod read_only; mod rollups; +mod scan_lock; mod scan_state; mod snapshot; mod sources; @@ -81,7 +83,12 @@ pub use pricing::{ apply_current_estimated_pricing, RepricingReport, APPLIED_PRICING_CATALOG_VERSION_KEY, APPLIED_PRICING_RULESET_VERSION_KEY, }; +pub use read_only::{ReadOnlyOpenError, ReadStore}; pub use rollups::{SessionFilter, SessionSort, SessionStats}; +pub use scan_lock::{ + acquire_scan_lock_with_timeout, scan_lock_path, try_acquire_scan_lock, ScanLockGuard, + SCAN_LOCK_FILE_NAME, +}; pub use snapshot::{ clone_database_to, database_applied_pricing_ruleset_version, database_schema_version, DatabaseClone, @@ -382,22 +389,10 @@ impl Store { /// subscriptions, account assignments, privacy identity — and none of that /// can be collected again from local files. Relaxing durability is scoped /// to the imports that can, in [`Store::relax_durability_for_bulk_import`]. - /// - /// The page cache is raised because these archives are far larger than the - /// 2MB default, which turns index maintenance into a stream of single-page - /// reads. - /// - /// `mmap_size` covers what the cache cannot. A store with a year of history runs - /// to several gigabytes, so the reads that miss the cache are the ones that hurt, - /// and serving those from a mapping spares them a `pread` apiece. A gigabyte is - /// deliberately short of the whole file: it comfortably spans the tables scanning - /// touches without mapping the archive content that only full-text search reads. fn configure_connection(&self) -> Result<()> { + configure_read_cache(&self.conn)?; self.conn.execute_batch( - "PRAGMA cache_size = -65536; - PRAGMA mmap_size = 1073741824; - PRAGMA temp_store = MEMORY; - CREATE TEMP TABLE IF NOT EXISTS incoming_records ( + "CREATE TEMP TABLE IF NOT EXISTS incoming_records ( source_record_id TEXT, item_id TEXT NOT NULL ); @@ -430,6 +425,22 @@ impl Store { Self::open(Path::new(path)) } + /// Opens an existing store for reading only, without changing it. + /// + /// Unlike [`Store::open`], this never creates the file, migrates, backfills, + /// reprices, or changes permissions, and refuses a schema version other than + /// [`CURRENT_SCHEMA_VERSION`] with a typed error, so a process that only + /// reads (a desktop dashboard, say) can share the store with the CLI and + /// daemon while they scan. See [`ReadStore`] for what it can read. + /// + /// # Errors + /// + /// Returns [`ReadOnlyOpenError`] when the file is missing, its schema is + /// older or newer than this binary's, or SQLite cannot open it. + pub fn open_read_only(path: &Path) -> std::result::Result { + ReadStore::open(path) + } + /// Returns SQLite's connection-local database generation. /// /// The value changes when another connection commits, which lets a @@ -643,6 +654,29 @@ impl Store { } } +/// Sizes the page cache and memory map for large stores, on any connection. +/// +/// The page cache is raised because these archives are far larger than the +/// 2MB default, which turns index maintenance into a stream of single-page +/// reads. +/// +/// `mmap_size` covers what the cache cannot. A store with a year of history runs +/// to several gigabytes, so the reads that miss the cache are the ones that hurt, +/// and serving those from a mapping spares them a `pread` apiece. A gigabyte is +/// deliberately short of the whole file: it comfortably spans the tables scanning +/// touches without mapping the archive content that only full-text search reads. +/// +/// These settings are local to the connection and write nothing to the file, +/// so read-only connections use them too. +fn configure_read_cache(conn: &Connection) -> Result<()> { + conn.execute_batch( + "PRAGMA cache_size = -65536; + PRAGMA mmap_size = 1073741824; + PRAGMA temp_store = MEMORY;", + )?; + Ok(()) +} + #[cfg(test)] mod sqlite_security; #[cfg(test)] diff --git a/crates/statsai-store/src/migrations/mod.rs b/crates/statsai-store/src/migrations/mod.rs index d0f5da1..bbbaba5 100644 --- a/crates/statsai-store/src/migrations/mod.rs +++ b/crates/statsai-store/src/migrations/mod.rs @@ -31,7 +31,10 @@ pub fn migrate(conn: &Connection) -> Result<()> { Ok(()) } -fn existing_schema_version(conn: &Connection) -> Result> { +/// The recorded schema version, or `None` before `schema_migrations` exists. +/// +/// Only reads, so a read-only connection can use it. +pub(crate) fn existing_schema_version(conn: &Connection) -> Result> { let has_migrations_table = conn.query_row( "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'schema_migrations')", [], diff --git a/crates/statsai-store/src/read_only.rs b/crates/statsai-store/src/read_only.rs new file mode 100644 index 0000000..108bf4b --- /dev/null +++ b/crates/statsai-store/src/read_only.rs @@ -0,0 +1,316 @@ +//! Read-only access for processes that must never write the store. + +use super::{ + configure_read_cache, enable_sqlite_defensive, migrations::existing_schema_version, + CacheReportQuery, QuotaQuery, QuotaStatus, SessionFilter, SessionSort, SessionStats, Store, + SyncState, WeeklyResetAnchor, CURRENT_SCHEMA_VERSION, SQLITE_BUSY_TIMEOUT, +}; +use anyhow::Result; +use chrono::{DateTime, Utc}; +use rusqlite::config::DbConfig; +use rusqlite::{Connection, OpenFlags}; +use statsai_core::{ + CacheReport, DailyRollup, ProviderAccount, ProviderAccountId, QuotaObservationRecordV1, + QuotaWindowV1, SessionRollupV1, SourceAccountAssignment, SourceLocation, Subscription, + UsageSummary, +}; +use std::cmp::Ordering; +use std::path::{Path, PathBuf}; + +/// Why [`Store::open_read_only`] did not open a store. +/// +/// The schema variants let a caller decide what to do: an older store needs a +/// writer such as any `statsai` command to migrate it, and a newer one needs a +/// newer StatsAI. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum ReadOnlyOpenError { + /// Nothing exists at the path. A read-only open never creates a store. + #[error("no StatsAI store at {}", .path.display())] + NotFound { path: PathBuf }, + /// The store was last migrated by an older StatsAI, or never migrated. + #[error( + "StatsAI store at {} has schema version {found}, older than the version this binary reads ({expected}); run statsai to migrate it", + .path.display() + )] + SchemaTooOld { + path: PathBuf, + found: i64, + expected: i64, + }, + /// The store was migrated by a newer StatsAI than this binary. + #[error( + "StatsAI store at {} has schema version {found}, newer than this binary supports ({expected}); upgrade StatsAI", + .path.display() + )] + SchemaTooNew { + path: PathBuf, + found: i64, + expected: i64, + }, + /// SQLite could not open the file or read its schema version. + #[error("open StatsAI store read-only at {}", .path.display())] + Open { + path: PathBuf, + #[source] + source: anyhow::Error, + }, +} + +/// A store opened with [`Store::open_read_only`]: the read side of [`Store`] +/// and nothing else. +/// +/// It is for processes that must never write the store, such as a dashboard +/// that reads it while `statsai scan` or `statsai daemon --watch` writes. +/// +/// - The connection is opened with `SQLITE_OPEN_READONLY`, the flag the +/// `mode=ro` URI parameter selects, and `PRAGMA query_only` is on, so SQLite +/// refuses any write. It is deliberately not `immutable`: it takes ordinary +/// WAL read locks and sees each transaction a writer commits. +/// - Opening never creates the file, migrates, backfills, reprices, changes +/// permissions, or runs `PRAGMA optimize`, and closing never checkpoints. +/// - Only methods that read are exposed. Where [`Store`] pairs a read with +/// upkeep, as [`Store::all_sync_rollup_summaries`] does, this returns what is +/// stored and leaves the upkeep to the next writer. +/// +/// Each call reads in its own transaction. Wrap several in +/// [`ReadStore::with_read_snapshot`] to see one point in time, and compare +/// [`ReadStore::data_version`] to notice that a writer has committed since. +/// +/// In WAL mode SQLite coordinates readers through the `-wal` and `-shm` files +/// beside the database. When no other connection has the store open they may +/// not exist, and SQLite creates them for this connection, which needs write +/// access to the directory (never to the database file). In a directory this +/// process cannot write, the open fails with [`ReadOnlyOpenError::Open`] +/// unless those files already exist. +pub struct ReadStore { + store: Store, +} + +impl std::fmt::Debug for ReadStore { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("ReadStore") + .field("path", &self.store.conn.path()) + .finish_non_exhaustive() + } +} + +impl ReadStore { + pub(crate) fn open(path: &Path) -> std::result::Result { + let missing = std::fs::metadata(path) + .is_err_and(|error| error.kind() == std::io::ErrorKind::NotFound); + if missing { + return Err(ReadOnlyOpenError::NotFound { + path: path.to_path_buf(), + }); + } + let open_error = |source: anyhow::Error| ReadOnlyOpenError::Open { + path: path.to_path_buf(), + source, + }; + let conn = open_read_only_connection(path).map_err(open_error)?; + // A store that predates `schema_migrations` has version zero, as + // `database_schema_version` reports it. + let found = existing_schema_version(&conn) + .map_err(open_error)? + .unwrap_or(0); + let expected = CURRENT_SCHEMA_VERSION; + match found.cmp(&expected) { + Ordering::Less => Err(ReadOnlyOpenError::SchemaTooOld { + path: path.to_path_buf(), + found, + expected, + }), + Ordering::Greater => Err(ReadOnlyOpenError::SchemaTooNew { + path: path.to_path_buf(), + found, + expected, + }), + Ordering::Equal => Ok(Self { + store: Store { conn }, + }), + } + } + + /// The store's schema version, which is [`CURRENT_SCHEMA_VERSION`] + /// whenever the open succeeded. + pub fn schema_version(&self) -> Result { + self.store.schema_version() + } + + /// See [`Store::data_version`]: changes when another connection commits. + pub fn data_version(&self) -> Result { + self.store.data_version() + } + + /// Runs several reads against one point-in-time view. See + /// [`Store::with_read_snapshot`]. + pub fn with_read_snapshot(&self, operation: impl FnOnce(&Self) -> Result) -> Result { + self.store.with_read_snapshot(|_| operation(self)) + } + + /// The pricing ruleset the store's costs were last computed with. A reader + /// never reprices, so this can trail the binary's own ruleset until a + /// writer such as `statsai scan` next opens the store. + pub fn applied_pricing_ruleset_version(&self) -> Result> { + self.store.applied_pricing_ruleset_version() + } + + pub fn event_count(&self) -> Result { + self.store.event_count() + } + + pub fn token_total(&self) -> Result { + self.store.token_total() + } + + /// See [`Store::session_rollups_in_period`]. + pub fn session_rollups_in_period( + &self, + since: Option>, + until: DateTime, + filter: &SessionFilter, + sort: SessionSort, + limit: usize, + offset: usize, + ) -> Result> { + self.store + .session_rollups_in_period(since, until, filter, sort, limit, offset) + } + + /// See [`Store::session_stats_in_period`]. + pub fn session_stats_in_period( + &self, + since: Option>, + until: DateTime, + filter: &SessionFilter, + ) -> Result { + self.store.session_stats_in_period(since, until, filter) + } + + /// Every daily summary as stored. Unlike + /// [`Store::all_sync_rollup_summaries`], summaries an older binary built + /// are returned as they are rather than rebuilt; the next writer that + /// reads them through [`Store`] rebuilds them. + pub fn all_sync_rollup_summaries(&self) -> Result> { + self.store.stored_sync_rollup_summaries() + } + + /// See [`Store::daily_rollups_between`]. + pub fn daily_rollups_between( + &self, + start_date: &str, + end_date: &str, + ) -> Result> { + self.store.daily_rollups_between(start_date, end_date) + } + + /// See [`Store::cache_report`]. + pub fn cache_report( + &self, + query: &CacheReportQuery, + timeline_zone: &Tz, + ) -> Result + where + Tz::Offset: std::fmt::Display, + { + self.store.cache_report(query, timeline_zone) + } + + /// See [`Store::quota_status`]. + pub fn quota_status(&self, query: &QuotaQuery) -> Result { + self.store.quota_status(query) + } + + /// See [`Store::quota_windows`]. + pub fn quota_windows(&self, query: &QuotaQuery) -> Result> { + self.store.quota_windows(query) + } + + /// See [`Store::quota_observations`]. + pub fn quota_observations( + &self, + query: &QuotaQuery, + collapse_duplicates: bool, + ) -> Result> { + self.store.quota_observations(query, collapse_duplicates) + } + + /// See [`Store::weekly_reset_anchor`]. + pub fn weekly_reset_anchor( + &self, + provider: &str, + provider_account_id: &ProviderAccountId, + ) -> Result> { + self.store + .weekly_reset_anchor(provider, provider_account_id) + } + + pub fn list_sources(&self) -> Result> { + self.store.list_sources() + } + + pub fn list_accounts(&self) -> Result> { + self.store.list_accounts() + } + + pub fn list_source_account_assignments(&self) -> Result> { + self.store.list_source_account_assignments() + } + + pub fn list_subscriptions(&self) -> Result> { + self.store.list_subscriptions() + } + + pub fn list_sync_states(&self) -> Result> { + self.store.list_sync_states() + } +} + +/// Opens `path` so that nothing this connection does can change the file. +/// +/// `SQLITE_OPEN_READONLY` is what the `mode=ro` URI parameter sets; passing the +/// flag avoids escaping the path into a URI. Without `SQLITE_OPEN_CREATE` a +/// missing file is an error rather than a new database. +fn open_read_only_connection(path: &Path) -> Result { + let conn = Connection::open_with_flags( + path, + OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, + )?; + enable_sqlite_defensive(&conn)?; + // A read-only connection cannot checkpoint anyway; this keeps it from + // trying when it is the last one to close. + conn.set_db_config(DbConfig::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, true)?; + conn.busy_timeout(SQLITE_BUSY_TIMEOUT)?; + conn.execute_batch("PRAGMA query_only = ON;")?; + configure_read_cache(&conn)?; + Ok(conn) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_read_only_connection_refuses_writes() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = directory.path().join("statsai.sqlite"); + drop(Store::open(&path).expect("create store")); + + let reader = Store::open_read_only(&path).expect("open read-only"); + let conn = &reader.store.conn; + assert!(conn.is_readonly(rusqlite::MAIN_DB).expect("main db")); + for write in [ + "INSERT INTO local_metadata (key, value, updated_at) VALUES ('probe', '1', 'now')", + "CREATE TABLE probe (id INTEGER)", + "CREATE TEMP TABLE probe (id INTEGER)", + "PRAGMA user_version = 7", + ] { + assert!( + conn.execute_batch(write).is_err(), + "the read-only connection accepted {write:?}" + ); + } + } +} diff --git a/crates/statsai-store/src/rollups/mod.rs b/crates/statsai-store/src/rollups/mod.rs index 339934f..4a99d53 100644 --- a/crates/statsai-store/src/rollups/mod.rs +++ b/crates/statsai-store/src/rollups/mod.rs @@ -27,6 +27,12 @@ impl Store { pub fn all_sync_rollup_summaries(&self) -> Result> { self.ensure_current_sync_rollup_versions()?; + self.stored_sync_rollup_summaries() + } + + /// Every daily summary exactly as stored, without first rebuilding the + /// ones an older binary wrote. Only reads, so a read-only store can use it. + pub(crate) fn stored_sync_rollup_summaries(&self) -> Result> { self.sync_rollup_summaries_by_sql( "SELECT payload FROM sync_rollups ORDER BY updated_at, summary_id", ) diff --git a/crates/statsai-store/src/scan_lock.rs b/crates/statsai-store/src/scan_lock.rs new file mode 100644 index 0000000..dd6ed84 --- /dev/null +++ b/crates/statsai-store/src/scan_lock.rs @@ -0,0 +1,188 @@ +//! The advisory lock that keeps two scanners off one store at a time. +//! +//! `statsai scan`, the scanning paths of `statsai sync`, `statsai daemon +//! --watch`, and any other process that collects into the store take this lock +//! first. It lives in a file beside the database rather than in SQLite, so a +//! process can learn that a scan is running without opening the store. +//! +//! The lock is the operating system's whole-file lock: `flock` on Unix and +//! `LockFileEx` on Windows, through [`File::try_lock`]. Both belong to the open +//! file rather than to the process, so two opens conflict even inside one +//! process, and the lock is released when the guard's file is closed, including +//! when the holder crashes. It is advisory: it keeps scanners apart and stops +//! nothing else from reading or writing the store. + +use super::restrict_dir_permissions; +use anyhow::{Context, Result}; +use std::fs::{File, OpenOptions, TryLockError}; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +/// Name of the lock file, kept in the same directory as the database. +pub const SCAN_LOCK_FILE_NAME: &str = "scan.lock"; + +/// How often a waiting caller tries the lock again. +const SCAN_LOCK_POLL_INTERVAL: Duration = Duration::from_millis(50); + +/// The scan lock for the store at `store_path`: `scan.lock` in the same +/// directory, so `~/.statsai/statsai.sqlite` is guarded by +/// `~/.statsai/scan.lock`. +#[must_use] +pub fn scan_lock_path(store_path: &Path) -> PathBuf { + store_path + .parent() + .unwrap_or_else(|| Path::new("")) + .join(SCAN_LOCK_FILE_NAME) +} + +/// Proof that this process holds the scan lock. Dropping it releases the lock. +/// +/// The lock file is left in place on release. Deleting it would let a waiter +/// that already opened the old file lock it while a newcomer creates and locks +/// a new one, and both would believe they hold the lock. +#[derive(Debug)] +#[must_use = "the scan lock is released as soon as the guard is dropped"] +pub struct ScanLockGuard { + file: File, + path: PathBuf, +} + +impl ScanLockGuard { + /// The lock file this guard holds. + #[must_use] + pub fn path(&self) -> &Path { + &self.path + } +} + +impl Drop for ScanLockGuard { + fn drop(&mut self) { + // Closing the file releases the lock as well; unlocking first only + // makes the release independent of when the descriptor is closed. + let _ = self.file.unlock(); + } +} + +/// Takes the scan lock at `lock_path` if no one holds it. +/// +/// Returns `Ok(None)` when another holder has it. Creates the lock file, and +/// its directory with owner-only permissions, when they do not exist yet. +/// +/// A process holds the lock through one guard: taking it again while a guard +/// is alive reports it as held, so take it once at the outermost level. +/// +/// # Errors +/// +/// Returns an error if the lock file cannot be created or opened, or the +/// operating system cannot lock it. +pub fn try_acquire_scan_lock(lock_path: &Path) -> Result> { + acquire_scan_lock_with_timeout(lock_path, Duration::ZERO) +} + +/// Takes the scan lock at `lock_path`, waiting up to `timeout` for the current +/// holder to release it. +/// +/// Returns `Ok(None)` when the lock is still held once `timeout` has passed. +/// A zero `timeout` tries once, like [`try_acquire_scan_lock`]. +/// +/// # Errors +/// +/// Returns an error if the lock file cannot be created or opened, or the +/// operating system cannot lock it. +pub fn acquire_scan_lock_with_timeout( + lock_path: &Path, + timeout: Duration, +) -> Result> { + let file = open_lock_file(lock_path)?; + // A timeout too large to represent waits indefinitely. + let deadline = Instant::now().checked_add(timeout); + loop { + match file.try_lock() { + Ok(()) => { + return Ok(Some(ScanLockGuard { + file, + path: lock_path.to_path_buf(), + })); + } + Err(TryLockError::WouldBlock) => {} + Err(TryLockError::Error(error)) => { + return Err(error) + .with_context(|| format!("lock scan lock {}", lock_path.display())); + } + } + let now = Instant::now(); + let pause = match deadline { + Some(deadline) if now >= deadline => return Ok(None), + Some(deadline) => SCAN_LOCK_POLL_INTERVAL.min(deadline - now), + None => SCAN_LOCK_POLL_INTERVAL, + }; + std::thread::sleep(pause); + } +} + +fn open_lock_file(lock_path: &Path) -> Result { + if let Some(parent) = lock_path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + if !parent.exists() { + std::fs::create_dir_all(parent) + .with_context(|| format!("create {}", parent.display()))?; + restrict_dir_permissions(parent)?; + } + } + let mut options = OpenOptions::new(); + // Never truncated: holders and waiters share the one file. + options.read(true).write(true).create(true).truncate(false); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + options + .open(lock_path) + .with_context(|| format!("open scan lock {}", lock_path.display())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_lock_lives_beside_the_store() { + assert_eq!( + scan_lock_path(Path::new("/home/me/.statsai/statsai.sqlite")), + PathBuf::from("/home/me/.statsai/scan.lock") + ); + assert_eq!( + scan_lock_path(Path::new("statsai.sqlite")), + PathBuf::from("scan.lock") + ); + } + + #[test] + #[cfg(unix)] + fn a_new_lock_directory_is_private_and_the_file_survives_release() { + use std::os::unix::fs::PermissionsExt; + + let directory = tempfile::tempdir().expect("tempdir"); + let lock_path = directory.path().join(".statsai").join(SCAN_LOCK_FILE_NAME); + + let guard = try_acquire_scan_lock(&lock_path) + .expect("acquire") + .expect("uncontended lock"); + assert_eq!(guard.path(), lock_path); + drop(guard); + + let mode = |path: &Path| { + std::fs::metadata(path) + .expect("metadata") + .permissions() + .mode() + & 0o777 + }; + assert_eq!(mode(lock_path.parent().expect("parent")), 0o700); + assert_eq!(mode(&lock_path), 0o600); + assert!(lock_path.exists(), "release must not delete the lock file"); + } +} diff --git a/crates/statsai-store/tests/read_only_store.rs b/crates/statsai-store/tests/read_only_store.rs new file mode 100644 index 0000000..db57f9f --- /dev/null +++ b/crates/statsai-store/tests/read_only_store.rs @@ -0,0 +1,477 @@ +//! `Store::open_read_only`: what a process that must never write can open and +//! read, including while another process scans into the same store. + +use chrono::{DateTime, Duration, TimeZone, Utc}; +use rusqlite::Connection; +use statsai_core::{ + Confidence, CostInfo, EventId, EventSource, LocationOrigin, ModelInfo, PrivacyInfo, + PrivacyMode, SessionInfo, SourceKind, SourceLocation, UsageCounts, UsageEvent, + USAGE_EVENT_SCHEMA_VERSION, +}; +use statsai_store::{ + CacheReportQuery, QuotaQuery, ReadOnlyOpenError, ReadStore, SessionFilter, SessionSort, Store, + CURRENT_SCHEMA_VERSION, +}; +use std::path::{Path, PathBuf}; + +fn started_at(index: u32) -> DateTime { + Utc.with_ymd_and_hms(2026, 7, 1, 9, 0, 0) + .single() + .expect("time") + + Duration::minutes(i64::from(index)) +} + +fn source() -> SourceLocation { + SourceLocation::local_adapter( + "codex", + "read-only-test", + "0", + Path::new("/tmp/statsai-read-only-test"), + LocationOrigin::Configured, + ) +} + +fn event(index: u32) -> UsageEvent { + let at = started_at(index); + UsageEvent { + schema_version: USAGE_EVENT_SCHEMA_VERSION.to_string(), + event_id: EventId(format!("event_read_only_{index:04}")), + device_id: "device_read_only".to_string(), + provider: "codex".to_string(), + source_id: source().source_id, + provider_account_id: None, + subscription_id: None, + source: EventSource { + adapter_id: "read-only-test".to_string(), + adapter_version: "0".to_string(), + source_kind: SourceKind::LocalAdapter, + location_origin: Some(LocationOrigin::Configured), + source_type: "jsonl".to_string(), + source_path_hash: None, + source_record_id: None, + parse_confidence: Confidence::High, + }, + session: SessionInfo { + session_id: format!("session_read_only_{}", index % 3), + local_session_id_hash: Some(format!("session_hash_read_only_{}", index % 3)), + // Session rollups list only sessions with a title or a project. + title: Some(format!("Read-only session {}", index % 3)), + started_at: at, + ended_at: None, + duration_seconds: None, + turn_started_at: None, + }, + model: Some(ModelInfo { + normalized_name: Some("gpt-6-sol".to_string()), + ..ModelInfo::default() + }), + usage: UsageCounts { + input_tokens: Some(1_000), + output_tokens: Some(100), + cache_read_tokens: Some(500), + requests: Some(1), + ..UsageCounts::default() + }, + runtime: None, + cost: CostInfo { + currency: "USD".to_string(), + estimated_api_equivalent_usd: None, + provider_reported_usd: None, + estimated_api_equivalent_micro_usd: None, + provider_reported_micro_usd: None, + pricing_source: None, + pricing_version: None, + confidence: Confidence::Low, + }, + parse_evidence: None, + project: None, + git: None, + privacy: PrivacyInfo { + mode: PrivacyMode::MetadataOnly, + contains_prompt_text: false, + contains_response_text: false, + contains_file_paths: false, + }, + created_at: at, + imported_at: at, + context: None, + } +} + +fn store_path(directory: &tempfile::TempDir) -> PathBuf { + directory.path().join("statsai.sqlite") +} + +/// A store as the CLI leaves it: written, then closed. +fn written_store(path: &Path, events: u32) { + let store = Store::open(path).expect("create store"); + store.upsert_source(&source()).expect("source"); + let events = (0..events).map(event).collect::>(); + store.insert_events(&events).expect("events"); +} + +fn recorded_schema_version(path: &Path) -> Option { + statsai_store::database_schema_version(path).expect("schema version") +} + +fn open_error(path: &Path) -> ReadOnlyOpenError { + match Store::open_read_only(path) { + Ok(_) => panic!("{} must not open read-only", path.display()), + Err(error) => error, + } +} + +/// Runs every read the read-only store exposes, so a write hidden in any of +/// them fails the test that calls this. +fn read_everything(reader: &ReadStore) { + let now = started_at(0) + Duration::days(1); + reader.schema_version().expect("schema version"); + reader.data_version().expect("data version"); + reader + .applied_pricing_ruleset_version() + .expect("pricing ruleset"); + reader.event_count().expect("event count"); + reader.token_total().expect("token total"); + reader + .session_rollups_in_period( + None, + now, + &SessionFilter::default(), + SessionSort::default(), + 100, + 0, + ) + .expect("session rollups"); + reader + .session_stats_in_period(None, now, &SessionFilter::default()) + .expect("session stats"); + reader + .all_sync_rollup_summaries() + .expect("sync rollup summaries"); + reader + .daily_rollups_between("2026-01-01", "2026-12-31") + .expect("daily rollups"); + reader + .cache_report(&CacheReportQuery::all_time(now), &Utc) + .expect("cache report"); + reader + .quota_status(&QuotaQuery::default()) + .expect("quota status"); + reader + .quota_windows(&QuotaQuery::default()) + .expect("quota windows"); + reader + .quota_observations(&QuotaQuery::default(), true) + .expect("quota observations"); + reader + .weekly_reset_anchor( + "claude_code", + &statsai_core::ProviderAccountId("acct_read_only".to_string()), + ) + .expect("weekly reset anchor"); + reader.list_sources().expect("sources"); + reader.list_accounts().expect("accounts"); + reader + .list_source_account_assignments() + .expect("source account assignments"); + reader.list_subscriptions().expect("subscriptions"); + reader.list_sync_states().expect("sync states"); + reader + .with_read_snapshot(|snapshot| snapshot.event_count()) + .expect("read snapshot"); +} + +#[test] +fn a_missing_store_is_reported_and_not_created() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = directory.path().join(".statsai").join("statsai.sqlite"); + + match open_error(&path) { + ReadOnlyOpenError::NotFound { path: reported } => assert_eq!(reported, path), + other => panic!("expected NotFound, got {other:?}"), + } + assert!(!path.exists(), "a read-only open created the database"); + assert!( + !path.parent().expect("parent").exists(), + "a read-only open created the store directory" + ); +} + +#[test] +fn a_fresh_store_opens_and_reads_empty() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + drop(Store::open(&path).expect("create store")); + + let reader = Store::open_read_only(&path).expect("open fresh store read-only"); + + assert_eq!( + reader.schema_version().expect("schema"), + CURRENT_SCHEMA_VERSION + ); + assert_eq!(reader.event_count().expect("events"), 0); + assert!(reader.list_sources().expect("sources").is_empty()); + assert!(reader + .all_sync_rollup_summaries() + .expect("summaries") + .is_empty()); + read_everything(&reader); +} + +/// Schema 31 is current as of this change; the test follows +/// `CURRENT_SCHEMA_VERSION`, so it keeps covering "a store this binary wrote". +#[test] +fn a_current_v31_store_reads_like_the_writer_and_is_left_byte_for_byte_unchanged() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + written_store(&path, 6); + #[cfg(unix)] + { + // Store::open would tighten this to 0600; a reader must not. + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)) + .expect("loosen permissions"); + } + let before = std::fs::read(&path).expect("database bytes"); + assert_eq!(recorded_schema_version(&path), Some(CURRENT_SCHEMA_VERSION)); + + let reader = Store::open_read_only(&path).expect("open read-only"); + read_everything(&reader); + assert_eq!(reader.event_count().expect("events"), 6); + assert_eq!(reader.list_sources().expect("sources").len(), 1); + // No writer has repriced this store, and a reader must not either. + assert_eq!( + reader.applied_pricing_ruleset_version().expect("ruleset"), + None + ); + let reader_summaries = reader.all_sync_rollup_summaries().expect("summaries"); + let reader_sessions = reader + .session_rollups_in_period( + None, + started_at(0) + Duration::days(1), + &SessionFilter::default(), + SessionSort::default(), + 100, + 0, + ) + .expect("sessions"); + assert!(!reader_summaries.is_empty()); + assert_eq!(reader_sessions.len(), 3); + drop(reader); + + assert_eq!( + std::fs::read(&path).expect("database bytes"), + before, + "reading changed the database file" + ); + let wal = PathBuf::from(format!("{}-wal", path.display())); + if wal.exists() { + assert_eq!( + std::fs::metadata(&wal).expect("wal").len(), + 0, + "a reader wrote to the WAL" + ); + } + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(&path) + .expect("metadata") + .permissions() + .mode() + & 0o777; + assert_eq!(mode, 0o644, "a read-only open changed permissions"); + } + + // The writer sees the same rows the reader did. + let writer = Store::open(&path).expect("writer"); + assert_eq!( + writer.all_sync_rollup_summaries().expect("summaries"), + reader_summaries + ); + assert_eq!( + writer + .session_rollups_in_period( + None, + started_at(0) + Duration::days(1), + &SessionFilter::default(), + SessionSort::default(), + 100, + 0, + ) + .expect("sessions"), + reader_sessions + ); +} + +#[test] +fn an_older_schema_is_refused_and_not_migrated() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + written_store(&path, 1); + let previous = CURRENT_SCHEMA_VERSION - 1; + Connection::open(&path) + .expect("raw connection") + .execute( + "DELETE FROM schema_migrations WHERE version > ?1", + [previous], + ) + .expect("roll the recorded schema back"); + + match open_error(&path) { + ReadOnlyOpenError::SchemaTooOld { + path: reported, + found, + expected, + } => { + assert_eq!(reported, path); + assert_eq!(found, previous); + assert_eq!(expected, CURRENT_SCHEMA_VERSION); + } + other => panic!("expected SchemaTooOld, got {other:?}"), + } + assert_eq!(recorded_schema_version(&path), Some(previous)); +} + +#[test] +fn a_store_from_before_schema_tracking_is_refused_as_version_zero() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + Connection::open(&path) + .expect("raw connection") + .execute_batch("CREATE TABLE sources (source_id TEXT PRIMARY KEY);") + .expect("legacy table"); + + match open_error(&path) { + ReadOnlyOpenError::SchemaTooOld { + found, expected, .. + } => { + assert_eq!((found, expected), (0, CURRENT_SCHEMA_VERSION)); + } + other => panic!("expected SchemaTooOld, got {other:?}"), + } + // A writer's open would have stamped and migrated it. + assert_eq!(recorded_schema_version(&path), Some(0)); +} + +#[test] +fn a_newer_schema_is_refused_with_both_versions() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + written_store(&path, 1); + let future = CURRENT_SCHEMA_VERSION + 1; + Connection::open(&path) + .expect("raw connection") + .execute( + "INSERT INTO schema_migrations (version, applied_at) VALUES (?1, '2026-10-01T00:00:00Z')", + [future], + ) + .expect("record a future schema"); + + let error = open_error(&path); + assert!( + error + .to_string() + .contains("newer than this binary supports"), + "{error}" + ); + match error { + ReadOnlyOpenError::SchemaTooNew { + found, expected, .. + } => assert_eq!((found, expected), (future, CURRENT_SCHEMA_VERSION)), + other => panic!("expected SchemaTooNew, got {other:?}"), + } + assert_eq!(recorded_schema_version(&path), Some(future)); +} + +#[test] +fn a_file_that_is_not_a_database_is_an_open_error() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + std::fs::write( + &path, + b"definitely not sqlite, but long enough to have a header....", + ) + .expect("write junk"); + + assert!(matches!(open_error(&path), ReadOnlyOpenError::Open { .. })); +} + +#[test] +fn reads_proceed_during_a_write_and_see_it_once_committed() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + let writer = Store::open(&path).expect("writer"); + let reader = Store::open_read_only(&path).expect("reader"); + let version_before = reader.data_version().expect("data version"); + + writer + .apply_scan_update(|store| { + store.upsert_source(&source())?; + store.insert_events(&[event(0), event(1)])?; + // The writer holds an open write transaction here. WAL lets the + // reader through, at the last committed state. + assert_eq!(reader.event_count()?, 0); + assert!(reader.list_sources()?.is_empty()); + Ok(()) + }) + .expect("scan update"); + + assert_eq!(reader.event_count().expect("events"), 2); + assert_eq!(reader.list_sources().expect("sources").len(), 1); + assert_ne!( + reader.data_version().expect("data version"), + version_before, + "data_version must move when another connection commits" + ); + + // A snapshot keeps its point in time while the writer commits more. + reader + .with_read_snapshot(|snapshot| { + assert_eq!(snapshot.event_count()?, 2); + writer.insert_events(&[event(2)])?; + assert_eq!(snapshot.event_count()?, 2); + assert_eq!(snapshot.all_sync_rollup_summaries()?.len(), 1); + Ok(()) + }) + .expect("snapshot"); + assert_eq!(reader.event_count().expect("events"), 3); +} + +#[test] +fn a_reader_thread_keeps_reading_while_a_writer_thread_commits() { + const WRITES: u32 = 25; + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + let writer = Store::open(&path).expect("writer"); + writer.upsert_source(&source()).expect("source"); + let done = std::sync::atomic::AtomicBool::new(false); + + std::thread::scope(|scope| { + let reader = scope.spawn(|| { + let reader = Store::open_read_only(&path).expect("reader"); + let mut last = 0; + let mut reads = 0u32; + loop { + let finished = done.load(std::sync::atomic::Ordering::Acquire); + let count = reader.event_count().expect("read during writes"); + assert!(count >= last, "event count went backwards"); + last = count; + reader + .all_sync_rollup_summaries() + .expect("summaries during writes"); + reads += 1; + if finished { + return (last, reads); + } + } + }); + for index in 0..WRITES { + writer.insert_events(&[event(index)]).expect("write"); + } + done.store(true, std::sync::atomic::Ordering::Release); + let (last, reads) = reader.join().expect("reader thread"); + assert_eq!(last, u64::from(WRITES)); + assert!(reads > 0); + }); +} diff --git a/crates/statsai-store/tests/scan_lock.rs b/crates/statsai-store/tests/scan_lock.rs new file mode 100644 index 0000000..c49c358 --- /dev/null +++ b/crates/statsai-store/tests/scan_lock.rs @@ -0,0 +1,153 @@ +//! The scan lock keeps two scanners apart, whether they are threads with their +//! own opens of the lock file or separate processes. + +use statsai_store::{acquire_scan_lock_with_timeout, scan_lock_path, try_acquire_scan_lock}; +use std::io::{BufRead, BufReader, Read, Write}; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; + +/// Set only on the child process the cross-process test spawns. +const CHILD_LOCK_PATH_ENV: &str = "STATSAI_SCAN_LOCK_TEST_CHILD_LOCK_PATH"; +const CHILD_HOLDS_LOCK: &str = "STATSAI-TEST-CHILD-HOLDS-THE-SCAN-LOCK"; + +fn lock_path(directory: &tempfile::TempDir) -> PathBuf { + scan_lock_path(&directory.path().join("statsai.sqlite")) +} + +#[test] +fn a_held_lock_is_busy_for_another_open_until_released() { + let directory = tempfile::tempdir().expect("tempdir"); + let lock = lock_path(&directory); + + let held = try_acquire_scan_lock(&lock) + .expect("acquire") + .expect("uncontended"); + assert!(try_acquire_scan_lock(&lock).expect("try").is_none()); + let started = Instant::now(); + assert!( + acquire_scan_lock_with_timeout(&lock, Duration::from_millis(150)) + .expect("wait") + .is_none() + ); + assert!( + started.elapsed() >= Duration::from_millis(150), + "gave up before the timeout" + ); + + drop(held); + assert!(try_acquire_scan_lock(&lock).expect("retry").is_some()); + assert!(lock.exists(), "the lock file outlives its holders"); +} + +#[test] +fn a_waiter_gets_the_lock_when_the_holder_releases_it() { + let directory = tempfile::tempdir().expect("tempdir"); + let lock = lock_path(&directory); + let held = try_acquire_scan_lock(&lock) + .expect("acquire") + .expect("uncontended"); + + std::thread::scope(|scope| { + let waiter = scope.spawn(|| { + acquire_scan_lock_with_timeout(&lock, Duration::from_secs(30)) + .expect("wait") + .is_some() + }); + std::thread::sleep(Duration::from_millis(200)); + assert!(!waiter.is_finished(), "the waiter did not wait"); + drop(held); + assert!(waiter.join().expect("waiter thread")); + }); +} + +#[test] +fn threads_with_their_own_opens_hold_the_lock_one_at_a_time() { + let directory = tempfile::tempdir().expect("tempdir"); + let lock = lock_path(&directory); + let inside = AtomicUsize::new(0); + let most_inside = AtomicUsize::new(0); + + std::thread::scope(|scope| { + for _ in 0..6 { + scope.spawn(|| { + for _ in 0..5 { + let guard = acquire_scan_lock_with_timeout(&lock, Duration::from_secs(30)) + .expect("wait") + .expect("the lock frees up within the timeout"); + let now_inside = inside.fetch_add(1, Ordering::SeqCst) + 1; + most_inside.fetch_max(now_inside, Ordering::SeqCst); + std::thread::sleep(Duration::from_millis(2)); + inside.fetch_sub(1, Ordering::SeqCst); + drop(guard); + } + }); + } + }); + + assert_eq!(most_inside.load(Ordering::SeqCst), 1); +} + +#[test] +fn a_lock_held_by_another_process_is_busy_until_it_exits() { + let directory = tempfile::tempdir().expect("tempdir"); + let lock = lock_path(&directory); + let mut child = Command::new(std::env::current_exe().expect("test binary")) + .args([ + "--exact", + "hold_the_scan_lock_for_the_parent_test", + "--ignored", + "--nocapture", + "--test-threads=1", + ]) + .env(CHILD_LOCK_PATH_ENV, &lock) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn the lock-holding child"); + let mut lines = BufReader::new(child.stdout.take().expect("child stdout")).lines(); + loop { + let line = lines + .next() + .expect("the child exited before taking the lock") + .expect("read child stdout"); + if line.contains(CHILD_HOLDS_LOCK) { + break; + } + } + + assert!(try_acquire_scan_lock(&lock).expect("try").is_none()); + assert!( + acquire_scan_lock_with_timeout(&lock, Duration::from_millis(100)) + .expect("wait") + .is_none() + ); + + // Closing its stdin tells the child to release the lock and exit. + drop(child.stdin.take()); + assert!(child.wait().expect("child exit").success()); + assert!( + acquire_scan_lock_with_timeout(&lock, Duration::from_secs(10)) + .expect("wait after exit") + .is_some() + ); +} + +/// The other process in the test above. Does nothing when run on its own. +#[test] +#[ignore = "the child process of a_lock_held_by_another_process_is_busy_until_it_exits"] +fn hold_the_scan_lock_for_the_parent_test() { + let Some(lock) = std::env::var_os(CHILD_LOCK_PATH_ENV) else { + return; + }; + let guard = try_acquire_scan_lock(Path::new(&lock)) + .expect("acquire") + .expect("the parent test leaves the lock free for the child"); + println!("{CHILD_HOLDS_LOCK}"); + std::io::stdout().flush().expect("flush"); + let mut rest = Vec::new(); + let _ = std::io::stdin().read_to_end(&mut rest); + drop(guard); +} diff --git a/crates/statsai/src/cli/daemon.rs b/crates/statsai/src/cli/daemon.rs index 366cb72..03d1095 100644 --- a/crates/statsai/src/cli/daemon.rs +++ b/crates/statsai/src/cli/daemon.rs @@ -1,14 +1,28 @@ use anyhow::Result; use statsai_store::Store; +use std::path::Path; use std::sync::{Arc, Mutex}; use super::args::DaemonCommand; -pub(crate) fn daemon(command: DaemonCommand, store: Store, device_id: &str) -> Result<()> { +pub(crate) fn daemon( + command: DaemonCommand, + store: Store, + device_id: &str, + store_path: &Path, +) -> Result<()> { let store = Arc::new(Mutex::new(store)); let auth_token = statsai::default_daemon_auth_token()?; if command.watch { - statsai_daemon::watch_and_serve(&command.api, store, device_id, &auth_token) + // The same lock `statsai scan` takes for this store, held per pass. + let scan_lock = statsai::scan_lock_path(store_path); + statsai_daemon::watch_and_serve( + &command.api, + store, + device_id, + &auth_token, + Some(&scan_lock), + ) } else { statsai_daemon::run(&command.api, store, &auth_token) } diff --git a/crates/statsai/src/cli/dispatch.rs b/crates/statsai/src/cli/dispatch.rs index a8ddb6a..ec9e9cf 100644 --- a/crates/statsai/src/cli/dispatch.rs +++ b/crates/statsai/src/cli/dispatch.rs @@ -1,10 +1,16 @@ use anyhow::Result; use clap::Parser; -use statsai::{default_device_id, default_store_path, snapshot}; +use statsai::{default_device_id, default_store_path, snapshot, ScanLockGuard}; use statsai_store::Store; +use std::path::{Path, PathBuf}; +use std::time::Duration; use super::*; +/// How long `statsai scan` and `statsai sync` wait for another scanner, such +/// as the watch daemon, to finish before skipping this run. +pub(crate) const SCAN_LOCK_WAIT: Duration = Duration::from_secs(5); + pub(crate) fn run() -> Result<()> { let cli = Cli::parse(); let store_path = cli.store.unwrap_or_else(default_store_path); @@ -18,6 +24,21 @@ pub(crate) fn run() -> Result<()> { Command::Service(command) => service(command), Command::Snapshot(command) => snapshot::run(command, &store_path, &device_id), command => { + // Taken before the store opens, so the migrations and repricing an + // open may run wait for a running scan as well, and held until the + // command returns. Nothing below takes it again. + let scan_lock = acquire_command_scan_lock(&command, &store_path, SCAN_LOCK_WAIT)?; + let _scan_lock = match scan_lock { + CommandScanLock::NotNeeded => None, + CommandScanLock::Held(guard) => Some(guard), + CommandScanLock::Busy(lock_path) => { + eprintln!( + "Another statsai scan is running (lock: {}); skipping.", + lock_path.display() + ); + return Ok(()); + } + }; let store = if command_reprices_persisted_usage(&command) { statsai::open_operational_store(&store_path)? } else { @@ -40,7 +61,7 @@ pub(crate) fn run() -> Result<()> { statsai::privacy_cli::run(command, &store, &store_path) } Command::Sync(command) => sync(command, &store, &device_id), - Command::Daemon(command) => daemon(command, store, &device_id), + Command::Daemon(command) => daemon(command, store, &device_id, &store_path), Command::Status => status(&store), Command::Schema(_) | Command::Store(_) @@ -68,3 +89,42 @@ pub(crate) fn command_reprices_persisted_usage(command: &Command) -> bool { | Command::Daemon(_) ) } + +/// Whether a command scans into the store and so must hold the scan lock. +/// +/// `sync` refreshes git code-change scans and rebuilds daily summaries before +/// it sends; only `--status`, `--verify`, and `--reset-remote` skip that. The +/// watch daemon is not listed: it takes the lock for each pass instead of for +/// its lifetime. +pub(crate) fn command_takes_scan_lock(command: &Command) -> bool { + match command { + Command::Scan(_) => true, + Command::Sync(sync) => !(sync.status || sync.verify || sync.reset_remote), + _ => false, + } +} + +pub(crate) enum CommandScanLock { + NotNeeded, + Held(ScanLockGuard), + /// Another process still held the lock at this path after the wait. + Busy(PathBuf), +} + +/// Takes the scan lock beside `store_path` for a command that scans, waiting +/// up to `wait` for another scanner to finish. +pub(crate) fn acquire_command_scan_lock( + command: &Command, + store_path: &Path, + wait: Duration, +) -> Result { + if !command_takes_scan_lock(command) { + return Ok(CommandScanLock::NotNeeded); + } + let lock_path = statsai::scan_lock_path(store_path); + let lock = statsai::acquire_scan_lock_with_timeout(&lock_path, wait)?; + Ok(match lock { + Some(guard) => CommandScanLock::Held(guard), + None => CommandScanLock::Busy(lock_path), + }) +} diff --git a/crates/statsai/src/cli/tests/mod.rs b/crates/statsai/src/cli/tests/mod.rs index bb1a76f..2e5b54f 100644 --- a/crates/statsai/src/cli/tests/mod.rs +++ b/crates/statsai/src/cli/tests/mod.rs @@ -8,6 +8,7 @@ mod import; mod quota; mod report; mod scan; +mod scan_lock; mod sessions; mod source; mod status; diff --git a/crates/statsai/src/cli/tests/scan_lock.rs b/crates/statsai/src/cli/tests/scan_lock.rs new file mode 100644 index 0000000..61bfe7f --- /dev/null +++ b/crates/statsai/src/cli/tests/scan_lock.rs @@ -0,0 +1,65 @@ +use super::*; + +fn parse(args: &[&str]) -> Command { + Cli::try_parse_from(args).expect("parse").command +} + +#[test] +fn scanning_commands_take_the_scan_lock_and_others_do_not() { + let takes = |args: &[&str]| command_takes_scan_lock(&parse(args)); + assert!(takes(&["statsai", "scan"])); + assert!(takes(&["statsai", "scan", "--preview"])); + assert!(takes(&["statsai", "sync"])); + assert!(takes(&[ + "statsai", + "sync", + "--sink", + "http", + "--since-last" + ])); + assert!(takes(&["statsai", "sync", "--dry-run"])); + assert!(!takes(&["statsai", "sync", "--status"])); + assert!(!takes(&["statsai", "sync", "--verify"])); + assert!(!takes(&["statsai", "sync", "--reset-remote", "--yes"])); + // The watch daemon takes it per pass, not for its lifetime. + assert!(!takes(&["statsai", "daemon", "--watch"])); + assert!(!takes(&["statsai", "report", "weekly"])); + assert!(!takes(&["statsai", "status"])); +} + +#[test] +fn a_scanning_command_reports_a_busy_lock_after_waiting() { + let directory = tempfile::tempdir().expect("tempdir"); + let store_path = directory.path().join("statsai.sqlite"); + let lock_path = statsai::scan_lock_path(&store_path); + let wait = StdDuration::from_millis(100); + let other_scanner = statsai::try_acquire_scan_lock(&lock_path) + .expect("acquire") + .expect("uncontended lock"); + + let started = std::time::Instant::now(); + match acquire_command_scan_lock(&parse(&["statsai", "scan"]), &store_path, wait) + .expect("scan lock") + { + CommandScanLock::Busy(reported) => assert_eq!(reported, lock_path), + _ => panic!("scan must see the lock as busy"), + } + assert!(started.elapsed() >= wait, "scan gave up without waiting"); + assert!(matches!( + acquire_command_scan_lock(&parse(&["statsai", "status"]), &store_path, wait) + .expect("no lock needed"), + CommandScanLock::NotNeeded + )); + + drop(other_scanner); + let held = match acquire_command_scan_lock(&parse(&["statsai", "sync"]), &store_path, wait) + .expect("scan lock") + { + CommandScanLock::Held(guard) => guard, + _ => panic!("sync must take the free lock"), + }; + assert_eq!(held.path(), lock_path); + assert!(statsai::try_acquire_scan_lock(&lock_path) + .expect("try") + .is_none()); +} diff --git a/crates/statsai/src/lib.rs b/crates/statsai/src/lib.rs index 06157bb..0c9bf97 100644 --- a/crates/statsai/src/lib.rs +++ b/crates/statsai/src/lib.rs @@ -13,6 +13,17 @@ use std::fs::OpenOptions; use std::io::{ErrorKind, Write}; use std::path::{Path, PathBuf}; +/// The lock that keeps scanners of one store from running at once. +/// +/// `statsai scan`, the scanning paths of `statsai sync`, and each pass of +/// `statsai daemon --watch` hold it; any other process that collects into the +/// store should take it too. It is implemented in `statsai-store` so the +/// daemon crate can share it, and re-exported here beside the default paths. +pub use statsai_store::{ + acquire_scan_lock_with_timeout, scan_lock_path, try_acquire_scan_lock, ScanLockGuard, + SCAN_LOCK_FILE_NAME, +}; + /// Opens a store for a command that reads or publishes price-derived data and /// applies the compiled pricing ruleset first. pub fn open_operational_store(path: &Path) -> Result { @@ -35,6 +46,14 @@ pub fn default_store_path() -> PathBuf { .join("statsai.sqlite") } +/// `~/.statsai/scan.lock`, the scan lock for [`default_store_path`]. +/// +/// A store opened elsewhere with `--store` has its lock beside it instead; use +/// [`scan_lock_path`] for that. +pub fn default_scan_lock_path() -> PathBuf { + scan_lock_path(&default_store_path()) +} + pub fn default_device_id() -> String { if let Ok(value) = std::env::var("STATSAI_DEVICE_ID") { let value = value.trim(); @@ -412,6 +431,17 @@ mod tests { ); } + #[test] + fn the_default_scan_lock_sits_beside_the_default_store() { + let lock = default_scan_lock_path(); + assert_eq!(lock.parent(), default_store_path().parent()); + assert_eq!( + lock.file_name(), + Some(std::ffi::OsStr::new(SCAN_LOCK_FILE_NAME)) + ); + assert!(lock.ends_with(".statsai/scan.lock")); + } + #[test] fn daemon_auth_token_is_random_persistent_and_private() { let temp = tempfile::tempdir().expect("temp dir"); diff --git a/crates/statsai/tests/scan_lock_cli.rs b/crates/statsai/tests/scan_lock_cli.rs new file mode 100644 index 0000000..9c7fdb2 --- /dev/null +++ b/crates/statsai/tests/scan_lock_cli.rs @@ -0,0 +1,95 @@ +//! `statsai scan` and `statsai sync` wait for a running scanner, then skip +//! without failing. + +use std::path::Path; +use std::process::{Command, Output}; +use std::time::Duration; + +fn run_statsai(home: &Path, store: &Path, args: &[&str]) -> std::process::Child { + Command::new(env!("CARGO_BIN_EXE_statsai")) + .arg("--store") + .arg(store) + .args(args) + // Nothing to collect, and nothing outside the test directory to find. + .env("HOME", home) + .env("CODEX_HOME", home.join("codex")) + .env("STATSAI_DEVICE_ID", "scan-lock-cli-test") + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .expect("run statsai") +} + +fn finish(child: std::process::Child) -> (Output, String) { + let output = child.wait_with_output().expect("statsai output"); + let stderr = String::from_utf8_lossy(&output.stderr).into_owned(); + assert!(output.status.success(), "statsai failed: {stderr}"); + (output, stderr) +} + +#[test] +fn scan_and_sync_skip_cleanly_while_another_scan_holds_the_lock() { + let directory = tempfile::tempdir().expect("tempdir"); + let store = directory.path().join("data").join("statsai.sqlite"); + let lock = statsai::scan_lock_path(&store); + let _other_scanner = statsai::try_acquire_scan_lock(&lock) + .expect("acquire") + .expect("uncontended lock"); + + // Both wait out the same few seconds, so run them side by side. + let scan = run_statsai(directory.path(), &store, &["scan", "--provider", "codex"]); + let batch = directory.path().join("batch.json"); + let sync = run_statsai( + directory.path(), + &store, + &[ + "sync", + "--sink", + "file", + "--output", + batch.to_str().expect("utf-8 path"), + ], + ); + let expected = format!( + "Another statsai scan is running (lock: {}); skipping.", + lock.display() + ); + for (name, child) in [("scan", scan), ("sync", sync)] { + let (output, stderr) = finish(child); + assert!( + stderr.contains(&expected), + "{name} did not report the busy lock: {stderr}" + ); + assert!(output.stdout.is_empty(), "{name} wrote output"); + } + assert!( + !store.exists(), + "a skipped command must not have opened (and created) the store" + ); + assert!(!batch.exists(), "a skipped sync must not have sent a batch"); +} + +#[test] +fn scan_waits_for_a_short_scan_to_finish_and_then_runs() { + let directory = tempfile::tempdir().expect("tempdir"); + let store = directory.path().join("statsai.sqlite"); + let lock = statsai::scan_lock_path(&store); + let other_scanner = statsai::try_acquire_scan_lock(&lock) + .expect("acquire") + .expect("uncontended lock"); + + let scan = run_statsai(directory.path(), &store, &["scan", "--provider", "codex"]); + std::thread::sleep(Duration::from_millis(500)); + drop(other_scanner); + let (_, stderr) = finish(scan); + + assert!( + !stderr.contains("Another statsai scan is running"), + "scan skipped instead of waiting: {stderr}" + ); + assert!(store.exists(), "the scan did not run"); + // The scan released the lock when it exited. + assert!(statsai::try_acquire_scan_lock(&lock) + .expect("try") + .is_some()); +} From 45cbdd2a797ad8875b2f4fbd2855929cfcc8d003 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:38:33 +0000 Subject: [PATCH 2/6] STA-5: Address review findings on the read-only store and scan lock - ReadStore::schema_version reads the version without writing. Store::schema_version created schema_migrations and set the journal mode, which failed on a rollback-journal store and inside a snapshot. - ReadStore::check_schema re-checks the schema, so a long-lived reader can notice a newer statsai migrating the store under it. - ReadOnlyOpenError::Open includes its cause in the message. - The read-only connection is built on snapshot::open_read_only instead of a second copy of the open flags. - sync validates its flags before waiting for the lock. A sync skipped for a running scan still saves its --include-*/--exclude-* preference flags, and says that nothing was sent. --reset-remote now takes the lock, so it cannot interleave with an upload. - The lock is advisory, so an unusable scan.lock (wrong owner, read-only mount) makes scan, sync and the daemon warn and carry on without it, instead of failing every run. - scan_lock_path resolves symlinks, so a linked store path shares the store's lock. - The daemon leaves the lock free for 200 ms after each pass, so a waiting scan or sync gets a turn when passes run back to back. - The comments and docs now name only the commands that take the lock. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G3EZeUTrrXQYg1NwFdgV3K --- README.md | 16 +-- crates/statsai-daemon/src/lib.rs | 3 +- crates/statsai-daemon/src/watch/mod.rs | 10 ++ crates/statsai-daemon/src/watch/scan.rs | 54 +++++++--- crates/statsai-daemon/src/watch/tests/scan.rs | 32 ++++++ crates/statsai-store/src/read_only.rs | 98 ++++++++++++------- crates/statsai-store/src/scan_lock.rs | 58 +++++++++-- crates/statsai-store/src/snapshot.rs | 8 +- crates/statsai-store/tests/read_only_store.rs | 75 +++++++++++++- crates/statsai/src/cli/dispatch.rs | 80 +++++++++++---- crates/statsai/src/cli/sync/mod.rs | 31 +++++- crates/statsai/src/cli/tests/scan_lock.rs | 46 +++++++-- crates/statsai/src/lib.rs | 23 +++-- crates/statsai/tests/scan_lock_cli.rs | 86 +++++++++++++++- 14 files changed, 510 insertions(+), 110 deletions(-) diff --git a/README.md b/README.md index a2a9483..8afebcc 100644 --- a/README.md +++ b/README.md @@ -432,12 +432,16 @@ statsai report cache --from 2026-05-01 --details Normal scans use a per-source file signature cache. `--no-cache` forces a one-off reread; `--replace` performs a destructive source rebuild. -One scan runs against a store at a time. `statsai scan`, `statsai sync` (other -than `--status`, `--verify`, and `--reset-remote`), and each pass of `statsai -daemon --watch` hold `scan.lock` beside the database (`~/.statsai/scan.lock` by -default). While another scan holds it, `scan` and `sync` wait up to 5 seconds, -then print `Another statsai scan is running (lock: …); skipping.` and exit -successfully; the daemon keeps changed files queued until the lock is free. +`statsai scan`, `statsai sync` (other than `--status` and `--verify`), and each +pass of `statsai daemon --watch` take turns through `scan.lock` beside the +database (`~/.statsai/scan.lock` by default). While another of them holds it, +`scan` and `sync` wait up to 5 seconds, then print `Another statsai scan is +running (lock: …); skipping.` (for `sync`, `…; skipping this sync. Nothing was +sent.`) and exit successfully. A skipped `sync` still saves its +`--include-*`/`--exclude-*` preference flags. The daemon keeps changed files +queued until the lock is free. The lock is advisory: if its file cannot be +opened or locked, these commands warn and run without it. `statsai import`, +`conversation collect`, and `source remove --delete-data` do not take it yet. JSONL input is streamed with a 16 MiB per-record ceiling. Invalid or oversized records are counted and discarded through the next newline, after which parsing diff --git a/crates/statsai-daemon/src/lib.rs b/crates/statsai-daemon/src/lib.rs index 52befbf..f428ad7 100644 --- a/crates/statsai-daemon/src/lib.rs +++ b/crates/statsai-daemon/src/lib.rs @@ -296,7 +296,8 @@ pub fn watch_and_serve( /// `scan_lock` is the lock file other scanners of this store take, normally /// `statsai_store::scan_lock_path(store_path)`. Each rescan holds it and only /// that long: while another process holds it the pass waits, and the changed -/// paths stay queued until the lock is free. `None` scans without it. +/// paths stay queued until the lock is free. `None` scans without it, and so +/// does a pass that cannot open or lock the file, after one warning. #[cfg(feature = "watch")] pub fn watch_and_serve( addr: &str, diff --git a/crates/statsai-daemon/src/watch/mod.rs b/crates/statsai-daemon/src/watch/mod.rs index 3f4b6f6..6e2c5cd 100644 --- a/crates/statsai-daemon/src/watch/mod.rs +++ b/crates/statsai-daemon/src/watch/mod.rs @@ -39,6 +39,16 @@ const WATCH_SCAN_LOCK_RETRY_DELAY: Duration = if cfg!(test) { } else { Duration::from_secs(1) }; +/// How long a pass leaves the scan lock free after releasing it. +/// +/// A busy watcher can start its next pass microseconds after the last one, +/// while `statsai scan` and `statsai sync` look for a free lock only every +/// 50 ms, so without a gap they could wait out their whole timeout and skip. +const WATCH_SCAN_LOCK_YIELD: Duration = if cfg!(test) { + Duration::ZERO +} else { + Duration::from_millis(200) +}; pub fn watch_and_serve( addr: &str, diff --git a/crates/statsai-daemon/src/watch/scan.rs b/crates/statsai-daemon/src/watch/scan.rs index 47ca922..f5d05c4 100644 --- a/crates/statsai-daemon/src/watch/scan.rs +++ b/crates/statsai-daemon/src/watch/scan.rs @@ -58,7 +58,10 @@ pub(super) enum BackgroundScanOutcome { /// Runs one pass of the background scan worker under the scan lock. /// /// The lock is taken for the scan alone, never while the worker waits for -/// file events or for a retry, so `statsai scan` gets its turn between passes. +/// file events or for a retry, and is left free for +/// [`super::WATCH_SCAN_LOCK_YIELD`] after each pass, so `statsai scan` gets +/// its turn between passes. The lock is advisory: when its file cannot be +/// opened or locked, the pass scans without it rather than failing. pub(super) fn process_background_scan( pending: &Arc>, signal: &mpsc::SyncSender<()>, @@ -70,19 +73,32 @@ pub(super) fn process_background_scan( if notice.paths().is_empty() && !notice.rescan_all() { return BackgroundScanOutcome::Completed; } - let result = match scan_lock.map(try_acquire_scan_lock).transpose() { - // Another process is scanning: keep the paths and look again shortly. - Ok(Some(None)) => { - requeue_background_scan(pending, signal, notice, super::WATCH_SCAN_LOCK_RETRY_DELAY); - return BackgroundScanOutcome::Deferred; - } - // Held (or not configured) for exactly the length of the scan. - Ok(guard) => { - let _guard = guard.flatten(); - scan(¬ice) - } - Err(error) => Err(error), + let guard = match scan_lock { + None => None, + Some(lock_path) => match try_acquire_scan_lock(lock_path) { + Ok(Some(guard)) => Some(guard), + // Another process is scanning: keep the paths and look again + // shortly. + Ok(None) => { + requeue_background_scan( + pending, + signal, + notice, + super::WATCH_SCAN_LOCK_RETRY_DELAY, + ); + return BackgroundScanOutcome::Deferred; + } + Err(error) => { + warn_scan_lock_unavailable(lock_path, &error); + None + } + }, }; + let result = scan(¬ice); + if let Some(guard) = guard { + drop(guard); + std::thread::sleep(super::WATCH_SCAN_LOCK_YIELD); + } match result { Ok(()) => BackgroundScanOutcome::Completed, Err(error) => { @@ -93,6 +109,18 @@ pub(super) fn process_background_scan( } } +/// Reports an unusable scan lock once per process: every later pass would +/// report the same thing. +fn warn_scan_lock_unavailable(lock_path: &Path, error: &anyhow::Error) { + static WARNED: std::sync::Once = std::sync::Once::new(); + WARNED.call_once(|| { + eprintln!( + "daemon: warning: could not use scan lock {}: {error:#}; scanning without it", + lock_path.display() + ); + }); +} + fn requeue_background_scan( pending: &Arc>, signal: &mpsc::SyncSender<()>, diff --git a/crates/statsai-daemon/src/watch/tests/scan.rs b/crates/statsai-daemon/src/watch/tests/scan.rs index b37af9e..4f2ff91 100644 --- a/crates/statsai-daemon/src/watch/tests/scan.rs +++ b/crates/statsai-daemon/src/watch/tests/scan.rs @@ -123,6 +123,38 @@ fn a_failed_background_scan_releases_the_scan_lock_before_its_retry() { .is_some()); } +#[test] +fn a_background_scan_runs_without_a_scan_lock_it_cannot_use() { + let directory = tempfile::tempdir().expect("tempdir"); + // A regular file where the lock's directory should be: the lock file can + // be neither created nor opened. + let not_a_directory = directory.path().join("not-a-directory"); + std::fs::write(¬_a_directory, b"").expect("regular file"); + let lock = not_a_directory.join(statsai_store::SCAN_LOCK_FILE_NAME); + let pending = Arc::new(Mutex::new(PendingWatch::default())); + let (signal_tx, signal_rx) = mpsc::sync_channel(1); + let scanned = std::cell::Cell::new(false); + + let outcome = process_background_scan( + &pending, + &signal_tx, + WatchNotice::RescanAll, + Duration::ZERO, + Some(&lock), + |_| { + scanned.set(true); + Ok(()) + }, + ); + + assert_eq!(outcome, BackgroundScanOutcome::Completed); + assert!(scanned.get(), "the pass did not scan"); + assert!( + signal_rx.try_recv().is_err(), + "a completed pass is not retried" + ); +} + pub(super) struct TestAdapter { pub(super) provider: &'static str, pub(super) verified_observation: VerifiedSourceObservation, diff --git a/crates/statsai-store/src/read_only.rs b/crates/statsai-store/src/read_only.rs index 108bf4b..22bd274 100644 --- a/crates/statsai-store/src/read_only.rs +++ b/crates/statsai-store/src/read_only.rs @@ -1,14 +1,14 @@ //! Read-only access for processes that must never write the store. use super::{ - configure_read_cache, enable_sqlite_defensive, migrations::existing_schema_version, - CacheReportQuery, QuotaQuery, QuotaStatus, SessionFilter, SessionSort, SessionStats, Store, - SyncState, WeeklyResetAnchor, CURRENT_SCHEMA_VERSION, SQLITE_BUSY_TIMEOUT, + configure_read_cache, migrations::existing_schema_version, snapshot, CacheReportQuery, + QuotaQuery, QuotaStatus, SessionFilter, SessionSort, SessionStats, Store, SyncState, + WeeklyResetAnchor, CURRENT_SCHEMA_VERSION, SQLITE_BUSY_TIMEOUT, }; use anyhow::Result; use chrono::{DateTime, Utc}; use rusqlite::config::DbConfig; -use rusqlite::{Connection, OpenFlags}; +use rusqlite::Connection; use statsai_core::{ CacheReport, DailyRollup, ProviderAccount, ProviderAccountId, QuotaObservationRecordV1, QuotaWindowV1, SessionRollupV1, SourceAccountAssignment, SourceLocation, Subscription, @@ -49,11 +49,13 @@ pub enum ReadOnlyOpenError { expected: i64, }, /// SQLite could not open the file or read its schema version. - #[error("open StatsAI store read-only at {}", .path.display())] + /// + /// The message includes `reason`. It is deliberately not the error's + /// `source`, so printing the error through a cause chain does not repeat it. + #[error("open StatsAI store read-only at {}: {reason:#}", .path.display())] Open { path: PathBuf, - #[source] - source: anyhow::Error, + reason: anyhow::Error, }, } @@ -77,6 +79,11 @@ pub enum ReadOnlyOpenError { /// [`ReadStore::with_read_snapshot`] to see one point in time, and compare /// [`ReadStore::data_version`] to notice that a writer has committed since. /// +/// The schema version is checked when the store opens, but a writer such as a +/// newer `statsai` may migrate the store while it stays open. A long-lived +/// reader should call [`ReadStore::check_schema`] whenever `data_version` +/// changes, and reopen or stop reading when it fails. +/// /// In WAL mode SQLite coordinates readers through the `-wal` and `-shm` files /// beside the database. When no other connection has the store open they may /// not exist, and SQLite creates them for this connection, which needs write @@ -85,13 +92,14 @@ pub enum ReadOnlyOpenError { /// unless those files already exist. pub struct ReadStore { store: Store, + path: PathBuf, } impl std::fmt::Debug for ReadStore { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter .debug_struct("ReadStore") - .field("path", &self.store.conn.path()) + .field("path", &self.path) .finish_non_exhaustive() } } @@ -105,40 +113,63 @@ impl ReadStore { path: path.to_path_buf(), }); } - let open_error = |source: anyhow::Error| ReadOnlyOpenError::Open { + let conn = open_read_only_connection(path).map_err(|reason| ReadOnlyOpenError::Open { + path: path.to_path_buf(), + reason, + })?; + let reader = Self { + store: Store { conn }, path: path.to_path_buf(), - source, }; - let conn = open_read_only_connection(path).map_err(open_error)?; - // A store that predates `schema_migrations` has version zero, as - // `database_schema_version` reports it. - let found = existing_schema_version(&conn) - .map_err(open_error)? - .unwrap_or(0); + reader.check_schema()?; + Ok(reader) + } + + /// The schema version recorded in the store now. + /// + /// It was [`CURRENT_SCHEMA_VERSION`] when the store opened, but a writer + /// may have migrated the store since; see [`ReadStore::check_schema`]. + /// A store that predates `schema_migrations` has version zero, as + /// [`crate::database_schema_version`] reports it. + pub fn schema_version(&self) -> Result { + // Not `Store::schema_version`: that creates `schema_migrations` and + // sets the journal mode first, which a read-only connection must not. + Ok(existing_schema_version(&self.store.conn)?.unwrap_or(0)) + } + + /// Checks again, as the open did, that the store's schema is the one this + /// binary reads. + /// + /// # Errors + /// + /// Returns [`ReadOnlyOpenError::SchemaTooOld`] or + /// [`ReadOnlyOpenError::SchemaTooNew`] when a writer has changed the + /// schema since the open, or [`ReadOnlyOpenError::Open`] when the version + /// cannot be read. + pub fn check_schema(&self) -> std::result::Result<(), ReadOnlyOpenError> { + let path = || self.path.clone(); + let found = self + .schema_version() + .map_err(|reason| ReadOnlyOpenError::Open { + path: path(), + reason, + })?; let expected = CURRENT_SCHEMA_VERSION; match found.cmp(&expected) { Ordering::Less => Err(ReadOnlyOpenError::SchemaTooOld { - path: path.to_path_buf(), + path: path(), found, expected, }), Ordering::Greater => Err(ReadOnlyOpenError::SchemaTooNew { - path: path.to_path_buf(), + path: path(), found, expected, }), - Ordering::Equal => Ok(Self { - store: Store { conn }, - }), + Ordering::Equal => Ok(()), } } - /// The store's schema version, which is [`CURRENT_SCHEMA_VERSION`] - /// whenever the open succeeded. - pub fn schema_version(&self) -> Result { - self.store.schema_version() - } - /// See [`Store::data_version`]: changes when another connection commits. pub fn data_version(&self) -> Result { self.store.data_version() @@ -268,17 +299,10 @@ impl ReadStore { } } -/// Opens `path` so that nothing this connection does can change the file. -/// -/// `SQLITE_OPEN_READONLY` is what the `mode=ro` URI parameter sets; passing the -/// flag avoids escaping the path into a URI. Without `SQLITE_OPEN_CREATE` a -/// missing file is an error rather than a new database. +/// [`snapshot::open_read_only`], set up for a reader that stays open while +/// other processes write. fn open_read_only_connection(path: &Path) -> Result { - let conn = Connection::open_with_flags( - path, - OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, - )?; - enable_sqlite_defensive(&conn)?; + let conn = snapshot::open_read_only(path)?; // A read-only connection cannot checkpoint anyway; this keeps it from // trying when it is the last one to close. conn.set_db_config(DbConfig::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, true)?; diff --git a/crates/statsai-store/src/scan_lock.rs b/crates/statsai-store/src/scan_lock.rs index dd6ed84..bd2c990 100644 --- a/crates/statsai-store/src/scan_lock.rs +++ b/crates/statsai-store/src/scan_lock.rs @@ -1,9 +1,12 @@ //! The advisory lock that keeps two scanners off one store at a time. //! -//! `statsai scan`, the scanning paths of `statsai sync`, `statsai daemon -//! --watch`, and any other process that collects into the store take this lock -//! first. It lives in a file beside the database rather than in SQLite, so a -//! process can learn that a scan is running without opening the store. +//! `statsai scan`, `statsai sync` (except `--status` and `--verify`), and each +//! pass of `statsai daemon --watch` take this lock first. Other commands that +//! write collected data, such as `statsai import`, `statsai conversation +//! collect`, and `statsai source remove --delete-data`, do not take it yet, so +//! they can still run alongside a scan; bringing them under it is a follow-up. +//! It lives in a file beside the database rather than in SQLite, so a process +//! can learn that a scan is running without opening the store. //! //! The lock is the operating system's whole-file lock: `flock` on Unix and //! `LockFileEx` on Windows, through [`File::try_lock`]. Both belong to the open @@ -27,12 +30,22 @@ const SCAN_LOCK_POLL_INTERVAL: Duration = Duration::from_millis(50); /// The scan lock for the store at `store_path`: `scan.lock` in the same /// directory, so `~/.statsai/statsai.sqlite` is guarded by /// `~/.statsai/scan.lock`. +/// +/// Symlinks are resolved, so a link to the store, or a path through a linked +/// directory, finds the same lock as the store's own path. When the store +/// does not exist yet, its directory is resolved instead, and when that does +/// not exist either, the path is used as written. Hard links to the store are +/// not detected: each directory that holds one has its own lock. #[must_use] pub fn scan_lock_path(store_path: &Path) -> PathBuf { - store_path - .parent() - .unwrap_or_else(|| Path::new("")) - .join(SCAN_LOCK_FILE_NAME) + let parent = store_path.parent().unwrap_or_else(|| Path::new("")); + let directory = match std::fs::canonicalize(store_path) { + Ok(store) => store + .parent() + .map_or_else(|| parent.to_path_buf(), Path::to_path_buf), + Err(_) => std::fs::canonicalize(parent).unwrap_or_else(|_| parent.to_path_buf()), + }; + directory.join(SCAN_LOCK_FILE_NAME) } /// Proof that this process holds the scan lock. Dropping it releases the lock. @@ -160,6 +173,35 @@ mod tests { ); } + #[test] + #[cfg(unix)] + fn a_symlink_to_the_store_shares_the_store_s_lock() { + let directory = tempfile::tempdir().expect("tempdir"); + let real_directory = directory.path().join("real"); + let link_directory = directory.path().join("link"); + std::fs::create_dir(&real_directory).expect("real directory"); + std::fs::create_dir(&link_directory).expect("link directory"); + let store = real_directory.join("statsai.sqlite"); + std::fs::write(&store, b"").expect("store file"); + let linked_store = link_directory.join("statsai.sqlite"); + std::os::unix::fs::symlink(&store, &linked_store).expect("symlink"); + + let lock = scan_lock_path(&store); + assert_eq!(scan_lock_path(&linked_store), lock); + assert_eq!( + lock, + std::fs::canonicalize(&real_directory) + .expect("canonical directory") + .join(SCAN_LOCK_FILE_NAME) + ); + let _held = try_acquire_scan_lock(&lock) + .expect("acquire") + .expect("uncontended lock"); + assert!(try_acquire_scan_lock(&scan_lock_path(&linked_store)) + .expect("try through the link") + .is_none()); + } + #[test] #[cfg(unix)] fn a_new_lock_directory_is_private_and_the_file_survives_release() { diff --git a/crates/statsai-store/src/snapshot.rs b/crates/statsai-store/src/snapshot.rs index 3098787..80a725e 100644 --- a/crates/statsai-store/src/snapshot.rs +++ b/crates/statsai-store/src/snapshot.rs @@ -389,7 +389,13 @@ fn checkpoint_wal_while_writer_is_held(connection: &Connection, source: &Path) - } } -fn open_read_only(path: &Path) -> Result { +/// Opens `path` so that nothing this connection does can change the file. +/// +/// `SQLITE_OPEN_READONLY` is what the `mode=ro` URI parameter sets; passing the +/// flag avoids escaping the path into a URI. Without `SQLITE_OPEN_CREATE` a +/// missing file is an error rather than a new database. [`crate::ReadStore`] +/// builds its connection on this one. +pub(crate) fn open_read_only(path: &Path) -> Result { let connection = Connection::open_with_flags( path, OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, diff --git a/crates/statsai-store/tests/read_only_store.rs b/crates/statsai-store/tests/read_only_store.rs index db57f9f..edcde48 100644 --- a/crates/statsai-store/tests/read_only_store.rs +++ b/crates/statsai-store/tests/read_only_store.rs @@ -394,7 +394,80 @@ fn a_file_that_is_not_a_database_is_an_open_error() { ) .expect("write junk"); - assert!(matches!(open_error(&path), ReadOnlyOpenError::Open { .. })); + let error = open_error(&path); + assert!(matches!(error, ReadOnlyOpenError::Open { .. }), "{error:?}"); + // The message says why, on its own and not again through `source`. + let message = error.to_string(); + assert!(message.contains("file is not a database"), "{message}"); + assert!(std::error::Error::source(&error).is_none()); +} + +/// Reading the version must not set the journal mode or create +/// `schema_migrations`, as the writer's version check does: on a store in +/// rollback-journal mode, or inside a read transaction, that is a write. +#[test] +fn the_schema_version_is_read_without_writing_even_from_a_rollback_journal_store() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + written_store(&path, 1); + let journal_mode: String = Connection::open(&path) + .expect("raw connection") + .query_row("PRAGMA journal_mode = DELETE", [], |row| row.get(0)) + .expect("leave WAL mode"); + assert_eq!(journal_mode, "delete"); + let before = std::fs::read(&path).expect("database bytes"); + + let reader = Store::open_read_only(&path).expect("open read-only"); + assert_eq!( + reader.schema_version().expect("schema version"), + CURRENT_SCHEMA_VERSION + ); + assert_eq!( + reader + .with_read_snapshot(|snapshot| snapshot.schema_version()) + .expect("schema version in a snapshot"), + CURRENT_SCHEMA_VERSION + ); + reader.check_schema().expect("current schema"); + read_everything(&reader); + drop(reader); + + assert_eq!( + std::fs::read(&path).expect("database bytes"), + before, + "reading the schema version changed the database file" + ); +} + +#[test] +fn check_schema_reports_a_migration_made_after_the_open() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + written_store(&path, 1); + let reader = Store::open_read_only(&path).expect("open read-only"); + reader.check_schema().expect("current schema at open"); + + let future = CURRENT_SCHEMA_VERSION + 1; + Connection::open(&path) + .expect("writer connection") + .execute( + "INSERT INTO schema_migrations (version, applied_at) VALUES (?1, '2026-10-01T00:00:00Z')", + [future], + ) + .expect("a newer binary migrates the store"); + + assert_eq!(reader.schema_version().expect("schema version"), future); + match reader.check_schema() { + Err(ReadOnlyOpenError::SchemaTooNew { + path: reported, + found, + expected, + }) => { + assert_eq!(reported, path); + assert_eq!((found, expected), (future, CURRENT_SCHEMA_VERSION)); + } + other => panic!("expected SchemaTooNew, got {other:?}"), + } } #[test] diff --git a/crates/statsai/src/cli/dispatch.rs b/crates/statsai/src/cli/dispatch.rs index ec9e9cf..564f887 100644 --- a/crates/statsai/src/cli/dispatch.rs +++ b/crates/statsai/src/cli/dispatch.rs @@ -24,19 +24,27 @@ pub(crate) fn run() -> Result<()> { Command::Service(command) => service(command), Command::Snapshot(command) => snapshot::run(command, &store_path, &device_id), command => { - // Taken before the store opens, so the migrations and repricing an - // open may run wait for a running scan as well, and held until the - // command returns. Nothing below takes it again. - let scan_lock = acquire_command_scan_lock(&command, &store_path, SCAN_LOCK_WAIT)?; - let _scan_lock = match scan_lock { + // Checked first, so a bad combination is an error even when a + // running scan turns this sync away. + if let Command::Sync(sync) = &command { + validate_sync_command(sync)?; + } + // For `scan` and `sync`, taken before the store opens, so this + // command's own open does not overlap a running scan, and held + // until the command returns. Nothing below takes it again. + let _scan_lock = match acquire_command_scan_lock(&command, &store_path, SCAN_LOCK_WAIT) + { CommandScanLock::NotNeeded => None, CommandScanLock::Held(guard) => Some(guard), CommandScanLock::Busy(lock_path) => { + return skip_for_running_scan(&command, &store_path, &lock_path); + } + CommandScanLock::Unavailable { path, error } => { eprintln!( - "Another statsai scan is running (lock: {}); skipping.", - lock_path.display() + "warning: could not use scan lock {}: {error:#}; continuing without it", + path.display() ); - return Ok(()); + None } }; let store = if command_reprices_persisted_usage(&command) { @@ -93,13 +101,14 @@ pub(crate) fn command_reprices_persisted_usage(command: &Command) -> bool { /// Whether a command scans into the store and so must hold the scan lock. /// /// `sync` refreshes git code-change scans and rebuilds daily summaries before -/// it sends; only `--status`, `--verify`, and `--reset-remote` skip that. The -/// watch daemon is not listed: it takes the lock for each pass instead of for -/// its lifetime. +/// it sends, and `--reset-remote` clears the sync tracking a running sync +/// records, so it must not interleave with one. Only `--status` and +/// `--verify`, which read, go without it. The watch daemon is not listed: it +/// takes the lock for each pass instead of for its lifetime. pub(crate) fn command_takes_scan_lock(command: &Command) -> bool { match command { Command::Scan(_) => true, - Command::Sync(sync) => !(sync.status || sync.verify || sync.reset_remote), + Command::Sync(sync) => !(sync.status || sync.verify), _ => false, } } @@ -109,6 +118,12 @@ pub(crate) enum CommandScanLock { Held(ScanLockGuard), /// Another process still held the lock at this path after the wait. Busy(PathBuf), + /// The lock file could not be opened or locked. The lock is advisory, so + /// the command runs without it rather than failing. + Unavailable { + path: PathBuf, + error: anyhow::Error, + }, } /// Takes the scan lock beside `store_path` for a command that scans, waiting @@ -117,14 +132,39 @@ pub(crate) fn acquire_command_scan_lock( command: &Command, store_path: &Path, wait: Duration, -) -> Result { +) -> CommandScanLock { if !command_takes_scan_lock(command) { - return Ok(CommandScanLock::NotNeeded); + return CommandScanLock::NotNeeded; + } + let path = statsai::scan_lock_path(store_path); + match statsai::acquire_scan_lock_with_timeout(&path, wait) { + Ok(Some(guard)) => CommandScanLock::Held(guard), + Ok(None) => CommandScanLock::Busy(path), + Err(error) => CommandScanLock::Unavailable { path, error }, + } +} + +/// What a command does instead of running while another scan holds the lock. +/// +/// A `sync` still records preference flags such as `--exclude-projects`, as +/// the run it skips would have: they apply to every later sync, and an +/// opt-out must not be lost because a scan happened to be running. It collects +/// and sends nothing. +fn skip_for_running_scan(command: &Command, store_path: &Path, lock_path: &Path) -> Result<()> { + let Command::Sync(sync) = command else { + eprintln!( + "Another statsai scan is running (lock: {}); skipping.", + lock_path.display() + ); + return Ok(()); + }; + // `--dry-run` and `--reset-remote` never record preferences. + if !sync.dry_run && !sync.reset_remote && sync_command_sets_preferences(sync) { + apply_sync_preference_overrides(&Store::open(store_path)?, sync)?; } - let lock_path = statsai::scan_lock_path(store_path); - let lock = statsai::acquire_scan_lock_with_timeout(&lock_path, wait)?; - Ok(match lock { - Some(guard) => CommandScanLock::Held(guard), - None => CommandScanLock::Busy(lock_path), - }) + eprintln!( + "Another statsai scan is running (lock: {}); skipping this sync. Nothing was sent.", + lock_path.display() + ); + Ok(()) } diff --git a/crates/statsai/src/cli/sync/mod.rs b/crates/statsai/src/cli/sync/mod.rs index cc744cb..2ace58c 100644 --- a/crates/statsai/src/cli/sync/mod.rs +++ b/crates/statsai/src/cli/sync/mod.rs @@ -62,7 +62,20 @@ pub(crate) fn effective_sync_preferences( Ok(preferences.normalized()) } -fn apply_sync_preference_overrides( +/// Whether the command carries any flag that changes the stored sync +/// preferences. +pub(crate) fn sync_command_sets_preferences(command: &SyncCommand) -> bool { + command.include_projects + || command.exclude_projects + || command.include_tasks + || command.exclude_tasks + || command.include_activity + || command.exclude_activity + || command.include_sessions + || command.exclude_sessions +} + +pub(crate) fn apply_sync_preference_overrides( store: &Store, command: &SyncCommand, ) -> Result { @@ -133,17 +146,25 @@ fn apply_sync_preference_overrides( Ok(preferences) } -pub(crate) fn sync(command: SyncCommand, store: &Store, device_id: &str) -> Result<()> { +/// Rejects flag combinations that cannot run together. It needs no store, so +/// the caller can check before waiting for the scan lock, and a sync that a +/// running scan turns away still reports them. +pub(crate) fn validate_sync_command(command: &SyncCommand) -> Result<()> { if command.since_last && (command.full || command.rebuild_rollups) { bail!("--since-last cannot be combined with --full or --rebuild-rollups"); } + if command.reset_remote && (command.status || command.verify) { + bail!("--reset-remote cannot be combined with --status or --verify"); + } + Ok(()) +} + +pub(crate) fn sync(command: SyncCommand, store: &Store, device_id: &str) -> Result<()> { + validate_sync_command(&command)?; let sync_preferences = effective_sync_preferences(store, &command)?; if command.reset_remote { - if command.status || command.verify { - bail!("--reset-remote cannot be combined with --status or --verify"); - } return sync_remote_reset(command, store); } diff --git a/crates/statsai/src/cli/tests/scan_lock.rs b/crates/statsai/src/cli/tests/scan_lock.rs index 61bfe7f..177eef2 100644 --- a/crates/statsai/src/cli/tests/scan_lock.rs +++ b/crates/statsai/src/cli/tests/scan_lock.rs @@ -18,9 +18,17 @@ fn scanning_commands_take_the_scan_lock_and_others_do_not() { "--since-last" ])); assert!(takes(&["statsai", "sync", "--dry-run"])); + // It clears the sync tracking a running sync records. + assert!(takes(&[ + "statsai", + "sync", + "--sink", + "http", + "--reset-remote", + "--yes" + ])); assert!(!takes(&["statsai", "sync", "--status"])); assert!(!takes(&["statsai", "sync", "--verify"])); - assert!(!takes(&["statsai", "sync", "--reset-remote", "--yes"])); // The watch daemon takes it per pass, not for its lifetime. assert!(!takes(&["statsai", "daemon", "--watch"])); assert!(!takes(&["statsai", "report", "weekly"])); @@ -38,23 +46,18 @@ fn a_scanning_command_reports_a_busy_lock_after_waiting() { .expect("uncontended lock"); let started = std::time::Instant::now(); - match acquire_command_scan_lock(&parse(&["statsai", "scan"]), &store_path, wait) - .expect("scan lock") - { + match acquire_command_scan_lock(&parse(&["statsai", "scan"]), &store_path, wait) { CommandScanLock::Busy(reported) => assert_eq!(reported, lock_path), _ => panic!("scan must see the lock as busy"), } assert!(started.elapsed() >= wait, "scan gave up without waiting"); assert!(matches!( - acquire_command_scan_lock(&parse(&["statsai", "status"]), &store_path, wait) - .expect("no lock needed"), + acquire_command_scan_lock(&parse(&["statsai", "status"]), &store_path, wait), CommandScanLock::NotNeeded )); drop(other_scanner); - let held = match acquire_command_scan_lock(&parse(&["statsai", "sync"]), &store_path, wait) - .expect("scan lock") - { + let held = match acquire_command_scan_lock(&parse(&["statsai", "sync"]), &store_path, wait) { CommandScanLock::Held(guard) => guard, _ => panic!("sync must take the free lock"), }; @@ -63,3 +66,28 @@ fn a_scanning_command_reports_a_busy_lock_after_waiting() { .expect("try") .is_none()); } + +#[test] +fn a_scan_lock_that_cannot_be_opened_is_reported_rather_than_fatal() { + let directory = tempfile::tempdir().expect("tempdir"); + // A regular file where the store's directory should be: the lock file + // beneath it can be neither created nor opened. + let not_a_directory = directory.path().join("not-a-directory"); + std::fs::write(¬_a_directory, b"").expect("regular file"); + let store_path = not_a_directory.join("statsai.sqlite"); + + match acquire_command_scan_lock( + &parse(&["statsai", "scan"]), + &store_path, + StdDuration::from_millis(100), + ) { + CommandScanLock::Unavailable { path, error } => { + assert_eq!(path, statsai::scan_lock_path(&store_path)); + assert!( + format!("{error:#}").contains("scan lock"), + "the error names the lock: {error:#}" + ); + } + _ => panic!("an unusable lock must be reported as unavailable"), + } +} diff --git a/crates/statsai/src/lib.rs b/crates/statsai/src/lib.rs index 0c9bf97..b571b23 100644 --- a/crates/statsai/src/lib.rs +++ b/crates/statsai/src/lib.rs @@ -15,10 +15,12 @@ use std::path::{Path, PathBuf}; /// The lock that keeps scanners of one store from running at once. /// -/// `statsai scan`, the scanning paths of `statsai sync`, and each pass of -/// `statsai daemon --watch` hold it; any other process that collects into the -/// store should take it too. It is implemented in `statsai-store` so the -/// daemon crate can share it, and re-exported here beside the default paths. +/// `statsai scan`, `statsai sync` (except `--status` and `--verify`), and each +/// pass of `statsai daemon --watch` hold it. `statsai import`, `statsai +/// conversation collect`, and `statsai source remove --delete-data` also write +/// collected data but do not take it yet. It is implemented in `statsai-store` +/// so the daemon crate can share it, and re-exported here beside the default +/// paths. pub use statsai_store::{ acquire_scan_lock_with_timeout, scan_lock_path, try_acquire_scan_lock, ScanLockGuard, SCAN_LOCK_FILE_NAME, @@ -434,12 +436,21 @@ mod tests { #[test] fn the_default_scan_lock_sits_beside_the_default_store() { let lock = default_scan_lock_path(); - assert_eq!(lock.parent(), default_store_path().parent()); assert_eq!( lock.file_name(), Some(std::ffi::OsStr::new(SCAN_LOCK_FILE_NAME)) ); - assert!(lock.ends_with(".statsai/scan.lock")); + // The same directory, though `scan_lock_path` resolves symlinks in it + // once it exists. + let resolved = |directory: &Path| { + std::fs::canonicalize(directory).unwrap_or_else(|_| directory.to_path_buf()) + }; + let store_directory = default_store_path(); + let store_directory = store_directory.parent().expect("store directory"); + assert_eq!( + resolved(lock.parent().expect("lock directory")), + resolved(store_directory) + ); } #[test] diff --git a/crates/statsai/tests/scan_lock_cli.rs b/crates/statsai/tests/scan_lock_cli.rs index 9c7fdb2..ddf90c7 100644 --- a/crates/statsai/tests/scan_lock_cli.rs +++ b/crates/statsai/tests/scan_lock_cli.rs @@ -1,6 +1,7 @@ //! `statsai scan` and `statsai sync` wait for a running scanner, then skip //! without failing. +use statsai_store::{Store, SyncPreferences}; use std::path::Path; use std::process::{Command, Output}; use std::time::Duration; @@ -50,11 +51,18 @@ fn scan_and_sync_skip_cleanly_while_another_scan_holds_the_lock() { batch.to_str().expect("utf-8 path"), ], ); - let expected = format!( - "Another statsai scan is running (lock: {}); skipping.", + let busy = format!( + "Another statsai scan is running (lock: {});", lock.display() ); - for (name, child) in [("scan", scan), ("sync", sync)] { + for (name, child, expected) in [ + ("scan", scan, format!("{busy} skipping.")), + ( + "sync", + sync, + format!("{busy} skipping this sync. Nothing was sent."), + ), + ] { let (output, stderr) = finish(child); assert!( stderr.contains(&expected), @@ -69,6 +77,78 @@ fn scan_and_sync_skip_cleanly_while_another_scan_holds_the_lock() { assert!(!batch.exists(), "a skipped sync must not have sent a batch"); } +#[test] +fn an_invalid_sync_is_an_error_even_while_another_scan_holds_the_lock() { + let directory = tempfile::tempdir().expect("tempdir"); + let store = directory.path().join("statsai.sqlite"); + let _other_scanner = statsai::try_acquire_scan_lock(&statsai::scan_lock_path(&store)) + .expect("acquire") + .expect("uncontended lock"); + + let output = run_statsai( + directory.path(), + &store, + &["sync", "--since-last", "--full"], + ) + .wait_with_output() + .expect("statsai output"); + let stderr = String::from_utf8_lossy(&output.stderr); + + assert!( + !output.status.success(), + "an invalid sync succeeded: {stderr}" + ); + assert!( + stderr.contains("--since-last cannot be combined with --full or --rebuild-rollups"), + "{stderr}" + ); + assert!( + !stderr.contains("Another statsai scan is running"), + "{stderr}" + ); +} + +#[test] +fn a_skipped_sync_still_records_a_preference_opt_out() { + let directory = tempfile::tempdir().expect("tempdir"); + let store_path = directory.path().join("statsai.sqlite"); + Store::open(&store_path) + .expect("create store") + .set_sync_preferences(SyncPreferences { + include_projects: true, + ..SyncPreferences::default() + }) + .expect("opt in to project sync"); + let lock = statsai::scan_lock_path(&store_path); + let _other_scanner = statsai::try_acquire_scan_lock(&lock) + .expect("acquire") + .expect("uncontended lock"); + + let sync = run_statsai( + directory.path(), + &store_path, + &["sync", "--sink", "stdout", "--exclude-projects"], + ); + let (output, stderr) = finish(sync); + + assert!( + stderr.contains(&format!( + "Another statsai scan is running (lock: {}); skipping this sync. Nothing was sent.", + lock.display() + )), + "{stderr}" + ); + assert!(output.stdout.is_empty(), "a skipped sync sent a batch"); + let preferences = Store::open(&store_path) + .expect("reopen store") + .sync_preferences() + .expect("sync preferences"); + assert!( + !preferences.include_projects, + "the --exclude-projects opt-out was lost: {preferences:?}" + ); +} + #[test] fn scan_waits_for_a_short_scan_to_finish_and_then_runs() { let directory = tempfile::tempdir().expect("tempdir"); From 107e50137eaf224c0d7ba66530ff8bd46ded25a2 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:50:52 +0000 Subject: [PATCH 3/6] STA-5: Fail a turned-away remote reset and check the schema on every read A `sync --reset-remote` that a running scan turns away now fails, having deleted nothing, instead of exiting 0 as a skipped sync does. Its flag errors (`--sink http` required, `--yes` unless a dry run) are checked before the lock wait. `--reset-remote --dry-run` only reads and no longer takes the lock. Every ReadStore read now checks the schema version as the first read of its own transaction, or joins a with_read_snapshot that did. A reader left open while a newer statsai migrates the store gets SchemaTooNew instead of misreading it, and no migration can land between the check and the read. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G3EZeUTrrXQYg1NwFdgV3K --- README.md | 20 +++--- crates/statsai-store/src/read_only.rs | 70 ++++++++++++------- crates/statsai-store/tests/read_only_store.rs | 30 ++++++++ crates/statsai/src/cli/dispatch.rs | 24 +++++-- crates/statsai/src/cli/sync/mod.rs | 20 +++--- crates/statsai/src/cli/tests/scan_lock.rs | 9 +++ crates/statsai/tests/scan_lock_cli.rs | 43 +++++++++++- 7 files changed, 164 insertions(+), 52 deletions(-) diff --git a/README.md b/README.md index 8afebcc..fd50dd8 100644 --- a/README.md +++ b/README.md @@ -432,13 +432,14 @@ statsai report cache --from 2026-05-01 --details Normal scans use a per-source file signature cache. `--no-cache` forces a one-off reread; `--replace` performs a destructive source rebuild. -`statsai scan`, `statsai sync` (other than `--status` and `--verify`), and each -pass of `statsai daemon --watch` take turns through `scan.lock` beside the -database (`~/.statsai/scan.lock` by default). While another of them holds it, -`scan` and `sync` wait up to 5 seconds, then print `Another statsai scan is -running (lock: …); skipping.` (for `sync`, `…; skipping this sync. Nothing was -sent.`) and exit successfully. A skipped `sync` still saves its -`--include-*`/`--exclude-*` preference flags. The daemon keeps changed files +`statsai scan`, `statsai sync` (other than `--status`, `--verify`, and +`--reset-remote --dry-run`), and each pass of `statsai daemon --watch` take +turns through `scan.lock` beside the database (`~/.statsai/scan.lock` by +default). While another of them holds it, `scan` and `sync` wait up to 5 +seconds, then print `Another statsai scan is running (lock: …); skipping.` (for +`sync`, `…; skipping this sync. Nothing was sent.`) and exit successfully. A +skipped `sync` still saves its `--include-*`/`--exclude-*` preference flags. +`sync --reset-remote` fails instead, having deleted nothing. The daemon keeps changed files queued until the lock is free. The lock is advisory: if its file cannot be opened or locked, these commands warn and run without it. `statsai import`, `conversation collect`, and `source remove --delete-data` do not take it yet. @@ -526,7 +527,10 @@ never creates, migrates, backfills, reprices, or changes the permissions of the file, and returns a `ReadStore` with only the read APIs: session, daily, and cache reports, quota and list queries, `data_version`, and `with_read_snapshot`. A missing store, or one whose schema is older or newer -than the binary's, is a typed `ReadOnlyOpenError` the caller can act on. A +than the binary's, is a typed `ReadOnlyOpenError` the caller can act on. Each +read checks the schema again in its own snapshot, so a reader left open while a +newer `statsai` migrates the store gets that error (inside the `anyhow::Error`) +instead of misreading the new schema. A process that collects into the store should hold the scan lock, via `statsai::try_acquire_scan_lock` or `statsai::acquire_scan_lock_with_timeout` with `statsai::default_scan_lock_path()`. diff --git a/crates/statsai-store/src/read_only.rs b/crates/statsai-store/src/read_only.rs index 22bd274..4c78267 100644 --- a/crates/statsai-store/src/read_only.rs +++ b/crates/statsai-store/src/read_only.rs @@ -79,10 +79,11 @@ pub enum ReadOnlyOpenError { /// [`ReadStore::with_read_snapshot`] to see one point in time, and compare /// [`ReadStore::data_version`] to notice that a writer has committed since. /// -/// The schema version is checked when the store opens, but a writer such as a -/// newer `statsai` may migrate the store while it stays open. A long-lived -/// reader should call [`ReadStore::check_schema`] whenever `data_version` -/// changes, and reopen or stop reading when it fails. +/// A writer such as a newer `statsai` may migrate the store while it stays +/// open, so the schema version is checked again at the start of each of those +/// transactions, in the same snapshot as the reads. A read of a store whose +/// schema has changed since the open fails with a [`ReadOnlyOpenError`] that +/// `anyhow::Error::downcast_ref` recovers; reopen, or stop reading. /// /// In WAL mode SQLite coordinates readers through the `-wal` and `-shm` files /// beside the database. When no other connection has the store open they may @@ -138,7 +139,8 @@ impl ReadStore { } /// Checks again, as the open did, that the store's schema is the one this - /// binary reads. + /// binary reads. Every read already does this; this checks without + /// reading anything else. /// /// # Errors /// @@ -175,25 +177,41 @@ impl ReadStore { self.store.data_version() } - /// Runs several reads against one point-in-time view. See + /// Runs several reads against one point-in-time view, after checking that + /// the schema in that view is still the one this binary reads. See /// [`Store::with_read_snapshot`]. pub fn with_read_snapshot(&self, operation: impl FnOnce(&Self) -> Result) -> Result { - self.store.with_read_snapshot(|_| operation(self)) + self.read(|_| operation(self)) + } + + /// Runs `operation` in a read transaction that first checks the schema. + /// The check is the transaction's first read, so it and `operation` see + /// the same snapshot and a migration cannot land between them. Inside + /// [`ReadStore::with_read_snapshot`] it joins that transaction, which has + /// already checked. + fn read(&self, operation: impl FnOnce(&Store) -> Result) -> Result { + if !self.store.conn.is_autocommit() { + return operation(&self.store); + } + self.store.with_read_snapshot(|store| { + self.check_schema()?; + operation(store) + }) } /// The pricing ruleset the store's costs were last computed with. A reader /// never reprices, so this can trail the binary's own ruleset until a /// writer such as `statsai scan` next opens the store. pub fn applied_pricing_ruleset_version(&self) -> Result> { - self.store.applied_pricing_ruleset_version() + self.read(|store| store.applied_pricing_ruleset_version()) } pub fn event_count(&self) -> Result { - self.store.event_count() + self.read(|store| store.event_count()) } pub fn token_total(&self) -> Result { - self.store.token_total() + self.read(|store| store.token_total()) } /// See [`Store::session_rollups_in_period`]. @@ -206,8 +224,9 @@ impl ReadStore { limit: usize, offset: usize, ) -> Result> { - self.store - .session_rollups_in_period(since, until, filter, sort, limit, offset) + self.read(|store| { + store.session_rollups_in_period(since, until, filter, sort, limit, offset) + }) } /// See [`Store::session_stats_in_period`]. @@ -217,7 +236,7 @@ impl ReadStore { until: DateTime, filter: &SessionFilter, ) -> Result { - self.store.session_stats_in_period(since, until, filter) + self.read(|store| store.session_stats_in_period(since, until, filter)) } /// Every daily summary as stored. Unlike @@ -225,7 +244,7 @@ impl ReadStore { /// are returned as they are rather than rebuilt; the next writer that /// reads them through [`Store`] rebuilds them. pub fn all_sync_rollup_summaries(&self) -> Result> { - self.store.stored_sync_rollup_summaries() + self.read(|store| store.stored_sync_rollup_summaries()) } /// See [`Store::daily_rollups_between`]. @@ -234,7 +253,7 @@ impl ReadStore { start_date: &str, end_date: &str, ) -> Result> { - self.store.daily_rollups_between(start_date, end_date) + self.read(|store| store.daily_rollups_between(start_date, end_date)) } /// See [`Store::cache_report`]. @@ -246,17 +265,17 @@ impl ReadStore { where Tz::Offset: std::fmt::Display, { - self.store.cache_report(query, timeline_zone) + self.read(|store| store.cache_report(query, timeline_zone)) } /// See [`Store::quota_status`]. pub fn quota_status(&self, query: &QuotaQuery) -> Result { - self.store.quota_status(query) + self.read(|store| store.quota_status(query)) } /// See [`Store::quota_windows`]. pub fn quota_windows(&self, query: &QuotaQuery) -> Result> { - self.store.quota_windows(query) + self.read(|store| store.quota_windows(query)) } /// See [`Store::quota_observations`]. @@ -265,7 +284,7 @@ impl ReadStore { query: &QuotaQuery, collapse_duplicates: bool, ) -> Result> { - self.store.quota_observations(query, collapse_duplicates) + self.read(|store| store.quota_observations(query, collapse_duplicates)) } /// See [`Store::weekly_reset_anchor`]. @@ -274,28 +293,27 @@ impl ReadStore { provider: &str, provider_account_id: &ProviderAccountId, ) -> Result> { - self.store - .weekly_reset_anchor(provider, provider_account_id) + self.read(|store| store.weekly_reset_anchor(provider, provider_account_id)) } pub fn list_sources(&self) -> Result> { - self.store.list_sources() + self.read(|store| store.list_sources()) } pub fn list_accounts(&self) -> Result> { - self.store.list_accounts() + self.read(|store| store.list_accounts()) } pub fn list_source_account_assignments(&self) -> Result> { - self.store.list_source_account_assignments() + self.read(|store| store.list_source_account_assignments()) } pub fn list_subscriptions(&self) -> Result> { - self.store.list_subscriptions() + self.read(|store| store.list_subscriptions()) } pub fn list_sync_states(&self) -> Result> { - self.store.list_sync_states() + self.read(|store| store.list_sync_states()) } } diff --git a/crates/statsai-store/tests/read_only_store.rs b/crates/statsai-store/tests/read_only_store.rs index edcde48..bd0e1b7 100644 --- a/crates/statsai-store/tests/read_only_store.rs +++ b/crates/statsai-store/tests/read_only_store.rs @@ -470,6 +470,36 @@ fn check_schema_reports_a_migration_made_after_the_open() { } } +#[test] +fn every_read_refuses_a_store_migrated_after_the_open() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = store_path(&directory); + written_store(&path, 1); + let reader = Store::open_read_only(&path).expect("open read-only"); + assert_eq!(reader.event_count().expect("read at open"), 1); + + let future = CURRENT_SCHEMA_VERSION + 1; + Connection::open(&path) + .expect("writer connection") + .execute( + "INSERT INTO schema_migrations (version, applied_at) VALUES (?1, '2026-10-01T00:00:00Z')", + [future], + ) + .expect("a newer binary migrates the store"); + + let refused = |error: anyhow::Error| match error.downcast_ref::() { + Some(ReadOnlyOpenError::SchemaTooNew { found, .. }) => assert_eq!(*found, future), + _ => panic!("expected SchemaTooNew, got {error:#}"), + }; + refused(reader.event_count().expect_err("a single read")); + refused(reader.list_sources().expect_err("a list read")); + refused( + reader + .with_read_snapshot(|reader| reader.event_count()) + .expect_err("a read snapshot"), + ); +} + #[test] fn reads_proceed_during_a_write_and_see_it_once_committed() { let directory = tempfile::tempdir().expect("tempdir"); diff --git a/crates/statsai/src/cli/dispatch.rs b/crates/statsai/src/cli/dispatch.rs index 564f887..c50d18d 100644 --- a/crates/statsai/src/cli/dispatch.rs +++ b/crates/statsai/src/cli/dispatch.rs @@ -1,4 +1,4 @@ -use anyhow::Result; +use anyhow::{bail, Result}; use clap::Parser; use statsai::{default_device_id, default_store_path, snapshot, ScanLockGuard}; use statsai_store::Store; @@ -102,13 +102,13 @@ pub(crate) fn command_reprices_persisted_usage(command: &Command) -> bool { /// /// `sync` refreshes git code-change scans and rebuilds daily summaries before /// it sends, and `--reset-remote` clears the sync tracking a running sync -/// records, so it must not interleave with one. Only `--status` and -/// `--verify`, which read, go without it. The watch daemon is not listed: it -/// takes the lock for each pass instead of for its lifetime. +/// records, so it must not interleave with one. Only `--status`, `--verify` +/// and `--reset-remote --dry-run`, which read, go without it. The watch daemon +/// is not listed: it takes the lock for each pass instead of for its lifetime. pub(crate) fn command_takes_scan_lock(command: &Command) -> bool { match command { Command::Scan(_) => true, - Command::Sync(sync) => !(sync.status || sync.verify), + Command::Sync(sync) => !(sync.status || sync.verify || (sync.reset_remote && sync.dry_run)), _ => false, } } @@ -150,6 +150,10 @@ pub(crate) fn acquire_command_scan_lock( /// the run it skips would have: they apply to every later sync, and an /// opt-out must not be lost because a scan happened to be running. It collects /// and sends nothing. +/// +/// `sync --reset-remote` fails instead. It is a one-off the user asked for, +/// and a script that resets and then resyncs must not carry on as if the reset +/// had happened. fn skip_for_running_scan(command: &Command, store_path: &Path, lock_path: &Path) -> Result<()> { let Command::Sync(sync) = command else { eprintln!( @@ -158,8 +162,14 @@ fn skip_for_running_scan(command: &Command, store_path: &Path, lock_path: &Path) ); return Ok(()); }; - // `--dry-run` and `--reset-remote` never record preferences. - if !sync.dry_run && !sync.reset_remote && sync_command_sets_preferences(sync) { + if sync.reset_remote { + bail!( + "Another statsai scan is running (lock: {}); --reset-remote did not run and nothing was deleted. Retry when it finishes.", + lock_path.display() + ); + } + // `--dry-run` never records preferences. + if !sync.dry_run && sync_command_sets_preferences(sync) { apply_sync_preference_overrides(&Store::open(store_path)?, sync)?; } eprintln!( diff --git a/crates/statsai/src/cli/sync/mod.rs b/crates/statsai/src/cli/sync/mod.rs index 2ace58c..d6cfc53 100644 --- a/crates/statsai/src/cli/sync/mod.rs +++ b/crates/statsai/src/cli/sync/mod.rs @@ -156,6 +156,14 @@ pub(crate) fn validate_sync_command(command: &SyncCommand) -> Result<()> { if command.reset_remote && (command.status || command.verify) { bail!("--reset-remote cannot be combined with --status or --verify"); } + if command.reset_remote && command.sink != "http" { + bail!("--reset-remote is currently supported only with --sink http"); + } + if command.reset_remote && !command.dry_run && !command.yes { + bail!( + "--reset-remote deletes mirrored hosted sync data for this paired device; rerun with --yes" + ); + } Ok(()) } @@ -405,11 +413,9 @@ fn maybe_reset_http_sync_tracking_if_remote_changed( Ok(None) } +/// Runs `--reset-remote`. [`validate_sync_command`] has already required +/// `--sink http`, and `--yes` unless this is a dry run. fn sync_remote_reset(command: SyncCommand, store: &Store) -> Result<()> { - if command.sink != "http" { - bail!("--reset-remote is currently supported only with --sink http"); - } - let endpoint = http_sync_endpoint(&command)?; if command.dry_run { println!( @@ -427,12 +433,6 @@ fn sync_remote_reset(command: SyncCommand, store: &Store) -> Result<()> { return Ok(()); } - if !command.yes { - bail!( - "--reset-remote deletes mirrored hosted sync data for this paired device; rerun with --yes" - ); - } - eprintln!( "warning: --reset-remote deletes mirrored hosted sync data for this paired device. Other paired devices are not affected." ); diff --git a/crates/statsai/src/cli/tests/scan_lock.rs b/crates/statsai/src/cli/tests/scan_lock.rs index 177eef2..24a98a9 100644 --- a/crates/statsai/src/cli/tests/scan_lock.rs +++ b/crates/statsai/src/cli/tests/scan_lock.rs @@ -27,6 +27,15 @@ fn scanning_commands_take_the_scan_lock_and_others_do_not() { "--reset-remote", "--yes" ])); + // A dry run only prints what it would reset. + assert!(!takes(&[ + "statsai", + "sync", + "--sink", + "http", + "--reset-remote", + "--dry-run" + ])); assert!(!takes(&["statsai", "sync", "--status"])); assert!(!takes(&["statsai", "sync", "--verify"])); // The watch daemon takes it per pass, not for its lifetime. diff --git a/crates/statsai/tests/scan_lock_cli.rs b/crates/statsai/tests/scan_lock_cli.rs index ddf90c7..d33a617 100644 --- a/crates/statsai/tests/scan_lock_cli.rs +++ b/crates/statsai/tests/scan_lock_cli.rs @@ -1,5 +1,5 @@ //! `statsai scan` and `statsai sync` wait for a running scanner, then skip -//! without failing. +//! without failing, except `sync --reset-remote`, which fails. use statsai_store::{Store, SyncPreferences}; use std::path::Path; @@ -108,6 +108,47 @@ fn an_invalid_sync_is_an_error_even_while_another_scan_holds_the_lock() { ); } +#[test] +fn a_remote_reset_fails_rather_than_skips_while_another_scan_holds_the_lock() { + let directory = tempfile::tempdir().expect("tempdir"); + let store = directory.path().join("statsai.sqlite"); + let lock = statsai::scan_lock_path(&store); + let _other_scanner = statsai::try_acquire_scan_lock(&lock) + .expect("acquire") + .expect("uncontended lock"); + + // The first two are flag errors, reported before any wait. + for (args, expected) in [ + ( + &["sync", "--reset-remote", "--yes"][..], + "--reset-remote is currently supported only with --sink http".to_owned(), + ), + ( + &["sync", "--sink", "http", "--reset-remote"][..], + "rerun with --yes".to_owned(), + ), + ( + &["sync", "--sink", "http", "--reset-remote", "--yes"][..], + format!( + "Another statsai scan is running (lock: {}); --reset-remote did not run and nothing was deleted.", + lock.display() + ), + ), + ] { + let output = run_statsai(directory.path(), &store, args) + .wait_with_output() + .expect("statsai output"); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + !output.status.success(), + "{args:?} succeeded while the lock was busy: {stderr}" + ); + assert!(stderr.contains(&expected), "{args:?}: {stderr}"); + assert!(output.stdout.is_empty(), "{args:?} wrote output"); + } + assert!(!store.exists(), "a turned-away reset opened the store"); +} + #[test] fn a_skipped_sync_still_records_a_preference_opt_out() { let directory = tempfile::tempdir().expect("tempdir"); From 6c5e77abe70b1bc417deaa3280aff96d6971b832 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 16:55:21 +0000 Subject: [PATCH 4/6] STA-5: Drop a sync assertion a turned-away sync can break; list every lock exemption A sync that a running scan turns away still saves its preference flags, so the running sync can find stored preferences that differ from the ones it read at the start. The debug_assert_eq that assumed otherwise would panic in debug builds; it is removed. The scan_lock module and the statsai re-export docs now also name `--reset-remote --dry-run` among the syncs that don't take the lock, matching the README. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G3EZeUTrrXQYg1NwFdgV3K --- crates/statsai-store/src/scan_lock.rs | 5 +++-- crates/statsai/src/cli/sync/mod.rs | 5 +++-- crates/statsai/src/lib.rs | 5 +++-- 3 files changed, 9 insertions(+), 6 deletions(-) diff --git a/crates/statsai-store/src/scan_lock.rs b/crates/statsai-store/src/scan_lock.rs index bd2c990..35acc4b 100644 --- a/crates/statsai-store/src/scan_lock.rs +++ b/crates/statsai-store/src/scan_lock.rs @@ -1,7 +1,8 @@ //! The advisory lock that keeps two scanners off one store at a time. //! -//! `statsai scan`, `statsai sync` (except `--status` and `--verify`), and each -//! pass of `statsai daemon --watch` take this lock first. Other commands that +//! `statsai scan`, `statsai sync` (except `--status`, `--verify`, and +//! `--reset-remote --dry-run`), and each pass of `statsai daemon --watch` take +//! this lock first. Other commands that //! write collected data, such as `statsai import`, `statsai conversation //! collect`, and `statsai source remove --delete-data`, do not take it yet, so //! they can still run alongside a scan; bringing them under it is a follow-up. diff --git a/crates/statsai/src/cli/sync/mod.rs b/crates/statsai/src/cli/sync/mod.rs index d6cfc53..2dfc249 100644 --- a/crates/statsai/src/cli/sync/mod.rs +++ b/crates/statsai/src/cli/sync/mod.rs @@ -255,8 +255,9 @@ pub(crate) fn sync(command: SyncCommand, store: &Store, device_id: &str) -> Resu return Ok(()); } - let persisted_sync_preferences = apply_sync_preference_overrides(store, &command)?; - debug_assert_eq!(persisted_sync_preferences, sync_preferences); + // The stored preferences can differ from `sync_preferences` by now: a + // sync that this run's scan lock turned away still saves its flags. + apply_sync_preference_overrides(store, &command)?; reached_send.set(true); let result = (|| -> Result<()> { diff --git a/crates/statsai/src/lib.rs b/crates/statsai/src/lib.rs index b571b23..82554aa 100644 --- a/crates/statsai/src/lib.rs +++ b/crates/statsai/src/lib.rs @@ -15,8 +15,9 @@ use std::path::{Path, PathBuf}; /// The lock that keeps scanners of one store from running at once. /// -/// `statsai scan`, `statsai sync` (except `--status` and `--verify`), and each -/// pass of `statsai daemon --watch` hold it. `statsai import`, `statsai +/// `statsai scan`, `statsai sync` (except `--status`, `--verify`, and +/// `--reset-remote --dry-run`), and each pass of `statsai daemon --watch` hold +/// it. `statsai import`, `statsai /// conversation collect`, and `statsai source remove --delete-data` also write /// collected data but do not take it yet. It is implemented in `statsai-store` /// so the daemon crate can share it, and re-exported here beside the default From a1ee0d9cf51da48823db71ac2bdd57bf3c37c837 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 17:00:20 +0000 Subject: [PATCH 5/6] STA-5: Give every process the same scan lock path for a store scan_lock_path fell back to the path as written when the store's directory did not exist yet, and resolved it once it did. Through a linked directory such as macOS's /var -> /private/var the two spellings differ, which failed the macOS CLI test. It is one file, but the path a command prints depended on creation order. A dangling symlink at the store path could split the lock outright: its own directory before the store existed, the target's after. The longest existing prefix of the store's directory is now canonicalized and the missing directories appended as written, a symlink at the store path is followed even before its target exists, and a bare file name resolves against the current directory. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G3EZeUTrrXQYg1NwFdgV3K --- crates/statsai-store/src/scan_lock.rs | 126 +++++++++++++++++++++++--- 1 file changed, 111 insertions(+), 15 deletions(-) diff --git a/crates/statsai-store/src/scan_lock.rs b/crates/statsai-store/src/scan_lock.rs index 35acc4b..34abd79 100644 --- a/crates/statsai-store/src/scan_lock.rs +++ b/crates/statsai-store/src/scan_lock.rs @@ -32,21 +32,67 @@ const SCAN_LOCK_POLL_INTERVAL: Duration = Duration::from_millis(50); /// directory, so `~/.statsai/statsai.sqlite` is guarded by /// `~/.statsai/scan.lock`. /// -/// Symlinks are resolved, so a link to the store, or a path through a linked -/// directory, finds the same lock as the store's own path. When the store -/// does not exist yet, its directory is resolved instead, and when that does -/// not exist either, the path is used as written. Hard links to the store are -/// not detected: each directory that holds one has its own lock. +/// Every process that names one store must get the same lock file, however it +/// spells the path and whatever exists yet: +/// +/// - A symlink at the store path is followed, even before its target exists, +/// so a link to the store shares the store's lock. +/// - The longest part of the directory that exists is resolved, and the +/// directories that don't exist yet are appended as written, so the path is +/// the same before and after a scanner creates them, including through a +/// linked directory such as macOS's `/var` → `/private/var`. +/// - A bare file name is taken relative to the current directory. +/// +/// Hard links to the store are not detected: each directory that holds one has +/// its own lock. #[must_use] pub fn scan_lock_path(store_path: &Path) -> PathBuf { - let parent = store_path.parent().unwrap_or_else(|| Path::new("")); - let directory = match std::fs::canonicalize(store_path) { - Ok(store) => store - .parent() - .map_or_else(|| parent.to_path_buf(), Path::to_path_buf), - Err(_) => std::fs::canonicalize(parent).unwrap_or_else(|_| parent.to_path_buf()), + let store = follow_symlinks(store_path); + let directory = match store.parent() { + Some(parent) if !parent.as_os_str().is_empty() => parent, + _ => Path::new("."), }; - directory.join(SCAN_LOCK_FILE_NAME) + resolve_existing_prefix(directory).join(SCAN_LOCK_FILE_NAME) +} + +/// `path` with any symlink at its last component followed, even one whose +/// target does not exist. +fn follow_symlinks(path: &Path) -> PathBuf { + // Bounded, so a link cycle cannot loop; the open reports those. + const MAX_LINKS: usize = 40; + let mut path = path.to_path_buf(); + for _ in 0..MAX_LINKS { + let Ok(target) = std::fs::read_link(&path) else { + break; + }; + // An absolute target replaces the parent in `join`. + path = path + .parent() + .map_or_else(|| target.clone(), |parent| parent.join(&target)); + } + path +} + +/// `path` with its longest existing prefix canonicalized and the rest appended +/// as written, or `path` itself when no prefix resolves. +fn resolve_existing_prefix(path: &Path) -> PathBuf { + let mut missing = Vec::new(); + let mut existing = path; + loop { + if let Ok(resolved) = std::fs::canonicalize(existing) { + return missing + .iter() + .rev() + .fold(resolved, |resolved, name| resolved.join(name)); + } + match (existing.parent(), existing.file_name()) { + (Some(parent), Some(name)) => { + missing.push(name); + existing = parent; + } + _ => return path.to_path_buf(), + } + } } /// Proof that this process holds the scan lock. Dropping it releases the lock. @@ -164,14 +210,64 @@ mod tests { #[test] fn the_lock_lives_beside_the_store() { + let root = std::fs::canonicalize("/").expect("root"); assert_eq!( - scan_lock_path(Path::new("/home/me/.statsai/statsai.sqlite")), - PathBuf::from("/home/me/.statsai/scan.lock") + scan_lock_path(Path::new("/statsai-test-missing/.statsai/statsai.sqlite")), + root.join("statsai-test-missing") + .join(".statsai") + .join(SCAN_LOCK_FILE_NAME) ); assert_eq!( scan_lock_path(Path::new("statsai.sqlite")), - PathBuf::from("scan.lock") + std::fs::canonicalize(".") + .expect("current directory") + .join(SCAN_LOCK_FILE_NAME) + ); + } + + #[test] + #[cfg(unix)] + fn the_lock_path_is_the_same_before_and_after_the_store_directory_exists() { + let directory = tempfile::tempdir().expect("tempdir"); + let real_directory = directory.path().join("real"); + let link_directory = directory.path().join("link"); + std::fs::create_dir(&real_directory).expect("real directory"); + std::os::unix::fs::symlink(&real_directory, &link_directory).expect("symlink"); + let store = link_directory.join("data").join("statsai.sqlite"); + + let before = scan_lock_path(&store); + assert_eq!( + before, + std::fs::canonicalize(&real_directory) + .expect("canonical directory") + .join("data") + .join(SCAN_LOCK_FILE_NAME) ); + std::fs::create_dir(link_directory.join("data")).expect("data directory"); + assert_eq!(scan_lock_path(&store), before); + std::fs::write(&store, b"").expect("store file"); + assert_eq!(scan_lock_path(&store), before); + } + + #[test] + #[cfg(unix)] + fn a_symlink_to_a_store_not_yet_created_shares_the_store_s_lock() { + let directory = tempfile::tempdir().expect("tempdir"); + let real_directory = directory.path().join("real"); + let link_directory = directory.path().join("link"); + std::fs::create_dir(&real_directory).expect("real directory"); + std::fs::create_dir(&link_directory).expect("link directory"); + let store = real_directory.join("statsai.sqlite"); + let linked_store = link_directory.join("statsai.sqlite"); + // Relative, and dangling until the store is created. + std::os::unix::fs::symlink(Path::new("../real/statsai.sqlite"), &linked_store) + .expect("symlink"); + + let lock = scan_lock_path(&store); + assert_eq!(scan_lock_path(&linked_store), lock); + std::fs::write(&store, b"").expect("store file"); + assert_eq!(scan_lock_path(&linked_store), lock); + assert_eq!(scan_lock_path(&store), lock); } #[test] From f10f02f2a328886e5f20ffa0ce67f54a55a5d54d Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 10 Oct 2026 17:12:59 +0000 Subject: [PATCH 6/6] STA-5: Type only schema mismatches on reads; reject an unknown sink up front - A read whose schema check fails to read the version (a busy store, an I/O error) now returns that error with context, not a ReadOnlyOpenError::Open that claims the open failed. Only SchemaTooOld and SchemaTooNew are typed. - ReadStore opens through a snapshot helper without its own path context, so the Open message names the path once instead of three times. - The ReadStore docs and README no longer promise that a migration still in progress is caught: a migration records its version after its schema changes. - validate_sync_command rejects an unknown --sink, so `sync --sink htpp` fails whether or not a running scan would have turned it away. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01G3EZeUTrrXQYg1NwFdgV3K --- README.md | 6 +- crates/statsai-store/src/read_only.rs | 103 ++++++++++++++++++++++---- crates/statsai-store/src/snapshot.rs | 14 +++- crates/statsai/src/cli/sync/mod.rs | 7 +- crates/statsai/tests/scan_lock_cli.rs | 41 +++++----- 5 files changed, 128 insertions(+), 43 deletions(-) diff --git a/README.md b/README.md index fd50dd8..047ee7b 100644 --- a/README.md +++ b/README.md @@ -528,9 +528,9 @@ file, and returns a `ReadStore` with only the read APIs: session, daily, and cache reports, quota and list queries, `data_version`, and `with_read_snapshot`. A missing store, or one whose schema is older or newer than the binary's, is a typed `ReadOnlyOpenError` the caller can act on. Each -read checks the schema again in its own snapshot, so a reader left open while a -newer `statsai` migrates the store gets that error (inside the `anyhow::Error`) -instead of misreading the new schema. A +read checks the schema version again in its own snapshot, so a reader left open +while a newer `statsai` migrates the store gets that error (inside the +`anyhow::Error`) once the migration has recorded its version. A process that collects into the store should hold the scan lock, via `statsai::try_acquire_scan_lock` or `statsai::acquire_scan_lock_with_timeout` with `statsai::default_scan_lock_path()`. diff --git a/crates/statsai-store/src/read_only.rs b/crates/statsai-store/src/read_only.rs index 4c78267..173758a 100644 --- a/crates/statsai-store/src/read_only.rs +++ b/crates/statsai-store/src/read_only.rs @@ -5,7 +5,7 @@ use super::{ QuotaQuery, QuotaStatus, SessionFilter, SessionSort, SessionStats, Store, SyncState, WeeklyResetAnchor, CURRENT_SCHEMA_VERSION, SQLITE_BUSY_TIMEOUT, }; -use anyhow::Result; +use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use rusqlite::config::DbConfig; use rusqlite::Connection; @@ -81,9 +81,11 @@ pub enum ReadOnlyOpenError { /// /// A writer such as a newer `statsai` may migrate the store while it stays /// open, so the schema version is checked again at the start of each of those -/// transactions, in the same snapshot as the reads. A read of a store whose -/// schema has changed since the open fails with a [`ReadOnlyOpenError`] that -/// `anyhow::Error::downcast_ref` recovers; reopen, or stop reading. +/// transactions, in the same snapshot as the reads. Once a migration has +/// recorded its version, a read fails with a [`ReadOnlyOpenError::SchemaTooNew`] +/// (or `SchemaTooOld`) that `anyhow::Error::downcast_ref` recovers; reopen, or +/// stop reading. A migration records its version after its schema changes, so +/// a read that lands while one is still running is not caught. /// /// In WAL mode SQLite coordinates readers through the `-wal` and `-shm` files /// beside the database. When no other connection has the store open they may @@ -149,26 +151,32 @@ impl ReadStore { /// schema since the open, or [`ReadOnlyOpenError::Open`] when the version /// cannot be read. pub fn check_schema(&self) -> std::result::Result<(), ReadOnlyOpenError> { - let path = || self.path.clone(); let found = self .schema_version() .map_err(|reason| ReadOnlyOpenError::Open { - path: path(), + path: self.path.clone(), reason, })?; + self.schema_mismatch(found).map_or(Ok(()), Err) + } + + /// The error for a store at schema version `found`, if this binary does + /// not read that version. + fn schema_mismatch(&self, found: i64) -> Option { + let path = self.path.clone(); let expected = CURRENT_SCHEMA_VERSION; match found.cmp(&expected) { - Ordering::Less => Err(ReadOnlyOpenError::SchemaTooOld { - path: path(), + Ordering::Less => Some(ReadOnlyOpenError::SchemaTooOld { + path, found, expected, }), - Ordering::Greater => Err(ReadOnlyOpenError::SchemaTooNew { - path: path(), + Ordering::Greater => Some(ReadOnlyOpenError::SchemaTooNew { + path, found, expected, }), - Ordering::Equal => Ok(()), + Ordering::Equal => None, } } @@ -194,7 +202,14 @@ impl ReadStore { return operation(&self.store); } self.store.with_read_snapshot(|store| { - self.check_schema()?; + // Only a mismatch is a `ReadOnlyOpenError`. A failure to read the + // version, such as a busy or I/O error, is the read's own error. + let found = self + .schema_version() + .context("read the store's schema version")?; + if let Some(mismatch) = self.schema_mismatch(found) { + return Err(mismatch.into()); + } operation(store) }) } @@ -317,10 +332,10 @@ impl ReadStore { } } -/// [`snapshot::open_read_only`], set up for a reader that stays open while -/// other processes write. +/// [`snapshot::open_read_only_unlabelled`], set up for a reader that stays open +/// while other processes write. [`ReadOnlyOpenError::Open`] names the path. fn open_read_only_connection(path: &Path) -> Result { - let conn = snapshot::open_read_only(path)?; + let conn = snapshot::open_read_only_unlabelled(path)?; // A read-only connection cannot checkpoint anyway; this keeps it from // trying when it is the last one to close. conn.set_db_config(DbConfig::SQLITE_DBCONFIG_NO_CKPT_ON_CLOSE, true)?; @@ -334,6 +349,64 @@ fn open_read_only_connection(path: &Path) -> Result { mod tests { use super::*; + #[test] + fn a_busy_store_is_the_read_s_own_error_not_a_schema_error() { + let directory = tempfile::tempdir().expect("tempdir"); + let path = directory.path().join("statsai.sqlite"); + drop(Store::open(&path).expect("create store")); + let writer = Connection::open(&path).expect("writer"); + writer + .execute_batch("PRAGMA journal_mode = DELETE;") + .expect("rollback journal"); + let reader = Store::open_read_only(&path).expect("open read-only"); + reader + .store + .conn + .busy_timeout(std::time::Duration::ZERO) + .expect("no busy wait"); + + // An exclusive lock in rollback-journal mode keeps every reader out. + writer + .execute_batch("BEGIN EXCLUSIVE;") + .expect("exclusive lock"); + let error = reader.event_count().expect_err("the store is locked"); + assert!( + error.downcast_ref::().is_none(), + "a busy store was reported as a schema or open error: {error:#}" + ); + assert!( + format!("{error:#}").contains("read the store's schema version"), + "{error:#}" + ); + + writer.execute_batch("COMMIT;").expect("release"); + reader.event_count().expect("read once the lock is gone"); + } + + #[test] + fn a_failed_open_names_the_path_once() { + let directory = tempfile::tempdir().expect("tempdir"); + // A directory exists, so it gets past the missing-file check, but + // SQLite cannot open it. + let path = directory.path().join("statsai.sqlite"); + std::fs::create_dir(&path).expect("directory"); + + let error = Store::open_read_only(&path).expect_err("not a database file"); + assert!(matches!(error, ReadOnlyOpenError::Open { .. }), "{error:?}"); + let message = error.to_string(); + assert!( + message.starts_with(&format!( + "open StatsAI store read-only at {}: ", + path.display() + )), + "{message}" + ); + assert!( + !message.contains("open database read-only at"), + "the snapshot open's context repeats the path: {message}" + ); + } + #[test] fn the_read_only_connection_refuses_writes() { let directory = tempfile::tempdir().expect("tempdir"); diff --git a/crates/statsai-store/src/snapshot.rs b/crates/statsai-store/src/snapshot.rs index 80a725e..ae14a0a 100644 --- a/crates/statsai-store/src/snapshot.rs +++ b/crates/statsai-store/src/snapshot.rs @@ -393,14 +393,20 @@ fn checkpoint_wal_while_writer_is_held(connection: &Connection, source: &Path) - /// /// `SQLITE_OPEN_READONLY` is what the `mode=ro` URI parameter sets; passing the /// flag avoids escaping the path into a URI. Without `SQLITE_OPEN_CREATE` a -/// missing file is an error rather than a new database. [`crate::ReadStore`] -/// builds its connection on this one. +/// missing file is an error rather than a new database. pub(crate) fn open_read_only(path: &Path) -> Result { + open_read_only_unlabelled(path) + .with_context(|| format!("open database read-only at {}", path.display())) +} + +/// [`open_read_only`] without the context naming the path, for a caller whose +/// own error already names it. [`crate::ReadStore`] builds its connection on +/// this one. +pub(crate) fn open_read_only_unlabelled(path: &Path) -> Result { let connection = Connection::open_with_flags( path, OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX, - ) - .with_context(|| format!("open database read-only at {}", path.display()))?; + )?; enable_sqlite_defensive(&connection)?; Ok(connection) } diff --git a/crates/statsai/src/cli/sync/mod.rs b/crates/statsai/src/cli/sync/mod.rs index 2dfc249..663a115 100644 --- a/crates/statsai/src/cli/sync/mod.rs +++ b/crates/statsai/src/cli/sync/mod.rs @@ -146,10 +146,15 @@ pub(crate) fn apply_sync_preference_overrides( Ok(preferences) } -/// Rejects flag combinations that cannot run together. It needs no store, so +/// Rejects an unknown sink and flag combinations that cannot run together, so +/// a bad command fails the same way whether or not it would wait. It needs no +/// store, so /// the caller can check before waiting for the scan lock, and a sync that a /// running scan turns away still reports them. pub(crate) fn validate_sync_command(command: &SyncCommand) -> Result<()> { + if !matches!(command.sink.as_str(), "stdout" | "file" | "http") { + bail!("unsupported sync sink {}", command.sink); + } if command.since_last && (command.full || command.rebuild_rollups) { bail!("--since-last cannot be combined with --full or --rebuild-rollups"); } diff --git a/crates/statsai/tests/scan_lock_cli.rs b/crates/statsai/tests/scan_lock_cli.rs index d33a617..8f78f5d 100644 --- a/crates/statsai/tests/scan_lock_cli.rs +++ b/crates/statsai/tests/scan_lock_cli.rs @@ -85,27 +85,28 @@ fn an_invalid_sync_is_an_error_even_while_another_scan_holds_the_lock() { .expect("acquire") .expect("uncontended lock"); - let output = run_statsai( - directory.path(), - &store, - &["sync", "--since-last", "--full"], - ) - .wait_with_output() - .expect("statsai output"); - let stderr = String::from_utf8_lossy(&output.stderr); + for (args, expected) in [ + ( + &["sync", "--since-last", "--full"][..], + "--since-last cannot be combined with --full or --rebuild-rollups", + ), + ( + &["sync", "--sink", "htpp"][..], + "unsupported sync sink htpp", + ), + ] { + let output = run_statsai(directory.path(), &store, args) + .wait_with_output() + .expect("statsai output"); + let stderr = String::from_utf8_lossy(&output.stderr); - assert!( - !output.status.success(), - "an invalid sync succeeded: {stderr}" - ); - assert!( - stderr.contains("--since-last cannot be combined with --full or --rebuild-rollups"), - "{stderr}" - ); - assert!( - !stderr.contains("Another statsai scan is running"), - "{stderr}" - ); + assert!(!output.status.success(), "{args:?} succeeded: {stderr}"); + assert!(stderr.contains(expected), "{args:?}: {stderr}"); + assert!( + !stderr.contains("Another statsai scan is running"), + "{args:?}: {stderr}" + ); + } } #[test]