fix(retention): queue writer admission after bounded deferrals

crates/tracedecay-store-runtime/src/writer_gate.rs

@@ -15,9 +15,10 @@
 //! # The lock hierarchy
 //!
 //! ```text
-//! daemon: RwLock          Daemon scope = write, Store scope = read
-//!   ├── owner:  Mutex
-//!   └── content: Mutex
+//! daemon: RwLock       Daemon scope = write, store scopes = read
+//!   └── store: RwLock  Maintenance = write, owner/content = read
+//!       ├── owner: Mutex
+//!       └── content: Mutex
 //! ```
 //!
 //! Every acquisition takes these in exactly that order, so no lock-order
@@ -66,6 +67,8 @@ pub enum WriterScope {
         data_root: PathBuf,
         class: StoreWriterClass,
     },
+    /// Fair exclusive maintenance for one store, without blocking unrelated stores.
+    StoreMaintenance { data_root: PathBuf },
 }
 
 impl WriterScope {
@@ -79,13 +82,23 @@ impl WriterScope {
             class,
         }
     }
+
+    /// Same canonical store identity as [`Self::store`], with exclusive admission
+    /// only against this store's owner and content operations.
+    #[must_use]
+    pub fn maintenance(data_root: impl Into<PathBuf>) -> Self {
+        Self::StoreMaintenance {
+            data_root: data_root.into(),
+        }
+    }
 }
 
 /// The logical lanes of one store family. Held by `Arc` from the registry and by
 /// every live guard, so a gate outlives any holder even if the registry prunes
 /// its weak entry.
 #[derive(Default)]
 struct StoreGate {
+    maintenance: Arc<RwLock<()>>,
     owner: Arc<Mutex<()>>,
     content: Arc<Mutex<()>>,
 }
