diff --git a/Cargo.lock b/Cargo.lock index 37170e76..77523ac5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -206,6 +206,7 @@ dependencies = [ "log", "open", "rayon", + "redb 4.2.0", "reqwest", "serde", "serde_json", diff --git a/README.md b/README.md index 951065c1..131f6151 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,31 @@ The **lock is the authority; the socket or named pipe is only transport**. The l Repositories created before `atomic.redb` kept graph state in `.atomic/pristine.redb` and the journal in `.atomic/changes.redb`. The first open by a newer `atomic` merges both into `atomic.redb`, verifies every table against its source, and moves the old files to `.atomic/legacy//`. If an owner started by an older `atomic` still holds `changes.redb`, run `atomic agent database-owner shutdown` in that repository first. +### Reclaiming database space + +Run `atomic compact` to reclaim unused space in an existing `atomic.redb`: + +```sh +atomic compact +atomic compact --repository /path/to/repo --json +``` + +The command starts or reconnects to the repository's database owner, waits for +in-flight database requests to finish, and compacts with exclusive access. +It reports `database`, `before_bytes`, `after_bytes`, and `reclaimed_bytes`. +Running it inside a sandbox compacts that sandbox's canonical repository. +Changes, provenance, views, and working-copy files are preserved; this does not +delete history or `.change` files. The amount reclaimed depends on unused space, +and may be zero. Compaction runs only when explicitly requested. + +Use a quiet period for large repositories: database operations wait while +compaction runs and can exhaust their normal lock-wait budgets. The owner's +`ATOMIC_DB_LOCK_WAIT_MS` setting bounds acquisition waits, not the duration of +an already-started compaction. Owners started by an older executable must be +shut down once before retrying. For a legacy repository, first open it with the +current Atomic to complete its database migration; `compact` does not migrate +or create a database. + Within one repository, redb write transactions are serialized by design, while reads and independent repositories can proceed concurrently. Startup, reconnect, and retry loops are bounded so a broken owner fails instead of hanging hooks indefinitely. Checkpoint attempts, event cutoffs, and fencing generations let interrupted turns resume without rewriting completed session turns. Current operational considerations: diff --git a/atomic-cli/Cargo.toml b/atomic-cli/Cargo.toml index a60709a4..efbae9cf 100644 --- a/atomic-cli/Cargo.toml +++ b/atomic-cli/Cargo.toml @@ -43,6 +43,9 @@ console = { workspace = true } # Async runtime tokio = { workspace = true } +# Exclusive maintenance inside the database owner +redb = { workspace = true } + # Error handling anyhow = { workspace = true } thiserror = { workspace = true } diff --git a/atomic-cli/src/commands/agent/mod.rs b/atomic-cli/src/commands/agent/mod.rs index dc4df1ea..806f8d34 100644 --- a/atomic-cli/src/commands/agent/mod.rs +++ b/atomic-cli/src/commands/agent/mod.rs @@ -47,6 +47,8 @@ mod lifecycle; mod owner; mod status; +pub(crate) use owner::compact_repository; + use clap::{Args, Subcommand}; use crate::commands::Command; diff --git a/atomic-cli/src/commands/agent/owner.rs b/atomic-cli/src/commands/agent/owner.rs index e1ba1f16..e363824c 100644 --- a/atomic-cli/src/commands/agent/owner.rs +++ b/atomic-cli/src/commands/agent/owner.rs @@ -12,7 +12,7 @@ use std::fs::{File, OpenOptions}; use std::ops::Deref; use std::path::{Path, PathBuf}; use std::process::{Command as ProcessCommand, Stdio}; -use std::sync::{Arc, Mutex, TryLockError, Weak}; +use std::sync::{Arc, Mutex, MutexGuard, TryLockError, Weak}; use std::time::{Duration, Instant}; use anyhow::{anyhow, Context}; @@ -28,7 +28,7 @@ use atomic_repository::redb_change_store::{ ProvenanceCheckpointSource, ProvenanceId, ProvenanceTurnState, RedbChangeStore, StopCause, StopState, StoredProvenanceTurn, MAX_FROZEN_PAGE_FRAGMENTS, }; -use atomic_repository::{ensure_database, Repository, RepositoryError}; +use atomic_repository::{ensure_database, Repository, RepositoryError, DATABASE_FILE}; use clap::{Args, Subcommand}; use fs2::FileExt; use serde::{Deserialize, Serialize}; @@ -39,6 +39,9 @@ use uuid::Uuid; use crate::commands::Command; use crate::error::CliResult; +mod maintenance; +use maintenance::{compact_database, is_database_busy, DatabaseCompaction}; + const PROTOCOL_VERSION: u16 = 1; // Local IPC only; a single oversized envelope (e.g. a huge recovered reasoning // block) must still fit one frame even after client-side chunking. @@ -118,6 +121,7 @@ struct RequestFrame { #[derive(Clone, Debug, Serialize, Deserialize)] enum OwnerRequest { Ping, + Compact, ReserveProvenanceTurn { session_id: String, turn_number: u32, @@ -203,6 +207,11 @@ pub(crate) enum OwnerResponse { // Additive v1 capability: old clients ignore it, old owners omit it. #[serde(default)] frozen_envelope_paging: bool, + #[serde(default)] + database_compaction: bool, + }, + Compacted { + report: DatabaseCompaction, }, ProvenanceTurn { turn: StoredProvenanceTurn, @@ -838,6 +847,29 @@ pub(crate) fn start_or_reconnect(repository: &Path) -> anyhow::Result anyhow::Result { + require_compaction(start_or_reconnect(repository)?)?; + match request(repository, OwnerRequest::Compact)? { + OwnerResponse::Compacted { report } => Ok(report), + other => Err(unexpected_response("compact", other)), + } +} + +fn require_compaction(response: OwnerResponse) -> anyhow::Result<()> { + match response { + OwnerResponse::Pong { + database_compaction: true, + .. + } => Ok(()), + OwnerResponse::Pong { .. } => Err(anyhow!( + "the running database owner does not support compaction; run \ + `atomic agent database-owner shutdown --repository ` and retry" + )), + other => Err(unexpected_response("ping", other)), + } +} + fn request_with_reconnect( repository: &Path, request_body: OwnerRequest, @@ -932,9 +964,23 @@ impl StoreLease { fn acquire(&self, started: Instant) -> anyhow::Result> { let timeout = atomic_agent::turn::orchestrator::wait_budget::database_wait(); - let mut current = loop { + let mut current = self.lock_current(started, timeout)?; + if let Some(store) = current.upgrade() { + return Ok(store); + } + let store = Arc::new(self.open_waiting(started, timeout)?); + *current = Arc::downgrade(&store); + Ok(store) + } + + fn lock_current( + &self, + started: Instant, + timeout: Duration, + ) -> anyhow::Result>> { + loop { match self.current.try_lock() { - Ok(guard) => break guard, + Ok(guard) => return Ok(guard), Err(TryLockError::Poisoned(_)) => { return Err(anyhow!("database-owner store lease is poisoned")); } @@ -949,13 +995,45 @@ impl StoreLease { )); } } - }; - if let Some(store) = current.upgrade() { - return Ok(store); } - let store = Arc::new(self.open_waiting(started, timeout)?); - *current = Arc::downgrade(&store); - Ok(store) + } + + fn compact(&self, started: Instant) -> anyhow::Result { + let timeout = atomic_agent::turn::orchestrator::wait_budget::database_wait(); + // Keep this gate locked through compaction. Existing requests can + // finish and release their Arcs, but no new store lease may open. + let current = self.lock_current(started, timeout)?; + while current.strong_count() != 0 { + if started.elapsed() >= timeout { + return Err(anyhow!( + "database compaction waited {timeout:?} for active requests" + )); + } + std::thread::sleep(DATABASE_RETRY_DELAY.min(timeout.saturating_sub(started.elapsed()))); + } + // The owner may have started with `--repository .` in a different + // working directory from this client (for example a sandbox). + let path = self.dot_dir.join(DATABASE_FILE); + let path = std::fs::canonicalize(&path) + .with_context(|| format!("failed to locate database {}", path.display()))?; + loop { + match compact_database(&path) { + Err(error) if is_database_busy(&error) && started.elapsed() < timeout => { + std::thread::sleep( + DATABASE_RETRY_DELAY.min(timeout.saturating_sub(started.elapsed())), + ); + } + Err(error) if is_database_busy(&error) => { + return Err(anyhow!( + "database compaction waited {timeout:?} for {} to become available", + path.display() + )); + } + result => { + return result.with_context(|| format!("failed to compact {}", path.display())) + } + } + } } fn open_waiting(&self, started: Instant, timeout: Duration) -> anyhow::Result { @@ -1082,6 +1160,14 @@ where }; let (response, shutdown) = match frame.request { OwnerRequest::Ping => (pong(), false), + // Maintenance needs StoreLease's exclusive gate, not a shared handle. + OwnerRequest::Compact => ( + OwnerResponse::Error { + code: "database-compaction".into(), + message: "compaction requires an exclusive owner lease".into(), + }, + false, + ), OwnerRequest::ReserveProvenanceTurn { session_id, turn_number, @@ -1426,6 +1512,7 @@ fn pong() -> OwnerResponse { OwnerResponse::Pong { pid: std::process::id(), frozen_envelope_paging: true, + database_compaction: true, } } @@ -1448,7 +1535,18 @@ where // Opening the database may wait for another process; keep that off the // runtime threads that accept connections and answer pings. let (response, should_shutdown) = tokio::task::spawn_blocking(move || { - handle_request_with(request, || lease.acquire(started)) + if request.version == PROTOCOL_VERSION && matches!(request.request, OwnerRequest::Compact) { + let response = match lease.compact(started) { + Ok(report) => OwnerResponse::Compacted { report }, + Err(error) => OwnerResponse::Error { + code: "database-compaction".into(), + message: format!("{error:#}"), + }, + }; + respond(request.request_id, response, false) + } else { + handle_request_with(request, || lease.acquire(started)) + } }) .await .context("database-owner request handler panicked")?; @@ -1704,6 +1802,56 @@ fn configure_detached(command: &mut ProcessCommand) -> anyhow::Result<()> { mod tests { use super::*; + #[test] + fn legacy_owner_requires_an_explicit_restart_for_compaction() { + let old: OwnerResponse = serde_json::from_str(r#"{"Pong":{"pid":42}}"#).unwrap(); + assert!(require_compaction(old) + .unwrap_err() + .to_string() + .contains("database-owner shutdown")); + assert!(require_compaction(pong()).is_ok()); + } + + #[test] + fn compaction_drains_active_leases_before_opening_the_database() { + let dir = tempfile::tempdir().unwrap(); + let repo = Repository::init(dir.path()).unwrap(); + let lease = Arc::new(StoreLease::new(repo.dot_dir().to_path_buf())); + drop(repo); + let held = lease.acquire(Instant::now()).unwrap(); + let worker_lease = Arc::clone(&lease); + let (sender, receiver) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + sender.send(worker_lease.compact(Instant::now())).unwrap(); + }); + let deadline = Instant::now() + Duration::from_secs(5); + // Observe the maintenance gate deterministically before releasing + // the old lease, rather than relying on a sleep racing the worker. + loop { + if matches!(lease.current.try_lock(), Err(TryLockError::WouldBlock)) { + break; + } + assert!( + Instant::now() < deadline, + "compaction did not acquire its gate" + ); + std::thread::yield_now(); + } + assert!(receiver.try_recv().is_err()); + drop(held); + let report = receiver + .recv_timeout(Duration::from_secs(10)) + .unwrap() + .unwrap(); + assert_eq!( + report.database, + std::fs::canonicalize(dir.path().join(".atomic").join(DATABASE_FILE)).unwrap() + ); + worker.join().unwrap(); + // Maintenance releases the gate and its redb handle for new requests. + assert!(lease.acquire(Instant::now()).is_ok()); + } + #[cfg(windows)] #[tokio::test] async fn busy_owner_pipe_waits_for_a_new_listening_instance() { @@ -1869,7 +2017,8 @@ mod tests { .contains("database-owner shutdown")); assert!(require_frozen_paging(OwnerResponse::Pong { pid: 42, - frozen_envelope_paging: true + frozen_envelope_paging: true, + database_compaction: false, }) .is_ok()); } diff --git a/atomic-cli/src/commands/agent/owner/maintenance.rs b/atomic-cli/src/commands/agent/owner/maintenance.rs new file mode 100644 index 00000000..214fe55e --- /dev/null +++ b/atomic-cli/src/commands/agent/owner/maintenance.rs @@ -0,0 +1,170 @@ +//! Exclusive maintenance operations used by the repository database owner. + +use std::path::{Path, PathBuf}; + +use redb::ReadableDatabase; +use serde::{Deserialize, Serialize}; + +/// File sizes before compaction and after the compacted database is closed. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub(crate) struct DatabaseCompaction { + pub database: PathBuf, + pub before_bytes: u64, + pub after_bytes: u64, + pub reclaimed_bytes: u64, +} + +/// Compact an existing database without creating tables or migrating its layout. +/// +/// The owner must first drain its store leases and prevent new ones until this +/// returns. redb's file lock excludes other processes; a busy file is reported +/// as a database-open error so the owner can apply its usual wait budget. +pub(super) fn compact_database(path: &Path) -> anyhow::Result { + let mut database = redb::Builder::new().open(path)?; + atomic_core::pristine::schema::check_schema_version(&database.begin_read()?)?; + let before_bytes = std::fs::metadata(path)?.len(); + database.compact()?; + // Closing redb persists allocator metadata and can change the file length. + // Include that final write in the reported size, while the owner's lease + // gate still prevents new requests from reopening the database. + drop(database); + let after_bytes = std::fs::metadata(path)?.len(); + Ok(DatabaseCompaction { + database: path.to_path_buf(), + before_bytes, + after_bytes, + reclaimed_bytes: before_bytes.saturating_sub(after_bytes), + }) +} + +pub(super) fn is_database_busy(error: &anyhow::Error) -> bool { + matches!( + error.downcast_ref::(), + Some(redb::DatabaseError::DatabaseAlreadyOpen) + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use redb::{Database, ReadableTableMetadata, TableDefinition}; + + const DATA: TableDefinition = TableDefinition::new("compaction_data"); + + #[test] + fn compaction_reclaims_space_and_preserves_surviving_values() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("atomic.redb"); + let db = Database::create(&path).unwrap(); + let value = vec![0x5a; 4096]; + let write = db.begin_write().unwrap(); + { + let mut table = write.open_table(DATA).unwrap(); + for key in 0..1024 { + table.insert(key, value.as_slice()).unwrap(); + } + } + write.commit().unwrap(); + // Keep the old pages pinned through deletion so the fixture has + // reclaimable space without depending on the allocator's layout. + let reader = db.begin_read().unwrap(); + let write = db.begin_write().unwrap(); + { + let mut table = write.open_table(DATA).unwrap(); + for key in 1..1024 { + table.remove(key).unwrap(); + } + } + write.commit().unwrap(); + drop(reader); + drop(db); + + let before = std::fs::metadata(&path).unwrap().len(); + let report = compact_database(&path).unwrap(); + assert_eq!(report.before_bytes, before); + assert!(report.after_bytes < report.before_bytes, "{report:?}"); + assert_eq!(report.after_bytes, std::fs::metadata(&path).unwrap().len()); + assert_eq!(report.reclaimed_bytes, before - report.after_bytes); + let db = redb::Builder::new().open_read_only(&path).unwrap(); + let read = db.begin_read().unwrap(); + assert_eq!(read.list_tables().unwrap().count(), 1); + let table = read.open_table(DATA).unwrap(); + assert_eq!(table.len().unwrap(), 1); + assert_eq!(table.get(0).unwrap().unwrap().value(), value); + drop(table); + drop(read); + drop(db); + // A repeated maintenance request remains valid. + compact_database(&path).unwrap(); + } + + #[test] + fn compaction_does_not_create_a_missing_database() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("atomic.redb"); + assert!(compact_database(&path).is_err()); + assert!(!path.exists()); + } + + #[test] + fn compaction_respects_existing_writer_and_reader_handles() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("atomic.redb"); + let db = Database::create(&path).unwrap(); + assert!(is_database_busy(&compact_database(&path).unwrap_err())); + drop(db); + let db = redb::Builder::new().open_read_only(&path).unwrap(); + assert!(is_database_busy(&compact_database(&path).unwrap_err())); + drop(db); + compact_database(&path).unwrap(); + } + + #[test] + fn compaction_rejects_a_future_repository_schema() { + use atomic_core::pristine::{schema::SCHEMA_VERSION, tables::ATOMIC_META}; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("atomic.redb"); + let db = Database::create(&path).unwrap(); + let write = db.begin_write().unwrap(); + { + let mut meta = write.open_table(ATOMIC_META).unwrap(); + meta.insert( + atomic_core::pristine::schema::SCHEMA_VERSION_KEY, + (SCHEMA_VERSION + 1).to_le_bytes().as_slice(), + ) + .unwrap(); + } + write.commit().unwrap(); + drop(db); + let error = compact_database(&path).unwrap_err().to_string(); + assert!(error.contains("schema"), "{error}"); + } + + #[test] + fn compaction_preserves_persistent_savepoints_and_reports_the_blocker() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("atomic.redb"); + let db = Database::create(&path).unwrap(); + let write = db.begin_write().unwrap(); + let savepoint = write.persistent_savepoint().unwrap(); + write.commit().unwrap(); + drop(db); + + assert!(matches!( + compact_database(&path) + .unwrap_err() + .downcast::() + .unwrap(), + redb::CompactionError::PersistentSavepointExists + )); + let db = redb::Builder::new().open(&path).unwrap(); + let write = db.begin_write().unwrap(); + assert_eq!( + write + .list_persistent_savepoints() + .unwrap() + .collect::>(), + vec![savepoint] + ); + } +} diff --git a/atomic-cli/src/commands/compact.rs b/atomic-cli/src/commands/compact.rs new file mode 100644 index 00000000..f81215d9 --- /dev/null +++ b/atomic-cli/src/commands/compact.rs @@ -0,0 +1,41 @@ +//! Explicit repository database compaction through the local owner. + +use std::path::PathBuf; + +use clap::Args; + +use crate::commands::{agent::compact_repository, Command}; +use crate::error::CliResult; + +/// Reclaim unused space in the repository database. +#[derive(Debug, Args)] +pub struct Compact { + /// Path inside the repository or one of its sandboxes. + #[arg(long, default_value = ".")] + repository: PathBuf, + + /// Emit the database path and before/after byte counts as JSON. + #[arg(long)] + json: bool, +} + +impl Command for Compact { + fn run(&self) -> CliResult<()> { + let report = compact_repository(&self.repository)?; + if self.json { + println!( + "{}", + serde_json::to_string(&report).map_err(anyhow::Error::from)? + ); + } else { + println!( + "Compacted {}: {} -> {} bytes ({} bytes reclaimed)", + report.database.display(), + report.before_bytes, + report.after_bytes, + report.reclaimed_bytes, + ); + } + Ok(()) + } +} diff --git a/atomic-cli/src/commands/mod.rs b/atomic-cli/src/commands/mod.rs index d5ff1859..e5e9f1da 100644 --- a/atomic-cli/src/commands/mod.rs +++ b/atomic-cli/src/commands/mod.rs @@ -69,6 +69,7 @@ use crate::error::{CliError, CliResult}; // Phase 2: Core Local Commands pub mod add; pub mod change; +pub mod compact; pub mod complete; pub mod completions; pub mod conflicts; @@ -136,6 +137,7 @@ pub use add::Add; pub use agent::Agent; pub use change::ChangeCmd; pub use clone::Clone; +pub use compact::Compact; pub use completions::Completions; pub use conflicts::Conflicts; pub use diff::Diff; diff --git a/atomic-cli/src/main.rs b/atomic-cli/src/main.rs index 1cd1e275..8da686c0 100644 --- a/atomic-cli/src/main.rs +++ b/atomic-cli/src/main.rs @@ -61,6 +61,7 @@ use commands::{ ChangeCmd, Clone, Command, + Compact, Completions, Conflicts, Diff, @@ -446,6 +447,9 @@ enum Commands { /// such as backfilling the dependency index for legacy repositories. Doctor(Doctor), + /// Reclaim unused space in the repository database. + Compact(Compact), + /// Git interoperability commands. /// /// Import Git repositories into Atomic, preserving history, authorship, @@ -995,6 +999,7 @@ fn main() { Commands::Diff(diff) => diff.run(), Commands::Doctor(doctor) => doctor.run(), + Commands::Compact(compact) => compact.run(), Commands::Git(git) => git.run(), diff --git a/atomic-cli/tests/database_owner_integration_test.rs b/atomic-cli/tests/database_owner_integration_test.rs index a3437dc9..4536a627 100644 --- a/atomic-cli/tests/database_owner_integration_test.rs +++ b/atomic-cli/tests/database_owner_integration_test.rs @@ -135,6 +135,161 @@ fn spawn_owner(repository: &std::path::Path) -> Child { .expect("spawn foreground database owner") } +// Reap the foreground owner even when a maintenance assertion panics. +struct MaintenanceOwner(Child); + +impl Drop for MaintenanceOwner { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn run_compact(repository: &std::path::Path) -> Output { + command_output( + Command::new(env!("CARGO_BIN_EXE_atomic")) + .arg("compact") + .arg("--repository") + .arg(repository) + .arg("--json"), + ) +} + +#[test] +#[serial] +fn compact_from_sandbox_preserves_history_pending_journal_and_dirty_files() { + let temp = TempDir::new().unwrap(); + let repository = temp.path().join("repo"); + let sandbox = temp.path().join("sandbox"); + let mut repo = Repository::init(&repository).unwrap(); + let contents = b"recorded contents\n"; + std::fs::write(repository.join("file.txt"), contents).unwrap(); + repo.add("file.txt", Default::default()).unwrap(); + let hash = *repo.record_all("compaction fixture").unwrap().hash(); + let change_path = ChangeStore::new(repo.changes_dir(), DEFAULT_CACHE_CAPACITY) + .unwrap() + .change_path(&hash); + let change_bytes = std::fs::read(&change_path).unwrap(); + let view = repo.current_view().to_owned(); + repo.create_view("compact-child").unwrap(); + let views = repo.list_views().unwrap(); + repo.provision_sandbox(&sandbox, &view).unwrap(); + let database = + std::fs::canonicalize(Repository::canonical_database_path(&sandbox).unwrap()).unwrap(); + let store = repo.redb_change_store().unwrap(); + let turn = store.reserve_provenance_turn("compact", 1, 1).unwrap(); + let envelope = serde_json::to_vec(&serde_json::json!({ + "schema_version":1,"event_id":"pending","session_id":"compact", + "turn_number":1,"generation":turn.generation,"timestamp_ms":1700000000000_i64, + "event":{"type":"tool","phase":"after","tool_name":"Read", + "tool_call_id":"pending","input":{"path":"file.txt"}, + "output":"recorded contents","status":"completed"} + })) + .unwrap(); + ProvenanceJournalEnvelope::from_json_bytes(&envelope).unwrap(); + store + .append_provenance_envelope(turn.provenance_id, turn.generation, "pending", &envelope, 2) + .unwrap(); + let pending = store.load_provenance_envelopes(turn.provenance_id).unwrap(); + let turn_before = store + .get_provenance_turn(turn.provenance_id) + .unwrap() + .unwrap(); + drop(store); + drop(repo); + std::fs::write(repository.join("file.txt"), b"unrecorded main edit\n").unwrap(); + std::fs::write(sandbox.join("file.txt"), b"unrecorded sandbox edit\n").unwrap(); + + // Start with a relative path, then ask from another working tree. Reports + // must identify the canonical database independently of the owner's cwd. + let mut owner = MaintenanceOwner( + owner_command(std::path::Path::new("."), "serve") + .current_dir(&repository) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), + ); + let ping_before = wait_for_ping(&repository, &mut owner.0); + let before_bytes = std::fs::metadata(&database).unwrap().len(); + let output = run_compact(&sandbox); + assert_hook_success("compact sandbox", &output); + let report: Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(report["database"], database.to_str().unwrap()); + assert_eq!(report["before_bytes"], before_bytes); + let after_bytes = std::fs::metadata(&database).unwrap().len(); + assert_eq!(report["after_bytes"], after_bytes); + assert_eq!( + report["reclaimed_bytes"], + before_bytes.saturating_sub(after_bytes) + ); + assert_eq!(wait_for_ping(&repository, &mut owner.0), ping_before); + assert_hook_success("repeat compact", &run_compact(&repository)); + // A subsequent store request proves maintenance released the owner gate. + assert_eq!(reserve(&repository)["committed"], true); + + let repo = Repository::open(&repository).unwrap(); + assert_eq!(repo.list_views().unwrap(), views); + assert_eq!(repo.current_view(), view); + assert_eq!(repo.log(Default::default()).unwrap()[0].hash, hash); + for visible_view in [&view, "compact-child"] { + assert_eq!( + repo.get_file_content_on_view("file.txt", visible_view) + .unwrap() + .unwrap(), + contents + ); + } + assert_eq!(std::fs::read(&change_path).unwrap(), change_bytes); + assert_eq!( + std::fs::read(repository.join("file.txt")).unwrap(), + b"unrecorded main edit\n" + ); + assert_eq!( + std::fs::read(sandbox.join("file.txt")).unwrap(), + b"unrecorded sandbox edit\n" + ); + let store = repo.redb_change_store().unwrap(); + assert_eq!( + store.load_provenance_envelopes(turn.provenance_id).unwrap(), + pending + ); + assert_eq!( + store + .get_provenance_turn(turn.provenance_id) + .unwrap() + .unwrap(), + turn_before + ); +} + +#[test] +#[serial] +fn compact_busy_database_times_out_and_owner_can_retry() { + let temp = TempDir::new().unwrap(); + let repository = temp.path().join("repo"); + let held = Repository::init(&repository).unwrap(); + let mut owner = MaintenanceOwner( + owner_command(&repository, "serve") + .env("ATOMIC_DB_LOCK_WAIT_MS", "100") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), + ); + wait_for_ping(&repository, &mut owner.0); + let output = run_compact(&repository); + assert!(!output.status.success()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("database compaction waited"), "{stderr}"); + wait_for_ping(&repository, &mut owner.0); + drop(held); + assert_hook_success("compact after lock release", &run_compact(&repository)); + assert_eq!(reserve(&repository)["committed"], true); +} + fn spawn_owner_with_failpoint( repository: &std::path::Path, failpoint: &str,