feat(desktop): wire owner-identity egress barrier into identity import (P29/P30-C1)

The identity-transition coordinator now drains owner-identity egress between
the runtime drain and the durable B dispatch, closing the split-state class
where an already-issued lease or durable capability could transmit under A
after B commits. Because the coordinator body holds two std-mutex Layer-2
guards it cannot carry across .await, the drain is synchronous: a new
`wait_egress_drain_blocking` parks on a Condvar beside the egress registry,
notified from both lease Drop impls. Both guards stay held continuously across
the barrier — releasing either permits A-runtime resurrection before B commits
(the three-phase guard-release shape was ruled UNSAFE). Deadlock-freedom rests
on leases taking only the egress mutex; a tree sweep found zero lease-vs-guard
sites.

The persist stage is now classified against durable fact (P27-C1) under the
retained commit fence, after a fsync'd IdentityTransitionPending journal
(P28-C1): Committed finishes the in-memory swap and clears the journal;
DefinitelyUnchanged compensates the drain; Indeterminate latches fail-closed
and leaves the journal for boot reconciliation. `commit_imported_identity` no
longer performs the durable persist — it takes the proven storage and does
only the in-memory swap, so it runs on the Committed arm alone.

Retires the dead_code allows the coordinator now consumes across the egress,
persistence, and journal modules. An RAII test guard resets the egress
registry on drop so a latched Indeterminate state cannot leak into crate tests
that read the indeterminate-gated key accessors without the registry lock.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
2026-08-18 17:37:54 -04:00
parent 09bc66427d
commit 998f1ef8b9
6 changed files with 272 additions and 119 deletions

View File

