diff --git a/Cargo.lock b/Cargo.lock index 7acc1ce..f69067a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1548,6 +1548,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs4" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8640e34b88f7652208ce9e88b1a37a2ae95227d84abec377ccd3c5cfeb141ed4" +dependencies = [ + "rustix", + "windows-sys 0.59.0", +] + [[package]] name = "fs_extra" version = "1.3.0" @@ -1817,6 +1827,7 @@ dependencies = [ "dashmap", "data-encoding", "dotenvy", + "fs4", "futures", "futures-util", "hex", @@ -5517,6 +5528,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets 0.52.6", +] + [[package]] name = "windows-sys" version = "0.60.2" diff --git a/Cargo.toml b/Cargo.toml index 880e414..6fb5132 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,6 +60,7 @@ regex = "1.12.3" semver = "1.0" async-stream = "0.3.6" blake3 = "1" +fs4 = "0.13" hkdf = "0.13" hmac = "0.13" sqlparser = "0.62.0" diff --git a/packages/docs/content/docs/api-reference/admin/admin-api.md b/packages/docs/content/docs/api-reference/admin/admin-api.md index 4794315..0cc8f56 100644 --- a/packages/docs/content/docs/api-reference/admin/admin-api.md +++ b/packages/docs/content/docs/api-reference/admin/admin-api.md @@ -115,6 +115,9 @@ Each admin API endpoint requires a specific permission. See the [Permissions gui | `GET /admin/records/collections` | `records:read` | | `DELETE /admin/records` | `records:delete` | | `DELETE /admin/records/collection` | `records:delete-collection`| +| `GET /admin/database/status` | `settings:manage` | +| `POST /admin/database/vacuum/schedule` | `settings:manage` | +| `DELETE /admin/database/vacuum/schedule` | `settings:manage` | | `GET /admin/settings` | `settings:manage` | | `GET /admin/settings/db-info` | `settings:manage` | | `PUT /admin/settings/{key}` | `settings:manage` | @@ -169,3 +172,7 @@ Each admin API endpoint requires a specific permission. See the [Permissions gui | `GET /admin/permissions` | `users:read` | | `GET /admin/settings/xrpc-proxy` | `settings:manage` | | `PUT /admin/settings/xrpc-proxy` | `settings:manage` | + +### Database disk usage and vacuum + +`GET /admin/database/status` reports the database and write-ahead log size, free disk space, and the status of the one-time SQLite vacuum described in [Configuration](../../getting-started/configuration.md). `POST /admin/database/vacuum/schedule` arms the vacuum to run at the next restart; `DELETE /admin/database/vacuum/schedule` disarms it. These only apply to SQLite — on a Postgres-backed instance, scheduling returns `400 Bad Request` rather than silently doing nothing. diff --git a/packages/docs/content/docs/getting-started/configuration.md b/packages/docs/content/docs/getting-started/configuration.md index 07e06e6..921e4f2 100644 --- a/packages/docs/content/docs/getting-started/configuration.md +++ b/packages/docs/content/docs/getting-started/configuration.md @@ -10,6 +10,7 @@ HappyView is configured via environment variables. A `.env` file in the project |----------|----------|---------|-------------| | `DATABASE_URL` | yes | --- | Database connection string. SQLite (`sqlite://path/to/db?mode=rwc`) or Postgres (`postgres://user:pass@host/db`) | | `DATABASE_BACKEND` | no | auto-detected | Force `sqlite` or `postgres`. Auto-detected from `DATABASE_URL` scheme if not set | +| `SQLITE_JOURNAL_SIZE_LIMIT` | no | `67108864` (64 MiB) | Bytes. Caps how large the SQLite write-ahead log is allowed to grow before it's truncated back down after a checkpoint, bounding disk usage from WAL growth. SQLite-only; ignored on Postgres | | `PUBLIC_URL` | yes | --- | Public-facing URL for HappyView (used for OAuth callbacks, e.g. `https://happyview.example.com`). **For local development, use `http://127.0.0.1:3000` — not `localhost`** (see note below). Do **not** include the base path — see `BASE_PATH` | | `BASE_PATH` | no | _(none)_ | Subpath prefix for mounting HappyView behind a reverse proxy (e.g. `/hv`). Must start with `/` and have no trailing slash. When set, all routes are served under this prefix and the dashboard is accessible at `https://example.com/hv/`. See [Reverse proxy subpath](deployment/production.md#reverse-proxy-subpath) | | `SESSION_SECRET` | no | dev default | Secret key for signing session cookies (at least 64 characters). **Must be set in production** | @@ -57,6 +58,7 @@ SESSION_SECRET=change-me-in-production # BASE_PATH=/hv # HOST=0.0.0.0 # PORT=3000 +# SQLITE_JOURNAL_SIZE_LIMIT=67108864 # JETSTREAM_URL=wss://jetstream1.us-east.bsky.network # RELAY_URL=https://bsky.network # PLC_URL=https://plc.directory diff --git a/src/admin/database.rs b/src/admin/database.rs new file mode 100644 index 0000000..d51abee --- /dev/null +++ b/src/admin/database.rs @@ -0,0 +1,83 @@ +//! `/admin/database` — database size, free space, and the one-time vacuum. + +use axum::Json; +use axum::extract::State; +use serde_json::json; + +use crate::AppState; +use crate::db::DatabaseBackend; +use crate::error::AppError; +use crate::maintenance::{disk, vacuum}; + +use super::auth::UserAuth; +use super::permissions::Permission; + +/// `GET /admin/database/status` +pub(super) async fn status( + State(state): State, + auth: UserAuth, +) -> Result, AppError> { + auth.require(Permission::SettingsManage).await?; + + // `disk::report` runs several blocking filesystem syscalls (`stat`, + // `statvfs`) — if the data directory or `TMPDIR` sits on a wedged mount, + // those can block uninterruptibly. Run it on the blocking pool so a stuck + // mount can't park a runtime worker thread. + let db_url = state.config.database_url.clone(); + let report = tokio::task::spawn_blocking(move || disk::report(&db_url)) + .await + .map_err(|e| AppError::Internal(format!("disk report task panicked: {e}")))?; + let feasibility = report.as_ref().map(|r| { + disk::feasibility( + r.db_bytes + r.wal_bytes, + r.db_fs_free, + r.temp_fs_free, + &r.db_path, + &r.temp_path, + r.same_filesystem, + ) + }); + let status = vacuum::read_status(&state.db, state.db_backend).await?; + + Ok(Json(json!({ + "backend": match state.db_backend { + DatabaseBackend::Sqlite => "sqlite", + DatabaseBackend::Postgres => "postgres", + }, + "disk": report, + "feasibility": feasibility, + "vacuum": status, + "journal_size_limit": state.config.sqlite_journal_size_limit, + }))) +} + +/// `POST /admin/database/vacuum/schedule` +/// +/// Rejected outright on Postgres: a VACUUM here is inapplicable, and arming +/// `requested_at` anyway would let an admin schedule something that +/// `vacuum::run_if_requested` (SQLite-only) will silently never run, leaving +/// `/database/status` reporting "scheduled" forever with nothing behind it. +/// A clear 400 up front beats that silent no-op. +pub(super) async fn schedule_vacuum( + State(state): State, + auth: UserAuth, +) -> Result, AppError> { + auth.require(Permission::SettingsManage).await?; + if state.db_backend != DatabaseBackend::Sqlite { + return Err(AppError::BadRequest( + "vacuum is only applicable to SQLite instances".into(), + )); + } + vacuum::request(&state.db, state.db_backend).await?; + Ok(Json(json!({ "scheduled": true }))) +} + +/// `DELETE /admin/database/vacuum/schedule` +pub(super) async fn cancel_vacuum( + State(state): State, + auth: UserAuth, +) -> Result, AppError> { + auth.require(Permission::SettingsManage).await?; + vacuum::cancel_request(&state.db, state.db_backend).await?; + Ok(Json(json!({ "scheduled": false }))) +} diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 553305b..38d28c3 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -2,6 +2,7 @@ mod api_clients; mod api_keys; pub(crate) mod auth; pub mod backfill; +mod database; mod dead_letters; mod domains; mod events; @@ -116,6 +117,11 @@ pub fn admin_routes(_state: AppState) -> Router { "/records/collection", delete(records::delete_collection_records), ) + .route("/database/status", get(database::status)) + .route( + "/database/vacuum/schedule", + post(database::schedule_vacuum).delete(database::cancel_vacuum), + ) .route( "/network-lexicons", post(network_lexicons::add).get(network_lexicons::list), diff --git a/src/admin/records.rs b/src/admin/records.rs index 9cfa6ad..ef1db2c 100644 --- a/src/admin/records.rs +++ b/src/admin/records.rs @@ -177,27 +177,38 @@ pub(super) struct DeleteCollectionParams { pub collection: String, } -/// DELETE /admin/records/collection?collection=X — delete all records in a collection. +/// DELETE /admin/records/collection?collection=X — enqueue deletion of every +/// record in a collection. +/// +/// Returns `202` with a job id rather than deleting inline: a large collection +/// takes long enough that a synchronous request would time out, and running it +/// as a job gives progress, pause and cancel through the existing jobs UI. pub(super) async fn delete_collection_records( State(state): State, auth: UserAuth, Query(params): Query, -) -> Result, AppError> { +) -> Result<(StatusCode, Json), AppError> { auth.require(Permission::RecordsDeleteCollection).await?; auth.require(Permission::RecordsDelete).await?; - let backend = state.db_backend; - let sql = adapt_sql( - "DELETE FROM happyview_records WHERE collection = ?", - backend, - ); - let result = crate::db::query(&sql) - .bind(¶ms.collection) - .execute(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to delete records: {e}")))?; - Ok(Json( - serde_json::json!({ "deleted": result.rows_affected() }), + if params.collection.trim().is_empty() { + return Err(AppError::BadRequest("collection is required".into())); + } + + let job_id = crate::jobs::db::create_job( + &state, + "happyview.delete-collection", + &serde_json::json!({ "collection": params.collection }), + &auth.did, + false, + None, + None, + ) + .await?; + + Ok(( + StatusCode::ACCEPTED, + Json(serde_json::json!({ "job_id": job_id })), )) } diff --git a/src/config.rs b/src/config.rs index 3a6827c..462ea72 100644 --- a/src/config.rs +++ b/src/config.rs @@ -69,6 +69,7 @@ pub struct Config { pub port: u16, pub database_url: String, pub database_backend: DatabaseBackend, + pub sqlite_journal_size_limit: u64, pub public_url: String, pub session_secret: String, pub jetstream_url: String, @@ -102,6 +103,7 @@ impl Config { .unwrap_or(3000), database_url, database_backend, + sqlite_journal_size_limit: crate::db::journal_size_limit_bytes(), public_url: env::var("PUBLIC_URL").expect("PUBLIC_URL must be set"), // Not required and never defaulted to a placeholder: an unset, // insecure, or too-short value is surfaced via `config_errors()` and @@ -238,6 +240,7 @@ mod tests { port: 8080, database_url: String::new(), database_backend: DatabaseBackend::Postgres, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: String::new(), jetstream_url: String::new(), @@ -565,6 +568,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: DatabaseBackend::Postgres, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: "https://example.com".into(), session_secret: String::new(), jetstream_url: String::new(), @@ -591,6 +595,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: DatabaseBackend::Postgres, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: "https://example.com".into(), session_secret: String::new(), jetstream_url: String::new(), @@ -617,6 +622,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: DatabaseBackend::Postgres, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: "https://example.com/".into(), session_secret: String::new(), jetstream_url: String::new(), @@ -643,6 +649,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: DatabaseBackend::Postgres, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: String::new(), jetstream_url: String::new(), @@ -672,6 +679,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: DatabaseBackend::Postgres, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: String::new(), jetstream_url: String::new(), diff --git a/src/db.rs b/src/db.rs index ee79fac..fa0d3b9 100644 --- a/src/db.rs +++ b/src/db.rs @@ -279,6 +279,56 @@ pub fn decode_cursor(cursor: &str) -> Option<(String, String)> { Some((ts.to_string(), uri.to_string())) } +/// Default `journal_size_limit`: 64 MiB. Caps how large the `-wal` file is left +/// after a checkpoint. Without it the WAL stays at its high-water mark forever, +/// so a single large delete permanently inflates disk usage. +pub const DEFAULT_JOURNAL_SIZE_LIMIT: u64 = 67_108_864; + +/// Parse `SQLITE_JOURNAL_SIZE_LIMIT`. `-1` means "no limit" and is represented +/// as `u64::MAX`; anything unparseable falls back to the default rather than +/// failing boot over a maintenance knob. +pub fn parse_journal_size_limit(raw: Option<&str>) -> u64 { + match raw { + None => DEFAULT_JOURNAL_SIZE_LIMIT, + Some("-1") => u64::MAX, + Some(s) => s + .trim() + .parse::() + .unwrap_or(DEFAULT_JOURNAL_SIZE_LIMIT), + } +} + +/// Resolve the configured journal size limit from the environment. +pub fn journal_size_limit_bytes() -> u64 { + parse_journal_size_limit(std::env::var("SQLITE_JOURNAL_SIZE_LIMIT").ok().as_deref()) +} + +/// Apply per-connection SQLite pragmas to every pooled connection. +/// +/// `journal_size_limit` is per-connection and not persisted in the database +/// file, so unlike `journal_mode` it cannot be set once against the pool — that +/// would reach only whichever single connection served the statement. +fn with_sqlite_pragmas( + opts: PoolOptions, + backend: DatabaseBackend, +) -> PoolOptions { + if backend != DatabaseBackend::Sqlite { + return opts; + } + let limit = journal_size_limit_bytes(); + opts.after_connect(move |conn, _meta| { + Box::pin(async move { + let sql = if limit == u64::MAX { + "PRAGMA journal_size_limit = -1".to_string() + } else { + format!("PRAGMA journal_size_limit = {limit}") + }; + crate::db::query(&sql).execute(&mut *conn).await?; + Ok(()) + }) + }) +} + /// Connect to the configured database and run migrations. pub async fn connect(url: &str, backend: DatabaseBackend) -> AnyPool { sqlx::any::install_default_drivers(); @@ -305,16 +355,28 @@ pub async fn connect(url: &str, backend: DatabaseBackend) -> AnyPool { DatabaseBackend::Postgres => 32, }); - let pool = PoolOptions::::new() - .max_connections(max_connections) - .acquire_timeout(std::time::Duration::from_secs(10)) - .idle_timeout(std::time::Duration::from_secs(300)) - .connect(url) - .await - .expect("Failed to connect to database"); + let pool = with_sqlite_pragmas( + PoolOptions::::new() + .max_connections(max_connections) + .acquire_timeout(std::time::Duration::from_secs(10)) + .idle_timeout(std::time::Duration::from_secs(300)), + backend, + ) + .connect(url) + .await + .expect("Failed to connect to database"); // Enable foreign keys and WAL mode for SQLite if backend == DatabaseBackend::Sqlite { + // Must run before any table exists: on a populated database this is a + // silent no-op until a full VACUUM rebuilds the file. New instances get + // incremental vacuum from birth; existing ones need the one-time repair + // in `maintenance::vacuum`. + crate::db::query("PRAGMA auto_vacuum = INCREMENTAL") + .execute(&pool) + .await + .expect("Failed to set auto_vacuum"); + crate::db::query("PRAGMA foreign_keys = ON") .execute(&pool) .await @@ -346,6 +408,19 @@ pub async fn connect(url: &str, backend: DatabaseBackend) -> AnyPool { pool } +/// Extract the filesystem path from a `sqlite://` URL, dropping query params. +/// Returns `None` for non-SQLite URLs and for in-memory databases. +pub fn sqlite_path_from_url(url: &str) -> Option { + let rest = url + .strip_prefix("sqlite://") + .or_else(|| url.strip_prefix("sqlite:"))?; + let path = rest.split('?').next().unwrap_or(rest); + if path.is_empty() || path == ":memory:" { + return None; + } + Some(std::path::PathBuf::from(path)) +} + pub fn backfill_pool_ceiling(backend: DatabaseBackend) -> u32 { match backend { DatabaseBackend::Sqlite => 64, @@ -390,13 +465,16 @@ pub async fn connect_backfill_pool(url: &str, backend: DatabaseBackend) -> AnyPo tracing::info!(max_connections, "backfill pool sized"); - let pool = PoolOptions::::new() - .max_connections(max_connections) - .acquire_timeout(std::time::Duration::from_secs(30)) - .idle_timeout(std::time::Duration::from_secs(300)) - .connect(url) - .await - .expect("Failed to connect backfill database pool"); + let pool = with_sqlite_pragmas( + PoolOptions::::new() + .max_connections(max_connections) + .acquire_timeout(std::time::Duration::from_secs(30)) + .idle_timeout(std::time::Duration::from_secs(300)), + backend, + ) + .connect(url) + .await + .expect("Failed to connect backfill database pool"); if backend == DatabaseBackend::Sqlite { crate::db::query("PRAGMA foreign_keys = ON") @@ -422,6 +500,7 @@ pub async fn connect_backfill_pool(url: &str, backend: DatabaseBackend) -> AnyPo mod tests { use super::*; use chrono::Datelike; + use serial_test::serial; const CAST_ON_READ: &[&str] = &[ "service_identity.setup_complete", @@ -859,4 +938,96 @@ mod tests { assert_eq!(dt.year(), 2025); assert_eq!(dt.month(), 3); } + + // ----------------------------------------------------------------------- + // journal_size_limit + // ----------------------------------------------------------------------- + + #[test] + fn journal_size_limit_defaults_to_64_mib() { + assert_eq!(parse_journal_size_limit(None), 67_108_864); + } + + #[test] + fn journal_size_limit_parses_env_value() { + assert_eq!(parse_journal_size_limit(Some("1048576")), 1_048_576); + } + + #[test] + fn journal_size_limit_rejects_garbage_and_uses_default() { + assert_eq!(parse_journal_size_limit(Some("banana")), 67_108_864); + assert_eq!(parse_journal_size_limit(Some("")), 67_108_864); + } + + #[test] + fn journal_size_limit_allows_unlimited_sentinel() { + // -1 disables the limit; it is the one negative value SQLite accepts. + assert_eq!(parse_journal_size_limit(Some("-1")), u64::MAX); + } + + // ----------------------------------------------------------------------- + // sqlite_path_from_url + // ----------------------------------------------------------------------- + + #[test] + fn sqlite_path_from_url_strips_scheme_and_params() { + assert_eq!( + sqlite_path_from_url("sqlite://data/happyview.db?mode=rwc"), + Some(std::path::PathBuf::from("data/happyview.db")) + ); + assert_eq!( + sqlite_path_from_url("sqlite:/var/lib/hv.db"), + Some(std::path::PathBuf::from("/var/lib/hv.db")) + ); + assert_eq!(sqlite_path_from_url("postgres://localhost/hv"), None); + assert_eq!(sqlite_path_from_url("sqlite://:memory:"), None); + } + + // ----------------------------------------------------------------------- + // with_sqlite_pragmas: journal_size_limit actually reaches pooled + // connections, not just the one `connect()` happens to touch first. + // ----------------------------------------------------------------------- + + #[tokio::test] + #[serial] + async fn journal_size_limit_pragma_reaches_pooled_connections() { + let path = + std::env::temp_dir().join(format!("hv-journal-pragma-{}.db", uuid::Uuid::new_v4())); + let url = format!("sqlite://{}?mode=rwc", path.display()); + + // SAFETY: this test is `#[serial]`; no other test reads or mutates + // this variable concurrently. + unsafe { + std::env::set_var("SQLITE_JOURNAL_SIZE_LIMIT", "1048576"); + } + let pool = connect(&url, DatabaseBackend::Sqlite).await; + unsafe { + std::env::remove_var("SQLITE_JOURNAL_SIZE_LIMIT"); + } + + // Hold several connections open simultaneously so the pool has to + // actually establish more than one, proving `after_connect` runs on + // each rather than on a single lucky connection reused by every + // acquire. + let mut conns = Vec::new(); + for _ in 0..4 { + conns.push(pool.acquire().await.expect("failed to acquire connection")); + } + + for mut conn in conns { + let (limit,): (i64,) = crate::db::query_as("PRAGMA journal_size_limit") + .fetch_one(&mut *conn) + .await + .expect("failed to read journal_size_limit"); + assert_eq!( + limit, 1_048_576, + "journal_size_limit pragma did not reach a pooled connection" + ); + } + + drop(pool); + let _ = std::fs::remove_file(&path); + let _ = std::fs::remove_file(format!("{}-wal", path.display())); + let _ = std::fs::remove_file(format!("{}-shm", path.display())); + } } diff --git a/src/jobs/mod.rs b/src/jobs/mod.rs index fd2bf24..1c8200c 100644 --- a/src/jobs/mod.rs +++ b/src/jobs/mod.rs @@ -1,5 +1,6 @@ pub(crate) mod db; pub(crate) mod logs; +pub mod native; pub mod worker; use serde::{Deserialize, Serialize}; diff --git a/src/jobs/native/delete_collection.rs b/src/jobs/native/delete_collection.rs new file mode 100644 index 0000000..2920a6f --- /dev/null +++ b/src/jobs/native/delete_collection.rs @@ -0,0 +1,117 @@ +//! `happyview.delete-collection` — delete every record in a collection. +//! +//! Batched deliberately. One unbounded `DELETE` is a single transaction whose +//! every dirtied page accumulates in the WAL until it commits, which is how a +//! large delete inflates disk usage instead of reducing it. Committing per batch +//! releases the write lock (so ingest can interleave) and lets the WAL +//! checkpoint between batches. + +use crate::AppState; +use crate::db::{DatabaseBackend, adapt_sql}; + +use super::super::{Job, db}; +use super::NativeOutcome; + +/// Rows per transaction. +const BATCH_SIZE: i64 = 5000; + +pub async fn run(state: &AppState, job: &Job) -> NativeOutcome { + let collection = match job.input.get("collection").and_then(|v| v.as_str()) { + Some(c) if !c.is_empty() => c.to_string(), + _ => return NativeOutcome::Failed("input.collection is required".into()), + }; + let backend = state.db_backend; + + let count_sql = adapt_sql( + "SELECT COUNT(*) FROM happyview_records WHERE collection = ?", + backend, + ); + let total: i64 = match crate::db::query_as::<(i64,)>(&count_sql) + .bind(&collection) + .fetch_one(&state.db) + .await + { + Ok((n,)) => n, + Err(e) => return NativeOutcome::Failed(format!("failed to count records: {e}")), + }; + + let delete_sql = adapt_sql( + "DELETE FROM happyview_records WHERE uri IN \ + (SELECT uri FROM happyview_records WHERE collection = ? LIMIT ?)", + backend, + ); + + let mut deleted: i64 = 0; + let mut stopped: Option<&'static str> = None; + + loop { + if let Some(reason) = db::should_stop(state, &job.id).await { + tracing::info!(job_id = %job.id, reason, deleted, "delete-collection stopping"); + stopped = Some(reason); + break; + } + + let affected = match crate::db::query(&delete_sql) + .bind(&collection) + .bind(BATCH_SIZE) + .execute(&state.db) + .await + { + Ok(r) => r.rows_affected(), + Err(e) => return NativeOutcome::Failed(format!("failed to delete batch: {e}")), + }; + deleted += affected as i64; + + let _ = db::update_progress( + state, + &job.id, + &serde_json::json!({ "deleted": deleted, "total": total }), + ) + .await; + + // A batch smaller than BATCH_SIZE means the collection, as it stood + // when this statement ran, has been drained. Waiting for a batch of + // exactly zero instead would never happen on a collection still + // receiving inserts (e.g. from Jetstream) — the loop would spin + // indefinitely, holding the write lock and starving every other + // queued job behind this one. Records that arrive after a job was + // asked to delete a collection are out of scope for that job. + if (affected as i64) < BATCH_SIZE { + break; + } + } + + if let Some(reason) = stopped { + // finalize()'s pause/cancel arms only set status — they don't persist + // a result, so this is the only place the partial count and the + // reason it stopped are recorded. + let _ = db::update_progress( + state, + &job.id, + &serde_json::json!({ "deleted": deleted, "total": total, "stopped": reason }), + ) + .await; + } + + // Incremental auto-vacuum makes reclamation possible but never automatic; + // this is what actually returns the freed pages to the filesystem. It is a + // no-op on instances that have not yet run the one-time vacuum. Runs + // whether the loop drained the collection or was interrupted — a cancel is + // the operator's escape hatch from a slow delete, and it should reclaim + // whatever was freed up to that point rather than reclaiming nothing. + if backend == DatabaseBackend::Sqlite + && let Err(e) = crate::db::query("PRAGMA incremental_vacuum") + .execute(&state.db) + .await + { + tracing::warn!(error = %e, "incremental_vacuum after delete failed"); + } + + match stopped { + Some(reason) => NativeOutcome::Completed(serde_json::json!({ + "deleted": deleted, + "stopped": reason, + })), + None => NativeOutcome::Completed(serde_json::json!({ "deleted": deleted })), + } +} diff --git a/src/jobs/native/mod.rs b/src/jobs/native/mod.rs new file mode 100644 index 0000000..9cb7854 --- /dev/null +++ b/src/jobs/native/mod.rs @@ -0,0 +1,73 @@ +//! Built-in job types implemented in Rust rather than Lua. +//! +//! The `happyview.` prefix is **reserved**, and that reservation is an +//! authorization boundary rather than a naming convention: a native job runs +//! privileged Rust with no script mediating its input, so if a Lua script could +//! enqueue `happyview.delete-collection` it would bypass the +//! `records:delete-collection` permission check that the admin endpoint +//! performs. Only internal Rust callers may enqueue these. + +pub mod delete_collection; + +use crate::AppState; + +use super::Job; + +/// Job types beginning with this prefix may only be enqueued by internal Rust +/// callers, and may not be used as script trigger ids. +pub const RESERVED_PREFIX: &str = "happyview."; + +/// Every implemented native job type. +const NATIVE_TYPES: &[&str] = &["happyview.delete-collection"]; + +/// Whether a job type is reserved. Reserved-but-unimplemented types are still +/// refused, so adding a handler later can never be shadowed by a user script +/// created in the meantime. +pub fn is_reserved(job_type: &str) -> bool { + job_type.starts_with(RESERVED_PREFIX) +} + +/// Whether a job type has a native handler. +pub fn is_native(job_type: &str) -> bool { + NATIVE_TYPES.contains(&job_type) +} + +/// What a native handler produced. +pub enum NativeOutcome { + Completed(serde_json::Value), + Failed(String), +} + +/// Dispatch to the handler for this job type. +pub async fn execute(state: &AppState, job: &Job) -> NativeOutcome { + match job.job_type.as_str() { + "happyview.delete-collection" => delete_collection::run(state, job).await, + other => NativeOutcome::Failed(format!("no native handler for job type '{other}'")), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn happyview_prefix_is_reserved() { + assert!(is_reserved("happyview.delete-collection")); + assert!(is_reserved("happyview.anything")); + } + + #[test] + fn user_job_types_are_not_reserved() { + assert!(!is_reserved("pet.trezy.reset-account")); + assert!(!is_reserved("happyviewish.thing")); + assert!(!is_reserved("my.happyview.thing")); + } + + #[test] + fn only_known_reserved_types_are_native() { + assert!(is_native("happyview.delete-collection")); + // Reserved but unimplemented: still refused to callers, but not dispatched. + assert!(!is_native("happyview.not-a-real-job")); + assert!(!is_native("pet.trezy.reset-account")); + } +} diff --git a/src/jobs/worker.rs b/src/jobs/worker.rs index 93277b5..68859af 100644 --- a/src/jobs/worker.rs +++ b/src/jobs/worker.rs @@ -77,6 +77,15 @@ async fn execute_job(state: &AppState, job: &super::Job) { ) .await; + if super::native::is_native(&job.job_type) { + let outcome = match super::native::execute(state, job).await { + super::native::NativeOutcome::Completed(v) => JobOutcome::Completed(v), + super::native::NativeOutcome::Failed(e) => JobOutcome::Failed(e), + }; + finalize(state, job, outcome).await; + return; + } + let trigger_id = format!("job.run:{}", job.job_type); let script = match scripts::resolve(state, &trigger_id).await { Some(s) => s, @@ -308,122 +317,99 @@ async fn execute_job(state: &AppState, job: &super::Job) { } }; - match handle.call_async::(()).await { + let outcome = match handle.call_async::(()).await { Ok(result) => { - let json_result: serde_json::Value = - lua.from_value(result).unwrap_or(serde_json::json!(null)); + JobOutcome::Completed(lua.from_value(result).unwrap_or(serde_json::json!(null))) + } + Err(e) => JobOutcome::Failed(format!("{e}")), + }; + finalize(state, job, outcome).await; +} - match db::should_stop(state, &job.id).await { - Some("pausing") => { - let _ = db::set_status(state, &job.id, "paused").await; - tracing::info!(job_id = %job.id, "job paused"); - log_event( - &state.db, - EventLog { - event_type: "job.paused".to_string(), - severity: Severity::Info, - actor_did: Some(job.created_by.clone()), - subject: Some(job.job_type.clone()), - detail: serde_json::json!({ "job_id": job.id }), - }, - backend, - ) - .await; - } - Some("cancelling") => { - let _ = db::set_status(state, &job.id, "cancelled").await; - tracing::info!(job_id = %job.id, "job cancelled"); - log_event( - &state.db, - EventLog { - event_type: "job.cancelled".to_string(), - severity: Severity::Info, - actor_did: Some(job.created_by.clone()), - subject: Some(job.job_type.clone()), - detail: serde_json::json!({ "job_id": job.id }), - }, - backend, - ) - .await; - } - _ => { - let _ = db::set_result(state, &job.id, &json_result).await; - tracing::info!(job_id = %job.id, "job completed"); - log_event( - &state.db, - EventLog { - event_type: "job.completed".to_string(), - severity: Severity::Info, - actor_did: Some(job.created_by.clone()), - subject: Some(job.job_type.clone()), - detail: serde_json::json!({ - "job_id": job.id, - "result": json_result, - }), - }, - backend, - ) - .await; - } - } +/// What a job's body produced, before pause/cancel is taken into account. +pub(crate) enum JobOutcome { + Completed(serde_json::Value), + Failed(String), +} + +/// Apply pause/cancel semantics and record the terminal state. +/// +/// Shared by the Lua and native paths so the two cannot drift apart on what +/// `pausing` and `cancelling` mean. +async fn finalize(state: &AppState, job: &super::Job, outcome: JobOutcome) { + let backend = state.db_backend; + let stop = db::should_stop(state, &job.id).await; + + let (event_type, severity, detail) = match (stop, &outcome) { + (Some("pausing"), JobOutcome::Failed(error)) => { + let _ = db::set_status(state, &job.id, "paused").await; + tracing::info!(job_id = %job.id, %error, "job paused (error during stop)"); + ( + "job.paused", + Severity::Info, + serde_json::json!({ "job_id": job.id }), + ) } - Err(e) => { - let error = format!("{e}"); - match db::should_stop(state, &job.id).await { - Some("pausing") => { - let _ = db::set_status(state, &job.id, "paused").await; - tracing::info!(job_id = %job.id, "job paused (script error during stop: {error})"); - log_event( - &state.db, - EventLog { - event_type: "job.paused".to_string(), - severity: Severity::Info, - actor_did: Some(job.created_by.clone()), - subject: Some(job.job_type.clone()), - detail: serde_json::json!({ "job_id": job.id }), - }, - backend, - ) - .await; - } - Some("cancelling") => { - let _ = db::set_status(state, &job.id, "cancelled").await; - tracing::info!(job_id = %job.id, "job cancelled (script error during stop: {error})"); - log_event( - &state.db, - EventLog { - event_type: "job.cancelled".to_string(), - severity: Severity::Info, - actor_did: Some(job.created_by.clone()), - subject: Some(job.job_type.clone()), - detail: serde_json::json!({ "job_id": job.id }), - }, - backend, - ) - .await; - } - _ => { - tracing::error!(job_id = %job.id, %error, "job script failed"); - let _ = db::set_error(state, &job.id, &error).await; - log_event( - &state.db, - EventLog { - event_type: "job.failed".to_string(), - severity: Severity::Error, - actor_did: Some(job.created_by.clone()), - subject: Some(job.job_type.clone()), - detail: serde_json::json!({ - "job_id": job.id, - "error": error, - }), - }, - backend, - ) - .await; - } - } + (Some("pausing"), JobOutcome::Completed(_)) => { + let _ = db::set_status(state, &job.id, "paused").await; + tracing::info!(job_id = %job.id, "job paused"); + ( + "job.paused", + Severity::Info, + serde_json::json!({ "job_id": job.id }), + ) } - } + (Some("cancelling"), JobOutcome::Failed(error)) => { + let _ = db::set_status(state, &job.id, "cancelled").await; + tracing::info!(job_id = %job.id, %error, "job cancelled (error during stop)"); + ( + "job.cancelled", + Severity::Info, + serde_json::json!({ "job_id": job.id }), + ) + } + (Some("cancelling"), JobOutcome::Completed(_)) => { + let _ = db::set_status(state, &job.id, "cancelled").await; + tracing::info!(job_id = %job.id, "job cancelled"); + ( + "job.cancelled", + Severity::Info, + serde_json::json!({ "job_id": job.id }), + ) + } + (_, JobOutcome::Completed(result)) => { + let _ = db::set_result(state, &job.id, result).await; + tracing::info!(job_id = %job.id, "job completed"); + ( + "job.completed", + Severity::Info, + serde_json::json!({ "job_id": job.id, "result": result }), + ) + } + (_, JobOutcome::Failed(error)) => { + // Not necessarily a script: native jobs fail here too. + tracing::error!(job_id = %job.id, %error, "job failed"); + let _ = db::set_error(state, &job.id, error).await; + ( + "job.failed", + Severity::Error, + serde_json::json!({ "job_id": job.id, "error": error }), + ) + } + }; + + log_event( + &state.db, + EventLog { + event_type: event_type.to_string(), + severity, + actor_did: Some(job.created_by.clone()), + subject: Some(job.job_type.clone()), + detail, + }, + backend, + ) + .await; } async fn load_env_vars( diff --git a/src/lib.rs b/src/lib.rs index fce8eb4..f3b410b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ pub mod lexicon; pub mod linked_repos; pub mod lua; pub mod lua_analysis; +pub mod maintenance; pub mod oauth; pub mod plc; pub mod plugin; diff --git a/src/lua/atproto_api.rs b/src/lua/atproto_api.rs index edf478d..6bdc58f 100644 --- a/src/lua/atproto_api.rs +++ b/src/lua/atproto_api.rs @@ -682,6 +682,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), diff --git a/src/lua/db_api.rs b/src/lua/db_api.rs index afe0878..7059d9b 100644 --- a/src/lua/db_api.rs +++ b/src/lua/db_api.rs @@ -786,6 +786,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), diff --git a/src/lua/http_api.rs b/src/lua/http_api.rs index 1e7a3c0..ff2f158 100644 --- a/src/lua/http_api.rs +++ b/src/lua/http_api.rs @@ -92,6 +92,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), diff --git a/src/lua/jobs_api.rs b/src/lua/jobs_api.rs index 070a1d5..7ead04d 100644 --- a/src/lua/jobs_api.rs +++ b/src/lua/jobs_api.rs @@ -53,6 +53,13 @@ pub fn register_jobs_api( )); } + if crate::jobs::native::is_reserved(&job_type) { + return Err(mlua::Error::runtime(format!( + "job_type prefix '{}' is reserved for built-in jobs", + crate::jobs::native::RESERVED_PREFIX + ))); + } + let caller = caller.as_ref().ok_or_else(|| { mlua::Error::runtime("jobs.create requires an authenticated caller") })?; @@ -221,6 +228,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), @@ -481,6 +489,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), diff --git a/src/lua/scripts.rs b/src/lua/scripts.rs index ea2b2e3..0edebeb 100644 --- a/src/lua/scripts.rs +++ b/src/lua/scripts.rs @@ -151,6 +151,12 @@ fn validate_job_type(job_type: &str) -> Result<(), String> { "invalid job type '{job_type}': must match /^[a-z0-9][a-z0-9._-]*$/" )); } + if crate::jobs::native::is_reserved(job_type) { + return Err(format!( + "invalid job type '{job_type}': the '{}' prefix is reserved for built-in jobs", + crate::jobs::native::RESERVED_PREFIX + )); + } Ok(()) } @@ -902,6 +908,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), diff --git a/src/lua/spaces_api.rs b/src/lua/spaces_api.rs index cf4cd67..10c9c65 100644 --- a/src/lua/spaces_api.rs +++ b/src/lua/spaces_api.rs @@ -554,6 +554,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), @@ -755,6 +756,7 @@ mod tests { port: 0, database_url: String::new(), database_backend: backend, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), diff --git a/src/lua/xrpc_api.rs b/src/lua/xrpc_api.rs index 9e84305..fc34b51 100644 --- a/src/lua/xrpc_api.rs +++ b/src/lua/xrpc_api.rs @@ -197,6 +197,7 @@ mod tests { port: 3000, database_url: String::new(), database_backend: DatabaseBackend::Sqlite, + sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), diff --git a/src/main.rs b/src/main.rs index 8beac92..dd411e3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -74,6 +74,29 @@ async fn main() { ), } + match happyview::maintenance::vacuum::run_if_requested( + &db_pool, + db_backend, + &config.database_url, + ) + .await + { + Ok(Some(result)) if result.status == "ok" => { + info!( + reclaimed = %happyview::maintenance::vacuum::human_bytes(result.reclaimed_bytes), + "scheduled vacuum complete" + ); + } + Ok(Some(result)) => { + tracing::error!( + error = %result.error.clone().unwrap_or_default(), + "scheduled vacuum failed" + ); + } + Ok(None) => { /* not scheduled */ } + Err(e) => tracing::error!(error = %e, "scheduled vacuum could not be evaluated"), + } + // Backfill record_refs in the background (first run after upgrade) { let db_bg = db_pool.clone(); diff --git a/src/maintenance/disk.rs b/src/maintenance/disk.rs new file mode 100644 index 0000000..a4e8570 --- /dev/null +++ b/src/maintenance/disk.rs @@ -0,0 +1,345 @@ +use std::path::{Path, PathBuf}; + +/// Sizes and free space relevant to a VACUUM. +#[derive(Debug, Clone, serde::Serialize)] +pub struct DiskReport { + pub db_bytes: u64, + pub wal_bytes: u64, + /// `None` means the free space on the database's filesystem could not be + /// measured (not that it measured zero). + pub db_fs_free: Option, + /// `None` means the free space on the temp filesystem could not be + /// measured (not that it measured zero). + pub temp_fs_free: Option, + /// Whether the database and temp directories resolve to the same + /// filesystem (compared by device id), which changes how much headroom a + /// VACUUM needs — see `feasibility`. + pub same_filesystem: bool, + pub db_path: String, + pub temp_path: String, +} + +/// Whether a VACUUM has room to run. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +#[serde(tag = "status", rename_all = "snake_case")] +pub enum VacuumFeasibility { + Ok, + Insufficient { + needed: u64, + available: u64, + path: String, + }, + /// Free space on `path` could not be measured at all. Treated as blocking + /// — a vacuum is not run on a number we don't have — but reported + /// distinctly from `Insufficient` so an operator isn't sent chasing disk + /// space that was never actually the problem. + Unknown { + path: String, + }, +} + +/// VACUUM rebuilds the whole database, so it needs the database's size again +/// plus a margin. `db_bytes` must already include the WAL — a stranded WAL is +/// exactly the case this feature targets, and the temp copy plus the rebuilt +/// WAL both have to fit alongside it, not just alongside the main file. Both +/// the database's filesystem and the temp directory's are checked, because +/// SQLite writes the rebuild's temporary file to the temp directory, which is +/// frequently a different filesystem inside a container. +/// +/// The margin depends on whether the two directories share a filesystem: +/// - Different filesystems: each needs the database's size again plus 20%, +/// checked independently. +/// - Same filesystem: that one filesystem must hold the temp copy *and* +/// absorb the whole rebuild through the WAL, so the margin doubles to 120% +/// (2.2x the database size in total). +pub fn feasibility( + db_bytes: u64, + db_fs_free: Option, + temp_fs_free: Option, + db_path: &str, + temp_path: &str, + same_filesystem: bool, +) -> VacuumFeasibility { + let needed = if same_filesystem { + db_bytes + (db_bytes * 12) / 10 + } else { + db_bytes + db_bytes / 5 + }; + if needed == 0 { + return VacuumFeasibility::Ok; + } + + match (db_fs_free, temp_fs_free) { + (None, _) => VacuumFeasibility::Unknown { + path: db_path.to_string(), + }, + (_, None) => VacuumFeasibility::Unknown { + path: temp_path.to_string(), + }, + (Some(db_free), Some(temp_free)) => { + // Report whichever filesystem has least room, so the message + // names the one that will actually fail. + let (available, path) = if db_free <= temp_free { + (db_free, db_path) + } else { + (temp_free, temp_path) + }; + if available >= needed { + VacuumFeasibility::Ok + } else { + VacuumFeasibility::Insufficient { + needed, + available, + path: path.to_string(), + } + } + } + } +} + +fn file_len(path: &Path) -> u64 { + std::fs::metadata(path).map(|m| m.len()).unwrap_or(0) +} + +/// Free space at `path`, or `None` if it could not be measured (e.g. the +/// directory doesn't exist or isn't readable) — distinct from a real zero. +fn free_space(path: &Path) -> Option { + fs4::available_space(path).ok() +} + +/// The directory containing `path`, falling back to `.` when `path` has no +/// parent component. `Path::parent()` returns `Some("")` for a bare filename +/// like `happyview.db`, not `None`, so the empty case must be checked +/// explicitly — the same guard `db::connect` uses when creating the data +/// directory. +fn containing_dir(path: &Path) -> &Path { + match path.parent() { + Some(p) if !p.as_os_str().is_empty() => p, + _ => Path::new("."), + } +} + +/// The WAL file's path for a given database path. SQLite names it by +/// appending `-wal` to the *entire* main file name, not by replacing its +/// extension — `/data/happyview.db` becomes `/data/happyview.db-wal`, and the +/// extensionless `/data/hv` becomes `/data/hv-wal`. +fn wal_path_for(db_path: &Path) -> PathBuf { + let mut name = db_path.as_os_str().to_os_string(); + name.push("-wal"); + PathBuf::from(name) +} + +/// Whether `a` and `b` live on the same filesystem, compared by device id. +/// Conservatively `false` if either path's metadata can't be read. +fn same_filesystem(a: &Path, b: &Path) -> bool { + use std::os::unix::fs::MetadataExt; + match (std::fs::metadata(a), std::fs::metadata(b)) { + (Ok(ma), Ok(mb)) => ma.dev() == mb.dev(), + _ => false, + } +} + +/// The directory SQLite will use for VACUUM's temporary database. +fn temp_dir() -> PathBuf { + std::env::var_os("SQLITE_TMPDIR") + .or_else(|| std::env::var_os("TMPDIR")) + .map(PathBuf::from) + .unwrap_or_else(std::env::temp_dir) +} + +/// Measure database and WAL sizes plus free space. Returns `None` when the URL +/// is not a file-backed SQLite database. +pub fn report(db_url: &str) -> Option { + let db_path = crate::db::sqlite_path_from_url(db_url)?; + let wal_path = wal_path_for(&db_path); + let dir = containing_dir(&db_path); + let tmp = temp_dir(); + + Some(DiskReport { + db_bytes: file_len(&db_path), + wal_bytes: file_len(&wal_path), + db_fs_free: free_space(dir), + temp_fs_free: free_space(&tmp), + same_filesystem: same_filesystem(dir, &tmp), + db_path: db_path.to_string_lossy().into_owned(), + temp_path: tmp.to_string_lossy().into_owned(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + const GIB: u64 = 1024 * 1024 * 1024; + + #[test] + fn feasible_when_both_filesystems_have_headroom() { + let v = feasibility( + 8 * GIB, + Some(20 * GIB), + Some(20 * GIB), + "/data", + "/tmp", + false, + ); + assert!(matches!(v, VacuumFeasibility::Ok)); + } + + #[test] + fn requires_twenty_percent_margin_over_database_size() { + // 8 GiB db needs 9.6 GiB; 9 GiB is not enough. + let v = feasibility( + 8 * GIB, + Some(9 * GIB), + Some(50 * GIB), + "/data", + "/tmp", + false, + ); + match v { + VacuumFeasibility::Insufficient { + needed, + available, + path, + } => { + assert_eq!(needed, 8 * GIB + (8 * GIB) / 5); + assert_eq!(available, 9 * GIB); + assert_eq!(path, "/data"); + } + other => panic!("expected Insufficient, got {other:?}"), + } + } + + #[test] + fn reports_the_temp_filesystem_when_it_is_the_binding_constraint() { + let v = feasibility(8 * GIB, Some(50 * GIB), Some(GIB), "/data", "/tmp", false); + match v { + VacuumFeasibility::Insufficient { + available, path, .. + } => { + assert_eq!(available, GIB); + assert_eq!(path, "/tmp"); + } + other => panic!("expected Insufficient, got {other:?}"), + } + } + + #[test] + fn an_empty_database_is_always_feasible() { + assert!(matches!( + feasibility(0, Some(0), Some(0), "/data", "/tmp", false), + VacuumFeasibility::Ok + )); + } + + #[test] + fn unknown_when_a_filesystem_cannot_be_measured() { + let v = feasibility(8 * GIB, None, Some(50 * GIB), "/data", "/tmp", false); + assert!(matches!(v, VacuumFeasibility::Unknown { ref path } if path == "/data")); + + let v = feasibility(8 * GIB, Some(50 * GIB), None, "/data", "/tmp", false); + assert!(matches!(v, VacuumFeasibility::Unknown { ref path } if path == "/tmp")); + } + + #[test] + fn same_filesystem_requires_the_higher_margin() { + // 8 GiB db on one volume needs 8 + 8*1.2 = 17.6 GiB. + let v = feasibility( + 8 * GIB, + Some(18 * GIB), + Some(18 * GIB), + "/data", + "/data", + true, + ); + assert!(matches!(v, VacuumFeasibility::Ok)); + + let v = feasibility( + 8 * GIB, + Some(17 * GIB), + Some(17 * GIB), + "/data", + "/data", + true, + ); + assert!(matches!(v, VacuumFeasibility::Insufficient { .. })); + } + + #[test] + fn reviewer_scenario_ten_gib_database_fifteen_gib_free_same_volume_is_refused() { + let v = feasibility( + 10 * GIB, + Some(15 * GIB), + Some(15 * GIB), + "/data", + "/data", + true, + ); + match v { + VacuumFeasibility::Insufficient { + needed, available, .. + } => { + assert_eq!(needed, 22 * GIB); + assert_eq!(available, 15 * GIB); + } + other => panic!("expected Insufficient, got {other:?}"), + } + } + + #[test] + fn reviewer_scenario_wal_pushes_needed_size_past_available_free_space() { + // 10 GiB database + 8 GiB WAL on one volume with 24 GiB free. Judged + // solely on `db_bytes` this needs 10 + 12 = 22 GiB and is wrongly + // green-lit; the WAL has to be folded into the size the requirement + // is computed from, making it 18 GiB -> 18 + 21.6 = 39.6 GiB needed, + // which 24 GiB does not cover. + let db_bytes = 10 * GIB; + let wal_bytes = 8 * GIB; + let v = feasibility( + db_bytes + wal_bytes, + Some(24 * GIB), + Some(24 * GIB), + "/data", + "/data", + true, + ); + match v { + VacuumFeasibility::Insufficient { + needed, available, .. + } => { + assert_eq!(needed, 18 * GIB + (18 * GIB * 12) / 10); + assert_eq!(available, 24 * GIB); + } + other => panic!("expected Insufficient, got {other:?}"), + } + } + + #[test] + fn containing_dir_falls_back_to_dot_for_a_bare_filename() { + assert_eq!(containing_dir(Path::new("happyview.db")), Path::new(".")); + assert_eq!( + containing_dir(Path::new("/data/happyview.db")), + Path::new("/data") + ); + } + + #[test] + fn wal_path_appends_suffix_without_touching_the_extension() { + assert_eq!( + wal_path_for(Path::new("/data/happyview.db")), + PathBuf::from("/data/happyview.db-wal") + ); + assert_eq!( + wal_path_for(Path::new("/data/hv")), + PathBuf::from("/data/hv-wal") + ); + } + + #[test] + fn free_space_returns_none_when_the_path_does_not_exist() { + assert_eq!( + free_space(Path::new("/definitely/does/not/exist/at/all")), + None + ); + } +} diff --git a/src/maintenance/mod.rs b/src/maintenance/mod.rs new file mode 100644 index 0000000..4e5f01e --- /dev/null +++ b/src/maintenance/mod.rs @@ -0,0 +1,5 @@ +//! One-off database maintenance: disk accounting and the one-time SQLite +//! vacuum that reclaims pages stranded before incremental auto-vacuum existed. + +pub mod disk; +pub mod vacuum; diff --git a/src/maintenance/vacuum.rs b/src/maintenance/vacuum.rs new file mode 100644 index 0000000..ea6c7ed --- /dev/null +++ b/src/maintenance/vacuum.rs @@ -0,0 +1,482 @@ +//! One-time SQLite VACUUM. +//! +//! `VACUUM` does not clean up in place — it rebuilds the database into a new +//! file and swaps it. That is why it needs the database's size again in free +//! space and exclusive access, and why it runs at startup (before any worker +//! spawns) rather than against a live instance. +//! +//! Existing instances need it because `PRAGMA auto_vacuum = INCREMENTAL` is a +//! no-op on a populated database until a rebuild applies it. + +use serde::{Deserialize, Serialize}; +use sqlx::AnyPool; +use tracing::info; + +use crate::db::{DatabaseBackend, adapt_sql, now_rfc3339}; +use crate::error::AppError; +use crate::maintenance::disk::{self, VacuumFeasibility}; + +const KEY_REQUESTED: &str = "sqlite_vacuum_requested_at"; +const KEY_ATTEMPT: &str = "sqlite_vacuum_attempt_started_at"; +const KEY_COMPLETED: &str = "sqlite_vacuum_completed_at"; +const KEY_RESULT: &str = "sqlite_vacuum_last_result"; + +/// The outcome of the most recent attempt, persisted so the dashboard can +/// report it long after the boot that produced it. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct VacuumResult { + /// `"ok"` or `"failed"`. + pub status: String, + pub at: String, + pub db_bytes_before: u64, + pub db_bytes_after: u64, + pub reclaimed_bytes: u64, + pub error: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct VacuumStatus { + pub requested_at: Option, + pub attempt_started_at: Option, + /// When a vacuum last completed, or when one was marked unnecessary via + /// `mark_not_needed`. Purely informational — it suppresses the + /// dashboard's "you should vacuum" prompt but does not gate + /// `run_if_requested`, which is driven by `requested_at` alone. + pub completed_at: Option, + pub last_result: Option, +} + +async fn get<'e, E>( + executor: E, + backend: DatabaseBackend, + key: &str, +) -> Result, AppError> +where + E: sqlx::Executor<'e, Database = sqlx::Any>, +{ + let sql = adapt_sql( + "SELECT value FROM happyview_instance_settings WHERE key = ?", + backend, + ); + let row: Option<(String,)> = crate::db::query_as(&sql) + .bind(key) + .fetch_optional(executor) + .await + .map_err(|e| AppError::Internal(format!("failed to read {key}: {e}")))?; + Ok(row.map(|r| r.0)) +} + +async fn put<'e, E>( + executor: E, + backend: DatabaseBackend, + key: &str, + value: &str, +) -> Result<(), AppError> +where + E: sqlx::Executor<'e, Database = sqlx::Any>, +{ + let sql = adapt_sql( + "INSERT INTO happyview_instance_settings (key, value, updated_at) VALUES (?, ?, ?) + ON CONFLICT (key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at", + backend, + ); + crate::db::query(&sql) + .bind(key) + .bind(value) + .bind(now_rfc3339()) + .execute(executor) + .await + .map_err(|e| AppError::Internal(format!("failed to write {key}: {e}")))?; + Ok(()) +} + +async fn clear<'e, E>(executor: E, backend: DatabaseBackend, key: &str) -> Result<(), AppError> +where + E: sqlx::Executor<'e, Database = sqlx::Any>, +{ + let sql = adapt_sql( + "DELETE FROM happyview_instance_settings WHERE key = ?", + backend, + ); + crate::db::query(&sql) + .bind(key) + .execute(executor) + .await + .map_err(|e| AppError::Internal(format!("failed to clear {key}: {e}")))?; + Ok(()) +} + +pub async fn read_status( + pool: &AnyPool, + backend: DatabaseBackend, +) -> Result { + let last_result = get(pool, backend, KEY_RESULT) + .await? + .and_then(|raw| serde_json::from_str::(&raw).ok()); + Ok(VacuumStatus { + requested_at: get(pool, backend, KEY_REQUESTED).await?, + attempt_started_at: get(pool, backend, KEY_ATTEMPT).await?, + completed_at: get(pool, backend, KEY_COMPLETED).await?, + last_result, + }) +} + +/// Arm the vacuum. It runs on the next startup. +pub async fn request(pool: &AnyPool, backend: DatabaseBackend) -> Result<(), AppError> { + put(pool, backend, KEY_REQUESTED, &now_rfc3339()).await +} + +/// Disarm a scheduled vacuum. +pub async fn cancel_request(pool: &AnyPool, backend: DatabaseBackend) -> Result<(), AppError> { + clear(pool, backend, KEY_REQUESTED).await +} + +/// Record that no vacuum is needed — used at setup, since a database created by +/// this version already has incremental auto-vacuum and nothing stranded. +/// This only suppresses the dashboard's prompt; it does not prevent a later +/// `request()` from arming (and running) a vacuum. +pub async fn mark_not_needed(pool: &AnyPool, backend: DatabaseBackend) -> Result<(), AppError> { + put(pool, backend, KEY_COMPLETED, &now_rfc3339()).await +} + +/// Persist the outcome of an attempt and disarm, all in one transaction — a +/// crash between clearing `requested_at` and writing `completed_at` after a +/// *successful* vacuum would otherwise leave the instance unarmed with no +/// record of success, nagging forever about a vacuum that already worked. +async fn record_result( + pool: &AnyPool, + backend: DatabaseBackend, + result: &VacuumResult, +) -> Result<(), AppError> { + let json = serde_json::to_string(result) + .map_err(|e| AppError::Internal(format!("failed to serialize vacuum result: {e}")))?; + + let mut tx = pool + .begin() + .await + .map_err(|e| AppError::Internal(format!("failed to begin transaction: {e}")))?; + + put(&mut *tx, backend, KEY_RESULT, &json).await?; + clear(&mut *tx, backend, KEY_REQUESTED).await?; + clear(&mut *tx, backend, KEY_ATTEMPT).await?; + if result.status == "ok" { + put(&mut *tx, backend, KEY_COMPLETED, &result.at).await?; + } + + tx.commit() + .await + .map_err(|e| AppError::Internal(format!("failed to commit vacuum result: {e}")))?; + Ok(()) +} + +fn failure(at: String, before: u64, error: String) -> VacuumResult { + VacuumResult { + status: "failed".into(), + at, + db_bytes_before: before, + db_bytes_after: before, + reclaimed_bytes: 0, + error: Some(error), + } +} + +/// Run the vacuum if one was requested. +/// +/// Gated on `requested_at` alone — see `VacuumStatus::completed_at`. Returns +/// `Ok(None)` when nothing is armed. Any attempt — successful or not — +/// disarms (clears `requested_at`), so a vacuum that cannot succeed does not +/// retry on every boot. +pub async fn run_if_requested( + pool: &AnyPool, + backend: DatabaseBackend, + db_url: &str, +) -> Result, AppError> { + if backend != DatabaseBackend::Sqlite { + return Ok(None); + } + let status = read_status(pool, backend).await?; + if status.requested_at.is_none() { + return Ok(None); + } + run_armed(pool, backend, &status, disk::report(db_url), db_url).await +} + +/// The decision tree for an armed vacuum, split out from `run_if_requested` +/// so it can be exercised directly against a synthetic `DiskReport` — real +/// free-disk numbers are normally far too large to hit the `Insufficient` or +/// `Unknown` branches in a test. +async fn run_armed( + pool: &AnyPool, + backend: DatabaseBackend, + status: &VacuumStatus, + report: Option, + db_url: &str, +) -> Result, AppError> { + let now = now_rfc3339(); + let before = report.as_ref().map(|r| r.db_bytes).unwrap_or(0); + + // An attempt marker left over from a previous boot means the process died + // mid-rebuild and never reached the code that records an error. + if status.attempt_started_at.is_some() { + let result = failure( + now, + before, + "a previous vacuum attempt did not complete; the process may have run \ + out of memory or disk, or it may have been killed by a container \ + healthcheck or orchestrator timeout while the rebuild was still in \ + progress (VACUUM can take minutes to hours on a large database). \ + Re-schedule it once the cause is addressed." + .into(), + ); + record_result(pool, backend, &result).await?; + return Ok(Some(result)); + } + + if let Some(r) = &report { + match disk::feasibility( + r.db_bytes + r.wal_bytes, + r.db_fs_free, + r.temp_fs_free, + &r.db_path, + &r.temp_path, + r.same_filesystem, + ) { + VacuumFeasibility::Ok => {} + VacuumFeasibility::Insufficient { + needed, + available, + path, + } => { + let result = failure( + now, + before, + format!( + "not enough free space: needs ~{} on {path}, {} available", + human_bytes(needed), + human_bytes(available) + ), + ); + record_result(pool, backend, &result).await?; + return Ok(Some(result)); + } + VacuumFeasibility::Unknown { path } => { + let result = failure( + now, + before, + format!( + "could not measure free space on {path}; skipping the vacuum until this is resolved" + ), + ); + record_result(pool, backend, &result).await?; + return Ok(Some(result)); + } + } + } + + put(pool, backend, KEY_ATTEMPT, &now).await?; + + // Log before the rebuild starts, not after: `VACUUM` on a large database + // blocks the server from binding its port for minutes to hours, and this + // is the only place that says so — an operator staring at a container + // that hasn't come up needs this line in the logs to know why, and how + // big a rebuild they're waiting on. + info!( + db_bytes = before, + db_size = %human_bytes(before), + "starting scheduled database vacuum; the server will not accept \ + connections until it completes" + ); + + let result = match run(pool, backend).await { + Ok(()) => { + let after = disk::report(db_url).map(|r| r.db_bytes).unwrap_or(before); + VacuumResult { + status: "ok".into(), + at: now_rfc3339(), + db_bytes_before: before, + db_bytes_after: after, + reclaimed_bytes: before.saturating_sub(after), + error: None, + } + } + Err(e) => failure(now_rfc3339(), before, e.to_string()), + }; + record_result(pool, backend, &result).await?; + Ok(Some(result)) +} + +/// Perform the rebuild. All three statements must share one connection: the +/// `auto_vacuum` change only takes effect via the VACUUM that follows it. +/// A no-op on Postgres — `VACUUM`/`auto_vacuum` here are SQLite pragmas. +pub async fn run(pool: &AnyPool, backend: DatabaseBackend) -> Result<(), AppError> { + if backend != DatabaseBackend::Sqlite { + return Ok(()); + } + + let mut conn = pool + .acquire() + .await + .map_err(|e| AppError::Internal(format!("failed to acquire connection: {e}")))?; + + crate::db::query("PRAGMA auto_vacuum = INCREMENTAL") + .execute(&mut *conn) + .await + .map_err(|e| AppError::Internal(format!("failed to set auto_vacuum: {e}")))?; + crate::db::query("VACUUM") + .execute(&mut *conn) + .await + .map_err(|e| AppError::Internal(format!("VACUUM failed: {e}")))?; + crate::db::query("PRAGMA wal_checkpoint(TRUNCATE)") + .execute(&mut *conn) + .await + .map_err(|e| AppError::Internal(format!("WAL checkpoint failed: {e}")))?; + Ok(()) +} + +/// Render a byte count for an operator-facing message. +pub fn human_bytes(bytes: u64) -> String { + const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"]; + let mut value = bytes as f64; + let mut unit = 0; + // The cutoff is slightly below 1024 so a value that would display as + // "1024.0" after rounding to one decimal place bumps to the next unit + // instead (e.g. 1_048_575 B is 1 byte short of 1 MiB but rounds to + // "1024.0 KiB" without this). + while value >= 1023.95 && unit < UNITS.len() - 1 { + value /= 1024.0; + unit += 1; + } + if unit == 0 { + format!("{bytes} B") + } else { + format!("{value:.1} {}", UNITS[unit]) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn human_bytes_scales_units() { + assert_eq!(human_bytes(512), "512 B"); + assert_eq!(human_bytes(1024), "1.0 KiB"); + assert_eq!(human_bytes(8 * 1024 * 1024 * 1024), "8.0 GiB"); + } + + #[test] + fn human_bytes_rounds_up_to_the_next_unit_at_the_boundary() { + // 1_048_575 B is 1 byte short of 1 MiB; formatting the KiB value to + // one decimal place rounds it to "1024.0", so the unit must bump too. + assert_eq!(human_bytes(1_048_575), "1.0 MiB"); + } + + async fn temp_sqlite_pool() -> (AnyPool, std::path::PathBuf) { + let path = std::env::temp_dir().join(format!("hv-vacuum-unit-{}.db", uuid::Uuid::new_v4())); + let url = format!("sqlite://{}?mode=rwc", path.display()); + let pool = crate::db::connect(&url, DatabaseBackend::Sqlite).await; + (pool, path) + } + + fn cleanup_sqlite_files(path: &std::path::Path) { + let _ = std::fs::remove_file(path); + let _ = std::fs::remove_file(format!("{}-wal", path.display())); + let _ = std::fs::remove_file(format!("{}-shm", path.display())); + } + + /// Real free-disk numbers are normally far too large to hit the + /// `Insufficient` branch, so this drives `run_armed` directly against a + /// synthetic `DiskReport` to prove the refusal actually disarms — a bug + /// here means production retries a doomed vacuum on every boot. + #[tokio::test] + async fn insufficient_disk_records_failure_and_disarms() { + let (pool, path) = temp_sqlite_pool().await; + + request(&pool, DatabaseBackend::Sqlite).await.unwrap(); + let status = read_status(&pool, DatabaseBackend::Sqlite).await.unwrap(); + + let report = disk::DiskReport { + db_bytes: 10 * 1024 * 1024 * 1024, + wal_bytes: 0, + db_fs_free: Some(1024), + temp_fs_free: Some(1024), + same_filesystem: false, + db_path: "/fake/db".into(), + temp_path: "/fake/tmp".into(), + }; + + let out = run_armed( + &pool, + DatabaseBackend::Sqlite, + &status, + Some(report), + "sqlite://unused", + ) + .await + .expect("run_armed failed") + .expect("expected a result"); + + assert_eq!(out.status, "failed"); + assert!( + out.error + .unwrap_or_default() + .contains("not enough free space"), + "error should name the disk shortfall" + ); + + let status_after = read_status(&pool, DatabaseBackend::Sqlite).await.unwrap(); + assert!( + status_after.requested_at.is_none(), + "must disarm on insufficient disk, not retry forever" + ); + + drop(pool); + cleanup_sqlite_files(&path); + } + + #[tokio::test] + async fn unmeasurable_disk_records_failure_and_disarms() { + let (pool, path) = temp_sqlite_pool().await; + + request(&pool, DatabaseBackend::Sqlite).await.unwrap(); + let status = read_status(&pool, DatabaseBackend::Sqlite).await.unwrap(); + + let report = disk::DiskReport { + db_bytes: 10 * 1024 * 1024 * 1024, + wal_bytes: 0, + db_fs_free: None, + temp_fs_free: Some(50 * 1024 * 1024 * 1024), + same_filesystem: false, + db_path: "/fake/db".into(), + temp_path: "/fake/tmp".into(), + }; + + let out = run_armed( + &pool, + DatabaseBackend::Sqlite, + &status, + Some(report), + "sqlite://unused", + ) + .await + .expect("run_armed failed") + .expect("expected a result"); + + assert_eq!(out.status, "failed"); + assert!( + out.error + .unwrap_or_default() + .contains("could not measure free space"), + "error should name the measurement failure, not claim the disk is full" + ); + + let status_after = read_status(&pool, DatabaseBackend::Sqlite).await.unwrap(); + assert!( + status_after.requested_at.is_none(), + "must disarm, not retry forever" + ); + + drop(pool); + cleanup_sqlite_files(&path); + } +} diff --git a/src/setup.rs b/src/setup.rs index 99578cd..8a2be7b 100644 --- a/src/setup.rs +++ b/src/setup.rs @@ -605,6 +605,12 @@ async fn complete(_auth: UserAuth, State(state): State) -> Result Value { + let bytes = resp.into_body().collect().await.unwrap().to_bytes(); + serde_json::from_slice(&bytes).unwrap() +} + +#[tokio::test] +#[serial] +async fn status_reports_backend_and_vacuum_state() { + common::require_db!(); + let app = TestApp::new().await; + + let resp = app + .router + .clone() + .oneshot( + Request::builder() + .uri("/admin/database/status") + .header(app.admin_cookie().0, app.admin_cookie().1) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let body = json_body(resp).await; + assert!(body["backend"].is_string()); + assert!(body["vacuum"].is_object()); + assert!( + body["journal_size_limit"].is_number(), + "journal_size_limit missing or not a number: {body}" + ); + + // `disk`/`feasibility` are only meaningful for a file-backed SQLite + // database (`disk::report` returns `None` for any other backend, which + // serializes to JSON `null`). Asserting the full shape here — rather than + // just `is_object()` on the whole body — is what would catch a future + // `unwrap_or(0)` on `db_fs_free`/`temp_fs_free`: that regression would + // turn a real disk's free space into a suspicious literal `0` rather than + // leaving it `null` or a genuine (non-zero, on any real test host) count. + if app.state.db_backend == DatabaseBackend::Sqlite { + assert!(body["disk"].is_object(), "disk missing: {body}"); + let disk = &body["disk"]; + assert!( + disk["db_fs_free"].is_u64() || disk["db_fs_free"].is_null(), + "db_fs_free must be a number or null, never a stand-in like 0 \ + collapsed from `unwrap_or(0)`: {body}" + ); + assert!( + disk["temp_fs_free"].is_u64() || disk["temp_fs_free"].is_null(), + "temp_fs_free must be a number or null: {body}" + ); + assert!( + disk["same_filesystem"].is_boolean(), + "same_filesystem missing or not a bool: {body}" + ); + assert!( + body["feasibility"]["status"].is_string(), + "feasibility.status missing: {body}" + ); + } else { + assert!( + body["disk"].is_null(), + "disk should be null on a non-SQLite backend: {body}" + ); + assert!( + body["feasibility"].is_null(), + "feasibility should be null on a non-SQLite backend: {body}" + ); + } +} + +#[tokio::test] +#[serial] +async fn scheduling_and_cancelling_a_vacuum_round_trips() { + common::require_db!(); + let app = TestApp::new().await; + + let post = |uri: &str, method: &str| { + Request::builder() + .method(method) + .uri(uri.to_string()) + .header(app.admin_cookie().0, app.admin_cookie().1) + .body(Body::empty()) + .unwrap() + }; + + let resp = app + .router + .clone() + .oneshot(post("/admin/database/vacuum/schedule", "POST")) + .await + .unwrap(); + + // VACUUM is a SQLite-only operation, so scheduling it on Postgres is + // refused outright rather than silently armed-but-inert: `status` would + // otherwise report "scheduled" forever since nothing ever runs it. + if app.state.db_backend != DatabaseBackend::Sqlite { + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + + let resp = app + .router + .clone() + .oneshot(post("/admin/database/status", "GET")) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = json_body(resp).await; + assert!( + body["vacuum"]["requested_at"].is_null(), + "should not have armed on a non-SQLite backend: {body}" + ); + return; + } + + assert_eq!(resp.status(), StatusCode::OK); + + let resp = app + .router + .clone() + .oneshot(post("/admin/database/status", "GET")) + .await + .unwrap(); + let body = json_body(resp).await; + assert!( + body["vacuum"]["requested_at"].is_string(), + "not armed: {body}" + ); + + let resp = app + .router + .clone() + .oneshot(post("/admin/database/vacuum/schedule", "DELETE")) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let resp = app + .router + .clone() + .oneshot(post("/admin/database/status", "GET")) + .await + .unwrap(); + let body = json_body(resp).await; + assert!( + body["vacuum"]["requested_at"].is_null(), + "not disarmed: {body}" + ); +} + +#[tokio::test] +#[serial] +async fn status_requires_auth() { + common::require_db!(); + let app = TestApp::new().await; + + let resp = app + .router + .clone() + .oneshot( + Request::builder() + .uri("/admin/database/status") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); +} diff --git a/tests/e2e_jobs.rs b/tests/e2e_jobs.rs index 11dc535..8913915 100644 --- a/tests/e2e_jobs.rs +++ b/tests/e2e_jobs.rs @@ -848,3 +848,345 @@ async fn list_jobs_includes_dpop_fields() { assert_eq!(jobs[0]["dpop_key_id"], "list_key"); assert_eq!(jobs[0]["inherit_auth"], true); } + +// --------------------------------------------------------------------------- +// Native job dispatch +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +async fn scripts_cannot_claim_a_reserved_job_trigger() { + common::require_db!(); + let app = TestApp::new().await; + + let resp = app + .router + .clone() + .oneshot(admin_post( + "/admin/scripts", + app.admin_cookie(), + &json!({ + "id": "job.run:happyview.delete-collection", + "body": "function handle() end", + "script_type": "lua" + }), + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + let body = json_body(resp).await; + assert!( + body.to_string().contains("reserved"), + "error should explain the reservation: {body}" + ); +} + +async fn seed_record(app: &TestApp, uri: &str, collection: &str) { + // created_at has no default on Postgres (dropped by the TEXT migration, + // see 20260318000000_uuid_to_text.sql), so it must be supplied explicitly + // to work on both backends. + let sql = adapt_sql( + "INSERT INTO happyview_records (uri, did, collection, rkey, record, cid, indexed_at, created_at) \ + VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + app.state.db_backend, + ); + let now = happyview::db::now_rfc3339(); + happyview::db::query(&sql) + .bind(uri) + .bind("did:plc:seed") + .bind(collection) + .bind("rkey1") + .bind(r#"{"text":"hi"}"#) + .bind("bafyseed") + .bind(&now) + .bind(&now) + .execute(&app.state.db) + .await + .expect("seed_record failed"); +} + +async fn count_in_collection(app: &TestApp, collection: &str) -> i64 { + let sql = adapt_sql( + "SELECT COUNT(*) FROM happyview_records WHERE collection = ?", + app.state.db_backend, + ); + let (n,): (i64,) = happyview::db::query_as(&sql) + .bind(collection) + .fetch_one(&app.state.db) + .await + .expect("count failed"); + n +} + +#[tokio::test] +#[serial] +async fn native_delete_collection_job_empties_the_collection() { + common::require_db!(); + let app = TestApp::new().await; + + for i in 0..25 { + seed_record( + &app, + &format!("at://did:plc:seed/app.test.post/{i}"), + "app.test.post", + ) + .await; + } + seed_record(&app, "at://did:plc:seed/app.other.post/1", "app.other.post").await; + assert_eq!(count_in_collection(&app, "app.test.post").await, 25); + + let id = Uuid::new_v4().to_string(); + let now = happyview::db::now_rfc3339(); + // inherit_auth is a bool column; on Postgres a literal `0` doesn't + // implicitly cast to boolean, so it must be a bound parameter (as + // seed_job_with_dpop does above), not embedded as a literal in the SQL. + let sql = adapt_sql( + "INSERT INTO happyview_jobs (id, job_type, status, input, progress, created_by, created_at, inherit_auth) \ + VALUES (?, 'happyview.delete-collection', 'pending', ?, '{}', ?, ?, ?)", + app.state.db_backend, + ); + happyview::db::query(&sql) + .bind(&id) + .bind(r#"{"collection":"app.test.post"}"#) + .bind(&app.admin_did) + .bind(&now) + .bind(false) + .execute(&app.state.db) + .await + .expect("enqueue failed"); + + let worker = tokio::spawn(happyview::jobs::worker::run_worker(app.state.clone())); + let (status, result, error) = await_terminal_status(&app, &id).await; + worker.abort(); + + assert_eq!(status, "completed", "job error: {error:?}"); + let result: Value = serde_json::from_str(&result.expect("no result")).unwrap(); + assert_eq!(result["deleted"], 25); + + assert_eq!(count_in_collection(&app, "app.test.post").await, 0); + assert_eq!( + count_in_collection(&app, "app.other.post").await, + 1, + "other collections must be untouched" + ); +} + +/// Insert `count` records into `collection` in chunks, staying well under +/// SQLite's default bound variable limit (999) per statement. +async fn seed_records_bulk(app: &TestApp, collection: &str, count: usize) { + const CHUNK: usize = 100; + let now = happyview::db::now_rfc3339(); + let mut i = 0usize; + while i < count { + let n = std::cmp::min(CHUNK, count - i); + let placeholders = vec!["(?, ?, ?, ?, ?, ?, ?, ?)"; n].join(", "); + let sql = adapt_sql( + &format!( + "INSERT INTO happyview_records (uri, did, collection, rkey, record, cid, indexed_at, created_at) \ + VALUES {placeholders}" + ), + app.state.db_backend, + ); + let mut q = happyview::db::query(&sql); + for j in 0..n { + let idx = i + j; + q = q + .bind(format!("at://did:plc:bulk/{collection}/{idx}")) + .bind("did:plc:bulk".to_string()) + .bind(collection.to_string()) + .bind(format!("rkey{idx}")) + .bind(r#"{"text":"bulk"}"#.to_string()) + .bind(format!("bafybulk{idx}")) + .bind(now.clone()) + .bind(now.clone()); + } + q.execute(&app.state.db) + .await + .expect("seed_records_bulk: insert failed"); + i += n; + } +} + +async fn seed_job_row_with_status( + app: &TestApp, + id: &str, + job_type: &str, + status: &str, + input: &Value, +) { + let sql = adapt_sql( + "INSERT INTO happyview_jobs (id, job_type, status, input, progress, created_by, created_at, inherit_auth) \ + VALUES (?, ?, ?, ?, '{}', ?, ?, ?)", + app.state.db_backend, + ); + let now = happyview::db::now_rfc3339(); + happyview::db::query(&sql) + .bind(id) + .bind(job_type) + .bind(status) + .bind(input.to_string()) + .bind(&app.admin_did) + .bind(&now) + .bind(false) + .execute(&app.state.db) + .await + .expect("seed_job_row_with_status failed"); +} + +// Regression guard for B1: the delete loop's only historical exit was +// `affected == 0`, which never happens on a collection still receiving +// inserts (Jetstream ingest). The fix breaks on a *short* batch instead. +// BATCH_SIZE is 5000 (see delete_collection.rs); seeding one row over a full +// batch forces two DELETE statements — a full batch (affected == BATCH_SIZE, +// loop must continue) then a short one (affected == 1 < BATCH_SIZE, loop must +// break) — so an inverted or off-by-one comparison (e.g. `<=`) would leave a +// row behind and fail this test's final count assertions. +#[tokio::test] +#[serial] +async fn native_delete_collection_handles_more_than_one_batch() { + common::require_db!(); + let app = TestApp::new().await; + + const SEEDED: usize = 5001; + seed_records_bulk(&app, "app.bulk.post", SEEDED).await; + assert_eq!( + count_in_collection(&app, "app.bulk.post").await, + SEEDED as i64 + ); + + let id = Uuid::new_v4().to_string(); + let now = happyview::db::now_rfc3339(); + let sql = adapt_sql( + "INSERT INTO happyview_jobs (id, job_type, status, input, progress, created_by, created_at, inherit_auth) \ + VALUES (?, 'happyview.delete-collection', 'pending', ?, '{}', ?, ?, ?)", + app.state.db_backend, + ); + happyview::db::query(&sql) + .bind(&id) + .bind(r#"{"collection":"app.bulk.post"}"#) + .bind(&app.admin_did) + .bind(&now) + .bind(false) + .execute(&app.state.db) + .await + .expect("enqueue failed"); + + let worker = tokio::spawn(happyview::jobs::worker::run_worker(app.state.clone())); + let (status, result, error) = await_terminal_status(&app, &id).await; + worker.abort(); + + assert_eq!(status, "completed", "job error: {error:?}"); + let result: Value = serde_json::from_str(&result.expect("no result")).unwrap(); + assert_eq!(result["deleted"], SEEDED); + assert_eq!(count_in_collection(&app, "app.bulk.post").await, 0); +} + +// Regression guard for B3: the should_stop branch was untested. Pre-setting +// the job row's status to `cancelling` before calling the handler makes +// should_stop's very first check (which runs before any batch) fire on +// iteration one — deterministic, no need to race the 5000-row batch. +#[tokio::test] +#[serial] +async fn native_delete_collection_stops_on_cancelling_without_failing() { + common::require_db!(); + let app = TestApp::new().await; + + seed_record(&app, "at://did:plc:seed/app.stop.post/1", "app.stop.post").await; + seed_record(&app, "at://did:plc:seed/app.stop.post/2", "app.stop.post").await; + seed_record(&app, "at://did:plc:seed/app.stop.post/3", "app.stop.post").await; + + let id = Uuid::new_v4().to_string(); + let input = json!({"collection": "app.stop.post"}); + seed_job_row_with_status( + &app, + &id, + "happyview.delete-collection", + "cancelling", + &input, + ) + .await; + + let job = happyview::jobs::Job { + id: id.clone(), + job_type: "happyview.delete-collection".into(), + status: "cancelling".into(), + input, + progress: json!({}), + result: None, + error: None, + created_by: app.admin_did.clone(), + started_at: None, + completed_at: None, + created_at: happyview::db::now_rfc3339(), + inherit_auth: false, + api_client_id: None, + dpop_key_id: None, + }; + + let outcome = happyview::jobs::native::delete_collection::run(&app.state, &job).await; + + match outcome { + happyview::jobs::native::NativeOutcome::Completed(result) => { + assert_eq!(result["deleted"], 0); + assert_eq!(result["stopped"], "cancelling"); + } + happyview::jobs::native::NativeOutcome::Failed(e) => { + panic!("a deliberate cancel must not be reported as a failure: {e}"); + } + } + + assert_eq!( + count_in_collection(&app, "app.stop.post").await, + 3, + "no records should be deleted once should_stop fires before the first batch" + ); +} + +#[tokio::test] +#[serial] +async fn native_delete_collection_stops_on_pausing_without_failing() { + common::require_db!(); + let app = TestApp::new().await; + + seed_record(&app, "at://did:plc:seed/app.pause.post/1", "app.pause.post").await; + + let id = Uuid::new_v4().to_string(); + let input = json!({"collection": "app.pause.post"}); + seed_job_row_with_status(&app, &id, "happyview.delete-collection", "pausing", &input).await; + + let job = happyview::jobs::Job { + id: id.clone(), + job_type: "happyview.delete-collection".into(), + status: "pausing".into(), + input, + progress: json!({}), + result: None, + error: None, + created_by: app.admin_did.clone(), + started_at: None, + completed_at: None, + created_at: happyview::db::now_rfc3339(), + inherit_auth: false, + api_client_id: None, + dpop_key_id: None, + }; + + let outcome = happyview::jobs::native::delete_collection::run(&app.state, &job).await; + + match outcome { + happyview::jobs::native::NativeOutcome::Completed(result) => { + assert_eq!(result["deleted"], 0); + assert_eq!(result["stopped"], "pausing"); + } + happyview::jobs::native::NativeOutcome::Failed(e) => { + panic!("a deliberate pause must not be reported as a failure: {e}"); + } + } + + assert_eq!( + count_in_collection(&app, "app.pause.post").await, + 1, + "no records should be deleted once should_stop fires before the first batch" + ); +} diff --git a/tests/e2e_setup.rs b/tests/e2e_setup.rs index 14c5746..17169ce 100644 --- a/tests/e2e_setup.rs +++ b/tests/e2e_setup.rs @@ -798,3 +798,58 @@ async fn plc_submit_rejects_non_attach_account_mode() { "plc_submit should reject non-attach_account mode" ); } + +// --------------------------------------------------------------------------- +// Setup exempts new instances from the one-time vacuum +// --------------------------------------------------------------------------- + +/// A database created by this version already has incremental auto-vacuum and +/// nothing stranded, so completing setup must record the one-time vacuum as +/// unnecessary — otherwise a brand-new instance would be prompted to rebuild a +/// database that has nothing to reclaim. +#[tokio::test] +#[serial] +async fn setup_complete_marks_vacuum_not_needed() { + common::require_db!(); + let app = TestApp::new().await; + + let resp = app + .router + .clone() + .oneshot( + app.authed_request() + .method("POST") + .uri("/api/setup/identity") + .header("content-type", "application/json") + .body(Body::from( + serde_json::to_vec(&json!({"mode": "not_exposed"})).unwrap(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NO_CONTENT); + + let resp = app + .router + .clone() + .oneshot( + app.authed_request() + .method("POST") + .uri("/api/setup/complete") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NO_CONTENT); + + let status = happyview::maintenance::vacuum::read_status(&app.state.db, app.state.db_backend) + .await + .expect("failed to read vacuum status"); + assert!( + status.completed_at.is_some(), + "completing setup must mark the vacuum as not needed, so a fresh \ + instance is never prompted to run one" + ); +} diff --git a/tests/lua_atproto_api.rs b/tests/lua_atproto_api.rs index 4643521..da71956 100644 --- a/tests/lua_atproto_api.rs +++ b/tests/lua_atproto_api.rs @@ -21,6 +21,7 @@ async fn test_state_with_pool(pool: sqlx::AnyPool, backend: DatabaseBackend) -> port: 3000, database_url: String::new(), database_backend: backend, + sqlite_journal_size_limit: happyview::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), diff --git a/tests/lua_db_api.rs b/tests/lua_db_api.rs index ce6bded..3b804ab 100644 --- a/tests/lua_db_api.rs +++ b/tests/lua_db_api.rs @@ -24,6 +24,7 @@ async fn test_state_with_pool(pool: sqlx::AnyPool, backend: DatabaseBackend) -> port: 3000, database_url: String::new(), database_backend: backend, + sqlite_journal_size_limit: happyview::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), diff --git a/tests/lua_record_api.rs b/tests/lua_record_api.rs index 9764601..346f4d0 100644 --- a/tests/lua_record_api.rs +++ b/tests/lua_record_api.rs @@ -29,6 +29,7 @@ async fn test_state_with_pool(pool: sqlx::AnyPool, backend: DatabaseBackend) -> port: 3000, database_url: String::new(), database_backend: backend, + sqlite_journal_size_limit: happyview::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), base_path: None, session_secret: "test-secret".into(), diff --git a/tests/maintenance_vacuum.rs b/tests/maintenance_vacuum.rs new file mode 100644 index 0000000..a705489 --- /dev/null +++ b/tests/maintenance_vacuum.rs @@ -0,0 +1,268 @@ +mod common; + +use happyview::db::DatabaseBackend; +use happyview::maintenance::vacuum; +use serial_test::serial; + +/// All tests below share one SQLite file (`TEST_DATABASE_URL`) and mutate the +/// same `happyview_instance_settings` keys, so they must not run concurrently +/// with each other (see `#[serial]` on each). +async fn sqlite_pool() -> Option { + if common::db::test_backend() != DatabaseBackend::Sqlite { + eprintln!("skipped (SQLite-only)"); + return None; + } + Some(common::db::test_pool().await) +} + +/// A database created by this version must have incremental auto-vacuum, which +/// is what makes `PRAGMA incremental_vacuum` able to release pages later. +#[tokio::test] +#[serial] +async fn fresh_sqlite_database_has_incremental_auto_vacuum() { + common::require_db!(); + let Some(pool) = sqlite_pool().await else { + return; + }; + + let (mode,): (i64,) = happyview::db::query_as("PRAGMA auto_vacuum") + .fetch_one(&pool) + .await + .expect("failed to read auto_vacuum"); + + assert_eq!(mode, 2, "expected auto_vacuum=INCREMENTAL (2), got {mode}"); +} + +#[tokio::test] +#[serial] +async fn run_if_requested_is_a_noop_when_not_armed() { + common::require_db!(); + let Some(pool) = sqlite_pool().await else { + return; + }; + common::db::truncate_all(&pool).await; + + let out = vacuum::run_if_requested(&pool, DatabaseBackend::Sqlite, &test_url()) + .await + .expect("run_if_requested failed"); + assert!(out.is_none(), "expected no run when unarmed"); +} + +#[tokio::test] +#[serial] +async fn requesting_then_running_records_completion_and_disarms() { + common::require_db!(); + let Some(pool) = sqlite_pool().await else { + return; + }; + common::db::truncate_all(&pool).await; + + vacuum::request(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + let status = vacuum::read_status(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + assert!(status.requested_at.is_some()); + assert!(status.completed_at.is_none()); + + let out = vacuum::run_if_requested(&pool, DatabaseBackend::Sqlite, &test_url()) + .await + .expect("run_if_requested failed") + .expect("expected a result"); + assert_eq!(out.status, "ok", "vacuum failed: {:?}", out.error); + + let status = vacuum::read_status(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + assert!(status.completed_at.is_some(), "completed_at not written"); + assert!(status.requested_at.is_none(), "requested_at not cleared"); + assert!( + status.attempt_started_at.is_none(), + "attempt marker not cleared" + ); + + // Second call must not run again. + let again = vacuum::run_if_requested(&pool, DatabaseBackend::Sqlite, &test_url()) + .await + .unwrap(); + assert!(again.is_none()); +} + +#[tokio::test] +#[serial] +async fn a_crashed_previous_attempt_is_detected_and_disarmed() { + common::require_db!(); + let Some(pool) = sqlite_pool().await else { + return; + }; + common::db::truncate_all(&pool).await; + + // Simulate a process that died mid-vacuum: armed, attempt started, no result. + vacuum::request(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + set_setting( + &pool, + "sqlite_vacuum_attempt_started_at", + "2026-01-01T00:00:00Z", + ) + .await; + + let out = vacuum::run_if_requested(&pool, DatabaseBackend::Sqlite, &test_url()) + .await + .expect("run_if_requested failed") + .expect("expected a result describing the crash"); + + assert_eq!(out.status, "failed"); + let err = out.error.unwrap_or_default(); + assert!( + err.contains("did not complete"), + "error should name the crash, got: {err}" + ); + + let status = vacuum::read_status(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + assert!( + status.requested_at.is_none(), + "must disarm, not retry forever" + ); + assert!(status.completed_at.is_none()); +} + +#[tokio::test] +#[serial] +async fn mark_not_needed_records_completion_without_running() { + common::require_db!(); + let Some(pool) = sqlite_pool().await else { + return; + }; + common::db::truncate_all(&pool).await; + + vacuum::mark_not_needed(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + let status = vacuum::read_status(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + assert!(status.completed_at.is_some()); +} + +/// `cancel_request` is currently called by nothing in production, but it is +/// the only way to back out of an armed vacuum before it runs. +#[tokio::test] +#[serial] +async fn cancel_request_disarms_without_running() { + common::require_db!(); + let Some(pool) = sqlite_pool().await else { + return; + }; + common::db::truncate_all(&pool).await; + + vacuum::request(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + vacuum::cancel_request(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + + let status = vacuum::read_status(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + assert!( + status.requested_at.is_none(), + "cancel_request must clear the arm" + ); + + let out = vacuum::run_if_requested(&pool, DatabaseBackend::Sqlite, &test_url()) + .await + .unwrap(); + assert!(out.is_none(), "a cancelled request must not run"); +} + +/// The other tests only prove VACUUM returned without error. This proves it +/// actually did something: bloat the file, delete the rows (which, under +/// incremental auto-vacuum, does *not* shrink the file on its own), then +/// vacuum and check the file got smaller. +#[tokio::test] +#[serial] +async fn a_real_vacuum_reclaims_freed_space() { + common::require_db!(); + let Some(pool) = sqlite_pool().await else { + return; + }; + common::db::truncate_all(&pool).await; + + happyview::db::query( + "CREATE TABLE IF NOT EXISTS hv_vacuum_scratch (id INTEGER PRIMARY KEY, data BLOB)", + ) + .execute(&pool) + .await + .expect("failed to create scratch table"); + + let blob = vec![0u8; 300_000]; + for _ in 0..20 { + happyview::db::query("INSERT INTO hv_vacuum_scratch (data) VALUES (?)") + .bind(blob.clone()) + .execute(&pool) + .await + .expect("failed to insert scratch row"); + } + happyview::db::query("DELETE FROM hv_vacuum_scratch") + .execute(&pool) + .await + .expect("failed to delete scratch rows"); + // Force the bloat out of the WAL and into the main file before measuring, + // so `db_bytes_before` actually reflects it. + happyview::db::query("PRAGMA wal_checkpoint(FULL)") + .execute(&pool) + .await + .expect("failed to checkpoint before measuring"); + + vacuum::request(&pool, DatabaseBackend::Sqlite) + .await + .unwrap(); + + let out = vacuum::run_if_requested(&pool, DatabaseBackend::Sqlite, &test_url()) + .await + .expect("run_if_requested failed") + .expect("expected a result"); + + assert_eq!(out.status, "ok", "vacuum failed: {:?}", out.error); + assert!(out.db_bytes_before > 0, "expected a nonzero before size"); + assert!( + out.db_bytes_after < out.db_bytes_before, + "expected the file to shrink: before={}, after={}", + out.db_bytes_before, + out.db_bytes_after + ); + assert!( + out.reclaimed_bytes > 0, + "expected reclaimed_bytes > 0, got {}", + out.reclaimed_bytes + ); + + happyview::db::query("DROP TABLE IF EXISTS hv_vacuum_scratch") + .execute(&pool) + .await + .ok(); +} + +fn test_url() -> String { + std::env::var("TEST_DATABASE_URL").expect("TEST_DATABASE_URL") +} + +async fn set_setting(pool: &sqlx::AnyPool, key: &str, value: &str) { + let sql = happyview::db::adapt_sql( + "INSERT INTO happyview_instance_settings (key, value, updated_at) VALUES (?, ?, ?)", + DatabaseBackend::Sqlite, + ); + happyview::db::query(&sql) + .bind(key) + .bind(value) + .bind(happyview::db::now_rfc3339()) + .execute(pool) + .await + .expect("set_setting failed"); +} diff --git a/web/src/app/dashboard/jobs/page.tsx b/web/src/app/dashboard/jobs/page.tsx index 4de942b..04474f6 100644 --- a/web/src/app/dashboard/jobs/page.tsx +++ b/web/src/app/dashboard/jobs/page.tsx @@ -1,6 +1,7 @@ "use client"; import { useCallback, useEffect, useRef, useState } from "react"; +import { useSearchParams } from "next/navigation"; import { toast } from "sonner"; import { CheckCircle2, @@ -15,6 +16,7 @@ import { useCurrentUser } from "@/hooks/use-current-user"; import { toastError } from "@/lib/format"; import { cancelJob, + getJob, getJobLogs, getJobs, pauseJob, @@ -156,9 +158,15 @@ function relativeTime(dateStr: string): string { export default function JobsPage() { const { hasPermission } = useCurrentUser(); + const searchParams = useSearchParams(); const [jobs, setJobs] = useState([]); const [statusFilter, setStatusFilter] = useState("all"); const [selectedJobId, setSelectedJobId] = useState(null); + // Holds a job opened via the `?job=` deep link that may not be present in + // the current (filtered/paginated) `jobs` list — fetched directly by id + // rather than relying on it showing up in the list. + const [linkedJob, setLinkedJob] = useState(null); + const appliedDeepLink = useRef(false); const [loading, setLoading] = useState(true); const load = useCallback(() => { @@ -193,7 +201,28 @@ export default function JobsPage() { return () => clearInterval(interval); }, [load, hasActiveJobs]); - const selectedJob = jobs.find((j) => j.id === selectedJobId) ?? null; + // Open a job's detail sheet from a `?job=` deep link (e.g. the "View + // job" toast after enqueueing a collection delete). Fetched directly by id + // rather than found in `jobs`, since the linked job may not be on the + // current page or may not match the default status filter. + useEffect(() => { + if (appliedDeepLink.current) return; + const jobParam = searchParams.get("job"); + if (!jobParam) return; + appliedDeepLink.current = true; + getJob(jobParam) + .then((job) => { + setLinkedJob(job); + setSelectedJobId(job.id); + }) + .catch((e) => { + toastError("Failed to open linked job", e); + }); + }, [searchParams]); + + const selectedJob = + jobs.find((j) => j.id === selectedJobId) ?? + (linkedJob?.id === selectedJobId ? linkedJob : null); const canManage = hasPermission("jobs:manage"); return ( @@ -294,6 +323,7 @@ export default function JobsPage() { onOpenChange={(open) => { if (!open) { setSelectedJobId(null); + setLinkedJob(null); load(); } }} diff --git a/web/src/app/dashboard/layout.tsx b/web/src/app/dashboard/layout.tsx index 8ce25b1..3384e06 100644 --- a/web/src/app/dashboard/layout.tsx +++ b/web/src/app/dashboard/layout.tsx @@ -8,6 +8,7 @@ import { useAuth } from "@/lib/auth-context" import { useConfig } from "@/lib/config-context" import { AppSidebar } from "@/components/app-sidebar" import { PluginUpdateProvider } from "@/components/plugin-update-provider" +import { VacuumPromptProvider } from "@/components/vacuum-prompt-provider" import { RestartProvider } from "@/lib/restart-context" import { SidebarInset, SidebarProvider } from "@/components/ui/sidebar" import { Toaster } from "@/components/ui/sonner" @@ -50,19 +51,27 @@ export default function DashboardLayout({ return ( - - - {children} - + {/* VacuumPromptProvider calls useRestart() to register the "vacuum + scheduled" restart reason, so it must be nested INSIDE + RestartProvider — a provider can't consume a context that is its + own descendant; React resolves context by tree position, not + call order, so getting this nesting backwards silently falls + through to the context's no-op default instead of erroring. */} + + + + {children} + + + - ) } diff --git a/web/src/app/dashboard/records/page.tsx b/web/src/app/dashboard/records/page.tsx index c011364..9394334 100644 --- a/web/src/app/dashboard/records/page.tsx +++ b/web/src/app/dashboard/records/page.tsx @@ -10,8 +10,9 @@ import { } from "@tanstack/react-table"; import { toast } from "sonner"; -import { useSearchParams } from "next/navigation"; +import { useRouter, useSearchParams } from "next/navigation"; import { useCurrentUser } from "@/hooks/use-current-user"; +import { useVacuumStatus } from "@/components/vacuum-prompt-provider"; import { toastError } from "@/lib/format"; import { getCollections, @@ -91,7 +92,15 @@ function formatCellValue(value: unknown): string { export default function RecordsPage() { const { hasPermission } = useCurrentUser(); + const router = useRouter(); const searchParams = useSearchParams(); + // `RecordsPage` sits inside `VacuumPromptProvider`, so it can tell whether + // a collection delete on this instance will actually shrink the database + // file: on SQLite, `incremental_vacuum` (the delete job's final step) is a + // no-op until the one-time vacuum has run once. + const { status: vacuumStatus } = useVacuumStatus(); + const vacuumNotDone = + vacuumStatus?.backend === "sqlite" && !vacuumStatus.vacuum.completed_at; const initialCollection = searchParams.get("collection") ?? ""; const appliedInitial = useRef(false); const [collections, setCollections] = useState([]); @@ -191,24 +200,30 @@ export default function RecordsPage() { if (!selectedCollection) return; setDeletingAll(true); try { - await deleteCollectionRecords(selectedCollection); + const { job_id } = await deleteCollectionRecords(selectedCollection); setBulkDeleteOpen(false); setBulkDeleteMode("selected"); setBulkDeleteConfirm(""); setRowSelection({}); - toast.success("All records deleted"); - const data = await getCollections(); - setCollections(data.collections); - setCursorStack([]); - setNextCursor(undefined); - setRecords([]); - setSelectedCollection(""); + toast.success("Deletion started", { + description: vacuumNotDone + ? "Records are being deleted in the background. Jobs run one at a time, so this may wait behind other work. This instance has not run its one-time database cleanup yet, so the freed space will not be returned to the disk until that runs." + : "Records are being deleted in the background. Jobs run one at a time, so this may wait behind other work.", + action: { + label: "View job", + onClick: () => router.push(`/dashboard/jobs?job=${job_id}`), + }, + }); + // Nothing has actually been deleted yet — the job may still be pending + // behind other work — so the view is left as-is rather than clearing + // the selection and records list, which previously read as "your + // collection is already gone" while it's still sitting right there. } catch (e: unknown) { toastError("Failed to delete collection", e); } finally { setDeletingAll(false); } - }, [selectedCollection]); + }, [selectedCollection, router, vacuumNotDone]); const handleBulkDelete = useCallback(async () => { setDeleting(true); @@ -665,6 +680,22 @@ export default function RecordsPage() { {bulkDeleteMode === "all" && (
+ {vacuumNotDone && ( +

+ This instance has not run its one-time database + cleanup yet, so deleted space will not be returned to + the disk until that runs.{" "} + +

+ )}