fix(rekey): a reader leaving mid-deferral must not strand the retirement

CHANGELOG.md

@@ -16,7 +16,7 @@ The first release. Pre-releases were published as `0.1.0-beta.N`; the entries be
 - **Cross-platform VFS** — Linux (`io_uring`), Windows (IOCP), macOS/iOS (Grand Central Dispatch), Android, WASM/OPFS and WASI backends, plus a tokio thread-pool fallback and an in-memory backend, with format-bit identity across targets. On Linux the backend is chosen at run time: a kernel that refuses an `io_uring` ring falls back to the thread pool with a warning instead of failing the open. All native backends share one advisory-lock implementation, so processes on different backends still exclude each other on one store.
 - **Snapshots** — `snapshot_to`, `restore_from`, and incremental apply, each authenticated against the state its manifest describes. Destinations must be empty; malformed or incomplete artifacts fail closed.
 - **Recovery** — open-flow GC, apply-journal replay, deep-walk `fsck`, and the `pagedb-fsck` binary.
-- **Online rekey** — rekey under a new key with mixed-cipher and mixed-epoch page coexistence; no full-file migration.
+- **Online rekey** — rekey under a new key with mixed-cipher and mixed-epoch page coexistence; no full-file migration. A rotation whose source epoch is still pinned by a reader defers retiring it and completes the retirement once the reader set drains — including when the last reader leaves during the deferral itself, so the superseded master key never stays leasable behind an `Ok(())`. A retirement that cannot be taken yet stays queued for the next attempt without holding up the others.
 - **Handle modes** — `Standalone`, `Follower`, `ReadOnly`, and `Observer`.
 - **Open refusals name the parameter, not the store** — `KeyMismatch`, `PageSizeMismatch`, and `RealmMismatch`, each decided before anything is read or written, and none reported as corruption.
 - **Failures report themselves** — an unreadable free-list chain, main file, or segment catalog fails `stats()` instead of reporting zero; compaction never skips a catalog entry whose file it cannot open; segment open distinguishes a missing file from a permission or backend error; and only genuine contention is reported as contention. Persisted named-counter rows are validated at open, and commit-history keys are rejected unless exactly eight bytes.

src/txn/db/core.rs

@@ -244,8 +244,17 @@ pub struct Db<V: Vfs + Clone> {
     pub(crate) visibility_test_hook: parking_lot::Mutex<Option<Arc<VisibilityTestHook>>>,
     #[cfg(test)]
     pub(crate) rekey_test_fault: parking_lot::Mutex<Option<RekeyTestFault>>,
+    #[cfg(test)]
+    pub(crate) retirement_defer_hook: parking_lot::Mutex<Option<RetirementDeferHook<V>>>,
 }
 
+/// Runs inside `retire_rekey_source_when_safe`'s critical section, once it has
+/// decided to defer and before the obligation is recorded. That instant is the
+/// only place a test can stand to prove the two are indivisible, because the
+/// failure it guards against needs the last reader to leave precisely there.
+#[cfg(test)]
+pub(crate) type RetirementDeferHook<V> = Arc<dyn Fn(&Db<V>) + Send + Sync>;
+
 /// Reader-visible state, refreshed by the writer at commit time.
 #[derive(Debug, Clone, Copy)]
 #[allow(clippy::struct_field_names)]
