From b81f11cce6b2efe02f3fb40c48a5c1148fcf2347 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 5 Aug 2026 07:11:15 +0700 Subject: [PATCH 1/6] perf: subtract reusable data ranges in one pass --- src/persistence/space/data.rs | 256 +++++++++++++++++++++++++++------- 1 file changed, 203 insertions(+), 53 deletions(-) diff --git a/src/persistence/space/data.rs b/src/persistence/space/data.rs index caa1edd..b4df72d 100644 --- a/src/persistence/space/data.rs +++ b/src/persistence/space/data.rs @@ -1,3 +1,4 @@ +use std::collections::HashSet; use std::io::SeekFrom; use std::path::Path; @@ -19,6 +20,121 @@ use rkyv::{Archive, Deserialize, Serialize}; use tokio::fs::File; use tokio::io::{AsyncSeekExt, AsyncWriteExt}; +fn link_sort_key(link: &Link) -> (u32, u32) { + (link.page_id.into(), link.offset) +} + +fn link_end(link: &Link) -> u64 { + u64::from(link.offset) + u64::from(link.length) +} + +/// Sorts and coalesces ranges within each page. +fn normalize_ranges(mut ranges: Vec) -> (Vec, bool) { + let was_sorted = ranges + .windows(2) + .all(|pair| link_sort_key(&pair[0]) <= link_sort_key(&pair[1])); + ranges.sort_unstable_by_key(link_sort_key); + + let mut changed = !was_sorted; + let mut normalized: Vec = Vec::with_capacity(ranges.len()); + for range in ranges { + if range.length == 0 { + changed = true; + continue; + } + + if let Some(last) = normalized.last_mut() + && last.page_id == range.page_id + && u64::from(range.offset) <= link_end(last) + { + let end = link_end(last).max(link_end(&range)); + last.length = (end - u64::from(last.offset)) as u32; + changed = true; + continue; + } + + normalized.push(range); + } + + (normalized, changed) +} + +/// Subtracts sorted, coalesced used ranges from sorted, coalesced free ranges. +/// +/// Both cursors only move forward, so the subtraction is O(f log f + u log u) +/// for sorting and O(f + u) for the scan instead of rebuilding the full free +/// list once per used link. +fn subtract_used_ranges(free_ranges: Vec, used_ranges: impl IntoIterator) -> (Vec, bool) { + let (free_ranges, mut changed) = normalize_ranges(free_ranges); + let (used_ranges, _) = normalize_ranges(used_ranges.into_iter().collect()); + if used_ranges.is_empty() { + return (free_ranges, changed); + } + + let mut remaining = Vec::with_capacity(free_ranges.len() + used_ranges.len()); + let mut used_index = 0; + + for free in free_ranges { + let free_page: u32 = free.page_id.into(); + let free_start = u64::from(free.offset); + let free_end = link_end(&free); + + while let Some(used) = used_ranges.get(used_index) { + let used_page: u32 = used.page_id.into(); + if used_page < free_page || (used_page == free_page && link_end(used) <= free_start) { + used_index += 1; + } else { + break; + } + } + + let mut cursor = free_start; + let mut scan = used_index; + while let Some(used) = used_ranges.get(scan) { + let used_page: u32 = used.page_id.into(); + let used_start = u64::from(used.offset); + let used_end = link_end(used); + if used_page != free_page || used_start >= free_end { + break; + } + + if used_end > cursor { + if cursor < used_start { + let segment_end = used_start.min(free_end); + remaining.push(Link { + page_id: free.page_id, + offset: cursor as u32, + length: (segment_end - cursor) as u32, + }); + } + + let overlap_end = used_end.min(free_end); + if cursor.max(used_start) < overlap_end { + changed = true; + cursor = overlap_end; + } + } + + if used_end <= free_end { + scan += 1; + } else { + break; + } + } + used_index = scan; + + if cursor < free_end { + remaining.push(Link { + page_id: free.page_id, + offset: cursor as u32, + length: (free_end - cursor) as u32, + }); + } + } + + (remaining, changed) +} + #[derive(Debug)] pub struct SpaceData { pub info: GeneralPage>, @@ -44,54 +160,9 @@ impl SpaceData

) -> bool { - let mut changed = false; - - for used in used_links { - let used_start = u64::from(used.offset); - let used_end = used_start + u64::from(used.length); - let mut remaining = Vec::with_capacity(self.info.inner.empty_links_list.len() + 1); - - for free in self.info.inner.empty_links_list.drain(..) { - if free.page_id != used.page_id { - remaining.push(free); - continue; - } - - let free_start = u64::from(free.offset); - let free_end = free_start + u64::from(free.length); - let overlap_start = free_start.max(used_start); - let overlap_end = free_end.min(used_end); - if overlap_start >= overlap_end { - remaining.push(free); - continue; - } - - changed = true; - if free_start < overlap_start { - remaining.push(Link { - page_id: free.page_id, - offset: free.offset, - length: (overlap_start - free_start) as u32, - }); - } - if overlap_end < free_end { - remaining.push(Link { - page_id: free.page_id, - offset: overlap_end as u32, - length: (free_end - overlap_end) as u32, - }); - } - } - - self.info.inner.empty_links_list = remaining; - } - - if changed { - self.info.inner.empty_links_list.sort_by_key(|link| { - let page_id: u32 = link.page_id.into(); - (page_id, link.offset) - }); - } + let free_ranges = std::mem::take(&mut self.info.inner.empty_links_list); + let (remaining, changed) = subtract_used_ranges(free_ranges, used_links); + self.info.inner.empty_links_list = remaining; changed } } @@ -184,7 +255,7 @@ where async fn save_batch_data(&mut self, batch_data: BatchData) -> eyre::Result<()> { let used_links = batch_data.values().flat_map(|ops| ops.iter().map(|(link, _)| *link)); - if self.consume_reusable_ranges(used_links.collect::>()) { + if self.consume_reusable_ranges(used_links) { self.save_info().await?; } @@ -248,15 +319,13 @@ where } async fn reclaim_data_pages(&mut self, page_ids: Vec) -> eyre::Result<()> { - let mut page_ids = page_ids + let page_ids = page_ids .into_iter() .filter(|page_id| { let id: u32 = (*page_id).into(); id != 0 && id <= self.last_page_id }) - .collect::>(); - page_ids.sort_unstable(); - page_ids.dedup(); + .collect::>(); if page_ids.is_empty() { return Ok(()); @@ -266,6 +335,8 @@ where .inner .empty_links_list .retain(|link| !page_ids.contains(&link.page_id)); + let mut page_ids = page_ids.into_iter().collect::>(); + page_ids.sort_unstable(); self.info .inner .empty_links_list @@ -294,3 +365,82 @@ where Ok(()) } } + +#[cfg(test)] +mod tests { + use data_bucket::page::PageId; + + use super::subtract_used_ranges; + use crate::prelude::Link; + + fn link(page_id: u32, offset: u32, length: u32) -> Link { + Link { + page_id: PageId::from(page_id), + offset, + length, + } + } + + #[test] + fn reusable_ranges_are_subtracted_in_one_sorted_scan() { + let free = vec![link(2, 0, 50), link(1, 0, 100)]; + let used = vec![link(1, 40, 20), link(2, 0, 10), link(1, 10, 20), link(1, 25, 30)]; + + let (remaining, changed) = subtract_used_ranges(free, used); + + assert!(changed); + assert_eq!(remaining, vec![link(1, 0, 10), link(1, 60, 40), link(2, 10, 40)]); + } + + #[test] + fn non_overlapping_used_ranges_leave_free_ranges_unchanged() { + let free = vec![link(1, 0, 10), link(1, 20, 10), link(2, 0, 10)]; + let used = vec![link(1, 10, 10), link(3, 0, 10)]; + + let (remaining, changed) = subtract_used_ranges(free.clone(), used); + + assert!(!changed); + assert_eq!(remaining, free); + } + + #[test] + fn randomized_subtraction_matches_byte_level_coverage() { + const PAGES: usize = 4; + const BYTES: usize = 64; + let mut rng = fastrand::Rng::with_seed(0x51ce_5eed); + + for case in 0..1_000 { + let mut free = Vec::new(); + let mut used = Vec::new(); + let mut expected = [[false; BYTES]; PAGES]; + + for _ in 0..rng.usize(0..20) { + let page = rng.usize(0..PAGES); + let start = rng.usize(0..BYTES); + let end = rng.usize(start + 1..=BYTES); + free.push(link((page + 1) as u32, start as u32, (end - start) as u32)); + expected[page][start..end].fill(true); + } + for _ in 0..rng.usize(0..20) { + let page = rng.usize(0..PAGES); + let start = rng.usize(0..BYTES); + let end = rng.usize(start + 1..=BYTES); + used.push(link((page + 1) as u32, start as u32, (end - start) as u32)); + expected[page][start..end].fill(false); + } + + let (remaining, _) = subtract_used_ranges(free, used); + let mut actual = [[false; BYTES]; PAGES]; + for range in remaining { + let page: u32 = range.page_id.into(); + let page = page as usize - 1; + let start = range.offset as usize; + let end = start + range.length as usize; + assert!(actual[page][start..end].iter().all(|occupied| !occupied), "case {case}"); + actual[page][start..end].fill(true); + } + + assert_eq!(actual, expected, "case {case}"); + } + } +} From b980f18da768e21fd6050c356c3bcadf8cace9ea Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 5 Aug 2026 07:12:17 +0700 Subject: [PATCH 2/6] perf: use FxHash for logical WTI mutation stripes --- Cargo.toml | 1 + src/index/persistent_wti.rs | 32 +++++++++++++++++++++++++++----- 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index b30b580..7b274bd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -58,6 +58,7 @@ prettytable-rs = "^0.10" psc-nanoid = { version = "3.1.1", features = ["rkyv", "packed"] } rkyv = { version = "0.8.17", features = ["uuid-1"] } reqwest = { version = "0.12", optional = true, default-features = false, features = ["rustls-tls-webpki-roots", "charset", "http2"] } +rustc-hash = "2.1.1" rusty-s3 = { version = "0.10.2", optional = true } smart-default = "0.7.1" tokio = { version = "1", features = ["full"] } diff --git a/src/index/persistent_wti.rs b/src/index/persistent_wti.rs index 0853ffd..abbc0ad 100644 --- a/src/index/persistent_wti.rs +++ b/src/index/persistent_wti.rs @@ -10,7 +10,6 @@ //! that marker and derives the real structural metadata itself. use std::array; -use std::collections::hash_map::DefaultHasher; use std::fmt::{self, Debug}; use std::hash::{Hash, Hasher}; use std::ops::RangeBounds; @@ -22,17 +21,28 @@ use indexset::cdc::change::{ChangeEvent, Id}; use indexset::core::node::NodeLike; use indexset::core::pair::Pair; use parking_lot::{Mutex, MutexGuard}; +use rustc_hash::FxHasher; use crate::index::UniqueIndex; use crate::util::OffsetEqLink; use crate::{IndexMap, TableIndexCdc}; -// A stripe provides per-key exclusion only: DefaultHasher has no relationship +// A stripe provides per-key exclusion only: FxHasher has no relationship // to key order, so callers must not infer range or cross-key ordering from the // selected mutex. Reads never touch this fixed inline table. Logical batches // are ordered independently by event id in the persistence worker. const MUTATION_STRIPES: usize = 64; +#[inline] +fn mutation_stripe_index(key: &Q) -> usize { + // Stripe selection is not a security boundary. FxHasher avoids SipHash's + // per-mutation cost and distributes the overwhelmingly common sequential + // integer keys across the power-of-two stripe table. + let mut hasher = FxHasher::default(); + key.hash(&mut hasher); + hasher.finish() as usize & (MUTATION_STRIPES - 1) +} + /// A persisted WorkTablesIndex whose foreground mutations emit logical CDC. /// /// Point reads delegate directly to the native index. There is no runtime @@ -45,6 +55,9 @@ where { inner: IndexMap, next_event_id: AtomicU64, + // Fixed inline allocation: 64 parking_lot mutexes per persisted WTI. The + // enclosing index's allocation size accounts for these; there is no + // per-mutation or per-key mutex allocation. mutation_stripes: [Mutex<()>; MUTATION_STRIPES], } @@ -96,10 +109,9 @@ where &self.inner } + #[inline] fn mutation_stripe(&self, key: &Q) -> MutexGuard<'_, ()> { - let mut hasher = DefaultHasher::new(); - key.hash(&mut hasher); - self.mutation_stripes[hasher.finish() as usize % MUTATION_STRIPES].lock() + self.mutation_stripes[mutation_stripe_index(key)].lock() } fn next_event_id(&self) -> Id { @@ -267,6 +279,8 @@ where #[cfg(test)] mod tests { + use std::collections::HashSet; + use data_bucket::page::PageId; use super::*; @@ -305,4 +319,12 @@ mod tests { assert!(index.insert_checked_cdc(8, link(8)).is_some()); assert_eq!(index.next_event_id.load(Ordering::Relaxed), 2); } + + #[test] + fn sequential_integer_keys_use_every_mutation_stripe() { + let stripes = (0_u64..4_096) + .map(|key| mutation_stripe_index(&key)) + .collect::>(); + assert_eq!(stripes.len(), MUTATION_STRIPES); + } } From 505785ccc765330aef67dd1fb482f74c2fa7bfc1 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 5 Aug 2026 07:14:22 +0700 Subject: [PATCH 3/6] perf: reuse locked row on in-place fallback --- .../generators/in_memory/queries/update.rs | 6 ++-- src/in_memory/data.rs | 4 +++ src/in_memory/pages.rs | 33 +++++++++++++++++++ 3 files changed, 41 insertions(+), 2 deletions(-) diff --git a/codegen/src/generators/in_memory/queries/update.rs b/codegen/src/generators/in_memory/queries/update.rs index a538c7f..d2c5f65 100644 --- a/codegen/src/generators/in_memory/queries/update.rs +++ b/codegen/src/generators/in_memory/queries/update.rs @@ -367,8 +367,10 @@ impl InMemoryGenerator { return core::result::Result::Ok(()); } - let row_old_for_reinsert = self.0.select(pk.clone()).expect("should not be deleted by other thread"); - if let Err(e) = self.reinsert(row_old_for_reinsert, row_new).await { + // `update_in_place` checks serialization and exact slot + // length before touching page bytes, so its error path + // leaves this locked snapshot authoritative for fallback. + if let Err(e) = self.reinsert(row_old, row_new).await { self.0.update_state.remove(&pk); return Err(e); } diff --git a/src/in_memory/data.rs b/src/in_memory/data.rs index 789bb83..a440dd0 100644 --- a/src/in_memory/data.rs +++ b/src/in_memory/data.rs @@ -135,6 +135,10 @@ impl Data { if length != link.length { return Err(ExecutionError::InvalidLink); } + debug_assert_eq!( + length, link.length, + "slot length was checked before archived bytes are overwritten" + ); let inner_data = unsafe { &mut *self.inner_data.get() }; inner_data[link.offset as usize..][..link.length as usize].copy_from_slice(bytes.as_slice()); diff --git a/src/in_memory/pages.rs b/src/in_memory/pages.rs index b5e8df9..156e506 100644 --- a/src/in_memory/pages.rs +++ b/src/in_memory/pages.rs @@ -645,6 +645,12 @@ where /// generated persisted update path deliberately keeps the reinsert path for /// this reason. /// + /// Serialization and the exact-length check finish before any page byte is + /// changed. `page_access` excludes low-level archived-page readers during + /// the copy, while generated reads continue from the old immutable + /// publication until [`Self::publish_wrapped_row`] replaces the complete + /// owned row and flags together. + /// /// # Safety /// Same contract as [`Self::update`]: `link` must be valid and no other /// mutable references to the row may exist during modification. @@ -909,6 +915,7 @@ mod tests { use crate::in_memory::pages::{DataPages, ExecutionError}; use crate::in_memory::{DATA_INNER_LENGTH, PagesExecutionError, RowWrapper, StorableRow}; use crate::prelude::ArchivedRowWrapper; + use data_bucket::Link; #[derive(Archive, Copy, Clone, Deserialize, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)] struct TestRow { @@ -1089,6 +1096,32 @@ mod tests { assert_eq!(pages.select_non_ghosted(link), Ok(TestRow { a: 1, b: 1 })); } + #[test] + fn failed_exact_length_update_preserves_page_bytes_and_publication() { + let pages = DataPages::::new(); + let old_row = TestRow { a: 10, b: 20 }; + let link = pages.insert(old_row).unwrap(); + unsafe { + pages.with_mut_ref(link, |row| row.unghost()).unwrap(); + } + let old_bytes = pages.select_raw(link).unwrap(); + let wrong_length = Link { + length: link.length - 1, + ..link + }; + + let result = unsafe { pages.update_in_place::(TestRow { a: 30, b: 40 }, wrong_length) }; + + assert!(matches!( + result, + Err(ExecutionError::DataPageError( + crate::in_memory::DataExecutionError::InvalidLink + )) + )); + assert_eq!(pages.select_raw(link).unwrap(), old_bytes); + assert_eq!(pages.select_non_ghosted(link), Ok(old_row)); + } + #[test] fn retired_version_survives_link_reuse_for_in_flight_reader() { let pages = DataPages::::new(); From 79f9b4ec4cc3df93d0b1f056ad120cfa16f14bc0 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 5 Aug 2026 07:15:48 +0700 Subject: [PATCH 4/6] test: cover async row-lock cancellation cleanup --- src/lock/map.rs | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/src/lock/map.rs b/src/lock/map.rs index af25c55..088aedf 100644 --- a/src/lock/map.rs +++ b/src/lock/map.rs @@ -98,6 +98,16 @@ impl Drop for MutationGuard { } } +/// Registry for per-row async locks and synchronous mutation stripes. +/// +/// # Sync/async lock boundary +/// +/// The `parking_lot` map guard is never returned and never crosses an +/// `.await`. Acquisition clones a tracked `Arc>` before +/// releasing the map guard. Cleanup may synchronously take the short-lived map +/// write guard, but only probes the per-row lock with `try_read`; it never waits +/// on a Tokio lock while holding the map. This one-way boundary prevents a +/// map-lock/per-row-lock cycle during cancellation and `Drop`. #[derive(Debug)] pub struct LockMap { map: RwLock>>, @@ -119,6 +129,12 @@ impl LockMap where PrimaryKey: Hash + Eq + Debug + Clone, { + /// Inserts a raw lock entry. + /// + /// A returned or externally retained `Arc` pins cleanup through + /// `Arc::strong_count`. Generated operations should prefer + /// [`Self::get_or_insert_with`], whose [`LockAcquirer`] makes cancellation + /// tracking explicit. pub fn insert( &self, key: PrimaryKey, @@ -136,6 +152,8 @@ where .map(|entry| entry.lock) } + /// Returns an untracked raw lock clone, which keeps the map entry alive + /// until that clone is dropped. pub fn get(&self, key: &PrimaryKey) -> Option>> { self.map.read().get(key).map(|entry| entry.lock.clone()) } @@ -278,4 +296,24 @@ mod tests { drop(second); assert!(!lock_map.map.read().contains_key(&33)); } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelling_async_waiter_releases_tracking_without_deadlock() { + let lock_map: Arc> = Arc::new(LockMap::default()); + let owner = lock_map.get_or_insert_with(41, FullRowLock::new); + let owner_guard = owner.write().await; + let waiter = lock_map.get_or_insert_with(41, FullRowLock::new); + let waiting_task = tokio::spawn(async move { + let _guard = waiter.write().await; + }); + tokio::task::yield_now().await; + + waiting_task.abort(); + assert!(waiting_task.await.unwrap_err().is_cancelled()); + assert!(lock_map.map.read().contains_key(&41)); + + drop(owner_guard); + drop(owner); + assert!(!lock_map.map.read().contains_key(&41)); + } } From eeb3b39be6b20a800eeaabba569b10b1fba50d90 Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 5 Aug 2026 07:19:36 +0700 Subject: [PATCH 5/6] test: harden persistence barriers and quarantine --- src/persistence/space/logical_index.rs | 57 +++++++++++++++++++++ src/persistence/task.rs | 68 +++++++++++++++++++++----- 2 files changed, 113 insertions(+), 12 deletions(-) diff --git a/src/persistence/space/logical_index.rs b/src/persistence/space/logical_index.rs index f5d637d..7f8fd86 100644 --- a/src/persistence/space/logical_index.rs +++ b/src/persistence/space/logical_index.rs @@ -306,6 +306,8 @@ where #[cfg(test)] mod tests { + use std::collections::BTreeMap as StdBTreeMap; + use data_bucket::page::PageId; use super::*; @@ -422,4 +424,59 @@ mod tests { assert!(!structural.is_empty()); assert_eq!(shadow.get(&7).map(|entry| entry.get().value), Some(link(8))); } + + #[test] + fn shuffled_large_logical_batch_replays_in_event_id_order() { + let shadow = BTreeMap::::default(); + let mut expected = StdBTreeMap::::new(); + let mut events = Vec::new(); + + for event_id in 0_u64..2_000 { + let key = event_id.wrapping_mul(17) % 97; + let event = if event_id % 5 == 0 { + if let Some(old_link) = expected.remove(&key) { + let pair = Pair { key, value: old_link }; + ChangeEvent::RemoveAt { + event_id: event_id.into(), + max_value: pair.clone(), + value: pair, + index: 0, + } + } else { + let new_link = link(event_id as u32); + expected.insert(key, new_link); + let pair = Pair { key, value: new_link }; + ChangeEvent::InsertAt { + event_id: event_id.into(), + max_value: pair.clone(), + value: pair, + index: 0, + } + } + } else { + let new_link = link(event_id as u32); + expected.insert(key, new_link); + let pair = Pair { key, value: new_link }; + ChangeEvent::InsertAt { + event_id: event_id.into(), + max_value: pair.clone(), + value: pair, + index: 0, + } + }; + events.push(event); + } + + fastrand::Rng::with_seed(0x10_91ca1).shuffle(&mut events); + let structural = translate_logical_batch(Path::new("test.wt.idx"), &shadow, events).unwrap(); + + assert!(!structural.is_empty()); + for key in 0..97 { + assert_eq!( + shadow.get(&key).map(|entry| entry.get().value), + expected.get(&key).copied(), + "key {key}" + ); + } + } } diff --git a/src/persistence/task.rs b/src/persistence/task.rs index 5e4b51e..a80a3a7 100644 --- a/src/persistence/task.rs +++ b/src/persistence/task.rs @@ -403,7 +403,14 @@ mod lifecycle_tests { batches: Arc, events: Arc>>, config: TestConfig, - fail: bool, + failure: TestFailure, + } + + #[derive(Clone, Copy)] + enum TestFailure { + None, + Engine, + IndexCorruption, } impl PersistenceEngine<(), u64, TestEvents, TestIndex> for TestEngine { @@ -414,7 +421,7 @@ mod lifecycle_tests { batches: Arc::new(AtomicUsize::new(0)), events: Arc::new(ParkingMutex::new(Vec::new())), config, - fail: false, + failure: TestFailure::None, }) } @@ -426,8 +433,14 @@ mod lifecycle_tests { &mut self, _batch_op: BatchOperation<(), u64, TestEvents, TestIndex>, ) -> eyre::Result<()> { - if self.fail { - return Err(eyre::eyre!("injected batch failure")); + match self.failure { + TestFailure::None => {} + TestFailure::Engine => return Err(eyre::eyre!("injected batch failure")), + TestFailure::IndexCorruption => { + return Err( + PersistenceIndexCorruption::new("table/primary.wt.idx", "injected shadow divergence").into(), + ); + } } self.batches.fetch_add(1, Ordering::Relaxed); self.events.lock().push("batch"); @@ -466,7 +479,7 @@ mod lifecycle_tests { batches: batches.clone(), events: Arc::new(ParkingMutex::new(Vec::new())), config: TestConfig, - fail: false, + failure: TestFailure::None, }); task.apply_operation(insert_operation(1)).unwrap(); @@ -481,7 +494,7 @@ mod lifecycle_tests { batches: Arc::new(AtomicUsize::new(0)), events: Arc::new(ParkingMutex::new(Vec::new())), config: TestConfig, - fail: true, + failure: TestFailure::Engine, }); task.apply_operation(insert_operation(1)).unwrap(); @@ -502,7 +515,7 @@ mod lifecycle_tests { batches: Arc::new(AtomicUsize::new(0)), events: events.clone(), config: TestConfig, - fail: false, + failure: TestFailure::None, }); task.apply_operation(insert_operation(1)).unwrap(); @@ -513,6 +526,25 @@ mod lifecycle_tests { assert_eq!(&*events.lock(), &["batch", "reclaim", "batch"]); } + #[tokio::test] + async fn index_corruption_from_engine_is_typed_and_terminal_for_callers() { + let task = PersistenceTask::run_engine(TestEngine { + batches: Arc::new(AtomicUsize::new(0)), + events: Arc::new(ParkingMutex::new(Vec::new())), + config: TestConfig, + failure: TestFailure::IndexCorruption, + }); + + task.apply_operation(insert_operation(1)).unwrap(); + let wait_error = task.wait_for_ops().await.unwrap_err(); + assert!(matches!(wait_error.as_ref(), PersistenceError::IndexCorruption(_))); + + let intake_error = task.apply_operation(insert_operation(2)).unwrap_err(); + assert!(Arc::ptr_eq(&wait_error, &intake_error)); + let close_error = task.close().await.unwrap_err(); + assert!(Arc::ptr_eq(&wait_error, &close_error)); + } + #[test] fn typed_index_corruption_quarantines_the_persistence_lifecycle() { let lifecycle = PersistenceLifecycle::new(); @@ -840,11 +872,23 @@ impl } else { tokio::time::sleep(Duration::from_millis(500)).await; } - } else if let Some(page_ids) = pending_reclaim.take() - && let Err(error) = engine.reclaim_data_pages(page_ids).await - { - engine_lifecycle.fail(error); - return; + } else if let Some(page_ids) = pending_reclaim.take() { + // `get_first_op_id_available() == None` is only sufficient + // when the analyzer itself is empty. If its operation-id + // index ever loses an entry, reclaiming here would make a + // source page reusable before its buffered row move became + // durable. Fail terminally instead of trusting that state. + let buffered_operations = analyzer.len(); + if buffered_operations != 0 { + engine_lifecycle.fail(eyre::eyre!( + "persistence reclamation barrier found {buffered_operations} buffered operations without an operation-id index entry" + )); + return; + } + if let Err(error) = engine.reclaim_data_pages(page_ids).await { + engine_lifecycle.fail(error); + return; + } } } }; From fe4b6424a2deab74137e810ddebea726c625bb4f Mon Sep 17 00:00:00 2001 From: meh Date: Wed, 5 Aug 2026 07:27:36 +0700 Subject: [PATCH 6/6] perf: generate distinct full-row update paths --- .../generators/in_memory/queries/update.rs | 85 ++++++++++--------- 1 file changed, 46 insertions(+), 39 deletions(-) diff --git a/codegen/src/generators/in_memory/queries/update.rs b/codegen/src/generators/in_memory/queries/update.rs index d2c5f65..949f9dc 100644 --- a/codegen/src/generators/in_memory/queries/update.rs +++ b/codegen/src/generators/in_memory/queries/update.rs @@ -66,8 +66,36 @@ impl InMemoryGenerator { // column and therefore always reinserts, correctly.) let const_name = name_generator.get_page_inner_size_const_ident(); let full_row_in_place_eligible = !self.columns.is_sized && self.columns.indexes.is_empty(); - let size_check = if self.columns.is_sized { - quote! {} + let update_body = if self.columns.is_sized { + quote! { + let mut bytes = rkyv::to_bytes::(&row) + .map_err(|_| WorkTableError::SerializeError)?; + let mut archived_row = unsafe { + rkyv::access_unchecked_mut::<<#row_ident as rkyv::Archive>::Archived>(&mut bytes[..]) + .unseal_unchecked() + }; + + let op_id = OperationId::Single(uuid::Uuid::now_v7()); + #diff_process_insert + #persist_op + + unsafe { + self.0 + .data + .with_mut_ref(link, move |archived| { + #(#row_updates)* + }) + .map_err(WorkTableError::PagesError)? + }; + + #diff_process_remove + + self.0.update_state.remove(&pk); + + #persist_call + + core::result::Result::Ok(()) + } } else if full_row_in_place_eligible { quote! { // No secondary indexes: same-size unsized full-row update may go @@ -93,29 +121,27 @@ impl InMemoryGenerator { return Err(e); } self.0.update_state.remove(&pk); - return core::result::Result::Ok(()); + core::result::Result::Ok(()) } } else { quote! { - if true { - drop(_guard); - let op_lock = { #full_row_lock }; - let _guard = LockGuard::new_with_mutation( - op_lock, - self.0.lock_manager.clone(), - pk.clone(), - ); - let row_old = self.0.data.select_non_ghosted(link)?; - if let Err(e) = self.reinsert(row_old, row).await { - self.0.update_state.remove(&pk); - - return Err(e); - } - + drop(_guard); + let op_lock = { #full_row_lock }; + let _guard = LockGuard::new_with_mutation( + op_lock, + self.0.lock_manager.clone(), + pk.clone(), + ); + let row_old = self.0.data.select_non_ghosted(link)?; + if let Err(e) = self.reinsert(row_old, row).await { self.0.update_state.remove(&pk); - return core::result::Result::Ok(()); + return Err(e); } + + self.0.update_state.remove(&pk); + + core::result::Result::Ok(()) } }; @@ -150,26 +176,7 @@ impl InMemoryGenerator { let row_old = self.0.data.select_non_ghosted(link)?; self.0.update_state.insert(pk.clone(), row_old); - let mut bytes = rkyv::to_bytes::(&row).map_err(|_| WorkTableError::SerializeError)?; - #size_check - - let mut archived_row = unsafe { rkyv::access_unchecked_mut::<<#row_ident as rkyv::Archive>::Archived>(&mut bytes[..]).unseal_unchecked() }; - - let op_id = OperationId::Single(uuid::Uuid::now_v7()); - #diff_process_insert - #persist_op - - unsafe { self.0.data.with_mut_ref(link, move |archived| { - #(#row_updates)* - }).map_err(WorkTableError::PagesError)? }; - - #diff_process_remove - - self.0.update_state.remove(&pk); - - #persist_call - - core::result::Result::Ok(()) + #update_body } } }