From 5948a6e7f69705ff9abf374265f8a9f20bedd22c Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 4 Aug 2026 03:10:02 +0700 Subject: [PATCH 1/2] test: cover persisted composite primary keys --- tests/persistence/mod.rs | 1 + tests/persistence/tuple_primary_key.rs | 89 ++++++++++++++++++++++++++ 2 files changed, 90 insertions(+) create mode 100644 tests/persistence/tuple_primary_key.rs diff --git a/tests/persistence/mod.rs b/tests/persistence/mod.rs index 4a965337..aa15496c 100644 --- a/tests/persistence/mod.rs +++ b/tests/persistence/mod.rs @@ -13,6 +13,7 @@ mod space_index; mod sync; mod toc; mod torn_shutdown; +mod tuple_primary_key; mod vacuum; #[cfg(feature = "s3-support")] diff --git a/tests/persistence/tuple_primary_key.rs b/tests/persistence/tuple_primary_key.rs new file mode 100644 index 00000000..e8b10f9c --- /dev/null +++ b/tests/persistence/tuple_primary_key.rs @@ -0,0 +1,89 @@ +use worktable::prelude::*; +use worktable::worktable; + +use crate::remove_dir_if_exists; + +worktable! ( + name: PersistedTuplePrimaryKey, + persist: true, + columns: { + tenant_id: u64 primary_key, + record_id: u64 primary_key, + value: i64, + }, +); + +#[tokio::test] +async fn composite_primary_key_survives_mutations_and_reload() { + let path = "tests/data/persisted_tuple_primary_key/reload"; + remove_dir_if_exists(path.to_string()).await; + + let config = DiskConfig::new_with_table_name( + path, + PersistedTuplePrimaryKeyWorkTable::name_snake_case(), + PersistedTuplePrimaryKeyWorkTable::version(), + ); + let rows = [ + PersistedTuplePrimaryKeyRow { + tenant_id: 7, + record_id: 41, + value: -10, + }, + PersistedTuplePrimaryKeyRow { + tenant_id: 7, + record_id: 42, + value: -11, + }, + PersistedTuplePrimaryKeyRow { + tenant_id: 8, + record_id: 1, + value: -12, + }, + ]; + + { + let engine = PersistedTuplePrimaryKeyPersistenceEngine::new(config.clone()) + .await + .unwrap(); + let table = PersistedTuplePrimaryKeyWorkTable::load(engine).await.unwrap(); + for row in &rows { + table.insert(row.clone()).unwrap(); + } + table.wait_for_ops().await; + for row in &rows { + assert_eq!(table.select((row.tenant_id, row.record_id)), Some(row.clone())); + } + } + + { + let engine = PersistedTuplePrimaryKeyPersistenceEngine::new(config.clone()) + .await + .unwrap(); + let table = PersistedTuplePrimaryKeyWorkTable::load(engine).await.unwrap(); + for row in &rows { + assert_eq!(table.select((row.tenant_id, row.record_id)), Some(row.clone())); + } + + let range = table.select_by_pk_range((7, 41)..=(7, 42)).execute().unwrap(); + assert_eq!(range, rows[..2]); + + let updated = PersistedTuplePrimaryKeyRow { + value: 99, + ..rows[1].clone() + }; + table.update(updated.clone()).await.unwrap(); + table.delete((7, 41)).await.unwrap(); + table.wait_for_ops().await; + + assert_eq!(table.select((7, 42)), Some(updated)); + assert!(table.select((7, 41)).is_none()); + } + + { + let engine = PersistedTuplePrimaryKeyPersistenceEngine::new(config).await.unwrap(); + let table = PersistedTuplePrimaryKeyWorkTable::load(engine).await.unwrap(); + assert!(table.select((7, 41)).is_none()); + assert_eq!(table.select((7, 42)).unwrap().value, 99); + assert_eq!(table.select((8, 1)), Some(rows[2].clone())); + } +} From ceb4e61ea23df86301ca4d14d77c7fd4001bbbe0 Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 4 Aug 2026 03:22:42 +0700 Subject: [PATCH 2/2] fix: make persistence failures terminal --- README.md | 16 + .../src/generators/persist/queries/delete.rs | 2 +- .../src/generators/persist/queries/update.rs | 4 +- codegen/src/generators/persist/table/impls.rs | 6 +- codegen/src/migration_engine/generator.rs | 2 +- .../generator/space_file/worktable_impls.rs | 24 +- src/lib.rs | 8 +- src/persistence/error.rs | 37 ++ src/persistence/mod.rs | 2 + src/persistence/operation/batch.rs | 14 +- src/persistence/task.rs | 348 ++++++++++++++++-- src/table/mod.rs | 2 + src/table/vacuum/mod.rs | 3 +- src/table/vacuum/vacuum.rs | 51 +-- tests/migration/mod.rs | 8 +- tests/persistence/bulk_load_stall.rs | 3 +- tests/persistence/concurrent/mod.rs | 2 +- .../persistence/duplicate_key_index_reload.rs | 18 +- tests/persistence/failure/insert.rs | 48 ++- tests/persistence/failure/reinsert.rs | 48 ++- tests/persistence/failure/update.rs | 16 +- .../persistence/failure/update_non_unique.rs | 16 +- tests/persistence/failure/update_unsized.rs | 32 +- tests/persistence/loaded_index_growth.rs | 9 +- tests/persistence/s3/mod.rs | 2 +- tests/persistence/sync/failure.rs | 12 +- tests/persistence/sync/failure_multi_index.rs | 6 +- tests/persistence/sync/many_strings.rs | 8 +- tests/persistence/sync/mod.rs | 18 +- tests/persistence/sync/option.rs | 32 +- .../persistence/sync/string_primary_index.rs | 16 +- tests/persistence/sync/string_re_read.rs | 52 +-- .../sync/string_secondary_index.rs | 18 +- .../persistence/sync/string_update_timeout.rs | 6 +- tests/persistence/sync/uuid_.rs | 8 +- tests/persistence/torn_shutdown.rs | 9 +- tests/persistence/tuple_primary_key.rs | 5 +- tests/persistence/vacuum.rs | 9 +- tests/worktable/index_backends.rs | 28 +- tests/worktable_version/basic.rs | 2 +- tests/worktable_version/string_primary_key.rs | 2 +- 41 files changed, 697 insertions(+), 255 deletions(-) create mode 100644 src/persistence/error.rs diff --git a/README.md b/README.md index 023fb7c2..85390100 100644 --- a/README.md +++ b/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 diff --git a/codegen/src/generators/persist/queries/delete.rs b/codegen/src/generators/persist/queries/delete.rs index e7fd6b58..0c3e2290 100644 --- a/codegen/src/generators/persist/queries/delete.rs +++ b/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 { diff --git a/codegen/src/generators/persist/queries/update.rs b/codegen/src/generators/persist/queries/update.rs index cce93e09..b9108081 100644 --- a/codegen/src/generators/persist/queries/update.rs +++ b/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())) } diff --git a/codegen/src/generators/persist/table/impls.rs b/codegen/src/generators/persist/table/impls.rs index c7474f6f..4f8267c6 100644 --- a/codegen/src/generators/persist/table/impls.rs +++ b/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 } diff --git a/codegen/src/migration_engine/generator.rs b/codegen/src/migration_engine/generator.rs index 4743adc0..b81ea25f 100644 --- a/codegen/src/migration_engine/generator.rs +++ b/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 }) } diff --git a/codegen/src/persist_table/generator/space_file/worktable_impls.rs b/codegen/src/persist_table/generator/space_file/worktable_impls.rs index c6d95ee5..579c24a2 100644 --- a/codegen/src/persist_table/generator/space_file/worktable_impls.rs +++ b/codegen/src/persist_table/generator/space_file/worktable_impls.rs @@ -10,12 +10,14 @@ 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 } } } @@ -23,17 +25,35 @@ impl Generator { 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(); diff --git a/src/lib.rs b/src/lib.rs index 59f3f289..e1d4dbe5 100644 --- a/src/lib.rs +++ b/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}; diff --git a/src/persistence/error.rs b/src/persistence/error.rs new file mode 100644 index 00000000..f8b48d5e --- /dev/null +++ b/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 = Result>; + +/// Observable state of the persistence worker. +#[derive(Clone, Debug)] +pub enum PersistenceState { + Running, + Closing, + Failed(Arc), + Closed, +} diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs index 681fbd89..3cbaefaf 100644 --- a/src/persistence/mod.rs +++ b/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; diff --git a/src/persistence/operation/batch.rs b/src/persistence/operation/batch.rs index 7d6e2e9c..02d4dc84 100644 --- a/src/persistence/operation/batch.rs +++ b/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); diff --git a/src/persistence/task.rs b/src/persistence/task.rs index 7fefe41c..30a35f05 100644 --- a/src/persistence/task.rs +++ b/src/persistence/task.rs @@ -9,10 +9,11 @@ use std::time::Duration; use data_bucket::page::PageId; use parking_lot::Mutex as ParkingMutex; use tokio::sync::Notify; +use tokio::task::JoinHandle; use worktable_codegen::worktable; -use crate::persistence::PersistenceEngine; use crate::persistence::operation::{BatchInnerRow, BatchInnerWorkTable, BatchOperation, OperationId, PosByOpIdQuery}; +use crate::persistence::{PersistenceEngine, PersistenceError, PersistenceResult, PersistenceState}; use crate::prelude::*; use crate::util::OptimizedVec; use crate::vacuum::VacuumPersistence; @@ -35,6 +36,70 @@ worktable! ( const MAX_PAGE_AMOUNT: usize = 16; +#[derive(Debug)] +struct PersistenceLifecycle { + state: ParkingMutex, + notify: Notify, +} + +impl PersistenceLifecycle { + fn new() -> Self { + Self { + state: ParkingMutex::new(PersistenceState::Running), + notify: Notify::new(), + } + } + + fn state(&self) -> PersistenceState { + self.state.lock().clone() + } + + fn begin_close(&self) -> PersistenceResult { + let mut state = self.state.lock(); + match &*state { + PersistenceState::Running => { + *state = PersistenceState::Closing; + self.notify.notify_waiters(); + Ok(()) + } + PersistenceState::Closing => Ok(()), + PersistenceState::Failed(error) => Err(error.clone()), + PersistenceState::Closed => Ok(()), + } + } + + fn finish_close(&self) { + let mut state = self.state.lock(); + if matches!(*state, PersistenceState::Closing) { + *state = PersistenceState::Closed; + } + self.notify.notify_waiters(); + } + + fn fail(&self, report: eyre::Report) -> Arc { + let mut state = self.state.lock(); + let error = match &*state { + PersistenceState::Failed(error) => error.clone(), + _ => { + let error = Arc::new(PersistenceError::Engine(report)); + *state = PersistenceState::Failed(error.clone()); + error + } + }; + self.notify.notify_waiters(); + error + } + + fn ensure_running(&self) -> PersistenceResult { + match self.state() { + PersistenceState::Running => Ok(()), + PersistenceState::Closing => Err(Arc::new(PersistenceError::Closing)), + PersistenceState::Closed => Err(Arc::new(PersistenceError::Closed)), + PersistenceState::Failed(error) => Err(error), + } + } +} + pub struct QueueAnalyzer { operations: OptimizedVec>, queue_inner_wt: Arc, @@ -271,6 +336,151 @@ where } } +#[cfg(test)] +mod lifecycle_tests { + use std::collections::HashMap; + use std::sync::atomic::{AtomicUsize, Ordering}; + + use super::*; + + #[derive(Clone, Debug, Default)] + struct TestConfig; + + #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] + enum TestIndex {} + + #[derive(Clone, Debug, Default)] + struct TestEvents; + + impl TableSecondaryIndexEventsOps for TestEvents { + fn extend(&mut self, _another: Self) {} + + fn remove(&mut self, _another: &Self) {} + + fn last_evs(&self) -> HashMap> { + HashMap::new() + } + + fn first_evs(&self) -> HashMap> { + HashMap::new() + } + + fn iter_event_ids(&self) -> impl Iterator { + std::iter::empty() + } + + fn sort(&mut self) {} + + fn validate(&mut self) -> Self { + Self + } + + fn is_empty(&self) -> bool { + true + } + + fn is_unit() -> bool { + true + } + } + + impl PersistenceConfig for TestConfig { + fn table_path(&self) -> &str { + "" + } + + fn version(&self) -> u32 { + 0 + } + } + + struct TestEngine { + batches: Arc, + config: TestConfig, + fail: bool, + } + + impl PersistenceEngine<(), u64, TestEvents, TestIndex> for TestEngine { + type Config = TestConfig; + + async fn new(config: Self::Config) -> eyre::Result { + Ok(Self { + batches: Arc::new(AtomicUsize::new(0)), + config, + fail: false, + }) + } + + async fn apply_operation(&mut self, _op: Operation<(), u64, TestEvents>) -> eyre::Result<()> { + Ok(()) + } + + async fn apply_batch_operation( + &mut self, + _batch_op: BatchOperation<(), u64, TestEvents, TestIndex>, + ) -> eyre::Result<()> { + if self.fail { + return Err(eyre::eyre!("injected batch failure")); + } + self.batches.fetch_add(1, Ordering::Relaxed); + Ok(()) + } + + fn config(&self) -> &Self::Config { + &self.config + } + } + + fn insert_operation(id: u128) -> Operation<(), u64, TestEvents> { + Operation::Insert(InsertOperation { + id: OperationId::Single(uuid::Uuid::from_u128(id)), + pk_gen_state: (), + primary_key_events: vec![], + secondary_keys_events: TestEvents, + bytes: vec![id as u8], + link: Link { + page_id: 1.into(), + offset: id as u32, + length: 1, + }, + }) + } + + #[tokio::test] + async fn close_drains_and_joins_the_engine() { + let batches = Arc::new(AtomicUsize::new(0)); + let task = PersistenceTask::run_engine(TestEngine { + batches: batches.clone(), + config: TestConfig, + fail: false, + }); + + task.apply_operation(insert_operation(1)).unwrap(); + task.close().await.unwrap(); + + assert_eq!(batches.load(Ordering::Relaxed), 1); + } + + #[tokio::test] + async fn engine_failure_is_terminal_and_reused_for_later_callers() { + let task = PersistenceTask::run_engine(TestEngine { + batches: Arc::new(AtomicUsize::new(0)), + config: TestConfig, + fail: true, + }); + + task.apply_operation(insert_operation(1)).unwrap(); + let wait_error = task.wait_for_ops().await.unwrap_err(); + assert!(wait_error.to_string().contains("injected batch failure")); + + let intake_error = task.apply_operation(insert_operation(2)).unwrap_err(); + assert!(Arc::ptr_eq(&wait_error, &intake_error)); + + let close_error = task.close().await.unwrap_err(); + assert!(Arc::ptr_eq(&wait_error, &close_error)); + } +} + #[derive(Debug)] pub struct Queue { // Not `lockfree::queue::Queue`: its `Removable::empty` materializes the @@ -283,21 +493,31 @@ pub struct Queue { // 65_536 queued operations, making the wait triggers see an "empty" // queue that still holds work. len: Arc, + lifecycle: Arc, } impl Queue { - pub fn new() -> Self { + fn new(lifecycle: Arc) -> Self { Self { queue: ParkingMutex::new(VecDeque::new()), notify: Notify::new(), len: Arc::new(AtomicUsize::new(0)), + lifecycle, } } - pub fn push(&self, value: Operation) { + pub fn push(&self, value: Operation) -> PersistenceResult { + let state = self.lifecycle.state.lock(); + match &*state { + PersistenceState::Running => {} + PersistenceState::Closing => return Err(Arc::new(PersistenceError::Closing)), + PersistenceState::Closed => return Err(Arc::new(PersistenceError::Closed)), + PersistenceState::Failed(error) => return Err(error.clone()), + } self.len.fetch_add(1, Ordering::Release); self.queue.lock().push_back(value); self.notify.notify_one(); + Ok(()) } /// Pops the next operation, marking `in_progress` `true` before the queue @@ -308,23 +528,32 @@ impl Queue Operation { + ) -> Option> { loop { + let notified = self.notify.notified(); // Drain values { let mut queue = self.queue.lock(); if let Some(value) = queue.pop_front() { in_progress.store(true, Ordering::Release); self.len.fetch_sub(1, Ordering::Release); - return value; + return Some(value); } } + if !matches!(self.lifecycle.state(), PersistenceState::Running) { + return None; + } + // Wait for values to be available - self.notify.notified().await; + notified.await; } } + fn wake(&self) { + self.notify.notify_waiters(); + } + pub fn immediate_pop(&self) -> Option> { if let Some(v) = self.queue.lock().pop_front() { self.len.fetch_sub(1, Ordering::Release); @@ -356,24 +585,24 @@ where new_link: Link, primary_key_events: Vec>>, secondary_keys_events: SecondaryKeys, - ) { + ) -> PersistenceResult { self.push(Operation::Update(UpdateOperation { id: OperationId::Single(uuid::Uuid::now_v7()), primary_key_events, secondary_keys_events, bytes, link: new_link, - })); + })) } } #[derive(Debug)] pub struct PersistenceTask { - engine_task_handle: tokio::task::AbortHandle, + engine_task_handle: Option>, queue: Arc>, analyzer_inner_wt: Arc, analyzer_in_progress: Arc, - progress_notify: Arc, + lifecycle: Arc, phantom_data: PhantomData, } @@ -396,8 +625,20 @@ impl Drop /// `close()` lifecycle (drain, join, surface terminal errors) is the /// long-term replacement for this heuristic. fn drop(&mut self) { + let Some(handle) = self.engine_task_handle.as_ref() else { + return; + }; + if handle.is_finished() { + return; + } + if matches!( + self.lifecycle.state(), + PersistenceState::Failed(_) | PersistenceState::Closed + ) { + return; + } if self.check_wait_triggers() { - self.engine_task_handle.abort(); + handle.abort(); } else { tracing::error!( "PersistenceTask dropped with work in flight; the engine task keeps running detached. Call wait_for_ops() before dropping to guarantee a clean shutdown." @@ -409,8 +650,16 @@ impl Drop impl PersistenceTask { - pub fn apply_operation(&self, op: Operation) { - self.queue.push(op); + pub fn apply_operation(&self, op: Operation) -> PersistenceResult { + self.queue.push(op) + } + + pub fn ensure_running(&self) -> PersistenceResult { + self.lifecycle.ensure_running() + } + + pub fn state(&self) -> PersistenceState { + self.lifecycle.state() } /// Returns a sink that lets vacuum queue persistence operations for row @@ -432,11 +681,11 @@ impl PrimaryKey: Clone + Debug + Send + Sync + 'static, AvailableIndexes: Copy + Clone + Debug + Hash + Eq + Send + Sync + 'static, { - let queue = Arc::new(Queue::new()); - let progress_notify = Arc::new(Notify::new()); + let lifecycle = Arc::new(PersistenceLifecycle::new()); + let queue = Arc::new(Queue::new(lifecycle.clone())); let engine_queue = queue.clone(); - let engine_progress_notify = progress_notify.clone(); + let engine_lifecycle = lifecycle.clone(); let analyzer_inner_wt: Arc = Default::default(); let mut analyzer = QueueAnalyzer::new(analyzer_inner_wt.clone()); let analyzer_in_progress = Arc::new(AtomicBool::new(true)); @@ -448,31 +697,39 @@ impl Some(next_op) } else if analyzer.len() == 0 { task_analyzer_in_progress.store(false, Ordering::Release); - engine_progress_notify.notify_waiters(); + engine_lifecycle.notify.notify_waiters(); + if matches!(engine_lifecycle.state(), PersistenceState::Closing) { + engine_lifecycle.finish_close(); + return; + } // The pop sets the flag back to `true` atomically with the // dequeue, so waiters never observe an empty queue with an // idle analyzer while an operation is in flight. - Some(engine_queue.pop_marking_in_progress(&task_analyzer_in_progress).await) + engine_queue.pop_marking_in_progress(&task_analyzer_in_progress).await } else { None }; if let Some(op) = op && let Err(err) = analyzer.push(op.clone()) { - tracing::warn!("Error while feeding data to analyzer: {}", err); + engine_lifecycle.fail(err); + return; } let ops_available_iter = engine_queue.pop_iter(); if let Err(err) = analyzer.extend_from_iter(ops_available_iter) { - tracing::warn!("Error while feeding data to analyzer: {}", err); + engine_lifecycle.fail(err); + return; } if let Some(op_id) = analyzer.get_first_op_id_available() { let batch_op = analyzer.collect_batch_from_op_id(op_id).await; if let Err(e) = batch_op { - tracing::warn!("Error collecting batch operation: {}", e); + engine_lifecycle.fail(e); + return; } else if let Some(batch_op) = batch_op.unwrap() { let res = engine.apply_batch_operation(batch_op).await; if let Err(e) = res { - tracing::warn!("Persistence engine failed while applying batch op: {}", e); + engine_lifecycle.fail(e); + return; } } else { tokio::time::sleep(Duration::from_millis(500)).await; @@ -480,13 +737,13 @@ impl } } }; - let engine_task_handle = tokio::spawn(task).abort_handle(); + let engine_task_handle = tokio::spawn(task); Self { queue, - engine_task_handle, + engine_task_handle: Some(engine_task_handle), analyzer_inner_wt, analyzer_in_progress, - progress_notify, + lifecycle, phantom_data: PhantomData, } } @@ -504,8 +761,22 @@ impl true } - pub async fn wait_for_ops(&self) { - while !self.check_wait_triggers() { + pub async fn wait_for_ops(&self) -> PersistenceResult { + loop { + match self.lifecycle.state() { + PersistenceState::Failed(error) => return Err(error), + PersistenceState::Closed => return Ok(()), + PersistenceState::Running if self.check_wait_triggers() => return Ok(()), + PersistenceState::Running | PersistenceState::Closing => {} + } + + if self.engine_task_handle.as_ref().is_some_and(JoinHandle::is_finished) { + let error = self + .lifecycle + .fail(eyre::eyre!("persistence engine task terminated unexpectedly")); + return Err(error); + } + let queue_count = self.queue.len(); let analyzer_count = self.analyzer_inner_wt.count(); let count = queue_count + analyzer_count; @@ -516,9 +787,30 @@ impl } tokio::select! { - _ = self.progress_notify.notified() => {}, + _ = self.lifecycle.notify.notified() => {}, _ = tokio::time::sleep(Duration::from_secs(1)) => {} } } } + + pub async fn close(mut self) -> PersistenceResult { + let begin_result = self.lifecycle.begin_close(); + self.queue.wake(); + + if let Some(handle) = self.engine_task_handle.take() + && let Err(error) = handle.await + { + return Err(self + .lifecycle + .fail(eyre::eyre!("persistence engine task failed to join: {error}"))); + } + + match self.lifecycle.state() { + PersistenceState::Closed => begin_result, + PersistenceState::Failed(error) => Err(error), + PersistenceState::Running | PersistenceState::Closing => Err(self + .lifecycle + .fail(eyre::eyre!("persistence engine task exited without a terminal state"))), + } + } } diff --git a/src/table/mod.rs b/src/table/mod.rs index 265c270e..aabf6d28 100644 --- a/src/table/mod.rs +++ b/src/table/mod.rs @@ -540,4 +540,6 @@ pub enum WorkTableError { SecondaryIndexError, PrimaryUpdateTry, PagesError(in_memory::PagesExecutionError), + #[display("{}", _0)] + PersistenceError(#[error(not(source))] std::sync::Arc), } diff --git a/src/table/vacuum/mod.rs b/src/table/vacuum/mod.rs index 21ab55e3..bcfb4148 100644 --- a/src/table/vacuum/mod.rs +++ b/src/table/vacuum/mod.rs @@ -4,6 +4,7 @@ use data_bucket::Link; use indexset::cdc::change::ChangeEvent; use indexset::core::pair::Pair; +use crate::persistence::PersistenceResult; use crate::vacuum::fragmentation_info::FragmentationInfo; mod fragmentation_info; @@ -36,7 +37,7 @@ pub trait VacuumPersistence: Send + Sync { new_link: Link, primary_key_events: Vec>>, secondary_keys_events: SecondaryEvents, - ); + ) -> PersistenceResult; } /// Trait for unifying different [`WorkTable`] related [`EmptyDataVacuum`]'s. diff --git a/src/table/vacuum/vacuum.rs b/src/table/vacuum/vacuum.rs index 0e62bcb3..5ff3f0d7 100644 --- a/src/table/vacuum/vacuum.rs +++ b/src/table/vacuum/vacuum.rs @@ -129,7 +129,7 @@ where self } - async fn defragment(&self) -> VacuumStats { + async fn defragment(&self) -> eyre::Result { let now = Instant::now(); let registry = self.data_pages.empty_links_registry(); @@ -164,7 +164,7 @@ where } else { unreachable!("I hope so") }; - match self.move_data_from(page_from, page_to).await { + match self.move_data_from(page_from, page_to).await? { (true, true) => { // from moved fully and on to no more space free_pages.push_back(page_from); @@ -198,19 +198,19 @@ where self.data_pages.mark_page_full(id) } - VacuumStats { + Ok(VacuumStats { pages_processed, pages_freed, bytes_freed: initial_bytes_freed, duration_ns: now.elapsed().as_nanos(), - } + }) } fn free_page(&self, page_id: PageId) { self.data_pages.reset_page(page_id).expect("should exist as called") } - async fn move_data_from(&self, from: PageId, to: PageId) -> (bool, bool) { + async fn move_data_from(&self, from: PageId, to: PageId) -> eyre::Result<(bool, bool)> { let to_page = self.data_pages.get_page(to).expect("should exist as link exists"); let to_free_space = to_page.free_space(); @@ -271,14 +271,14 @@ where .move_row_for_vacuum(from_link.0, to) .expect("links and destination capacity were checked") }; - self.update_index_after_move(pk.clone(), from_link.0, new_link, raw_data); + self.update_index_after_move(pk.clone(), from_link.0, new_link, raw_data)?; self.data_pages.retire_published_link(from_link.0); lock.unlock(); self.lock_manager.remove_with_lock_check(&pk); } - (from_page_will_be_moved, to_page_will_be_filled) + Ok((from_page_will_be_moved, to_page_will_be_filled)) } async fn full_row_lock(&self, pk: &PrimaryKey) -> Arc { @@ -294,7 +294,13 @@ where op_lock } - fn update_index_after_move(&self, pk: PrimaryKey, old_link: Link, new_link: Link, raw_data: Vec) { + fn update_index_after_move( + &self, + pk: PrimaryKey, + old_link: Link, + new_link: Link, + raw_data: Vec, + ) -> eyre::Result<()> { let row = self .data_pages .select(new_link) @@ -309,13 +315,14 @@ where .reinsert_row_cdc(row.clone(), old_link, row, new_link); res.expect("should be ok as index were no violated"); let (_, primary_key_events) = self.primary_index.insert_cdc(pk.clone(), new_link); - persistence.apply_move(raw_data, new_link, primary_key_events, secondary_keys_events); + persistence.apply_move(raw_data, new_link, primary_key_events, secondary_keys_events)?; } else { self.secondary_indexes .reinsert_row(row.clone(), old_link, row, new_link) .expect("should be ok as index were no violated"); self.primary_index.insert(pk.clone(), new_link); } + Ok(()) } } @@ -378,7 +385,7 @@ where } async fn vacuum(&self) -> eyre::Result { - Ok(self.defragment().await) + self.defragment().await } } @@ -454,7 +461,7 @@ mod tests { table.delete(first_two_ids[1]).await.unwrap(); let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); for (id, expected) in ids.into_iter().skip(2) { let row = table.select(id); @@ -485,7 +492,7 @@ mod tests { table.delete(ids_to_delete[1]).await.unwrap(); let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); for (id, expected) in ids .into_iter() @@ -519,7 +526,7 @@ mod tests { table.delete(last_two_ids[0]).await.unwrap(); let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); for (id, expected) in ids .into_iter() @@ -554,7 +561,7 @@ mod tests { } let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); for (id, expected) in ids.into_iter().filter(|(i, _)| !ids_to_delete.contains(i)) { let row = table.select(id); @@ -586,7 +593,7 @@ mod tests { } let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); let row = table.select(remaining_id); assert_eq!(row, Some(ids[0].1.clone())); @@ -612,7 +619,7 @@ mod tests { table.delete(ids.last().unwrap().0).await.unwrap(); let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); for (id, expected) in ids.into_iter().take(4) { let row = table.select(id); @@ -654,7 +661,7 @@ mod tests { } let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); for (id, expected) in ids.into_iter().filter(|(i, _)| !ids_to_delete.contains(i)) { let row = table.select(id); @@ -685,7 +692,7 @@ mod tests { } let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); let mut new_ids = HashMap::new(); for i in 0..3 { @@ -735,7 +742,7 @@ mod tests { } let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); for (id, expected) in ids.into_iter().filter(|(i, _)| !ids_to_delete.contains(i)) { let row = table.select(id); @@ -767,7 +774,7 @@ mod tests { } let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); for (id, expected) in ids.into_iter().filter(|(id, _)| !ids_to_delete.contains(id)) { let row = table.select(id); @@ -796,7 +803,7 @@ mod tests { table.delete(ids.last().unwrap().0).await.unwrap(); let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); for (id, expected) in ids.into_iter().take(499) { let row = table.select(id); @@ -830,7 +837,7 @@ mod tests { } let vacuum = create_vacuum(&table); - vacuum.defragment().await; + vacuum.defragment().await.unwrap(); assert!(!table.0.data.get_empty_pages().is_empty()); diff --git a/tests/migration/mod.rs b/tests/migration/mod.rs index c1818cec..080f5216 100644 --- a/tests/migration/mod.rs +++ b/tests/migration/mod.rs @@ -128,7 +128,7 @@ fn test_migrate_v1_to_current() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } // Verify source data is readable @@ -211,7 +211,7 @@ fn test_migrate_v2_to_current() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } let ctx = UserMigrationContext { @@ -280,7 +280,7 @@ fn test_next_pk_and_indexes_after_migration() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } let ctx = UserMigrationContext { @@ -308,7 +308,7 @@ fn test_next_pk_and_indexes_after_migration() { }; table.insert(inserted.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); assert_eq!(table.count(), 3); assert_eq!(table.select(inserted.id), Some(inserted.clone())); diff --git a/tests/persistence/bulk_load_stall.rs b/tests/persistence/bulk_load_stall.rs index cf55b028..665ea2ce 100644 --- a/tests/persistence/bulk_load_stall.rs +++ b/tests/persistence/bulk_load_stall.rs @@ -88,7 +88,8 @@ fn test_bulk_insert_delete_persistence() { timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled on bulk insert+delete"); + .expect("persistence stalled on bulk insert+delete") + .expect("persistence engine failed"); for id in &deleted { rows.remove(id); diff --git a/tests/persistence/concurrent/mod.rs b/tests/persistence/concurrent/mod.rs index 22f68ac9..ecec914f 100644 --- a/tests/persistence/concurrent/mod.rs +++ b/tests/persistence/concurrent/mod.rs @@ -105,7 +105,7 @@ fn test_concurrent() { let _ = h.await; } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { let engine = TestConcurrentPersistenceEngine::new(config.clone()).await.unwrap(); diff --git a/tests/persistence/duplicate_key_index_reload.rs b/tests/persistence/duplicate_key_index_reload.rs index ee6058d0..dfd995fd 100644 --- a/tests/persistence/duplicate_key_index_reload.rs +++ b/tests/persistence/duplicate_key_index_reload.rs @@ -216,7 +216,8 @@ fn test_duplicate_key_secondary_index_survives_reload() { timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled on bulk insert"); + .expect("persistence stalled on bulk insert") + .expect("persistence engine failed"); } assert_straddling_topology(dir).await; @@ -277,7 +278,8 @@ fn test_duplicate_key_secondary_index_survives_reload() { timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled on post-reload mutations"); + .expect("persistence stalled on post-reload mutations") + .expect("persistence engine failed"); } { let engine = DuplicateKeyReloadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -328,7 +330,8 @@ fn test_single_key_all_duplicates_survives_reload() { timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled on bulk insert"); + .expect("persistence stalled on bulk insert") + .expect("persistence engine failed"); } { let engine = DuplicateKeyReloadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -360,7 +363,8 @@ fn test_single_key_all_duplicates_survives_reload() { .unwrap(); timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled on post-reload insert"); + .expect("persistence stalled on post-reload insert") + .expect("persistence engine failed"); assert_eq!(table.select_by_score(42).execute().unwrap().len() as u64, ROWS + 1); } }) @@ -410,7 +414,8 @@ fn test_duplicate_key_mutations_without_reload() { } timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled on bulk insert"); + .expect("persistence stalled on bulk insert") + .expect("persistence engine failed"); for j in 0..500u64 { let id = 20_000 + j; @@ -457,6 +462,7 @@ fn test_duplicate_key_mutations_without_reload() { timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled on mutations without any reload"); + .expect("persistence stalled on mutations without any reload") + .expect("persistence engine failed"); }) } diff --git a/tests/persistence/failure/insert.rs b/tests/persistence/failure/insert.rs index d6011ea0..7273a899 100644 --- a/tests/persistence/failure/insert.rs +++ b/tests/persistence/failure/insert.rs @@ -25,7 +25,7 @@ fn test_insert_two_indexes_first_fail() { unique_b: 200, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.unique_a }; @@ -67,7 +67,9 @@ fn test_insert_two_indexes_first_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!( wait_result.is_ok(), "BUG: persistence blocked after insert failure on first index!" @@ -86,7 +88,7 @@ fn test_insert_two_indexes_first_fail() { unique_b: 4001, }; assert!(table.insert(new_row).is_ok()); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -115,7 +117,7 @@ fn test_insert_two_indexes_second_fail() { unique_b: 200, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.unique_b }; @@ -157,7 +159,9 @@ fn test_insert_two_indexes_second_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!( wait_result.is_ok(), "BUG: persistence blocked after insert failure on second index!" @@ -175,7 +179,7 @@ fn test_insert_two_indexes_second_fail() { unique_b: 301, }; assert!(table.insert(new_row).is_ok(), "BUG: orphaned entry in unique_a_idx!"); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -205,7 +209,7 @@ fn test_insert_three_indexes_first_fail() { unique_c: 300, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.unique_a }; @@ -250,7 +254,9 @@ fn test_insert_three_indexes_first_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -266,7 +272,7 @@ fn test_insert_three_indexes_first_fail() { unique_c: 4002, }; assert!(table.insert(new_row).is_ok()); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -296,7 +302,7 @@ fn test_insert_three_indexes_middle_fail() { unique_c: 300, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.unique_b }; @@ -341,7 +347,9 @@ fn test_insert_three_indexes_middle_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -357,7 +365,7 @@ fn test_insert_three_indexes_middle_fail() { unique_c: 500, }; assert!(table.insert(new_row).is_ok(), "BUG: orphaned entry in unique_a_idx!"); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -387,7 +395,7 @@ fn test_insert_three_indexes_last_fail() { unique_c: 300, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.unique_c }; @@ -432,7 +440,9 @@ fn test_insert_three_indexes_last_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -448,7 +458,7 @@ fn test_insert_three_indexes_last_fail() { unique_c: 301, }; assert!(table.insert(new_row).is_ok(), "BUG: orphaned entries in indexes!"); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -476,7 +486,7 @@ fn test_insert_primary_duplicate() { data: 100, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -513,7 +523,9 @@ fn test_insert_primary_duplicate() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok()); } @@ -524,7 +536,7 @@ fn test_insert_primary_duplicate() { let original = table.select(existing_pk).unwrap(); assert_eq!(original.data, 100); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } diff --git a/tests/persistence/failure/reinsert.rs b/tests/persistence/failure/reinsert.rs index 19bbeab7..0d889167 100644 --- a/tests/persistence/failure/reinsert.rs +++ b/tests/persistence/failure/reinsert.rs @@ -25,7 +25,7 @@ fn test_reinsert_pk_mismatch() { unique_b: 200, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -66,7 +66,9 @@ fn test_reinsert_pk_mismatch() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok()); } @@ -78,7 +80,7 @@ fn test_reinsert_pk_mismatch() { let original = table.select(existing_pk).unwrap(); assert_eq!(original.unique_a, 100); assert_eq!(original.unique_b, 200); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -114,7 +116,7 @@ fn test_reinsert_two_indexes_first_fail() { unique_b: 400, }; table.insert(row2.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (row1.id, row2.id, row1.unique_a) }; @@ -156,7 +158,9 @@ fn test_reinsert_two_indexes_first_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked after reinsert failure!"); } @@ -174,7 +178,7 @@ fn test_reinsert_two_indexes_first_fail() { assert_eq!(row2.unique_a, 300); assert_eq!(row2.unique_b, 400); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -210,7 +214,7 @@ fn test_reinsert_two_indexes_second_fail() { unique_b: 400, }; table.insert(row2.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row1.unique_b }; @@ -254,7 +258,9 @@ fn test_reinsert_two_indexes_second_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -269,7 +275,7 @@ fn test_reinsert_two_indexes_second_fail() { unique_b: 600, }; assert!(table.insert(new_row).is_ok(), "BUG: orphaned entry in unique_a_idx!"); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -307,7 +313,7 @@ fn test_reinsert_three_indexes_first_fail() { unique_c: 600, }; table.insert(row2.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row1.unique_a }; @@ -355,7 +361,9 @@ fn test_reinsert_three_indexes_first_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -368,7 +376,7 @@ fn test_reinsert_three_indexes_first_fail() { assert_eq!(row2.unique_a, 400); assert_eq!(row2.unique_b, 500); assert_eq!(row2.unique_c, 600); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -406,7 +414,7 @@ fn test_reinsert_three_indexes_middle_fail() { unique_c: 600, }; table.insert(row2.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row1.unique_b }; @@ -454,7 +462,9 @@ fn test_reinsert_three_indexes_middle_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -470,7 +480,7 @@ fn test_reinsert_three_indexes_middle_fail() { unique_c: 1000, }; assert!(table.insert(new_row).is_ok(), "BUG: orphaned entry!"); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -508,7 +518,7 @@ fn test_reinsert_three_indexes_last_fail() { unique_c: 600, }; table.insert(row2.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row1.unique_c }; @@ -556,7 +566,9 @@ fn test_reinsert_three_indexes_last_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -572,7 +584,7 @@ fn test_reinsert_three_indexes_last_fail() { unique_c: 900, }; assert!(table.insert(new_row).is_ok(), "BUG: orphaned entries!"); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } diff --git a/tests/persistence/failure/update.rs b/tests/persistence/failure/update.rs index 80756abe..9e50c0e0 100644 --- a/tests/persistence/failure/update.rs +++ b/tests/persistence/failure/update.rs @@ -33,7 +33,7 @@ fn test_update_unique_secondary_conflict() { unique_b: 400, }; table.insert(row2.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row1.id }; @@ -74,7 +74,9 @@ fn test_update_unique_secondary_conflict() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -86,7 +88,7 @@ fn test_update_unique_secondary_conflict() { let row1 = table.select(row1_pk).unwrap(); assert_eq!(row1.unique_a, 100); assert_eq!(row1.unique_b, 200); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -115,7 +117,7 @@ fn test_update_pk_based_success() { unique_b: 200, }; table.insert(row1.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row1.id }; @@ -156,7 +158,9 @@ fn test_update_pk_based_success() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok()); } @@ -168,7 +172,7 @@ fn test_update_pk_based_success() { let row1 = table.select(row1_pk).unwrap(); assert_eq!(row1.unique_a, 150); assert_eq!(row1.unique_b, 250); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } diff --git a/tests/persistence/failure/update_non_unique.rs b/tests/persistence/failure/update_non_unique.rs index c756d889..36e7b8b5 100644 --- a/tests/persistence/failure/update_non_unique.rs +++ b/tests/persistence/failure/update_non_unique.rs @@ -43,7 +43,7 @@ fn test_update_non_unique_middle_fail() { data: 300, }; table.insert(row3.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (row1.id, row2.id, row3.id) }; @@ -83,7 +83,9 @@ fn test_update_non_unique_middle_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -95,7 +97,7 @@ fn test_update_non_unique_middle_fail() { assert!(table.select(row1_pk).is_some()); assert!(table.select(row2_pk).is_some()); assert!(table.select(row3_pk).is_some()); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -149,7 +151,7 @@ fn test_update_non_unique_last_fail() { data: 300, }; table.insert(row3.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); conflict_row.id }; @@ -189,7 +191,9 @@ fn test_update_non_unique_last_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -200,7 +204,7 @@ fn test_update_non_unique_last_fail() { let conflict = table.select(conflict_pk).unwrap(); assert_eq!(conflict.unique_value, 99); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } diff --git a/tests/persistence/failure/update_unsized.rs b/tests/persistence/failure/update_unsized.rs index ccfe78c8..8c860587 100644 --- a/tests/persistence/failure/update_unsized.rs +++ b/tests/persistence/failure/update_unsized.rs @@ -43,7 +43,7 @@ fn test_update_unsized_same_size() { name: "ccc".to_string(), }; table.insert(row3.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (row1.id, row2.id, row3.id) }; @@ -86,7 +86,9 @@ fn test_update_unsized_same_size() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -98,7 +100,7 @@ fn test_update_unsized_same_size() { assert!(table.select(row1_pk).is_some()); assert!(table.select(row2_pk).is_some()); assert!(table.select(row3_pk).is_some()); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -128,7 +130,7 @@ fn test_update_unsized_larger_all_success() { name: "a".to_string(), }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -171,7 +173,9 @@ fn test_update_unsized_larger_all_success() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok()); } @@ -182,7 +186,7 @@ fn test_update_unsized_larger_all_success() { let row = table.select(row_pk).unwrap(); assert_eq!(row.unique_value, 20); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -236,7 +240,7 @@ fn test_update_unsized_larger_middle_fail() { name: "c".to_string(), }; table.insert(row3.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (conflict.id, row2.id, row3.id) }; @@ -280,7 +284,9 @@ fn test_update_unsized_larger_middle_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -299,7 +305,7 @@ fn test_update_unsized_larger_middle_fail() { let conflict = table.select(conflict_pk).unwrap(); assert_eq!(conflict.unique_value, 99); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } @@ -337,7 +343,7 @@ fn test_update_unsized_larger_last_fail() { name: "b".to_string(), }; table.insert(row2.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (row1.id, row2.id) }; @@ -381,7 +387,9 @@ fn test_update_unsized_larger_last_fail() { }; table.insert(valid_row3).unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!(wait_result.is_ok(), "BUG: persistence blocked!"); } @@ -399,7 +407,7 @@ fn test_update_unsized_larger_last_fail() { let row1 = table.select(row1_pk).unwrap(); assert_eq!(row1.name, "larger".to_string()); assert_eq!(row1.unique_value, 99); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } diff --git a/tests/persistence/loaded_index_growth.rs b/tests/persistence/loaded_index_growth.rs index f6ab4e2d..db353cea 100644 --- a/tests/persistence/loaded_index_growth.rs +++ b/tests/persistence/loaded_index_growth.rs @@ -97,7 +97,8 @@ fn test_primary_index_grows_on_a_loaded_table() { } timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled building the initial store"); + .expect("persistence stalled building the initial store") + .expect("persistence engine failed"); } let idx_when_loaded = primary_idx_size(dir); @@ -115,7 +116,8 @@ fn test_primary_index_grows_on_a_loaded_table() { } timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled appending to the loaded store"); + .expect("persistence stalled appending to the loaded store") + .expect("persistence engine failed"); } let idx_after_appends = primary_idx_size(dir); @@ -165,7 +167,8 @@ fn test_primary_index_grows_on_a_loaded_table() { table.insert(row(ROWS_BEFORE_RELOAD + ROWS_AFTER_RELOAD)).unwrap(); timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled on the post-reload insert"); + .expect("persistence stalled on the post-reload insert") + .expect("persistence engine failed"); } }) } diff --git a/tests/persistence/s3/mod.rs b/tests/persistence/s3/mod.rs index 45acfa7e..9b5a7f3c 100644 --- a/tests/persistence/s3/mod.rs +++ b/tests/persistence/s3/mod.rs @@ -53,7 +53,7 @@ fn test_s3_engine_compiles() { }) .unwrap(); assert!(!table.select_all().execute().unwrap().is_empty()); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } }); } diff --git a/tests/persistence/sync/failure.rs b/tests/persistence/sync/failure.rs index 0ae622b5..4a20f7ee 100644 --- a/tests/persistence/sync/failure.rs +++ b/tests/persistence/sync/failure.rs @@ -35,7 +35,7 @@ fn test_failed_update_by_pk_doesnt_corrupt_persistence() { table.insert(row.clone()).unwrap(); pks.push(row.id); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); pks }; @@ -60,7 +60,7 @@ fn test_failed_update_by_pk_doesnt_corrupt_persistence() { .await .unwrap(); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { @@ -108,7 +108,7 @@ fn test_failed_update_by_unique_index_doesnt_corrupt_persistence() { table.insert(row.clone()).unwrap(); pks.push(row.id); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); pks }; @@ -133,7 +133,7 @@ fn test_failed_update_by_unique_index_doesnt_corrupt_persistence() { .await .unwrap(); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { @@ -181,7 +181,7 @@ fn test_failed_delete_by_pk_doesnt_corrupt_persistence() { table.insert(row.clone()).unwrap(); pks.push(row.id); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); pks }; @@ -193,7 +193,7 @@ fn test_failed_delete_by_pk_doesnt_corrupt_persistence() { assert!(result.is_err()); assert!(matches!(result.unwrap_err(), WorkTableError::NotFound)); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { diff --git a/tests/persistence/sync/failure_multi_index.rs b/tests/persistence/sync/failure_multi_index.rs index e1997b2a..469e87f1 100644 --- a/tests/persistence/sync/failure_multi_index.rs +++ b/tests/persistence/sync/failure_multi_index.rs @@ -57,7 +57,7 @@ fn test_multi_index_insert_failure_doesnt_corrupt_persistence() { unique_b: 0, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -113,7 +113,9 @@ fn test_multi_index_insert_failure_doesnt_corrupt_persistence() { // Use timeout to detect if persistence is stuck // If this hangs, the bug exists - CDC queue is blocked - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); if wait_result.is_err() { panic!( diff --git a/tests/persistence/sync/many_strings.rs b/tests/persistence/sync/many_strings.rs index 88381285..4f266bb4 100644 --- a/tests/persistence/sync/many_strings.rs +++ b/tests/persistence/sync/many_strings.rs @@ -51,7 +51,7 @@ fn test_space_update_query_pk_sync() { id: "Some string before 2".to_string(), }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -64,7 +64,7 @@ fn test_space_update_query_pk_sync() { another: 0, }; table.update_field_another_by_id(q, pk.clone()).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { let engine = TestSyncPersistenceEngine::new(config.clone()).await.unwrap(); @@ -109,7 +109,7 @@ fn test_space_update_query_pk_many_times_sync() { id: "Some string before 2".to_string(), }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -125,7 +125,7 @@ fn test_space_update_query_pk_many_times_sync() { table.update_field_another_by_id(q, pk.clone()).await.unwrap(); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { let engine = TestSyncPersistenceEngine::new(config.clone()).await.unwrap(); diff --git a/tests/persistence/sync/mod.rs b/tests/persistence/sync/mod.rs index c2f3c605..1306a34f 100644 --- a/tests/persistence/sync/mod.rs +++ b/tests/persistence/sync/mod.rs @@ -61,7 +61,7 @@ fn test_wait_for_ops_for_empty() { let engine = TestSyncPersistenceEngine::new(config.clone()).await.unwrap(); let table = TestSyncWorkTable::load(engine).await.unwrap(); tokio::time::sleep(Duration::from_millis(200)).await; - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); }); } @@ -93,7 +93,7 @@ fn test_space_insert_sync() { id: table.get_next_pk().0, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -140,7 +140,7 @@ fn test_space_insert_many_sync() { }; pks.push(pk); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { @@ -192,7 +192,7 @@ fn test_space_update_full_sync() { }) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -237,7 +237,7 @@ fn test_space_update_query_pk_sync() { .update_another_by_id(AnotherByIdQuery { another: 13 }, row.id) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -282,7 +282,7 @@ fn test_space_update_query_unique_sync() { .update_field_by_another(FieldByAnotherQuery { field: 1.0 }, 42) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -327,7 +327,7 @@ fn test_space_update_query_non_unique_sync() { .update_another_by_non_unique(AnotherByNonUniqueQuery { another: 13 }, 10) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -369,7 +369,7 @@ fn test_space_delete_sync() { }; table.insert(row.clone()).unwrap(); table.delete(row.id).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -410,7 +410,7 @@ fn test_space_delete_query_sync() { }; table.insert(row.clone()).unwrap(); table.delete_by_another(row.another).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { diff --git a/tests/persistence/sync/option.rs b/tests/persistence/sync/option.rs index f663957c..7c2ac3b4 100644 --- a/tests/persistence/sync/option.rs +++ b/tests/persistence/sync/option.rs @@ -54,7 +54,7 @@ fn test_option_insert_none_sync() { exchange: 1, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -96,7 +96,7 @@ fn test_option_insert_some_sync() { exchange: 1, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -148,7 +148,7 @@ fn test_option_update_full_sync() { }) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -195,7 +195,7 @@ fn test_option_update_by_id_sync() { .update_test_by_id(TestByIdQuery { test: Some(42) }, row.id) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -242,7 +242,7 @@ fn test_option_update_none_to_some_sync() { .update_test_by_id(TestByIdQuery { test: Some(55) }, row.id) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -289,7 +289,7 @@ fn test_option_update_some_to_none_sync() { .update_test_by_id(TestByIdQuery { test: None }, row.id) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -336,7 +336,7 @@ fn test_option_update_by_another_sync() { .update_test_by_another(TestByAnotherQuery { test: Some(77) }, 123) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -383,7 +383,7 @@ fn test_option_update_by_exchange_sync() { .update_test_by_exchange(TestByExchangeQuery { test: Some(88) }, 456) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -440,7 +440,7 @@ fn test_option_multiple_rows_sync() { .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (pk1, pk2) }; @@ -504,7 +504,7 @@ fn test_option_indexed_insert_none_sync() { exchange: 1, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -546,7 +546,7 @@ fn test_option_indexed_insert_some_sync() { exchange: 1, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -593,7 +593,7 @@ fn test_option_indexed_update_none_to_some_by_id_sync() { .update_index_test_by_id(IndexTestByIdQuery { test: Some(55) }, row.id) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -640,7 +640,7 @@ fn test_option_indexed_update_some_to_none_by_id_sync() { .update_index_test_by_id(IndexTestByIdQuery { test: None }, row.id) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -687,7 +687,7 @@ fn test_option_indexed_update_by_another_sync() { .update_index_test_by_another(IndexTestByAnotherQuery { test: Some(77) }, 123) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; @@ -757,7 +757,7 @@ fn test_option_indexed_multiple_rows_sync() { .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (pk1, pk2, pk3) }; @@ -809,7 +809,7 @@ fn test_option_indexed_full_row_update_sync() { }) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; diff --git a/tests/persistence/sync/string_primary_index.rs b/tests/persistence/sync/string_primary_index.rs index 82a138dd..0de9853c 100644 --- a/tests/persistence/sync/string_primary_index.rs +++ b/tests/persistence/sync/string_primary_index.rs @@ -57,7 +57,7 @@ fn test_space_insert_sync() { id: "Some string to test".to_string(), }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -103,7 +103,7 @@ fn test_space_insert_many_sync() { }; pks.push(pk); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { @@ -153,7 +153,7 @@ fn test_space_update_full_sync() { }) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -197,7 +197,7 @@ fn test_space_update_query_pk_sync() { .update_another_by_id(AnotherByIdQuery { another: 13 }, row.id.clone()) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -241,7 +241,7 @@ fn test_space_update_query_unique_sync() { .update_field_by_another(FieldByAnotherQuery { field: 1.0 }, 42) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -285,7 +285,7 @@ fn test_space_update_query_non_unique_sync() { .update_another_by_non_unique(AnotherByNonUniqueQuery { another: 13 }, 10) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -333,7 +333,7 @@ fn test_space_delete_sync() { }; table.insert(another_row.clone()).unwrap(); table.delete(another_row.id.clone()).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); another_row.id }; { @@ -373,7 +373,7 @@ fn test_space_delete_query_sync() { }; table.insert(row.clone()).unwrap(); table.delete_by_another(row.another).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { diff --git a/tests/persistence/sync/string_re_read.rs b/tests/persistence/sync/string_re_read.rs index 2cb8e709..ff478294 100644 --- a/tests/persistence/sync/string_re_read.rs +++ b/tests/persistence/sync/string_re_read.rs @@ -68,7 +68,7 @@ fn test_key() { }) .unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -82,7 +82,7 @@ fn test_key() { last: "_________________________last_____________________".to_string(), }) .unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -132,7 +132,7 @@ fn test_key_delete_scenario() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (pk0, pk) }; { @@ -140,7 +140,7 @@ fn test_key_delete_scenario() { let table = StringReReadWorkTable::load(engine).await.unwrap(); table.delete(pk.clone()).await.unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -160,14 +160,14 @@ fn test_key_delete_scenario() { }) .unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); let table = StringReReadWorkTable::load(engine).await.unwrap(); table.delete(pk0.clone()).await.unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -228,7 +228,7 @@ fn test_key_delete() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); pk }; { @@ -236,7 +236,7 @@ fn test_key_delete() { let table = StringReReadWorkTable::load(engine).await.unwrap(); table.delete(pk.clone()).await.unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -290,7 +290,7 @@ fn test_key_delete_all() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (pk0, pk1) }; { @@ -299,7 +299,7 @@ fn test_key_delete_all() { table.delete(pk0.clone()).await.unwrap(); table.delete(pk1.clone()).await.unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -355,7 +355,7 @@ fn test_key_delete_all_and_insert() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (pk0, pk1) }; { @@ -364,7 +364,7 @@ fn test_key_delete_all_and_insert() { table.delete(pk0.clone()).await.unwrap(); table.delete(pk1.clone()).await.unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } let pk = { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -381,7 +381,7 @@ fn test_key_delete_all_and_insert() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); pk }; { @@ -437,7 +437,7 @@ fn test_key_delete_by_unique() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); pk }; { @@ -445,7 +445,7 @@ fn test_key_delete_by_unique() { let table = StringReReadWorkTable::load(engine).await.unwrap(); table.delete_by_second("second_again".to_string()).await.unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -499,7 +499,7 @@ fn test_key_delete_by_non_unique() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); (pk0, pk1) }; { @@ -507,7 +507,7 @@ fn test_key_delete_by_non_unique() { let table = StringReReadWorkTable::load(engine).await.unwrap(); table.delete_by_first("first".to_string()).await.unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -565,7 +565,7 @@ fn test_toc_not_updated_when_index_value_same_but_link_changes() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); pk1 }; @@ -584,7 +584,7 @@ fn test_toc_not_updated_when_index_value_same_but_link_changes() { .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { @@ -601,7 +601,9 @@ fn test_toc_not_updated_when_index_value_same_but_link_changes() { assert!(result.is_ok(), "TOC entry is stale after update with same index value"); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); if wait_result.is_err() { panic!("BUG DETECTED: Persistence system is stuck - wait_for_ops() timed out"); } @@ -653,7 +655,7 @@ fn test_big_amount_reread() { .unwrap(); } - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -668,7 +670,7 @@ fn test_big_amount_reread() { }) .unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = StringReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -712,7 +714,7 @@ fn test_unique_index_same_value_link_changes() { }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); pk1 }; @@ -748,7 +750,9 @@ fn test_unique_index_same_value_link_changes() { ); // Timeout check for stuck persistence - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); assert!( wait_result.is_ok(), "BUG: persistence blocked after unique index update" diff --git a/tests/persistence/sync/string_secondary_index.rs b/tests/persistence/sync/string_secondary_index.rs index f0c03a5a..ff2bb1e0 100644 --- a/tests/persistence/sync/string_secondary_index.rs +++ b/tests/persistence/sync/string_secondary_index.rs @@ -57,7 +57,7 @@ fn test_space_insert_sync() { id: table.get_next_pk().0, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -104,7 +104,7 @@ fn test_space_insert_many_sync() { }; pks.push(pk); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); } { @@ -156,7 +156,7 @@ fn test_space_update_full_sync() { }) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); assert_eq!( table.select(row.id).unwrap().another, "Some string to test updated".to_string() @@ -213,7 +213,7 @@ fn test_space_update_query_pk_sync() { ) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -261,7 +261,7 @@ fn test_space_update_query_unique_sync() { .update_field_by_another(FieldByAnotherQuery { field: 1.0 }, "Some string before".to_string()) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -311,7 +311,7 @@ fn test_space_update_query_non_unique_sync() { ) .await .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -356,7 +356,7 @@ fn test_space_delete_sync() { }; table.insert(row.clone()).unwrap(); table.delete(row.id).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -397,7 +397,7 @@ fn test_space_delete_query_sync() { }; table.insert(row.clone()).unwrap(); table.delete_by_another(row.another).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row.id }; { @@ -441,7 +441,7 @@ fn test_space_delete_query_sync() { // table.insert(row.clone()).unwrap(); // } // -// table.wait_for_ops().await; +// table.wait_for_ops().await.unwrap(); // }; // { // let table = TestSyncWorkTable::load_from_file(config).await.unwrap(); diff --git a/tests/persistence/sync/string_update_timeout.rs b/tests/persistence/sync/string_update_timeout.rs index d54c6dd4..4eef4cd3 100644 --- a/tests/persistence/sync/string_update_timeout.rs +++ b/tests/persistence/sync/string_update_timeout.rs @@ -75,7 +75,7 @@ fn test_string_update_doesnt_block_persistence() { honey_app_role: 2, }; table.insert(row.clone()).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); row }; @@ -85,7 +85,9 @@ fn test_string_update_doesnt_block_persistence() { table.update(row.clone()).await.unwrap(); - let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()).await; + let wait_result = timeout(Duration::from_secs(4), table.wait_for_ops()) + .await + .expect("persistence timed out"); if wait_result.is_err() { panic!( diff --git a/tests/persistence/sync/uuid_.rs b/tests/persistence/sync/uuid_.rs index 51d820e1..7725f7ea 100644 --- a/tests/persistence/sync/uuid_.rs +++ b/tests/persistence/sync/uuid_.rs @@ -55,7 +55,7 @@ fn test_uuid() { }) .unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = UuidReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -67,7 +67,7 @@ fn test_uuid() { second: Uuid::now_v7(), }) .unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = UuidReReadPersistenceEngine::new(config.clone()).await.unwrap(); @@ -108,7 +108,7 @@ fn test_big_amount_reread() { .unwrap(); } - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } let second_last = Uuid::now_v7(); { @@ -122,7 +122,7 @@ fn test_big_amount_reread() { second: second_last, }) .unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { let engine = UuidReReadPersistenceEngine::new(config.clone()).await.unwrap(); diff --git a/tests/persistence/torn_shutdown.rs b/tests/persistence/torn_shutdown.rs index 2c9bca07..1c20cebe 100644 --- a/tests/persistence/torn_shutdown.rs +++ b/tests/persistence/torn_shutdown.rs @@ -122,7 +122,8 @@ fn tear_the_store_repeatedly() { } timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled building the base store"); + .expect("persistence stalled building the base store") + .expect("persistence engine failed"); } }); @@ -253,7 +254,8 @@ fn test_store_survives_torn_shutdowns() { table.insert(row(9_000_000)).unwrap(); timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled appending to the survivor store"); + .expect("persistence stalled appending to the survivor store") + .expect("persistence engine failed"); }); }); if let Err(panic) = outcome { @@ -309,7 +311,8 @@ fn test_many_clean_sessions_stay_readable() { } timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .unwrap_or_else(|_| panic!("session {session}: drain stalled")); + .unwrap_or_else(|_| panic!("session {session}: drain stalled")) + .expect("persistence engine failed"); } let table = open().await; diff --git a/tests/persistence/tuple_primary_key.rs b/tests/persistence/tuple_primary_key.rs index e8b10f9c..03e4b1bc 100644 --- a/tests/persistence/tuple_primary_key.rs +++ b/tests/persistence/tuple_primary_key.rs @@ -49,7 +49,7 @@ async fn composite_primary_key_survives_mutations_and_reload() { for row in &rows { table.insert(row.clone()).unwrap(); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); for row in &rows { assert_eq!(table.select((row.tenant_id, row.record_id)), Some(row.clone())); } @@ -73,10 +73,9 @@ async fn composite_primary_key_survives_mutations_and_reload() { }; table.update(updated.clone()).await.unwrap(); table.delete((7, 41)).await.unwrap(); - table.wait_for_ops().await; - assert_eq!(table.select((7, 42)), Some(updated)); assert!(table.select((7, 41)).is_none()); + table.close().await.unwrap(); } { diff --git a/tests/persistence/vacuum.rs b/tests/persistence/vacuum.rs index 1e510e3f..5623c8e9 100644 --- a/tests/persistence/vacuum.rs +++ b/tests/persistence/vacuum.rs @@ -74,7 +74,8 @@ fn test_vacuum_on_persisted_table_survives_reload() { timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence should catch up before vacuum"); + .expect("persistence should catch up before vacuum") + .expect("persistence engine failed"); let vacuum = table.vacuum(); let stats = vacuum.vacuum().await.unwrap(); @@ -96,7 +97,8 @@ fn test_vacuum_on_persisted_table_survives_reload() { if i % 50 == 49 { timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled after vacuum on persisted table"); + .expect("persistence stalled after vacuum on persisted table") + .expect("persistence engine failed"); } } @@ -105,7 +107,8 @@ fn test_vacuum_on_persisted_table_survives_reload() { // consumed, leaving a permanent gap the batch validator defers on. timeout(Duration::from_secs(30), table.wait_for_ops()) .await - .expect("persistence stalled after vacuum on persisted table"); + .expect("persistence stalled after vacuum on persisted table") + .expect("persistence engine failed"); for id in &deleted { rows.remove(id); diff --git a/tests/worktable/index_backends.rs b/tests/worktable/index_backends.rs index 2465d787..0924722b 100644 --- a/tests/worktable/index_backends.rs +++ b/tests/worktable/index_backends.rs @@ -258,7 +258,7 @@ async fn upstream_indexset_survives_persist_reload_and_more_writes() { }) .unwrap(); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let engine = PersistedUpstreamPersistenceEngine::new(config.clone()).await.unwrap(); @@ -274,7 +274,7 @@ async fn upstream_indexset_survives_persist_reload_and_more_writes() { .unwrap(); let added_id: u64 = added_pk.clone().into(); table.delete(10).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let engine = PersistedUpstreamPersistenceEngine::new(config).await.unwrap(); @@ -283,7 +283,7 @@ async fn upstream_indexset_survives_persist_reload_and_more_writes() { assert!(table.select(10).is_none()); assert_eq!(table.select(added_pk).unwrap().unique_key, 2_000); assert_eq!(table.select_by_unique_key(2_000).unwrap().id, added_id); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); remove_dir_if_exists(ROOT.to_string()).await; @@ -329,7 +329,7 @@ async fn native_art_backends_survive_wal_reload_and_further_mutation() { congee_key: 300, }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let engine = PersistedArcticPersistenceEngine::new(arctic_config.clone()) @@ -341,14 +341,14 @@ async fn native_art_backends_survive_wal_reload_and_further_mutation() { assert_eq!(table.select(accepted_id).unwrap().congee_key, 300); assert_eq!(table.select_by_congee_key(77).unwrap().congee_key, 77); table.delete(77).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let engine = PersistedArcticPersistenceEngine::new(arctic_config).await.unwrap(); let table = PersistedArcticWorkTable::load(engine).await.unwrap(); assert!(table.select(77).is_none()); assert!(table.select_by_congee_key(77).is_none()); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let congee_config = DiskConfig::new_with_table_name( @@ -368,7 +368,7 @@ async fn native_art_backends_survive_wal_reload_and_further_mutation() { }) .unwrap(); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let engine = PersistedCongeePersistenceEngine::new(congee_config.clone()) @@ -378,14 +378,14 @@ async fn native_art_backends_survive_wal_reload_and_further_mutation() { assert_eq!(table.count(), 256); assert_eq!(table.select_by_arctic_key(199).unwrap().arctic_key, 199); table.delete(199).await.unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let engine = PersistedCongeePersistenceEngine::new(congee_config).await.unwrap(); let table = PersistedCongeeWorkTable::load(engine).await.unwrap(); assert!(table.select(199).is_none()); assert!(table.select_by_arctic_key(199).is_none()); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); remove_dir_if_exists(ARCTIC_ROOT.to_string()).await; @@ -437,7 +437,7 @@ async fn native_art_backends_recover_concurrent_same_row_updates() { } let expected = table.select(id).unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let engine = PersistedArcticPersistenceEngine::new(config).await.unwrap(); @@ -449,7 +449,7 @@ async fn native_art_backends_recover_concurrent_same_row_updates() { assert!(table.select_by_congee_key(key).is_none()); } } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); remove_dir_if_exists(ROOT.to_string()).await; @@ -480,7 +480,7 @@ async fn persisted_tables_can_switch_between_wti_and_upstream_without_rebuild() }) .unwrap(); } - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let upstream_config = DiskConfig::new_with_table_name( @@ -501,7 +501,7 @@ async fn persisted_tables_can_switch_between_wti_and_upstream_without_rebuild() unique_key: 2_000, }) .unwrap(); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); let engine = wti::ProviderSwitchPersistenceEngine::new(wti_config).await.unwrap(); @@ -509,7 +509,7 @@ async fn persisted_tables_can_switch_between_wti_and_upstream_without_rebuild() assert_eq!(table.count(), 1_024); assert!(table.select(10).is_none()); assert_eq!(table.select_by_unique_key(2_000).unwrap().unique_key, 2_000); - table.wait_for_ops().await; + table.wait_for_ops().await.unwrap(); drop(table); remove_dir_if_exists(ROOT.to_string()).await; diff --git a/tests/worktable_version/basic.rs b/tests/worktable_version/basic.rs index b480b93a..5264efed 100644 --- a/tests/worktable_version/basic.rs +++ b/tests/worktable_version/basic.rs @@ -66,7 +66,7 @@ fn test_version_reads_persisted_data() { }) .unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } { diff --git a/tests/worktable_version/string_primary_key.rs b/tests/worktable_version/string_primary_key.rs index 3372a9cd..60f844de 100644 --- a/tests/worktable_version/string_primary_key.rs +++ b/tests/worktable_version/string_primary_key.rs @@ -69,7 +69,7 @@ fn test_version_reads_persisted_data_with_string_primary_key() { }) .unwrap(); - table.wait_for_ops().await + table.wait_for_ops().await.unwrap() } {