jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-13 12:51:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(managed-agents): close apply_workspace Unready window and add RAII panic recovery

Browse Source 
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<ReadinessClaim> (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 <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
2026-07-30 21:18:19 -04:00
parent 7d096ae299
commit cf23ad6943
5 changed files with 328 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
desktop/src-tauri/src/commands/workspace.rs
View File

@@ -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) => {

55
desktop/src-tauri/src/event_sync.rs
View File

@@ -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;
});
}

43
desktop/src-tauri/src/managed_agents/config_barrier.rs
View File

@@ -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::<AppState>();
@@ -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).
}
}
}

171
desktop/src-tauri/src/managed_agents/config_sync_readiness.rs
View File

@@ -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<ReadinessClaim> {
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
}
}

138
desktop/src-tauri/src/managed_agents/persona_events/flush_barrier_tests.rs
View File

@@ -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
}