Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 115 additions & 3 deletions crates/tracedecay-store-runtime/src/writer_gate.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand All @@ -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<()>>,
}
Expand All @@ -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
Expand All @@ -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<()>>,
Expand Down Expand Up @@ -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(),
Expand All @@ -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),
}
Expand All @@ -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),
})
Expand All @@ -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::*;
Expand Down Expand Up @@ -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();
Expand Down
53 changes: 49 additions & 4 deletions crates/tracedecay/src/daemon/branch_admin.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand Down
Loading
Loading