@@ -329,33 +338,75 @@ impl<V: Vfs + Clone> Db<V> {
     /// Retire an obsolete source epoch immediately when no reader can still
     /// resolve a pre-cutover snapshot; otherwise defer retirement until the
     /// tracked reader set drains.
+    ///
+    /// Deciding to defer and recording what was deferred happen under one
+    /// lock, held across both. They are one step, not two: the reader set is
+    /// only a reason to defer for as long as the obligation is not yet visible
+    /// to the drain, and the last reader can leave at any instant. Split them
+    /// and that reader's drain reads an obligation list that is still empty,
+    /// then this pushes into a list nothing will visit again — the rotation
+    /// reports success while the superseded key stays leasable for the life of
+    /// the handle. `drain_pending_key_retirements` takes the same two locks in
+    /// the same order.
+    ///
+    /// The immediate path records the obligation too, then retires the whole
+    /// list: it is the only place that sees an empty reader set together with
+    /// the backlog, and the drain that would otherwise retry a failed entry
+    /// only runs when the last reader leaves — which by then has happened.
     pub(crate) fn retire_rekey_source_when_safe(
         &self,
         epoch: u64,
         cipher_id: CipherId,
     ) -> Result<()> {
-        if self.tracked_readers.lock().is_empty() {
-            return self.pager.retire_mk_epoch(epoch, cipher_id);
+        let mut retirements = self.pending_key_retirements.lock();
+        let readers_present = !self.tracked_readers.lock().is_empty();
+        if readers_present {
+            #[cfg(test)]
+            self.await_retirement_defer_hook();
         }
         let pending = PendingKeyRetirement { epoch, cipher_id };
-        let mut retirements = self.pending_key_retirements.lock();
         if !retirements.contains(&pending) {
             retirements.push(pending);
         }
-        Ok(())
+        if readers_present {
+            return Ok(());
+        }
+        self.retire_recorded(&mut retirements)
     }
 
     /// Drain deferred source-epoch retirements once no tracked reader remains.
     pub(crate) fn drain_pending_key_retirements(&self) -> Result<()> {
+        let mut retirements = self.pending_key_retirements.lock();
         if !self.tracked_readers.lock().is_empty() {
             return Ok(());
         }
-        let pending = std::mem::take(&mut *self.pending_key_retirements.lock());
-        for retirement in pending {
-            self.pager
-                .retire_mk_epoch(retirement.epoch, retirement.cipher_id)?;
+        self.retire_recorded(&mut retirements)
+    }
+
+    /// Retire every recorded obligation, keeping the ones that fail.
+    ///
+    /// An entry leaves the list only once its retirement has succeeded, so a
+    /// failure keeps that obligation queued for the next attempt — and keeps
+    /// only that one: the rest are independent epochs, and stopping at the
+    /// first error would let one unretirable epoch stand in front of them
+    /// forever. The first error is reported once the whole pass is done.
+    fn retire_recorded(&self, retirements: &mut Vec<PendingKeyRetirement>) -> Result<()> {
+        let mut first_error = None;
+        retirements.retain(|pending| {
+            match self.pager.retire_mk_epoch(pending.epoch, pending.cipher_id) {
+                Ok(()) => false,
+                Err(error) => {
+                    if first_error.is_none() {
+                        first_error = Some(error);
+                    }
+                    true
+                }
+            }
+        });
+        match first_error {
+            Some(error) => Err(error),
+            None => Ok(()),
         }
-        Ok(())
     }
 
     pub(crate) fn ensure_usable(&self) -> Result<()> {
@@ -425,6 +476,27 @@ impl<V: Vfs + Clone> Db<V> {
         *self.visibility_test_hook.lock() = Some(hook);
     }
 
+    #[cfg(test)]
+    pub(crate) fn install_retirement_defer_hook(&self, hook: RetirementDeferHook<V>) {
+        *self.retirement_defer_hook.lock() = Some(hook);
+    }
+
+    /// Detach the deferral hook, for the same reason as
+    /// [`Self::clear_visibility_test_hook`]: it is a rendezvous, and a later
+    /// deferral would park on a counterpart that is no longer coming.
+    #[cfg(test)]
+    pub(crate) fn clear_retirement_defer_hook(&self) {
+        *self.retirement_defer_hook.lock() = None;
+    }
+
+    #[cfg(test)]
+    fn await_retirement_defer_hook(&self) {
+        let hook = self.retirement_defer_hook.lock().clone();
+        if let Some(hook) = hook {
+            hook(self);
+        }
+    }
+
     /// Detach the hook so the rest of a test runs on unrehearsed paths. Every
     /// pause point is a rendezvous that only completes when its counterpart
     /// notifies, so a hook left installed past the interleaving it was staging

src/txn/db/open/create.rs

@@ -134,6 +134,8 @@ impl<V: Vfs + Clone> Db<V> {
             #[cfg(test)]
             visibility_test_hook: parking_lot::Mutex::new(None),
             #[cfg(test)]
+            retirement_defer_hook: parking_lot::Mutex::new(None),
+            #[cfg(test)]
             rekey_test_fault: parking_lot::Mutex::new(None),
         })
     }

src/txn/db/open/existing.rs

@@ -325,6 +325,8 @@ impl<V: Vfs + Clone> Db<V> {
             #[cfg(test)]
             visibility_test_hook: parking_lot::Mutex::new(None),
             #[cfg(test)]
+            retirement_defer_hook: parking_lot::Mutex::new(None),
+            #[cfg(test)]
             rekey_test_fault: parking_lot::Mutex::new(None),
         };
 

src/txn/db/rekey/main.rs

@@ -944,6 +944,167 @@ mod tests {
         );
     }
 
+    /// The last reader may leave *during* the deferral, and the retirement must
+    /// still complete.
+    ///
+    /// Deferring is two facts that have to agree: a reader exists, and an
+    /// obligation is recorded where the drain will find it. Nothing pins the
+    /// first fact in place — the reader's `drop` is a plain unregister that can
+    /// land at any instant, including between the two. When it did, the drain
+    /// read an empty obligation list and the record arrived after the only
+    /// thing that ever reads it had gone: `rekey_db` returned `Ok(())` while the
+    /// superseded master key stayed leasable for the life of the handle, with no
+    /// error, no log line, and nothing to query. This drives the reader out at
+    /// exactly that instant and requires the epoch to retire anyway.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn a_reader_leaving_mid_deferral_still_retires_the_source_epoch() {
+        let db = Db::open_internal(MemVfs::new(), SOURCE_KEK, PAGE, REALM)
+            .await
+            .unwrap();
+        // Move the active epoch off 0 so retiring it is a real retirement
+        // rather than the active-epoch refusal.
+        let target_mk = derive_mk(&TARGET_KEK, &db.kek_salt, 1).unwrap();
+        db.pager.set_active_mk_epoch(target_mk, 1);
+
+        // Left to race, the reader can be gone before the critical section is
+        // even entered, the deferral is never taken, and every assertion below
+        // passes against the unfixed code. So the hook drives the reader out
+        // itself, from the one instant the bug needed.
+        let departure = std::sync::Arc::new(std::sync::Barrier::new(2));
+        let deferrals = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let hook_departure = departure.clone();
+        let hook_deferrals = deferrals.clone();
+        db.install_retirement_defer_hook(std::sync::Arc::new(move |db: &Db<MemVfs>| {
+            hook_deferrals.fetch_add(1, Ordering::Relaxed);
+            hook_departure.wait();
+            // `unregister_read` releases `tracked_readers` before it touches
+            // the retirement list held here, so this cannot deadlock.
+            for _ in 0..5_000 {
+                if db.tracked_readers.lock().is_empty() {
+                    return;
+                }
+                std::thread::sleep(std::time::Duration::from_millis(1));
+            }
+            panic!("the reader never left; the window under test was not reached");
+        }));
+
+        let read = db.begin_read().await.unwrap();
+        assert!(
+            db.pager.mk_for(0, db.cipher_id).is_ok(),
+            "the source epoch must still be leasable while a reader holds it"
+        );
+
+        std::thread::scope(|scope| {
+            scope.spawn(|| {
+                departure.wait();
+                drop(read);
+            });
+            db.retire_rekey_source_when_safe(0, db.cipher_id)
+        })
+        .unwrap();
+        db.clear_retirement_defer_hook();
+
+        assert_eq!(
+            deferrals.load(Ordering::Relaxed),
+            1,
+            "the retirement must have been deferred; otherwise the reader left \
+             before the window and the race was never staged"
+        );
+        assert!(
+            db.tracked_readers.lock().is_empty(),
+            "the reader must be gone before the retirement is judged"
+        );
+        assert!(
+            db.pending_key_retirements.lock().is_empty(),
+            "an obligation no reader is waiting on must not survive the deferral"
+        );
+        assert!(
+            matches!(
+                db.pager.mk_for(0, db.cipher_id),
+                Err(PagedbError::MissingPersistedKey { mk_epoch: 0, .. })
+            ),
+            "a reader leaving mid-deferral must not leave the source epoch leasable"
+        );
+    }
+
+    /// Queue two obligations, then make the first one refuse to retire.
+    /// Retirements are independent epochs, so the refusal must be kept and
+    /// stepped over, not treated as a wall.
+    #[tokio::test(flavor = "current_thread")]
+    async fn a_refused_retirement_does_not_hold_up_the_obligations_behind_it() {
+        let db = Db::open_internal(MemVfs::new(), SOURCE_KEK, PAGE, REALM)
+            .await
+            .unwrap();
+        let target_mk = derive_mk(&TARGET_KEK, &db.kek_salt, 1).unwrap();
+        db.pager.set_active_mk_epoch(target_mk, 1);
+        let spare_mk = derive_mk(&TARGET_KEK, &db.kek_salt, 2).unwrap();
+        db.pager.install_mk_epoch(spare_mk, 2, db.cipher_id);
+
+        let read = db.begin_read().await.unwrap();
+        db.retire_rekey_source_when_safe(0, db.cipher_id).unwrap();
+        db.retire_rekey_source_when_safe(2, db.cipher_id).unwrap();
+
+        // Epoch 0 is the active one by the time the drain runs, which is the
+        // one condition `retire_mk_epoch` refuses.
+        let source_mk = derive_mk(&SOURCE_KEK, &db.kek_salt, 0).unwrap();
+        db.pager.set_active_mk_epoch(source_mk, 0);
+        drop(read);
+
+        assert!(
+            db.pager.mk_for(2, db.cipher_id).is_err(),
+            "an epoch queued behind a refusal must still be retired"
+        );
+        let queued: Vec<u64> = db
+            .pending_key_retirements
+            .lock()
+            .iter()
+            .map(|pending| pending.epoch)
+            .collect();
+        assert_eq!(
+            queued,
+            vec![0],
+            "the refused epoch — and only it — must stay queued for the next attempt"
+        );
+    }
+
+    /// A drain runs only when the last reader leaves, so an obligation it
+    /// failed to clear has no second drain coming while the reader set stays
+    /// empty. The next rotation is the only thing left that can retry it.
+    #[tokio::test(flavor = "current_thread")]
+    async fn a_rotation_with_no_readers_retires_the_backlog_it_inherits() {
+        let db = Db::open_internal(MemVfs::new(), SOURCE_KEK, PAGE, REALM)
+            .await
+            .unwrap();
+        let target_mk = derive_mk(&TARGET_KEK, &db.kek_salt, 1).unwrap();
+        db.pager.set_active_mk_epoch(target_mk, 1);
+
+        let read = db.begin_read().await.unwrap();
+        db.retire_rekey_source_when_safe(0, db.cipher_id).unwrap();
+        let source_mk = derive_mk(&SOURCE_KEK, &db.kek_salt, 0).unwrap();
+        db.pager.set_active_mk_epoch(source_mk, 0);
+        drop(read);
+        assert_eq!(
+            db.pending_key_retirements.lock().len(),
+            1,
+            "the refused epoch must have been kept"
+        );
+
+        let target_mk = derive_mk(&TARGET_KEK, &db.kek_salt, 1).unwrap();
+        db.pager.set_active_mk_epoch(target_mk, 1);
+        let spare_mk = derive_mk(&TARGET_KEK, &db.kek_salt, 2).unwrap();
+        db.pager.install_mk_epoch(spare_mk, 2, db.cipher_id);
+        db.retire_rekey_source_when_safe(2, db.cipher_id).unwrap();
+
+        assert!(
+            db.pending_key_retirements.lock().is_empty(),
+            "a rotation that finds no reader must clear the backlog, not just its own epoch"
+        );
+        assert!(
+            db.pager.mk_for(0, db.cipher_id).is_err(),
+            "the inherited obligation must actually have been retired"
+        );
+    }
+
     #[tokio::test(flavor = "current_thread")]
     async fn direct_segment_reader_keeps_source_lease_after_retirement() {
         let db = Db::open_internal(MemVfs::new(), SOURCE_KEK, PAGE, REALM)