Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

29 changes: 29 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -432,6 +432,18 @@ 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`, `--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.

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.
Expand Down Expand Up @@ -506,6 +518,23 @@ 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. Each
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()`.

### Rust SDK

`crates/statsai-sdk` exposes a facade for embedding collection and reporting in
Expand Down
11 changes: 10 additions & 1 deletion crates/statsai-daemon/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -284,20 +284,29 @@ pub fn watch_and_serve(
_store: Arc<Mutex<Store>>,
_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, 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,
store: Arc<Mutex<Store>>,
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<SocketAddr> {
Expand Down
58 changes: 50 additions & 8 deletions crates/statsai-daemon/src/watch/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -32,12 +32,30 @@ 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)
};
/// 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,
store: Arc<Mutex<Store>>,
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();
Expand Down Expand Up @@ -91,6 +109,7 @@ pub fn watch_and_serve(
pending_changed_paths,
startup_executable,
bind_addr,
scan_lock,
)
}
#[cfg(not(target_os = "macos"))]
Expand All @@ -114,6 +133,7 @@ pub fn watch_and_serve(
pending_changed_paths,
startup_executable,
bind_addr,
scan_lock,
)
}
}
Expand Down Expand Up @@ -146,11 +166,13 @@ fn run_watch_loop<W: Watcher>(
pending_changed_paths: Arc<Mutex<PendingWatch>>,
startup_executable: Option<ExecutableStamp>,
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()));
Expand All @@ -162,6 +184,7 @@ fn run_watch_loop<W: Watcher>(
.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()
Expand All @@ -174,11 +197,12 @@ fn run_watch_loop<W: Watcher>(
.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,
Expand All @@ -205,13 +229,31 @@ fn run_watch_loop<W: Watcher>(
}
},
);
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")?;
Expand Down
101 changes: 85 additions & 16 deletions crates/statsai-daemon/src/watch/scan.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -43,27 +43,96 @@ 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, 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<Mutex<PendingWatch>>,
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(&notice) {
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 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(&notice);
if let Some(guard) = guard {
drop(guard);
std::thread::sleep(super::WATCH_SCAN_LOCK_YIELD);
}
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
}

/// 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<Mutex<PendingWatch>>,
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(
Expand Down
Loading
Loading