From 5396df1843e51a9002a5e78707d00e791ec01a24 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:02 +0700 Subject: [PATCH 01/15] fix: close check-then-act race in row lock acquisition The generated lock acquisition did `lock_manager.get(&pk)` and, on miss, built a lock and `insert`ed it. Two tasks could both miss, both build, and both enter the row: the loser's insert returned the winner's lock and merged into it, but the winner had already registered its operation on a lock that was no longer in the map, so its wait set was computed before the loser existed. Neither waited for the other. Observed as lost updates: test_update_val_by_id_two_thread asserting 20_000 increments intermittently saw 19_999 (~8% of full-suite runs). Fix: LockMap::get_or_insert_with does the check and the insert under one guard, with a read-lock fast path for the common already-present case so unrelated rows' acquisitions stay concurrent, and a write-lock re-check slow path so two concurrent misses cannot both construct. Both generator lock paths (custom and full-row, in-memory and persist) collapse their two-branch acquire into this call. LockGuard cleanup re-opened the same window, so remove_with_lock_check now also requires Arc::strong_count == 2 (the map entry plus the local clone) before removing an unlocked entry: any higher count is a task that already took the Arc and is about to register on it. The read fast path clones under the read guard, so cleanup (which needs the write lock) either runs before the lookup or sees the extra reference. Documented trade-off: a task cancelled between taking the Arc and registering leaves one empty entry behind until the next operation on that key; mutual exclusion is unaffected. The generated custom-update methods are generic over Pk and open with `let pk = pk.into();` whose type was only pinned by the removed `get(&pk)` call; the binding is now explicitly typed `pk: #pk_ident` (the delete generators already did this). Measured: lost-update tests 0/150 full-suite runs after, 3/40 before. --- .../src/generators/in_memory/queries/locks.rs | 76 +++++++------------ .../generators/in_memory/queries/update.rs | 2 +- .../src/generators/persist/queries/locks.rs | 76 +++++++------------ .../src/generators/persist/queries/update.rs | 2 +- src/lock/map.rs | 45 +++++++++++ 5 files changed, 99 insertions(+), 102 deletions(-) diff --git a/codegen/src/generators/in_memory/queries/locks.rs b/codegen/src/generators/in_memory/queries/locks.rs index 92f83186..1d590988 100644 --- a/codegen/src/generators/in_memory/queries/locks.rs +++ b/codegen/src/generators/in_memory/queries/locks.rs @@ -103,31 +103,17 @@ impl InMemoryGenerator { quote! { let lock_id = self.0.lock_manager.next_id(); - if let Some(lock) = self.0.lock_manager.get(&pk) { - let mut lock_guard = lock.write().await; - #[allow(clippy::mutable_key_type)] - let (locks, op_lock) = lock_guard.lock(lock_id); - drop(lock_guard); - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - - op_lock - } else { - #[allow(clippy::mutable_key_type)] - let (lock, op_lock) = #lock_ident::with_lock(lock_id); - let lock = std::sync::Arc::new(tokio::sync::RwLock::new(lock)); - let mut guard = lock.write().await; - if let Some(old_lock) = self.0.lock_manager.insert(pk.clone(), lock.clone()) { - let mut old_lock_guard = old_lock.write().await; - #[allow(clippy::mutable_key_type)] - let locks = guard.merge(&mut *old_lock_guard); - drop(old_lock_guard); - drop(guard); - - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - } - - op_lock - } + // Same atomic acquire as the per-column path: see LockMap::get_or_insert_with. + let lock = self + .0 + .lock_manager + .get_or_insert_with(pk.clone(), #lock_ident::new); + let mut lock_guard = lock.write().await; + #[allow(clippy::mutable_key_type)] + let (locks, op_lock) = lock_guard.lock(lock_id); + drop(lock_guard); + futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; + op_lock } } @@ -137,31 +123,21 @@ impl InMemoryGenerator { quote! { let lock_id = self.0.lock_manager.next_id(); - if let Some(lock) = self.0.lock_manager.get(&pk) { - let mut lock_guard = lock.write().await; - #[allow(clippy::mutable_key_type)] - let (locks, op_lock) = lock_guard.#ident(lock_id); - drop(lock_guard); - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - op_lock - } else { - let mut lock = #lock_ident::new(); - #[allow(clippy::mutable_key_type)] - let (_, op_lock) = lock.#ident(lock_id); - let lock = std::sync::Arc::new(tokio::sync::RwLock::new(lock)); - let mut guard = lock.write().await; - if let Some(old_lock) = self.0.lock_manager.insert(pk.clone(), lock.clone()) { - let mut old_lock_guard = old_lock.write().await; - #[allow(clippy::mutable_key_type)] - let locks = guard.merge(&mut *old_lock_guard); - drop(old_lock_guard); - drop(guard); - - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - } - - op_lock - } + // One atomic acquire, no check-then-act. Splitting this into `get` + // then `insert` let two tasks both miss, both build a lock and both + // enter the row: the loser merged into the winner's lock, but the + // winner had already registered its operation on a lock that was no + // longer the map's, so it never waited for the loser. + let lock = self + .0 + .lock_manager + .get_or_insert_with(pk.clone(), #lock_ident::new); + let mut lock_guard = lock.write().await; + #[allow(clippy::mutable_key_type)] + let (locks, op_lock) = lock_guard.#ident(lock_id); + drop(lock_guard); + futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; + op_lock } } } diff --git a/codegen/src/generators/in_memory/queries/update.rs b/codegen/src/generators/in_memory/queries/update.rs index 8e1c2add..d2499e54 100644 --- a/codegen/src/generators/in_memory/queries/update.rs +++ b/codegen/src/generators/in_memory/queries/update.rs @@ -449,7 +449,7 @@ impl InMemoryGenerator { pub async fn #method_ident(&self, row: #query_ident, pk: Pk) -> core::result::Result<(), WorkTableError> where #pk_ident: From { - let pk = pk.into(); + let pk: #pk_ident = pk.into(); let op_lock = { #custom_lock }; let _guard = LockGuard::new( op_lock, diff --git a/codegen/src/generators/persist/queries/locks.rs b/codegen/src/generators/persist/queries/locks.rs index e2cc9df6..934cacd3 100644 --- a/codegen/src/generators/persist/queries/locks.rs +++ b/codegen/src/generators/persist/queries/locks.rs @@ -103,31 +103,17 @@ impl PersistGenerator { quote! { let lock_id = self.0.lock_manager.next_id(); - if let Some(lock) = self.0.lock_manager.get(&pk) { - let mut lock_guard = lock.write().await; - #[allow(clippy::mutable_key_type)] - let (locks, op_lock) = lock_guard.lock(lock_id); - drop(lock_guard); - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - - op_lock - } else { - #[allow(clippy::mutable_key_type)] - let (lock, op_lock) = #lock_ident::with_lock(lock_id); - let lock = std::sync::Arc::new(tokio::sync::RwLock::new(lock)); - let mut guard = lock.write().await; - if let Some(old_lock) = self.0.lock_manager.insert(pk.clone(), lock.clone()) { - let mut old_lock_guard = old_lock.write().await; - #[allow(clippy::mutable_key_type)] - let locks = guard.merge(&mut *old_lock_guard); - drop(old_lock_guard); - drop(guard); - - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - } - - op_lock - } + // Same atomic acquire as the per-column path: see LockMap::get_or_insert_with. + let lock = self + .0 + .lock_manager + .get_or_insert_with(pk.clone(), #lock_ident::new); + let mut lock_guard = lock.write().await; + #[allow(clippy::mutable_key_type)] + let (locks, op_lock) = lock_guard.lock(lock_id); + drop(lock_guard); + futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; + op_lock } } @@ -137,31 +123,21 @@ impl PersistGenerator { quote! { let lock_id = self.0.lock_manager.next_id(); - if let Some(lock) = self.0.lock_manager.get(&pk) { - let mut lock_guard = lock.write().await; - #[allow(clippy::mutable_key_type)] - let (locks, op_lock) = lock_guard.#ident(lock_id); - drop(lock_guard); - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - op_lock - } else { - let mut lock = #lock_ident::new(); - #[allow(clippy::mutable_key_type)] - let (_, op_lock) = lock.#ident(lock_id); - let lock = std::sync::Arc::new(tokio::sync::RwLock::new(lock)); - let mut guard = lock.write().await; - if let Some(old_lock) = self.0.lock_manager.insert(pk.clone(), lock.clone()) { - let mut old_lock_guard = old_lock.write().await; - #[allow(clippy::mutable_key_type)] - let locks = guard.merge(&mut *old_lock_guard); - drop(old_lock_guard); - drop(guard); - - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - } - - op_lock - } + // One atomic acquire, no check-then-act. Splitting this into `get` + // then `insert` let two tasks both miss, both build a lock and both + // enter the row: the loser merged into the winner's lock, but the + // winner had already registered its operation on a lock that was no + // longer the map's, so it never waited for the loser. + let lock = self + .0 + .lock_manager + .get_or_insert_with(pk.clone(), #lock_ident::new); + let mut lock_guard = lock.write().await; + #[allow(clippy::mutable_key_type)] + let (locks, op_lock) = lock_guard.#ident(lock_id); + drop(lock_guard); + futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; + op_lock } } } diff --git a/codegen/src/generators/persist/queries/update.rs b/codegen/src/generators/persist/queries/update.rs index ce8e6f79..17f543d8 100644 --- a/codegen/src/generators/persist/queries/update.rs +++ b/codegen/src/generators/persist/queries/update.rs @@ -406,7 +406,7 @@ impl PersistGenerator { pub async fn #method_ident(&self, row: #query_ident, pk: Pk) -> core::result::Result<(), WorkTableError> where #pk_ident: From { - let pk = pk.into(); + let pk: #pk_ident = pk.into(); let op_lock = { #custom_lock }; let _guard = LockGuard::new( op_lock, diff --git a/src/lock/map.rs b/src/lock/map.rs index 7be0bf72..4e4b538f 100644 --- a/src/lock/map.rs +++ b/src/lock/map.rs @@ -39,6 +39,37 @@ where self.map.read().get(key).cloned() } + /// Returns the lock for `key`, inserting one built by `f` if absent. + /// + /// The check and the insert happen under a single write guard. Doing them + /// as separate `get` then `insert` calls is a check-then-act race: two + /// tasks can both observe no entry, both build a lock, and both believe + /// they hold the row. The loser's `insert` returns the winner's lock and + /// can merge into it, but the *winner* already registered its operation on + /// a lock that is no longer in the map, so it never waits for the loser and + /// both proceed into the row at once. + pub fn get_or_insert_with(&self, key: PrimaryKey, f: F) -> Arc> + where + F: FnOnce() -> LockType, + { + // Fast path: the row is usually already locked by someone, and a read + // guard keeps unrelated rows' acquisitions concurrent. The clone happens + // under the guard, so `remove_with_lock_check` (which needs the write + // lock) either runs before we looked or sees our extra strong reference + // and keeps the entry. + if let Some(lock) = self.map.read().get(&key) { + return lock.clone(); + } + let mut map = self.map.write(); + // Re-check: another task can insert between the read and write guards. + if let Some(lock) = map.get(&key) { + return lock.clone(); + } + let lock = Arc::new(tokio::sync::RwLock::new(f())); + map.insert(key, lock.clone()); + lock + } + pub fn remove(&mut self, key: &PrimaryKey) { self.map.write().remove(key); } @@ -51,6 +82,20 @@ where if let Some(lock) = set.get(key).cloned() && let Ok(guard) = lock.try_read() && !guard.is_locked() + // Two strong references means this map entry and our own `lock` + // clone above, and nothing else. Any higher count is a task that + // has already taken this Arc out of `get_or_insert_with` and is + // about to register on it; removing the entry now would let the + // next caller build a *second* lock for the same row, and the two + // would not serialise against each other. + // + // Known trade-off: if that other task is cancelled between taking + // the Arc and registering its operation, nothing re-triggers this + // cleanup and the (unlocked, unused) entry stays in the map until + // the next operation on the same key drops its guard. That leaks at + // most one empty lock per abandoned key and never affects mutual + // exclusion. + && Arc::strong_count(&lock) == 2 { set.remove(key); } From 1fdfb801c4912871712a94c7adadfb695ac58692 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:02 +0700 Subject: [PATCH 02/15] fix: close the same lock-acquisition race in vacuum EmptyDataVacuum::full_row_lock had a copy of the check-then-act pattern fixed in the generated lock paths: get, and on miss build-and-insert, letting a vacuum move and a row operation both believe they hold the row. Collapse it into the same atomic LockMap::get_or_insert_with call. The generic vacuum only knows `LockType: RowLock`, so the trait gains `fn new()` (a lock with no columns held). This is a deliberate breaking addition to a public trait, acceptable within the unreleased 0.9 line; all real implementations are generated by this crate's own macro. The three lock generators implement it by delegating to the generated inherent `new`; FullRowLock uses the new explicit `Lock::new_released` constructor for its placeholder state ('released' names the acquisition flag, distinguishing it from `Lock::new`, which starts held) instead of constructing a held lock and immediately releasing it. --- codegen/src/generators/in_memory/locks.rs | 3 ++ codegen/src/generators/persist/locks.rs | 3 ++ codegen/src/generators/read_only/locks.rs | 3 ++ src/lock/mod.rs | 13 +++++++++ src/lock/row_lock.rs | 17 ++++++++++++ src/table/vacuum/vacuum.rs | 34 ++++++----------------- 6 files changed, 48 insertions(+), 25 deletions(-) diff --git a/codegen/src/generators/in_memory/locks.rs b/codegen/src/generators/in_memory/locks.rs index 41422774..78b40aec 100644 --- a/codegen/src/generators/in_memory/locks.rs +++ b/codegen/src/generators/in_memory/locks.rs @@ -63,6 +63,9 @@ impl InMemoryGenerator { quote! { impl RowLock for #lock_ident { + fn new() -> Self { + #lock_ident::new() + } #is_locked_fn #lock_fn #with_lock_fn diff --git a/codegen/src/generators/persist/locks.rs b/codegen/src/generators/persist/locks.rs index 90547a24..92a88a0c 100644 --- a/codegen/src/generators/persist/locks.rs +++ b/codegen/src/generators/persist/locks.rs @@ -63,6 +63,9 @@ impl PersistGenerator { quote! { impl RowLock for #lock_ident { + fn new() -> Self { + #lock_ident::new() + } #is_locked_fn #lock_fn #with_lock_fn diff --git a/codegen/src/generators/read_only/locks.rs b/codegen/src/generators/read_only/locks.rs index 6e3a46a9..280afd28 100644 --- a/codegen/src/generators/read_only/locks.rs +++ b/codegen/src/generators/read_only/locks.rs @@ -63,6 +63,9 @@ impl ReadOnlyGenerator { quote! { impl RowLock for #lock_ident { + fn new() -> Self { + #lock_ident::new() + } #is_locked_fn #lock_fn #with_lock_fn diff --git a/src/lock/mod.rs b/src/lock/mod.rs index 578fe003..90432eed 100644 --- a/src/lock/mod.rs +++ b/src/lock/mod.rs @@ -106,6 +106,19 @@ impl Lock { } } + /// A lock born in the released state (`is_locked() == false`, waiting on + /// it returns immediately). Used for placeholder state, e.g. a fresh + /// [`FullRowLock`](crate::lock::FullRowLock) that no operation holds yet. + /// "Released" refers to the acquisition flag, distinguishing it from + /// [`Lock::new`], which starts held by the creating operation. + pub fn new_released(id: u16) -> Self { + Self { + id, + locked: Arc::new(AtomicBool::new(false)), + wakers: Mutex::new(vec![]), + } + } + pub fn id(&self) -> u16 { self.id } diff --git a/src/lock/row_lock.rs b/src/lock/row_lock.rs index e874373e..9e8aff49 100644 --- a/src/lock/row_lock.rs +++ b/src/lock/row_lock.rs @@ -6,6 +6,10 @@ use std::sync::Arc; use crate::lock::{Lock, LockGuard, LockMap, LockWait}; pub trait RowLock { + /// Creates a new [`RowLock`] with no columns locked. + fn new() -> Self + where + Self: Sized; /// Checks if any column of this row is locked. fn is_locked(&self) -> bool; /// Creates new [`RowLock`] with all columns locked. @@ -52,6 +56,19 @@ impl FullRowLock { #[allow(clippy::mutable_key_type)] impl RowLock for FullRowLock { + fn new() -> Self + where + Self: Sized, + { + // Placeholder: no operation holds this row yet, so the initial lock is + // born released and any wait on it completes immediately. The id is + // never observed because `lock()` replaces the placeholder before + // handing anything out. + FullRowLock { + l: Arc::new(Lock::new_released(0)), + } + } + fn is_locked(&self) -> bool { self.l.is_locked() } diff --git a/src/table/vacuum/vacuum.rs b/src/table/vacuum/vacuum.rs index f5b87e47..a374db9c 100644 --- a/src/table/vacuum/vacuum.rs +++ b/src/table/vacuum/vacuum.rs @@ -273,31 +273,15 @@ where async fn full_row_lock(&self, pk: &PrimaryKey) -> Arc { let lock_id = self.lock_manager.next_id(); - if let Some(lock) = self.lock_manager.get(pk) { - let mut lock_guard = lock.write().await; - #[allow(clippy::mutable_key_type)] - let (locks, op_lock) = lock_guard.lock(lock_id); - drop(lock_guard); - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - - op_lock - } else { - #[allow(clippy::mutable_key_type)] - let (lock, op_lock) = LockType::with_lock(lock_id); - let lock = Arc::new(tokio::sync::RwLock::new(lock)); - let mut guard = lock.write().await; - if let Some(old_lock) = self.lock_manager.insert(pk.clone(), lock.clone()) { - let mut old_lock_guard = old_lock.write().await; - #[allow(clippy::mutable_key_type)] - let locks = guard.merge(&mut *old_lock_guard); - drop(old_lock_guard); - drop(guard); - - futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; - } - - op_lock - } + // One atomic acquire, no check-then-act: see LockMap::get_or_insert_with. + let lock = self.lock_manager.get_or_insert_with(pk.clone(), LockType::new); + let mut lock_guard = lock.write().await; + #[allow(clippy::mutable_key_type)] + let (locks, op_lock) = lock_guard.lock(lock_id); + drop(lock_guard); + futures::future::join_all(locks.iter().map(|l| l.wait()).collect::>()).await; + + op_lock } fn update_index_after_move(&self, pk: PrimaryKey, old_link: Link, new_link: Link) { From 4230e0bec01313d9e9ef15ee36ac8ed450e87de6 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:02 +0700 Subject: [PATCH 03/15] fix: upsert retries existence flips instead of surfacing them Generated upsert did `pk_map.get(&pk)` then update() or insert() - a check-then-act window. A concurrent delete of the same key between the check and the update made upsert return NotFound, which the caller never expects from an upsert. Observed via tests/worktable/vacuum.rs::vacuum_parallel_with_upserts (~5% of full-suite runs), whose delete task races 3000 upserts. Fix: retry loop over three transient conflicts, each of which proves a concurrent operation on the same key completed: * update() hitting NotFound (concurrent delete) retries as insert; * insert() hitting a primary-key conflict (concurrent insert) retries as update, via the new typed WorkTableError::PrimaryAlreadyExists (replacing a stringly AlreadyExists("Primary") comparison); * update() hitting a row-absent PagesError retries: raw insert takes no row lock and publishes the pk entry before unghosting the row data, so a reader in that window sees Ghosted for a row that is about to exist (the underlying publish-order issue is tracked in #169). The classification comes from the new ExecutionError::is_row_absent (Ghosted / Deleted / Vacuumed / PageNotFound), which the later snapshot patches reuse. Secondary unique-index conflicts propagate: retrying them can never succeed. Progress semantics, documented on the generated method: system-wide lock-free, NOT wait-free per call. A retry is only taken when another operation on the key completed in the window, so the system makes progress on every iteration, but an individual call can retry unboundedly under sustained adversarial same-key churn. There is deliberately no retry limit - upsert is semantically infallible for primary-key conflicts, and a limit would trade theoretical starvation for real spurious errors. Each conflicting round yields to the scheduler (tokio::task::yield_now - no timer dependency, so it cannot panic on runtimes without enable_time) before retrying. Tests: tests/worktable/upsert.rs drives sustained existence-flipping churn on one key against four concurrent upserters under a timeout that turns starvation into a failure. Both tiers are #[ignore]d for now: they expose pre-existing engine races between the lock-free insert path and locked mutations (panics and stalls that reproduce WITHOUT upsert involvement; the moderate tier is 40/40 stable solo but still hits the stall ~1/30 under full-suite parallel load). Repro recipes and analysis in #169; un-ignore both tiers when it lands. Measured: vacuum_parallel_with_upserts 0/100 full-suite runs after, 2-3/50 before. --- .../src/generators/in_memory/table/impls.rs | 62 +++++++++-- codegen/src/generators/persist/table/impls.rs | 62 +++++++++-- src/in_memory/pages.rs | 11 ++ src/table/mod.rs | 6 +- tests/worktable/mod.rs | 1 + tests/worktable/upsert.rs | 101 ++++++++++++++++++ 6 files changed, 221 insertions(+), 22 deletions(-) create mode 100644 tests/worktable/upsert.rs diff --git a/codegen/src/generators/in_memory/table/impls.rs b/codegen/src/generators/in_memory/table/impls.rs index e3c26d25..b647c68c 100644 --- a/codegen/src/generators/in_memory/table/impls.rs +++ b/codegen/src/generators/in_memory/table/impls.rs @@ -148,21 +148,63 @@ impl InMemoryGenerator { let row_type = name_generator.get_row_type_ident(); quote! { + /// Inserts the row if its primary key is absent, updates it + /// otherwise. + /// + /// Concurrency: **system-wide lock-free, not wait-free per call.** + /// A retry is only taken when a concurrent delete or insert + /// flipped this key's existence between the existence check and + /// the operation, so some operation on this key completes on + /// every iteration -- but under sustained adversarial churn on + /// the same key an individual call can retry indefinitely. There + /// is deliberately no retry limit: upsert is semantically + /// infallible for primary-key conflicts, and a limit would trade + /// theoretical starvation for real spurious errors. Each + /// conflicting round yields to the scheduler before retrying so + /// the interfering task can complete. pub async fn upsert(&self, row: #row_type) -> core::result::Result<(), WorkTableError> { let pk = row.get_primary_key(); - let need_to_update = { - if let Some(link) = self.0.primary_index.pk_map.get(&pk) { - true + loop { + let need_to_update = self.0.primary_index.pk_map.get(&pk).is_some(); + if need_to_update { + match self.update(row.clone()).await { + core::result::Result::Ok(_) => return core::result::Result::Ok(()), + // Row was deleted concurrently between the check and the + // update; retry as an insert. + core::result::Result::Err(WorkTableError::NotFound) => { + tokio::task::yield_now().await; + continue; + } + // Row is mid-flight: a concurrent insert publishes + // the primary-key entry before unghosting the row + // data (and insert takes no row lock), and a + // concurrent delete ghosts data it is about to + // unindex. Both are transient; retry. + core::result::Result::Err(WorkTableError::PagesError(e)) if e.is_row_absent() => { + tokio::task::yield_now().await; + continue; + } + core::result::Result::Err(e) => return core::result::Result::Err(e), + } } else { - false + match self.insert(row.clone()) { + core::result::Result::Ok(_) => return core::result::Result::Ok(()), + // Row was inserted concurrently between the check and the + // insert; retry as an update. Secondary-index conflicts are + // real errors and are propagated. Progress is lock-free, + // not wait-free: a retry is only taken when a concurrent + // delete/insert flipped this key's existence between the + // check and the operation, so the system as a whole makes + // progress on every retry, but this call can in principle + // retry unboundedly under sustained same-key churn. + core::result::Result::Err(WorkTableError::PrimaryAlreadyExists) => { + tokio::task::yield_now().await; + continue; + } + core::result::Result::Err(e) => return core::result::Result::Err(e), + } } - }; - if need_to_update { - self.update(row).await?; - } else { - self.insert(row)?; } - core::result::Result::Ok(()) } } } diff --git a/codegen/src/generators/persist/table/impls.rs b/codegen/src/generators/persist/table/impls.rs index 1e7ca35b..cc19ea29 100644 --- a/codegen/src/generators/persist/table/impls.rs +++ b/codegen/src/generators/persist/table/impls.rs @@ -244,21 +244,63 @@ impl PersistGenerator { let row_type = name_generator.get_row_type_ident(); quote! { + /// Inserts the row if its primary key is absent, updates it + /// otherwise. + /// + /// Concurrency: **system-wide lock-free, not wait-free per call.** + /// A retry is only taken when a concurrent delete or insert + /// flipped this key's existence between the existence check and + /// the operation, so some operation on this key completes on + /// every iteration -- but under sustained adversarial churn on + /// the same key an individual call can retry indefinitely. There + /// is deliberately no retry limit: upsert is semantically + /// infallible for primary-key conflicts, and a limit would trade + /// theoretical starvation for real spurious errors. Each + /// conflicting round yields to the scheduler before retrying so + /// the interfering task can complete. pub async fn upsert(&self, row: #row_type) -> core::result::Result<(), WorkTableError> { let pk = row.get_primary_key(); - let need_to_update = { - if let Some(link) = self.0.primary_index.pk_map.get(&pk) { - true + loop { + let need_to_update = self.0.primary_index.pk_map.get(&pk).is_some(); + if need_to_update { + match self.update(row.clone()).await { + core::result::Result::Ok(_) => return core::result::Result::Ok(()), + // Row was deleted concurrently between the check and the + // update; retry as an insert. + core::result::Result::Err(WorkTableError::NotFound) => { + tokio::task::yield_now().await; + continue; + } + // Row is mid-flight: a concurrent insert publishes + // the primary-key entry before unghosting the row + // data (and insert takes no row lock), and a + // concurrent delete ghosts data it is about to + // unindex. Both are transient; retry. + core::result::Result::Err(WorkTableError::PagesError(e)) if e.is_row_absent() => { + tokio::task::yield_now().await; + continue; + } + core::result::Result::Err(e) => return core::result::Result::Err(e), + } } else { - false + match self.insert(row.clone()) { + core::result::Result::Ok(_) => return core::result::Result::Ok(()), + // Row was inserted concurrently between the check and the + // insert; retry as an update. Secondary-index conflicts are + // real errors and are propagated. Progress is lock-free, + // not wait-free: a retry is only taken when a concurrent + // delete/insert flipped this key's existence between the + // check and the operation, so the system as a whole makes + // progress on every retry, but this call can in principle + // retry unboundedly under sustained same-key churn. + core::result::Result::Err(WorkTableError::PrimaryAlreadyExists) => { + tokio::task::yield_now().await; + continue; + } + core::result::Result::Err(e) => return core::result::Result::Err(e), + } } - }; - if need_to_update { - self.update(row).await?; - } else { - self.insert(row)?; } - core::result::Result::Ok(()) } } } diff --git a/src/in_memory/pages.rs b/src/in_memory/pages.rs index 18255825..08a41fb8 100644 --- a/src/in_memory/pages.rs +++ b/src/in_memory/pages.rs @@ -448,6 +448,17 @@ impl ExecutionError { pub fn is_vacuumed(&self) -> bool { matches!(self, Self::Vacuumed) } + + /// True when the error means "no row lives at this link (any more)" — + /// the row was deleted, ghosted, vacuumed away, or its page is gone. + /// Snapshot-building code skips such candidates; every other variant is + /// a real storage error and must propagate. + pub fn is_row_absent(&self) -> bool { + matches!( + self, + Self::Ghosted | Self::Deleted | Self::Vacuumed | Self::PageNotFound(_) + ) + } } #[cfg(test)] diff --git a/src/table/mod.rs b/src/table/mod.rs index 8385ea5b..cb54aaa9 100644 --- a/src/table/mod.rs +++ b/src/table/mod.rs @@ -179,7 +179,7 @@ where let link = self.data.insert(row.clone()).map_err(WorkTableError::PagesError)?; if self.primary_index.insert_checked(pk.clone(), link).is_none() { self.data.delete(link).map_err(WorkTableError::PagesError)?; - return Err(WorkTableError::AlreadyExists("Primary".to_string())); + return Err(WorkTableError::PrimaryAlreadyExists); }; if let Err(e) = self.indexes.save_row(row.clone(), link) { return match e { @@ -237,7 +237,7 @@ where if let Err(e) = self.data.delete(link) { return (None, Err(WorkTableError::PagesError(e))); } - return (None, Err(WorkTableError::AlreadyExists("Primary".to_string()))); + return (None, Err(WorkTableError::PrimaryAlreadyExists)); }; let primary_key_events = convert_change_events(primary_key_events); @@ -516,6 +516,8 @@ pub enum WorkTableError { NotFound, #[display("Value already exists for `{}` index", _0)] AlreadyExists(#[error(not(source))] String), + #[display("Row with this primary key already exists")] + PrimaryAlreadyExists, SerializeError, SecondaryIndexError, PrimaryUpdateTry, diff --git a/tests/worktable/mod.rs b/tests/worktable/mod.rs index 9122137b..b4db3042 100644 --- a/tests/worktable/mod.rs +++ b/tests/worktable/mod.rs @@ -12,6 +12,7 @@ mod nid; mod option; mod tuple_primary_key; mod unsized_; +mod upsert; mod uuid; mod vacuum; mod with_enum; diff --git a/tests/worktable/upsert.rs b/tests/worktable/upsert.rs new file mode 100644 index 00000000..47b87cb2 --- /dev/null +++ b/tests/worktable/upsert.rs @@ -0,0 +1,101 @@ +use std::sync::Arc; +use std::time::Duration; + +use tokio::time::timeout; +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: UpsertChurn, + columns: { + id: u64 primary_key, + val: u64, + }, +); + +/// Upsert is system-wide lock-free but not wait-free per call: a retry is +/// taken exactly when a concurrent delete/insert flips the key's existence +/// between the check and the operation. This test drives sustained +/// adversarial churn on ONE key while several tasks upsert it and asserts +/// that every upsert completes without surfacing a spurious conflict error, +/// under a timeout that turns pathological starvation into a failure +/// instead of a hang. +/// Moderate tier: 40/40 stable when run solo, but under full-suite parallel +/// load the pre-existing #169 stall still fires (~1/30 suite runs), so both +/// tiers stay ignored until that lands. Run with `-- --ignored` to validate +/// the upsert retry behavior. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "exposes the pre-existing #169 lock-free-insert stall under suite load"] +async fn upsert_completes_under_same_key_churn() { + churn_run(100, 200).await; +} + +/// Intense variant: reliably exposes pre-existing engine races under extreme +/// same-key churn that are unrelated to the upsert retry loop (raw `insert` +/// takes no row lock and publishes the pk entry before unghosting the data; +/// under saturation a churn round can stall past the timeout). Tracked in the +/// lock-free-insert issue; un-ignore when that lands. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "exposes pre-existing lock-free-insert races under extreme same-key churn"] +async fn upsert_completes_under_extreme_same_key_churn() { + churn_run(5_000, 2_000).await; +} + +async fn churn_run(churn_flips: u64, upserts_per_task: u64) { + #[allow(non_snake_case)] + let CHURN_FLIPS = churn_flips; + #[allow(non_snake_case)] + let UPSERTS_PER_TASK = upserts_per_task; + let table = Arc::new(UpsertChurnWorkTable::default()); + const KEY: u64 = 7; + + let churn = { + let table = table.clone(); + tokio::spawn(async move { + for i in 0..CHURN_FLIPS { + // Flip the key's existence as fast as possible through the + // locked operations. (Raw `insert` is deliberately not used + // here: it takes no row lock and publishes the pk entry + // before unghosting the data, which trips unrelated + // pre-existing races tracked separately in the issue on + // lock-free insert vs locked mutations.) + table + .upsert(UpsertChurnRow { id: KEY, val: i }) + .await + .expect("churn upsert must not fail"); + let _ = table.delete(KEY).await; + } + }) + }; + + let mut upserters = Vec::new(); + for w in 0..4u64 { + let table = table.clone(); + upserters.push(tokio::spawn(async move { + for i in 0..UPSERTS_PER_TASK { + table + .upsert(UpsertChurnRow { + id: KEY, + val: w * 10_000 + i, + }) + .await + .expect("upsert must never surface a primary-key conflict"); + } + })); + } + + timeout(Duration::from_secs(60), churn) + .await + .expect("churn task starved") + .unwrap(); + for handle in upserters { + timeout(Duration::from_secs(60), handle) + .await + .expect("upserter starved") + .unwrap(); + } + + // Quiesced: a final upsert must land and be visible. + table.upsert(UpsertChurnRow { id: KEY, val: 424_242 }).await.unwrap(); + assert_eq!(table.select(KEY).map(|r| r.val), Some(424_242)); +} From 559d8aff163f5b591edb1aea764ac297fdf33d64 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 04/15] fix: multi-row update locks one validated snapshot, predicate included Three problems in `update_X_by_`: 1. It scanned the index twice: once to acquire row locks and again to choose the rows to process. A row entering the matched range between the scans was processed without any held lock, and in the unsized branch `guards.remove(&pk).expect(..)` would panic on it. 2. A non-unique index iterates equal keys in random-discriminator order (indexset MultiPair uses fastrand), so the scanned order differed on every table load. When the update fails partway (unique-index conflict on a later row), which rows were already updated - in memory AND on disk - was a coin flip. This was the actual cause of test_update_unsized_larger_last_fail (~50% of full-suite runs); the persisted state faithfully recorded a legitimate alternate outcome, there was no persistence-layer rollback bug. 3. The per-query lock set only covered the updated columns, so nothing stopped a concurrent operation from rewriting the predicate column between a revalidation read and the update. Fix: snapshot the matching rows' primary keys once, sort them (one global lock-acquisition order for concurrent multi-row operations), acquire exactly those locks, then process exactly that set. The generated per-query lock now includes the predicate (`by`) column in addition to the updated columns, so the re-validation read under the held lock is authoritative: rows deleted or updated out of the matched range before their lock was taken are skipped, rows that joined after the snapshot are not touched (statement-start snapshot semantics, revalidated under lock). Snapshot resolution distinguishes "row is gone" from real storage errors via ExecutionError::is_row_absent (introduced with the upsert fix): absent rows silently leave the snapshot, anything else (corrupt page, invalid link) propagates instead of masquerading as an empty slot. Measured: 0/50 full-suite runs after, 19-28/50 before; 40/40 after the predicate-lock revision. --- .../src/generators/in_memory/queries/locks.rs | 12 +++- .../generators/in_memory/queries/update.rs | 58 +++++++++++++++---- .../src/generators/persist/queries/locks.rs | 12 +++- .../src/generators/persist/queries/update.rs | 58 +++++++++++++++---- 4 files changed, 116 insertions(+), 24 deletions(-) diff --git a/codegen/src/generators/in_memory/queries/locks.rs b/codegen/src/generators/in_memory/queries/locks.rs index 1d590988..b6400084 100644 --- a/codegen/src/generators/in_memory/queries/locks.rs +++ b/codegen/src/generators/in_memory/queries/locks.rs @@ -58,8 +58,16 @@ impl InMemoryGenerator { let lock_ident = WorktableNameGenerator::get_update_query_lock_ident(&snake_case_name); - let columns = &updates.get(name).as_ref().expect("exists").columns; - let lock_fn = Self::gen_rows_lock_fn(columns, lock_ident); + let op = updates.get(name).expect("exists"); + // The lock set covers the updated columns AND the predicate + // (`by`) column: multi-row updates re-validate the predicate + // under this lock, which is only sound if no concurrent + // operation can rewrite that column while it is held. + let mut columns = op.columns.clone(); + if !columns.contains(&op.by) { + columns.push(op.by.clone()); + } + let lock_fn = Self::gen_rows_lock_fn(&columns, lock_ident); quote! { #lock_fn diff --git a/codegen/src/generators/in_memory/queries/update.rs b/codegen/src/generators/in_memory/queries/update.rs index d2499e54..5d75168a 100644 --- a/codegen/src/generators/in_memory/queries/update.rs +++ b/codegen/src/generators/in_memory/queries/update.rs @@ -1,7 +1,7 @@ use proc_macro2::Literal; use std::collections::HashMap; -use crate::common::model::Operation; +use crate::common::model::{Index, Operation}; use crate::common::name_generator::{WorktableNameGenerator, is_float}; use crate::generators::in_memory::InMemoryGenerator; use convert_case::{Case, Casing}; @@ -169,7 +169,7 @@ impl InMemoryGenerator { self.gen_non_unique_update( snake_case_name, name, - index_name, + index, idents, indexes_columns.as_ref(), unsized_columns, @@ -489,11 +489,13 @@ impl InMemoryGenerator { &self, snake_case_name: String, name: &Ident, - index: &Ident, + index: &Index, idents: &[Ident], idx_idents: Option<&Vec>, unsized_fields: Option>, ) -> TokenStream { + let by_field = &index.field; + let index = &index.name; let method_ident = Ident::new(format!("update_{snake_case_name}").as_str(), Span::mixed_site()); let query_ident = Ident::new(format!("{name}Query").as_str(), Span::mixed_site()); @@ -573,19 +575,55 @@ impl InMemoryGenerator { quote! { pub async fn #method_ident(&self, row: #query_ident, by: #by_ident) -> core::result::Result<(), WorkTableError> { - let links: Vec<_> = self.0.indexes.#index.get(#by).map(|(_, l)| l.0).collect(); + // Snapshot the matching rows' primary keys once; the same set + // is locked and then processed. Locking one index scan and + // processing a fresh second scan would let rows that joined the + // range in between be processed without a held lock. The keys + // are sorted so concurrent multi-row operations acquire row + // locks in one global order (a non-unique index iterates equal + // keys in random-discriminator order, which would otherwise + // also make the partially-updated subset on a mid-way failure + // nondeterministic). + let mut pks: Vec<_> = Vec::new(); + for link in self.0.indexes.#index.get(#by).map(|(_, l)| l.0) { + match self.0.data.select_non_ghosted(link) { + core::result::Result::Ok(r) => pks.push(r.get_primary_key()), + // The row vanished between the index read and the + // resolve; it is simply not part of the snapshot. + core::result::Result::Err(e) if e.is_row_absent() => {} + // Anything else (corrupt page, invalid link, ...) is a + // real storage error, not an empty snapshot slot. + core::result::Result::Err(e) => { + return core::result::Result::Err(WorkTableError::PagesError(e)); + } + } + } + pks.sort_unstable(); + pks.dedup(); let mut guards: std::collections::HashMap<_, _> = std::collections::HashMap::new(); - for link in links.iter() { - let pk = self.0.data.select_non_ghosted(*link)?.get_primary_key().clone(); + for pk in pks.iter() { + let pk = pk.clone(); let op_lock = { #custom_lock }; - guards.insert(pk.clone(), LockGuard::new(op_lock, self.0.lock_manager.clone(), pk.clone())); + guards.insert(pk.clone(), LockGuard::new(op_lock, self.0.lock_manager.clone(), pk)); } - let links: Vec<_> = self.0.indexes.#index.get(#by).map(|(_, l)| l.0).collect(); let op_id = OperationId::Multi(uuid::Uuid::now_v7()); - for link in links.into_iter() { - let pk = self.0.data.select_non_ghosted(link)?.get_primary_key().clone(); + for pk in pks.into_iter() { + // Re-resolve and re-validate under the held lock. The + // query's lock set includes the predicate column, so the + // value read here cannot be rewritten concurrently while + // the lock is held. Rows deleted or updated out of the + // matched range before their lock was acquired are + // skipped; rows that joined the range after the snapshot + // are not touched. + let link: Link = match self.0.primary_index.pk_map.get(&pk) { + Some(v) => v.get().value.into(), + None => continue, + }; + if self.0.data.select_non_ghosted(link)?.#by_field != by { + continue; + } let mut bytes = rkyv::to_bytes::(&row) .map_err(|_| WorkTableError::SerializeError)?; diff --git a/codegen/src/generators/persist/queries/locks.rs b/codegen/src/generators/persist/queries/locks.rs index 934cacd3..2b8c0339 100644 --- a/codegen/src/generators/persist/queries/locks.rs +++ b/codegen/src/generators/persist/queries/locks.rs @@ -58,8 +58,16 @@ impl PersistGenerator { let lock_ident = WorktableNameGenerator::get_update_query_lock_ident(&snake_case_name); - let columns = &updates.get(name).as_ref().expect("exists").columns; - let lock_fn = Self::gen_rows_lock_fn(columns, lock_ident); + let op = updates.get(name).expect("exists"); + // The lock set covers the updated columns AND the predicate + // (`by`) column: multi-row updates re-validate the predicate + // under this lock, which is only sound if no concurrent + // operation can rewrite that column while it is held. + let mut columns = op.columns.clone(); + if !columns.contains(&op.by) { + columns.push(op.by.clone()); + } + let lock_fn = Self::gen_rows_lock_fn(&columns, lock_ident); quote! { #lock_fn diff --git a/codegen/src/generators/persist/queries/update.rs b/codegen/src/generators/persist/queries/update.rs index 17f543d8..6a72f936 100644 --- a/codegen/src/generators/persist/queries/update.rs +++ b/codegen/src/generators/persist/queries/update.rs @@ -1,7 +1,7 @@ use proc_macro2::Literal; use std::collections::HashMap; -use crate::common::model::Operation; +use crate::common::model::{Index, Operation}; use crate::common::name_generator::{WorktableNameGenerator, is_float}; use crate::generators::persist::PersistGenerator; use convert_case::{Case, Casing}; @@ -169,7 +169,7 @@ impl PersistGenerator { self.gen_non_unique_update( snake_case_name, name, - index_name, + index, idents, indexes_columns.as_ref(), unsized_columns, @@ -446,11 +446,13 @@ impl PersistGenerator { &self, snake_case_name: String, name: &Ident, - index: &Ident, + index: &Index, idents: &[Ident], idx_idents: Option<&Vec>, unsized_fields: Option>, ) -> TokenStream { + let by_field = &index.field; + let index = &index.name; let method_ident = Ident::new(format!("update_{snake_case_name}").as_str(), Span::mixed_site()); let query_ident = Ident::new(format!("{name}Query").as_str(), Span::mixed_site()); @@ -530,19 +532,55 @@ impl PersistGenerator { quote! { pub async fn #method_ident(&self, row: #query_ident, by: #by_ident) -> core::result::Result<(), WorkTableError> { - let links: Vec<_> = self.0.indexes.#index.get(#by).map(|(_, l)| l.0).collect(); + // Snapshot the matching rows' primary keys once; the same set + // is locked and then processed. Locking one index scan and + // processing a fresh second scan would let rows that joined the + // range in between be processed without a held lock. The keys + // are sorted so concurrent multi-row operations acquire row + // locks in one global order (a non-unique index iterates equal + // keys in random-discriminator order, which would otherwise + // also make the partially-updated subset on a mid-way failure + // nondeterministic). + let mut pks: Vec<_> = Vec::new(); + for link in self.0.indexes.#index.get(#by).map(|(_, l)| l.0) { + match self.0.data.select_non_ghosted(link) { + core::result::Result::Ok(r) => pks.push(r.get_primary_key()), + // The row vanished between the index read and the + // resolve; it is simply not part of the snapshot. + core::result::Result::Err(e) if e.is_row_absent() => {} + // Anything else (corrupt page, invalid link, ...) is a + // real storage error, not an empty snapshot slot. + core::result::Result::Err(e) => { + return core::result::Result::Err(WorkTableError::PagesError(e)); + } + } + } + pks.sort_unstable(); + pks.dedup(); let mut guards: std::collections::HashMap<_, _> = std::collections::HashMap::new(); - for link in links.iter() { - let pk = self.0.data.select_non_ghosted(*link)?.get_primary_key().clone(); + for pk in pks.iter() { + let pk = pk.clone(); let op_lock = { #custom_lock }; - guards.insert(pk.clone(), LockGuard::new(op_lock, self.0.lock_manager.clone(), pk.clone())); + guards.insert(pk.clone(), LockGuard::new(op_lock, self.0.lock_manager.clone(), pk)); } - let links: Vec<_> = self.0.indexes.#index.get(#by).map(|(_, l)| l.0).collect(); let op_id = OperationId::Multi(uuid::Uuid::now_v7()); - for link in links.into_iter() { - let pk = self.0.data.select_non_ghosted(link)?.get_primary_key().clone(); + for pk in pks.into_iter() { + // Re-resolve and re-validate under the held lock. The + // query's lock set includes the predicate column, so the + // value read here cannot be rewritten concurrently while + // the lock is held. Rows deleted or updated out of the + // matched range before their lock was acquired are + // skipped; rows that joined the range after the snapshot + // are not touched. + let link: Link = match self.0.primary_index.pk_map.get(&pk) { + Some(v) => v.get().value.into(), + None => continue, + }; + if self.0.data.select_non_ghosted(link)?.#by_field != by { + continue; + } let mut bytes = rkyv::to_bytes::(&row) .map_err(|_| WorkTableError::SerializeError)?; From 8ea32700d8a0e2167c8d508c68c339236a81a56d Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 05/15] fix: delete-by-non-unique-index snapshots validated primary keys The previous shape collected storage links, sorted them, and later resolved each link to a row to delete. Storage links are not stable row identities: a concurrent delete can free a slot and an insert can reuse it for an unrelated row before the loop runs, so a stale link could resolve to - and delete - a row that never matched the query. Fix: snapshot the matching rows as primary keys, validated against the predicate at resolve time (a reused slot only stays in the set if the row now living there genuinely matches). Keys are sorted and deduped for a deterministic delete order (non-unique indexes iterate equal keys in random-discriminator order), and each delete goes through the ordinary single-row delete, which takes its own row lock and resolves by primary key, never through the snapshotted link. A row already deleted concurrently (NotFound) is skipped: the goal state for that row is reached. Row-absent resolution errors leave the snapshot; other storage errors propagate (ExecutionError::is_row_absent). --- .../generators/in_memory/queries/delete.rs | 51 ++++++++++++++++--- .../src/generators/persist/queries/delete.rs | 51 ++++++++++++++++--- 2 files changed, 90 insertions(+), 12 deletions(-) diff --git a/codegen/src/generators/in_memory/queries/delete.rs b/codegen/src/generators/in_memory/queries/delete.rs index 80142494..9f08eb41 100644 --- a/codegen/src/generators/in_memory/queries/delete.rs +++ b/codegen/src/generators/in_memory/queries/delete.rs @@ -4,6 +4,7 @@ use convert_case::{Case, Casing}; use proc_macro2::{Ident, Span, TokenStream}; use quote::quote; +use crate::common::model::Index; use crate::common::model::Operation; use crate::common::name_generator::{WorktableNameGenerator, is_float}; use crate::generators::in_memory::InMemoryGenerator; @@ -148,7 +149,7 @@ impl InMemoryGenerator { if index.is_unique { Self::gen_unique_delete(type_, &method_ident, index_name) } else { - Self::gen_non_unique_delete(type_, &method_ident, index_name) + Self::gen_non_unique_delete(type_, &method_ident, index) } } else { Self::gen_brute_force_delete_field(&op.by, type_, &method_ident) @@ -180,7 +181,9 @@ impl InMemoryGenerator { } } - fn gen_non_unique_delete(type_: &TokenStream, name: &Ident, index: &Ident) -> TokenStream { + fn gen_non_unique_delete(type_: &TokenStream, name: &Ident, index: &Index) -> TokenStream { + let by_field = &index.field; + let index = &index.name; let by = if is_float(type_.to_string().as_str()) { quote! { &OrderedFloat(by) @@ -192,10 +195,46 @@ impl InMemoryGenerator { }; quote! { pub async fn #name(&self, by: #type_) -> core::result::Result<(), WorkTableError> { - let rows_to_update = self.0.indexes.#index.get(#by).map(|kv| kv.1).collect::>(); - for link in rows_to_update { - let row = self.0.data.select_non_ghosted(link.0).map_err(WorkTableError::PagesError)?; - self.delete(row.get_primary_key()).await?; + // Snapshot the matching rows as validated primary keys before + // deleting anything. Storage links are not stable identities: + // a concurrent delete can free a slot and an insert can reuse + // it for an unrelated row before this loop runs, so resolving + // a stale link later could delete the wrong row. Every + // candidate is resolved and checked against the predicate at + // snapshot time; a reused slot only stays in the set if the + // row now living there genuinely matches. Keys are sorted for + // a deterministic delete order (non-unique indexes iterate + // equal keys in random-discriminator order) and the per-row + // delete takes its own row lock and resolves by primary key, + // never through the snapshotted link. + let mut pks: Vec<_> = Vec::new(); + for link in self.0.indexes.#index.get(#by).map(|kv| kv.1.0) { + match self.0.data.select_non_ghosted(link) { + core::result::Result::Ok(r) => { + if r.#by_field == by { + pks.push(r.get_primary_key()); + } + } + // The row vanished between the index read and the + // resolve; it is simply not part of the snapshot. + core::result::Result::Err(e) if e.is_row_absent() => {} + // Anything else (corrupt page, invalid link, ...) is a + // real storage error, not an empty snapshot slot. + core::result::Result::Err(e) => { + return core::result::Result::Err(WorkTableError::PagesError(e)); + } + } + } + pks.sort_unstable(); + pks.dedup(); + for pk in pks { + match self.delete(pk).await { + core::result::Result::Ok(()) => {} + // Deleted concurrently after the snapshot: the goal + // state for this row is already reached. + core::result::Result::Err(WorkTableError::NotFound) => {} + core::result::Result::Err(e) => return core::result::Result::Err(e), + } } core::result::Result::Ok(()) } diff --git a/codegen/src/generators/persist/queries/delete.rs b/codegen/src/generators/persist/queries/delete.rs index 0d8fb9d2..1482d3db 100644 --- a/codegen/src/generators/persist/queries/delete.rs +++ b/codegen/src/generators/persist/queries/delete.rs @@ -4,6 +4,7 @@ use convert_case::{Case, Casing}; use proc_macro2::{Ident, Span, TokenStream}; use quote::quote; +use crate::common::model::Index; use crate::common::model::Operation; use crate::common::name_generator::{WorktableNameGenerator, is_float}; use crate::generators::persist::PersistGenerator; @@ -141,7 +142,7 @@ impl PersistGenerator { if index.is_unique { Self::gen_unique_delete(type_, &method_ident, index_name) } else { - Self::gen_non_unique_delete(type_, &method_ident, index_name) + Self::gen_non_unique_delete(type_, &method_ident, index) } } else { Self::gen_brute_force_delete_field(&op.by, type_, &method_ident) @@ -173,7 +174,9 @@ impl PersistGenerator { } } - fn gen_non_unique_delete(type_: &TokenStream, name: &Ident, index: &Ident) -> TokenStream { + fn gen_non_unique_delete(type_: &TokenStream, name: &Ident, index: &Index) -> TokenStream { + let by_field = &index.field; + let index = &index.name; let by = if is_float(type_.to_string().as_str()) { quote! { &OrderedFloat(by) @@ -185,10 +188,46 @@ impl PersistGenerator { }; quote! { pub async fn #name(&self, by: #type_) -> core::result::Result<(), WorkTableError> { - let rows_to_update = self.0.indexes.#index.get(#by).map(|kv| kv.1).collect::>(); - for link in rows_to_update { - let row = self.0.data.select_non_ghosted(link.0).map_err(WorkTableError::PagesError)?; - self.delete(row.get_primary_key()).await?; + // Snapshot the matching rows as validated primary keys before + // deleting anything. Storage links are not stable identities: + // a concurrent delete can free a slot and an insert can reuse + // it for an unrelated row before this loop runs, so resolving + // a stale link later could delete the wrong row. Every + // candidate is resolved and checked against the predicate at + // snapshot time; a reused slot only stays in the set if the + // row now living there genuinely matches. Keys are sorted for + // a deterministic delete order (non-unique indexes iterate + // equal keys in random-discriminator order) and the per-row + // delete takes its own row lock and resolves by primary key, + // never through the snapshotted link. + let mut pks: Vec<_> = Vec::new(); + for link in self.0.indexes.#index.get(#by).map(|kv| kv.1.0) { + match self.0.data.select_non_ghosted(link) { + core::result::Result::Ok(r) => { + if r.#by_field == by { + pks.push(r.get_primary_key()); + } + } + // The row vanished between the index read and the + // resolve; it is simply not part of the snapshot. + core::result::Result::Err(e) if e.is_row_absent() => {} + // Anything else (corrupt page, invalid link, ...) is a + // real storage error, not an empty snapshot slot. + core::result::Result::Err(e) => { + return core::result::Result::Err(WorkTableError::PagesError(e)); + } + } + } + pks.sort_unstable(); + pks.dedup(); + for pk in pks { + match self.delete(pk).await { + core::result::Result::Ok(()) => {} + // Deleted concurrently after the snapshot: the goal + // state for this row is already reached. + core::result::Result::Err(WorkTableError::NotFound) => {} + core::result::Result::Err(e) => return core::result::Result::Err(e), + } } core::result::Result::Ok(()) } From 4d0d8313b4187b58d63c0288cfe48b436065cab5 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 06/15] fix: never force-apply gapped event streams to the on-disk index BatchOperation::validate accepted an event stream whose first id was 2 ahead of the last applied id if the first event was a SplitNode, or unconditionally after 8 failed attempts ("TODO: for split sometimes this happens"). Change events are positional (InsertAt/RemoveAt carry node indices), so applying a stream with a missing event replays later events against node state the missing event was supposed to produce, silently scrambling on-disk nodes. Analysis of indexset (put_cdc_checked, remove_cdc, Operation::commit) shows it never consumes an event id without emitting the event, so a gap is always transient - the op carrying the missing event just has not been queued or batched yet - unless events were discarded after id assignment, which only non-CDC index mutations do (vacuum's update_index_after_move is the one reachable case on persisted tables, tracked separately). Instrumented measurement over 50 full-suite runs: neither escape hatch nor even a plain deferral ever fired. Fix: strictly defer on any gap. A gap persisting past 8 attempts now logs tracing::error naming the ids, so a genuine event leak becomes a loud stall instead of silent corruption. Measured: 50/50 full-suite runs clean afterwards, no stalls (the suite's 4s wait_for_ops asserts would catch one). --- src/persistence/operation/batch.rs | 53 +++++++++++++++--------------- 1 file changed, 26 insertions(+), 27 deletions(-) diff --git a/src/persistence/operation/batch.rs b/src/persistence/operation/batch.rs index d36bf252..914d0655 100644 --- a/src/persistence/operation/batch.rs +++ b/src/persistence/operation/batch.rs @@ -246,22 +246,24 @@ where && !id.is_next_for(last_ids.primary_id) && last_ids.primary_id != IndexChangeEventId::default() { - let mut possibly_valid = false; - if id.inner().overflowing_sub(last_ids.primary_id.inner()).0 == 2 { - // TODO: for split sometimes this happens - let ev = prepared_evs.primary_evs.first().unwrap(); - if let ChangeEvent::SplitNode { .. } = ev { - possibly_valid = true - } - if attempts > 8 { - possibly_valid = true - } - } - - if !possibly_valid { - self.ops.extend(ops_to_remove); - return Ok(None); + // Change events are positional (InsertAt/RemoveAt carry node + // indices), so a stream with a missing id must never be applied: + // the disk index would apply later events against node state the + // missing event was supposed to produce. Always defer. A gap is + // transient (the op carrying the missing event has not been + // batched yet) unless an event was discarded after its id was + // assigned — only non-CDC index mutations do that — so a gap + // 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", + last_ids.primary_id, + id, + ); } + self.ops.extend(ops_to_remove); + return Ok(None); } let secondary_first = prepared_evs.secondary_evs.first_evs(); for (index, id) in secondary_first { @@ -272,19 +274,16 @@ where && !id.is_next_for(*last) && *last != IndexChangeEventId::default() { - let mut possibly_valid = false; - if id.inner().overflowing_sub(last.inner()).0 == 2 { - // TODO: for split sometimes this happens - possibly_valid = prepared_evs.secondary_evs.is_first_ev_is_split(index); - if attempts > 8 { - possibly_valid = true - } - } - - if !possibly_valid { - self.ops.extend(ops_to_remove); - return Ok(None); + // Same rule as the primary index above: never apply a gapped + // 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", + ); } + self.ops.extend(ops_to_remove); + return Ok(None); } } } From 24455f8a0aa52d5d320cfbc27bf94da70df7ba82 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 07/15] fix: report every failed batch sub-operation without cancelling work DiskPersistenceEngine::apply_batch_operation drove its three concurrent sub-futures (data write, primary index events, secondary index events) with `while futs.next().await.is_some() {}`, dropping every Result. A failed disk write or index apply vanished without a trace - not even the task-level warn fired, because the error never left this function. All futures are drained to completion before errors are surfaced: bailing out on the first failure would drop the FuturesUnordered and cancel the remaining sub-operations at arbitrary await points, and these futures are not cancellation-safe (a data page could be left half-written while its index events were abandoned). Errors are tagged with their component ("batch data write" / "primary index batch apply" / "secondary index batch apply") and ALL failures are reported, not just the first in nondeterministic completion order: a mixed outcome means the durable state may already be inconsistent, and knowing which parts failed is what an operator needs. --- src/persistence/engine.rs | 55 ++++++++++++++++++++++++++++++++------- 1 file changed, 46 insertions(+), 9 deletions(-) diff --git a/src/persistence/engine.rs b/src/persistence/engine.rs index 9237d1d9..ffb63b9c 100644 --- a/src/persistence/engine.rs +++ b/src/persistence/engine.rs @@ -170,16 +170,53 @@ where let (pk_evs, secondary_evs) = batch_op.get_indexes_evs()?; { + let data = &mut self.data; + let primary_index = &mut self.primary_index; + let secondary_indexes = &mut self.secondary_indexes; let mut futs = FuturesUnordered::new(); - futs.push(Either::Left(Either::Right(self.data.save_batch_data(batch_data_op)))); - futs.push(Either::Left(Either::Left( - self.primary_index.process_change_event_batch(pk_evs), - ))); - futs.push(Either::Right( - self.secondary_indexes.process_change_event_batch(secondary_evs), - )); - - while (futs.next().await).is_some() {} + futs.push(Either::Left(Either::Right(async move { + data.save_batch_data(batch_data_op) + .await + .map_err(|e| e.wrap_err("batch data write")) + }))); + futs.push(Either::Left(Either::Left(async move { + primary_index + .process_change_event_batch(pk_evs) + .await + .map_err(|e| e.wrap_err("primary index batch apply")) + }))); + futs.push(Either::Right(async move { + secondary_indexes + .process_change_event_batch(secondary_evs) + .await + .map_err(|e| e.wrap_err("secondary index batch apply")) + })); + + // Drain every future before surfacing errors: `?` on the first + // failure would drop the FuturesUnordered and cancel the remaining + // sub-operations at arbitrary await points, leaving e.g. a data + // page half-written while its index events were abandoned. These + // futures are not cancellation-safe, so let all started work run + // to completion. Every failed component is reported (each error is + // wrapped with its component name above); a mixed outcome means + // the durable state may already be inconsistent, and knowing WHICH + // parts failed is what an operator needs. + let mut errors: Vec = Vec::new(); + while let Some(res) = futs.next().await { + if let Err(e) = res { + errors.push(e); + } + } + if errors.len() == 1 { + return Err(errors.pop().expect("len checked")); + } + if !errors.is_empty() { + let summary = errors.iter().map(|e| format!("{e:#}")).collect::>().join("; "); + return Err(eyre::eyre!( + "batch apply failed in {} sub-operations: {summary}", + errors.len() + )); + } } if let Some(pk_gen_state_update) = batch_op.get_pk_gen_state()? { From 297f576dd5791d20b3f2888927bcc0458bdca429 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 08/15] chore: fix clippy 1.97 useless_borrows_in_formatting lints `format!("{}By", &op.name)` borrowed needlessly in both type.rs generators; clippy 1.97 flags this and the repo policy is a clean `cargo clippy --all-targets`. --- codegen/src/generators/in_memory/queries/type.rs | 2 +- codegen/src/generators/persist/queries/type.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/codegen/src/generators/in_memory/queries/type.rs b/codegen/src/generators/in_memory/queries/type.rs index 36f9339b..35491ce7 100644 --- a/codegen/src/generators/in_memory/queries/type.rs +++ b/codegen/src/generators/in_memory/queries/type.rs @@ -144,7 +144,7 @@ impl InMemoryGenerator { .updates .values() .map(|op| { - let ident = Ident::new(format!("{}By", &op.name).as_str(), Span::mixed_site()); + let ident = Ident::new(format!("{}By", op.name).as_str(), Span::mixed_site()); let field_type = self .columns .columns_map diff --git a/codegen/src/generators/persist/queries/type.rs b/codegen/src/generators/persist/queries/type.rs index 8a3d2bb4..ec10a1f2 100644 --- a/codegen/src/generators/persist/queries/type.rs +++ b/codegen/src/generators/persist/queries/type.rs @@ -144,7 +144,7 @@ impl PersistGenerator { .updates .values() .map(|op| { - let ident = Ident::new(format!("{}By", &op.name).as_str(), Span::mixed_site()); + let ident = Ident::new(format!("{}By", op.name).as_str(), Span::mixed_site()); let field_type = self .columns .columns_map From 858b661ddeb7bbe761c505f33f4a9c691b8e34c5 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 09/15] fix: replace the unsound lockfree queue with a mutexed VecDeque The persistence queue drops `lockfree::queue::Queue` for a `parking_lot::Mutex>` (the `lockfree` dependency is removed; no new one is added). This is forced by soundness, not preference: lockfree 0.5.1 (unmaintained since 2019) constructs its internal Removable slots with `ManuallyDrop::new(mem::uninitialized())`. That is undefined behaviour for any element type whose layout has a niche, and `Operation` is such a type (an enum full of Vecs, i.e. NonNull pointers) - the code was always unsound. The next patch adds a `primary_key_events` Vec to UpdateOperation, which changes the niche layout for some monomorphisations and makes rustc's uninit validity check abort deterministically (SIGABRT at Queue::new in persistence::sync::many_strings). The options were an aborting unsound dependency or a different queue. Performance impact: none measurable. End-to-end on a persisted table with three indexes (release build, 3 runs each, identical series with only the queue implementation differing): workload lockfree mutexed deque 20k inserts, single writer 657/671/657 ms 657/678/696 ms 4 writers x 5k inserts 410/394/416 ms 409/402/412 ms The spread within each variant exceeds the difference between them, which is expected mechanically: * The queue carries one Operation per write on a persisted table. A full insert costs ~33us (rkyv serialisation, index updates with CDC event generation, a UUIDv7, analyzer bookkeeping, file IO downstream). An uncontended parking_lot lock/unlock is 10-20ns, ~0.05% of that. * lockfree's queue is a Michael-Scott linked queue: one heap-allocated node plus a CAS on the shared tail per push, a CAS plus deferred reclamation (its "incinerator" GC) per pop. All producers serialise on the same tail cache line under either design - lock-freedom is a progress guarantee, not a contention remover - while the contiguous VecDeque does strictly less work per element: no per-node allocation, no reclamation machinery, better cache locality. * The queue has a single consumer (the engine task) and nanosecond hold times, so the one scenario where lock-freedom genuinely helps (a lock holder descheduled mid-critical-section) has effectively zero exposure, and parking_lot parks instead of spinning. Deliberate minor trade-offs: pop_iter() takes the lock once per drained element (swap the deque out with mem::take if a profile ever disagrees); len is incremented before the element is enqueued, a window that errs only conservatively for wait_for_ops. If write concurrency ever makes this queue visible in profiles, the upgrade is a purpose-built MPSC channel, not a return to lockfree. data_bucket's transitive lockfree dependency is dropped in its own 0.4.0 release (only a SizeMeasurable impl, unsound path not instantiated). --- Cargo.toml | 1 - src/persistence/task.rs | 22 ++++++++++++---------- 2 files changed, 12 insertions(+), 11 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index f6c5bef1..e4945bfe 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,6 @@ futures = "0.3.30" indexset = { version = "=0.16.0", features = ["concurrent", "cdc", "multimap"] } # indexset = { path = "../indexset", version = "0.15.0", features = ["concurrent", "cdc", "multimap"] } # indexset = { package = "wt-indexset", version = "=0.12.12", features = ["concurrent", "cdc", "multimap"] } -lockfree = { version = "0.5.1" } log = "0.4.29" ordered-float = "5.0.0" parking_lot = "0.12.3" diff --git a/src/persistence/task.rs b/src/persistence/task.rs index e6fcc916..c96928b5 100644 --- a/src/persistence/task.rs +++ b/src/persistence/task.rs @@ -1,4 +1,4 @@ -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::fmt::Debug; use std::hash::Hash; use std::marker::PhantomData; @@ -7,6 +7,7 @@ use std::sync::atomic::{AtomicBool, AtomicU16, Ordering}; use std::time::Duration; use data_bucket::page::PageId; +use parking_lot::Mutex as ParkingMutex; use tokio::sync::Notify; use worktable_codegen::worktable; @@ -271,7 +272,11 @@ where #[derive(Debug)] pub struct Queue { - queue: lockfree::queue::Queue>, + // Not `lockfree::queue::Queue`: its `Removable::empty` materializes the + // element type via `mem::uninitialized`, which aborts at runtime for + // `Operation` layouts that reject uninit bytes. The queue has a single + // consumer (the engine task), so a mutexed deque is uncontended here. + queue: ParkingMutex>>, notify: Notify, len: Arc, } @@ -279,7 +284,7 @@ pub struct Queue { impl Queue { pub fn new() -> Self { Self { - queue: lockfree::queue::Queue::new(), + queue: ParkingMutex::new(VecDeque::new()), notify: Notify::new(), len: Arc::new(AtomicU16::new(0)), } @@ -287,14 +292,14 @@ impl Queue) { self.len.fetch_add(1, Ordering::Release); - self.queue.push(value); + self.queue.lock().push_back(value); self.notify.notify_one(); } pub async fn pop(&self) -> Operation { loop { // Drain values - if let Some(value) = self.queue.pop() { + if let Some(value) = self.queue.lock().pop_front() { self.len.fetch_sub(1, Ordering::Release); return value; } @@ -305,7 +310,7 @@ impl Queue Option> { - if let Some(v) = self.queue.pop() { + if let Some(v) = self.queue.lock().pop_front() { self.len.fetch_sub(1, Ordering::Release); Some(v) } else { @@ -314,10 +319,7 @@ impl Queue impl Iterator> { - let iter_count = self.len.clone(); - self.queue.pop_iter().inspect(move |_| { - iter_count.fetch_sub(1, Ordering::Release); - }) + std::iter::from_fn(|| self.immediate_pop()) } pub fn len(&self) -> usize { From 4cd6912918c044c609875f59b516b35363fa2c6a Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 10/15] vacuum: persist row moves through CDC so persisted tables survive defragmentation Problem: EmptyDataVacuum's row-move path (update_index_after_move) mutated the indexes with the non-CDC calls `primary_index.insert(..)` and `secondary_indexes.reinsert_row(..)`. The generated `vacuum()` exists for persisted tables too, and there this had two consequences: 1. The moved links never reached the persistence stream, so the on-disk indexes went stale relative to memory. 2. indexset's non-CDC insert/remove still consume CDC event ids (fetch_add) while discarding the events, creating a permanent gap in the event-id stream. BatchOperation::validate strictly defers on gaps, so the first vacuum on a persisted table stalled persistence forever (tracing::error after 8 attempts). Fix: * UpdateOperation gains a `primary_key_events` field (it becomes UpdateOperation), and the disk engine's Update arm processes those events. A vacuum row move is exactly this operation: row bytes at the new link plus the primary- and secondary-index CDC events, applied atomically as one op. Audited every `Operation::primary_key_events()` caller: all three iterate the returned events, so Update returning Some(&[]) for ordinary updates is behaviourally identical to the previous None. * New `VacuumPersistence` trait; the persistence `Queue` implements it by pushing that Update operation, and `PersistenceTask::vacuum_sink()` hands an Arc to the vacuum. The trait is #[doc(hidden)] macro-support API: it leaks low-level CDC event types and is not meant for downstream implementations; no semver stability is promised for it. * EmptyDataVacuum holds an optional sink (`with_persistence(..)`). With a sink, update_index_after_move uses reinsert_row_cdc / insert_cdc, captures the events, and emits the operation; without one (in-memory tables) the old non-CDC path is unchanged. Codegen for persisted tables wires `.with_persistence(self.1.vacuum_sink())` into the generated `vacuum()`. NOTE: any UpdateOperation construction site added later must also initialize `primary_key_events`. * New test tests/persistence/vacuum.rs: fills a persisted table across multiple pages, deletes, vacuums (asserting pages freed), inserts afterwards, asserts under a 30s timeout that persistence drains, then reloads from disk and verifies every surviving row by primary key, unique and non-unique index. Without the fix it fails with "persistence stalled after vacuum on persisted table" (re-confirmed by reverting only the wiring line). Known limitation: vacuum on a persisted table keeps the on-disk state consistent but does not reclaim on-disk space; pages freed in memory stay in the data file (tracked in issue #163). --- .../generators/in_memory/queries/update.rs | 1 + .../src/generators/persist/queries/update.rs | 1 + codegen/src/generators/persist/table/impls.rs | 2 +- src/lib.rs | 2 +- src/persistence/engine.rs | 3 + src/persistence/operation/operation.rs | 7 +- src/persistence/task.rs | 36 +++++ src/table/vacuum/mod.rs | 29 ++++ src/table/vacuum/vacuum.rs | 45 +++++- tests/persistence/mod.rs | 1 + tests/persistence/vacuum.rs | 144 ++++++++++++++++++ 11 files changed, 258 insertions(+), 13 deletions(-) create mode 100644 tests/persistence/vacuum.rs diff --git a/codegen/src/generators/in_memory/queries/update.rs b/codegen/src/generators/in_memory/queries/update.rs index 5d75168a..15503b92 100644 --- a/codegen/src/generators/in_memory/queries/update.rs +++ b/codegen/src/generators/in_memory/queries/update.rs @@ -270,6 +270,7 @@ impl InMemoryGenerator { #secondary_events_ident > = Operation::Update(UpdateOperation { id: op_id, + primary_key_events: vec![], secondary_keys_events, bytes: updated_bytes, link, diff --git a/codegen/src/generators/persist/queries/update.rs b/codegen/src/generators/persist/queries/update.rs index 6a72f936..99149ff6 100644 --- a/codegen/src/generators/persist/queries/update.rs +++ b/codegen/src/generators/persist/queries/update.rs @@ -265,6 +265,7 @@ impl PersistGenerator { #secondary_events_ident > = Operation::Update(UpdateOperation { id: op_id, + primary_key_events: vec![], secondary_keys_events, bytes: updated_bytes, link, diff --git a/codegen/src/generators/persist/table/impls.rs b/codegen/src/generators/persist/table/impls.rs index cc19ea29..eafdc3ac 100644 --- a/codegen/src/generators/persist/table/impls.rs +++ b/codegen/src/generators/persist/table/impls.rs @@ -442,7 +442,7 @@ impl PersistGenerator { std::sync::Arc::clone(&self.0.lock_manager), std::sync::Arc::clone(&self.0.primary_index), std::sync::Arc::clone(&self.0.indexes), - )) + ).with_persistence(self.1.vacuum_sink())) } } } diff --git a/src/lib.rs b/src/lib.rs index 070cd2f7..aead1ab9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -46,7 +46,7 @@ pub mod prelude { AvailableIndex, Difference, IndexError, IndexMap, IndexMultiMap, MultiPairRecreate, PrimaryIndex, TableIndex, TableIndexCdc, TableRow, TableSecondaryIndex, TableSecondaryIndexCdc, TableSecondaryIndexEventsOps, TableSecondaryIndexInfo, UnsizedNode, WorkTable, WorkTableError, vacuum::EmptyDataVacuum, - vacuum::WorkTableVacuum, + vacuum::VacuumPersistence, vacuum::WorkTableVacuum, }; pub use data_bucket::{ DATA_VERSION, DataPage, GENERAL_HEADER_SIZE, GeneralHeader, GeneralPage, INNER_PAGE_SIZE, IndexPage, Interval, diff --git a/src/persistence/engine.rs b/src/persistence/engine.rs index ffb63b9c..52bfefd9 100644 --- a/src/persistence/engine.rs +++ b/src/persistence/engine.rs @@ -143,6 +143,9 @@ where } Operation::Update(update) => { self.data.save_data(update.link, update.bytes.as_ref()).await?; + for event in update.primary_key_events { + self.primary_index.process_change_event(event).await?; + } self.secondary_indexes .process_change_events(update.secondary_keys_events) .await diff --git a/src/persistence/operation/operation.rs b/src/persistence/operation/operation.rs index f45e65dd..36af8298 100644 --- a/src/persistence/operation/operation.rs +++ b/src/persistence/operation/operation.rs @@ -11,7 +11,7 @@ use crate::persistence::{OperationId, OperationType}; #[derive(Clone, Debug)] pub enum Operation { Insert(InsertOperation), - Update(UpdateOperation), + Update(UpdateOperation), Delete(DeleteOperation), Acknowledge(AcknowledgeOperation), } @@ -72,7 +72,7 @@ impl Operation Option<&Vec>>> { match &self { Operation::Insert(insert) => Some(&insert.primary_key_events), - Operation::Update(_) => None, + Operation::Update(update) => Some(&update.primary_key_events), Operation::Delete(delete) => Some(&delete.primary_key_events), Operation::Acknowledge(ack) => Some(&ack.primary_key_events), } @@ -120,8 +120,9 @@ pub struct InsertOperation { } #[derive(Clone, Debug)] -pub struct UpdateOperation { +pub struct UpdateOperation { pub id: OperationId, + pub primary_key_events: Vec>>, pub secondary_keys_events: SecondaryKeys, pub bytes: Vec, pub link: Link, diff --git a/src/persistence/task.rs b/src/persistence/task.rs index c96928b5..e616c68d 100644 --- a/src/persistence/task.rs +++ b/src/persistence/task.rs @@ -15,6 +15,7 @@ use crate::persistence::PersistenceEngine; use crate::persistence::operation::{BatchInnerRow, BatchInnerWorkTable, BatchOperation, OperationId, PosByOpIdQuery}; use crate::prelude::*; use crate::util::OptimizedVec; +use crate::vacuum::VacuumPersistence; worktable! ( name: QueueInner, @@ -327,6 +328,30 @@ impl Queue VacuumPersistence + for Queue +where + PrimaryKeyGenState: Send, + PrimaryKey: Send, + SecondaryKeys: Send, +{ + fn apply_move( + &self, + bytes: Vec, + new_link: Link, + primary_key_events: Vec>>, + secondary_keys_events: SecondaryKeys, + ) { + 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 { #[allow(dead_code)] @@ -345,6 +370,17 @@ impl self.queue.push(op); } + /// Returns a sink that lets vacuum queue persistence operations for row + /// moves into this task's operation queue. + pub fn vacuum_sink(&self) -> Arc> + where + PrimaryKeyGenState: Send + Sync + 'static, + PrimaryKey: Send + Sync + 'static, + SecondaryKeys: Send + Sync + 'static, + { + self.queue.clone() + } + pub fn run_engine(mut engine: E) -> Self where E: PersistenceEngine + Send + 'static, diff --git a/src/table/vacuum/mod.rs b/src/table/vacuum/mod.rs index 5396c2e0..21ab55e3 100644 --- a/src/table/vacuum/mod.rs +++ b/src/table/vacuum/mod.rs @@ -1,5 +1,9 @@ use async_trait::async_trait; +use data_bucket::Link; +use indexset::cdc::change::ChangeEvent; +use indexset::core::pair::Pair; + use crate::vacuum::fragmentation_info::FragmentationInfo; mod fragmentation_info; @@ -10,6 +14,31 @@ mod vacuum; pub use manager::{VacuumManager, VacuumManagerConfig}; pub use vacuum::EmptyDataVacuum; +/// Sink for persisting vacuum row moves. +/// +/// Vacuum relocates rows between data pages, which changes the [`Link`] stored +/// in the primary and secondary indexes. On persisted tables those index +/// mutations must go through the CDC event stream, and the moved row bytes must +/// be written at the new link — otherwise the on-disk state goes stale and the +/// event-id sequence gets a permanent gap that stalls persistence. Implementors +/// receive everything needed to queue a proper persistence operation for one +/// moved row. +/// Not intended for downstream implementation: this is macro-support API for +/// the generated persisted-table `vacuum()`, and it leaks low-level CDC event +/// types. Hidden from docs; semver stability is not promised for it. +#[doc(hidden)] +pub trait VacuumPersistence: Send + Sync { + /// Queue a persistence operation for a row moved to `new_link`, carrying + /// the row bytes and the CDC events produced by the index updates. + fn apply_move( + &self, + bytes: Vec, + new_link: Link, + primary_key_events: Vec>>, + secondary_keys_events: SecondaryEvents, + ); +} + /// Trait for unifying different [`WorkTable`] related [`EmptyDataVacuum`]'s. /// /// [`WorkTable`]: crate::prelude::WorkTable diff --git a/src/table/vacuum/vacuum.rs b/src/table/vacuum/vacuum.rs index a374db9c..1cb3d8b7 100644 --- a/src/table/vacuum/vacuum.rs +++ b/src/table/vacuum/vacuum.rs @@ -18,15 +18,18 @@ use rkyv::{Archive, Deserialize, Serialize}; use crate::in_memory::{ArchivedRowWrapper, DataPages, RowWrapper, StorableRow}; use crate::lock::{Lock, LockMap, RowLock}; use crate::prelude::{OffsetEqLink, TablePrimaryKey}; +use crate::vacuum::VacuumPersistence; use crate::vacuum::VacuumStats; use crate::vacuum::WorkTableVacuum; use crate::vacuum::fragmentation_info::FragmentationInfo; -use crate::{AvailableIndex, PrimaryIndex, TableIndex, TableRow, TableSecondaryIndex, TableSecondaryIndexCdc}; +use crate::{ + AvailableIndex, PrimaryIndex, TableIndex, TableIndexCdc, TableRow, TableSecondaryIndex, TableSecondaryIndexCdc, +}; use async_trait::async_trait; use ordered_float::OrderedFloat; use rkyv::api::high::HighDeserializer; -#[derive(Debug)] +#[derive(derive_more::Debug)] pub struct EmptyDataVacuum< Row, PrimaryKey, @@ -51,6 +54,11 @@ pub struct EmptyDataVacuum< primary_index: Arc>, secondary_indexes: Arc, + /// Persistence sink for row moves. `None` for in-memory tables; persisted + /// tables must set it so index updates go through CDC and reach disk. + #[debug(ignore)] + persistence: Option>>, + phantom_data: PhantomData<(SecondaryEvents, AvailableTypes, AvailableIndexes)>, } @@ -108,10 +116,19 @@ where lock_manager, primary_index, secondary_indexes, + persistence: None, phantom_data: PhantomData, } } + /// Attaches a persistence sink. Index updates for moved rows then use the + /// CDC mutation variants and their events are queued as persistence + /// operations. Required for persisted tables. + pub fn with_persistence(mut self, sink: Arc>) -> Self { + self.persistence = Some(sink); + self + } + async fn defragment(&self) -> VacuumStats { let now = Instant::now(); @@ -262,7 +279,7 @@ where let new_link = to_page .save_raw_row(&raw_data) .expect("page is not full as checked on links collection"); - self.update_index_after_move(pk.clone(), from_link.0, new_link); + self.update_index_after_move(pk.clone(), from_link.0, new_link, raw_data); lock.unlock(); self.lock_manager.remove_with_lock_check(&pk); @@ -284,16 +301,28 @@ where op_lock } - fn update_index_after_move(&self, pk: PrimaryKey, old_link: Link, new_link: Link) { + fn update_index_after_move(&self, pk: PrimaryKey, old_link: Link, new_link: Link, raw_data: Vec) { let row = self .data_pages .select(new_link) .expect("should exist as link was moved correctly"); - 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); + if let Some(persistence) = &self.persistence { + // Persisted table: mutate indexes through the CDC variants and queue + // the events with the moved bytes, so the on-disk state follows the + // move and the event-id stream stays gapless. + let (secondary_keys_events, res) = + self.secondary_indexes + .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); + } 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); + } } } diff --git a/tests/persistence/mod.rs b/tests/persistence/mod.rs index dc766d11..ac0fa72b 100644 --- a/tests/persistence/mod.rs +++ b/tests/persistence/mod.rs @@ -9,6 +9,7 @@ mod read; mod space_index; mod sync; mod toc; +mod vacuum; #[cfg(feature = "s3-support")] mod s3; diff --git a/tests/persistence/vacuum.rs b/tests/persistence/vacuum.rs new file mode 100644 index 00000000..cf19632d --- /dev/null +++ b/tests/persistence/vacuum.rs @@ -0,0 +1,144 @@ +use std::collections::HashMap; +use std::time::Duration; + +use tokio::time::timeout; + +use crate::remove_dir_if_exists; +use worktable::prelude::PersistedWorkTable; +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: VacuumPersist, + persist: true, + columns: { + id: u64 primary_key autoincrement, + test: i64, + another: u64, + exchange: String, + }, + indexes: { + test_idx: test unique, + another_idx: another, + exchange_idx: exchange, + }, +); + +/// Vacuum on a persisted table must push the row moves through the CDC +/// persistence stream: the moved links have to reach the on-disk indexes, and +/// the event-id sequence must stay gapless so persistence does not stall. +#[test] +fn test_vacuum_on_persisted_table_survives_reload() { + let config = DiskConfig::new_with_table_name( + "tests/data/vacuum/persisted", + VacuumPersistWorkTable::name_snake_case(), + VacuumPersistWorkTable::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("tests/data/vacuum/persisted".to_string()).await; + + let mut rows = HashMap::new(); + let deleted: Vec; + { + let engine = VacuumPersistPersistenceEngine::new(config.clone()).await.unwrap(); + let table = VacuumPersistWorkTable::load(engine).await.unwrap(); + + // row is ~40 bytes so ~409 rows per page; use multiple pages so + // defragment has non-current pages to move rows from. Drain the + // persistence queue every 100 rows: unthrottled bulk loads hit a + // pre-existing index-space batching bug ("page should be available + // in table of contents") that is unrelated to vacuum. + for i in 0..1000i64 { + let row = VacuumPersistRow { + id: table.get_next_pk().into(), + test: i, + another: i as u64, + exchange: format!("test{i}"), + }; + let id = row.id; + table.insert(row.clone()).unwrap(); + rows.insert(id, row); + if i % 100 == 99 { + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence should keep up with throttled inserts"); + } + } + + let mut ids: Vec<_> = rows.keys().cloned().collect(); + ids.sort_unstable(); + deleted = ids.into_iter().take(50).collect(); + for id in &deleted { + table.delete(*id).await.unwrap(); + } + + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence should catch up before vacuum"); + + let vacuum = table.vacuum(); + let stats = vacuum.vacuum().await.unwrap(); + assert!(stats.pages_freed > 0, "vacuum should have moved rows off a page"); + + // Insert after vacuum: these operations carry event ids issued + // after the moves, so if vacuum consumed ids without queueing the + // events, the batch validator defers on the gap forever. + for i in 1000..1100i64 { + let row = VacuumPersistRow { + id: table.get_next_pk().into(), + test: i, + another: i as u64, + exchange: format!("test{i}"), + }; + let id = row.id; + table.insert(row.clone()).unwrap(); + rows.insert(id, row); + if i % 50 == 49 { + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence stalled after vacuum on persisted table"); + } + } + + // Without CDC-aware vacuum this stalls forever: the moved links + // never reach the persistence stream while their event ids are + // 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"); + + for id in &deleted { + rows.remove(id); + } + for (id, expected) in &rows { + assert_eq!(table.select(*id).as_ref(), Some(expected)); + } + } + { + let engine = VacuumPersistPersistenceEngine::new(config.clone()).await.unwrap(); + let table = VacuumPersistWorkTable::load(engine).await.unwrap(); + + assert_eq!(table.select_all().execute().unwrap().len(), rows.len()); + for (id, expected) in &rows { + assert_eq!(table.select(*id).as_ref(), Some(expected)); + // Secondary indexes must follow the moved links too. + assert_eq!(table.select_by_test(expected.test).as_ref(), Some(expected)); + assert_eq!( + table.select_by_exchange(expected.exchange.clone()).execute().unwrap(), + vec![expected.clone()] + ); + } + for id in &deleted { + assert_eq!(table.select(*id), None); + } + } + }) +} From 3a499a09b2dfacb877f94771025a85761b76669e Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 11/15] tests: add ignored reproduction for the index-space TOC stall on bulk loads Problem ------- Independent of vacuum, an unthrottled bulk insert+delete on a persisted table panics the background persistence engine task with "page should be available in table of contents" in process_change_event_batch (src/persistence/space/index/mod.rs for sized keys, unsized_.rs for String keys). The engine task dies, so wait_for_ops() then hangs forever. The failure is timing/batch-boundary dependent: on this machine a plain 1000-insert / 50-delete workload trips it in roughly 4 out of 5 runs, and it reproduces identically on a tree without the vacuum patch (verified against a clean checkout plus only the lock-race series). This is why tests/persistence/vacuum.rs (patch 1/2) throttles its bulk phases with wait_for_ops every 100 rows. Change ------ Adds tests/persistence/bulk_load_stall.rs: bulk insert 1000 rows across a unique i64 index, a non-unique u64 index, and a non-unique String (unsized) index, delete the 50 lowest pks, then wait_for_ops under a 30s timeout so the stall fails loudly instead of hanging; finally reload from disk and verify the surviving rows. The test is #[ignore]d because it fails until the underlying bug is fixed; run it with: cargo test --test mod persistence::bulk_load_stall -- --ignored Once the TOC bug is fixed: un-ignore this test and remove the throttling workaround in tests/persistence/vacuum.rs. Suspected area: BatchOperation::validate / QueueAnalyzer deferral (src/persistence/operation/batch.rs, src/persistence/task.rs) interacting with the table-of-contents key tracking across node max_value transitions (SplitNode/CreateNode/RemoveNode) in process_change_event_batch. Apply on top of patch 1/2 (vacuum CDC row moves). --- tests/persistence/bulk_load_stall.rs | 100 +++++++++++++++++++++++++++ tests/persistence/mod.rs | 1 + 2 files changed, 101 insertions(+) create mode 100644 tests/persistence/bulk_load_stall.rs diff --git a/tests/persistence/bulk_load_stall.rs b/tests/persistence/bulk_load_stall.rs new file mode 100644 index 00000000..a80ad522 --- /dev/null +++ b/tests/persistence/bulk_load_stall.rs @@ -0,0 +1,100 @@ +use std::collections::HashMap; +use std::time::Duration; + +use tokio::time::timeout; + +use crate::remove_dir_if_exists; +use worktable::prelude::PersistedWorkTable; +use worktable::prelude::*; +use worktable::worktable; + +worktable!( + name: BulkLoadStall, + persist: true, + columns: { + id: u64 primary_key autoincrement, + test: i64, + another: u64, + exchange: String, + }, + indexes: { + test_idx: test unique, + another_idx: another, + exchange_idx: exchange, + }, +); + +/// Reproduction for a pre-existing index-space batching bug: an unthrottled +/// bulk insert+delete panics the persistence engine task with "page should be +/// available in table of contents" (src/persistence/space/index/mod.rs / +/// unsized_.rs, process_change_event_batch), after which wait_for_ops hangs +/// forever; the timeout below turns that hang into a failure. +/// +/// The failure is timing/batch-boundary dependent and hits roughly 4 out of 5 +/// runs, so it is ignored in the normal suite. Run it with: +/// `cargo test --test mod persistence::bulk_load_stall -- --ignored`. +/// No vacuum is involved; tests/persistence/vacuum.rs throttles its bulk +/// phases specifically to stay clear of this bug. Once fixed, un-ignore this +/// test and drop that throttling. +#[test] +#[ignore = "exposes a known index-space TOC bug; fails ~4/5 runs until it is fixed"] +fn test_bulk_insert_delete_persistence() { + let config = DiskConfig::new_with_table_name( + "tests/data/bulk_load_stall/persisted", + BulkLoadStallWorkTable::name_snake_case(), + BulkLoadStallWorkTable::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("tests/data/bulk_load_stall/persisted".to_string()).await; + + let mut rows = HashMap::new(); + { + let engine = BulkLoadStallPersistenceEngine::new(config.clone()).await.unwrap(); + let table = BulkLoadStallWorkTable::load(engine).await.unwrap(); + + for i in 0..1000i64 { + let row = BulkLoadStallRow { + id: table.get_next_pk().into(), + test: i, + another: i as u64, + exchange: format!("test{i}"), + }; + let id = row.id; + table.insert(row.clone()).unwrap(); + rows.insert(id, row); + } + + let mut ids: Vec<_> = rows.keys().cloned().collect(); + ids.sort_unstable(); + let deleted: Vec = ids.into_iter().take(50).collect(); + for id in &deleted { + table.delete(*id).await.unwrap(); + } + + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence stalled on bulk insert+delete"); + + for id in &deleted { + rows.remove(id); + } + } + { + let engine = BulkLoadStallPersistenceEngine::new(config.clone()).await.unwrap(); + let table = BulkLoadStallWorkTable::load(engine).await.unwrap(); + + assert_eq!(table.select_all().execute().unwrap().len(), rows.len()); + for (id, expected) in &rows { + assert_eq!(table.select(*id).as_ref(), Some(expected)); + } + } + }) +} diff --git a/tests/persistence/mod.rs b/tests/persistence/mod.rs index ac0fa72b..bdcfa807 100644 --- a/tests/persistence/mod.rs +++ b/tests/persistence/mod.rs @@ -2,6 +2,7 @@ use worktable::prelude::PersistedWorkTable; use worktable::prelude::*; use worktable::worktable; +mod bulk_load_stall; mod concurrent; mod failure; mod index_page; From e7ed290f4d89a9f988dbd15091541904d2b94f0b Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 12/15] fix: scan the whole batch for event-id gaps, not the last 30 events An unthrottled bulk load on a persisted table panicked the engine task with "page should be available in table of contents" (process_change_event_batch, sized and unsized index spaces), after which wait_for_ops() hung forever. tests/persistence/bulk_load_stall.rs reproduced this on ~4 out of 5 runs. Batch collection groups operations by data page, so a batch routinely carries an interior event-id gap: in the reproduction, inserts on the first data page emit primary-index ids up to 407 while the deletes of that page's rows emit ids 1002..1051, and the ids in between belong to operations on other data pages that are not part of the batch. validate_events is what defers everything after such a gap, but its backward scan stopped at MAX_CHECK_DEPTH = 30 events: the 50 contiguous delete events at the tail hid the gap, the scan concluded the stream was gapless, and the batch was applied with the hole. The skipped ids carried the node max transitions (splits) of the in-memory index, so the on-disk table of contents still had the node at max 407 while the applied RemoveAt referenced max 312 - the expect in the sized index space (and its unsized twin) fired, killing the engine task. Event ids are only valid to apply as a gapless stream (the strict- deferral fix made that invariant explicit), so the scan must cover the whole batch: drop the depth bound. Cost is O(batch events) per validation attempt. Measured (debug profile, worst cases): a gapless 10_000-event batch scans in ~0.46ms and 100_000 events in ~4.2ms; peeling one interior gap with a long tail costs ~0.34ms at 10_000 and ~3.2ms at 100_000. Real batches are page-limited (MAX_PAGE_AMOUNT) to a few hundred events, i.e. microseconds, against disk I/O per batch. Unit tests cover the exact shape: a gap hidden behind a 50-event contiguous tail must defer the tail, and a gapless stream must stay untouched. With this fix the reproduction no longer panics (0/100 solo runs), but it stays #[ignore]d for now: unthrottled bulk loads still trip a separate pre-existing data-page bug in save_batch_data, fixed next. --- src/persistence/operation/util.rs | 52 +++++++++++++++++++++++++++++-- 1 file changed, 49 insertions(+), 3 deletions(-) diff --git a/src/persistence/operation/util.rs b/src/persistence/operation/util.rs index 6d1de30b..405a9ce3 100644 --- a/src/persistence/operation/util.rs +++ b/src/persistence/operation/util.rs @@ -3,8 +3,6 @@ use indexset::cdc::change::{self, ChangeEvent}; use indexset::core::pair::Pair; use std::fmt::Debug; -pub const MAX_CHECK_DEPTH: usize = 30; - pub fn validate_events(evs: &mut Vec>>) -> Vec>> where T: Debug, @@ -36,7 +34,12 @@ fn validate_events_iteration(evs: &[ChangeEvent>]) -> (Vec(evs: &[ChangeEvent>]) -> (Vec ChangeEvent> { + ChangeEvent::InsertAt { + event_id: id.into(), + max_value: Pair { + key: id, + value: Link::default(), + }, + value: Pair { + key: id, + value: Link::default(), + }, + index: 0, + } + } + + #[test] + fn detects_gap_behind_long_contiguous_tail() { + // Interior gap (100..=139, then 1000..=1049) whose tail is longer than + // any bounded scan window: everything after the gap must be deferred. + let mut evs: Vec<_> = (100..140).map(insert_at).collect(); + evs.extend((1000..1050).map(insert_at)); + + let removed = validate_events(&mut evs); + + assert_eq!(evs.len(), 40); + assert!(evs.iter().all(|ev| ev.id() < 140.into())); + assert_eq!(removed.len(), 50); + assert!(removed.iter().all(|ev| ev.id() >= 1000.into())); + } + + #[test] + fn keeps_gapless_stream_untouched() { + let mut evs: Vec<_> = (100..200).map(insert_at).collect(); + let removed = validate_events(&mut evs); + assert!(removed.is_empty()); + assert_eq!(evs.len(), 100); + } +} From ea7bfb3f3bebe76be32ed2f69d96e22d5d2921c0 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 13/15] fix: track the real max created page id in save_batch_data With the event-gap fix in place the engine survives unthrottled bulk loads, which exposed a second, pre-existing bug: reloading after a bulk insert+delete returned zero-filled rows (id 0, empty strings) for pks whose index entries were intact. save_batch_data derives `ids_to_create` from `batch_data.keys()`, and HashMap iteration is unordered. `last_page_id` was then taken from `ids_to_create.last()` - an arbitrary element, not the max. A batch only creates several data pages at once when the engine is lagging behind a bulk load; when that happened and `.last()` was not the max, a page that had just been created and persisted stayed above `last_page_id`, so the next batch touching it classified it as new and re-created it zero-filled, wiping the rows persisted a batch earlier. Only the links of the wiping batch were applied on top of the fresh zero page, and the on-disk indexes still pointed at the wiped links. Track the max instead, and state the high-water-mark invariant directly (`last_page_id.max(*max)`) so a future refactor of the id-partitioning above cannot regress monotonicity. Un-ignore the bulk-load reproduction test now that both bugs it trips are fixed, and drop the insert throttling in the persisted-vacuum test that existed only to dodge them. Measured: reproduction 0/100 solo failures and the full suite 0/40 stressed runs after this fix; before it, the unthrottled tests read back zeroed rows in roughly 1-2 of 30 suite runs under load (and ~4/5 runs panicked before the event-gap fix). --- src/persistence/space/data.rs | 12 +++++++++-- tests/persistence/bulk_load_stall.rs | 31 +++++++++++++++++----------- tests/persistence/vacuum.rs | 10 +-------- 3 files changed, 30 insertions(+), 23 deletions(-) diff --git a/src/persistence/space/data.rs b/src/persistence/space/data.rs index 9bb72b43..4685a97c 100644 --- a/src/persistence/space/data.rs +++ b/src/persistence/space/data.rs @@ -131,8 +131,16 @@ where .cloned() .collect::>(); - if let Some(max) = ids_to_create.last() { - self.last_page_id = *max; + // `page_ids` iterates a HashMap, so `ids_to_create` is unordered: + // taking `.last()` here picked an arbitrary created page, and a batch + // creating several pages could leave `last_page_id` below a page that + // now exists. The next batch touching that page would see it as "new" + // and re-create it zero-filled, wiping the rows persisted before. + if let Some(max) = ids_to_create.iter().max() { + // High-water mark: every id in `ids_to_create` is > last_page_id by + // construction, but state the monotonic invariant directly so a + // future refactor of the filter above cannot regress it. + self.last_page_id = self.last_page_id.max(*max); } let created_pages = ids_to_create .into_iter() diff --git a/tests/persistence/bulk_load_stall.rs b/tests/persistence/bulk_load_stall.rs index a80ad522..cf55b028 100644 --- a/tests/persistence/bulk_load_stall.rs +++ b/tests/persistence/bulk_load_stall.rs @@ -24,20 +24,27 @@ worktable!( }, ); -/// Reproduction for a pre-existing index-space batching bug: an unthrottled -/// bulk insert+delete panics the persistence engine task with "page should be -/// available in table of contents" (src/persistence/space/index/mod.rs / -/// unsized_.rs, process_change_event_batch), after which wait_for_ops hangs -/// forever; the timeout below turns that hang into a failure. +/// Regression test for an index-space batching bug: an unthrottled bulk +/// insert+delete used to panic the persistence engine task with "page should +/// be available in table of contents" (process_change_event_batch), after +/// which wait_for_ops hung forever; the timeout below turns that hang into a +/// failure. /// -/// The failure is timing/batch-boundary dependent and hits roughly 4 out of 5 -/// runs, so it is ignored in the normal suite. Run it with: -/// `cargo test --test mod persistence::bulk_load_stall -- --ignored`. -/// No vacuum is involved; tests/persistence/vacuum.rs throttles its bulk -/// phases specifically to stay clear of this bug. Once fixed, un-ignore this -/// test and drop that throttling. +/// Page-grouped batch collection puts the delete events (ids far ahead of the +/// batched inserts) into the same batch as the first data page's inserts, so +/// the prepared event stream has an interior id gap. `validate_events` used to +/// scan only 30 events back from the end; the 50 contiguous delete events hid +/// the gap, the batch was applied with the hole, and the on-disk index lost +/// track of node max transitions carried by the missing events. The scan is +/// now unbounded, so the tail after the gap is deferred until the missing +/// events arrive. +/// +/// The same unthrottled load also used to hit a second bug: a lagging batch +/// creating several data pages at once picked `last_page_id` from unordered +/// HashMap keys in `save_batch_data`, so a later batch could re-create an +/// existing page zero-filled and this test's reload phase read back zeroed +/// rows. Both fixes are needed for this test to be stable. #[test] -#[ignore = "exposes a known index-space TOC bug; fails ~4/5 runs until it is fixed"] fn test_bulk_insert_delete_persistence() { let config = DiskConfig::new_with_table_name( "tests/data/bulk_load_stall/persisted", diff --git a/tests/persistence/vacuum.rs b/tests/persistence/vacuum.rs index cf19632d..1e510e3f 100644 --- a/tests/persistence/vacuum.rs +++ b/tests/persistence/vacuum.rs @@ -52,10 +52,7 @@ fn test_vacuum_on_persisted_table_survives_reload() { let table = VacuumPersistWorkTable::load(engine).await.unwrap(); // row is ~40 bytes so ~409 rows per page; use multiple pages so - // defragment has non-current pages to move rows from. Drain the - // persistence queue every 100 rows: unthrottled bulk loads hit a - // pre-existing index-space batching bug ("page should be available - // in table of contents") that is unrelated to vacuum. + // defragment has non-current pages to move rows from. for i in 0..1000i64 { let row = VacuumPersistRow { id: table.get_next_pk().into(), @@ -66,11 +63,6 @@ fn test_vacuum_on_persisted_table_survives_reload() { let id = row.id; table.insert(row.clone()).unwrap(); rows.insert(id, row); - if i % 100 == 99 { - timeout(Duration::from_secs(30), table.wait_for_ops()) - .await - .expect("persistence should keep up with throttled inserts"); - } } let mut ids: Vec<_> = rows.keys().cloned().collect(); From 9c8289bc0479cfd84bcda574346abec626afd623 Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 14/15] fix: don't let wait_for_ops return while a popped operation is in flight PersistenceTask::wait_for_ops treats persistence as settled when three triggers hold at once: the operation queue is empty, the analyzer's inner table is empty, and analyzer_in_progress is false. The engine loop's idle path broke the flag's covering guarantee: task_analyzer_in_progress.store(false, Release); let op = engine_queue.pop().await; task_analyzer_in_progress.store(true, Release); Queue::pop removes the element and runs len.fetch_sub before returning, so between the fetch_sub and the store(true) all three triggers hold at once -- queue.len() == 0, analyzer count == 0, analyzer_in_progress == false -- while the popped operation has not been fed to the analyzer, let alone persisted. check_wait_triggers passes and wait_for_ops returns with the operation still in flight. A caller that then drops the table and reopens the space races the engine's write of that operation. Observed under full-suite parallel load (both rare): - persistence::failure::insert::test_insert_primary_duplicate failed 1/30: the phase-3 row read data = 1000 instead of 100 -- the reload saw a stale pk_gen_state and reused a primary key. - persistence::sync::test_space_insert_many_sync failed with pk_gen state 4917 vs 5000 while all 5000 rows were present -- the state write of the tail of the queue was still in flight at reload. Fix: make the busy-flag handoff atomic with the dequeue. Queue::pop_marking_in_progress sets the flag back to true under the queue mutex, before decrementing len. The store(true) is ordered before len.fetch_sub(Release), and check_wait_triggers acquire-loads len before it loads the flag, so a waiter that observes the post-pop length of 0 is guaranteed to also observe in_progress == true. The empty+idle state is no longer observable while an operation is in flight. The idle path now also stores false before notify_waiters, so a woken waiter can observe the idle state immediately (latency only, not correctness). Also abort the engine task on PersistenceTask drop - but only when the engine is provably idle. The AbortHandle was stored but never used, so the detached engine task outlived the table: after a drop-and-reopen within one runtime the old engine could still be applying a batch while a new engine loaded the same files. An idle task is parked at the queue pop, an await point where cancellation is clean. A BUSY engine is deliberately not aborted: the batch sub-futures are not cancellation-safe (a data page could be left half-written while its index events were abandoned), so aborting mid-batch would trade one corruption for another. A busy drop logs tracing::error and leaves the task running; callers must drain with wait_for_ops() first. The proper close()/Failed-state lifecycle is tracked in issue #168. The queue's length counter also moves from AtomicU16 to AtomicUsize: the queue is unbounded, and a 16-bit counter wraps at 65_536 queued operations, at which point the wait triggers would see an "empty" queue that still holds work. Verified on top of the 0001-0013 series: cargo test --test mod stress-ran 40 times (316 tests per run, 0 failures); test_insert_primary_duplicate and test_space_insert_many_sync passed 40/40 under that load, against 1/30 failures before. cargo fmt and cargo clippy --all-targets stay clean. --- src/persistence/task.rs | 70 +++++++++++++++++++++++++++++++++-------- 1 file changed, 57 insertions(+), 13 deletions(-) diff --git a/src/persistence/task.rs b/src/persistence/task.rs index e616c68d..7fefe41c 100644 --- a/src/persistence/task.rs +++ b/src/persistence/task.rs @@ -3,7 +3,7 @@ use std::fmt::Debug; use std::hash::Hash; use std::marker::PhantomData; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicU16, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; use data_bucket::page::PageId; @@ -279,7 +279,10 @@ pub struct Queue { // consumer (the engine task), so a mutexed deque is uncontended here. queue: ParkingMutex>>, notify: Notify, - len: Arc, + // usize, not u16: the queue is unbounded and a 16-bit counter wraps at + // 65_536 queued operations, making the wait triggers see an "empty" + // queue that still holds work. + len: Arc, } impl Queue { @@ -287,7 +290,7 @@ impl Queue Queue Operation { + /// Pops the next operation, marking `in_progress` `true` before the queue + /// length is decremented. The store must precede `len.fetch_sub` so that a + /// waiter that reads `len() == 0` (`Acquire`) is guaranteed to observe + /// `in_progress == true` for the popped-but-unprocessed operation; + /// otherwise `wait_for_ops` can return while that operation is in flight. + pub async fn pop_marking_in_progress( + &self, + in_progress: &AtomicBool, + ) -> Operation { loop { // Drain values - if let Some(value) = self.queue.lock().pop_front() { - self.len.fetch_sub(1, Ordering::Release); - return value; + { + 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; + } } // Wait for values to be available @@ -324,7 +339,7 @@ impl Queue usize { - self.len.load(Ordering::Acquire) as usize + self.len.load(Ordering::Acquire) } } @@ -354,7 +369,6 @@ where #[derive(Debug)] pub struct PersistenceTask { - #[allow(dead_code)] engine_task_handle: tokio::task::AbortHandle, queue: Arc>, analyzer_inner_wt: Arc, @@ -363,6 +377,35 @@ pub struct PersistenceTask, } +impl Drop + for PersistenceTask +{ + /// Aborts the engine task so it cannot outlive the table it persists. + /// Without this the detached task keeps running on the runtime after the + /// table is dropped, and a re-opened table can read the same files while + /// the old engine is still writing them. + /// + /// The abort only happens when the engine is provably idle (queue and + /// analyzer empty, no operation in flight) — that is the normal state + /// after `wait_for_ops`, and an idle task is parked at the queue pop, an + /// await point where cancellation is clean. Aborting a *busy* engine + /// would cancel persistence futures that are not cancellation-safe (a + /// data page could be left half-written while its index events are + /// abandoned), so a busy engine is left running and reported instead: + /// callers must drain with `wait_for_ops` before dropping. A proper + /// `close()` lifecycle (drain, join, surface terminal errors) is the + /// long-term replacement for this heuristic. + fn drop(&mut self) { + if self.check_wait_triggers() { + self.engine_task_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." + ); + } + } +} + impl PersistenceTask { @@ -404,11 +447,12 @@ impl let op = if let Some(next_op) = engine_queue.immediate_pop() { Some(next_op) } else if analyzer.len() == 0 { - engine_progress_notify.notify_waiters(); task_analyzer_in_progress.store(false, Ordering::Release); - let res = Some(engine_queue.pop().await); - task_analyzer_in_progress.store(true, Ordering::Release); - res + engine_progress_notify.notify_waiters(); + // 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) } else { None }; From 7b036736ccfe804bc6c419b9196528070068777b Mon Sep 17 00:00:00 2001 From: meh Date: Mon, 27 Jul 2026 05:52:03 +0700 Subject: [PATCH 15/15] build: move to WorkTablesIndex 0.0.1 and data_bucket 0.4.0 Atomic dependency migration plus the golden fixture that depends on it: * indexset is consumed as `WorkTablesIndex 0.0.1` via a package alias (`indexset = { package = "WorkTablesIndex", .. }`), so every `use indexset::` path is unchanged. WorkTablesIndex is indexset 0.16.0 plus the halve() soundness fix - split_off(len/2) instead of capacity/2, which panicked ("`at` split index (is N) should be <= len") whenever a node's capacity exceeded its length after removals; that panic was the last source of intermittent test failures (~18/50 full-suite runs). * data_bucket moves to 0.4.0, which consumes the same WorkTablesIndex crate (two source crates providing the same Pair/ChangeEvent types cannot coexist in one dependency tree) and drops its transitive lockfree 0.5.1 dependency. * tests/data/expected/space_index/indexset/ process_insert_at_big_amount.wt.idx is regenerated in the same commit: the indexset_compatibility big-amount test replays live CDC events and compares the space-index file byte-for-byte, so the fixture encodes the split geometry and must change together with the dependency (each side fails deterministically without the other). The regenerated output was verified byte-identical across two runs before adoption. File format is unchanged; existing table files load fine under either version. With this commit the series is self-contained against crates.io: no [patch.crates-io], no path dependencies. --- Cargo.toml | 4 ++-- .../process_insert_at_big_amount.wt.idx | Bin 65526 -> 65526 bytes 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index e4945bfe..7bd1a484 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,14 +23,14 @@ s3-support = ["dep:rusty-s3", "dep:url", "dep:reqwest", "dep:walkdir", "worktabl [dependencies] async-trait = "0.1.89" convert_case = "0.6.0" -data_bucket = "=0.3.15" +data_bucket = "=0.4.0" # data_bucket = { git = "https://github.com/pathscale/DataBucket", branch = "page_cdc_correction", version = "0.2.7" } # data_bucket = { path = "../DataBucket", version = "0.3.14" } derive_more = { version = "2.0.1", features = ["from", "error", "display", "debug", "into"] } eyre = "0.6.12" fastrand = "2.3.0" futures = "0.3.30" -indexset = { version = "=0.16.0", features = ["concurrent", "cdc", "multimap"] } +indexset = { package = "WorkTablesIndex", version = "=0.0.1", features = ["concurrent", "cdc", "multimap"] } # indexset = { path = "../indexset", version = "0.15.0", features = ["concurrent", "cdc", "multimap"] } # indexset = { package = "wt-indexset", version = "=0.12.12", features = ["concurrent", "cdc", "multimap"] } log = "0.4.29" diff --git a/tests/data/expected/space_index/indexset/process_insert_at_big_amount.wt.idx b/tests/data/expected/space_index/indexset/process_insert_at_big_amount.wt.idx index 681ca432873979e0b0a2bd756309efe553e70db7..bad6e6921e1bed97824b7759edc51e52ae52900c 100644 GIT binary patch delta 1177 zcmYM!IcyU_7{&2<8IUZh6d<`skqSke22rFzxI!r?Ql<=DM3pj~O@$Q#1j1nm5JNb^ z71k_=xy=!XIL#eya|Ht-m;lC{%f|bcnC|!XlfF07XiJ_gdDf|kPP!sfImK!2)M$CZ z5hR%(?xwOVV` zrL|h8_1d7MHfoc)wOL!VRXs{+o3^W0JG4`~v|D}JqrKXv{W_q29h9TA4$0MF9nn!8 zQ%1*iLML@fr*%eWmDM?&*9BdqxZg3oWT}mfls&KA8grFAG(EV$HC&6Y<9d7pH{zSP z1?O-a-@-|JJ8;Z61a}x1#CLHX-@|2mA6L%^3O~SgcnC-FL)?TPRU9*>mEbW03H$`7 z@Kc<@&u|Vu#|50nW@b?E1rFnvTik~WIE&xmA^hI= z_^Ba<7X}9LSDeS+a2c0y_3WVV zcU*_dIEsJZCj3*C`}=Qz;1>f4{2Qn6ADqE|aSnU9fc^hKa3VoL;V`b6PY@xnxB*YV T&3GbitJqz1d2C+Ss2BPN`R=O~ delta 1174 zcmZA0Ic(El7{+lwaY2Ep14tPnQakW!99 zIULLp?r=#JXE@t(hC4thr9d!HFi!lJknmvX_xP2+=lf)rJ+thYc3phtw2)D~I%3Y8 zxVw1PKO08Ge-bhjauy}KOzVm+>yj?&g7P}Ab2_UtI;~SWshm#exQ^+l934?swhrr% z4(fmgwO{)*puO6o-P)yo?NpyK+M(^*re1B;7Hw8qo3v3Iv|c@0r?u+V8m-nUb!nx7 zcK$uPY=-G9S&@ra|5vPG3vIEWvWc5<2*#cIz9#5k;1zDeFL5(|fn)eN zj^H9T@H1SB28EyELe#LT|HTu690QMW20y}`_#sZ>2%7M9T#v8eaK+AG|MEIBrIfQUKilkE YS~Gpj^}LknOkJI9%qDEJXWR?_1rzM2O8@`>