diff --git a/desktop/src-tauri/src/commands/identity.rs b/desktop/src-tauri/src/commands/identity.rs index f4c444b37a..c224073705 100644 --- a/desktop/src-tauri/src/commands/identity.rs +++ b/desktop/src-tauri/src/commands/identity.rs @@ -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, + 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. diff --git a/desktop/src-tauri/src/commands/identity_transition.rs b/desktop/src-tauri/src/commands/identity_transition.rs index 4ff8b72597..e922ed4453 100644 --- a/desktop/src-tauri/src/commands/identity_transition.rs +++ b/desktop/src-tauri/src/commands/identity_transition.rs @@ -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>, + store_guard: Option>, + 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)?; diff --git a/desktop/src-tauri/src/identity_persistence.rs b/desktop/src-tauri/src/identity_persistence.rs index 32a4ab18e6..9299c58cbb 100644 --- a/desktop/src-tauri/src/identity_persistence.rs +++ b/desktop/src-tauri/src/identity_persistence.rs @@ -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, diff --git a/desktop/src-tauri/src/identity_transition_journal.rs b/desktop/src-tauri/src/identity_transition_journal.rs index 4f49181932..6e8c0462d2 100644 --- a/desktop/src-tauri/src/identity_transition_journal.rs +++ b/desktop/src-tauri/src/identity_transition_journal.rs @@ -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) { diff --git a/desktop/src-tauri/src/owner_identity_egress/durable.rs b/desktop/src-tauri/src/owner_identity_egress/durable.rs index f5faaa938a..5e5f7d4890 100644 --- a/desktop/src-tauri/src/owner_identity_egress/durable.rs +++ b/desktop/src-tauri/src/owner_identity_egress/durable.rs @@ -396,8 +396,6 @@ fn register_durable( /// [`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 = durable diff --git a/desktop/src-tauri/src/owner_identity_egress/mod.rs b/desktop/src-tauri/src/owner_identity_egress/mod.rs index 86e09eac3f..aeeba89e22 100644 --- a/desktop/src-tauri/src/owner_identity_egress/mod.rs +++ b/desktop/src-tauri/src/owner_identity_egress/mod.rs @@ -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 = LazyLock::new(|| Registry { @@ -174,6 +180,7 @@ static REGISTRY: LazyLock = 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 /// 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 { 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();