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/tests/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/tests/mod.rs index 6d8e8864d1c..3b0d532f484 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/tests/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/tests/mod.rs @@ -24,6 +24,7 @@ use tracedecay_store::{ use super::*; use crate::{ StorageOperationExecutor, + admission::QueueItem, checkpoint::{ CheckpointBlockers, CheckpointDecision, CheckpointInterruption, CheckpointMode, CheckpointOutcome, CheckpointPressure, CheckpointReport, CheckpointResult, WalPressure, @@ -315,6 +316,58 @@ impl RuntimeRequestProbeV1 for Probe { } } +#[test] +fn faulted_writer_retires_without_waiting_for_registered_client_senders() { + let database = TestDatabase::new(); + let request = Arc::new(request(metadata( + "operation.faulted-retirement", + "key.faulted-retirement", + 'f', + ))); + let writer = start(&database, &request, Arc::new(AtomicU64::new(0))); + let permit = writer + .admission + .reserve(&request.envelope().metadata) + .expect("faulted retirement fixture fits admission"); + let (reply, mut response) = oneshot::channel(); + let accepted = AcceptedRequest::new( + Arc::clone(&request), + Arc::new(Probe::new(&request, None)), + Arc::new(UnrestrictedRuntimeWriteAuthority), + reply, + permit, + ); + writer.telemetry.admitted(accepted.admission_bytes()); + let sender = writer.sender.lock().unwrap().as_ref().unwrap().clone(); + writer + .state + .store(WriterState::Faulted as u8, Ordering::Release); + assert!(sender.try_send(accepted).is_ok()); + + // The sender is intentionally retained: the old fault path settled the + // request, then parked forever waiting for every client to disconnect. + let deadline = std::time::Instant::now() + Duration::from_secs(3); + while !writer.worker_exited() && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(5)); + } + assert!( + writer.worker_exited(), + "faulted writer retained its SQLite connection while clients remained registered" + ); + assert_eq!( + response + .try_recv() + .expect("accepted request receives a fault reply"), + Ok(RuntimeSubmitOutcomeV1::Unavailable { + reason: UnavailableReasonV1::Faulted, + }) + ); + assert_eq!(writer.telemetry_snapshot().queue.queued_operations, 0); + assert_eq!(writer.telemetry_snapshot().queue.queued_bytes, 0); + drop(sender); + writer.shutdown_and_join().unwrap(); +} + #[test] fn writer_start_leaves_schema_unchanged_before_the_first_request() { let database = TestDatabase::new(); diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index b18573ea65f..ee20c0cec58 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -65,8 +65,9 @@ use ingress::{ select_auxiliary_work, wait_for_work, }; use rejection::{ - cancel_waiting, reject_all, reject_all_exact_sql, reject_all_incremental_vacuum, - reject_incremental_vacuum, reject_unauthorized, + cancel_waiting, reject_all_exact_sql, reject_all_incremental_vacuum, + reject_faulted_execution_batches, reject_incremental_vacuum, reject_retired_write_ingress, + reject_unauthorized, }; /// When the blocked writer retries its drain: the fence's short interval, or @@ -569,7 +570,33 @@ impl Worker { } }; if lost { - reject_all(&mut queue, &self.telemetry, UnavailableReasonV1::FileLost); + // A sender may have passed the Ready check before the loss + // probe flipped state. Close ingress before the final drain + // so accepted work cannot disappear when this worker retires. + self.exact_sql_receiver.close(); + self.incremental_vacuum_receiver.close(); + self.checkpoint_receiver.close(); + reject_retired_write_ingress( + &mut self.receiver, + &mut queue, + &self.telemetry, + UnavailableReasonV1::FileLost, + ); + drain_command_ingress( + &mut self.exact_sql_receiver, + &mut exact_sql_queue, + &mut exact_sql_closed, + ); + drain_command_ingress( + &mut self.incremental_vacuum_receiver, + &mut incremental_vacuum_queue, + &mut incremental_vacuum_closed, + ); + drain_command_ingress( + &mut self.checkpoint_receiver, + &mut checkpoint_queue, + &mut checkpoint_closed, + ); reject_all_exact_sql(&mut exact_sql_queue); reject_all_incremental_vacuum(&mut incremental_vacuum_queue); checkpoint_queue.clear(); @@ -593,45 +620,38 @@ impl Worker { break; } if self.state.load(Ordering::Acquire) == WriterState::Faulted as u8 { - reject_all(&mut queue, &self.telemetry, UnavailableReasonV1::Faulted); - reject_all_exact_sql(&mut exact_sql_queue); - reject_all_incremental_vacuum(&mut incremental_vacuum_queue); - checkpoint_queue.clear(); - if input_closed - && exact_sql_closed - && incremental_vacuum_closed - && checkpoint_closed - { - break; - } - let wake = wait_for_work( - &self.cx, + // A permanent fault must release its SQLite connection, even + // when registered clients still hold idle channel senders. + // Close ingress before rejecting already-admitted writes so + // a racing sender cannot lose its typed fault reply. + self.exact_sql_receiver.close(); + self.incremental_vacuum_receiver.close(); + self.checkpoint_receiver.close(); + reject_retired_write_ingress( &mut self.receiver, - &mut self.exact_sql_receiver, - &mut self.incremental_vacuum_receiver, - &mut self.checkpoint_receiver, - &mut self.shutdown_receiver, - input_closed, - exact_sql_closed, - incremental_vacuum_closed, - checkpoint_closed, - earliest(None, probe.as_ref()), - ) - .await; - apply_wake( - wake, &mut queue, - &mut inflight, - &mut exact_sql_queue, - &mut incremental_vacuum_queue, - &mut checkpoint_queue, &self.telemetry, - &mut input_closed, + UnavailableReasonV1::Faulted, + ); + drain_command_ingress( + &mut self.exact_sql_receiver, + &mut exact_sql_queue, &mut exact_sql_closed, + ); + drain_command_ingress( + &mut self.incremental_vacuum_receiver, + &mut incremental_vacuum_queue, &mut incremental_vacuum_closed, + ); + drain_command_ingress( + &mut self.checkpoint_receiver, + &mut checkpoint_queue, &mut checkpoint_closed, ); - continue; + reject_all_exact_sql(&mut exact_sql_queue); + reject_all_incremental_vacuum(&mut incremental_vacuum_queue); + checkpoint_queue.clear(); + return; } if let Some(command) = checkpoint_queue.pop_front() { self.run_requested_checkpoint(&mut checkpoint, command) @@ -731,7 +751,46 @@ impl Worker { tracing::trace_span!("rusqlite.writer.incremental_vacuum") .entered(); { - run_incremental_vacuum(checkpoint.connection_mut(), command); + if run_incremental_vacuum(checkpoint.connection_mut(), command) + { + // A failed SQLite rollback must not leak its + // transaction into the next accepted write. + self.state + .store(WriterState::Faulted as u8, Ordering::Release); + // Close channels before retiring the connection: + // accepted commands may still be in ingress. + self.exact_sql_receiver.close(); + self.incremental_vacuum_receiver.close(); + self.checkpoint_receiver.close(); + reject_retired_write_ingress( + &mut self.receiver, + &mut queue, + &self.telemetry, + UnavailableReasonV1::Faulted, + ); + drain_command_ingress( + &mut self.exact_sql_receiver, + &mut exact_sql_queue, + &mut exact_sql_closed, + ); + drain_command_ingress( + &mut self.incremental_vacuum_receiver, + &mut incremental_vacuum_queue, + &mut incremental_vacuum_closed, + ); + drain_command_ingress( + &mut self.checkpoint_receiver, + &mut checkpoint_queue, + &mut checkpoint_closed, + ); + reject_all_exact_sql(&mut exact_sql_queue); + reject_all_incremental_vacuum( + &mut incremental_vacuum_queue, + ); + checkpoint_queue.clear(); + // Closing the connection releases any live transaction. + return; + } } }; } else { @@ -879,6 +938,10 @@ impl Worker { break; } if self.state.load(Ordering::Acquire) == WriterState::Faulted as u8 { + // Earlier batch work may fault the writer. All remaining + // accepted batches have left the fair queue, so a normal + // faulted-queue drain cannot settle or account for them. + reject_faulted_execution_batches(batches, &self.telemetry); break; } } @@ -1151,10 +1214,48 @@ 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 rollback_incremental_vacuum( + mut transaction: rusqlite::Transaction<'_>, + failure: WriterActorError, +) -> WriterActorError { + // SQLITE_FULL can roll back the transaction automatically. Preserve its + // real cause rather than manufacture a rollback error. + if transaction.is_autocommit() { + transaction.set_drop_behavior(rusqlite::DropBehavior::Ignore); + return failure; + } + match transaction.rollback() { + Ok(()) => failure, + Err(error) => { + tracing::error!( + error = %error, + cause = %failure, + "SQLite vacuum rollback failed" + ); + WriterActorError::IncrementalVacuumFailed(format!( + "rollback after {failure} failed: {error}" + )) + } + } +} + fn run_incremental_vacuum( connection: &mut rusqlite::Connection, command: IncrementalVacuumCommand, -) { +) -> bool { if command .authority .verify(RuntimeWriteAuthorityStage::Dequeued) @@ -1163,20 +1264,22 @@ fn run_incremental_vacuum( command.settle(Err(WriterActorError::AuthorityDenied { stage: RuntimeWriteAuthorityStage::Dequeued, })); - return; + return !connection.is_autocommit(); } - let match_result = { + // A named `Result, _>` stays drop-live (and so keeps the + // mutable connection borrow) through the failure arm. Reduce the begin + // result to an owned error inside a `let ... else` scrutinee: its + // temporaries are dropped before the else block inspects the connection. + let mut begin_failure = None; + let Ok(mut transaction) = ({ let _span = tracing::trace_span!("rusqlite.begin_immediate").entered(); connection.transaction_with_behavior(TransactionBehavior::Immediate) - }; - let transaction = match match_result { - Ok(transaction) => transaction, - Err(error) => { - command.settle(Err(WriterActorError::IncrementalVacuumFailed( - error.to_string(), - ))); - return; + }) + .map_err(|error| begin_failure = Some(incremental_vacuum_error(error))) else { + if let Some(failure) = begin_failure { + command.settle(Err(failure)); } + return !connection.is_autocommit(); }; // Each result row releases one page. pragma_update only steps once, so // drain this bounded pragma before deciding whether its transaction commits. @@ -1191,31 +1294,42 @@ fn run_incremental_vacuum( Ok(()) }); if let Err(error) = reclaimed { - command.settle(Err(WriterActorError::IncrementalVacuumFailed( - error.to_string(), - ))); - return; + let failure = rollback_incremental_vacuum(transaction, incremental_vacuum_error(error)); + let unclosed_transaction = !connection.is_autocommit(); + command.settle(Err(failure)); + return unclosed_transaction; } if command .authority .verify(RuntimeWriteAuthorityStage::BeforeCommit) .is_err() { - let result = match transaction.rollback() { - Ok(()) => Err(WriterActorError::AuthorityDenied { + let failure = rollback_incremental_vacuum( + transaction, + WriterActorError::AuthorityDenied { stage: RuntimeWriteAuthorityStage::BeforeCommit, - }), - Err(error) => Err(WriterActorError::IncrementalVacuumFailed(format!( - "rollback after authority loss: {error}" - ))), - }; - command.settle(result); - return; + }, + ); + command.settle(Err(failure)); + return !connection.is_autocommit(); } - let result = transaction - .commit() - .map_err(|error| WriterActorError::IncrementalVacuumFailed(error.to_string())); + // Keep ownership after a failed COMMIT so rollback errors cannot be lost + // by Transaction::drop. A successful COMMIT has already closed the SQLite + // transaction, so prevent a redundant rollback on Drop. + let result = match transaction.execute_batch("COMMIT") { + Ok(()) => { + transaction.set_drop_behavior(rusqlite::DropBehavior::Ignore); + drop(transaction); + Ok(()) + } + Err(error) => Err(rollback_incremental_vacuum( + transaction, + incremental_vacuum_error(error), + )), + }; + let unclosed_transaction = !connection.is_autocommit(); command.settle(result); + unclosed_transaction } fn build_batches( @@ -1327,7 +1441,9 @@ 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, incremental_vacuum_error, reject_faulted_execution_batches, + reject_incremental_vacuum, reject_retired_write_ingress, rollback_incremental_vacuum, + run_incremental_vacuum, select_auxiliary_work, }; struct BatchProbe { @@ -1443,6 +1559,146 @@ mod auxiliary_scheduling_tests { ) } + #[test] + fn faulted_vacuum_rejects_unread_ingress_and_balances_telemetry() { + let telemetry = WriterTelemetry::default(); + let (sender, mut receiver) = mpsc::channel(2); + let mut queue = FairQueue::default(); + + let (parked, mut parked_reply) = + accepted_request_with_reply("operation.fault-parked", "key.fault-parked"); + telemetry.admitted(parked.admission_bytes()); + assert!(queue.push(parked).is_ok()); + + let (unread, mut unread_reply) = + accepted_request_with_reply("operation.fault-unread", "key.fault-unread"); + telemetry.admitted(unread.admission_bytes()); + assert!(sender.try_send(unread).is_ok()); + assert_eq!(telemetry.snapshot().queue.queued_operations, 2); + + reject_retired_write_ingress( + &mut receiver, + &mut queue, + &telemetry, + UnavailableReasonV1::Faulted, + ); + + let fault = RuntimeSubmitOutcomeV1::Unavailable { + reason: UnavailableReasonV1::Faulted, + }; + assert_eq!(parked_reply.try_recv().unwrap(), Ok(fault.clone())); + assert_eq!(unread_reply.try_recv().unwrap(), Ok(fault)); + assert!(queue.is_empty()); + let snapshot = telemetry.snapshot(); + assert_eq!(snapshot.queue.queued_operations, 0); + assert_eq!(snapshot.queue.queued_bytes, 0); + assert_eq!(snapshot.operations.admitted_operations, 2); + assert_eq!(snapshot.operations.completed_operations, 2); + + let (late, _) = accepted_request_with_reply("operation.fault-late", "key.fault-late"); + assert!(matches!( + sender.try_send(late), + Err(mpsc::SendError::Disconnected(_)) + )); + } + + #[test] + fn lost_file_settles_parked_and_unread_writes_with_typed_reason() { + let telemetry = WriterTelemetry::default(); + let (sender, mut receiver) = mpsc::channel(2); + let mut queue = FairQueue::default(); + + let (parked, mut parked_reply) = + accepted_request_with_reply("operation.lost-parked", "key.lost-parked"); + telemetry.admitted(parked.admission_bytes()); + assert!(queue.push(parked).is_ok()); + + let (unread, mut unread_reply) = + accepted_request_with_reply("operation.lost-unread", "key.lost-unread"); + telemetry.admitted(unread.admission_bytes()); + assert!(sender.try_send(unread).is_ok()); + assert_eq!(telemetry.snapshot().queue.queued_operations, 2); + + // The loss probe may race a sender that passed the Ready check. + // Both queued and unread admissions must receive FileLost, not a + // dropped reply, and the old store must retain no queue accounting. + reject_retired_write_ingress( + &mut receiver, + &mut queue, + &telemetry, + UnavailableReasonV1::FileLost, + ); + + let lost = RuntimeSubmitOutcomeV1::Unavailable { + reason: UnavailableReasonV1::FileLost, + }; + assert_eq!(parked_reply.try_recv().unwrap(), Ok(lost.clone())); + assert_eq!(unread_reply.try_recv().unwrap(), Ok(lost)); + assert!(queue.is_empty()); + let snapshot = telemetry.snapshot(); + assert_eq!(snapshot.queue.queued_operations, 0); + assert_eq!(snapshot.queue.queued_bytes, 0); + assert_eq!(snapshot.operations.admitted_operations, 2); + assert_eq!(snapshot.operations.completed_operations, 2); + + let (late, _) = accepted_request_with_reply("operation.lost-late", "key.lost-late"); + assert!(matches!( + sender.try_send(late), + Err(mpsc::SendError::Disconnected(_)) + )); + } + + #[test] + fn faulted_writer_settles_selected_but_unexecuted_batches() { + let telemetry = WriterTelemetry::default(); + let mut config = tracedecay_store::AdmissionConfigV1::default(); + config.foreground_batch.max_operations = 1; + config.background_batch.max_operations = 1; + let mut accepted = Vec::new(); + let mut replies = Vec::new(); + for index in 0..3 { + let (item, reply) = accepted_request_with_reply( + &format!("operation.faulted-batch.{index}"), + &format!("key.faulted-batch.{index}"), + ); + telemetry.admitted(item.admission_bytes()); + accepted.push(item); + replies.push(reply); + } + assert_eq!(telemetry.snapshot().queue.queued_operations, 3); + + // Only one batch reached execution before the writer faulted. The + // other two were already selected out of the fair queue. + let mut batches = build_batches(accepted, &config).into_iter(); + assert_eq!(batches.len(), 3); + let executed = batches.next().expect("first batch exists"); + for item in executed.items { + telemetry.released(1, item.admission_bytes()); + let result = Ok(RuntimeSubmitOutcomeV1::Unavailable { + reason: UnavailableReasonV1::Faulted, + }); + telemetry.completed(&result); + item.settle(result); + } + reject_faulted_execution_batches(batches, &telemetry); + + for mut reply in replies { + assert_eq!( + reply + .try_recv() + .expect("every admitted request must be settled"), + Ok(RuntimeSubmitOutcomeV1::Unavailable { + reason: UnavailableReasonV1::Faulted, + }) + ); + } + let snapshot = telemetry.snapshot(); + assert_eq!(snapshot.queue.queued_operations, 0); + assert_eq!(snapshot.queue.queued_bytes, 0); + assert_eq!(snapshot.operations.admitted_operations, 3); + assert_eq!(snapshot.operations.completed_operations, 3); + } + #[test] fn duplicate_operation_attach_settles_both_requesters_with_the_same_outcome() { let telemetry = WriterTelemetry::default(); @@ -2007,6 +2263,253 @@ 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(); + assert!( + !run_incremental_vacuum(&mut connection, vacuum), + "a settled SQLite busy/read-only attempt must not strand a transaction" + ); + 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(); + assert!( + !run_incremental_vacuum(&mut connection, vacuum), + "a settled SQLite busy/read-only attempt must not strand a transaction" + ); + 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 incremental_vacuum_commit_busy_preserves_rollback_outcome() { + for deny_rollback in [false, true] { + 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(); + connection + .pragma_update(None, "journal_mode", "DELETE") + .unwrap(); + connection + .pragma_update(None, "auto_vacuum", "INCREMENTAL") + .unwrap(); + connection + .execute_batch( + "CREATE TABLE retained(value TEXT NOT NULL); + INSERT INTO retained VALUES ('keep the live evidence'); + CREATE TABLE released(value BLOB NOT NULL); + WITH RECURSIVE copies(n) AS ( + VALUES(1) UNION ALL SELECT n + 1 FROM copies WHERE n < 64 + ) INSERT INTO released SELECT zeroblob(16384) FROM copies; + DELETE FROM released;", + ) + .unwrap(); + let before = vacuum_page_counts(&connection); + assert!(before.1 > 8); + // In rollback-journal mode a reader allows BEGIN IMMEDIATE and + // reclamation but prevents COMMIT from gaining the exclusive lock. + let reader = rusqlite::Connection::open(path).unwrap(); + reader + .execute_batch("BEGIN; SELECT value FROM retained;") + .unwrap(); + let rollback_observed = Arc::new(AtomicBool::new(false)); + let observed = Arc::clone(&rollback_observed); + connection + .authorizer(Some(move |context: rusqlite::hooks::AuthContext<'_>| { + if matches!( + context.action, + rusqlite::hooks::AuthAction::Transaction { + operation: rusqlite::hooks::TransactionOperation::Rollback + } + ) { + observed.store(true, Ordering::SeqCst); + if deny_rollback { + return rusqlite::hooks::Authorization::Deny; + } + } + rusqlite::hooks::Authorization::Allow + })) + .unwrap(); + let stages = Arc::new(Mutex::new(Vec::new())); + let authority = Arc::new(RecordingAuthority { + stages: Arc::clone(&stages), + deny_before_commit: false, + }); + let (reply, mut response) = oneshot::channel(); + let must_retire_writer = run_incremental_vacuum( + &mut connection, + IncrementalVacuumCommand::new(8, authority, reply), + ); + assert_eq!( + *stages.lock().unwrap(), + [ + RuntimeWriteAuthorityStage::Dequeued, + RuntimeWriteAuthorityStage::BeforeCommit + ] + ); + assert!(rollback_observed.load(Ordering::SeqCst)); + let result = response.try_recv().unwrap(); + if deny_rollback { + assert!(must_retire_writer, "failed cleanup must fault the writer"); + assert!(matches!( + result, + Err(WriterActorError::IncrementalVacuumFailed(message)) + if message.contains("rollback") && message.contains("not authorized") + )); + assert!(!connection.is_autocommit()); + connection + .authorizer(Some(|_: rusqlite::hooks::AuthContext<'_>| { + rusqlite::hooks::Authorization::Allow + })) + .unwrap(); + connection.execute_batch("ROLLBACK").unwrap(); + } else { + assert!(!must_retire_writer, "successful cleanup permits retry"); + assert!(matches!( + result, + Err(WriterActorError::IncrementalVacuumBusy) + )); + } + assert!(connection.is_autocommit()); + assert_eq!(vacuum_page_counts(&connection), before); + reader.execute_batch("ROLLBACK").unwrap(); + let (reply, mut response) = oneshot::channel(); + assert!(!run_incremental_vacuum( + &mut connection, + IncrementalVacuumCommand::new( + 8, + Arc::new(UnrestrictedRuntimeWriteAuthority), + reply, + ), + )); + assert!(response.try_recv().unwrap().is_ok()); + assert_eq!(before.1 - vacuum_page_counts(&connection).1, 8); + assert_eq!( + connection + .query_row("SELECT value FROM retained", [], |row| row + .get::<_, String>(0)) + .unwrap(), + "keep the live evidence" + ); + } + } + + #[test] + fn incremental_vacuum_execution_failure_preserves_rollback_error() { + let mut connection = vacuum_connection_with_free_pages(); + let before = vacuum_page_counts(&connection); + let rollback_observed = Arc::new(AtomicBool::new(false)); + let observed = Arc::clone(&rollback_observed); + connection + .authorizer(Some( + move |context: rusqlite::hooks::AuthContext<'_>| match context.action { + rusqlite::hooks::AuthAction::Pragma { .. } => { + rusqlite::hooks::Authorization::Deny + } + rusqlite::hooks::AuthAction::Transaction { + operation: rusqlite::hooks::TransactionOperation::Rollback, + } => { + observed.store(true, Ordering::SeqCst); + rusqlite::hooks::Authorization::Deny + } + _ => rusqlite::hooks::Authorization::Allow, + }, + )) + .unwrap(); + let stages = Arc::new(Mutex::new(Vec::new())); + let authority = Arc::new(RecordingAuthority { + stages: Arc::clone(&stages), + deny_before_commit: false, + }); + let (reply, mut response) = oneshot::channel(); + assert!(run_incremental_vacuum( + &mut connection, + IncrementalVacuumCommand::new(8, authority, reply), + )); + assert_eq!( + *stages.lock().unwrap(), + [RuntimeWriteAuthorityStage::Dequeued] + ); + assert!(rollback_observed.load(Ordering::SeqCst)); + assert!(matches!( + response.try_recv().unwrap(), + Err(WriterActorError::IncrementalVacuumFailed(message)) + if message.contains("rollback") && message.contains("not authorized") + )); + assert!(!connection.is_autocommit()); + connection + .authorizer(Some(|_: rusqlite::hooks::AuthContext<'_>| { + rusqlite::hooks::Authorization::Allow + })) + .unwrap(); + connection.execute_batch("ROLLBACK").unwrap(); + assert!(connection.is_autocommit()); + assert_eq!(vacuum_page_counts(&connection), before); + } + + #[test] + fn full_sqlite_disk_quota_defers_only_optional_vacuum() { + let mut 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 transaction = connection.transaction().unwrap(); + let full = transaction + .execute("INSERT INTO quota_probe VALUES (zeroblob(65536))", []) + .unwrap_err(); + assert_eq!( + full.sqlite_error_code(), + Some(rusqlite::ErrorCode::DiskFull) + ); + assert!( + transaction.is_autocommit(), + "SQLITE_FULL must not leave the old write transaction open" + ); + assert!(matches!( + rollback_incremental_vacuum(transaction, incremental_vacuum_error(full)), + WriterActorError::IncrementalVacuumNoSpace + )); + assert!(connection.is_autocommit()); + } + #[test] fn incremental_vacuum_samples_worker_authority_stages() { let mut connection = rusqlite::Connection::open_in_memory().unwrap(); @@ -2106,9 +2609,12 @@ mod auxiliary_scheduling_tests { }); let (reply, mut response) = oneshot::channel(); - run_incremental_vacuum( - &mut connection, - IncrementalVacuumCommand::new(8, authority, reply), + assert!( + !run_incremental_vacuum( + &mut connection, + IncrementalVacuumCommand::new(8, authority, reply), + ), + "a successful rollback must leave the writer usable" ); assert!(matches!( @@ -2120,4 +2626,60 @@ mod auxiliary_scheduling_tests { assert!(connection.is_autocommit()); assert_eq!(vacuum_page_counts(&connection), before); } + + #[test] + fn incremental_vacuum_preserves_rollback_failure_after_revocation() { + let mut connection = vacuum_connection_with_free_pages(); + let before = vacuum_page_counts(&connection); + let rollback_denied = Arc::new(AtomicBool::new(false)); + let observed = Arc::clone(&rollback_denied); + // Deny SQLite's actual ROLLBACK statement, not a fabricated error + // passed into the writer. BEGIN and the vacuum itself remain allowed. + connection + .authorizer(Some(move |context: rusqlite::hooks::AuthContext<'_>| { + if matches!( + context.action, + rusqlite::hooks::AuthAction::Transaction { + operation: rusqlite::hooks::TransactionOperation::Rollback + } + ) { + observed.store(true, Ordering::SeqCst); + rusqlite::hooks::Authorization::Deny + } else { + rusqlite::hooks::Authorization::Allow + } + })) + .unwrap(); + let authority = Arc::new(RecordingAuthority { + stages: Arc::new(Mutex::new(Vec::new())), + deny_before_commit: true, + }); + let (reply, mut response) = oneshot::channel(); + let must_retire_writer = run_incremental_vacuum( + &mut connection, + IncrementalVacuumCommand::new(8, authority, reply), + ); + assert!( + must_retire_writer, + "a failed rollback with an open transaction must retire the writer" + ); + + assert!(rollback_denied.load(Ordering::SeqCst)); + assert!(matches!( + response.try_recv().unwrap(), + Err(WriterActorError::IncrementalVacuumFailed(message)) + if message.contains("rollback") && message.contains("not authorized") + )); + // The failed rollback leaves a live transaction. Clear the test-only + // authorizer and recover it explicitly, then verify no data was lost. + assert!(!connection.is_autocommit()); + connection + .authorizer(Some(|_: rusqlite::hooks::AuthContext<'_>| { + rusqlite::hooks::Authorization::Allow + })) + .unwrap(); + connection.execute_batch("ROLLBACK").unwrap(); + assert!(connection.is_autocommit()); + assert_eq!(vacuum_page_counts(&connection), before); + } } diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs index b1c17f1c16d..6d260449f69 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs @@ -5,6 +5,8 @@ use std::collections::VecDeque; +use asupersync::channel::mpsc; + use tracedecay_store::{RuntimeCancellationStageV1, RuntimeSubmitOutcomeV1, UnavailableReasonV1}; use crate::{ @@ -16,7 +18,7 @@ use crate::{ use super::super::{ WriterActorError, - request::{AcceptedRequest, IncrementalVacuumCommand}, + request::{AcceptedRequest, ExecutionBatch, IncrementalVacuumCommand}, settlement::{interruption_outcome, missing_authority}, }; @@ -53,20 +55,57 @@ pub(super) fn reject_unauthorized( } } +/// A retired writer must settle every accepted request, including those not yet +/// transferred from ingress to its fair queue. Closing first prevents a racing +/// sender from losing its reply and leaving queue telemetry inflated. +pub(super) fn reject_retired_write_ingress( + receiver: &mut mpsc::Receiver, + queue: &mut FairQueue, + telemetry: &WriterTelemetry, + reason: UnavailableReasonV1, +) { + receiver.close(); + reject_all(queue, telemetry, reason); + while let Ok(item) = receiver.try_recv() { + settle_unavailable(item, telemetry, reason); + } +} + pub(super) fn reject_all( queue: &mut FairQueue, telemetry: &WriterTelemetry, reason: UnavailableReasonV1, ) { for item in queue.drain_all() { - let bytes = item.admission_bytes(); - let result = Ok(RuntimeSubmitOutcomeV1::Unavailable { reason }); - telemetry.released(1, bytes); - telemetry.completed(&result); - item.settle(result); + settle_unavailable(item, telemetry, reason); + } +} + +/// Selected batches are no longer in the fair queue. If an earlier batch +/// faults the writer, settle every remaining admission before retiring it. +pub(super) fn reject_faulted_execution_batches( + batches: impl IntoIterator, + telemetry: &WriterTelemetry, +) { + for batch in batches { + for item in batch.items { + settle_unavailable(item, telemetry, UnavailableReasonV1::Faulted); + } } } +fn settle_unavailable( + item: AcceptedRequest, + telemetry: &WriterTelemetry, + reason: UnavailableReasonV1, +) { + let bytes = item.admission_bytes(); + let result = Ok(RuntimeSubmitOutcomeV1::Unavailable { reason }); + telemetry.released(1, bytes); + telemetry.completed(&result); + item.settle(result); +} + pub(super) fn reject_all_exact_sql(queue: &mut VecDeque) { for command in queue.drain(..) { reject_writer_command(command); 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;