fix: replace vacuum delay with epoch grace period

benches/cases/simple.rs

@@ -90,6 +90,28 @@ fn delete(c: &mut Criterion) {
     });
 }
 
+fn delete_insert_reuse(c: &mut Criterion) {
+    let rt = Runtime::new().unwrap();
+    let table = Arc::new(SimpleWorkTable::default());
+    let initial = SimpleRow {
+        id: table.get_next_pk().into(),
+        value: fastrand::u64(..),
+    };
+    let initial_pk = table.insert(initial).unwrap();
+    rt.block_on(table.delete(initial_pk)).unwrap();
+
+    c.bench_function("simple_delete_insert_reuse", |b| {
+        b.to_async(&rt).iter(|| async {
+            let row = SimpleRow {
+                id: table.get_next_pk().into(),
+                value: fastrand::u64(..),
+            };
+            let pk = table.insert(black_box(row)).unwrap();
+            black_box(table.delete(pk).await)
+        })
+    });
+}
+
 fn upsert_insert(c: &mut Criterion) {
     let rt = Runtime::new().unwrap();
     let table = Arc::new(SimpleWorkTable::default());
@@ -195,6 +217,7 @@ criterion_group! {
         select_by_pk,
         update,
         delete,
+        delete_insert_reuse,
         upsert_insert,
         upsert_update,
         batch_insert,

src/in_memory/empty_link_registry.rs

@@ -1,4 +1,5 @@
-use std::sync::atomic::{AtomicU32, Ordering};
+use std::ops::Deref;
+use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering};
 
 use data_bucket::Link;
 use data_bucket::page::PageId;
@@ -85,6 +86,38 @@ pub struct EmptyLinkRegistry<const DATA_LENGTH: usize = DATA_INNER_LENGTH> {
 
     pub(crate) op_lock: FairMutex<()>,
     vacuum_lock: tokio::sync::Mutex<()>,
+    reuse_epoch: AtomicU64,
+    active_reusers: [AtomicUsize; 2],
+    reuse_quiescent: tokio::sync::Notify,
+}
+
+/// A free link reserved by an insert operation.
+///
+/// Dropping this lease announces that the link is no longer in flight. Vacuum
+/// advances the reuse epoch and waits for all leases from the preceding epoch
+/// before it starts moving or resetting pages.
+#[derive(Debug)]
+#[must_use = "the reservation must live until the free-link write finishes"]
+pub(crate) struct EmptyLinkReservation<'a, const DATA_LENGTH: usize = DATA_INNER_LENGTH> {
+    registry: &'a EmptyLinkRegistry<DATA_LENGTH>,
+    link: Link,
+    epoch: usize,
+}
+
+impl<const DATA_LENGTH: usize> Deref for EmptyLinkReservation<'_, DATA_LENGTH> {
+    type Target = Link;
+
+    fn deref(&self) -> &Self::Target {
+        &self.link
+    }
+}
+
+impl<const DATA_LENGTH: usize> Drop for EmptyLinkReservation<'_, DATA_LENGTH> {
+    fn drop(&mut self) {
+        if self.registry.active_reusers[self.epoch].fetch_sub(1, Ordering::AcqRel) == 1 {
+            self.registry.reuse_quiescent.notify_one();
+        }
+    }
 }
 
 impl<const DATA_LENGTH: usize> Default for EmptyLinkRegistry<DATA_LENGTH> {
@@ -96,6 +129,9 @@ impl<const DATA_LENGTH: usize> Default for EmptyLinkRegistry<DATA_LENGTH> {
             sum_links_len: Default::default(),
             op_lock: Default::default(),
             vacuum_lock: Default::default(),
+            reuse_epoch: AtomicU64::new(0),
+            active_reusers: [AtomicUsize::new(0), AtomicUsize::new(0)],
+            reuse_quiescent: tokio::sync::Notify::new(),
         }
     }
 }
@@ -164,20 +200,36 @@ impl<const DATA_LENGTH: usize> EmptyLinkRegistry<DATA_LENGTH> {
         self.insert_link(index_ord_link);
     }
 
