diff --git a/crates/tracedecay-global-db/src/registered.rs b/crates/tracedecay-global-db/src/registered.rs index 4402e809fbf..0bb2792f9f3 100644 --- a/crates/tracedecay-global-db/src/registered.rs +++ b/crates/tracedecay-global-db/src/registered.rs @@ -124,6 +124,10 @@ impl From> for SessionAuthorityV1 { } } +// Short-lived store owners have no periodic daemon compactor. Keep their +// one admission reclaim bounded; daemon admission performs no foreground reclaim. +const SHORT_LIVED_RECLAIM_PAGES: u64 = 4096; + /// The sole map owner for one registered global-database publication. /// /// It can issue independently counted client leases, but cannot be cloned or @@ -213,6 +217,18 @@ impl RegisteredGlobalDbOwnerV1 { } RegisteredSchemaAttachmentV1::SessionsRefused(refused) => Some(refused), }; + // A short-lived client has no later daemon maintenance turn. Reclaim + // pages released by earlier writes after schema convergence, without + // introducing the same foreground work into daemon admission. Refused + // authorities remain byte-preserving until their typed reset. + if refused_authority.is_none() { + // Defer only optional SQLite maintenance contention. A revoked + // write authority or closed writer must still fail admission. + registered + .database + .reclaim_short_lived_pages(SHORT_LIVED_RECLAIM_PAGES) + .await?; + } drop(registered); Ok(Self { database, diff --git a/crates/tracedecay-global-db/src/registered/workflow_schema_tests.rs b/crates/tracedecay-global-db/src/registered/workflow_schema_tests.rs index 3dd339b74f0..1618df734ae 100644 --- a/crates/tracedecay-global-db/src/registered/workflow_schema_tests.rs +++ b/crates/tracedecay-global-db/src/registered/workflow_schema_tests.rs @@ -129,6 +129,65 @@ async fn schema_snapshot( snapshot } +#[asupersync::test] +async fn short_lived_admission_reclaims_freed_pages_without_daemon_maintenance() { + crate::register_registered_schema_installer(); + let directory = TempDir::new().unwrap(); + let path = directory.path().join("project/sessions.db"); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + let scope = || TestDatabaseRuntimeScope::ProjectSessions { + project_id: ProjectId::new("project.short-lived-reclaim").unwrap(), + }; + + drop( + open_registered_test_database_fixture(&path, scope()) + .await + .unwrap(), + ); + let connection = rusqlite::Connection::open(&path).unwrap(); + let auto_vacuum: i64 = connection + .query_row("PRAGMA auto_vacuum", [], |row| row.get(0)) + .unwrap(); + assert_eq!( + auto_vacuum, 2, + "new stores must support incremental reclamation" + ); + connection + .execute_batch( + "CREATE TABLE ephemeral_reclaim(payload BLOB); + INSERT INTO ephemeral_reclaim VALUES (zeroblob(8388608)); + DROP TABLE ephemeral_reclaim;", + ) + .unwrap(); + let before_pages: i64 = connection + .query_row("PRAGMA page_count", [], |row| row.get(0)) + .unwrap(); + let before_free: i64 = connection + .query_row("PRAGMA freelist_count", [], |row| row.get(0)) + .unwrap(); + assert!( + before_free > 0, + "the fixture must contain reclaimable pages" + ); + drop(connection); + + // Reopen through the actual short-lived owner: there is no background + // compactor after this call, so admission must make physical progress. + let (lease, owner) = open_registered_test_database_fixture(&path, scope()) + .await + .unwrap(); + let (_, after_pages, after_free) = lease.storage_page_counts().await.unwrap(); + assert!( + after_pages < before_pages as u64, + "short-lived admission did not reduce the physical page count: {before_pages} -> {after_pages}" + ); + assert!( + after_free < before_free as u64, + "short-lived admission left the original free-page backlog: {before_free} -> {after_free}" + ); + drop((lease, owner)); +} + #[asupersync::test] async fn partial_workflow_store_requires_reset_without_mutation() { assert_workflow_schema_reset_without_mutation( @@ -857,6 +916,11 @@ mod published_artifact_ceiling { INSERT INTO workflow_artifact_payloads SELECT 'sha256:large-' || i, 1048576, randomblob(1048576) FROM n;", ); + store.run( + "CREATE TABLE freed_pages(payload BLOB); + INSERT INTO freed_pages VALUES (zeroblob(2097152)); + DROP TABLE freed_pages;", + ); let before = store.state(); let wal_path = format!("{}-wal", store.path.display()); let wal_bytes = || fs::metadata(&wal_path).map_or(0, |metadata| metadata.len()); @@ -874,6 +938,11 @@ mod published_artifact_ceiling { ); assert!(owner.reset_required().is_none()); let lease = owner.issue_lease().unwrap(); + let (_, _, free_pages) = lease.storage_page_counts().await.unwrap(); + assert!( + free_pages > 0, + "admission reclaimed pages before convergence" + ); let issuer = owner.weak_lease_issuer(); assert_workflow_recovering(&lease).await; assert_workflow_recovering(&issuer.issue_lease().unwrap()).await; diff --git a/crates/tracedecay-global-db/src/schema_stages.rs b/crates/tracedecay-global-db/src/schema_stages.rs index 63e7e42226c..f7f1c8a71ef 100644 --- a/crates/tracedecay-global-db/src/schema_stages.rs +++ b/crates/tracedecay-global-db/src/schema_stages.rs @@ -55,8 +55,6 @@ use tracedecay_sessions::runtime::git_correlation::{ }; use tracedecay_store::StoreShardScopeV1; -const CANONICAL_BODY_RECLAIM_PAGES: u64 = 4096; - const REGISTRY_SCHEMA: &str = " CREATE TABLE IF NOT EXISTS projects ( path TEXT PRIMARY KEY, @@ -1453,9 +1451,6 @@ pub(crate) async fn ensure_attached_registered_schema( ) } })?; - database - .run_incremental_vacuum(CANONICAL_BODY_RECLAIM_PAGES) - .await?; Ok(match refused_authority { Some(refused) => RegisteredSchemaAttachmentV1::SessionsRefused(refused), None => RegisteredSchemaAttachmentV1::Admitted(RegisteredSchemaConvergence { diff --git a/crates/tracedecay-global-db/src/schema_stages/workflow_artifact_ceiling_upgrade.rs b/crates/tracedecay-global-db/src/schema_stages/workflow_artifact_ceiling_upgrade.rs index 135b3d078ad..89e33b37ab7 100644 --- a/crates/tracedecay-global-db/src/schema_stages/workflow_artifact_ceiling_upgrade.rs +++ b/crates/tracedecay-global-db/src/schema_stages/workflow_artifact_ceiling_upgrade.rs @@ -34,13 +34,14 @@ use tracedecay_rusqlite_runtime::workflow::{ }; use super::{ - CANONICAL_BODY_RECLAIM_PAGES, CancellableSchemaTransaction, WorkflowSchemaAdmission, - inspect_workflow_schema_for_admission, read_recorded_workflow_identity, - schema_install_checkpoint, settle_schema_transaction, workflow_schema_refusal, + CancellableSchemaTransaction, WorkflowSchemaAdmission, inspect_workflow_schema_for_admission, + read_recorded_workflow_identity, schema_install_checkpoint, settle_schema_transaction, + workflow_schema_refusal, }; use crate::registered::RefusedAuthorityV1; use crate::{global_db_operation_error, global_db_operation_message}; +const WORKFLOW_ARTIFACT_RECLAIM_PAGES: u64 = 4096; const ARTIFACT_TABLE: &str = "workflow_artifact_payloads"; const PUBLISHED_ARTIFACT_TABLE_SQL: &str = "CREATE TABLE workflow_artifact_payloads ( @@ -292,7 +293,7 @@ pub(crate) async fn reclaim_freed_artifact_pages( } remaining = free_pages; database - .run_incremental_vacuum(CANONICAL_BODY_RECLAIM_PAGES) + .run_incremental_vacuum(WORKFLOW_ARTIFACT_RECLAIM_PAGES) .await?; } } diff --git a/crates/tracedecay-lcm/src/summary_convergence.rs b/crates/tracedecay-lcm/src/summary_convergence.rs index acf0cd1e233..3c603485bd5 100644 --- a/crates/tracedecay-lcm/src/summary_convergence.rs +++ b/crates/tracedecay-lcm/src/summary_convergence.rs @@ -646,6 +646,10 @@ pub async fn record_invalidation_page_yield( Ok(()) } +/// A failed attempt is settled against the raw revision it actually read. +/// Older dirty raw work may remain after invalidation fails; it must not +/// discard a retry deadline or resurrect a terminal refusal. A truly new +/// raw revision increments the generation, so its stale outcome loses the CAS. pub async fn record_outcome( conn: &(impl Executor + ?Sized), candidate: &LcmSummaryConvergenceCandidate, @@ -659,30 +663,38 @@ pub async fn record_outcome( "UPDATE lcm_summary_convergence_queue SET attempted_raw_store_id = ?3, state = CASE - WHEN newest_raw_store_id > ?3 OR EXISTS ( - SELECT 1 FROM lcm_summary_convergence_dirty_raw AS dirty - WHERE dirty.provider = ?1 AND dirty.session_id = ?2 + WHEN newest_raw_store_id > ?3 OR ( + ?4 IN ('pending', 'current') AND EXISTS ( + SELECT 1 FROM lcm_summary_convergence_dirty_raw AS dirty + WHERE dirty.provider = ?1 AND dirty.session_id = ?2 + ) ) THEN 'pending' ELSE ?4 END, failure_code = CASE - WHEN newest_raw_store_id > ?3 OR EXISTS ( - SELECT 1 FROM lcm_summary_convergence_dirty_raw AS dirty - WHERE dirty.provider = ?1 AND dirty.session_id = ?2 + WHEN newest_raw_store_id > ?3 OR ( + ?4 IN ('pending', 'current') AND EXISTS ( + SELECT 1 FROM lcm_summary_convergence_dirty_raw AS dirty + WHERE dirty.provider = ?1 AND dirty.session_id = ?2 + ) ) THEN NULL ELSE ?5 END, failure_count = CASE - WHEN newest_raw_store_id > ?3 OR EXISTS ( - SELECT 1 FROM lcm_summary_convergence_dirty_raw AS dirty - WHERE dirty.provider = ?1 AND dirty.session_id = ?2 + WHEN newest_raw_store_id > ?3 OR ( + ?4 IN ('pending', 'current') AND EXISTS ( + SELECT 1 FROM lcm_summary_convergence_dirty_raw AS dirty + WHERE dirty.provider = ?1 AND dirty.session_id = ?2 + ) ) THEN 0 ELSE ?6 END, next_attempt_at_ms = CASE - WHEN newest_raw_store_id > ?3 OR EXISTS ( - SELECT 1 FROM lcm_summary_convergence_dirty_raw AS dirty - WHERE dirty.provider = ?1 AND dirty.session_id = ?2 + WHEN newest_raw_store_id > ?3 OR ( + ?4 IN ('pending', 'current') AND EXISTS ( + SELECT 1 FROM lcm_summary_convergence_dirty_raw AS dirty + WHERE dirty.provider = ?1 AND dirty.session_id = ?2 + ) ) THEN 0 ELSE ?7 END, diff --git a/crates/tracedecay-lcm/src/summary_convergence_tests.rs b/crates/tracedecay-lcm/src/summary_convergence_tests.rs index d0c6bc90a48..c7c119a4d28 100644 --- a/crates/tracedecay-lcm/src/summary_convergence_tests.rs +++ b/crates/tracedecay-lcm/src/summary_convergence_tests.rs @@ -601,6 +601,215 @@ async fn disjoint_raw_revisions_drain_as_distinct_restart_safe_work_items() { ); } +async fn dirty_summary_queue_store(temp: &tempfile::TempDir) -> TestConnection { + let conn = TestConnection::open(&temp.path().join("sessions.db")); + conn.execute_batch( + "CREATE TABLE sessions ( + provider TEXT NOT NULL, + session_id TEXT NOT NULL, + project_key TEXT NOT NULL, + project_path TEXT NOT NULL, + PRIMARY KEY(provider, session_id) + ); + INSERT INTO sessions(provider, session_id, project_key, project_path) + VALUES ('cursor', 'dirty-session', 'project.dirty', '/dirty');", + ) + .await + .unwrap(); + schema::ensure_lcm_schema(&conn).await.unwrap(); + conn.execute_batch( + "INSERT INTO lcm_raw_messages ( + provider, message_id, session_id, role, ordinal, content, + content_hash, storage_kind, metadata_json + ) VALUES ( + 'cursor', 'message-1', 'dirty-session', 'assistant', 1, 'body', + 'old-hash', 'inline', '{}' + ); + UPDATE lcm_raw_messages SET content_hash = 'revised-hash' + WHERE provider = 'cursor' AND message_id = 'message-1';", + ) + .await + .unwrap(); + conn +} + +#[asupersync::test] +async fn dirty_raw_failure_backoff_survives_restart_and_new_revision_recovers() { + let temp = tempfile::tempdir().unwrap(); + let conn = dirty_summary_queue_store(&temp).await; + let first = summary_convergence::next_candidate(&conn, 0) + .await + .unwrap() + .expect("changed raw row must enqueue invalidation"); + assert_eq!(first.stale_from_store_id, Some(1)); + + assert!( + summary_convergence::record_outcome( + &conn, + &first, + summary_convergence::LcmSummaryConvergenceQueueState::Retryable, + Some("storage_unavailable"), + 1, + 100_000, + ) + .await + .unwrap() + ); + assert!( + summary_convergence::next_candidate(&conn, 99_999) + .await + .unwrap() + .is_none(), + "undrained dirty input must not erase a retry deadline" + ); + + drop(conn); + let conn = TestConnection::open(&temp.path().join("sessions.db")); + assert!( + summary_convergence::next_candidate(&conn, 99_999) + .await + .unwrap() + .is_none(), + "backoff must remain durable across a daemon/store reopening" + ); + let retry = summary_convergence::next_candidate(&conn, 100_000) + .await + .unwrap() + .expect("the failed input becomes due at its persisted deadline"); + assert_eq!(retry.failure_count, 1); + assert_eq!(retry.stale_from_store_id, Some(1)); + assert!( + summary_convergence::record_outcome( + &conn, + &retry, + summary_convergence::LcmSummaryConvergenceQueueState::Retryable, + Some("storage_unavailable"), + 2, + 300_000, + ) + .await + .unwrap() + ); + assert!( + summary_convergence::next_candidate(&conn, 299_999) + .await + .unwrap() + .is_none(), + "a second identical failure must not hot-loop" + ); + + conn.execute( + "UPDATE lcm_raw_messages SET content_hash = 'new-revision' + WHERE provider = 'cursor' AND message_id = 'message-1'", + (), + ) + .await + .unwrap(); + let revised = summary_convergence::next_candidate(&conn, 0) + .await + .unwrap() + .expect("a new raw revision must wake the previously failed session"); + assert!(revised.raw_revision_generation > retry.raw_revision_generation); + assert_eq!(revised.failure_count, 0); + assert_eq!(revised.stale_from_store_id, Some(1)); + assert!( + !summary_convergence::record_outcome( + &conn, + &retry, + summary_convergence::LcmSummaryConvergenceQueueState::Permanent, + Some("storage_unavailable"), + 3, + 0, + ) + .await + .unwrap(), + "superseded failures must not overwrite the newly queued work" + ); +} + +#[asupersync::test] +async fn dirty_raw_terminal_outcomes_park_until_an_actual_raw_change() { + let temp = tempfile::tempdir().unwrap(); + let conn = dirty_summary_queue_store(&temp).await; + + for (state, expected_state) in [ + ( + summary_convergence::LcmSummaryConvergenceQueueState::Unavailable, + "unavailable", + ), + ( + summary_convergence::LcmSummaryConvergenceQueueState::Permanent, + "permanent", + ), + ] { + let candidate = summary_convergence::next_candidate(&conn, 0) + .await + .unwrap() + .expect("new or changed raw input is due"); + assert!( + summary_convergence::record_outcome( + &conn, + &candidate, + state, + Some("unavailable"), + 1, + 0, + ) + .await + .unwrap() + ); + assert_eq!( + fetch_i64( + &conn, + "SELECT COUNT(*) FROM lcm_summary_convergence_queue + WHERE state = ?1 AND failure_count = 1", + params![expected_state], + ) + .await, + 1 + ); + assert!( + summary_convergence::next_candidate(&conn, i64::MAX) + .await + .unwrap() + .is_none(), + "old dirty rows must not reoffer a parked session" + ); + conn.execute( + "UPDATE lcm_raw_messages SET content_hash = content_hash || 'x' + WHERE provider = 'cursor' AND message_id = 'message-1'", + (), + ) + .await + .unwrap(); + } + + let candidate = summary_convergence::next_candidate(&conn, 0) + .await + .unwrap() + .expect("a new raw revision must release a terminally parked session"); + assert_eq!(candidate.failure_count, 0); + assert!( + summary_convergence::record_outcome( + &conn, + &candidate, + summary_convergence::LcmSummaryConvergenceQueueState::Current, + None, + 0, + 0, + ) + .await + .unwrap() + ); + assert!( + summary_convergence::next_candidate(&conn, 0) + .await + .unwrap() + .is_some(), + "successful processing cannot claim complete with dirty invalidation owed" + ); +} + async fn parked_queue_store(temp: &tempfile::TempDir, sessions: &[&str]) -> TestConnection { let conn = TestConnection::open(&temp.path().join("sessions.db")); conn.execute_batch( diff --git a/crates/tracedecay-runtime-core/src/db/connection/retained_maintenance.rs b/crates/tracedecay-runtime-core/src/db/connection/retained_maintenance.rs index 371b82ee8a5..72080014718 100644 --- a/crates/tracedecay-runtime-core/src/db/connection/retained_maintenance.rs +++ b/crates/tracedecay-runtime-core/src/db/connection/retained_maintenance.rs @@ -1,3 +1,5 @@ +use crate::shard_runtime::registry::StoreRuntimeRegistryFailure; + use super::{ Arc, CheckpointBlockers, CheckpointOutcome, CheckpointRequest, DATABASE_HEALTH_GATE, Database, DatabaseAuthorityRole, DatabaseHealth, DatabaseStorageTelemetryHandle, Result, TraceDecayError, @@ -241,17 +243,47 @@ impl Database { /// Runs bounded incremental vacuum through the canonical writer lane. pub async fn run_incremental_vacuum(&self, pages: u64) -> Result<()> { + self.incremental_vacuum_attempt(pages) + .await? + .map_err(compaction_error) + } + + /// Short-lived store admission may defer optional page reclamation, but + /// cannot serve a store whose writer or write authority has failed. + pub async fn reclaim_short_lived_pages(&self, pages: u64) -> Result<()> { + match self.incremental_vacuum_attempt(pages).await? { + Ok(()) => Ok(()), + Err(StoreRuntimeRegistryFailure::CompactionDeferred { message }) => { + tracing::warn!( + reason = %message, + pages, + "short-lived registered-store reclamation deferred" + ); + Ok(()) + } + Err(error) => Err(compaction_error(error)), + } + } + + async fn incremental_vacuum_attempt( + &self, + pages: u64, + ) -> Result> { let cx = crate::native_runtime::cx().ok_or_else(|| TraceDecayError::Config { message: "database compaction requires an Asupersync context".to_owned(), })?; let authority = self.write_authority()?; - self.client + Ok(self + .client .runtime() .run_bounded_incremental_compaction(&cx, pages, authority) - .await - .map_err(|error| TraceDecayError::Database { - message: format!("failed to compact SQLite store: {error:?}"), - operation: "run bounded SQLite-store compaction".to_owned(), - }) + .await) + } +} + +fn compaction_error(error: StoreRuntimeRegistryFailure) -> TraceDecayError { + TraceDecayError::Database { + message: format!("failed to compact SQLite store: {error:?}"), + operation: "run bounded SQLite-store compaction".to_owned(), } } diff --git a/crates/tracedecay-runtime-core/src/shard_runtime/registry.rs b/crates/tracedecay-runtime-core/src/shard_runtime/registry.rs index da1e1615004..ef754fff9b7 100644 --- a/crates/tracedecay-runtime-core/src/shard_runtime/registry.rs +++ b/crates/tracedecay-runtime-core/src/shard_runtime/registry.rs @@ -1292,6 +1292,10 @@ pub enum StoreRuntimeRegistryFailure { operation: &'static str, message: String, }, + // Recoverable page reclamation is not an admission or writer-authority failure. + CompactionDeferred { + message: String, + }, /// Every reader was leased for the whole admission wait. Nothing failed; /// the same read can succeed once a reader returns. ReaderSaturated, diff --git a/crates/tracedecay-runtime-core/src/shard_runtime/registry/ports.rs b/crates/tracedecay-runtime-core/src/shard_runtime/registry/ports.rs index 1a951d8b2c9..6a0ef2fe41a 100644 --- a/crates/tracedecay-runtime-core/src/shard_runtime/registry/ports.rs +++ b/crates/tracedecay-runtime-core/src/shard_runtime/registry/ports.rs @@ -618,9 +618,14 @@ impl PhysicalRuntimeAttachment for RepositoryRuntimePhysicalAttachment { self, cx, max_pages, authority, ) .await - .map_err(|error| StoreRuntimeRegistryFailure::PhysicalRuntimeFailed { - operation: "run bounded repository compaction", - message: error.to_string(), + .map_err(|error| match error { + RepositoryDispatchError::CompactionDeferred(message) => { + StoreRuntimeRegistryFailure::CompactionDeferred { message } + } + error => StoreRuntimeRegistryFailure::PhysicalRuntimeFailed { + operation: "run bounded repository compaction", + message: error.to_string(), + }, }) }) } diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/attachment.rs b/crates/tracedecay-rusqlite-runtime/src/repository/attachment.rs index 2ad9b08b1d7..4c47642b365 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/attachment.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/attachment.rs @@ -631,7 +631,7 @@ impl RepositoryRuntimePhysicalAttachment { writer .bounded_incremental_vacuum(cx, max_pages, authority) .await - .map_err(|error| RepositoryDispatchError::Writer(error.to_string())) + .map_err(compaction_dispatch_error) } pub async fn run_checkpoint( @@ -835,12 +835,24 @@ impl Drop for RepositoryRuntimePhysicalAttachment { } } +// Only a failed SQLite vacuum or saturated maintenance channel is deferrable. +// A lost writer, revoked authority or failed reply is fatal to admission. +fn compaction_dispatch_error(error: WriterActorError) -> RepositoryDispatchError { + match error { + WriterActorError::IncrementalVacuumBusy | WriterActorError::IncrementalVacuumNoSpace => { + RepositoryDispatchError::CompactionDeferred(error.to_string()) + } + other => RepositoryDispatchError::Writer(other.to_string()), + } +} + #[derive(Debug)] pub enum RepositoryDispatchError { Closed, Reader(ReaderAcquireError), ReaderWorker(String), Writer(String), + CompactionDeferred(String), WriterRuntime(StorageRuntimeErrorV1), } @@ -851,6 +863,9 @@ impl fmt::Display for RepositoryDispatchError { Self::Reader(error) => write!(formatter, "repository read failed: {error}"), Self::ReaderWorker(error) => write!(formatter, "repository snapshot failed: {error}"), Self::Writer(error) => write!(formatter, "repository write failed: {error}"), + Self::CompactionDeferred(reason) => { + write!(formatter, "SQLite compaction deferred: {reason}") + } Self::WriterRuntime(error) => write!(formatter, "repository write failed: {error}"), } } @@ -860,7 +875,10 @@ impl Error for RepositoryDispatchError { fn source(&self) -> Option<&(dyn Error + 'static)> { match self { Self::Reader(error) => Some(error), - Self::Closed | Self::ReaderWorker(_) | Self::Writer(_) => None, + Self::Closed + | Self::ReaderWorker(_) + | Self::Writer(_) + | Self::CompactionDeferred(_) => None, Self::WriterRuntime(error) => Some(error), } } @@ -967,6 +985,39 @@ mod tests { use super::*; + #[test] + fn vacuum_dispatch_does_not_hide_writer_or_authority_failure() { + let revoked = compaction_dispatch_error(WriterActorError::AuthorityDenied { + stage: crate::RuntimeWriteAuthorityStage::BeforeAdmission, + }); + assert!( + matches!(revoked, RepositoryDispatchError::Writer(_)), + "lost write authority must deny admission" + ); + let closed = compaction_dispatch_error(WriterActorError::ReplyDropped); + assert!( + matches!(closed, RepositoryDispatchError::Writer(_)), + "a stopped writer must deny admission" + ); + let blocked = compaction_dispatch_error(WriterActorError::IncrementalVacuumBusy); + assert!( + matches!(blocked, RepositoryDispatchError::CompactionDeferred(_)), + "an occupied maintenance queue or busy SQLite transaction may defer" + ); + let disk_full = compaction_dispatch_error(WriterActorError::IncrementalVacuumNoSpace); + assert!( + matches!(disk_full, RepositoryDispatchError::CompactionDeferred(_)), + "an optional vacuum must not turn disk pressure into a store-open failure" + ); + let sqlite = compaction_dispatch_error(WriterActorError::IncrementalVacuumFailed( + "database is read-only".to_owned(), + )); + assert!( + matches!(sqlite, RepositoryDispatchError::Writer(_)), + "database and worker failures must deny admission" + ); + } + fn native_runtime() -> RuntimeHandle { crate::native_runtime::handle().expect("native runtime") } diff --git a/crates/tracedecay-rusqlite-runtime/src/writer.rs b/crates/tracedecay-rusqlite-runtime/src/writer.rs index 87a5bff3c14..b586a6e4353 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer.rs @@ -371,6 +371,8 @@ pub enum WriterActorError { WaitCancelled, StorageFailure(StorageRuntimeErrorV1), IncrementalVacuumFailed(String), + IncrementalVacuumBusy, + IncrementalVacuumNoSpace, InvalidWorkerOutcome(StorageRuntimeContractErrorV1), TaskPanicked, } @@ -391,6 +393,12 @@ impl fmt::Display for WriterActorError { Self::IncrementalVacuumFailed(message) => { write!(f, "SQLite incremental vacuum failed: {message}") } + Self::IncrementalVacuumBusy => { + f.write_str("SQLite vacuum deferred by writer or database contention") + } + Self::IncrementalVacuumNoSpace => { + f.write_str("SQLite vacuum deferred by insufficient disk space") + } Self::InvalidWorkerOutcome(error) => { write!(f, "SQLite writer returned an invalid outcome: {error}") } @@ -776,9 +784,7 @@ impl PersistentWriter { stage: RuntimeWriteAuthorityStage::BeforeAdmission, })?; if self.state() != WriterState::Ready { - return Err(WriterActorError::IncrementalVacuumFailed( - "writer is unavailable".to_owned(), - )); + return Err(WriterActorError::ReplyDropped); } let (reply, mut response) = oneshot::channel(); let command = IncrementalVacuumCommand::new(max_pages, authority, reply); @@ -796,11 +802,9 @@ impl PersistentWriter { } }; send_result.map_err(|error| match error { - mpsc::SendError::Full(_) => { - WriterActorError::IncrementalVacuumFailed("command channel is busy".to_owned()) - } + mpsc::SendError::Full(_) => WriterActorError::IncrementalVacuumBusy, mpsc::SendError::Disconnected(_) | mpsc::SendError::Cancelled(_) => { - WriterActorError::IncrementalVacuumFailed("writer is unavailable".to_owned()) + WriterActorError::ReplyDropped } })?; response.recv(cx).await.map_err(writer_reply_error)? diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index b18573ea65f..d134e686c0f 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -1151,6 +1151,19 @@ fn queue_wait_micros(enqueued_at: impl IntoIterator, dequeued_at }) } +fn incremental_vacuum_error(error: rusqlite::Error) -> WriterActorError { + match error.sqlite_error_code() { + Some(rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked) => { + WriterActorError::IncrementalVacuumBusy + } + // A large optional vacuum may need more WAL space than the next + // ordinary small write. Disk-full on that vacuum is not evidence of + // lost writer authority or an unusable registered store. + Some(rusqlite::ErrorCode::DiskFull) => WriterActorError::IncrementalVacuumNoSpace, + _ => WriterActorError::IncrementalVacuumFailed(error.to_string()), + } +} + fn run_incremental_vacuum( connection: &mut rusqlite::Connection, command: IncrementalVacuumCommand, @@ -1172,9 +1185,7 @@ fn run_incremental_vacuum( let transaction = match match_result { Ok(transaction) => transaction, Err(error) => { - command.settle(Err(WriterActorError::IncrementalVacuumFailed( - error.to_string(), - ))); + command.settle(Err(incremental_vacuum_error(error))); return; } }; @@ -1191,9 +1202,7 @@ fn run_incremental_vacuum( Ok(()) }); if let Err(error) = reclaimed { - command.settle(Err(WriterActorError::IncrementalVacuumFailed( - error.to_string(), - ))); + command.settle(Err(incremental_vacuum_error(error))); return; } if command @@ -1201,20 +1210,18 @@ fn run_incremental_vacuum( .verify(RuntimeWriteAuthorityStage::BeforeCommit) .is_err() { - let result = match transaction.rollback() { - Ok(()) => Err(WriterActorError::AuthorityDenied { - stage: RuntimeWriteAuthorityStage::BeforeCommit, - }), - Err(error) => Err(WriterActorError::IncrementalVacuumFailed(format!( - "rollback after authority loss: {error}" - ))), - }; - command.settle(result); + if let Err(error) = transaction.rollback() { + tracing::error!( + error = %error, + "SQLite vacuum rollback failed after write authority was revoked" + ); + } + command.settle(Err(WriterActorError::AuthorityDenied { + stage: RuntimeWriteAuthorityStage::BeforeCommit, + })); return; } - let result = transaction - .commit() - .map_err(|error| WriterActorError::IncrementalVacuumFailed(error.to_string())); + let result = transaction.commit().map_err(incremental_vacuum_error); command.settle(result); } @@ -1327,7 +1334,7 @@ mod auxiliary_scheduling_tests { use super::{ AcceptedRequest, AuxiliaryWork, BatchCoalescingWindow, IncrementalVacuumCommand, PendingBatchDwell, WriterActorError, build_batches, checkpoint_config, dwell_for_batch, - enqueue, run_incremental_vacuum, select_auxiliary_work, + enqueue, reject_incremental_vacuum, run_incremental_vacuum, select_auxiliary_work, }; struct BatchProbe { @@ -2007,6 +2014,71 @@ mod auxiliary_scheduling_tests { ); } + #[test] + fn incremental_vacuum_defers_contention_but_reports_unusable_writers() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("vacuum.db"); + let mut connection = rusqlite::Connection::open(&path).unwrap(); + connection.busy_timeout(Duration::ZERO).unwrap(); + let competing_writer = rusqlite::Connection::open(path).unwrap(); + competing_writer.execute_batch("BEGIN IMMEDIATE").unwrap(); + let command = || { + let (reply, response) = oneshot::channel(); + ( + IncrementalVacuumCommand::new( + 1, + Arc::new(UnrestrictedRuntimeWriteAuthority), + reply, + ), + response, + ) + }; + + let (vacuum, mut result) = command(); + run_incremental_vacuum(&mut connection, vacuum); + assert!(matches!( + result.try_recv().unwrap(), + Err(WriterActorError::IncrementalVacuumBusy) + )); + competing_writer.execute_batch("ROLLBACK").unwrap(); + + connection.pragma_update(None, "query_only", true).unwrap(); + let (vacuum, mut result) = command(); + run_incremental_vacuum(&mut connection, vacuum); + assert!(matches!( + result.try_recv().unwrap(), + Err(WriterActorError::IncrementalVacuumFailed(_)) + )); + + let (vacuum, mut result) = command(); + reject_incremental_vacuum(vacuum); + assert!(matches!( + result.try_recv().unwrap(), + Err(WriterActorError::IncrementalVacuumFailed(_)) + )); + } + + #[test] + fn full_sqlite_disk_quota_defers_only_optional_vacuum() { + let connection = rusqlite::Connection::open_in_memory().unwrap(); + connection.pragma_update(None, "page_size", 512).unwrap(); + connection.pragma_update(None, "max_page_count", 8).unwrap(); + connection + .execute_batch("CREATE TABLE quota_probe(payload BLOB)") + .unwrap(); + // Exercise SQLite's actual SQLITE_FULL result instead of inventing + // an error string. An optional vacuum must not block an otherwise + // admitted store solely because its write would exceed the quota. + let full = connection + .execute("INSERT INTO quota_probe VALUES (zeroblob(65536))", []) + .unwrap_err(); + assert_eq!(full.sqlite_error_code(), Some(rusqlite::ErrorCode::DiskFull)); + assert!(matches!( + incremental_vacuum_error(full), + WriterActorError::IncrementalVacuumNoSpace + )); + } + #[test] fn incremental_vacuum_samples_worker_authority_stages() { let mut connection = rusqlite::Connection::open_in_memory().unwrap(); diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index ee3ab998e87..c4c1632528d 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -17,6 +17,7 @@ use tracedecay_global_db::RegisteredGlobalDbLeaseV1; use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; #[cfg(unix)] use tracedecay_global_db::tests::harness::RegisteredGlobalDbTestRuntime; +use tracedecay_lcm::summary_convergence::{next_candidate, next_retry_at_ms}; use tracedecay_lcm::{LcmRelationProjectionStatus, LcmSourceRef, LcmSummarizerMode}; use tracedecay_runtime_core::db::engine::params; #[cfg(unix)] @@ -2125,6 +2126,94 @@ async fn permanent_session_failure_does_not_starve_later_sessions_after_restart( assert!(third.sessions.is_empty(), "permanent work was reselected"); } +#[tokio::test] +async fn failed_revision_invalidation_defers_the_same_dirty_input_until_retry() { + let harness = RegisteredGlobalDbHarness::open("lcm-invalidation-retry").await; + let db = &harness.registered; + let session_id = "failed-invalidation"; + let root = db.db_path().parent().unwrap(); + let mut raw = message(session_id, 1); + seed_session_messages(db, &session("cursor", session_id), &[raw.clone()]).await; + raw.text = "revised raw input".to_owned(); + db.lcm_ingest_raw_message(root, &raw).await.unwrap(); + + // Fail the final deletion in the real invalidation transaction. The + // subsequent queue settlement must still be able to commit its backoff. + let transaction = db.begin_write_transaction().await.unwrap(); + transaction + .execute_batch( + "CREATE TRIGGER refuse_dirty_completion BEFORE DELETE ON lcm_summary_convergence_dirty_raw + BEGIN SELECT RAISE(ABORT, 'dirty completion unavailable'); END;", + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let before_run_ms = i64::try_from( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis(), + ) + .unwrap(); + let failed = crate::lcm_summary_convergence::run_summary_convergence_page(db.clone(), 1) + .await + .unwrap(); + assert_eq!(failed.sessions.len(), 1); + assert_eq!( + failed.sessions[0].disposition, + crate::lcm_summary_convergence::LcmSummaryConvergenceDisposition::Retryable { + reason: "storage_unavailable".to_owned(), + }, + ); + assert!(failed.next_retry_delay.is_some()); + let snapshot = db.read_snapshot().await.unwrap(); + let deadline = next_retry_at_ms(&snapshot).await.unwrap().unwrap(); + assert!( + deadline > before_run_ms, + "persist a future deadline for this attempt" + ); + assert!( + next_candidate(&snapshot, deadline - 1) + .await + .unwrap() + .is_none() + ); + let due = next_candidate(&snapshot, deadline).await.unwrap().unwrap(); + assert_eq!(due.session_id, session_id); + assert_eq!(due.failure_count, 1); + assert_eq!(due.stale_from_store_id, Some(due.newest_raw_store_id)); + snapshot.commit().await.unwrap(); + + let transaction = db.begin_write_transaction().await.unwrap(); + transaction + .execute_batch("DROP TRIGGER refuse_dirty_completion") + .await + .unwrap(); + transaction.commit().await.unwrap(); + raw.text = "new revision wakes the parked session".to_owned(); + db.lcm_ingest_raw_message(root, &raw).await.unwrap(); + let recovered = crate::lcm_summary_convergence::run_summary_convergence_page(db.clone(), 1) + .await + .unwrap(); + assert_eq!(recovered.sessions.len(), 1); + let snapshot = db.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT raw_revision_generation, stale_from_store_id, + (SELECT COUNT(*) FROM lcm_summary_convergence_dirty_raw + WHERE provider = 'cursor' AND session_id = ?1) + FROM lcm_summary_convergence_queue WHERE provider = 'cursor' AND session_id = ?1", + params![session_id], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + assert!(row.get::(0).unwrap() > due.raw_revision_generation); + assert_eq!(row.get::>(1).unwrap(), None); + assert_eq!(row.get::(2).unwrap(), 0); +} + #[tokio::test] async fn partial_revision_invalidation_hides_replay_and_yields_to_a_due_peer() { let harness = RegisteredGlobalDbHarness::open("lcm-partial-invalidation-fairness").await;