@@ -105,6 +118,7 @@ impl StoreGate {
 /// exist to pin the guards.
 pub struct WriterAdmissionGuard {
     _class: Option<OwnedMutexGuard<()>>,
+    _maintenance: Option<StoreMaintenanceGuard>,
     _daemon: DaemonGuard,
     /// Keeps the store's gate alive for as long as it is held, so a registry
     /// prune between acquisition and release cannot hand a second acquirer a
@@ -120,6 +134,13 @@ enum DaemonGuard {
     Exclusive(OwnedRwLockWriteGuard<()>),
 }
 
+/// Store-local readers and the queued exclusive retention guard.
+#[allow(dead_code)]
+enum StoreMaintenanceGuard {
+    Shared(OwnedRwLockReadGuard<()>),
+    Exclusive(OwnedRwLockWriteGuard<()>),
+}
+
 /// Writer-gate registry: one process-wide lane plus one lane set per store family.
 pub struct StoreWriterGates {
     daemon: Arc<RwLock<()>>,
@@ -173,6 +194,7 @@ impl StoreWriterGates {
         match scope {
             WriterScope::Daemon => WriterAdmissionGuard {
                 _class: None,
+                _maintenance: None,
                 _daemon: DaemonGuard::Exclusive(
                     tracing::Instrument::instrument(
                         Arc::clone(&self.daemon).write_owned(),
@@ -189,13 +211,38 @@ impl StoreWriterGates {
                 )
                 .await;
                 let gate = self.store_gate(data_root);
+                let maintenance = tracing::Instrument::instrument(
+                    Arc::clone(&gate.maintenance).read_owned(),
+                    tracing::trace_span!("daemon.writer_gate.acquire.store_read"),
+                )
+                .await;
                 let class_guard = tracing::Instrument::instrument(
                     gate.class_mutex(*class).lock_owned(),
                     tracing::trace_span!("daemon.writer_gate.acquire.class"),
                 )
                 .await;
                 WriterAdmissionGuard {
                     _class: Some(class_guard),
+                    _maintenance: Some(StoreMaintenanceGuard::Shared(maintenance)),
+                    _daemon: DaemonGuard::Shared(daemon),
+                    _gate: Some(gate),
+                }
+            }
+            WriterScope::StoreMaintenance { data_root } => {
+                let daemon = tracing::Instrument::instrument(
+                    Arc::clone(&self.daemon).read_owned(),
+                    tracing::trace_span!("daemon.writer_gate.acquire.store"),
+                )
+                .await;
+                let gate = self.store_gate(data_root);
+                let maintenance = tracing::Instrument::instrument(
+                    Arc::clone(&gate.maintenance).write_owned(),
+                    tracing::trace_span!("daemon.writer_gate.acquire.store_maintenance"),
+                )
+                .await;
+                WriterAdmissionGuard {
+                    _class: None,
+                    _maintenance: Some(StoreMaintenanceGuard::Exclusive(maintenance)),
                     _daemon: DaemonGuard::Shared(daemon),
                     _gate: Some(gate),
                 }
@@ -209,15 +256,29 @@ impl StoreWriterGates {
         match scope {
             WriterScope::Daemon => Some(WriterAdmissionGuard {
                 _class: None,
+                _maintenance: None,
                 _daemon: DaemonGuard::Exclusive(Arc::clone(&self.daemon).try_write_owned().ok()?),
                 _gate: None,
             }),
             WriterScope::Store { data_root, class } => {
                 let daemon = Arc::clone(&self.daemon).try_read_owned().ok()?;
                 let gate = self.store_gate(data_root);
+                let maintenance = Arc::clone(&gate.maintenance).try_read_owned().ok()?;
                 let class_guard = gate.class_mutex(*class).try_lock_owned().ok()?;
                 Some(WriterAdmissionGuard {
                     _class: Some(class_guard),
+                    _maintenance: Some(StoreMaintenanceGuard::Shared(maintenance)),
+                    _daemon: DaemonGuard::Shared(daemon),
+                    _gate: Some(gate),
+                })
+            }
+            WriterScope::StoreMaintenance { data_root } => {
+                let daemon = Arc::clone(&self.daemon).try_read_owned().ok()?;
+                let gate = self.store_gate(data_root);
+                let maintenance = Arc::clone(&gate.maintenance).try_write_owned().ok()?;
+                Some(WriterAdmissionGuard {
+                    _class: None,
+                    _maintenance: Some(StoreMaintenanceGuard::Exclusive(maintenance)),
                     _daemon: DaemonGuard::Shared(daemon),
                     _gate: Some(gate),
                 })
@@ -228,6 +289,8 @@ impl StoreWriterGates {
 
 #[cfg(test)]
 mod tests {
+    use std::future::{Future, poll_fn};
+    use std::task::Poll;
     use std::time::Duration;
 
     use super::*;
@@ -322,6 +385,55 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn queued_store_maintenance_preserves_other_project_admission() {
+        let gates = StoreWriterGates::default();
+        let held = gates.acquire(&store("/a", StoreWriterClass::Content)).await;
+        let maintenance_scope = WriterScope::maintenance("/a");
+        let queued = gates.acquire(&maintenance_scope);
+        tokio::pin!(queued);
+        poll_fn(|cx| {
+            assert!(queued.as_mut().poll(cx).is_pending());
+            Poll::Ready(())
+        })
+        .await;
+
+        let following_scope = store("/a", StoreWriterClass::Owner);
+        let following = gates.acquire(&following_scope);
+        tokio::pin!(following);
+        poll_fn(|cx| {
+            assert!(
+                following.as_mut().poll(cx).is_pending(),
+                "same-store operations wait behind queued maintenance"
+            );
+            Poll::Ready(())
+        })
+        .await;
+
+        assert!(
+            gates
+                .try_acquire(&store("/b", StoreWriterClass::Owner))
+                .is_some(),
+            "an unrelated project can open while maintenance queues behind a long sync"
+        );
+        drop(held);
+        let admitted = tokio::time::timeout(Duration::from_secs(5), &mut queued)
+            .await
+            .expect("queued maintenance acquires when its own store is free");
+        assert!(
+            gates
+                .try_acquire(&store("/a", StoreWriterClass::Content))
+                .is_none(),
+            "admitted maintenance excludes writes to its own store"
+        );
+        drop(admitted);
+        let follower = tokio::time::timeout(Duration::from_secs(5), &mut following)
+            .await
+            .expect("same-store owner resumes after maintenance");
+        drop(follower);
+        assert!(gates.try_acquire(&maintenance_scope).is_some());
+    }
+
     #[tokio::test]
     async fn retired_store_gates_are_reclaimed() {
         let gates = StoreWriterGates::default();

crates/tracedecay/src/daemon/branch_admin.rs

@@ -125,6 +125,16 @@ pub(super) fn store_writer_scope(data_root: &Path, class: StoreWriterClass) -> W
     }
 }
 
+/// Resolves maintenance to the same canonical store gate as logical writers.
+/// A missing or unresolvable store refuses isolation by falling back to daemon
+/// administration rather than silently splitting the writer identity.
+pub(super) fn store_maintenance_writer_scope(data_root: &Path) -> WriterScope {
+    match authority::canonical_identity_path(data_root) {
+        Ok(canonical) => WriterScope::maintenance(canonical),
+        Err(_) => WriterScope::Daemon,
+    }
+}
+
 /// Owner-lane scope for one registered database owner.
 ///
 /// [`StoreOwnerKey::store_root`](tracedecay_session_runtime::StoreOwnerKey) is already the
@@ -1923,6 +1933,30 @@ impl StoreAdministration {
         operation().await
     }
 
+    /// Cancellation applies only while queued for writer admission.
+    /// Once the guard is acquired, the operation must finish cooperatively:
+    /// dropping an in-flight journal/retention future would abandon physical
+    /// blocking work whose client survives the daemon's shutdown join.
+    pub(super) async fn with_writer_cancellable_acquisition<Operation, OperationFuture, Output>(
+        &self,
+        cancellation: &tracedecay_runtime_core::cancellation::CancellationToken,
+        scope: WriterScope,
+        operation: Operation,
+    ) -> Option<Output>
+    where
+        Operation: FnOnce() -> OperationFuture,
+        OperationFuture: Future<Output = Output>,
+    {
+        let _writer = tokio::select! {
+            biased;
+            () = cancellation.cancelled() => return None,
+            writer = crate::daemon::core_admission::park_admission(
+                self.gate.acquire(&scope)
+            ) => writer,
+        };
+        Some(operation().await)
+    }
+
     pub(super) async fn try_with_writer<Operation, OperationFuture, Output>(
         &self,
         operation: Operation,
@@ -1931,10 +1965,21 @@ impl StoreAdministration {
         Operation: FnOnce() -> OperationFuture,
         OperationFuture: Future<Output = Output>,
     {
-        let writer = self.gate.try_acquire(&WriterScope::Daemon)?;
-        let output = operation().await;
-        drop(writer);
-        Some(output)
+        self.try_with_writer_in(WriterScope::Daemon, operation)
+            .await
+    }
+
+    pub(super) async fn try_with_writer_in<Operation, OperationFuture, Output>(
+        &self,
+        scope: WriterScope,
+        operation: Operation,
+    ) -> Option<Output>
+    where
+        Operation: FnOnce() -> OperationFuture,
+        OperationFuture: Future<Output = Output>,
+    {
+        let _writer = self.gate.try_acquire(&scope)?;
+        Some(operation().await)
     }
 
     /// Resolves the authenticated client's project layout and runs destructive