From 5a9e081ae6b6e6a3b57579e37268fa00661ecfe9 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:31:53 -0800 Subject: [PATCH 01/16] fix(storage): preserve idle reclaim and LCM retry backoff Refs #4481 Rebase the accepted 13-file issue-specific patch intact onto current master without rewriting any existing owner branch. Persist retry and terminal settlement across dirty LCM invalidation, bound store page reclamation, and preserve fatal writer/authority failures while deferring optional SQLite BUSY, LOCKED, or FULL compaction. This clean successor replaces the blocked PR #4808 branch, whose last commit header exceeded the enforced 72-character limit. Carry the exact same 13 file blobs and preserve unrelated master changes. --- crates/tracedecay-global-db/src/registered.rs | 16 ++ .../src/registered/workflow_schema_tests.rs | 69 ++++++ .../tracedecay-global-db/src/schema_stages.rs | 5 - .../workflow_artifact_ceiling_upgrade.rs | 9 +- .../tracedecay-lcm/src/summary_convergence.rs | 36 ++- .../src/summary_convergence_tests.rs | 209 ++++++++++++++++++ .../src/db/connection/retained_maintenance.rs | 44 +++- .../src/shard_runtime/registry.rs | 4 + .../src/shard_runtime/registry/ports.rs | 11 +- .../src/repository/attachment.rs | 55 ++++- .../tracedecay-rusqlite-runtime/src/writer.rs | 18 +- .../src/writer/worker/mod.rs | 110 +++++++-- .../src/lcm_effects/tests.rs | 89 ++++++++ 13 files changed, 617 insertions(+), 58 deletions(-) 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; From b1c09c1781525702c16671b21eb2e34e39825fc0 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:01:09 -0800 Subject: [PATCH 02/16] fix(sqlite): report failed vacuum rollback after revocation Refs #4481. Propagate an actual SQLite rollback failure instead of masking it with an authority denial; preserve the authority result on success. Exercise a real SQLite authorization refusal during rollback and verify cleanup preserves live rows. --- .../src/writer/worker/mod.rs | 76 ++++++++++++++++--- 1 file changed, 67 insertions(+), 9 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index d134e686c0f..c4c891590de 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -1210,15 +1210,21 @@ fn run_incremental_vacuum( .verify(RuntimeWriteAuthorityStage::BeforeCommit) .is_err() { - 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, - })); + let failure = match transaction.rollback() { + Ok(()) => WriterActorError::AuthorityDenied { + stage: RuntimeWriteAuthorityStage::BeforeCommit, + }, + Err(error) => { + tracing::error!( + error = %error, + "SQLite vacuum rollback failed after write authority was revoked" + ); + WriterActorError::IncrementalVacuumFailed(format!( + "rollback after write authority revocation failed: {error}" + )) + } + }; + command.settle(Err(failure)); return; } let result = transaction.commit().map_err(incremental_vacuum_error); @@ -2192,4 +2198,56 @@ 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(); + run_incremental_vacuum( + &mut connection, + IncrementalVacuumCommand::new(8, authority, reply), + ); + + 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); + } } From 8ea62537017205b56d4abee4e4033cd2a5ac122f Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:22:30 -0800 Subject: [PATCH 03/16] test(sqlite): format quota-vacuum error assertion --- crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index c4c891590de..8290862f522 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -2078,7 +2078,10 @@ mod auxiliary_scheduling_tests { let full = connection .execute("INSERT INTO quota_probe VALUES (zeroblob(65536))", []) .unwrap_err(); - assert_eq!(full.sqlite_error_code(), Some(rusqlite::ErrorCode::DiskFull)); + assert_eq!( + full.sqlite_error_code(), + Some(rusqlite::ErrorCode::DiskFull) + ); assert!(matches!( incremental_vacuum_error(full), WriterActorError::IncrementalVacuumNoSpace From d6f2976f2fdfe87c2ec6d3225bf6f14c5dddf017 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 12:25:50 -0800 Subject: [PATCH 04/16] fix(sqlite): import vacuum classifier in quota regression Refs #4481 --- crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index 8290862f522..86acbbcf486 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -1340,7 +1340,8 @@ mod auxiliary_scheduling_tests { use super::{ AcceptedRequest, AuxiliaryWork, BatchCoalescingWindow, IncrementalVacuumCommand, PendingBatchDwell, WriterActorError, build_batches, checkpoint_config, dwell_for_batch, - enqueue, reject_incremental_vacuum, run_incremental_vacuum, select_auxiliary_work, + enqueue, incremental_vacuum_error, reject_incremental_vacuum, run_incremental_vacuum, + select_auxiliary_work, }; struct BatchProbe { From 41bd7ab65311d4c97b91c1021c89893a72fad93c Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 15:46:45 -0800 Subject: [PATCH 05/16] fix(storage): fault writer on failed vacuum rollback Refs #4481 Never reuse a SQLite writer connection if optional page reclamation leaves an open transaction. Drop the connection and reject queued work before a later write can accidentally join that transaction. Settle recoverable vacuum errors normally when SQLite is back in autocommit. Exercise a real SQLite rollback refusal plus successful rollback, contention and read-only paths to distinguish fatal and safe outcomes. --- .../src/writer/worker/mod.rs | 60 +++++++++++++++---- 1 file changed, 48 insertions(+), 12 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index 86acbbcf486..d871401088a 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -731,7 +731,26 @@ 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, + ); + 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(); + // Closing the connection releases any live transaction. + return; + } } }; } else { @@ -1167,7 +1186,7 @@ fn incremental_vacuum_error(error: rusqlite::Error) -> WriterActorError { fn run_incremental_vacuum( connection: &mut rusqlite::Connection, command: IncrementalVacuumCommand, -) { +) -> bool { if command .authority .verify(RuntimeWriteAuthorityStage::Dequeued) @@ -1176,7 +1195,7 @@ fn run_incremental_vacuum( command.settle(Err(WriterActorError::AuthorityDenied { stage: RuntimeWriteAuthorityStage::Dequeued, })); - return; + return !connection.is_autocommit(); } let match_result = { let _span = tracing::trace_span!("rusqlite.begin_immediate").entered(); @@ -1186,7 +1205,7 @@ fn run_incremental_vacuum( Ok(transaction) => transaction, Err(error) => { command.settle(Err(incremental_vacuum_error(error))); - return; + return !connection.is_autocommit(); } }; // Each result row releases one page. pragma_update only steps once, so @@ -1202,8 +1221,10 @@ fn run_incremental_vacuum( Ok(()) }); if let Err(error) = reclaimed { + drop(transaction); + let unclosed_transaction = !connection.is_autocommit(); command.settle(Err(incremental_vacuum_error(error))); - return; + return unclosed_transaction; } if command .authority @@ -1225,10 +1246,12 @@ fn run_incremental_vacuum( } }; command.settle(Err(failure)); - return; + return !connection.is_autocommit(); } let result = transaction.commit().map_err(incremental_vacuum_error); + let unclosed_transaction = !connection.is_autocommit(); command.settle(result); + unclosed_transaction } fn build_batches( @@ -2042,7 +2065,10 @@ mod auxiliary_scheduling_tests { }; let (vacuum, mut result) = command(); - run_incremental_vacuum(&mut connection, vacuum); + 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) @@ -2051,7 +2077,10 @@ mod auxiliary_scheduling_tests { connection.pragma_update(None, "query_only", true).unwrap(); let (vacuum, mut result) = command(); - run_incremental_vacuum(&mut connection, vacuum); + 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(_)) @@ -2188,9 +2217,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!( @@ -2231,10 +2263,14 @@ mod auxiliary_scheduling_tests { deny_before_commit: true, }); let (reply, mut response) = oneshot::channel(); - run_incremental_vacuum( + 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!( From be64067457c10949f601c8be264a04f83d64ed6b Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 15:47:45 -0800 Subject: [PATCH 06/16] style(storage): wrap vacuum retirement call Refs #4481 Keep the fail-closed writer fix within the Rust formatting line budget. --- crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index d871401088a..403bbf25e45 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -731,7 +731,10 @@ impl Worker { tracing::trace_span!("rusqlite.writer.incremental_vacuum") .entered(); { - if 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( From 1c4193af7ff58187696bc2eb527d9f932d01c0ce Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:20:22 -0800 Subject: [PATCH 07/16] style(sqlite): format incremental vacuum fault handling --- .../src/writer/worker/mod.rs | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index 403bbf25e45..02be8062cae 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -731,16 +731,12 @@ impl Worker { tracing::trace_span!("rusqlite.writer.incremental_vacuum") .entered(); { - if 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, - ); + self.state + .store(WriterState::Faulted as u8, Ordering::Release); reject_all( &mut queue, &self.telemetry, From b09a1e7263c28960d37babca4bfff88f36d75651 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:37:58 -0800 Subject: [PATCH 08/16] fix(sqlite): settle unread writes on vacuum fault Refs #4481 Close buffered writer ingress before retirement, drain already-admitted commands, and reconcile typed fault replies and queue telemetry. Real channel regression covers buffered and parked writes plus late admission. --- .../src/writer/worker/mod.rs | 67 +++++++++++++++++-- .../src/writer/worker/rejection.rs | 23 +++++++ 2 files changed, 86 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index 02be8062cae..01721e5cfeb 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -66,7 +66,7 @@ use ingress::{ }; use rejection::{ cancel_waiting, reject_all, reject_all_exact_sql, reject_all_incremental_vacuum, - reject_incremental_vacuum, reject_unauthorized, + reject_faulted_write_ingress, reject_incremental_vacuum, reject_unauthorized, }; /// When the blocked writer retries its drain: the fence's short interval, or @@ -737,10 +737,30 @@ impl Worker { // transaction into the next accepted write. self.state .store(WriterState::Faulted as u8, Ordering::Release); - reject_all( + // 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_faulted_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( @@ -1362,7 +1382,8 @@ mod auxiliary_scheduling_tests { use super::{ AcceptedRequest, AuxiliaryWork, BatchCoalescingWindow, IncrementalVacuumCommand, PendingBatchDwell, WriterActorError, build_batches, checkpoint_config, dwell_for_batch, - enqueue, incremental_vacuum_error, reject_incremental_vacuum, run_incremental_vacuum, + enqueue, incremental_vacuum_error, reject_faulted_write_ingress, + reject_incremental_vacuum, run_incremental_vacuum, select_auxiliary_work, }; @@ -1479,6 +1500,44 @@ 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_faulted_write_ingress(&mut receiver, &mut queue, &telemetry); + + 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 duplicate_operation_attach_settles_both_requesters_with_the_same_outcome() { let telemetry = WriterTelemetry::default(); diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs index b1c17f1c16d..7bd4f589faa 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::{ @@ -53,6 +55,27 @@ pub(super) fn reject_unauthorized( } } +/// Shut off ingress before settling buffered requests on fatal writer +/// retirement. Dropping this receiver would lose accepted request replies and +/// leave their queued-operation telemetry inflated. +pub(super) fn reject_faulted_write_ingress( + receiver: &mut mpsc::Receiver, + queue: &mut FairQueue, + telemetry: &WriterTelemetry, +) { + receiver.close(); + reject_all(queue, telemetry, UnavailableReasonV1::Faulted); + while let Ok(item) = receiver.try_recv() { + let bytes = item.admission_bytes(); + let result = Ok(RuntimeSubmitOutcomeV1::Unavailable { + reason: UnavailableReasonV1::Faulted, + }); + telemetry.released(1, bytes); + telemetry.completed(&result); + item.settle(result); + } +} + pub(super) fn reject_all( queue: &mut FairQueue, telemetry: &WriterTelemetry, From 3d62dded169529f9473ae2d3ea867c3de3e96c2d Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:40:26 -0800 Subject: [PATCH 09/16] fix(storage): preserve vacuum cleanup errors after failed commits Use explicit rollback for vacuum execution and commit failures so a failed rollback cannot masquerade as a retryable busy or disk-full condition. Preserve automatic SQLite rollback on disk-full and retire writers when a transaction cannot be closed. Add real-SQLite contention, refusal, recovery, and data-retention regressions. Refs #4481 --- .../src/writer/worker/mod.rs | 237 ++++++++++++++++-- 1 file changed, 217 insertions(+), 20 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index 01721e5cfeb..4ed806d731f 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -1202,6 +1202,31 @@ fn incremental_vacuum_error(error: rusqlite::Error) -> WriterActorError { } } +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, @@ -1220,7 +1245,7 @@ fn run_incremental_vacuum( let _span = tracing::trace_span!("rusqlite.begin_immediate").entered(); connection.transaction_with_behavior(TransactionBehavior::Immediate) }; - let transaction = match match_result { + let mut transaction = match match_result { Ok(transaction) => transaction, Err(error) => { command.settle(Err(incremental_vacuum_error(error))); @@ -1240,9 +1265,9 @@ fn run_incremental_vacuum( Ok(()) }); if let Err(error) = reclaimed { - drop(transaction); + let failure = rollback_incremental_vacuum(transaction, incremental_vacuum_error(error)); let unclosed_transaction = !connection.is_autocommit(); - command.settle(Err(incremental_vacuum_error(error))); + command.settle(Err(failure)); return unclosed_transaction; } if command @@ -1250,24 +1275,29 @@ fn run_incremental_vacuum( .verify(RuntimeWriteAuthorityStage::BeforeCommit) .is_err() { - let failure = match transaction.rollback() { - Ok(()) => WriterActorError::AuthorityDenied { + let failure = rollback_incremental_vacuum( + transaction, + WriterActorError::AuthorityDenied { stage: RuntimeWriteAuthorityStage::BeforeCommit, }, - Err(error) => { - tracing::error!( - error = %error, - "SQLite vacuum rollback failed after write authority was revoked" - ); - WriterActorError::IncrementalVacuumFailed(format!( - "rollback after write authority revocation failed: {error}" - )) - } - }; + ); command.settle(Err(failure)); return !connection.is_autocommit(); } - let result = transaction.commit().map_err(incremental_vacuum_error); + // 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 @@ -1383,7 +1413,7 @@ mod auxiliary_scheduling_tests { AcceptedRequest, AuxiliaryWork, BatchCoalescingWindow, IncrementalVacuumCommand, PendingBatchDwell, WriterActorError, build_batches, checkpoint_config, dwell_for_batch, enqueue, incremental_vacuum_error, reject_faulted_write_ingress, - reject_incremental_vacuum, run_incremental_vacuum, + reject_incremental_vacuum, rollback_incremental_vacuum, run_incremental_vacuum, select_auxiliary_work, }; @@ -2152,9 +2182,170 @@ mod auxiliary_scheduling_tests { )); } + #[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 connection = rusqlite::Connection::open_in_memory().unwrap(); + 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 @@ -2163,17 +2354,23 @@ mod auxiliary_scheduling_tests { // 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 + 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!( - incremental_vacuum_error(full), + rollback_incremental_vacuum(transaction, incremental_vacuum_error(full)), WriterActorError::IncrementalVacuumNoSpace )); + assert!(connection.is_autocommit()); } #[test] From 3501f921fc5f77021324df5cf4eba95db4330c06 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:20:31 -0800 Subject: [PATCH 10/16] style(storage): format vacuum rollback regression fixtures Apply the four Bazel rustfmt_check_2024 hunks from CI run 42737462258. Retain test logic, admissions, rollback observations and assertions. Refs #4481 --- .../src/writer/worker/mod.rs | 25 +++++++++++-------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index 4ed806d731f..4e89859236d 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -1412,9 +1412,8 @@ mod auxiliary_scheduling_tests { use super::{ AcceptedRequest, AuxiliaryWork, BatchCoalescingWindow, IncrementalVacuumCommand, PendingBatchDwell, WriterActorError, build_batches, checkpoint_config, dwell_for_batch, - enqueue, incremental_vacuum_error, reject_faulted_write_ingress, - reject_incremental_vacuum, rollback_incremental_vacuum, run_incremental_vacuum, - select_auxiliary_work, + enqueue, incremental_vacuum_error, reject_faulted_write_ingress, reject_incremental_vacuum, + rollback_incremental_vacuum, run_incremental_vacuum, select_auxiliary_work, }; struct BatchProbe { @@ -2189,8 +2188,12 @@ mod auxiliary_scheduling_tests { 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 + .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); @@ -2207,7 +2210,9 @@ mod auxiliary_scheduling_tests { // 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(); + reader + .execute_batch("BEGIN; SELECT value FROM retained;") + .unwrap(); let rollback_observed = Arc::new(AtomicBool::new(false)); let observed = Arc::clone(&rollback_observed); connection @@ -2297,8 +2302,8 @@ mod auxiliary_scheduling_tests { 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 { + .authorizer(Some( + move |context: rusqlite::hooks::AuthContext<'_>| match context.action { rusqlite::hooks::AuthAction::Pragma { .. } => { rusqlite::hooks::Authorization::Deny } @@ -2309,8 +2314,8 @@ mod auxiliary_scheduling_tests { rusqlite::hooks::Authorization::Deny } _ => rusqlite::hooks::Authorization::Allow, - } - })) + }, + )) .unwrap(); let stages = Arc::new(Mutex::new(Vec::new())); let authority = Arc::new(RecordingAuthority { From 3b94037b3f12b5574fb5d0bdf365b7d390e5c179 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 18:37:04 -0800 Subject: [PATCH 11/16] fix(storage): settle unexecuted batches after writer faults An execution-batch fault formerly dropped later selected batches, losing caller replies and leaving queue telemetry inflated. Fail those accepted requests with typed Faulted outcomes before retiring the worker. Refs #4481 --- .../src/writer/worker/mod.rs | 61 ++++++++++++++++++- .../src/writer/worker/rejection.rs | 41 +++++++++---- 2 files changed, 86 insertions(+), 16 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index 4e89859236d..b4102e177f8 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -66,7 +66,8 @@ use ingress::{ }; use rejection::{ cancel_waiting, reject_all, reject_all_exact_sql, reject_all_incremental_vacuum, - reject_faulted_write_ingress, reject_incremental_vacuum, reject_unauthorized, + reject_faulted_execution_batches, reject_faulted_write_ingress, reject_incremental_vacuum, + reject_unauthorized, }; /// When the blocked writer retries its drain: the fence's short interval, or @@ -917,6 +918,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; } } @@ -1412,8 +1417,9 @@ mod auxiliary_scheduling_tests { use super::{ AcceptedRequest, AuxiliaryWork, BatchCoalescingWindow, IncrementalVacuumCommand, PendingBatchDwell, WriterActorError, build_batches, checkpoint_config, dwell_for_batch, - enqueue, incremental_vacuum_error, reject_faulted_write_ingress, reject_incremental_vacuum, - rollback_incremental_vacuum, run_incremental_vacuum, select_auxiliary_work, + enqueue, incremental_vacuum_error, reject_faulted_execution_batches, + reject_faulted_write_ingress, reject_incremental_vacuum, rollback_incremental_vacuum, + run_incremental_vacuum, select_auxiliary_work, }; struct BatchProbe { @@ -1567,6 +1573,55 @@ mod auxiliary_scheduling_tests { )); } + #[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(); diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs index 7bd4f589faa..cab3134882f 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs @@ -18,7 +18,7 @@ use crate::{ use super::super::{ WriterActorError, - request::{AcceptedRequest, IncrementalVacuumCommand}, + request::{AcceptedRequest, ExecutionBatch, IncrementalVacuumCommand}, settlement::{interruption_outcome, missing_authority}, }; @@ -66,13 +66,7 @@ pub(super) fn reject_faulted_write_ingress( receiver.close(); reject_all(queue, telemetry, UnavailableReasonV1::Faulted); while let Ok(item) = receiver.try_recv() { - let bytes = item.admission_bytes(); - let result = Ok(RuntimeSubmitOutcomeV1::Unavailable { - reason: UnavailableReasonV1::Faulted, - }); - telemetry.released(1, bytes); - telemetry.completed(&result); - item.settle(result); + settle_unavailable(item, telemetry, UnavailableReasonV1::Faulted); } } @@ -82,14 +76,35 @@ pub(super) fn reject_all( 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); From def8c54751095a0f29ef5a7cf873dea109fca9b9 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 19:46:33 -0800 Subject: [PATCH 12/16] style(storage): format writer fault regression Align the existing behavioral regression with rustfmt so the Bazel repository gate can complete and exercise the source fix. Refs #4481 --- crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index b4102e177f8..a745a3363ac 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -1609,7 +1609,9 @@ mod auxiliary_scheduling_tests { for mut reply in replies { assert_eq!( - reply.try_recv().expect("every admitted request must be settled"), + reply + .try_recv() + .expect("every admitted request must be settled"), Ok(RuntimeSubmitOutcomeV1::Unavailable { reason: UnavailableReasonV1::Faulted, }) From ce06bfc59ef965866c2c1bead9cab610a12120e1 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 21:24:26 -0800 Subject: [PATCH 13/16] fix(storage): retire faulted SQLite writers with live clients Refs #4481 Faulted actors must not retain SQLite connections until registered senders disconnect. Close ingress and drain accepted work on retirement. Exercise the live writer and verify typed replies and zero queued bytes. --- .../src/writer/tests/mod.rs | 50 +++++++++++++++++ .../src/writer/worker/mod.rs | 54 ++++++++----------- 2 files changed, 71 insertions(+), 33 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/tests/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/tests/mod.rs index 6d8e8864d1c..6df8c24c047 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/tests/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/tests/mod.rs @@ -315,6 +315,56 @@ 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 a745a3363ac..db788fd0bfc 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -594,45 +594,33 @@ 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, - &mut self.receiver, + // 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_faulted_write_ingress(&mut self.receiver, &mut queue, &self.telemetry); + drain_command_ingress( &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, &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) From 86a61da37342910c788d54d90678b1f4028d46ad Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:15:45 -0800 Subject: [PATCH 14/16] fix(storage): compile and format writer fault regressions Refs #4481 The vacuum worker kept a named begin-transaction Result live through its failure arm, so inspecting the connection there conflicted with the mutable borrow (E0502). Reduce the begin result to an owned error in a let-else scrutinee, preserving settle-then-report ordering. The fault regression also needs the QueueItem trait in scope and a pinned-rustfmt wrap. No retry, vacuum, authority, or assertion behavior changes. Co-Authored-By: Claude Sonnet 5.5 --- .../src/writer/tests/mod.rs | 5 ++++- .../src/writer/worker/mod.rs | 18 +++++++++++------- 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/tests/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/tests/mod.rs index 6df8c24c047..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, @@ -354,7 +355,9 @@ fn faulted_writer_retires_without_waiting_for_registered_client_senders() { "faulted writer retained its SQLite connection while clients remained registered" ); assert_eq!( - response.try_recv().expect("accepted request receives a fault reply"), + response + .try_recv() + .expect("accepted request receives a fault reply"), Ok(RuntimeSubmitOutcomeV1::Unavailable { reason: UnavailableReasonV1::Faulted, }) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index db788fd0bfc..9483e455797 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -1234,16 +1234,20 @@ fn run_incremental_vacuum( })); 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 mut transaction = match match_result { - Ok(transaction) => transaction, - Err(error) => { - command.settle(Err(incremental_vacuum_error(error))); - return !connection.is_autocommit(); + }) + .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. From baab26e867746d99e5f0dca3a1c148cda0e96c68 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 10 Nov 2026 00:19:13 -0800 Subject: [PATCH 15/16] fix(sqlite): settle accepted writes when store file disappears Refs #4481 --- .../src/writer/worker/mod.rs | 97 +++++++++++++++++-- .../src/writer/worker/rejection.rs | 13 +-- 2 files changed, 97 insertions(+), 13 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index 9483e455797..c7870871278 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -65,8 +65,8 @@ use ingress::{ select_auxiliary_work, wait_for_work, }; use rejection::{ - cancel_waiting, reject_all, reject_all_exact_sql, reject_all_incremental_vacuum, - reject_faulted_execution_batches, reject_faulted_write_ingress, reject_incremental_vacuum, + cancel_waiting, reject_all_exact_sql, reject_all_incremental_vacuum, + reject_faulted_execution_batches, reject_incremental_vacuum, reject_retired_write_ingress, reject_unauthorized, }; @@ -570,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(); @@ -601,7 +627,12 @@ impl Worker { self.exact_sql_receiver.close(); self.incremental_vacuum_receiver.close(); self.checkpoint_receiver.close(); - reject_faulted_write_ingress(&mut self.receiver, &mut queue, &self.telemetry); + 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, @@ -731,10 +762,11 @@ impl Worker { self.exact_sql_receiver.close(); self.incremental_vacuum_receiver.close(); self.checkpoint_receiver.close(); - reject_faulted_write_ingress( + reject_retired_write_ingress( &mut self.receiver, &mut queue, &self.telemetry, + UnavailableReasonV1::Faulted, ); drain_command_ingress( &mut self.exact_sql_receiver, @@ -1410,7 +1442,7 @@ mod auxiliary_scheduling_tests { AcceptedRequest, AuxiliaryWork, BatchCoalescingWindow, IncrementalVacuumCommand, PendingBatchDwell, WriterActorError, build_batches, checkpoint_config, dwell_for_batch, enqueue, incremental_vacuum_error, reject_faulted_execution_batches, - reject_faulted_write_ingress, reject_incremental_vacuum, rollback_incremental_vacuum, + reject_retired_write_ingress, reject_incremental_vacuum, rollback_incremental_vacuum, run_incremental_vacuum, select_auxiliary_work, }; @@ -1544,7 +1576,12 @@ mod auxiliary_scheduling_tests { assert!(sender.try_send(unread).is_ok()); assert_eq!(telemetry.snapshot().queue.queued_operations, 2); - reject_faulted_write_ingress(&mut receiver, &mut queue, &telemetry); + reject_retired_write_ingress( + &mut receiver, + &mut queue, + &telemetry, + UnavailableReasonV1::Faulted, + ); let fault = RuntimeSubmitOutcomeV1::Unavailable { reason: UnavailableReasonV1::Faulted, @@ -1565,6 +1602,52 @@ mod auxiliary_scheduling_tests { )); } + #[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(); diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs index cab3134882f..6d260449f69 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/rejection.rs @@ -55,18 +55,19 @@ pub(super) fn reject_unauthorized( } } -/// Shut off ingress before settling buffered requests on fatal writer -/// retirement. Dropping this receiver would lose accepted request replies and -/// leave their queued-operation telemetry inflated. -pub(super) fn reject_faulted_write_ingress( +/// 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, UnavailableReasonV1::Faulted); + reject_all(queue, telemetry, reason); while let Ok(item) = receiver.try_recv() { - settle_unavailable(item, telemetry, UnavailableReasonV1::Faulted); + settle_unavailable(item, telemetry, reason); } } From 8b5ee8fe256014beeb5681ca3d42312b0902599e Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 10 Nov 2026 00:21:42 -0800 Subject: [PATCH 16/16] style(sqlite): order retired-writer test imports Refs #4481 --- crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs index c7870871278..ee20c0cec58 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/worker/mod.rs @@ -1442,7 +1442,7 @@ mod auxiliary_scheduling_tests { AcceptedRequest, AuxiliaryWork, BatchCoalescingWindow, IncrementalVacuumCommand, PendingBatchDwell, WriterActorError, build_batches, checkpoint_config, dwell_for_batch, enqueue, incremental_vacuum_error, reject_faulted_execution_batches, - reject_retired_write_ingress, reject_incremental_vacuum, rollback_incremental_vacuum, + reject_incremental_vacuum, reject_retired_write_ingress, rollback_incremental_vacuum, run_incremental_vacuum, select_auxiliary_work, };