+    /// Removes and returns the largest free link.
     pub fn pop_max(&self) -> Option<Link> {
-        if self.vacuum_lock.try_lock().is_err() {
-            return None;
-        }
+        self.reserve_max().map(|reservation| *reservation)
+    }
+
+    /// Reserves the largest free link for internal page reuse.
+    ///
+    /// The returned lease must remain alive until the caller has published the
+    /// row written into the link. Vacuum admission is closed while the lease is
+    /// registered, so a concurrent epoch advance cannot miss it.
+    pub(crate) fn reserve_max(&self) -> Option<EmptyLinkReservation<'_, DATA_LENGTH>> {
+        let _vacuum_admission = self.vacuum_lock.try_lock().ok()?;
 
         let _g = self.op_lock.lock();
 
         let mut iter = self.length_ord_links.iter().rev();
         let (_, max_length_link) = iter.next()?;
+        let max_length_link = *max_length_link;
         drop(iter);
 
-        self.remove_link(*max_length_link);
+        self.remove_link(max_length_link);
 
-        Some(*max_length_link)
+        let epoch = self.reuse_epoch.load(Ordering::Acquire) as usize & 1;
+        self.active_reusers[epoch].fetch_add(1, Ordering::AcqRel);
+
+        Some(EmptyLinkReservation {
+            registry: self,
+            link: max_length_link,
+            epoch,
+        })
     }
 
     pub fn iter(&self) -> impl Iterator<Item = Link> + '_ {
@@ -188,8 +240,17 @@ impl<const DATA_LENGTH: usize> EmptyLinkRegistry<DATA_LENGTH> {
         self.sum_links_len.load(Ordering::Acquire)
     }
 
+    /// Closes free-link reuse admission, advances the epoch, and waits for all
+    /// reservations admitted in the previous epoch to finish.
     pub async fn lock_vacuum(&self) -> tokio::sync::MutexGuard<'_, ()> {
-        self.vacuum_lock.lock().await
+        let guard = self.vacuum_lock.lock().await;
+        let previous_epoch = self.reuse_epoch.fetch_add(1, Ordering::AcqRel) as usize & 1;
+
+        while self.active_reusers[previous_epoch].load(Ordering::Acquire) != 0 {
+            self.reuse_quiescent.notified().await;
+        }
+
+        guard
     }
 }
 
@@ -422,7 +483,7 @@ mod tests {
     fn test_empty_registry() {
         let registry = EmptyLinkRegistry::<DATA_INNER_LENGTH>::default();
 
-        assert_eq!(registry.pop_max(), None);
+        assert!(registry.pop_max().is_none());
         assert_eq!(registry.iter().count(), 0);
     }
 
@@ -489,4 +550,30 @@ mod tests {
         );
         assert_eq!(popped_after_unlock.unwrap().length, 100);
     }
+
+    #[tokio::test]
+    async fn test_lock_vacuum_waits_for_previous_epoch_reuser() {
+        let registry = EmptyLinkRegistry::<DATA_INNER_LENGTH>::default();
+        registry.push(Link {
+            page_id: 1.into(),
+            offset: 0,
+            length: 100,
+        });
+
+        let reservation = registry.reserve_max().unwrap();
+        let vacuum = registry.lock_vacuum();
+        tokio::pin!(vacuum);
+
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(50), &mut vacuum)
+                .await
+                .is_err(),
+            "vacuum passed the grace period while a prior-epoch link was in flight"
+        );
+
+        drop(reservation);
+        let _guard = tokio::time::timeout(std::time::Duration::from_secs(1), vacuum)
+            .await
+            .expect("vacuum should resume when the prior epoch becomes quiescent");
+    }
 }

src/in_memory/pages.rs

@@ -17,7 +17,7 @@ use std::{
     sync::atomic::{AtomicU32, AtomicU64, Ordering},
 };
 
-use crate::in_memory::empty_link_registry::EmptyLinkRegistry;
+use crate::in_memory::empty_link_registry::{EmptyLinkRegistry, EmptyLinkReservation};
 use crate::prelude::ArchivedRowWrapper;
 use crate::{
     in_memory::{
@@ -31,6 +31,17 @@ fn page_id_mapper(page_id: usize) -> usize {
     page_id - 1usize
 }
 
+pub(crate) struct InsertedRow<'a, const DATA_LENGTH: usize> {
+    link: Link,
+    _reuse_reservation: Option<EmptyLinkReservation<'a, DATA_LENGTH>>,
+}
+
+impl<const DATA_LENGTH: usize> InsertedRow<'_, DATA_LENGTH> {
+    pub(crate) fn link(&self) -> Link {
+        self.link
+    }
+}
+
 #[derive(Debug)]
 pub struct DataPages<Row, const DATA_LENGTH: usize = DATA_INNER_LENGTH>
 where
