diff --git a/codegen/src/persist_table/generator/space_file/worktable_impls.rs b/codegen/src/persist_table/generator/space_file/worktable_impls.rs index 0b69fe6..20745e6 100644 --- a/codegen/src/persist_table/generator/space_file/worktable_impls.rs +++ b/codegen/src/persist_table/generator/space_file/worktable_impls.rs @@ -30,7 +30,7 @@ impl Generator { } else { quote! { /// Returns the physical size of this table's `.wt.data` file. - /// Persisted vacuum currently compacts logical/in-memory pages + /// Persisted vacuum makes freed pages reusable across reloads, /// but does not truncate this file. pub async fn persisted_data_file_size_bytes(&self) -> std::io::Result { self.1.persisted_data_file_size_bytes().await diff --git a/src/persistence/engine.rs b/src/persistence/engine.rs index 6e929bc..37254f6 100644 --- a/src/persistence/engine.rs +++ b/src/persistence/engine.rs @@ -285,6 +285,10 @@ where Ok(()) } + async fn reclaim_data_pages(&mut self, page_ids: Vec) -> eyre::Result<()> { + self.data.reclaim_data_pages(page_ids).await + } + async fn ensure_schema( &mut self, row_schema: Vec<(String, String)>, diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs index df98e79..206d812 100644 --- a/src/persistence/mod.rs +++ b/src/persistence/mod.rs @@ -1,5 +1,7 @@ use std::future::Future; +use data_bucket::page::PageId; + use crate::persistence::operation::BatchOperation; pub use engine::DiskConfig; @@ -57,6 +59,15 @@ pub trait PersistenceEngine, ) -> impl Future> + Send; + /// Persists whole data pages made reusable by vacuum. + /// + /// The persistence task invokes this only after every row move queued + /// before the reclamation barrier has reached the engine. Custom engines + /// that do not manage data pages may keep the default no-op. + fn reclaim_data_pages(&mut self, _page_ids: Vec) -> impl Future> + Send { + async { Ok(()) } + } + /// Installs the generated table schema and rejects a non-empty schema that /// belongs to a different table shape. /// Custom engines may keep the default no-op when they do not expose diff --git a/src/persistence/space/data.rs b/src/persistence/space/data.rs index 1fbffea..caa1edd 100644 --- a/src/persistence/space/data.rs +++ b/src/persistence/space/data.rs @@ -37,6 +37,63 @@ 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) + }); + } + changed + } } impl SpaceDataOps @@ -100,6 +157,9 @@ where } async fn save_data(&mut self, link: Link, bytes: &[u8]) -> eyre::Result<()> { + if self.consume_reusable_ranges([link]) { + self.save_info().await?; + } if link.page_id > self.last_page_id.into() { let mut page = GeneralPage { header: GeneralHeader::new(link.page_id, PageType::Data, 0.into()), @@ -123,6 +183,11 @@ 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::>()) { + self.save_info().await?; + } + let page_ids = batch_data.keys().map(|id| (*id).into()).collect::>(); let ids_to_create = page_ids .iter() @@ -182,6 +247,40 @@ where Ok(()) } + async fn reclaim_data_pages(&mut self, page_ids: Vec) -> eyre::Result<()> { + let mut 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(); + + if page_ids.is_empty() { + return Ok(()); + } + + self.info + .inner + .empty_links_list + .retain(|link| !page_ids.contains(&link.page_id)); + self.info + .inner + .empty_links_list + .extend(page_ids.into_iter().map(|page_id| Link { + page_id, + offset: 0, + length: INNER_PAGE_SIZE as u32, + })); + self.info.inner.empty_links_list.sort_by_key(|link| { + let page_id: u32 = link.page_id.into(); + (page_id, link.offset) + }); + self.save_info().await + } + fn get_mut_info(&mut self) -> &mut GeneralPage> { &mut self.info } diff --git a/src/persistence/space/mod.rs b/src/persistence/space/mod.rs index dee296f..12ca9c3 100644 --- a/src/persistence/space/mod.rs +++ b/src/persistence/space/mod.rs @@ -33,6 +33,7 @@ pub trait SpaceDataOps { fn bootstrap(file: &mut File, table_name: String, version: u32) -> impl Future> + Send; fn save_data(&mut self, link: Link, bytes: &[u8]) -> impl Future> + Send; fn save_batch_data(&mut self, batch_data: BatchData) -> impl Future> + Send; + fn reclaim_data_pages(&mut self, page_ids: Vec) -> impl Future> + Send; fn get_mut_info(&mut self) -> &mut GeneralPage>; fn save_info(&mut self) -> impl Future> + Send; } diff --git a/src/persistence/task.rs b/src/persistence/task.rs index 578fe75..f639341 100644 --- a/src/persistence/task.rs +++ b/src/persistence/task.rs @@ -396,6 +396,7 @@ mod lifecycle_tests { struct TestEngine { batches: Arc, + events: Arc>>, config: TestConfig, fail: bool, } @@ -406,6 +407,7 @@ mod lifecycle_tests { async fn new(config: Self::Config) -> eyre::Result { Ok(Self { batches: Arc::new(AtomicUsize::new(0)), + events: Arc::new(ParkingMutex::new(Vec::new())), config, fail: false, }) @@ -423,6 +425,12 @@ mod lifecycle_tests { return Err(eyre::eyre!("injected batch failure")); } self.batches.fetch_add(1, Ordering::Relaxed); + self.events.lock().push("batch"); + Ok(()) + } + + async fn reclaim_data_pages(&mut self, _page_ids: Vec) -> eyre::Result<()> { + self.events.lock().push("reclaim"); Ok(()) } @@ -451,6 +459,7 @@ mod lifecycle_tests { let batches = Arc::new(AtomicUsize::new(0)); let task = PersistenceTask::run_engine(TestEngine { batches: batches.clone(), + events: Arc::new(ParkingMutex::new(Vec::new())), config: TestConfig, fail: false, }); @@ -465,6 +474,7 @@ mod lifecycle_tests { async fn engine_failure_is_terminal_and_reused_for_later_callers() { let task = PersistenceTask::run_engine(TestEngine { batches: Arc::new(AtomicUsize::new(0)), + events: Arc::new(ParkingMutex::new(Vec::new())), config: TestConfig, fail: true, }); @@ -479,6 +489,30 @@ mod lifecycle_tests { let close_error = task.close().await.unwrap_err(); assert!(Arc::ptr_eq(&wait_error, &close_error)); } + + #[tokio::test] + async fn vacuum_reclamation_waits_for_preceding_row_moves() { + let events = Arc::new(ParkingMutex::new(Vec::new())); + let task = PersistenceTask::run_engine(TestEngine { + batches: Arc::new(AtomicUsize::new(0)), + events: events.clone(), + config: TestConfig, + fail: false, + }); + + task.apply_operation(insert_operation(1)).unwrap(); + task.queue.reclaim_pages(vec![1.into()]).unwrap(); + task.apply_operation(insert_operation(2)).unwrap(); + task.wait_for_ops().await.unwrap(); + + assert_eq!(&*events.lock(), &["batch", "reclaim", "batch"]); + } +} + +#[derive(Debug)] +enum PersistenceMessage { + Operation(Operation), + ReclaimPages(Vec), } #[derive(Debug)] @@ -487,7 +521,7 @@ pub struct Queue { // 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>>, + queue: ParkingMutex>>, notify: Notify, // 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" @@ -507,6 +541,13 @@ impl Queue) -> PersistenceResult { + self.push_message(PersistenceMessage::Operation(value)) + } + + fn push_message( + &self, + value: PersistenceMessage, + ) -> PersistenceResult { let state = self.lifecycle.state.lock(); match &*state { PersistenceState::Running => {} @@ -525,10 +566,10 @@ impl Queue Option> { + ) -> Option> { loop { let notified = self.notify.notified(); // Drain values @@ -554,7 +595,7 @@ impl Queue Option> { + fn immediate_pop(&self) -> Option> { if let Some(v) = self.queue.lock().pop_front() { self.len.fetch_sub(1, Ordering::Release); Some(v) @@ -563,10 +604,6 @@ impl Queue impl Iterator> { - std::iter::from_fn(|| self.immediate_pop()) - } - pub fn len(&self) -> usize { self.len.load(Ordering::Acquire) } @@ -594,6 +631,10 @@ where link: new_link, })) } + + fn reclaim_pages(&self, page_ids: Vec) -> PersistenceResult { + self.push_message(PersistenceMessage::ReclaimPages(page_ids)) + } } #[derive(Debug)] @@ -665,9 +706,9 @@ impl /// Returns the current physical size of the table data file. /// - /// This is intentionally separate from `VacuumStats`: vacuum currently - /// compacts and reuses in-memory pages but does not truncate `.wt.data`. - /// Operators can sample this value to observe physical growth. + /// This is intentionally separate from `VacuumStats`: online vacuum makes + /// freed pages durably reusable, but does not truncate `.wt.data`. + /// Operators can sample this value to observe physical growth and reuse. pub async fn persisted_data_file_size_bytes(&self) -> std::io::Result { tokio::fs::metadata(format!( "{}/{}", @@ -709,10 +750,16 @@ impl let task_analyzer_in_progress = analyzer_in_progress.clone(); let task = async move { + let mut pending_reclaim: Option> = None; loop { - let op = if let Some(next_op) = engine_queue.immediate_pop() { - Some(next_op) - } else if analyzer.len() == 0 { + let message = if pending_reclaim.is_none() { + engine_queue.immediate_pop() + } else { + None + }; + let message = if message.is_some() { + message + } else if analyzer.len() == 0 && pending_reclaim.is_none() { task_analyzer_in_progress.store(false, Ordering::Release); engine_lifecycle.notify.notify_waiters(); if matches!(engine_lifecycle.state(), PersistenceState::Closing) { @@ -726,17 +773,35 @@ impl } else { None }; - if let Some(op) = op - && let Err(err) = analyzer.push(op.clone()) - { - engine_lifecycle.fail(err); - return; + + if let Some(message) = message { + match message { + PersistenceMessage::Operation(op) => { + if let Err(err) = analyzer.push(op) { + engine_lifecycle.fail(err); + return; + } + } + PersistenceMessage::ReclaimPages(page_ids) => pending_reclaim = Some(page_ids), + } } - let ops_available_iter = engine_queue.pop_iter(); - if let Err(err) = analyzer.extend_from_iter(ops_available_iter) { - engine_lifecycle.fail(err); - return; + + // Pull operations up to, but never past, a reclamation + // barrier. This gives the analyzer every CDC event required + // for a batch while preserving FIFO ordering for maintenance. + while pending_reclaim.is_none() { + match engine_queue.immediate_pop() { + Some(PersistenceMessage::Operation(op)) => { + if let Err(err) = analyzer.push(op) { + engine_lifecycle.fail(err); + return; + } + } + Some(PersistenceMessage::ReclaimPages(page_ids)) => pending_reclaim = Some(page_ids), + None => break, + } } + if let Some(op_id) = analyzer.get_first_op_id_available() { let batch_op = analyzer.collect_batch_from_op_id(op_id).await; if let Err(e) = batch_op { @@ -751,6 +816,11 @@ 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; } } }; diff --git a/src/table/vacuum/mod.rs b/src/table/vacuum/mod.rs index bcfb414..727ceed 100644 --- a/src/table/vacuum/mod.rs +++ b/src/table/vacuum/mod.rs @@ -1,6 +1,7 @@ use async_trait::async_trait; use data_bucket::Link; +use data_bucket::page::PageId; use indexset::cdc::change::ChangeEvent; use indexset::core::pair::Pair; @@ -38,6 +39,10 @@ pub trait VacuumPersistence: Send + Sync { primary_key_events: Vec>>, secondary_keys_events: SecondaryEvents, ) -> PersistenceResult; + + /// Queue a barrier that makes pages reusable only after all preceding row + /// moves have become durable. + fn reclaim_pages(&self, page_ids: Vec) -> PersistenceResult; } /// Trait for unifying different [`WorkTable`] related [`EmptyDataVacuum`]'s. diff --git a/src/table/vacuum/vacuum.rs b/src/table/vacuum/vacuum.rs index a2fced8..e555c0a 100644 --- a/src/table/vacuum/vacuum.rs +++ b/src/table/vacuum/vacuum.rs @@ -202,13 +202,24 @@ where // expose the whole page for reuse. Otherwise a concurrent insert // could claim a stale fragment between retirement and cleanup. registry.remove_link_for_page(page_from); + if let Some(persistence) = &self.persistence { + // Queue the durable-free marker before publishing this page to + // in-memory allocators. Any concurrent reuse is then ordered + // after the marker and consumes the durable free range again. + persistence.reclaim_pages(vec![page_from])?; + } self.data_pages.mark_page_empty(page_from); pages_freed += 1; } pages_freed += free_pages.len(); + if let Some(persistence) = &self.persistence + && !free_pages.is_empty() + { + persistence.reclaim_pages(free_pages.iter().copied().collect())?; + } for id in free_pages { - self.data_pages.mark_page_empty(id) + self.data_pages.mark_page_empty(id); } for id in defragmented_pages { self.data_pages.mark_page_full(id) diff --git a/tests/persistence/vacuum.rs b/tests/persistence/vacuum.rs index 9ea77bc..3a9ba92 100644 --- a/tests/persistence/vacuum.rs +++ b/tests/persistence/vacuum.rs @@ -47,6 +47,7 @@ fn test_vacuum_on_persisted_table_survives_reload() { let mut rows = HashMap::new(); let deleted: Vec; + let reused_after_reload_id: u64; { let engine = VacuumPersistPersistenceEngine::new(config.clone()).await.unwrap(); let table = VacuumPersistWorkTable::load(engine).await.unwrap(); @@ -88,7 +89,7 @@ fn test_vacuum_on_persisted_table_survives_reload() { let physical_bytes_after = table.persisted_data_file_size_bytes().await.unwrap(); assert!( physical_bytes_after >= physical_bytes_before, - "persisted vacuum is logical compaction and must not report implicit file truncation" + "online vacuum may append a relocation page; durable reclamation is reuse, not truncation" ); // Insert after vacuum: these operations carry event ids issued @@ -131,6 +132,41 @@ fn test_vacuum_on_persisted_table_survives_reload() { let engine = VacuumPersistPersistenceEngine::new(config.clone()).await.unwrap(); let table = VacuumPersistWorkTable::load(engine).await.unwrap(); + let physical_bytes_before_reuse = table.persisted_data_file_size_bytes().await.unwrap(); + let durable_free_bytes: u64 = table + .0 + .data + .get_empty_links() + .iter() + .map(|link| u64::from(link.length)) + .sum(); + assert!( + durable_free_bytes > 0, + "vacuum-freed ranges must survive reload so later inserts can reuse them" + ); + + // Exercise reuse after reload. Without durable free-page metadata, + // this insert allocates a new page and grows `.wt.data` again. + let reused_row = VacuumPersistRow { + id: table.get_next_pk().into(), + test: 1_100, + another: 1_100, + exchange: "reused-after-reload".to_string(), + }; + let reused_id = reused_row.id; + reused_after_reload_id = reused_id; + table.insert(reused_row.clone()).unwrap(); + rows.insert(reused_id, reused_row); + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence should catch up after durable page reuse") + .expect("persistence engine failed"); + assert_eq!( + table.persisted_data_file_size_bytes().await.unwrap(), + physical_bytes_before_reuse, + "an insert after reload must consume vacuum-freed space instead of extending the file" + ); + assert_eq!(table.select_all().execute().unwrap().len(), rows.len()); for (id, expected) in &rows { assert_eq!(table.select(*id).as_ref(), Some(expected)); @@ -145,5 +181,30 @@ fn test_vacuum_on_persisted_table_survives_reload() { assert_eq!(table.select(*id), None); } } + { + // Reload once more and allocate from the remaining durable range. + // The first reused slot must have been removed from the free + // metadata before its bytes were written, or this insert could + // overwrite it after reopening the table. + let engine = VacuumPersistPersistenceEngine::new(config.clone()).await.unwrap(); + let table = VacuumPersistWorkTable::load(engine).await.unwrap(); + let second_reused_row = VacuumPersistRow { + id: table.get_next_pk().into(), + test: 1_101, + another: 1_101, + exchange: "second-reuse-after-reload".to_string(), + }; + table.insert(second_reused_row).unwrap(); + timeout(Duration::from_secs(30), table.wait_for_ops()) + .await + .expect("persistence should catch up after a second durable page reuse") + .expect("persistence engine failed"); + + assert_eq!( + table.select(reused_after_reload_id).as_ref(), + rows.get(&reused_after_reload_id), + "consumed durable free ranges must not be offered again after another reload" + ); + } }) }