mirror of
https://github.com/block/buzz.git
synced 2026-11-01 06:51:17 +00:00
fix(managed-agents): add generation counter to ReadinessClaim to prevent stale-claim mutations
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 <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
parent
cf23ad6943
commit
6c2d26e884
@@ -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
|
||||
|
||||
@@ -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::<AppState>();
|
||||
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<BarrierOutcome, String> {
|
||||
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.
|
||||
|
||||
@@ -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<ReadinessState> = std::sync::Mutex::new(ReadinessState::Unready);
|
||||
static READINESS: std::sync::Mutex<LatchState> = 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<ReadinessState> =
|
||||
const { std::sync::Mutex::new(ReadinessState::Unready) };
|
||||
static READINESS: std::sync::Mutex<LatchState> =
|
||||
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<ReadinessClaim> {
|
||||
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<ReadinessState> {
|
||||
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<R>(f: impl FnOnce(&mut ReadinessState) -> R) -> Result<R, String> {
|
||||
fn with_lock<R>(f: impl FnOnce(&mut LatchState) -> R) -> Result<R, String> {
|
||||
READINESS
|
||||
.lock()
|
||||
.map(|mut g| f(&mut g))
|
||||
@@ -177,7 +233,7 @@ fn with_lock<R>(f: impl FnOnce(&mut ReadinessState) -> R) -> Result<R, String> {
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn with_lock<R>(f: impl FnOnce(&mut ReadinessState) -> R) -> Result<R, String> {
|
||||
fn with_lock<R>(f: impl FnOnce(&mut LatchState) -> R) -> Result<R, String> {
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user