volume: a read plan waits for a vacuum commit instead of answering 404 (#12375)

* volume: a read plan waits for a vacuum commit instead of answering 404 A vacuum commit closes the needle map, swaps the compacted files in and reloads the map. It holds the data-file write lease for that and takes no store lock. needle_read_plan, which the HTTP GET path uses, read the needle map without the data-file read lease, so a GET planned in that window found no map and answered 404 for a live needle. needle_read_plan now takes the read lease first, as read_needle_with_option does, so the plan waits for the commit and then reads the reloaded map. Go's readNeedle does the same under dataFileAccessLock. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: plan a read with the store guard released A plan that waits on a vacuum commit's data-file lease under the store read guard parks store writers (allocate, mount, EC ops) on that commit, and every store reader queued behind them. Look the volume up under the guard and plan after releasing it, as Go's readNeedle does, and look the volume up again if its handle was closed while the plan waited. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: retry a read plan while its handle keeps closing A plan that waited out an unmount looked the volume up only once more, so a replacement handle that also closed before its plan took the lease answered NotFound for a needle a later mount holds. Repeat the lookup while the planned handle is closed, at most three times, so a closed handle that stays in the map still ends the loop. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: start the swap-window check once the plan holds the volume The swap-window test gave the plan 300 ms to answer early, so a reader thread that started late passed the check without waiting on the commit. Share the writer test's wait for the plan's volume handle and start both timed checks after it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: run the swap-window read-plan test on a redb map too A commit reloads the index through load_index_redb for the redb kinds, a path the in-memory run never took during the swap window. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: say a needle read plan is resolved under the data-file lease The plan no longer resolves under the store guard, and resolving it takes the data-file read lease. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: read the needle through the plan in the read-plan tests The tests checked only that the plan resolved and then read the needle through Store::read_volume_needle, so a plan with a wrong offset or size after the commit still passed. Read through the plan, as GET does, and compare the data. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * volume: test the read plan's lookup after an unmount closed the handle No test reached the closed-handle retry, so dropping it left every test green. Take the lookup as a closure so a test can hand the plan the handle an unmount closed and then the remounted volume, and check that the plan reads the needle from the remount. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
2026-11-04 11:07:35 +01:00 · 2026-11-03 14:50:30 -08:00
1 parent 9b5f7bf65b
commit ca4ce2c581
@@ -198,11 +198,8 @@ fn read_needle_for_get(
    Option<(Needle, Option<crate::storage::volume::NeedleStreamInfo>)>,
    crate::storage::volume::VolumeError,
 > {
-    let mut plan = state
-        .store
-        .read()
-        .unwrap()
-        .needle_read_plan(vid, needle_id, read_deleted)?;
+    let mut plan =
+        crate::storage::store::Store::needle_read_plan(&state.store, vid, needle_id, read_deleted)?;
    let blank = || Needle {
        id: needle_id,
        cookie,
@@ -8079,12 +8076,13 @@ mod tests {
            .delete_volume_needle(VolumeId(1), &mut filler)
            .unwrap();
        let data_offset = || {
-            let mut plan = state
-                .store
-                .read()
-                .unwrap()
-                .needle_read_plan(VolumeId(1), NeedleId(ID), false)
-                .unwrap();
+            let mut plan = crate::storage::store::Store::needle_read_plan(
+                &state.store,
+                VolumeId(1),
+                NeedleId(ID),
+                false,
+            )
+            .unwrap();
            let mut n = Needle::default();
            plan.read_meta(&mut n).unwrap();
            plan.into_stream_info(&n).data_file_offset
@@ -948,14 +948,39 @@ impl Store {
        vol.read_needle_opt(n, read_deleted)
    }

-    /// Resolve a needle read under this guard, to run after it is released.
+    /// Resolve a needle read with the store guard held only for the lookup:
+    /// the plan can wait on a vacuum commit's data-file lease, and waiting
+    /// under the guard would park store writers, and every reader queued
+    /// behind them, on that commit. A handle closed while the plan waited is
+    /// looked up again, a bounded number of times.
    pub(crate) fn needle_read_plan(
-        &self,
+        store: &std::sync::RwLock<Store>,
        vid: VolumeId,
        id: NeedleId,
        read_deleted: bool,
    ) -> Result<crate::storage::volume::NeedleReadPlan, VolumeError> {
-        let (_, vol) = self.find_volume(vid).ok_or(VolumeError::NotFound)?;
+        let find = || {
+            let store = store.read().unwrap();
+            store
+                .find_volume(vid)
+                .map(|(_, vol)| vol)
+                .ok_or(VolumeError::NotFound)
+        };
+        Self::plan_on_open_handle(find, id, read_deleted)
+    }
+
+    fn plan_on_open_handle(
+        mut find: impl FnMut() -> Result<Arc<Volume>, VolumeError>,
+        id: NeedleId,
+        read_deleted: bool,
+    ) -> Result<crate::storage::volume::NeedleReadPlan, VolumeError> {
+        let mut vol = find()?;
+        for _ in 0..3 {
+            match vol.needle_read_plan(id, read_deleted) {
+                Err(_) if vol.is_closed() => vol = find()?,
+                plan => return plan,
+            }
+        }
        vol.needle_read_plan(id, read_deleted)
    }

@@ -3385,6 +3410,207 @@ mod tests {
        assert_eq!(v.dat_file_size().unwrap(), volume_size);
    }

+    /// A volume `vid` holding needle 1, compacted and ready to commit, behind
+    /// the store lock the server's handlers use.
+    fn store_ready_to_commit(
+        dir: &str,
+        vid: VolumeId,
+        kind: NeedleMapKind,
+    ) -> Arc<std::sync::RwLock<Store>> {
+        let mut store = make_test_store(&[dir]);
+        store.needle_map_kind = kind;
+        store
+            .add_volume(vid, DiskType::HardDrive, &VolumeSpec::default())
+            .unwrap();
+        write_test_needle(&mut store, vid, 1, b"live needle");
+        store.compact_volume(vid, 0, 0, |_| true).unwrap();
+        Arc::new(std::sync::RwLock::new(store))
+    }
+
+    /// Commits as the vacuum-commit RPC does: the volume is looked up under
+    /// the store guard, which is dropped before the commit takes its locks.
+    /// The result comes back on a channel, so a deadlock fails a test by
+    /// timing out instead of hanging it.
+    fn commit_like_the_rpc(
+        store: Arc<std::sync::RwLock<Store>>,
+        vid: VolumeId,
+    ) -> std::sync::mpsc::Receiver<Result<(), VolumeError>> {
+        let (committed, result) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let (_, vol) = store.read().unwrap().find_volume(vid).unwrap();
+            let _sync_guard = vol.data_file_sync_lock().lock().unwrap();
+            let _lease = vol.instance().write_lock();
+            let _ = committed.send(vol.commit_compact_held());
+        });
+        result
+    }
+
+    /// Holds the next commit of `vid` inside its swap window, where the needle
+    /// map is closed, until the returned sender is used.
+    fn hold_commit_in_swap_window(
+        vid: VolumeId,
+    ) -> (
+        crate::storage::volume::commit_swap_hook::Registration,
+        std::sync::mpsc::Receiver<()>,
+        std::sync::mpsc::Sender<()>,
+    ) {
+        let (entered, in_window) = std::sync::mpsc::channel();
+        let (release, released) = std::sync::mpsc::channel::<()>();
+        let released = std::sync::Mutex::new(released);
+        let entered = std::sync::Mutex::new(entered);
+        let hook = crate::storage::volume::commit_swap_hook::register(vid, move || {
+            entered.lock().unwrap().send(()).unwrap();
+            released.lock().unwrap().recv().unwrap();
+        });
+        (hook, in_window, release)
+    }
+
+    /// Plans and reads needle 1 of `vid` on another thread, and returns once
+    /// the plan holds the volume handle, so a timed check cannot pass on a
+    /// late start.
+    fn plan_in_background(
+        store: &Arc<std::sync::RwLock<Store>>,
+        vid: VolumeId,
+    ) -> std::sync::mpsc::Receiver<Result<Vec<u8>, VolumeError>> {
+        let (_, vol) = store.read().unwrap().find_volume(vid).unwrap();
+        let handles = Arc::strong_count(&vol);
+        let (planned, plan) = std::sync::mpsc::channel();
+        let reader = store.clone();
+        std::thread::spawn(move || {
+            let mut n = Needle {
+                id: NeedleId(1),
+                cookie: Cookie(1),
+                ..Needle::default()
+            };
+            let read = Store::needle_read_plan(&reader, vid, NeedleId(1), false)
+                .and_then(|mut plan| plan.read_full(&mut n))
+                .map(|()| n.data);
+            let _ = planned.send(read);
+        });
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
+        while Arc::strong_count(&vol) == handles {
+            assert!(
+                std::time::Instant::now() < deadline,
+                "the plan never looked up the volume"
+            );
+            std::thread::sleep(std::time::Duration::from_millis(1));
+        }
+        plan
+    }
+
+    /// A GET planned while a vacuum commit has the needle map closed waits for
+    /// the commit, as Go's readNeedle does, instead of answering NotFound for a
+    /// live needle.
+    fn check_read_plan_waits_out_a_commit_swap_window(kind: NeedleMapKind, vid: VolumeId) {
+        let tmp = TempDir::new().unwrap();
+        let store = store_ready_to_commit(tmp.path().to_str().unwrap(), vid, kind);
+        let (_hook, in_window, release) = hold_commit_in_swap_window(vid);
+        let commit = commit_like_the_rpc(store.clone(), vid);
+        in_window
+            .recv_timeout(std::time::Duration::from_secs(10))
+            .expect("the commit never reached its swap window");
+
+        let plan = plan_in_background(&store, vid);
+        if let Ok(early) = plan.recv_timeout(std::time::Duration::from_millis(300)) {
+            release.send(()).unwrap();
+            panic!("planned during the swap window instead of waiting: {early:?}");
+        }
+
+        release.send(()).unwrap();
+        commit
+            .recv_timeout(std::time::Duration::from_secs(10))
+            .expect("the commit never finished")
+            .unwrap();
+        let after = plan
+            .recv_timeout(std::time::Duration::from_secs(10))
+            .expect("the plan still waits after the commit finished");
+        assert_eq!(after.unwrap(), b"live needle");
+    }
+
+    #[test]
+    fn test_read_plan_waits_out_a_commit_swap_window_in_memory() {
+        check_read_plan_waits_out_a_commit_swap_window(NeedleMapKind::InMemory, VolumeId(81));
+    }
+
+    #[test]
+    fn test_read_plan_waits_out_a_commit_swap_window_redb() {
+        check_read_plan_waits_out_a_commit_swap_window(NeedleMapKind::Redb, VolumeId(83));
+    }
+
+    /// A plan on a handle an unmount closed looks the volume up again and
+    /// reads the remounted volume.
+    #[test]
+    fn test_read_plan_retries_a_handle_closed_by_a_remount() {
+        let tmp = TempDir::new().unwrap();
+        let vid = VolumeId(1);
+        let mut store = make_test_store(&[tmp.path().to_str().unwrap()]);
+        store
+            .add_volume(vid, DiskType::HardDrive, &VolumeSpec::default())
+            .unwrap();
+        write_test_needle(&mut store, vid, 1, b"live needle");
+        let (_, stale) = store.find_volume(vid).unwrap();
+        assert!(store.unmount_volume(vid).unwrap());
+        store.mount_volume(vid, "", DiskType::HardDrive).unwrap();
+
+        let mut stale = Some(stale);
+        let mut plan = Store::plan_on_open_handle(
+            || match stale.take() {
+                Some(vol) => Ok(vol),
+                None => store
+                    .find_volume(vid)
+                    .map(|(_, vol)| vol)
+                    .ok_or(VolumeError::NotFound),
+            },
+            NeedleId(1),
+            false,
+        )
+        .unwrap();
+        let mut n = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(1),
+            ..Needle::default()
+        };
+        plan.read_full(&mut n).unwrap();
+        assert_eq!(n.data, b"live needle");
+    }
+
+    /// A plan waiting for a commit holds no store guard, so a store writer
+    /// finishes while the commit is still in its swap window.
+    #[test]
+    fn test_read_plan_waiting_on_a_commit_does_not_block_a_store_writer() {
+        let tmp = TempDir::new().unwrap();
+        let vid = VolumeId(82);
+        let store =
+            store_ready_to_commit(tmp.path().to_str().unwrap(), vid, NeedleMapKind::InMemory);
+        let (_hook, in_window, release) = hold_commit_in_swap_window(vid);
+        let commit = commit_like_the_rpc(store.clone(), vid);
+        in_window
+            .recv_timeout(std::time::Duration::from_secs(10))
+            .expect("the commit never reached its swap window");
+
+        let plan = plan_in_background(&store, vid);
+        let (wrote, written) = std::sync::mpsc::channel();
+        let writer = store.clone();
+        std::thread::spawn(move || {
+            let _guard = writer.write().unwrap();
+            let _ = wrote.send(());
+        });
+        let writer_done = written.recv_timeout(std::time::Duration::from_secs(10));
+        let planned_early = plan.try_recv().is_ok();
+        release.send(()).unwrap();
+        writer_done.expect("a store writer waited on the commit");
+        assert!(!planned_early, "planned during the swap window");
+
+        let committed = commit
+            .recv_timeout(std::time::Duration::from_secs(10))
+            .expect("the commit never finished");
+        assert!(committed.is_ok(), "{committed:?}");
+        let after = plan
+            .recv_timeout(std::time::Duration::from_secs(10))
+            .expect("the plan still waits after the commit finished");
+        assert_eq!(after.unwrap(), b"live needle");
+    }
+
    fn write_test_needle(store: &mut Store, vid: VolumeId, id: u64, data: &[u8]) {
        let mut n = Needle {
            id: NeedleId(id),
@@ -552,6 +552,43 @@ pub(crate) mod needle_read_hook {
    }
 }

+#[cfg(test)]
+pub(crate) mod commit_swap_hook {
+    use super::VolumeId;
+    use std::sync::{Arc, Mutex};
+
+    type Hook = Arc<dyn Fn() + Send + Sync>;
+    static HOOKS: Mutex<Vec<(VolumeId, Hook)>> = Mutex::new(Vec::new());
+
+    pub(crate) struct Registration(VolumeId);
+
+    impl Drop for Registration {
+        fn drop(&mut self) {
+            HOOKS.lock().unwrap().retain(|(id, _)| *id != self.0);
+        }
+    }
+
+    /// Runs inside a compaction commit, after the needle map is closed and
+    /// before the swap. Volume ids are the key, so parallel tests must use
+    /// distinct ones.
+    pub(crate) fn register(id: VolumeId, hook: impl Fn() + Send + Sync + 'static) -> Registration {
+        HOOKS.lock().unwrap().push((id, Arc::new(hook)));
+        Registration(id)
+    }
+
+    pub(crate) fn fire(id: VolumeId) {
+        let hook = HOOKS
+            .lock()
+            .unwrap()
+            .iter()
+            .find(|(hook_id, _)| *hook_id == id)
+            .map(|(_, hook)| hook.clone());
+        if let Some(hook) = hook {
+            hook();
+        }
+    }
+}
+
 #[cfg(test)]
 pub(crate) mod needle_write_hook {
    use super::NeedleId;
@@ -797,9 +834,8 @@ impl IoUnavailable {
    }
 }

-/// A needle read resolved under a store guard and run after it is released;
-/// the handle pins the inode, as for `DatScanPlan`. It takes no data-file
-/// lease: writers wait for one while holding the store write lock.
+/// A needle read resolved under the data-file read lease and run after it is
+/// released; the handle pins the inode, as for `DatScanPlan`.
 pub(crate) struct NeedleReadPlan {
    source: NeedleStreamSource,
    offset: i64,
@@ -2641,12 +2677,17 @@ impl Volume {
        Ok(())
    }

-    /// Resolve a needle read under the caller's store guard, without data I/O.
+    /// Resolve a needle read without data I/O. It takes the data-file read
+    /// lease, as `read_needle_with_option` does, so a plan made while a vacuum
+    /// commit has the needle map closed waits for the commit, as Go's
+    /// readNeedle does, instead of answering NotFound for a live needle.
+    /// Callers must not hold the store guard; see `Store::needle_read_plan`.
    pub(crate) fn needle_read_plan(
        &self,
        id: NeedleId,
        read_deleted: bool,
    ) -> Result<NeedleReadPlan, VolumeError> {
+        let _guard = self.data_file_access_control.read_lock();
        if let Some(e) = self.unavailable_error() {
            return Err(e);
        }
@@ -5558,6 +5599,8 @@ impl Volume {
            io.dat_file = None;
            io.remote_dat_file = None;
        }
+        #[cfg(test)]
+        commit_swap_hook::fire(self.id);

        // makeup_diff has fsynced the .cpd/.cpx contents. Persist a durable .cpc
        // commit marker BEFORE the renames so the two-rename swap is atomic