fix: make persistence failures terminal

README.md

@@ -49,6 +49,22 @@ worktable = { version = "=1.0.0-beta.2", features = ["s3-support"] }   # S3 sync
 
 Persisted indexes default to WorkTablesIndex. Vanilla IndexSet can be selected explicitly with `using indexset` while retaining the existing disk/S3 representation. Congee and Arctic are explicitly memory-only and require `persist: false`. The full syntax and capability matrix are documented in [Per-index backends with `using`](docs/index-backend-dsl-proposal.md).
 
+### Persistence lifecycle
+
+Persisted tables expose fallible draining and graceful shutdown:
+
+```rust
+table.wait_for_ops().await?; // drain currently queued operations
+table.close().await?;        // stop intake, drain, and join the engine task
+```
+
+An unrecoverable event gap, queue-analysis error, batch-apply error, or engine-task
+failure moves persistence into a terminal failed state. The original error is
+returned to waiters, graceful close, and later mutation attempts; later durable
+operations are not applied after that failure. Dropping a busy table remains a
+last-resort diagnostic path, so applications should call `close()` during orderly
+shutdown.
+
 WorkTablesIndex uses its predictable branch-based node search by default in WorkTable. This avoids a measured regression for sequential numeric-key workloads. Alternative search policies remain compile-time feature gates: disable WorkTable's default features and enable one of `wti-hybrid-search`, `wti-std-search`, or `wti-superslice-search` (plus any other features such as `s3-support`). Prefer one search feature for an unambiguous build. If Cargo feature unification enables several, WorkTablesIndex applies the documented deterministic precedence rather than rejecting the graph.
 
 ## Concurrent read/write publication

codegen/src/generators/persist/queries/delete.rs

@@ -95,7 +95,7 @@ impl PersistGenerator {
                 primary_key_events,
                 link,
             });
-            self.1.apply_operation(op);
+            self.1.apply_operation(op)?;
         };
 
         if is_locked {

codegen/src/generators/persist/queries/update.rs

@@ -212,7 +212,7 @@ impl PersistGenerator {
             } else {
                 unreachable!("")
             };
-            self.1.apply_operation(op);
+            self.1.apply_operation(op)?;
         }
     }
 
@@ -352,7 +352,7 @@ impl PersistGenerator {
                                     primary_key_events: vec![],
                                     secondary_keys_events: merged_events,
                                 });
-                                self.1.apply_operation(ack_op);
+                                self.1.apply_operation(ack_op)?;
 
                                 Err(WorkTableError::AlreadyExists(at.to_string_value()))
                             }

codegen/src/generators/persist/table/impls.rs