@@ -96,14 +107,24 @@ where
     }
 
     pub fn insert(&self, row: Row) -> Result<Link, ExecutionError>
+    where
+        Row: Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
+        <Row as StorableRow>::WrappedRow:
+            Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
+    {
+        Ok(self.insert_with_reservation(row)?.link())
+    }
+
+    pub(crate) fn insert_with_reservation(&self, row: Row) -> Result<InsertedRow<'_, DATA_LENGTH>, ExecutionError>
     where
         Row: Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
     {
         let general_row = <Row as StorableRow>::WrappedRow::from_inner(row);
 
-        if let Some(link) = self.empty_links.pop_max() {
+        if let Some(reservation) = self.empty_links.reserve_max() {
+            let link = *reservation;
             let pages = self.pages.read();
             let current_page: usize = page_id_mapper(link.page_id.into());
             let page = &pages[current_page];
@@ -113,7 +134,10 @@ where
                     if let Some(l) = left_link {
                         self.empty_links.push(l);
                     }
-                    return Ok(link);
+                    return Ok(InsertedRow {
+                        link,
+                        _reuse_reservation: Some(reservation),
+                    });
                 }
                 Err(e) => match e {
                     DataExecutionError::InvalidLink => {
@@ -138,7 +162,10 @@ where
             match link {
                 Ok(link) => {
                     self.row_count.fetch_add(1, Ordering::Relaxed);
-                    return Ok(link);
+                    return Ok(InsertedRow {
+                        link,
+                        _reuse_reservation: None,
+                    });
                 }
                 Err(e) => match e {
                     DataExecutionError::PageIsFull { .. } => {
@@ -170,12 +197,27 @@ where
         <Row as StorableRow>::WrappedRow:
             Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
     {
-        let link = self.insert(row.clone())?;
+        let (inserted, bytes) = self.insert_cdc_with_reservation(row)?;
+        Ok((inserted.link(), bytes))
+    }
+
+    pub(crate) fn insert_cdc_with_reservation(
+        &self,
+        row: Row,
+    ) -> Result<(InsertedRow<'_, DATA_LENGTH>, Vec<u8>), ExecutionError>
+    where
+        Row: Archive
+            + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>
+            + Clone,
+        <Row as StorableRow>::WrappedRow:
+            Archive + for<'a> Serialize<Strategy<Serializer<AlignedVec, ArenaHandle<'a>, Share>, rkyv::rancor::Error>>,
+    {
+        let inserted = self.insert_with_reservation(row.clone())?;
         let general_row = <Row as StorableRow>::WrappedRow::from_inner(row);
         let bytes = rkyv::to_bytes(&general_row)
             .expect("should be ok as insert not failed")
             .into_vec();
-        Ok((link, bytes))
+        Ok((inserted, bytes))
     }
 
     fn add_next_page(&self, tried_page: usize) {
@@ -728,6 +770,30 @@ mod tests {
         assert_eq!(pages.select(link).unwrap(), TestRow { a: 10, b: 20 })
     }
 
+    #[tokio::test]
+    async fn reused_link_reservation_lives_through_caller_publication() {
+        let pages = DataPages::<TestRow>::new();
+        let old_link = pages.insert(TestRow { a: 10, b: 20 }).unwrap();
+        pages.delete(old_link).unwrap();
+
+        let inserted = pages.insert_with_reservation(TestRow { a: 30, b: 40 }).unwrap();
+        assert_eq!(inserted.link(), old_link);
+
+        let vacuum = pages.empty_links.lock_vacuum();
+        tokio::pin!(vacuum);
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_millis(50), &mut vacuum)
+                .await
+                .is_err(),
+            "vacuum passed the grace period before the caller published the reused link"
+        );
+
+        drop(inserted);
+        let _guard = tokio::time::timeout(std::time::Duration::from_secs(1), vacuum)
+            .await
+            .expect("vacuum should resume after publication releases the reused link");
+    }
+
     //#[test]
     fn _bench() {
         let pages = Arc::new(DataPages::<TestRow>::new());