From d0c45d14e9afafb507064cd3af50b77fb9b8fbe5 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 5 Aug 2026 08:53:13 +0700 Subject: [PATCH 1/2] fix: revalidate vacuum links after row locking --- src/table/vacuum/vacuum.rs | 130 ++++++++++++++++++++++++++++++------- 1 file changed, 108 insertions(+), 22 deletions(-) diff --git a/src/table/vacuum/vacuum.rs b/src/table/vacuum/vacuum.rs index e555c0a..32e8ed3 100644 --- a/src/table/vacuum/vacuum.rs +++ b/src/table/vacuum/vacuum.rs @@ -14,7 +14,7 @@ use rkyv::util::AlignedVec; use rkyv::{Archive, Deserialize, Serialize}; use crate::in_memory::{ArchivedRowWrapper, DataPages, RowWrapper, StorableRow}; -use crate::lock::{Lock, LockMap, RowLock}; +use crate::lock::{Lock, LockGuard, LockMap, RowLock}; use crate::prelude::{OffsetEqLink, TablePrimaryKey}; use crate::vacuum::VacuumPersistence; use crate::vacuum::VacuumStats; @@ -279,32 +279,48 @@ where drop(range); for (from_link, pk) in links { - let lock = self.full_row_lock(&pk).await; - let _mutation_guard = self.lock_manager.mutation_guard(&pk); - if self - .data_pages - .with_ref(from_link.0, |r| r.is_deleted()) - .expect("link should be valid") - { - lock.unlock(); - self.lock_manager.remove_with_lock_check(&pk); - continue; - } - let (raw_data, new_link) = unsafe { - self.data_pages - .move_row_for_vacuum(from_link.0, to) - .expect("links and destination capacity were checked") - }; - self.update_index_after_move(pk.clone(), from_link.0, new_link, raw_data)?; - self.data_pages.retire_published_link(from_link.0); - - lock.unlock(); - self.lock_manager.remove_with_lock_check(&pk); + self.move_candidate_if_current(from_link.0, pk, to).await?; } Ok((from_page_will_be_moved, to_page_will_be_filled)) } + /// Moves a reverse-index candidate only if it is still the forward-index + /// location after the row lock is acquired. + /// + /// `move_data_from` snapshots the reverse index before taking per-row + /// locks. A concurrent reinsert may move the key and recycle the captured + /// slot for a different row in that interval. Revalidating under the row + /// lock prevents vacuum from publishing that replacement row under the + /// stale candidate's primary key. + async fn move_candidate_if_current(&self, from_link: Link, pk: PrimaryKey, to: PageId) -> eyre::Result { + let lock = self.full_row_lock(&pk).await; + let _guard = LockGuard::new_with_mutation(lock, self.lock_manager.clone(), pk.clone()); + + let current_link: Option = self.primary_index.pk_map.lookup_for_select(&pk).map(Into::into); + if current_link != Some(from_link) { + return Ok(false); + } + + if self + .data_pages + .with_ref(from_link, |r| r.is_deleted()) + .expect("a current primary-index link should be valid") + { + return Ok(false); + } + + let (raw_data, new_link) = unsafe { + self.data_pages + .move_row_for_vacuum(from_link, to) + .expect("links and destination capacity were checked") + }; + self.update_index_after_move(pk, from_link, new_link, raw_data)?; + self.data_pages.retire_published_link(from_link); + + Ok(true) + } + async fn full_row_lock(&self, pk: &PrimaryKey) -> Arc { let lock_id = self.lock_manager.next_id(); // One atomic acquire, no check-then-act: see LockMap::get_or_insert_with. @@ -942,4 +958,74 @@ mod tests { drop(read_guard); } + + #[tokio::test] + async fn vacuum_skips_a_stale_candidate_after_its_link_is_reused() { + let table = TestWorkTable::default(); + let target = TestRow { + id: table.get_next_pk().into(), + test: 10, + another: 10, + exchange: "target00".to_string(), + }; + let target_id = target.id; + table.insert(target).unwrap(); + + // Model the reverse-index snapshot taken before vacuum waits for the + // row lock. + let stale_link = table + .0 + .primary_index + .pk_map + .get_value(&TestPrimaryKey::from(target_id)) + .unwrap() + .0; + + // A same-sized reinsert moves the target and retires its old slot. + let updated_target = TestRow { + id: target_id, + test: 11, + another: 11, + exchange: "updated0".to_string(), + }; + table.update(updated_target.clone()).await.unwrap(); + let current_target_link = table + .0 + .primary_index + .pk_map + .get_value(&TestPrimaryKey::from(target_id)) + .unwrap() + .0; + assert_ne!(current_target_link, stale_link); + + // Reuse the retired physical slot for a different row. Without the + // post-lock forward-index check, vacuum would move this row and bind + // it to `target_id`. + let replacement = TestRow { + id: table.get_next_pk().into(), + test: 12, + another: 12, + exchange: "reused00".to_string(), + }; + let replacement_id = replacement.id; + table.insert(replacement.clone()).unwrap(); + let replacement_link = table + .0 + .primary_index + .pk_map + .get_value(&TestPrimaryKey::from(replacement_id)) + .unwrap() + .0; + assert_eq!(replacement_link, stale_link, "test setup must recycle the stale slot"); + + let destination = table.0.data.allocate_new_or_pop_free().id; + let moved = create_vacuum(&table) + .move_candidate_if_current(stale_link, TestPrimaryKey::from(target_id), destination) + .await + .unwrap(); + + assert!(!moved, "vacuum must reject a candidate whose key moved"); + assert_eq!(table.select(target_id), Some(updated_target)); + assert_eq!(table.select(replacement_id), Some(replacement)); + } } From df7a1c42b3aaf9694406a0ff3af61a2b9c5958f1 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 5 Aug 2026 12:51:04 +0700 Subject: [PATCH 2/2] perf: keep fixed-width updates in place --- .../generators/in_memory/queries/update.rs | 23 +++++------- codegen/src/worktable/mod.rs | 37 +++++++++++++++++++ tests/worktable/update_in_place_unsized.rs | 32 ++++++++++++++++ 3 files changed, 79 insertions(+), 13 deletions(-) diff --git a/codegen/src/generators/in_memory/queries/update.rs b/codegen/src/generators/in_memory/queries/update.rs index 949f9dc..893e4c9 100644 --- a/codegen/src/generators/in_memory/queries/update.rs +++ b/codegen/src/generators/in_memory/queries/update.rs @@ -75,7 +75,6 @@ impl InMemoryGenerator { .unseal_unchecked() }; - let op_id = OperationId::Single(uuid::Uuid::now_v7()); #diff_process_insert #persist_op @@ -385,14 +384,9 @@ impl InMemoryGenerator { return core::result::Result::Ok(()); } } - } else if self.columns.is_sized { - // Sized rows never change length; the caller's field swap is safe and - // no size check / reinsert is needed. - quote! {} - } else { - // Unsized rows where an updated column IS indexed: keep the original - // always-reinsert path so secondary-index maintenance and unique - // checks run through reinsert. + } else if touches_index { + // Updating an indexed column must keep the index-maintaining + // reinsert path, regardless of the row's storage shape. let row_updates = idents .iter() .map(|i| quote! { row_new.#i = row.#i.clone(); }) @@ -420,6 +414,11 @@ impl InMemoryGenerator { return core::result::Result::Ok(()); } } + } else { + // Other columns make the row unsized, but this query updates only + // fixed-width, unindexed fields. The caller can safely swap those + // archived fields in place without rebuilding the full row. + quote! {} } } @@ -605,6 +604,7 @@ impl InMemoryGenerator { }) .collect::>(); + let archived_swap_is_safe = self.columns.is_sized || (unsized_fields.is_none() && idx_idents.is_none()); let size_check = self.gen_size_check(unsized_fields, idents, idx_idents); let diff_process_insert = self.gen_process_diffs_insert_on_index(idents, idx_idents); let diff_process_remove = self.gen_process_diffs_remove_on_index(idx_idents); @@ -612,7 +612,7 @@ impl InMemoryGenerator { let persist_op = self.gen_persist_op(); let custom_lock = self.gen_custom_lock_for_update(lock_ident); - let finish_update = if self.columns.is_sized { + let finish_update = if archived_swap_is_safe { quote! { #diff_process_insert #persist_op @@ -653,7 +653,6 @@ impl InMemoryGenerator { let mut bytes = rkyv::to_bytes::(&row).map_err(|_| WorkTableError::SerializeError)?; let mut archived_row = unsafe { rkyv::access_unchecked_mut::<<#query_ident as rkyv::Archive>::Archived>(&mut bytes[..]).unseal_unchecked() }; - let op_id = OperationId::Single(uuid::Uuid::now_v7()); #size_check #finish_update } @@ -784,7 +783,6 @@ impl InMemoryGenerator { guards.insert(pk.clone(), LockGuard::new(op_lock, self.0.lock_manager.clone(), pk)); } - let op_id = OperationId::Multi(uuid::Uuid::now_v7()); 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 @@ -931,7 +929,6 @@ impl InMemoryGenerator { } }; - let op_id = OperationId::Single(uuid::Uuid::now_v7()); #size_check #finish_update } diff --git a/codegen/src/worktable/mod.rs b/codegen/src/worktable/mod.rs index e9334af..f456767 100644 --- a/codegen/src/worktable/mod.rs +++ b/codegen/src/worktable/mod.rs @@ -218,6 +218,43 @@ mod tests { } } + #[test] + fn fixed_width_update_on_unsized_table_uses_archived_swap() { + let output = expand(quote! { + name: MixedWidthUpdate, + persist: false, + columns: { + id: u64 primary_key, + payload: String, + balance: f64, + }, + queries: { + update: { + Balance(balance) by id, + } + } + }) + .unwrap() + .to_string(); + + let update = output + .split("pub async fn update_balance") + .nth(1) + .expect("generated balance update"); + assert!( + update.contains("data . with_mut_ref"), + "fixed-width unindexed field must update archived storage in place" + ); + assert!( + !update.contains("self . reinsert"), + "an unrelated String column must not force a fixed-width update through reinsert" + ); + assert!( + !update.contains("Uuid :: now_v7"), + "non-persistent updates must not generate an unused operation id" + ); + } + #[cfg(feature = "logical-index-persistence")] #[test] fn logical_persistence_wraps_only_default_wti_backends() { diff --git a/tests/worktable/update_in_place_unsized.rs b/tests/worktable/update_in_place_unsized.rs index 9aba3a0..5460177 100644 --- a/tests/worktable/update_in_place_unsized.rs +++ b/tests/worktable/update_in_place_unsized.rs @@ -44,10 +44,12 @@ macro_rules! unsized_in_place_suite { columns: { id: u64 primary_key using $using, payload: String, + balance: f64, }, queries: { update: { Payload(payload) by id, + Balance(balance) by id, } } ); @@ -72,6 +74,7 @@ macro_rules! unsized_in_place_suite { .insert(UnsizedUpdateRow { id: 1, payload: "abcdefgh".to_string(), // 8 bytes + balance: 1.0, }) .unwrap(); @@ -105,6 +108,7 @@ macro_rules! unsized_in_place_suite { .insert(UnsizedUpdateRow { id: 1, payload: "abcdefghij".to_string(), + balance: 1.0, }) .unwrap(); @@ -146,6 +150,7 @@ macro_rules! unsized_in_place_suite { .insert(UnsizedUpdateRow { id: 1, payload: "0000".to_string(), + balance: 1.0, }) .unwrap(); @@ -201,6 +206,33 @@ macro_rules! unsized_in_place_suite { format!("{:04}", (20_000u64 - 1) % 10000) ); } + + #[tokio::test] + async fn fixed_width_update_on_unsized_row_stays_in_place() { + let table = UnsizedUpdateWorkTable::default(); + table + .insert(UnsizedUpdateRow { + id: 1, + payload: "out-of-line payload that must remain unchanged".to_string(), + balance: 1.0, + }) + .unwrap(); + let before = link_of(&table, 1); + + table + .update_balance(BalanceQuery { balance: 42.5 }, 1) + .await + .unwrap(); + + let after = link_of(&table, 1); + let row = table.select(1).unwrap(); + assert_eq!(before, after); + assert_eq!(row.balance, 42.5); + assert_eq!( + row.payload, + "out-of-line payload that must remain unchanged" + ); + } } }; }