@@ -242,9 +242,10 @@ impl PersistGenerator {
 
         quote! {
             pub fn insert(&self, row: #row_type) -> core::result::Result<#primary_key_type, WorkTableError> {
+                self.1.ensure_running()?;
                 let (op, res) = self.0.insert_cdc::<#secondary_events_ident>(row);
                 if let Some(op) = op {
-                    self.1.apply_operation(op);
+                    self.1.apply_operation(op)?;
                 }
                 res
             }
@@ -259,9 +260,10 @@ impl PersistGenerator {
 
         quote! {
             pub async fn reinsert(&self, row_old: #row_type, row_new: #row_type) -> core::result::Result<#primary_key_type, WorkTableError> {
+                self.1.ensure_running()?;
                 let (op, res) = self.0.reinsert_cdc::<#secondary_events_ident>(row_old, row_new);
                 if let Some(op) = op {
-                    self.1.apply_operation(op);
+                    self.1.apply_operation(op)?;
                 }
                 res
             }

codegen/src/migration_engine/generator.rs

@@ -85,7 +85,7 @@ pub fn generate(input: MigrationEngineInput) -> TokenStream {
                     v => return Err(eyre::eyre!("Unsupported version: {}", v)),
                 };
 
-                target.wait_for_ops().await;
+                target.wait_for_ops().await?;
 
                 Ok(MigrationReport { source_version: version })
             }

codegen/src/persist_table/generator/space_file/worktable_impls.rs

@@ -10,30 +10,50 @@ impl Generator {
         let space_info_fn = self.gen_worktable_space_info_fn();
         let persisted_pk_fn = self.gen_worktable_persisted_primary_key_fn();
         let wait_for_ops_fn = self.gen_worktable_wait_for_ops_fn();
+        let close_fn = self.gen_worktable_close_fn();
 
         quote! {
             impl #ident {
                 #space_info_fn
                 #persisted_pk_fn
                 #wait_for_ops_fn
+                #close_fn
             }
         }
     }
 
     fn gen_worktable_wait_for_ops_fn(&self) -> TokenStream {
         if self.attributes.read_only {
             quote! {
-                pub async fn wait_for_ops(&self) {}
+                pub async fn wait_for_ops(&self) -> PersistenceResult {
+                    Ok(())
+                }
             }
         } else {
             quote! {
-                pub async fn wait_for_ops(&self) {
+                pub async fn wait_for_ops(&self) -> PersistenceResult {
                    self.1.wait_for_ops().await
                 }
             }
         }
     }
 
+    fn gen_worktable_close_fn(&self) -> TokenStream {
+        if self.attributes.read_only {
+            quote! {
+                pub async fn close(self) -> PersistenceResult {
+                    Ok(())
+                }
+            }
+        } else {
+            quote! {
+                pub async fn close(self) -> PersistenceResult {
+                    self.1.close().await
+                }
+            }
+        }
+    }
+
     fn gen_worktable_space_info_fn(&self) -> TokenStream {
         let name_generator = WorktableNameGenerator::from_struct_ident(&self.struct_def.ident);
         let pk = name_generator.get_primary_key_type_ident();

src/lib.rs

@@ -34,10 +34,10 @@ pub mod prelude {
     pub use crate::persistence::{
         AcknowledgeOperation, ArtPersistenceKey, DeleteOperation, DiskConfig, DiskPersistenceEngine,
         IndexTableOfContents, InsertOperation, Operation, OperationId, PersistedWorkTable, PersistenceConfig,
-        PersistenceEngine, PersistenceTask, ReadOnlyPersistenceEngine, SpaceArcticIndex, SpaceCongeeIndex, SpaceData,
-        SpaceDataOps, SpaceIndex, SpaceIndexOps, SpaceIndexUnsized, SpaceSecondaryIndexOps, UpdateOperation,
-        map_index_pages_to_toc_and_general, map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes,
-        validate_events,
+        PersistenceEngine, PersistenceError, PersistenceResult, PersistenceState, PersistenceTask,
+        ReadOnlyPersistenceEngine, SpaceArcticIndex, SpaceCongeeIndex, SpaceData, SpaceDataOps, SpaceIndex,
+        SpaceIndexOps, SpaceIndexUnsized, SpaceSecondaryIndexOps, UpdateOperation, map_index_pages_to_toc_and_general,
+        map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, validate_events,
     };
     pub use crate::primary_key::{PrimaryKeyGenerator, PrimaryKeyGeneratorState, TablePrimaryKey};
     pub use crate::table::select::{Order, QueryParams, SelectQueryBuilder, SelectQueryExecutor};

src/persistence/error.rs

@@ -0,0 +1,37 @@
+use std::error::Error;
+use std::fmt::{Display, Formatter};
+use std::sync::Arc;
+
+/// Terminal and lifecycle errors reported by a persistence task.
+#[derive(Debug)]
+pub enum PersistenceError {
+    /// New work was submitted after graceful shutdown began.
+    Closing,
+    /// New work was submitted after graceful shutdown completed.
+    Closed,
+    /// The persistence engine or its queue analyzer failed permanently.
+    Engine(eyre::Report),
+}
+
+impl Display for PersistenceError {
+    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Closing => formatter.write_str("persistence task is closing"),
+            Self::Closed => formatter.write_str("persistence task is closed"),
+            Self::Engine(error) => write!(formatter, "persistence engine failed: {error:#}"),
+        }
+    }
+}
+
+impl Error for PersistenceError {}
+
+pub type PersistenceResult<T = ()> = Result<T, Arc<PersistenceError>>;
+
+/// Observable state of the persistence worker.
+#[derive(Clone, Debug)]
+pub enum PersistenceState {
+    Running,
+    Closing,
+    Failed(Arc<PersistenceError>),
+    Closed,
+}

src/persistence/mod.rs

@@ -4,6 +4,7 @@ use crate::persistence::operation::BatchOperation;
 
 pub use engine::DiskConfig;
 pub use engine::DiskPersistenceEngine;
+pub use error::{PersistenceError, PersistenceResult, PersistenceState};
 pub use operation::{
     AcknowledgeOperation, DeleteOperation, InsertOperation, Operation, OperationId, OperationType, UpdateOperation,
     validate_events,
@@ -17,6 +18,7 @@ pub use space::{
 pub use task::PersistenceTask;
 
 mod engine;
+mod error;
 pub mod operation;
 mod readonly_engine;
 mod space;

src/persistence/operation/batch.rs

@@ -296,11 +296,11 @@ where
                 // that persists is a bug upstream of the analyzer; report it
                 // loudly instead of force-applying and corrupting the file.
                 if attempts > 8 {
-                    tracing::error!(
-                        "persistence stalled on primary index event gap: last applied {:?}, next available {:?}                          (attempt {attempts}); an event id was likely consumed without its event being queued",
+                    return Err(eyre::eyre!(
+                        "persistence stalled on primary index event gap: last applied {:?}, next available {:?} (attempt {attempts}); an event id was likely consumed without its event being queued",
                         last_ids.primary_id,
-                        id,
-                    );
+                        id
+                    ));
                 }
                 self.ops.extend(ops_to_remove);
                 return Ok(None);
@@ -318,9 +318,9 @@ where
                     // stream, defer until the missing event arrives, and report
                     // a persistent gap as the bug it is.
                     if attempts > 8 {
-                        tracing::error!(
-                            "persistence stalled on secondary index {index:?} event gap: last applied {last:?},                              next available {id:?} (attempt {attempts}); an event id was likely consumed without                              its event being queued",
-                        );
+                        return Err(eyre::eyre!(
+                            "persistence stalled on secondary index {index:?} event gap: last applied {last:?}, next available {id:?} (attempt {attempts}); an event id was likely consumed without its event being queued"
+                        ));
                     }
                     self.ops.extend(ops_to_remove);
                     return Ok(None);