From c5f7a6b1e983137e03c8bf9c2f9f0e72b37949cd Mon Sep 17 00:00:00 2001 From: meh Date: Sat, 8 Aug 2026 16:20:35 +0700 Subject: [PATCH] fix: carry TOC inserts across persisted segments --- .../space/index/table_of_contents.rs | 82 +++++++++++++------ .../sync/repeated_string_upsert.rs | 10 +++ 2 files changed, 68 insertions(+), 24 deletions(-) diff --git a/src/persistence/space/index/table_of_contents.rs b/src/persistence/space/index/table_of_contents.rs index 181ef08..2f44b9a 100644 --- a/src/persistence/space/index/table_of_contents.rs +++ b/src/persistence/space/index/table_of_contents.rs @@ -58,31 +58,47 @@ where where T: Clone + SizeMeasurable, { + // Replacing an existing identity must not leave duplicate mappings in + // separate TOC segments or inflate the page's estimated size. + let mut first_existing = None; + for index in 0..self.pages.len() { + if self.pages[index].inner.contains(&node_id) { + self.pages[index].inner.remove_without_record(&node_id); + first_existing.get_or_insert(index); + } + } + if let Some(first) = first_existing { + self.current_page = first; + } + let next_page_id = self.next_page_id.clone(); - let page = self.get_current_page_mut(); - page.inner.insert(node_id.clone(), page_id); - if page.inner.estimated_size() > DATA_LENGTH as usize { + loop { + let page = self.get_current_page_mut(); + page.inner.insert(node_id.clone(), page_id); + if page.inner.estimated_size() <= DATA_LENGTH as usize { + return; + } page.inner.remove_without_record(&node_id); - if page.header.next_id.is_empty() { - let next_page_id = next_page_id.fetch_add(1, Ordering::Relaxed); - let header = page.header.follow_with_page_id(next_page_id.into()); - page.header.next_id = next_page_id.into(); - self.pages.push(GeneralPage { - header, - inner: TableOfContentsPage::default(), - }); - self.current_page += 1; - let page = self.get_current_page_mut(); - page.inner.insert(node_id.clone(), page_id); - } else { - let mut i = self.current_page; - while !self.pages[i].header.next_id.is_empty() { - i += 1; - } - self.current_page = i; + if !page.header.next_id.is_empty() { + self.current_page += 1; + continue; } + + let next_page_id = next_page_id.fetch_add(1, Ordering::Relaxed); + let header = page.header.follow_with_page_id(next_page_id.into()); + page.header.next_id = next_page_id.into(); + let mut next_page = GeneralPage { + header, + inner: TableOfContentsPage::default(), + }; + // Preserve the old behavior for an entry larger than one segment: + // it gets its own page rather than creating pages forever. + next_page.inner.insert(node_id, page_id); + self.pages.push(next_page); + self.current_page = self.pages.len() - 1; + return; } } @@ -211,6 +227,7 @@ where #[cfg(test)] mod tests { use crate::persistence::space::index::table_of_contents::IndexTableOfContents; + use data_bucket::page::PageId; use std::sync::Arc; use std::sync::atomic::AtomicU32; @@ -280,6 +297,23 @@ mod tests { assert!(keys.is_empty(), "Some keys was not inserted: {keys:?}") } + #[test] + fn insert_reaches_existing_tail_after_reload_resets_cursor() { + let mut toc = IndexTableOfContents::::new(0.into(), Arc::new(AtomicU32::new(0))); + for key in 0..10 { + toc.insert(key, u32::from(key).into()); + } + assert!(toc.pages.len() > 1, "fixture must span TOC pages"); + + // `parse_from_file` starts at the first TOC segment. When that segment + // is already full and has a successor, a new page identity must carry + // forward until an existing or newly-created tail can accept it. + toc.current_page = 0; + toc.insert(200, PageId::from(200)); + + assert_eq!(toc.get(&200), Some(PageId::from(200))); + } + #[test] fn reinsert_on_empty_space() { let mut toc = IndexTableOfContents::::new(0.into(), Arc::new(AtomicU32::new(0))); @@ -313,10 +347,10 @@ mod tests { let new_key = keys.last().unwrap() + 1; let id = toc.pop_empty_page_id().unwrap(); toc.insert(new_key, id); - assert_eq!( - before_remove_current_page, toc.current_page, - "`current_page` not moved back to before remove state and is {}", - toc.current_page, + assert_eq!(toc.get(&new_key), Some(id), "reused page id was not recorded"); + assert!( + toc.current_page >= before_remove_current_page, + "`current_page` did not advance to a segment that could hold the identity" ); assert_eq!( toc.pages[after_remove_current_page].inner.clone().pop_empty_page(), diff --git a/tests/persistence/sync/repeated_string_upsert.rs b/tests/persistence/sync/repeated_string_upsert.rs index de9e97a..837fc18 100644 --- a/tests/persistence/sync/repeated_string_upsert.rs +++ b/tests/persistence/sync/repeated_string_upsert.rs @@ -54,6 +54,14 @@ fn repeated_varying_string_upserts_keep_the_worker_healthy() { }) .unwrap(); table.wait_for_ops().await.unwrap(); + table.close().await.unwrap(); + } + { + // Loading a multi-segment TOC resets its insertion cursor to the + // first segment. The next split must carry a new identity through + // the existing chain instead of dropping it at that full page. + let engine = StringBlobPersistenceEngine::new(config.clone()).await.unwrap(); + let table = StringBlobWorkTable::load(engine).await.unwrap(); for revision in 0..1_000 { table @@ -70,11 +78,13 @@ fn repeated_varying_string_upserts_keep_the_worker_healthy() { .await .expect("persistence stalled after repeated string upserts") .expect("persistence worker failed after repeated string upserts"); + table.close().await.unwrap(); } { let engine = StringBlobPersistenceEngine::new(config).await.unwrap(); let table = StringBlobWorkTable::load(engine).await.unwrap(); assert_eq!(table.select("settings".to_string()).unwrap().value.len(), 1_063); + table.close().await.unwrap(); } }); }