@@ -387,27 +387,24 @@ pub async fn import_identity(
run_identity_transition(app_handle, nsec, password, None, || Ok(())).await
}
/// Commit an imported identity: durably persist, swap in-memory keys, clear
/// recovery flags, then remove the previous identity's stale app-managed
/// backup. Caller must hold `state.identity_mutation`.
/// Finish an imported identity commit whose durable persist already succeeded
/// (`PersistenceOutcome::Committed`): swap in-memory keys, clear recovery flags,
/// then remove the previous identity's stale app-managed backup. Caller must
/// hold `state.identity_mutation` and have proven B durably canonical via
/// [`persist_imported_identity_classified`](crate::identity_persistence::persist_imported_identity_classified).
///
/// Ordering is the contract:
///
/// 1. `persist` runs FIRST. If it fails (`Err` from both keyring and file
/// fallback), nothing has changed — the previous identity stays live in
/// memory AND its valid canonical `identity.ncryptsec` stays on disk.
/// 2. Only after durable persistence do we swap `state.keys` and clear the
/// recovery flags.
/// 3. Stale-backup cleanup runs LAST and is deliberately best-effort: at that
/// point the import is durably committed, so reporting a cleanup failure
/// as a command `Err` would claim a half-applied import that actually
/// succeeded. The leftover blob is still passphrase-encrypted and is
/// replaced by the next backup creation; we log and move on.
/// `storage` records where B durably landed (reported on the success result).
/// Splitting the durable persist (now the classifier's, run under the egress
/// barrier) from this in-memory swap is the P27-C1 shape: this function runs
/// ONLY on the `Committed` arm, so it never restores live A beside durable B.
/// Stale-backup cleanup runs LAST and is best-effort: B is already durable, so
/// a cleanup failure must not be reported as a failed commit. The leftover blob
/// is still passphrase-encrypted and is replaced by the next backup creation.
pub(crate) fn commit_imported_identity(
state: &AppState,
data_dir: &std::path::Path,
keys: nostr::Keys,
persist: impl FnOnce(&nostr::Keys) -> Result<crate::app_state::IdentityStorage, String>,
storage: crate::app_state::IdentityStorage,
) -> Result<(nostr::PublicKey, crate::app_state::IdentityStorage), String> {
// Capture the previous pubkey up front for post-commit cleanup.
let previous_pubkey = state
@@ -415,8 +412,6 @@ pub(crate) fn commit_imported_identity(
.map_err(|e| e.to_string())?
.public_key();
let storage = persist(&keys)?;
// Update in-memory keys BEFORE clearing recovery flags. The Release
// stores below pair with Acquire loads in get_identity: a reader
// observing false is guaranteed to see the updated keys.

View File

@@ -25,6 +25,39 @@ fn drain_managed_agent_runtimes_for_import(
}
}
/// Compensate a drained runtime set with NO durable identity write — the shared
/// unwind for every barrier abort (drain failure, fenced-gate/journal failure,
/// `DefinitelyUnchanged`). The store guard is dropped FIRST because
/// [`compensate_drain`](crate::managed_agents::compensate_drain) re-acquires it,
/// and the runtime-transition guard is passed BY VALUE so compensation runs
/// without any interleave window. `scope`/`rt_guard` are `None` on the no-scope
/// path (nothing drained); either being `None` skips compensation. Returns a
/// combined diagnostic string when compensation itself fails.
fn compensate_import_drain(
app_handle: &tauri::AppHandle,
stopped: &[crate::managed_agents::DrainJournalEntry],
scope: Option<&crate::managed_agents::scope::WorkspaceAgentScope>,
rt_guard: Option<std::sync::MutexGuard<'_, ()>>,
store_guard: Option<std::sync::MutexGuard<'_, ()>>,
reason: &str,
) -> String {
// Drop the store lock before compensating (compensate_drain re-acquires it).
drop(store_guard);
let comp_err = match (scope, rt_guard) {
(Some(scope), Some(rt_guard)) => {
crate::managed_agents::compensate_drain(app_handle, stopped, scope, rt_guard)
}
(_, leftover_guard) => {
drop(leftover_guard);
None
}
};
match comp_err {
Some(comp) => format!("{reason}; compensation failed: {comp}"),
None => reason.to_string(),
}
}
/// The shared identity-transition coordinator (P25-C1): the ONE sink both
/// runtime identity-swap callers route through. Acquires the transition locks
/// in ONE unconditional order — `identity_mutation` → `workspace_transition`
@@ -190,6 +223,15 @@ fn import_identity_blocking(
let pre_import_scope = state.capture_active_scope();
let has_active_scope = pre_import_scope.is_some();
// The identity active before this transition (A), captured for the P28-C1
// journal BEFORE any swap. `to_pubkey` is the imported identity (B).
let from_pubkey = state
.identity_lifecycle_keys_guard()
.map_err(|e| format!("read current identity for transition journal: {e}"))?
.public_key()
.to_hex();
let to_pubkey = keys.public_key().to_hex();
let _rt_transition_guard = if has_active_scope {
Some(
state
@@ -216,28 +258,14 @@ fn import_identity_blocking(
match drain_managed_agent_runtimes_for_import(&app_handle, &state) {
Ok(stopped) => stopped,
Err((stopped, drain_err)) => {
// Drain failed — drop the store lock BEFORE compensating
// (compensate_drain re-acquires it), but pass the transition
// guard into compensate_drain so there is no interleave window.
drop(_store_guard);
let comp_err = match (pre_import_scope.as_ref(), _rt_transition_guard) {
(Some(scope), Some(rt_guard)) => crate::managed_agents::compensate_drain(
&app_handle,
&stopped,
scope,
rt_guard,
),
(_, leftover_guard) => {
drop(leftover_guard);
None
}
};
let msg = match comp_err {
Some(comp) => format!(
"identity import drain failed: {drain_err}; compensation failed: {comp}"
),
None => format!("identity import drain failed: {drain_err}"),
};
let msg = compensate_import_drain(
&app_handle,
&stopped,
pre_import_scope.as_ref(),
_rt_transition_guard,
_store_guard,
&format!("identity import drain failed: {drain_err}"),
);
return Err(msg);
}
}
@@ -245,66 +273,125 @@ fn import_identity_blocking(
vec![]
};
// ── Fenced pre-commit gate + durable commit (P25-C1 / P26-C1) ─────────
// After a successful drain, immediately before the durable commit. The
// commit fence (when supplied) is acquired FIRST, the validity check runs
// under it, and the durable commit runs under the SAME held guard — so a
// racing supersession can neither slip between the check and the commit nor
// interleave with it. A failed check short-circuits and NEVER commits.
let commit_result = commit_under_fence(commit_fence.as_deref(), validity_check, || {
super::commit_imported_identity(&state, &data_dir, keys, |keys| {
// Persist into the OS keyring first (store → read-back verify →
// marker → delete file). Falls back to the 0o600 file when the
// keyring is unavailable; returns Err only when both backends fail.
let store =
crate::secret_store::SecretStore::shared(crate::app_state::keyring_service());
crate::app_state::persist_imported_identity(store, keys, &key_path, &data_dir)
})
// ── Owner-identity egress barrier (P29/P30-C1) ────────────────────────
// Between the runtime drain and the durable dispatch, drain owner-identity
// egress: bump the identity-persistence generation, refuse new lease
// admission, await in-flight leases, and revoke old-generation durable
// capabilities (sessions/bearers). Both Layer-2 guards remain held
// continuously across this barrier — releasing either permits A-runtime
// resurrection before B commits (Thufir UNSAFE verdict on the three-phase
// split, Paul-ruled `092acdeb75`). The wait is synchronous
// (`wait_egress_drain_blocking`) because this body holds two std mutex
// guards that cannot cross `.await`; deadlock-freedom rests on leases
// taking only the egress registry mutex (zero lease-vs-guard sites).
//
// The barrier runs UNCONDITIONALLY, even on the no-scope path: owner sends
// are scope-independent, so an in-flight owner lease can exist with no
// active agent scope.
let winning_generation = crate::owner_identity_egress::begin_egress_drain()?;
crate::owner_identity_egress::wait_egress_drain_blocking();
crate::owner_identity_egress::revoke_durable_capabilities_before(winning_generation);
// ── Fenced pre-commit gate → journal → classified durable persist ─────
// The commit fence (when supplied) is acquired FIRST and held across the
// journal write AND the durable persist (P26-C1 fence retention): a racing
// supersession that must take the same fence can neither slip between the
// check and the durable dispatch nor interleave with it. The P28-C1 journal
// is written ONLY after the validity gate passes, so a superseded recovery
// leaves no journal residue. The persist is CLASSIFIED against durable fact
// (P27-C1) — never the kernel's `Ok`/`Err`.
let store = crate::secret_store::SecretStore::shared(crate::app_state::keyring_service());
let pending = crate::identity_transition_journal::IdentityTransitionPending {
from_pubkey,
to_pubkey,
};
let barrier_result = commit_under_fence(commit_fence.as_deref(), validity_check, || {
crate::identity_transition_journal::write_pending(&data_dir, &pending)?;
Ok(
crate::identity_persistence::persist_imported_identity_classified(
&keys,
store,
&key_path,
|| crate::app_state::persist_imported_identity(store, &keys, &key_path, &data_dir),
),
)
});
// If the fenced gate or the durable persist failed after a successful
// drain, compensate the drained runtimes with NO durable identity write.
// Drop the store lock BEFORE calling compensate_drain; pass the transition
// guard into it so there is no interleave window.
let (pubkey, storage) = match commit_result {
Ok(result) => result,
use crate::identity_persistence::PersistenceOutcome;
let (pubkey, storage) = match barrier_result {
// Validity gate or journal write failed BEFORE any durable B persist:
// reopen admission, compensate the drained runtimes, no durable write.
Err(e) => {
if !stopped_entries.is_empty() {
drop(_store_guard);
let comp_err = match (pre_import_scope.as_ref(), _rt_transition_guard) {
(Some(scope), Some(rt_guard)) => crate::managed_agents::compensate_drain(
&app_handle,
&stopped_entries,
scope,
rt_guard,
),
(_, leftover_guard) => {
drop(leftover_guard);
None
}
};
if let Some(comp_err) = comp_err {
eprintln!(
"buzz-desktop: identity import persist failed, compensation failed: {comp_err}"
);
crate::owner_identity_egress::resume_egress_live();
let msg = compensate_import_drain(
&app_handle,
&stopped_entries,
pre_import_scope.as_ref(),
_rt_transition_guard,
_store_guard,
&e,
);
return Err(msg);
}
// Durable B proven canonical: finish the in-memory swap through the
// already-held guards (no fallible acquisition after durability,
// P26-C2), clear the journal, and reopen admission at generation B.
Ok(PersistenceOutcome::Committed(storage)) => {
let committed = super::commit_imported_identity(&state, &data_dir, keys, storage);
match committed {
Ok(committed) => committed,
// The in-memory swap itself failed after durable B landed —
// a poisoned `state.keys` guard. B IS durable, so this is NOT
// compensatable: latch fail-closed and leave the journal for
// boot reconciliation rather than restoring live A beside B.
Err(e) => {
crate::owner_identity_egress::latch_identity_indeterminate();
return Err(format!(
"identity B durably committed but the in-memory swap failed: {e}; \
latched indeterminate — relaunch to reconcile from durable fact"
));
}
}
return Err(e);
}
// B never landed, A intact: the only outcome permitted to compensate.
Ok(PersistenceOutcome::DefinitelyUnchanged) => {
let _ = crate::identity_transition_journal::clear_pending(&data_dir);
crate::owner_identity_egress::resume_egress_live();
let msg = compensate_import_drain(
&app_handle,
&stopped_entries,
pre_import_scope.as_ref(),
_rt_transition_guard,
_store_guard,
"identity import persist failed and durable A is still canonical",
);
return Err(msg);
}
// Neither identity provable: latch the durable fail-closed state, LEAVE
// the journal, do NOT compensate or clear scope — runtimes stay down.
// Boot/reconciliation resolves it later from durable fact (P28-C1).
Ok(PersistenceOutcome::Indeterminate(reason)) => {
crate::owner_identity_egress::latch_identity_indeterminate();
return Err(format!(
"identity import could not prove either identity canonical: {reason}; \
latched indeterminate — relaunch to reconcile"
));
}
};
// ── Clear active scope and bump generation ────────────────────────────
// For no-active-scope path: no scope was ever set; clearing is a no-op
// but bumping generation invalidates any in-flight stale operations.
// For live-active path: agents are stopped; clearing scope makes all
// agent commands fail closed until the frontend re-applies a workspace.
//
// Invariant: the fallback relay can never claim legacy data — claims
// are only written inside apply_workspace's prepare stage.
//
// `clear_active_scope()` internally calls `next_scope_generation()` —
// no additional bump is needed here.
// ── Proven Committed: clear journal, scope, reopen admission ──────────
// The journal is cleared only here, at the proven `Committed` exit after
// the in-memory swap completed (best-effort — a proven-canonical identity
// must never be blocked by a stale delete). `clear_active_scope()` bumps
// the scope generation, making all agent commands fail closed until the
// frontend re-applies a workspace. `resume_egress_live()` reopens egress at
// generation B. The fallback relay can never claim legacy data — claims are
// only written inside apply_workspace's prepare stage.
if let Err(e) = crate::identity_transition_journal::clear_pending(&data_dir) {
eprintln!("buzz-desktop: identity committed, but journal clear failed: {e}");
}
state.clear_active_scope();
crate::owner_identity_egress::resume_egress_live();
let pubkey_hex = pubkey.to_hex();
let display_name = super::truncated_display_name(&pubkey)?;

View File

@@ -47,14 +47,9 @@ pub(crate) enum PersistenceOutcome {
DefinitelyUnchanged,
/// Durable B is proven canonical. The coordinator must finish the B commit;
/// compensation is forbidden. `storage` records where B durably landed.
// `storage` consumed by C5 (P29/P30 barrier finishes the B commit); remove
// allow when the coordinator reads it.
#[allow(dead_code)]
Committed(IdentityStorage),
/// Neither state is proven. The coordinator latches the durable fail-closed
/// state (P28-C1). `reason` is a diagnostic, never a control signal.
// Consumed by C5 (P28 latch); remove allow when the coordinator wires it.
#[allow(dead_code)]
Indeterminate(String),
}
@@ -76,9 +71,6 @@ pub(crate) enum PersistenceOutcome {
///
/// `expected` is the identity being imported (B). `store` and `legacy_path`
/// name the durable backends to re-read.
// Consumed by C5 (P25/P28 coordinator); remove allow when the coordinator
// routes its persist through this classifier.
#[allow(dead_code)]
pub(crate) fn persist_imported_identity_classified(
keys: &Keys,
store: &impl IdentityKeyStore,

View File

@@ -39,8 +39,6 @@ fn journal_path(data_dir: &Path) -> PathBuf {
/// durable — the coordinator must treat that as a hard failure and NOT proceed
/// to the durable dispatch, because an undurable journal cannot fail closed on
/// restart.
// Consumed by C5 (P28 coordinator); remove allow when the coordinator writes it.
#[allow(dead_code)]
pub(crate) fn write_pending(
data_dir: &Path,
pending: &IdentityTransitionPending,
@@ -63,8 +61,6 @@ pub(crate) fn write_pending(
/// identity must never be blocked by a stale journal delete failure; the next
/// coordinator run or a manual retry can clear it. Returns `Err` only to
/// surface a diagnostic; callers on the proven-exit path log and continue.
// Consumed by C5 (P28 coordinator); remove allow when the coordinator clears it.
#[allow(dead_code)]
pub(crate) fn clear_pending(data_dir: &Path) -> Result<(), String> {
let path = journal_path(data_dir);
match std::fs::remove_file(&path) {

View File

@@ -396,8 +396,6 @@ fn register_durable<P: CapabilityPolicy>(
/// [`begin_egress_drain`]/[`await_egress_drain`] and BEFORE the journal write +
/// durable B dispatch, so no old-generation session or bearer can transmit
/// across the durable boundary. It only *invokes* the handles C2 registered.
// Consumed by C5 (P25/P28 coordinator barrier); remove allow when C5 lands.
#[allow(dead_code)]
pub fn revoke_durable_capabilities_before(winning_generation: u64) -> usize {
let mut durable = lock_durable();
let stale: Vec<u64> = durable

View File

@@ -118,7 +118,7 @@
//! 5. the four `project_git_workflow` PR-status/merge sends (items 5–8).
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{LazyLock, Mutex};
use std::sync::{Condvar, LazyLock, Mutex};
use tokio::sync::Notify;
@@ -165,6 +165,12 @@ struct Registry {
generation: AtomicU64,
/// Notified when `in_flight` reaches zero so the drain can wake.
drained: Notify,
/// Blocking analogue of `drained` for the C5 coordinator's synchronous
/// drain: the barrier runs inside a `spawn_blocking` critical section that
/// holds two `std::sync::Mutex` guards it cannot carry across `.await`, so
/// it awaits lease-drain on this condvar (paired with `inner`) rather than
/// the async `drained`. Both are notified from the lease `Drop` impls.
drained_blocking: Condvar,
}
static REGISTRY: LazyLock<Registry> = LazyLock::new(|| Registry {
@@ -174,6 +180,7 @@ static REGISTRY: LazyLock<Registry> = LazyLock::new(|| Registry {
}),
generation: AtomicU64::new(0),
drained: Notify::new(),
drained_blocking: Condvar::new(),
});
fn lock_inner() -> std::sync::MutexGuard<'static, RegistryInner> {
@@ -242,8 +249,12 @@ impl Drop for OwnerIdentityEgressLease {
drop(inner);
if drained {
// Wake a drain awaiting the last in-flight lease. Harmless when no
// drain is in progress (no waiter registered).
// drain is in progress (no waiter registered). Both the async
// (`drained`) and blocking (`drained_blocking`) drains are woken;
// each re-checks `in_flight` under the lock, so notifying the one
// with no waiter is a no-op.
REGISTRY.drained.notify_waiters();
REGISTRY.drained_blocking.notify_all();
}
}
}
@@ -307,8 +318,6 @@ fn admit_owner_identity_after_wait() -> Result<OwnerIdentityEgressLease, String>
/// Idempotent-guard: only transitions from `Live`. Returns `Err` if a
/// transition is already in flight or latched — the coordinator serializes
/// transitions under its own guards, so this is a defensive invariant check.
// Consumed by C5 (P25/P28 coordinator); remove allow when C5 lands.
#[allow(dead_code)]
pub fn begin_egress_drain() -> Result<u64, String> {
let mut inner = lock_inner();
if inner.state != IdentityPersistenceState::Live {
@@ -344,12 +353,40 @@ pub async fn await_egress_drain() {
}
}
/// Synchronously await every in-flight lease admitted before the drain began —
/// the C5 coordinator's blocking analogue of [`await_egress_drain`]. Returns
/// once no lease is outstanding.
///
/// The coordinator barrier runs inside a `spawn_blocking` critical section that
/// holds two `std::sync::Mutex` guards (`managed_agent_runtime_transition`,
/// `managed_agents_store_lock`) it cannot carry across `.await`, so it drains
/// on the `drained_blocking` condvar rather than the async `drained`. Called
/// after [`begin_egress_drain`] and BEFORE revoke/journal/persist.
///
/// # Deadlock-freedom (load-bearing invariant)
///
/// Blocking on lease-drain while the caller owns both Layer-2 guards cannot
/// cycle: a lease `Drop` acquires ONLY the egress registry mutex (never either
/// managed-agent guard), and a tree sweep found ZERO sites that hold an egress
/// lease while acquiring either guard. `wait_while` releases the registry mutex
/// while parked, so an in-flight lease's `Drop` always makes progress and
/// notifies. The wait has NO timeout, matching [`await_egress_drain`]: every
/// in-flight lease is a network op with its own timeout, so the wait is bounded
/// transitively — a defensive timeout would only mask a leaked lease.
pub fn wait_egress_drain_blocking() {
let inner = lock_inner();
// wait_while re-checks under the lock and releases it while parked, so a
// lease Drop that reaches zero between the check and the park is not lost.
let _guard = REGISTRY
.drained_blocking
.wait_while(inner, |inner| inner.in_flight != 0)
.unwrap_or_else(|poisoned| poisoned.into_inner());
}
/// Resume admission at a proven transition exit (`Committed` finished, or
/// `DefinitelyUnchanged` / verified-rollback): state → `Live`. The generation
/// is already current for the winning identity, so leases admitted from here
/// carry it.
// Consumed by C5 (P25/P28 coordinator); remove allow when C5 lands.
#[allow(dead_code)]
pub fn resume_egress_live() {
lock_inner().state = IdentityPersistenceState::Live;
}
@@ -358,8 +395,6 @@ pub fn resume_egress_live() {
/// → `Indeterminate`. Owner-identity egress admission and the checked key
/// accessors refuse until [`resume_egress_live`] is called after reconciliation
/// proves one durable identity canonical.
// Consumed by C5 (P25/P28 coordinator); remove allow when C5 lands.
#[allow(dead_code)]
pub fn latch_identity_indeterminate() {
lock_inner().state = IdentityPersistenceState::Indeterminate;
}
@@ -449,7 +484,10 @@ impl Drop for ManagedAgentEgressLease {
let drained = inner.in_flight == 0;
drop(inner);
if drained {
// Wake both the async and blocking drains; each re-checks
// `in_flight` under the lock, so the one with no waiter is a no-op.
REGISTRY.drained.notify_waiters();
REGISTRY.drained_blocking.notify_all();
}
}
}
@@ -505,10 +543,9 @@ pub use durable::{
register_owner_artifact, register_owner_bearer, register_owner_session, BearerPolicy,
OwnerIdentityCapability, SessionPolicy, StampedArtifact,
};
// The C5 coordinator barrier and its registration-completeness reader; no
// production caller until C5, so the re-exports are unused today. Consumed by
// C5 (P25/P28); remove allow when C5 lands.
#[allow(unused_imports)]
// The C5 coordinator barrier's durable-capability revocation, consumed by
// `import_identity_blocking`. Its registration-completeness reader
// (`live_durable_capability_count`) stays test-only; retired in F.
pub use durable::revoke_durable_capabilities_before;
// The artifact policy marker and its wire stamp are named at the frontend
// application sites in C6/C7 (generation-compare) and by the §7 artifact
@@ -553,12 +590,27 @@ pub(crate) fn test_owner_egress_lease() -> EgressLease {
mod tests {
use super::*;
fn guard() -> std::sync::MutexGuard<'static, ()> {
/// RAII test guard: resets the process-global registry on BOTH entry and
/// drop. Resetting on drop is load-bearing — a test that latches
/// `Indeterminate` (or leaves the state `Draining`) must not leak that into
/// tests elsewhere in the crate that read the indeterminate-gated key
/// accessors (`signing_keys`) WITHOUT holding `EGRESS_REGISTRY_TEST_LOCK`.
/// Holding only on entry left the last-run state resident until the next
/// registry test entered, which those unguarded readers observed as a leak.
struct RegistryGuard(#[allow(dead_code)] std::sync::MutexGuard<'static, ()>);
impl Drop for RegistryGuard {
fn drop(&mut self) {
reset_registry_for_test();
}
}
fn guard() -> RegistryGuard {
let g = EGRESS_REGISTRY_TEST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
reset_registry_for_test();
g
RegistryGuard(g)
}
#[test]
@@ -679,6 +731,39 @@ mod tests {
handle.await.unwrap();
}
#[test]
fn blocking_drain_returns_immediately_with_no_in_flight_leases() {
let _g = guard();
begin_egress_drain().unwrap();
// No leases outstanding — the synchronous drain returns without parking.
wait_egress_drain_blocking();
assert_eq!(lock_inner().in_flight, 0);
}
#[test]
fn blocking_drain_awaits_an_in_flight_lease_then_completes() {
let _g = guard();
let lease = admit_owner_identity_after_wait().unwrap();
begin_egress_drain().unwrap();
assert_eq!(lock_inner().in_flight, 1);
// The blocking drain must not return while the lease is held. Drop it
// from another THREAD (not a task — the wait parks the OS thread) after
// the drain has begun parking; the condvar notify on lease Drop wakes it.
let handle = std::thread::spawn(move || {
std::thread::sleep(std::time::Duration::from_millis(50));
drop(lease);
});
wait_egress_drain_blocking();
assert_eq!(
lock_inner().in_flight,
0,
"blocking drain awaited the lease drop"
);
handle.join().unwrap();
}
#[test]
fn lease_admitted_before_drain_carries_pre_bump_generation() {
let _g = guard();