From 6c2d26e884a17c42dc8907a1ac7261047050bd2d Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Thu, 30 Jul 2026 22:45:04 -0400 Subject: [PATCH] fix(managed-agents): add generation counter to ReadinessClaim to prevent stale-claim mutations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A preempted ReadinessClaim had no ownership identity: after force_claim_in_progress() seized the latch from a flush-owned InProgress, the flush's in-flight barrier could still call complete()/drop and unconditionally mutate the latch. Two legs: - Leg 1 (same-scope case): stale flush barrier finishes, calls resolve()+mark_ready() → latch certified Ready while reconcile was still retaining rows → stale rows publish unarbitrated. - Leg 2: stale barrier errors → RAII drop resets force-claim's InProgress to Unready mid-migration → next flush tick wins a fresh claim and certifies against mid-sequence state. - Leg 3: run_boot_barrier_for_scope's abandon arm returned Ok(()) which the enforcing wrapper treated as Success. Fix: the latch carries a u64 generation counter. force_claim_in_progress increments it; claim_in_progress records the current generation. Latch mutations move onto the claim: - claim.complete(db_path) → Ready only if latch generation matches. - Drop → Unready only if generation matches. - Preempted claims are silent no-ops in both paths. BarrierOutcome enum added; abandon arm returns Abandoned (not Success) so run_boot_barrier_enforcing never calls complete() for an abandoned scope. mark_ready() and resolve() removed from the public API; mark_unready() gated to #[cfg(test)] since all production transitions now go through the claim. Tests (g)/(h): stale complete/drop are no-ops after force-claim. Test (i): leg-1 end-to-end interleaving — stale barrier complete cannot certify Ready; rows retained after the stale complete cannot publish until force-claim's own barrier passes. Tests (c)/(e)/(f) updated to claim.complete() API. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- desktop/src-tauri/src/event_sync.rs | 2 +- .../src/managed_agents/config_barrier.rs | 47 ++- .../managed_agents/config_sync_readiness.rs | 320 +++++++++++++----- .../persona_events/flush_barrier_tests.rs | 128 ++++++- 4 files changed, 374 insertions(+), 123 deletions(-) diff --git a/desktop/src-tauri/src/event_sync.rs b/desktop/src-tauri/src/event_sync.rs index 49e1dc9d6f..7b66766ff3 100644 --- a/desktop/src-tauri/src/event_sync.rs +++ b/desktop/src-tauri/src/event_sync.rs @@ -30,7 +30,7 @@ pub fn run_event_sync(app: &tauri::AppHandle, owner_keys: &nostr::Keys, db_path: /// ```text /// force_claim_in_progress() → migrate_legacy_retention_into() /// → spawn_event_sync_with_held_claim() → run_event_sync() -/// → run_boot_barrier_after_claim() → mark_ready / Unready +/// → run_boot_barrier_after_claim() → Ready / Unready /// ``` /// /// This closes the `Unready` window that previously existed between diff --git a/desktop/src-tauri/src/managed_agents/config_barrier.rs b/desktop/src-tauri/src/managed_agents/config_barrier.rs index 4fe7d77767..2a7d466805 100644 --- a/desktop/src-tauri/src/managed_agents/config_barrier.rs +++ b/desktop/src-tauri/src/managed_agents/config_barrier.rs @@ -90,44 +90,59 @@ pub(crate) async fn run_boot_barrier_after_claim( run_boot_barrier_enforcing(app, claim).await; } +/// Outcome of a single barrier scope run. +#[derive(Debug)] +enum BarrierOutcome { + /// Decision pass completed; scope is ready for publication. + Success, + /// Workspace switched mid-flight; the old scope's barrier was abandoned. + /// The latch is left to its current owner — no transition is applied. + Abandoned, +} + /// Shared enforcement body: scope resolution, relay lookups, decision pass. /// /// Takes ownership of the `ReadinessClaim` RAII guard. On the success path, -/// the guard is resolved before `mark_ready` sets the final state. On any -/// error path (including panics), the guard's `Drop` impl resets to `Unready` -/// — so a panic inside `run_boot_barrier_for_scope` can never leave the latch -/// permanently wedged at `InProgress`. +/// `claim.complete(db_path)` transitions the latch to `Ready` only if the +/// claim's generation still matches (i.e., it has not been preempted by a +/// concurrent `force_claim_in_progress`). On any error path (including panics +/// and abandonment), the claim is dropped without calling `complete`, so the +/// RAII guard resets to `Unready` — but only if this claim is still current. +/// A preempted claim's drop is a structural no-op. async fn run_boot_barrier_enforcing( app: &tauri::AppHandle, claim: super::config_sync_readiness::ReadinessClaim, ) { - use super::config_sync_readiness; - let state = app.state::(); let scope = match active_retention_scope(app, &state) { Ok(scope) => scope, Err(error) => { tracing::warn!(target: "buzz::config_sync", %error, "boot barrier skipped: no active scope"); - // claim drops here → mark_unready via RAII + // claim drops here → Unready via RAII if still current gen return; } }; match run_boot_barrier_for_scope(app, &state, &scope).await { - Ok(()) => { - // Resolve the claim so the RAII drop does not reset to Unready. - claim.resolve(); - config_sync_readiness::mark_ready(scope.db_path.clone()); + Ok(BarrierOutcome::Success) => { + // complete() transitions to Ready only if this claim's generation + // still matches — a preempted claim is a silent no-op. tracing::info!( target: "buzz::config_sync", db_path = %scope.db_path.display(), "boot barrier complete; scope is ready for publication" ); + claim.complete(scope.db_path); + } + Ok(BarrierOutcome::Abandoned) => { + tracing::info!(target: "buzz::config_sync", "boot barrier abandoned: workspace switched"); + // Drop the claim without completing — RAII resets to Unready if + // still current gen, no-op if preempted. Either way, the latch + // is not certified Ready for an abandoned scope. } Err(error) => { tracing::warn!(target: "buzz::config_sync", %error, "boot barrier failed; scope stays closed for retry"); - // claim drops here → mark_unready via RAII (explicit call would - // double-reset, but that is idempotent; RAII is the safety net). + // claim drops here → Unready via RAII if still current gen } } } @@ -175,7 +190,7 @@ async fn run_boot_barrier_for_scope( app: &tauri::AppHandle, state: &AppState, scope: &RetentionScope, -) -> Result<(), String> { +) -> Result { let owner_pubkey = scope.owner_keys.public_key().to_hex(); // Enumerate BEFORE the lookups: the set of coordinates to decide is the @@ -238,7 +253,7 @@ async fn run_boot_barrier_for_scope( target: "buzz::config_sync", "boot barrier abandoned: workspace switched during relay lookups" ); - return Ok(()); + return Ok(BarrierOutcome::Abandoned); } let conn = open_retention_db(&scope.db_path)?; @@ -260,7 +275,7 @@ async fn run_boot_barrier_for_scope( } run_decision_pass(&conn, &owner_pubkey, &states)?; - Ok(()) + Ok(BarrierOutcome::Success) } /// Read the three record stores and enumerate their coordinates. diff --git a/desktop/src-tauri/src/managed_agents/config_sync_readiness.rs b/desktop/src-tauri/src/managed_agents/config_sync_readiness.rs index bea5627d01..dbb50fb18e 100644 --- a/desktop/src-tauri/src/managed_agents/config_sync_readiness.rs +++ b/desktop/src-tauri/src/managed_agents/config_sync_readiness.rs @@ -11,7 +11,7 @@ //! apply_workspace (force) ─────────────────────────► InProgress (from any state) //! flush retry (claim, only when Unready) ───────────► InProgress //! run_boot_barrier success ─────────────────────────► Ready(db_path) -//! run_boot_barrier error / abandon ─────────────────► Unready +//! run_boot_barrier error / abandon ─────────────────► Unready (if still current gen) //! flush: InProgress → skip tick; Ready(p) → publish //! ``` //! @@ -22,14 +22,27 @@ //! path. //! //! - [`force_claim_in_progress`] — preempting claim; transitions from **any** -//! state (including `InProgress` or `Ready`) to `InProgress`. Used by -//! `apply_workspace`, which invalidates whatever the prior scope certified. +//! state (including `InProgress` or `Ready`) to `InProgress` **and bumps +//! the generation counter**. Used by `apply_workspace`, which invalidates +//! whatever the prior scope certified. //! -//! Both variants return a [`ReadinessClaim`] RAII guard. The guard resets the -//! latch to `Unready` on drop unless [`ReadinessClaim::resolve`] has been -//! called first. This ensures a panic inside the reconcile+barrier window -//! always leaves the scope fail-closed and retryable rather than permanently -//! wedged at `InProgress`. +//! Both variants return a [`ReadinessClaim`] RAII guard that records the +//! generation it was issued under. Latch mutations (`complete` and `Drop`) +//! only take effect if the latch generation still matches the claim's. A +//! preempted claim is a structural no-op — it cannot certify or clear +//! readiness after `force_claim_in_progress` has invalidated it. +//! +//! # Ownership identity +//! +//! The generation counter is the ownership identity. When `force_claim_in_progress` +//! bumps the generation, every outstanding claim from the previous generation +//! becomes permanently inert: its `complete` call is a no-op, and its `Drop` +//! does not reset the latch. This closes both legs of the preempted-claim race: +//! +//! - Leg 1 (stale barrier certifies Ready): stale claim's `complete()` sees +//! a generation mismatch → no-op → latch stays `InProgress`. +//! - Leg 2 (stale drop destroys InProgress): stale claim's `Drop` sees a +//! generation mismatch → no-op → force-claim's `InProgress` is preserved. use std::path::{Path, PathBuf}; @@ -44,53 +57,92 @@ pub enum ReadinessState { Ready(PathBuf), } +// ── Internal latch state ────────────────────────────────────────────────────── + +struct LatchState { + state: ReadinessState, + /// Monotonically-increasing generation. Incremented by every + /// `force_claim_in_progress` call. A claim records the generation it was + /// issued under; transitions are only applied when the current generation + /// matches the claim's, structurally invalidating all prior claims. + generation: u64, +} + +impl LatchState { + const fn new() -> Self { + Self { + state: ReadinessState::Unready, + generation: 0, + } + } +} + // ── Process-global latch ────────────────────────────────────────────────────── #[cfg(not(test))] -static READINESS: std::sync::Mutex = std::sync::Mutex::new(ReadinessState::Unready); +static READINESS: std::sync::Mutex = std::sync::Mutex::new(LatchState::new()); // In tests, each test thread gets an isolated latch so tests cannot interfere. #[cfg(test)] thread_local! { - static READINESS: std::sync::Mutex = - const { std::sync::Mutex::new(ReadinessState::Unready) }; + static READINESS: std::sync::Mutex = + const { std::sync::Mutex::new(LatchState::new()) }; } // ── Public interface ────────────────────────────────────────────────────────── /// RAII guard that holds the `InProgress` claim. /// -/// Resets the latch to `Unready` on drop unless [`ReadinessClaim::resolve`] -/// has been called first. This ensures a panic (or early return) inside the -/// reconcile+barrier window always leaves the scope fail-closed and retryable. +/// Owns a generation stamp matching the latch generation at the time the claim +/// was issued. All latch mutations go through the claim: +/// +/// - [`ReadinessClaim::complete`] — transitions to `Ready(db_path)` if and +/// only if the latch generation still matches. Preempted claims are no-ops. +/// - `Drop` — resets to `Unready` if and only if the latch generation still +/// matches. A panic or early return in the reconcile+barrier window leaves +/// the scope fail-closed and retryable, but only if this claim is still +/// current. A preempted claim's drop is a no-op. /// /// Created by [`claim_in_progress`] or [`force_claim_in_progress`]. -#[must_use = "dropping a ReadinessClaim without calling resolve() resets latch to Unready"] +#[must_use = "dropping a ReadinessClaim without calling complete() resets latch to Unready if still current generation"] pub struct ReadinessClaim { - resolved: bool, + generation: u64, } impl ReadinessClaim { - fn new() -> Self { - Self { resolved: false } + fn new(generation: u64) -> Self { + Self { generation } } - /// Mark the claim as explicitly resolved (either `mark_ready` or - /// `mark_unready` will handle the final state transition). Call this - /// before the explicit resolution to prevent the drop guard from - /// redundantly resetting to `Unready` again. - pub fn resolve(mut self) { - self.resolved = true; - // Drop runs immediately but the flag prevents the reset. + /// Transition the latch to `Ready(db_path)` if and only if this claim is + /// still current (latch generation matches). A preempted claim's call is + /// a silent no-op — the preemptor owns the latch now. + /// + /// Consumes the claim so that `Drop` does not also attempt a reset. + pub fn complete(self, db_path: PathBuf) { + let _ = with_lock(|latch| { + if latch.generation == self.generation { + latch.state = ReadinessState::Ready(db_path); + } + // Mismatch → preempted claim, no-op. + }); + // Prevent Drop from resetting to Unready: we consumed self, so Drop + // will not run (the value is moved into this function's scope and the + // mem::forget below prevents it). We use mem::forget here because there + // is no other way to prevent Drop after consuming self. + std::mem::forget(self); } } impl Drop for ReadinessClaim { fn drop(&mut self) { - if !self.resolved { - // Panic or early return: leave the scope fail-closed and retryable. - mark_unready(); - } + // Panic or early return: reset to Unready only if this claim is still + // current. A preempted claim (generation mismatch) is a no-op. + let _ = with_lock(|latch| { + if latch.generation == self.generation { + latch.state = ReadinessState::Unready; + } + }); } } @@ -100,76 +152,80 @@ impl Drop for ReadinessClaim { /// owns the reconcile+barrier run. Returns `None` if the state was already /// `InProgress` or `Ready` — the caller must not start a barrier. /// -/// The returned guard resets to `Unready` on drop unless -/// [`ReadinessClaim::resolve`] is called first. +/// The returned guard's `Drop` resets to `Unready` only if the claim is +/// still current (generation matches). Call [`ReadinessClaim::complete`] on +/// success. pub fn claim_in_progress() -> Option { - let won = with_lock(|state| { - if *state == ReadinessState::Unready { - *state = ReadinessState::InProgress; - true + with_lock(|latch| { + if latch.state == ReadinessState::Unready { + latch.state = ReadinessState::InProgress; + Some(ReadinessClaim::new(latch.generation)) } else { - false + None } }) - .unwrap_or(false); - if won { - Some(ReadinessClaim::new()) - } else { - None - } + .unwrap_or(None) } /// Preempting claim: transitions from **any** state (including `InProgress` -/// or `Ready`) to `InProgress`. +/// or `Ready`) to `InProgress` and **bumps the generation counter**. /// /// Used by `apply_workspace`, which invalidates whatever the prior scope -/// certified. Because it transitions unconditionally, the flush loop will see -/// `InProgress` immediately after this call and skip its tick — even if the -/// prior scope was `Ready`. +/// certified. Bumping the generation structurally invalidates all outstanding +/// claims from the previous generation: their `complete` and `Drop` become +/// no-ops, so a stale flush barrier cannot certify readiness or reset +/// `InProgress` mid-migration. /// -/// Returns a `ReadinessClaim` RAII guard (always succeeds). +/// Returns a `ReadinessClaim` RAII guard stamped with the new generation +/// (always succeeds). pub fn force_claim_in_progress() -> ReadinessClaim { - let _ = with_lock(|state| { - *state = ReadinessState::InProgress; - }); - ReadinessClaim::new() + let gen = with_lock(|latch| { + latch.generation = latch.generation.wrapping_add(1); + latch.state = ReadinessState::InProgress; + latch.generation + }) + .unwrap_or(0); + ReadinessClaim::new(gen) } -/// Mark the scope `Ready` for the given `db_path`. -pub fn mark_ready(db_path: PathBuf) { - let _ = with_lock(|state| { - *state = ReadinessState::Ready(db_path); - }); -} - -/// Reset to `Unready`. Called on barrier error, workspace switch, or any -/// path that must leave the scope fail-closed. +/// Reset to `Unready` unconditionally. +/// +/// Use for test cleanup and exceptional reset paths only. Normal transitions +/// go through [`ReadinessClaim::complete`] or the `Drop` impl. +#[cfg(test)] pub fn mark_unready() { - let _ = with_lock(|state| { - *state = ReadinessState::Unready; + let _ = with_lock(|latch| { + latch.state = ReadinessState::Unready; }); } /// `true` iff the scope is `Ready` for exactly `db_path`. pub fn is_ready_for(db_path: &Path) -> bool { - with_lock(|state| matches!(&*state, ReadinessState::Ready(p) if p == db_path)).unwrap_or(false) + with_lock(|latch| matches!(&latch.state, ReadinessState::Ready(p) if p == db_path)) + .unwrap_or(false) } /// `true` iff a reconcile+barrier is currently in-flight. #[cfg(test)] pub fn is_in_progress() -> bool { - with_lock(|state| *state == ReadinessState::InProgress).unwrap_or(false) + with_lock(|latch| latch.state == ReadinessState::InProgress).unwrap_or(false) } /// Read a snapshot of the current state. pub fn readiness_state() -> Option { - with_lock(|state| state.clone()).ok() + with_lock(|latch| latch.state.clone()).ok() +} + +/// Read the current generation counter (for tests only). +#[cfg(test)] +pub fn current_generation() -> u64 { + with_lock(|latch| latch.generation).unwrap_or(0) } // ── Internal helpers ────────────────────────────────────────────────────────── #[cfg(not(test))] -fn with_lock(f: impl FnOnce(&mut ReadinessState) -> R) -> Result { +fn with_lock(f: impl FnOnce(&mut LatchState) -> R) -> Result { READINESS .lock() .map(|mut g| f(&mut g)) @@ -177,7 +233,7 @@ fn with_lock(f: impl FnOnce(&mut ReadinessState) -> R) -> Result { } #[cfg(test)] -fn with_lock(f: impl FnOnce(&mut ReadinessState) -> R) -> Result { +fn with_lock(f: impl FnOnce(&mut LatchState) -> R) -> Result { READINESS.with(|m| m.lock().map(|mut g| f(&mut g)).map_err(|e| e.to_string())) } @@ -195,23 +251,25 @@ mod tests { #[test] fn test_claim_in_progress_from_unready_succeeds() { mark_unready(); + let path = PathBuf::from("/scope/db"); let claim = claim_in_progress(); assert!(claim.is_some(), "first claim must succeed"); assert!(is_in_progress()); - claim.unwrap().resolve(); + claim.unwrap().complete(path); mark_unready(); // cleanup } #[test] fn test_claim_in_progress_is_single_flight() { mark_unready(); + let path = PathBuf::from("/scope/db"); let claim = claim_in_progress(); assert!(claim.is_some(), "first claim wins"); assert!( claim_in_progress().is_none(), "second claim rejected while InProgress" ); - claim.unwrap().resolve(); + claim.unwrap().complete(path); mark_unready(); // cleanup } @@ -219,19 +277,19 @@ mod tests { fn test_claim_rejected_when_ready() { mark_unready(); let path = PathBuf::from("/scope/db"); - mark_ready(path.clone()); + let claim = claim_in_progress().unwrap(); + claim.complete(path.clone()); assert!(claim_in_progress().is_none(), "cannot claim when Ready"); assert!(is_ready_for(&path)); mark_unready(); // cleanup } #[test] - fn test_mark_ready_then_is_ready_for() { + fn test_complete_sets_ready_for_path() { mark_unready(); let path = PathBuf::from("/scope/db"); let claim = claim_in_progress().unwrap(); - claim.resolve(); - mark_ready(path.clone()); + claim.complete(path.clone()); assert!(is_ready_for(&path)); assert!(!is_ready_for(Path::new("/other/db"))); mark_unready(); // cleanup @@ -240,14 +298,14 @@ mod tests { #[test] fn test_mark_unready_resets_to_claimable() { mark_unready(); + let path = PathBuf::from("/scope/db"); let claim = claim_in_progress().unwrap(); - claim.resolve(); + claim.complete(path); mark_unready(); assert_eq!(readiness_state(), Some(ReadinessState::Unready)); let claim2 = claim_in_progress(); assert!(claim2.is_some(), "after reset, claim must succeed"); - claim2.unwrap().resolve(); - mark_unready(); // cleanup + drop(claim2); // drop without complete → resets to Unready } #[test] @@ -255,30 +313,28 @@ mod tests { mark_unready(); let path = PathBuf::from("/scope/db"); let claim = claim_in_progress().unwrap(); - claim.resolve(); - mark_ready(path.clone()); + claim.complete(path.clone()); assert!(is_ready_for(&path)); mark_unready(); assert!(!is_ready_for(&path)); let claim2 = claim_in_progress().unwrap(); - claim2.resolve(); - mark_ready(path.clone()); + claim2.complete(path.clone()); assert!(is_ready_for(&path)); mark_unready(); // cleanup } #[test] - fn test_claim_drop_without_resolve_resets_to_unready() { + fn test_claim_drop_without_complete_resets_to_unready() { mark_unready(); { let _claim = claim_in_progress().unwrap(); assert!(is_in_progress()); - // drop without resolve + // drop without complete } assert_eq!( readiness_state(), Some(ReadinessState::Unready), - "dropped claim without resolve must reset to Unready" + "dropped claim without complete must reset to Unready" ); } @@ -286,16 +342,16 @@ mod tests { fn test_force_claim_preempts_ready() { mark_unready(); let path = PathBuf::from("/scope/db"); - mark_ready(path.clone()); + let first = claim_in_progress().unwrap(); + first.complete(path.clone()); assert!(is_ready_for(&path)); // force_claim must preempt Ready - let claim = force_claim_in_progress(); + let fc = force_claim_in_progress(); assert!( is_in_progress(), "force_claim must set InProgress even from Ready" ); - claim.resolve(); - mark_unready(); // cleanup + drop(fc); // cleanup via RAII } #[test] @@ -303,12 +359,94 @@ mod tests { mark_unready(); let first = claim_in_progress().unwrap(); assert!(is_in_progress()); + let gen_before = current_generation(); // force_claim must preempt even an existing InProgress let second = force_claim_in_progress(); assert!(is_in_progress()); - // resolve the first claim (already replaced, no-op since resolved=true) - first.resolve(); - second.resolve(); - mark_unready(); // cleanup + let gen_after = current_generation(); + assert_eq!( + gen_after, + gen_before + 1, + "force_claim must bump generation" + ); + // first is now stale — its drop must be a no-op (generation mismatch) + drop(first); + assert!( + is_in_progress(), + "stale drop must not reset latch to Unready" + ); + drop(second); // cleanup via RAII + } + + // ── (g) stale-claim completion is a no-op ──────────────────────────────── + // + // Setup: flush wins claim → force_claim preempts → stale claim calls + // complete() → latch stays InProgress; flush claim_in_progress still + // rejected (current owner is the force-claim generation). + #[test] + fn test_stale_claim_complete_is_noop() { + mark_unready(); + let path = PathBuf::from("/scope/db"); + + // Flush wins the initial claim. + let stale_claim = claim_in_progress().unwrap(); + let stale_gen = current_generation(); + + // apply_workspace force-claims, bumping the generation. + let force_claim = force_claim_in_progress(); + let new_gen = current_generation(); + assert_eq!(new_gen, stale_gen + 1, "generation must increment"); + assert!(is_in_progress()); + + // Stale barrier finishes and tries to certify Ready — must be a no-op. + stale_claim.complete(path.clone()); + assert!( + is_in_progress(), + "stale complete must not transition latch to Ready" + ); + assert!( + !is_ready_for(&path), + "stale complete must not set Ready(path)" + ); + + // Force-claim's InProgress must still block a new flush claim. + assert!( + claim_in_progress().is_none(), + "flush claim must still be rejected after stale complete" + ); + + // Cleanup: force_claim still owns the latch. + drop(force_claim); + assert_eq!(readiness_state(), Some(ReadinessState::Unready)); + } + + // ── (h) stale-claim drop is a no-op ────────────────────────────────────── + // + // Setup: flush wins claim → force_claim preempts → stale claim is dropped + // unresolved → latch stays InProgress (force-claim still owns it). + #[test] + fn test_stale_claim_drop_is_noop() { + mark_unready(); + + // Flush wins the initial claim. + let stale_claim = claim_in_progress().unwrap(); + let stale_gen = current_generation(); + + // apply_workspace force-claims. + let force_claim = force_claim_in_progress(); + let new_gen = current_generation(); + assert_eq!(new_gen, stale_gen + 1); + assert!(is_in_progress()); + + // Stale barrier errors out — its RAII drop must be a no-op. + drop(stale_claim); + assert!( + is_in_progress(), + "stale drop must not reset force-claim's InProgress to Unready" + ); + + // Cleanup. + drop(force_claim); + assert_eq!(readiness_state(), Some(ReadinessState::Unready)); } } diff --git a/desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs b/desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs index 31aa61720a..d13a70436d 100644 --- a/desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs +++ b/desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs @@ -15,6 +15,7 @@ #![cfg(not(target_os = "windows"))] use nostr::{EventBuilder, JsonUtil, Kind, Tag}; +use std::path::PathBuf; use tempfile::tempdir; use crate::app_state::build_app_state; @@ -166,7 +167,8 @@ fn test_in_progress_state_blocks_second_claim() { Some(ReadinessState::InProgress) ); - first_claim.unwrap().resolve(); + let path = PathBuf::from("/scope/db"); + first_claim.unwrap().complete(path); config_sync_readiness::mark_unready(); // cleanup } @@ -183,9 +185,9 @@ fn test_barrier_failure_resets_to_unready_for_retry() { let claim = config_sync_readiness::claim_in_progress(); assert!(claim.is_some()); - // Barrier error: dropping the claim without resolve() resets to Unready - // (RAII guard). Simulates run_boot_barrier_enforcing returning Err. - drop(claim); // intentional drop-without-resolve + // Barrier error: dropping the claim without complete() resets to Unready + // (RAII guard, generation matches). Simulates run_boot_barrier_enforcing returning Err. + drop(claim); // intentional drop-without-complete assert_eq!( config_sync_readiness::readiness_state(), @@ -200,7 +202,8 @@ fn test_barrier_failure_resets_to_unready_for_retry() { "after failure, next claim must succeed so the retry can run" ); - retry_claim.unwrap().resolve(); + let path = PathBuf::from("/scope/db"); + retry_claim.unwrap().complete(path); config_sync_readiness::mark_unready(); // cleanup } @@ -258,8 +261,8 @@ async fn test_inline_barrier_cannot_certify_readiness_before_reconcile() { set_publish_blocked(&conn, KIND_PERSONA, &pubkey, "stale-agent", true) .expect("barrier gates stale row"); } - claim.unwrap().resolve(); - config_sync_readiness::mark_ready(db_path.clone()); + claim.unwrap().complete(db_path.clone()); + // latch is now Ready // Step 5: scope is Ready; flush runs but the row is blocked — 0 publish. assert!( @@ -291,7 +294,7 @@ async fn test_inline_barrier_cannot_certify_readiness_before_reconcile() { // // End-to-end: spawn_event_sync claims InProgress → barrier fails (RAII drop // resets to Unready) → next tick re-claims InProgress → barrier succeeds -// (mark_ready) → row publishes on the following flush call. +// (claim.complete()) → row publishes on the following flush call. #[tokio::test] async fn test_retry_after_barrier_failure_publishes_when_ready() { config_sync_readiness::mark_unready(); @@ -312,11 +315,11 @@ async fn test_retry_after_barrier_failure_publishes_when_ready() { *state.relay_url_override.lock().unwrap() = Some(relay_url); // Phase 1: spawn_event_sync claims InProgress → barrier fails → drop - // resets to Unready via RAII. + // resets to Unready via RAII (generation matches, this is the current claim). { let claim = config_sync_readiness::claim_in_progress(); assert!(claim.is_some()); - // Simulate barrier error: drop without resolve(). + // Simulate barrier error: drop without complete(). drop(claim); } @@ -333,8 +336,7 @@ async fn test_retry_after_barrier_failure_publishes_when_ready() { retry_claim.is_some(), "after mark_unready, claim must succeed for retry" ); - retry_claim.unwrap().resolve(); - config_sync_readiness::mark_ready(db_path.clone()); + retry_claim.unwrap().complete(db_path.clone()); // Row is unblocked — flush must publish it now that scope is Ready. let flushed = flush_pending_events(&db_path, &state).await.expect("flush"); @@ -411,9 +413,8 @@ async fn test_apply_workspace_flush_cannot_interleave_before_reconcile() { .expect("gate reconciled row"); } - // Step 6: barrier completes — resolve claim and mark Ready. - claim.resolve(); - config_sync_readiness::mark_ready(db_path.clone()); + // Step 6: barrier completes — complete claim (transitions to Ready only if still current gen). + claim.complete(db_path.clone()); // Step 7: flush runs — both rows are blocked, 0 published. let state = build_app_state(); @@ -440,3 +441,100 @@ async fn test_apply_workspace_flush_cannot_interleave_before_reconcile() { config_sync_readiness::mark_unready(); // cleanup } + +// ── (i) leg-1 end-to-end: stale flush barrier cannot publish after force-claim ─ +// +// Paul's second bounce, leg 1: flush wins claim at t≈0, its inline barrier +// completes its relay round-trips and calls `complete()`, but by that time +// apply_workspace's `force_claim_in_progress` has bumped the generation. The +// stale `complete()` is a structural no-op — latch stays InProgress (the +// force-claim's generation). Rows retained AFTER the stale complete cannot +// publish until the force-claim's own barrier passes. +// +// Deterministic interleaving: +// 1. Flush wins claim at Unready → InProgress (gen 0). +// 2. apply_workspace fires → force_claim bumps to gen 1, sets InProgress. +// Stale flush claim is now generation 0 in a generation-1 latch. +// 3. Stale barrier "completes" its relay work and calls complete(path). +// Generation mismatch (0 ≠ 1) → no-op; latch stays InProgress. +// 4. Rows retained by reconcile land in the DB (publish_blocked=false). +// 5. Force-claim's barrier gates all rows (sets publish_blocked=true), then +// calls complete(path) with gen 1 → latch transitions to Ready. +// 6. Flush runs: scope is Ready, but rows are blocked — 0 published. +#[tokio::test] +async fn test_stale_flush_barrier_complete_cannot_publish_after_force_claim() { + config_sync_readiness::mark_unready(); + + let keys = nostr::Keys::generate(); + let pubkey = keys.public_key().to_hex(); + let dir = tempdir().expect("tempdir"); + let db_path = dir.path().join("retention.db"); + + // Step 1: flush wins the initial claim (gen 0). + let stale_flush_claim = config_sync_readiness::claim_in_progress().unwrap(); + assert!(config_sync_readiness::is_in_progress()); + let stale_gen = config_sync_readiness::current_generation(); + assert_eq!(stale_gen, 0); + + // Step 2: apply_workspace fires — force_claim bumps generation to 1. + let force_claim = config_sync_readiness::force_claim_in_progress(); + assert_eq!(config_sync_readiness::current_generation(), 1); + assert!(config_sync_readiness::is_in_progress()); + + // Stale flush cannot claim again (force-claim holds InProgress). + assert!(config_sync_readiness::claim_in_progress().is_none()); + + // Step 3: stale barrier "finishes" and calls complete(path) — gen mismatch → no-op. + stale_flush_claim.complete(db_path.clone()); + assert!( + config_sync_readiness::is_in_progress(), + "stale complete (gen 0 vs latch gen 1) must not certify Ready" + ); + assert!( + !config_sync_readiness::is_ready_for(&db_path), + "latch must not be Ready after stale complete" + ); + + // Step 4: reconcile retains rows — these are the rows that would have + // escaped if the stale complete had succeeded. + { + let conn = open_retention_db(&db_path).expect("open db"); + retain_pending(&conn, &keys, "post-reconcile-row"); + } + + // Stale barrier cannot certify again (latch is still InProgress gen 1). + assert!(config_sync_readiness::claim_in_progress().is_none()); + + // Step 5: force-claim's barrier gates all rows, then completes with gen 1 → Ready. + { + let conn = open_retention_db(&db_path).expect("open db for force barrier"); + set_publish_blocked(&conn, KIND_PERSONA, &pubkey, "post-reconcile-row", true) + .expect("force barrier gates reconcile row"); + } + force_claim.complete(db_path.clone()); + assert!( + config_sync_readiness::is_ready_for(&db_path), + "force-claim complete must transition to Ready" + ); + + // Step 6: flush runs — scope is Ready but rows are blocked by the + // force-claim's barrier. 0 published. + let state = build_app_state(); + *state.keys.lock().unwrap() = keys.clone(); + *state.relay_url_override.lock().unwrap() = Some(spawn_stub_relay().await); + + let flushed = flush_pending_events(&db_path, &state).await.expect("flush"); + assert_eq!( + flushed, 0, + "rows retained after stale complete must not publish — force-claim's barrier gated them" + ); + + let conn = open_retention_db(&db_path).expect("reopen db"); + let row = get_retained_event(&conn, KIND_PERSONA, &pubkey, "post-reconcile-row") + .unwrap() + .unwrap(); + assert!(row.publish_blocked, "post-reconcile row must stay blocked"); + assert!(row.pending_sync, "blocked row must stay pending"); + + config_sync_readiness::mark_unready(); // cleanup +}