From cf23ad694340551b7393ab2a42d93c2b6f1013e8 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Thu, 30 Jul 2026 21:18:19 -0400 Subject: [PATCH] fix(managed-agents): close apply_workspace Unready window and add RAII panic recovery apply_workspace previously called mark_unready() then relied on spawn_event_sync's inner claim_in_progress() CAS to reclaim InProgress before migration and reconcile ran. The flush loop (alive since process start) could win that CAS in the window between mark_unready() and the spawn, certifying readiness against pre-migration, pre-reconcile DB state. Two consequences: reconcile silently dropped (spawn_event_sync returns early on failed CAS), and legacy migration rows published unarbitrated. Fixes: PRIMARY: apply_workspace now calls force_claim_in_progress() atomically BEFORE migrate_legacy_retention_into. The claim (an RAII ReadinessClaim guard) is held across the entire sequence: legacy migration -> reconcile -> barrier -> mark_ready/Unready. spawn_event_sync_with_held_claim takes the held claim instead of racing its own CAS. flush loop sees InProgress throughout and cannot interleave. SECONDARY: ReadinessClaim RAII guard replaces the manual mark_unready/ mark_ready calls at barrier entry/exit sites. On success, claim.resolve() prevents the drop from resetting to Unready; on error or panic, the drop resets to Unready automatically. A panic inside run_boot_barrier_for_scope no longer leaves the latch permanently wedged at InProgress. API changes: - claim_in_progress() returns Option (was bool) - force_claim_in_progress() added: preempting, any-state -> InProgress - ReadinessClaim: RAII guard with resolve() explicit success path - run_boot_barrier_after_claim() takes ReadinessClaim (transfers ownership) - spawn_event_sync removed (dead code); spawn_event_sync_with_held_claim is now the sole spawn entry point New test (f): apply_workspace window -- force_claim preempts any state; flush claim between invalidation and reconcile is rejected; migrated and reconciled rows gated by post-reconcile barrier cannot publish. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- desktop/src-tauri/src/commands/workspace.rs | 15 +- desktop/src-tauri/src/event_sync.rs | 55 +++--- .../src/managed_agents/config_barrier.rs | 43 +++-- .../managed_agents/config_sync_readiness.rs | 171 +++++++++++++++--- .../persona_events/flush_barrier_tests.rs | 138 ++++++++++++-- 5 files changed, 328 insertions(+), 94 deletions(-) diff --git a/desktop/src-tauri/src/commands/workspace.rs b/desktop/src-tauri/src/commands/workspace.rs index 237d4082de..2db2faf406 100644 --- a/desktop/src-tauri/src/commands/workspace.rs +++ b/desktop/src-tauri/src/commands/workspace.rs @@ -217,20 +217,23 @@ pub async fn apply_workspace( // collapse every community into one pending-event store. match crate::managed_agents::retention::active_retention_scope(&restore_app, &state) { Ok(scope) => { - // Clear the readiness latch before spawning so the new scope is - // unready until its own boot barrier completes. Without this, - // a prior scope's latch could let the new scope's flush loop - // publish before its gate has run. - crate::managed_agents::config_sync_readiness::mark_unready(); + // Preempt the readiness latch from any state (including a prior + // scope's Ready) and hold InProgress across the entire sequence: + // legacy migration → reconcile → barrier. This closes the window + // between the old mark_unready() and spawn_event_sync's inner CAS, + // during which the flush loop could win the claim and certify + // readiness against pre-migration, pre-reconcile database state. + let claim = crate::managed_agents::config_sync_readiness::force_claim_in_progress(); // Adopt whatever the pre-scoping release left queued in the global // retention database BEFORE the scoped reconcile and flush run, so // stranded tombstones and archive requests publish on this boot // instead of being abandoned by the storage cutover. migrate_legacy_retention_into(&restore_app, &scope); - crate::event_sync::spawn_event_sync( + crate::event_sync::spawn_event_sync_with_held_claim( restore_app.clone(), scope.owner_keys, scope.db_path, + claim, ) } Err(error) => { diff --git a/desktop/src-tauri/src/event_sync.rs b/desktop/src-tauri/src/event_sync.rs index 87db42fe4e..49e1dc9d6f 100644 --- a/desktop/src-tauri/src/event_sync.rs +++ b/desktop/src-tauri/src/event_sync.rs @@ -19,45 +19,35 @@ pub fn run_event_sync(app: &tauri::AppHandle, owner_keys: &nostr::Keys, db_path: crate::managed_agents::reconcile::reconcile_agents_to_events(app, owner_keys, db_path); } -/// Spawn the best-effort event reconcile off the synchronous Tauri setup path. +/// Variant for workspace applies: spawn the reconcile+barrier with an +/// already-held `InProgress` claim. /// -/// The owner keys are cloned before spawning so the task never touches the -/// `AppState::keys` mutex. The reconcile itself is still synchronous JSON, -/// SQLite, and signing work, so it runs on the blocking pool rather than an -/// async worker. +/// `apply_workspace` calls +/// [`crate::managed_agents::config_sync_readiness::force_claim_in_progress`] +/// **before** `migrate_legacy_retention_into` runs, then passes the claim +/// here. The claim covers the entire sequence: /// -/// # Ordering: reconcile, then barrier, enforced by the single-flight latch +/// ```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 +/// ``` /// -/// The reconcile marks changed coordinates `pending_sync = 1`; the boot barrier -/// ([`crate::managed_agents::config_barrier::run_boot_barrier_after_claim`]) -/// then decides which may go out and gates the rest. Ordering is enforced -/// structurally: +/// This closes the `Unready` window that previously existed between +/// `mark_unready()` and `spawn_event_sync`'s inner `claim_in_progress`, during +/// which the flush loop could win the CAS and certify readiness against the +/// pre-migration, pre-reconcile database state. /// -/// 1. This function claims `InProgress` on the process-global readiness latch -/// BEFORE the reconcile task starts, preventing any concurrent flush tick -/// from running its own barrier against the pre-reconcile database state. -/// -/// 2. The blocking task runs reconcile; `run_boot_barrier_after_claim` then -/// completes the transition to `Ready(db_path)` (or back to `Unready` on -/// error). InProgress is held continuously — there is no reset between -/// reconcile and barrier, so no concurrent flush tick can win the CAS and -/// certify readiness against a pre-reconcile snapshot. -/// -/// 3. The flush loop sees `InProgress` during the entire reconcile+barrier -/// window and skips its tick rather than racing to certify readiness early. -pub fn spawn_event_sync( +/// The reconcile itself is still synchronous JSON, SQLite, and signing work, +/// so it runs on the blocking pool rather than an async worker. The owner keys +/// are cloned before spawning so the task never touches the `AppState::keys` +/// mutex. +pub fn spawn_event_sync_with_held_claim( app: tauri::AppHandle, owner_keys: nostr::Keys, db_path: std::path::PathBuf, + claim: crate::managed_agents::config_sync_readiness::ReadinessClaim, ) { - // Claim the single-flight slot synchronously before spawning. The flush - // loop sees InProgress immediately and skips until reconcile+barrier - // completes — no concurrent inline barrier can beat the reconcile. - if !crate::managed_agents::config_sync_readiness::claim_in_progress() { - // Already InProgress or Ready: another path owns this cycle. - return; - } - tauri::async_runtime::spawn(async move { let barrier_app = app.clone(); if let Err(e) = tauri::async_runtime::spawn_blocking(move || { @@ -70,7 +60,8 @@ pub fn spawn_event_sync( // InProgress is still held here — no mark_unready between reconcile // and the barrier. run_boot_barrier_after_claim owns the transition // to Ready or Unready without re-claiming. - crate::managed_agents::config_barrier::run_boot_barrier_after_claim(&barrier_app).await; + crate::managed_agents::config_barrier::run_boot_barrier_after_claim(&barrier_app, claim) + .await; }); } diff --git a/desktop/src-tauri/src/managed_agents/config_barrier.rs b/desktop/src-tauri/src/managed_agents/config_barrier.rs index c3e5559f5e..4fe7d77767 100644 --- a/desktop/src-tauri/src/managed_agents/config_barrier.rs +++ b/desktop/src-tauri/src/managed_agents/config_barrier.rs @@ -65,34 +65,42 @@ pub async fn run_boot_barrier(app: &tauri::AppHandle) { // Single-flight: only the first caller to claim InProgress runs the // barrier. If the state is already InProgress (spawn_event_sync's // reconcile is in-flight) or Ready, skip. - if !config_sync_readiness::claim_in_progress() { + let Some(claim) = config_sync_readiness::claim_in_progress() else { return; - } + }; - run_boot_barrier_enforcing(app).await; + run_boot_barrier_enforcing(app, claim).await; } /// Execute the barrier enforcement assuming the caller already claimed /// `InProgress`. /// -/// Called from `spawn_event_sync` after reconcile completes. The caller holds -/// `InProgress` through the entire reconcile+barrier window, preventing any -/// concurrent flush tick from certifying readiness against the pre-reconcile -/// database state. This function never calls `claim_in_progress` — ownership -/// is the caller's invariant. +/// Called from `spawn_event_sync` after reconcile completes, and from +/// `apply_workspace` (via `spawn_event_sync_with_held_claim`) after legacy +/// migration + reconcile. The caller holds a [`ReadinessClaim`] through the +/// entire reconcile+barrier window, preventing any concurrent flush tick from +/// running its own barrier against the pre-reconcile database state. /// /// On success: transitions to `Ready(db_path)`. On any error: resets to /// `Unready` (fail-closed, retryable on the next flush tick). -pub(crate) async fn run_boot_barrier_after_claim(app: &tauri::AppHandle) { - run_boot_barrier_enforcing(app).await; +pub(crate) async fn run_boot_barrier_after_claim( + app: &tauri::AppHandle, + claim: super::config_sync_readiness::ReadinessClaim, +) { + run_boot_barrier_enforcing(app, claim).await; } /// Shared enforcement body: scope resolution, relay lookups, decision pass. /// -/// Caller must own `InProgress` (either via `claim_in_progress` in -/// `run_boot_barrier`, or via the pre-spawn claim in `spawn_event_sync` passed -/// through `run_boot_barrier_after_claim`). -async fn run_boot_barrier_enforcing(app: &tauri::AppHandle) { +/// 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`. +async fn run_boot_barrier_enforcing( + app: &tauri::AppHandle, + claim: super::config_sync_readiness::ReadinessClaim, +) { use super::config_sync_readiness; let state = app.state::(); @@ -100,13 +108,15 @@ async fn run_boot_barrier_enforcing(app: &tauri::AppHandle) { Ok(scope) => scope, Err(error) => { tracing::warn!(target: "buzz::config_sync", %error, "boot barrier skipped: no active scope"); - config_sync_readiness::mark_unready(); + // claim drops here → mark_unready via RAII 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()); tracing::info!( target: "buzz::config_sync", @@ -116,7 +126,8 @@ async fn run_boot_barrier_enforcing(app: &tauri::AppHandle) { } Err(error) => { tracing::warn!(target: "buzz::config_sync", %error, "boot barrier failed; scope stays closed for retry"); - config_sync_readiness::mark_unready(); + // claim drops here → mark_unready via RAII (explicit call would + // double-reset, but that is idempotent; RAII is the safety net). } } } 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 b2f435ef02..bea5627d01 100644 --- a/desktop/src-tauri/src/managed_agents/config_sync_readiness.rs +++ b/desktop/src-tauri/src/managed_agents/config_sync_readiness.rs @@ -2,26 +2,34 @@ //! //! Tracks the single-flight state of the `Unready → InProgress → Ready` //! transition for the active retention scope, preventing a concurrent inline -//! flush barrier from certifying readiness before the reconcile that follows -//! `spawn_event_sync` has finished queuing its rows. +//! flush barrier from certifying readiness before the reconcile+migration that +//! follows `spawn_event_sync_with_held_claim` has finished queuing its rows. //! //! # State machine //! //! ```text -//! apply_workspace ─────────────────────────────────► Unready -//! spawn_event_sync (claim) ─────────────────────────► InProgress +//! 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 //! flush: InProgress → skip tick; Ready(p) → publish //! ``` //! -//! Exactly one caller at a time owns the `Unready → InProgress` transition. -//! `claim_in_progress` is a compare-and-swap under the process-global lock: -//! it succeeds only from `Unready` and returns `false` when the state is -//! already `InProgress` or `Ready`. An inline flush retry that fires while -//! `spawn_event_sync`'s reconcile is running will see `InProgress` and skip -//! its tick — it cannot certify readiness against the pre-reconcile database. +//! Two claim variants: +//! +//! - [`claim_in_progress`] — CAS from `Unready` only. Returns `None` when +//! the state is already `InProgress` or `Ready`. Used by the flush retry +//! 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. +//! +//! 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`. use std::path::{Path, PathBuf}; @@ -50,13 +58,52 @@ thread_local! { // ── 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. +/// +/// Created by [`claim_in_progress`] or [`force_claim_in_progress`]. +#[must_use = "dropping a ReadinessClaim without calling resolve() resets latch to Unready"] +pub struct ReadinessClaim { + resolved: bool, +} + +impl ReadinessClaim { + fn new() -> Self { + Self { resolved: false } + } + + /// 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. + } +} + +impl Drop for ReadinessClaim { + fn drop(&mut self) { + if !self.resolved { + // Panic or early return: leave the scope fail-closed and retryable. + mark_unready(); + } + } +} + /// Attempt a CAS from `Unready` → `InProgress`. /// -/// Returns `true` if the caller won the transition and now owns the -/// reconcile+barrier run. Returns `false` if the state was already +/// Returns `Some(ReadinessClaim)` if the caller won the transition and now +/// owns the reconcile+barrier run. Returns `None` if the state was already /// `InProgress` or `Ready` — the caller must not start a barrier. -pub fn claim_in_progress() -> bool { - with_lock(|state| { +/// +/// The returned guard resets to `Unready` on drop unless +/// [`ReadinessClaim::resolve`] is called first. +pub fn claim_in_progress() -> Option { + let won = with_lock(|state| { if *state == ReadinessState::Unready { *state = ReadinessState::InProgress; true @@ -64,7 +111,28 @@ pub fn claim_in_progress() -> bool { false } }) - .unwrap_or(false) + .unwrap_or(false); + if won { + Some(ReadinessClaim::new()) + } else { + None + } +} + +/// Preempting claim: transitions from **any** state (including `InProgress` +/// or `Ready`) to `InProgress`. +/// +/// 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`. +/// +/// Returns a `ReadinessClaim` RAII guard (always succeeds). +pub fn force_claim_in_progress() -> ReadinessClaim { + let _ = with_lock(|state| { + *state = ReadinessState::InProgress; + }); + ReadinessClaim::new() } /// Mark the scope `Ready` for the given `db_path`. @@ -127,19 +195,23 @@ mod tests { #[test] fn test_claim_in_progress_from_unready_succeeds() { mark_unready(); - assert!(claim_in_progress(), "first claim must succeed"); + let claim = claim_in_progress(); + assert!(claim.is_some(), "first claim must succeed"); assert!(is_in_progress()); + claim.unwrap().resolve(); mark_unready(); // cleanup } #[test] fn test_claim_in_progress_is_single_flight() { mark_unready(); - assert!(claim_in_progress(), "first claim wins"); + let claim = claim_in_progress(); + assert!(claim.is_some(), "first claim wins"); assert!( - !claim_in_progress(), + claim_in_progress().is_none(), "second claim rejected while InProgress" ); + claim.unwrap().resolve(); mark_unready(); // cleanup } @@ -148,7 +220,7 @@ mod tests { mark_unready(); let path = PathBuf::from("/scope/db"); mark_ready(path.clone()); - assert!(!claim_in_progress(), "cannot claim when Ready"); + assert!(claim_in_progress().is_none(), "cannot claim when Ready"); assert!(is_ready_for(&path)); mark_unready(); // cleanup } @@ -157,7 +229,8 @@ mod tests { fn test_mark_ready_then_is_ready_for() { mark_unready(); let path = PathBuf::from("/scope/db"); - claim_in_progress(); + let claim = claim_in_progress().unwrap(); + claim.resolve(); mark_ready(path.clone()); assert!(is_ready_for(&path)); assert!(!is_ready_for(Path::new("/other/db"))); @@ -167,10 +240,13 @@ mod tests { #[test] fn test_mark_unready_resets_to_claimable() { mark_unready(); - claim_in_progress(); + let claim = claim_in_progress().unwrap(); + claim.resolve(); mark_unready(); assert_eq!(readiness_state(), Some(ReadinessState::Unready)); - assert!(claim_in_progress(), "after reset, claim must succeed"); + let claim2 = claim_in_progress(); + assert!(claim2.is_some(), "after reset, claim must succeed"); + claim2.unwrap().resolve(); mark_unready(); // cleanup } @@ -178,14 +254,61 @@ mod tests { fn test_workspace_switch_cycle() { mark_unready(); let path = PathBuf::from("/scope/db"); - claim_in_progress(); + let claim = claim_in_progress().unwrap(); + claim.resolve(); mark_ready(path.clone()); assert!(is_ready_for(&path)); mark_unready(); assert!(!is_ready_for(&path)); - assert!(claim_in_progress()); + let claim2 = claim_in_progress().unwrap(); + claim2.resolve(); mark_ready(path.clone()); assert!(is_ready_for(&path)); mark_unready(); // cleanup } + + #[test] + fn test_claim_drop_without_resolve_resets_to_unready() { + mark_unready(); + { + let _claim = claim_in_progress().unwrap(); + assert!(is_in_progress()); + // drop without resolve + } + assert_eq!( + readiness_state(), + Some(ReadinessState::Unready), + "dropped claim without resolve must reset to Unready" + ); + } + + #[test] + fn test_force_claim_preempts_ready() { + mark_unready(); + let path = PathBuf::from("/scope/db"); + mark_ready(path.clone()); + assert!(is_ready_for(&path)); + // force_claim must preempt Ready + let claim = force_claim_in_progress(); + assert!( + is_in_progress(), + "force_claim must set InProgress even from Ready" + ); + claim.resolve(); + mark_unready(); // cleanup + } + + #[test] + fn test_force_claim_preempts_in_progress() { + mark_unready(); + let first = claim_in_progress().unwrap(); + assert!(is_in_progress()); + // 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 + } } 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 d6bb945bb7..31aa61720a 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 @@ -145,8 +145,9 @@ fn test_in_progress_state_blocks_second_claim() { config_sync_readiness::mark_unready(); // First caller (spawn_event_sync) claims InProgress. + let first_claim = config_sync_readiness::claim_in_progress(); assert!( - config_sync_readiness::claim_in_progress(), + first_claim.is_some(), "first claim must succeed (Unready → InProgress)" ); assert!(config_sync_readiness::is_in_progress()); @@ -154,7 +155,7 @@ fn test_in_progress_state_blocks_second_claim() { // Second caller (flush retry) must be rejected — it would run a barrier // against the pre-reconcile database state. assert!( - !config_sync_readiness::claim_in_progress(), + config_sync_readiness::claim_in_progress().is_none(), "second claim must be rejected while InProgress" ); @@ -165,6 +166,7 @@ fn test_in_progress_state_blocks_second_claim() { Some(ReadinessState::InProgress) ); + first_claim.unwrap().resolve(); config_sync_readiness::mark_unready(); // cleanup } @@ -178,10 +180,12 @@ fn test_barrier_failure_resets_to_unready_for_retry() { config_sync_readiness::mark_unready(); // spawn_event_sync claims InProgress before reconcile. - assert!(config_sync_readiness::claim_in_progress()); + let claim = config_sync_readiness::claim_in_progress(); + assert!(claim.is_some()); - // Barrier error: run_boot_barrier_enforcing calls mark_unready on Err. - config_sync_readiness::mark_unready(); + // 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 assert_eq!( config_sync_readiness::readiness_state(), @@ -190,11 +194,13 @@ fn test_barrier_failure_resets_to_unready_for_retry() { ); // Next flush tick must be able to claim for retry. + let retry_claim = config_sync_readiness::claim_in_progress(); assert!( - config_sync_readiness::claim_in_progress(), + retry_claim.is_some(), "after failure, next claim must succeed so the retry can run" ); + retry_claim.unwrap().resolve(); config_sync_readiness::mark_unready(); // cleanup } @@ -222,14 +228,15 @@ async fn test_inline_barrier_cannot_certify_readiness_before_reconcile() { let db_path = dir.path().join("retention.db"); // Step 1: spawn_event_sync claims InProgress before reconcile starts. + let claim = config_sync_readiness::claim_in_progress(); assert!( - config_sync_readiness::claim_in_progress(), + claim.is_some(), "spawn_event_sync must win the InProgress claim" ); // Step 2: flush tick fires and tries to claim — must be rejected. assert!( - !config_sync_readiness::claim_in_progress(), + config_sync_readiness::claim_in_progress().is_none(), "inline flush barrier must be rejected — InProgress already claimed" ); assert!( @@ -251,6 +258,7 @@ 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()); // Step 5: scope is Ready; flush runs but the row is blocked — 0 publish. @@ -281,9 +289,9 @@ async fn test_inline_barrier_cannot_certify_readiness_before_reconcile() { // ── (e) retry path: after failure, publish succeeds when scope becomes Ready ── // -// End-to-end: spawn_event_sync claims InProgress → barrier fails (mark_unready) -// → next tick re-claims InProgress → barrier succeeds (mark_ready) → row -// publishes on the following flush call. +// 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. #[tokio::test] async fn test_retry_after_barrier_failure_publishes_when_ready() { config_sync_readiness::mark_unready(); @@ -303,10 +311,14 @@ async fn test_retry_after_barrier_failure_publishes_when_ready() { let relay_url = spawn_stub_relay().await; *state.relay_url_override.lock().unwrap() = Some(relay_url); - // Phase 1: spawn_event_sync claims InProgress → barrier fails → Unready. - assert!(config_sync_readiness::claim_in_progress()); - // Simulate barrier error: run_boot_barrier_after_claim calls mark_unready. - config_sync_readiness::mark_unready(); + // Phase 1: spawn_event_sync claims InProgress → barrier fails → drop + // resets to Unready via RAII. + { + let claim = config_sync_readiness::claim_in_progress(); + assert!(claim.is_some()); + // Simulate barrier error: drop without resolve(). + drop(claim); + } // Latch is Unready — retry is possible. assert_eq!( @@ -316,10 +328,12 @@ async fn test_retry_after_barrier_failure_publishes_when_ready() { ); // Phase 2: next flush tick re-claims InProgress, barrier succeeds → Ready. + let retry_claim = config_sync_readiness::claim_in_progress(); assert!( - config_sync_readiness::claim_in_progress(), + retry_claim.is_some(), "after mark_unready, claim must succeed for retry" ); + retry_claim.unwrap().resolve(); config_sync_readiness::mark_ready(db_path.clone()); // Row is unblocked — flush must publish it now that scope is Ready. @@ -334,3 +348,95 @@ async fn test_retry_after_barrier_failure_publishes_when_ready() { config_sync_readiness::mark_unready(); // cleanup } + +// ── (f) apply_workspace window: flush claim between invalidation and reconcile ── +// +// Paul's bounce defect: between `mark_unready()` and `spawn_event_sync`'s +// inner `claim_in_progress()`, the flush loop could win the CAS and certify +// readiness against the pre-migration, pre-reconcile database state. Migrated +// or reconciled rows then publish unarbitrated. +// +// Fix: `apply_workspace` calls `force_claim_in_progress()` BEFORE migration, +// then passes the held claim to `spawn_event_sync_with_held_claim`. The flush +// loop sees `InProgress` throughout and cannot interleave. +// +// This test reproduces the race deterministically: +// 1. `apply_workspace` calls `force_claim_in_progress()` (InProgress from any state). +// 2. A flush tick fires immediately and calls `claim_in_progress()` — rejected. +// 3. Legacy migration runs and retains a row with `publish_blocked=false`. +// 4. Reconcile retains a second row. +// 5. Post-reconcile barrier gates both rows (sets `publish_blocked=true`). +// 6. Scope transitions to Ready. +// 7. Flush runs — both rows are blocked, 0 published. +#[tokio::test] +async fn test_apply_workspace_flush_cannot_interleave_before_reconcile() { + 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: apply_workspace atomically claims InProgress (preempting). + let claim = config_sync_readiness::force_claim_in_progress(); + assert!( + config_sync_readiness::is_in_progress(), + "force_claim must set InProgress" + ); + + // Step 2: flush tick fires and tries to claim — must be rejected. + assert!( + config_sync_readiness::claim_in_progress().is_none(), + "flush claim during apply_workspace window must be rejected" + ); + + // Step 3: legacy migration runs and retains a row. + { + let conn = open_retention_db(&db_path).expect("open db"); + retain_pending(&conn, &keys, "legacy-row"); + } + + // Step 4: reconcile retains a second row. + { + let conn = open_retention_db(&db_path).expect("open db"); + retain_pending(&conn, &keys, "reconciled-row"); + } + + // Step 5: post-reconcile barrier gates both rows. + { + let conn = open_retention_db(&db_path).expect("open db for barrier"); + set_publish_blocked(&conn, KIND_PERSONA, &pubkey, "legacy-row", true) + .expect("gate legacy row"); + set_publish_blocked(&conn, KIND_PERSONA, &pubkey, "reconciled-row", true) + .expect("gate reconciled row"); + } + + // Step 6: barrier completes — resolve claim and mark Ready. + claim.resolve(); + config_sync_readiness::mark_ready(db_path.clone()); + + // Step 7: flush runs — both rows are blocked, 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, + "migrated and reconciled rows gated by post-reconcile barrier must not publish" + ); + + let conn = open_retention_db(&db_path).expect("reopen db"); + for d_tag in ["legacy-row", "reconciled-row"] { + let row = get_retained_event(&conn, KIND_PERSONA, &pubkey, d_tag) + .unwrap() + .unwrap(); + assert!( + row.publish_blocked, + "{d_tag}: publish_blocked must persist after gating" + ); + assert!(row.pending_sync, "{d_tag}: gated row must stay pending"); + } + + config_sync_readiness::mark_unready(); // cleanup +}