diff --git a/Cargo.toml b/Cargo.toml index f92e8cc..8dbe242 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ members = ["codegen", "examples", "performance_measurement", "performance_measur [package] name = "worktable" -version = "1.0.0-beta.8" +version = "1.0.0-beta.9" edition = "2024" authors = ["Handy-caT"] license = "MIT" @@ -66,7 +66,7 @@ tracing = "0.1" url = { version = "2", optional = true } uuid = { version = "1.24.0", features = ["v4", "v7"] } walkdir = { version = "2", optional = true } -worktable_codegen = { path = "codegen", version = "=1.0.0-beta.8" } +worktable_codegen = { path = "codegen", version = "=1.0.0-beta.9" } [dev-dependencies] chrono = "0.4.43" diff --git a/codegen/Cargo.toml b/codegen/Cargo.toml index 45016bf..a5daf24 100644 --- a/codegen/Cargo.toml +++ b/codegen/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "worktable_codegen" -version = "1.0.0-beta.8" +version = "1.0.0-beta.9" edition = "2024" license = "MIT" description = "Proc-macro companion crate for worktable: the worktable! macro and its derives." 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 20745e6..bd3dac7 100644 --- a/codegen/src/persist_table/generator/space_file/worktable_impls.rs +++ b/codegen/src/persist_table/generator/space_file/worktable_impls.rs @@ -10,6 +10,7 @@ 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 persistence_monitor_fn = self.gen_worktable_persistence_monitor_fn(); let close_fn = self.gen_worktable_close_fn(); let persisted_data_file_size_fn = self.gen_persisted_data_file_size_fn(); @@ -18,6 +19,7 @@ impl Generator { #space_info_fn #persisted_pk_fn #wait_for_ops_fn + #persistence_monitor_fn #close_fn #persisted_data_file_size_fn } @@ -55,6 +57,20 @@ impl Generator { } } + fn gen_worktable_persistence_monitor_fn(&self) -> TokenStream { + if self.attributes.read_only { + quote! {} + } else { + quote! { + /// Returns a cloneable terminal-state monitor that does not + /// borrow the table and can therefore observe `close()`. + pub fn persistence_monitor(&self) -> PersistenceMonitor { + self.1.monitor() + } + } + } + } + fn gen_worktable_close_fn(&self) -> TokenStream { if self.attributes.read_only { quote! { diff --git a/codegen/src/persist_table/mod.rs b/codegen/src/persist_table/mod.rs index e0d12cf..c57e3ad 100644 --- a/codegen/src/persist_table/mod.rs +++ b/codegen/src/persist_table/mod.rs @@ -78,6 +78,10 @@ mod tests { output.contains("fn into_worktable_with_mode"), "read_only should generate explicit recovery-mode conversion" ); + assert!( + !output.contains("persistence_monitor"), + "read_only should not expose monitoring for a worker it does not have" + ); assert!(output.contains("LoadMode :: Strict")); } @@ -107,6 +111,7 @@ mod tests { output.contains("async fn into_worktable_with_mode"), "normal should generate explicit recovery-mode conversion" ); + assert!(output.contains("fn persistence_monitor")); assert!(output.contains("LoadMode :: Strict")); } diff --git a/src/lib.rs b/src/lib.rs index 9e28c54..d39af05 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -36,11 +36,12 @@ pub mod prelude { pub use crate::persistence::{ AcknowledgeOperation, ArtPersistenceKey, DeleteOperation, DiskConfig, DiskPersistenceEngine, IndexTableOfContents, InsertOperation, LoadMode, Operation, OperationId, PersistedWorkTable, PersistenceConfig, - PersistenceEngine, PersistenceError, PersistenceIndexCorruption, PersistenceLoadError, PersistenceResult, - PersistenceState, PersistenceTask, ReadOnlyPersistenceEngine, SpaceArcticIndex, SpaceCongeeIndex, SpaceData, - SpaceDataOps, SpaceIndex, SpaceIndexOps, SpaceIndexUnsized, SpaceLogicalIndex, SpaceLogicalIndexUnsized, - SpaceSecondaryIndexOps, UpdateOperation, load_persisted_state, map_index_pages_to_toc_and_general, - map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, validate_events, + PersistenceEngine, PersistenceError, PersistenceIndexCorruption, PersistenceLoadError, PersistenceMonitor, + PersistenceResult, PersistenceState, PersistenceTask, ReadOnlyPersistenceEngine, SpaceArcticIndex, + SpaceCongeeIndex, SpaceData, SpaceDataOps, SpaceIndex, SpaceIndexOps, SpaceIndexUnsized, SpaceLogicalIndex, + SpaceLogicalIndexUnsized, SpaceSecondaryIndexOps, UpdateOperation, load_persisted_state, + 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/mod.rs b/src/persistence/mod.rs index 121e27a..c875061 100644 --- a/src/persistence/mod.rs +++ b/src/persistence/mod.rs @@ -20,7 +20,7 @@ pub use space::{ SpaceIndexOps, SpaceIndexUnsized, SpaceLogicalIndex, SpaceLogicalIndexUnsized, SpaceSecondaryIndexOps, map_index_pages_to_toc_and_general, map_unsized_index_pages_to_toc_and_general, reconstruct_multi_index_nodes, }; -pub use task::PersistenceTask; +pub use task::{PersistenceMonitor, PersistenceTask}; mod engine; mod error; diff --git a/src/persistence/space/index/table_of_contents.rs b/src/persistence/space/index/table_of_contents.rs index 5703c34..181ef08 100644 --- a/src/persistence/space/index/table_of_contents.rs +++ b/src/persistence/space/index/table_of_contents.rs @@ -113,15 +113,30 @@ where pub fn update_key(&mut self, old_key: &T, new_key: T) where T: Clone + Debug, + { + assert!( + self.try_update_key(old_key, new_key), + "Page with key {old_key:?} not found" + ); + } + + /// Updates a page identity without panicking when the old identity is + /// absent. Batch replay uses this checked form because an absent key is a + /// persistence invariant failure that must surface through `Result`. + pub fn try_update_key(&mut self, old_key: &T, new_key: T) -> bool + where + T: Clone, { let page = self.get_current_page_mut(); if page.inner.update_key(old_key, new_key.clone()).is_none() { for page in self.pages.iter_mut() { if page.inner.update_key(old_key, new_key.clone()).is_some() { - return; + return true; } } - panic!("Page with key {old_key:?} not found"); + false + } else { + true } } @@ -229,6 +244,16 @@ mod tests { ); } + #[test] + fn checked_update_reports_a_missing_identity_without_mutating_the_toc() { + let mut toc = IndexTableOfContents::::new(0.into(), Arc::new(AtomicU32::new(1))); + toc.insert(7, 2.into()); + + assert!(!toc.try_update_key(&8, 9)); + assert_eq!(toc.get(&7), Some(2.into())); + assert_eq!(toc.get(&9), None); + } + #[test] fn insert_more_than_one_page() { let mut toc = IndexTableOfContents::::new(0.into(), Arc::new(AtomicU32::new(0))); diff --git a/src/persistence/space/index/unsized_.rs b/src/persistence/space/index/unsized_.rs index 42070a4..b7efef1 100644 --- a/src/persistence/space/index/unsized_.rs +++ b/src/persistence/space/index/unsized_.rs @@ -28,6 +28,11 @@ use crate::persistence::space::BatchChangeEvent; use crate::persistence::{IndexTableOfContents, SpaceIndex, SpaceIndexOps}; use crate::prelude::WT_INDEX_EXTENSION; +// Normal persistence batches begin at 16 source pages. Keep that common case +// inline while allowing analyzer retries to grow beyond it without turning a +// recovery batch into a terminal capacity error. +const INLINE_BATCH_ALIASED_PAGES: usize = 16; + #[derive(Debug)] pub struct SpaceIndexUnsized { space_id: SpaceId, @@ -59,6 +64,21 @@ where + Debug + for<'a> rkyv::bytecheck::CheckBytes>, { + fn resolve_batch_page( + &self, + aliases: &PageAliases, + event_page_key: &(T, Link), + ) -> Option<(PageId, Option<(T, Link)>)> { + self.table_of_contents + .get(event_page_key) + .map(|page_id| (page_id, None)) + .or_else(|| { + aliases + .resolve(event_page_key) + .map(|(page_id, current_key)| (page_id, Some(current_key.clone()))) + }) + } + fn compact_page_if_needed(page: &mut UnsizedIndexPage) -> eyre::Result<()> { let persisted_size = UnsizedIndexPageUtility::::persisted_size(page.slots_size as usize, page.node_id_size as usize) @@ -390,15 +410,33 @@ where async fn process_change_event_batch(&mut self, events: BatchChangeEvent) -> eyre::Result<()> { let mut pages: HashMap = HashMap::new(); + // A split can change a page's maximum and therefore its table-of- + // contents key while later events in the same CDC batch still refer + // to the pre-split maximum. Keep those historical identities scoped + // to this batch so the event reaches the page it was generated from. + // At most two transitional identities per buffered page: the event's + // identity and the page's actual pre-apply identity. Keeping both is + // required when a split is followed by a remove/insert pair that still + // names the pre-split maximum. Memory is bounded by pages, not events. + let mut page_aliases = PageAliases::default(); for ev in events { match &ev { ChangeEvent::InsertAt { max_value, .. } | ChangeEvent::RemoveAt { max_value, .. } => { - let page_id = &(max_value.key.clone(), max_value.value); - - let page_index = self - .table_of_contents - .get(page_id) - .expect("page should be available in table of contents"); + let event_page_key = (max_value.key.clone(), max_value.value); + // A direct TOC hit means the event key is the page's + // canonical pre-event identity. An alias hit carries the + // current canonical identity captured when the alias was + // installed. This lets us compare the actual post-apply + // identity without predicting DataBucket's mutation rules. + let Some((page_index, aliased_page_key)) = self.resolve_batch_page(&page_aliases, &event_page_key) + else { + return Err(eyre!( + "unsized index event references a missing page (toc_segments={}, buffered_pages={}, aliases={})", + self.table_of_contents.pages.len(), + pages.len(), + page_aliases.len() + )); + }; let page = pages.get_mut(&page_index); let page_to_update = if let Some(page) = page { page @@ -413,19 +451,35 @@ where .get_mut(&page_index) .expect("should be available as was just inserted before") }; + let canonical_page_key = aliased_page_key.as_ref().unwrap_or(&event_page_key); page_to_update.inner.apply_change_event(ev.clone())?; - if &( - page_to_update.inner.node_id.key.clone(), - page_to_update.inner.node_id.link, - ) != page_id + if page_to_update.inner.node_id.key != canonical_page_key.0 + || page_to_update.inner.node_id.link != canonical_page_key.1 { - self.table_of_contents.update_key( - page_id, - ( - page_to_update.inner.node_id.key.clone(), - page_to_update.inner.node_id.link, - ), + let pre_event_page_key = aliased_page_key.unwrap_or_else(|| event_page_key.clone()); + let updated_page_key = ( + page_to_update.inner.node_id.key.clone(), + page_to_update.inner.node_id.link, ); + // The TOC owns the buffered page's actual identity. + // `event_page_key` may be a historical alias, so using + // it as the canonical update target can remove or + // rewrite the wrong segment. + if !self + .table_of_contents + .try_update_key(&pre_event_page_key, updated_page_key.clone()) + { + return Err(eyre!( + "unsized index page identity is absent from the table of contents (page={page_index:?}, toc_segments={})", + self.table_of_contents.pages.len() + )); + } + if self.table_of_contents.get(&updated_page_key) != Some(page_index) { + return Err(eyre!( + "unsized index page identity update did not become canonical (page={page_index:?})" + )); + } + page_aliases.replace(page_index, updated_page_key, event_page_key, pre_event_page_key)?; } } ChangeEvent::CreateNode { event_id: _, max_value } => { @@ -452,12 +506,17 @@ where max_value, split_index, } => { - let page_id = &(max_value.key.clone(), max_value.value); - - let page_index = self - .table_of_contents - .get(page_id) - .expect("page should be available in table of contents"); + let event_page_key = (max_value.key.clone(), max_value.value); + + let Some((page_index, aliased_page_key)) = self.resolve_batch_page(&page_aliases, &event_page_key) + else { + return Err(eyre!( + "unsized index split references a missing page (toc_segments={}, buffered_pages={}, aliases={})", + self.table_of_contents.pages.len(), + pages.len(), + page_aliases.len() + )); + }; let page = pages.get_mut(&page_index); let page_to_update = if let Some(page) = page { page @@ -472,6 +531,15 @@ where .get_mut(&page_index) .expect("should be available as was just inserted before") }; + let canonical_page_key = aliased_page_key.as_ref().unwrap_or(&event_page_key); + if page_to_update.inner.node_id.key != canonical_page_key.0 + || page_to_update.inner.node_id.link != canonical_page_key.1 + { + return Err(eyre!( + "unsized index split found a buffered page with a mismatched identity (page={page_index:?})" + )); + } + let pre_split_page_key = aliased_page_key.unwrap_or_else(|| event_page_key.clone()); let splitted_page = page_to_update.inner.split(*split_index); let new_page_id = if let Some(id) = self.table_of_contents.pop_empty_page_id() { @@ -480,23 +548,36 @@ where self.next_page_id.fetch_add(1, Ordering::Relaxed).into() }; - self.table_of_contents.update_key( - page_id, - ( - page_to_update.inner.node_id.key.clone(), - page_to_update.inner.node_id.link, - ), - ); - self.table_of_contents.insert( - (splitted_page.node_id.key.clone(), splitted_page.node_id.link), - new_page_id, + let left_page_key = ( + page_to_update.inner.node_id.key.clone(), + page_to_update.inner.node_id.link, ); + if !self + .table_of_contents + .try_update_key(&pre_split_page_key, left_page_key) + { + return Err(eyre!( + "unsized index split identity is absent from the table of contents (page={page_index:?})" + )); + } + let right_page_key = (splitted_page.node_id.key.clone(), splitted_page.node_id.link); + self.table_of_contents.insert(right_page_key.clone(), new_page_id); + if self.table_of_contents.get(&right_page_key) != Some(new_page_id) { + return Err(eyre!( + "unsized index split identity did not become canonical (page={new_page_id:?})" + )); + } let header = GeneralHeader::new(new_page_id, PageType::Index, self.space_id); let general_page = GeneralPage { inner: splitted_page, header, }; pages.insert(new_page_id, general_page); + // The pre-split maximum remains the right page's identity. + // A following remove/insert pair can still name it even + // after the remove temporarily lowers that maximum. + page_aliases.remove_page(page_index); + page_aliases.replace(new_page_id, right_page_key, event_page_key, pre_split_page_key)?; } } } @@ -513,3 +594,272 @@ where Ok(()) } } + +/// Transitional event identities for pages whose canonical TOC key changed +/// earlier in the same CDC batch. +/// +/// Each page owns at most the event identity and its actual pre-event identity. +/// The current canonical identity is retained alongside them so alias lookup +/// never has to predict page mutation semantics. Normal batches use the inline +/// slots; analyzer retries may spill into `overflow` without losing events. +struct PageAliases { + inline: [Option>; INLINE_BATCH_ALIASED_PAGES], + overflow: Vec>, +} + +struct PageAliasEntry { + page_id: PageId, + current_key: (T, Link), + aliases: [Option<(T, Link)>; 2], +} + +impl Default for PageAliases { + fn default() -> Self { + Self { + inline: std::array::from_fn(|_| None), + overflow: Vec::new(), + } + } +} + +impl PageAliases { + fn entries(&self) -> impl Iterator> { + self.inline.iter().flatten().chain(self.overflow.iter()) + } + + fn resolve(&self, key: &(T, Link)) -> Option<(PageId, &(T, Link))> { + self.entries().find_map(|entry| { + entry + .aliases + .iter() + .flatten() + .any(|alias| alias == key) + .then_some((entry.page_id, &entry.current_key)) + }) + } + + #[cfg(test)] + fn get(&self, key: &(T, Link)) -> Option { + self.resolve(key).map(|(page_id, _)| page_id) + } + + fn len(&self) -> usize { + self.entries().map(|entry| entry.aliases.iter().flatten().count()).sum() + } + + #[cfg(test)] + fn page_len(&self) -> usize { + self.entries().count() + } + + fn remove_page(&mut self, page_id: PageId) { + if let Some(slot) = self + .inline + .iter_mut() + .find(|slot| slot.as_ref().is_some_and(|entry| entry.page_id == page_id)) + { + *slot = None; + } else if let Some(index) = self.overflow.iter().position(|entry| entry.page_id == page_id) { + self.overflow.swap_remove(index); + } + } + + fn replace( + &mut self, + page_id: PageId, + current_key: (T, Link), + event_key: (T, Link), + pre_event_key: (T, Link), + ) -> eyre::Result<()> { + let first_alias = (event_key != current_key).then_some(event_key); + let second_alias = + (pre_event_key != current_key && first_alias.as_ref() != Some(&pre_event_key)).then_some(pre_event_key); + + if first_alias.is_none() && second_alias.is_none() { + self.remove_page(page_id); + return Ok(()); + } + + for alias in [first_alias.as_ref(), second_alias.as_ref()].into_iter().flatten() { + if let Some(owner) = self + .entries() + .find(|entry| entry.page_id != page_id && entry.aliases.iter().flatten().any(|stored| stored == alias)) + { + return Err(eyre!( + "page alias ownership collision between {:?} and {page_id:?}", + owner.page_id + )); + } + } + + let entry = PageAliasEntry { + page_id, + current_key, + aliases: [first_alias, second_alias], + }; + if let Some(slot) = self + .inline + .iter_mut() + .find(|slot| slot.as_ref().is_some_and(|stored| stored.page_id == page_id)) + { + *slot = Some(entry); + } else if let Some(slot) = self.overflow.iter_mut().find(|stored| stored.page_id == page_id) { + *slot = entry; + } else if let Some(slot) = self.inline.iter_mut().find(|slot| slot.is_none()) { + *slot = Some(entry); + } else { + self.overflow.push(entry); + } + Ok(()) + } +} + +#[cfg(test)] +mod alias_tests { + use super::*; + + fn link(offset: u32) -> Link { + Link { + page_id: 1.into(), + offset, + length: 8, + } + } + + #[test] + fn repeated_maximum_changes_keep_one_alias_per_page() { + let page_id = PageId::from(7); + let mut aliases = PageAliases::default(); + for revision in 0..1_000 { + let current = (format!("current-{revision}"), link(revision + 1_000)); + aliases + .replace( + page_id, + current, + (format!("key-{revision}"), link(revision)), + (format!("key-{revision}"), link(revision)), + ) + .unwrap(); + } + + assert_eq!(aliases.len(), 1); + assert_eq!(aliases.get(&("key-999".into(), link(999))), Some(page_id)); + assert_eq!(aliases.page_len(), 1); + } + + #[test] + fn canonical_only_transition_stores_no_alias_entry() { + let page_id = PageId::from(7); + let canonical = ("current".to_string(), link(1)); + let mut aliases = PageAliases::default(); + + aliases + .replace(page_id, canonical.clone(), canonical.clone(), canonical) + .unwrap(); + + assert_eq!(aliases.page_len(), 0); + assert_eq!(aliases.len(), 0); + } + + #[test] + fn split_keeps_only_the_two_live_transitional_identities() { + let old_page = PageId::from(3); + let right_page = PageId::from(4); + let mut aliases = PageAliases::default(); + aliases + .replace( + old_page, + ("old-current".to_string(), link(9)), + ("older".to_string(), link(1)), + ("older".to_string(), link(1)), + ) + .unwrap(); + aliases.remove_page(old_page); + aliases + .replace( + right_page, + ("right-current".to_string(), link(4)), + ("event".to_string(), link(2)), + ("pre-split".to_string(), link(3)), + ) + .unwrap(); + + assert_eq!(aliases.len(), 2); + assert_eq!(aliases.page_len(), 1); + } + + #[test] + fn split_remove_insert_preserves_the_event_identity() { + let right_page = PageId::from(4); + let pre_split = ("pre-split".to_string(), link(1)); + let post_split = ("post-split".to_string(), link(2)); + let after_remove = ("after-remove".to_string(), link(3)); + let mut aliases = PageAliases::default(); + + aliases + .replace(right_page, post_split.clone(), pre_split.clone(), pre_split.clone()) + .unwrap(); + aliases + .replace(right_page, after_remove.clone(), pre_split.clone(), post_split.clone()) + .unwrap(); + + assert_eq!(aliases.get(&pre_split), Some(right_page)); + assert_eq!(aliases.get(&post_split), Some(right_page)); + assert_eq!( + aliases.resolve(&pre_split).map(|(_, current)| current), + Some(&after_remove) + ); + } + + #[test] + fn batches_beyond_inline_capacity_preserve_every_alias() { + let mut aliases = PageAliases::default(); + let page_count = INLINE_BATCH_ALIASED_PAGES as u32 + 8; + for page in 1..=page_count { + aliases + .replace( + PageId::from(page), + (format!("current-{page}"), link(page + 1_000)), + (format!("old-{page}"), link(page)), + (format!("old-{page}"), link(page)), + ) + .unwrap(); + } + assert_eq!(aliases.page_len(), page_count as usize); + assert_eq!(aliases.overflow.len(), 8); + assert_eq!(aliases.get(&("old-1".into(), link(1))), Some(PageId::from(1))); + assert_eq!( + aliases.get(&(format!("old-{page_count}"), link(page_count))), + Some(PageId::from(page_count)) + ); + } + + #[test] + fn alias_invariants_fail_without_corrupting_existing_ownership() { + let first_page = PageId::from(1); + let second_page = PageId::from(2); + let shared = ("shared".to_string(), link(1)); + let mut aliases = PageAliases::default(); + aliases + .replace( + first_page, + ("first-current".to_string(), link(9)), + shared.clone(), + shared.clone(), + ) + .unwrap(); + + assert!( + aliases + .replace( + second_page, + ("second-current".to_string(), link(10)), + shared.clone(), + shared.clone(), + ) + .is_err() + ); + assert_eq!(aliases.get(&shared), Some(first_page)); + assert_eq!(aliases.page_len(), 1); + } +} diff --git a/src/persistence/task.rs b/src/persistence/task.rs index f26a5c2..88ca551 100644 --- a/src/persistence/task.rs +++ b/src/persistence/task.rs @@ -41,14 +41,18 @@ const MAX_PAGE_AMOUNT: usize = 16; #[derive(Debug)] struct PersistenceLifecycle { state: ParkingMutex, - notify: Notify, + /// Terminal transitions only: `Failed` or `Closed`. + terminal_notify: Notify, + /// Queue-drain and lifecycle progress observed by `wait_for_ops`. + progress_notify: Notify, } impl PersistenceLifecycle { fn new() -> Self { Self { state: ParkingMutex::new(PersistenceState::Running), - notify: Notify::new(), + terminal_notify: Notify::new(), + progress_notify: Notify::new(), } } @@ -61,7 +65,7 @@ impl PersistenceLifecycle { match &*state { PersistenceState::Running => { *state = PersistenceState::Closing; - self.notify.notify_waiters(); + self.progress_notify.notify_waiters(); Ok(()) } PersistenceState::Closing => Ok(()), @@ -75,7 +79,8 @@ impl PersistenceLifecycle { if matches!(*state, PersistenceState::Closing) { *state = PersistenceState::Closed; } - self.notify.notify_waiters(); + self.terminal_notify.notify_waiters(); + self.progress_notify.notify_waiters(); } fn fail(&self, report: eyre::Report) -> Arc { @@ -91,7 +96,8 @@ impl PersistenceLifecycle { error } }; - self.notify.notify_waiters(); + self.terminal_notify.notify_waiters(); + self.progress_notify.notify_waiters(); error } @@ -105,6 +111,66 @@ impl PersistenceLifecycle { } } +/// Marks every non-graceful worker exit as terminal, including cancellation +/// before the worker future's first poll and panic unwinding during a poll. +struct WorkerCompletionGuard { + lifecycle: Arc, + armed: bool, +} + +impl WorkerCompletionGuard { + fn new(lifecycle: Arc) -> Self { + Self { lifecycle, armed: true } + } + + fn disarm(mut self) { + self.armed = false; + } +} + +impl Drop for WorkerCompletionGuard { + fn drop(&mut self) { + if self.armed { + let reason = if std::thread::panicking() { + "persistence worker panicked" + } else { + "persistence worker was cancelled" + }; + self.lifecycle.fail(eyre::eyre!(reason)); + } + } +} + +/// Cloneable terminal-state handle independent of table ownership. +/// +/// Create this handle before spawning a supervisor. The table can then still +/// be moved into `close()`, while the supervisor observes either graceful +/// closure or a terminal persistence failure. +#[derive(Clone, Debug)] +pub struct PersistenceMonitor { + lifecycle: Arc, +} + +impl PersistenceMonitor { + /// Waits until the worker fails or closes. + pub async fn wait_for_failure(self) -> PersistenceResult { + loop { + let notified = self.lifecycle.terminal_notify.notified(); + tokio::pin!(notified); + // `notify_waiters` does not retain a permit. Register this waiter + // before reading the lifecycle state so a terminal transition + // cannot land between the state read and the first poll of + // `notified` and be lost forever. + notified.as_mut().enable(); + match self.lifecycle.state() { + PersistenceState::Failed(error) => return Err(error), + PersistenceState::Closed => return Ok(()), + PersistenceState::Running | PersistenceState::Closing => notified.await, + } + } + } +} + pub struct QueueAnalyzer { operations: OptimizedVec>, queue_inner_wt: Arc, @@ -417,6 +483,7 @@ mod lifecycle_tests { None, Engine, IndexCorruption, + Panic, } impl PersistenceEngine<(), u64, TestEvents, TestIndex> for TestEngine { @@ -447,6 +514,7 @@ mod lifecycle_tests { PersistenceIndexCorruption::new("table/primary.wt.idx", "injected shadow divergence").into(), ); } + TestFailure::Panic => panic!("injected persistence worker panic"), } self.batches.fetch_add(1, Ordering::Relaxed); self.events.lock().push("batch"); @@ -547,6 +615,22 @@ mod lifecycle_tests { assert_eq!(batches.load(Ordering::Relaxed), 1); } + #[tokio::test] + async fn detached_monitor_observes_graceful_close() { + let task = PersistenceTask::run_engine(TestEngine { + batches: Arc::new(AtomicUsize::new(0)), + events: Arc::new(ParkingMutex::new(Vec::new())), + config: TestConfig, + failure: TestFailure::None, + }); + let monitor = task.monitor(); + let waiter = tokio::spawn(monitor.wait_for_failure()); + + task.close().await.unwrap(); + + waiter.await.unwrap().unwrap(); + } + #[tokio::test] async fn engine_failure_is_terminal_and_reused_for_later_callers() { let task = PersistenceTask::run_engine(TestEngine { @@ -567,6 +651,63 @@ mod lifecycle_tests { assert!(Arc::ptr_eq(&wait_error, &close_error)); } + #[tokio::test] + async fn engine_panic_is_terminal_and_reported_to_waiters() { + let task = PersistenceTask::run_engine(TestEngine { + batches: Arc::new(AtomicUsize::new(0)), + events: Arc::new(ParkingMutex::new(Vec::new())), + config: TestConfig, + failure: TestFailure::Panic, + }); + + task.apply_operation(insert_operation(1)).unwrap(); + let wait_error = task.wait_for_failure().await.unwrap_err(); + assert_eq!( + wait_error.to_string(), + "persistence engine failed: persistence worker panicked" + ); + + let intake_error = task.apply_operation(insert_operation(2)).unwrap_err(); + assert!(Arc::ptr_eq(&wait_error, &intake_error)); + } + + #[test] + fn runtime_shutdown_is_terminal_and_rejects_later_operations() { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .unwrap(); + let task = runtime.block_on(async { + let task = PersistenceTask::run_engine(TestEngine { + batches: Arc::new(AtomicUsize::new(0)), + events: Arc::new(ParkingMutex::new(Vec::new())), + config: TestConfig, + failure: TestFailure::None, + }); + tokio::task::yield_now().await; + task + }); + + drop(runtime); + + let verifier = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let wait_error = verifier + .block_on(async { tokio::time::timeout(Duration::from_secs(1), task.wait_for_failure()).await }) + .expect("cancelled worker must notify terminal waiters") + .unwrap_err(); + assert_eq!( + wait_error.to_string(), + "persistence engine failed: persistence worker was cancelled" + ); + + let intake_error = task.apply_operation(insert_operation(1)).unwrap_err(); + assert!(Arc::ptr_eq(&wait_error, &intake_error)); + } + #[tokio::test] async fn vacuum_reclamation_waits_for_preceding_row_moves() { let events = Arc::new(ParkingMutex::new(Vec::new())); @@ -864,7 +1005,7 @@ impl let analyzer_in_progress = Arc::new(AtomicBool::new(true)); let task_analyzer_in_progress = analyzer_in_progress.clone(); - let task = async move { + let worker = async move { let mut pending_reclaim: Option> = None; loop { let message = if pending_reclaim.is_none() { @@ -876,7 +1017,7 @@ impl message } else if analyzer.len() == 0 && pending_reclaim.is_none() { task_analyzer_in_progress.store(false, Ordering::Release); - engine_lifecycle.notify.notify_waiters(); + engine_lifecycle.progress_notify.notify_waiters(); if matches!(engine_lifecycle.state(), PersistenceState::Closing) { engine_lifecycle.finish_close(); return; @@ -951,6 +1092,13 @@ impl } } }; + // Constructed outside the async block so cancellation before its first + // poll still drops the guard and publishes terminal failure. + let completion_guard = WorkerCompletionGuard::new(lifecycle.clone()); + let task = async move { + worker.await; + completion_guard.disarm(); + }; let engine_task_handle = tokio::spawn(task); Self { queue, @@ -1002,12 +1150,27 @@ impl } tokio::select! { - _ = self.lifecycle.notify.notified() => {}, + _ = self.lifecycle.progress_notify.notified() => {}, _ = tokio::time::sleep(Duration::from_secs(1)) => {} } } } + /// Returns a cloneable monitor independent of this task's ownership. + pub fn monitor(&self) -> PersistenceMonitor { + PersistenceMonitor { + lifecycle: self.lifecycle.clone(), + } + } + + /// Waits until the worker fails or closes. + /// + /// Prefer [`Self::monitor`] when another task must keep waiting while this + /// task is moved into [`Self::close`]. + pub async fn wait_for_failure(&self) -> PersistenceResult { + self.monitor().wait_for_failure().await + } + pub async fn close(mut self) -> PersistenceResult { let begin_result = self.lifecycle.begin_close(); self.queue.wake(); diff --git a/tests/persistence/sync/mod.rs b/tests/persistence/sync/mod.rs index 1306a34..23bbb71 100644 --- a/tests/persistence/sync/mod.rs +++ b/tests/persistence/sync/mod.rs @@ -9,6 +9,7 @@ mod failure; mod failure_multi_index; mod many_strings; mod option; +mod repeated_string_upsert; mod string_primary_index; mod string_re_read; mod string_secondary_index; diff --git a/tests/persistence/sync/repeated_string_upsert.rs b/tests/persistence/sync/repeated_string_upsert.rs new file mode 100644 index 0000000..de9e97a --- /dev/null +++ b/tests/persistence/sync/repeated_string_upsert.rs @@ -0,0 +1,80 @@ +use std::time::Duration; + +use tokio::time::timeout; +use worktable::prelude::*; +use worktable::worktable; + +use crate::remove_dir_if_exists; + +worktable!( + name: StringBlob, + persist: true, + columns: { + key: String primary_key, + value: String, + updated_at: String, + }, +); + +#[test] +fn repeated_varying_string_upserts_keep_the_worker_healthy() { + let path = "tests/data/sync/repeated_string_upsert"; + let config = DiskConfig::new_with_table_name( + path, + StringBlobWorkTable::name_snake_case(), + StringBlobWorkTable::version(), + ); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_io() + .enable_time() + .build() + .unwrap(); + + runtime.block_on(async { + remove_dir_if_exists(path.to_string()).await; + { + let engine = StringBlobPersistenceEngine::new(config.clone()).await.unwrap(); + let table = StringBlobWorkTable::load(engine).await.unwrap(); + + for index in 0..256 { + table + .insert(StringBlobRow { + key: format!("marker:{index:04}"), + value: format!("initial-{index}"), + updated_at: format!("2026-08-08T00:{:02}:00Z", index % 60), + }) + .unwrap(); + } + table + .insert(StringBlobRow { + key: "settings".into(), + value: "{}".into(), + updated_at: "2026-08-08T00:00:00Z".into(), + }) + .unwrap(); + table.wait_for_ops().await.unwrap(); + + for revision in 0..1_000 { + table + .upsert(StringBlobRow { + key: "settings".into(), + value: "x".repeat(64 + revision % 4_096), + updated_at: format!("2026-08-08T01:{:02}:{:02}Z", revision % 60, revision % 60), + }) + .await + .unwrap(); + } + + timeout(Duration::from_secs(15), table.wait_for_ops()) + .await + .expect("persistence stalled after repeated string upserts") + .expect("persistence worker failed after repeated string upserts"); + } + { + let engine = StringBlobPersistenceEngine::new(config).await.unwrap(); + let table = StringBlobWorkTable::load(engine).await.unwrap(); + assert_eq!(table.select("settings".to_string()).unwrap().value.len(), 1_063); + } + }); +}