mirror of
https://github.com/block/buzz.git
synced 2026-11-07 09:52:03 +00:00
fix(desktop): force-claim before scope resolution; propagate base-dir errors
IMPORTANT#1: move force_claim_in_progress() before active_retention_scope()
in both workspace.rs and identity.rs. On scope-resolution failure the claim
now drops via RAII → Unready, so the flush loop can retry. Previously a
transient scope failure after a key/relay swap left the stale Ready(old_path)
latch alive; claim_in_progress rejects from any Ready state, so publication
was permanently stopped for the session.
New test: test_scope_resolution_failure_after_force_claim_leaves_latch_unready_and_retryable
covers Ready(A) → swap → scope failure → Unready → retry claims and runs the
full transition for scope B.
IMPORTANT#2: replace let Ok(base_dir) = managed_agents_base_dir(app) else { return Ok(()) }
with let base_dir = managed_agents_base_dir(app)? in all three reconcile
wrappers (migrate_personas_to_events, migrate_teams_to_events,
reconcile_agents_to_events). Base-dir failures now propagate through
run_event_sync's Result; a failed leg cannot silently certify Ready.
New test: test_base_dir_failure_propagates_through_run_event_sync drives the
new run_event_sync_in_dir seam with a bad db_path, asserts Err returns and
scope stays Unready, then proves the full retry with a valid path enqueues
the row and certifies Ready.
MINOR: rewrite config_barrier.rs enforcement comment to state the real
invariant — managed_agents_store_lock does NOT freeze the readiness
generation; force_claim_in_progress takes only the readiness mutex. A
preempted force-claim sets InProgress on a new generation and its own
barrier (or a full flush retry) supplies the final gate writes.
Prose nits: fix flush_barrier_tests.rs:216 reference to deleted
run_boot_barrier; fix identity.rs comment that referenced is_ready_for
(now #[cfg(test)]) — flush uses readiness_state() directly.
Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
parent
2b442f9e3e
commit
c2f2f9a833
@@ -376,8 +376,16 @@ pub async fn import_identity(
|
||||
|
||||
// Enqueue the new owner's disk projections and run the boot barrier.
|
||||
// Without this, config publication silently stops until the next
|
||||
// restart: the flush loop checks `is_ready_for(db_path)` against the
|
||||
// OLD owner's scope, and the new owner's scope was never certified.
|
||||
// restart: the flush loop checks `readiness_state()` against the
|
||||
// active scope's db_path, and the new owner's scope was never certified.
|
||||
//
|
||||
// Force-claim BEFORE scope resolution: if resolution fails transiently,
|
||||
// the stale Ready(old_path) latch must not survive — it would cause
|
||||
// `flush_active_pending_events` to refuse every retry claim for the
|
||||
// session (claim_in_progress rejects from any Ready state, including
|
||||
// the wrong old-path one). Dropping the claim on failure leaves the
|
||||
// latch Unready so the flush loop can retry with the full transition
|
||||
// once scope resolution recovers.
|
||||
//
|
||||
// Mirrors `apply_workspace`'s force-claim sequence. The legacy-
|
||||
// migration step is skipped here: a stranded legacy row for the
|
||||
@@ -387,13 +395,13 @@ pub async fn import_identity(
|
||||
// pubkey — so the reconcile below is necessary to queue the new
|
||||
// owner's disk projections for the new scope.
|
||||
{
|
||||
let claim =
|
||||
crate::managed_agents::config_sync_readiness::force_claim_in_progress();
|
||||
match crate::managed_agents::retention::active_retention_scope(
|
||||
&app_handle,
|
||||
&state,
|
||||
) {
|
||||
Ok(scope) => {
|
||||
let claim =
|
||||
crate::managed_agents::config_sync_readiness::force_claim_in_progress();
|
||||
crate::event_sync::spawn_event_sync_with_held_claim(
|
||||
app_handle.clone(),
|
||||
scope.owner_keys,
|
||||
@@ -403,6 +411,10 @@ pub async fn import_identity(
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("buzz-desktop: import-identity: scope unavailable, config barrier skipped: {e}");
|
||||
// Claim drops here via RAII → latch becomes Unready; the
|
||||
// flush loop will retry with the full transition once scope
|
||||
// resolves.
|
||||
drop(claim);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -215,15 +215,15 @@ pub async fn apply_workspace(
|
||||
// Backfill this exact relay+owner scope only after the workspace has been
|
||||
// applied. Running at process boot would target the fallback relay and
|
||||
// collapse every community into one pending-event store.
|
||||
//
|
||||
// Force-claim BEFORE scope resolution: if resolution fails transiently,
|
||||
// the stale Ready(old_path) latch must not survive — it would cause
|
||||
// `flush_active_pending_events` to refuse every retry claim for the session.
|
||||
// Dropping the claim on failure leaves the latch Unready so the flush loop
|
||||
// can claim and run the full transition once scope resolution recovers.
|
||||
let claim = crate::managed_agents::config_sync_readiness::force_claim_in_progress();
|
||||
match crate::managed_agents::retention::active_retention_scope(&restore_app, &state) {
|
||||
Ok(scope) => {
|
||||
// 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
|
||||
@@ -238,6 +238,9 @@ pub async fn apply_workspace(
|
||||
}
|
||||
Err(error) => {
|
||||
eprintln!("buzz-desktop: scoped event-sync unavailable after workspace apply: {error}");
|
||||
// Claim drops here via RAII → latch becomes Unready; the flush
|
||||
// loop will retry with the full transition once scope resolves.
|
||||
drop(claim);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -119,9 +119,7 @@ pub fn migrate_personas_to_events(
|
||||
) -> Result<(), String> {
|
||||
use crate::managed_agents::managed_agents_base_dir;
|
||||
|
||||
let Ok(base_dir) = managed_agents_base_dir(app) else {
|
||||
return Ok(());
|
||||
};
|
||||
let base_dir = managed_agents_base_dir(app)?;
|
||||
|
||||
match migrate_personas_in_dir_at(&base_dir, keys, db_path) {
|
||||
Ok(0) => Ok(()),
|
||||
@@ -273,9 +271,7 @@ pub fn migrate_teams_to_events(
|
||||
) -> Result<(), String> {
|
||||
use crate::managed_agents::managed_agents_base_dir;
|
||||
|
||||
let Ok(base_dir) = managed_agents_base_dir(app) else {
|
||||
return Ok(());
|
||||
};
|
||||
let base_dir = managed_agents_base_dir(app)?;
|
||||
|
||||
match migrate_teams_in_dir_at(&base_dir, keys, db_path) {
|
||||
Ok(0) => Ok(()),
|
||||
@@ -376,6 +372,27 @@ fn migrate_teams_in_dir_at(
|
||||
Ok(migrated)
|
||||
}
|
||||
|
||||
/// Test seam: `run_event_sync` with an explicit `base_dir` instead of an
|
||||
/// `AppHandle`. Lets tests drive all three reconcile legs and the propagation
|
||||
/// of their `Result`s without a running Tauri application.
|
||||
///
|
||||
/// Used by `test_base_dir_failure_propagates_through_run_event_sync` to prove
|
||||
/// that a failing leg returns `Err`, the claim does NOT certify `Ready`, and
|
||||
/// a subsequent full retry enqueues the row.
|
||||
#[cfg(test)]
|
||||
pub(crate) fn run_event_sync_in_dir(
|
||||
base_dir: &Path,
|
||||
owner_keys: &nostr::Keys,
|
||||
db_path: &Path,
|
||||
) -> Result<(), String> {
|
||||
migrate_personas_in_dir_at(base_dir, owner_keys, db_path)?;
|
||||
migrate_teams_in_dir_at(base_dir, owner_keys, db_path)?;
|
||||
crate::managed_agents::reconcile::reconcile_agents_in_dir_at_test(
|
||||
base_dir, owner_keys, db_path,
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "event_sync_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
@@ -381,3 +381,85 @@ fn test_reconcile_failure_retry_with_full_transition_certifies_and_enqueues() {
|
||||
|
||||
config_sync_readiness::mark_unready(); // cleanup
|
||||
}
|
||||
|
||||
// ── Base-dir failure propagates through run_event_sync ────────────────────────
|
||||
//
|
||||
// Thufir IMPORTANT#2 coverage: all three reconcile wrappers formerly returned
|
||||
// `Ok(())` when `managed_agents_base_dir` failed (via `let Ok(...) else { return Ok(()) }`).
|
||||
// The fix replaces that with `let base_dir = ...?`, making the failure propagate.
|
||||
//
|
||||
// This test drives `run_event_sync_in_dir` (the AppHandle-free seam) with a
|
||||
// `db_path` pointing inside a non-existent directory so SQLite cannot open it,
|
||||
// forcing the first leg to return `Err`. Proves:
|
||||
// 1. `run_event_sync_in_dir` returns `Err` — a real reconcile failure, not a
|
||||
// manual `drop(claim)`.
|
||||
// 2. The claim does NOT call `complete` when the leg fails — scope stays
|
||||
// `Unready` (claim drops via RAII when the caller observes `Err`).
|
||||
// 3. A subsequent full retry with a valid `db_path` enqueues the row.
|
||||
#[test]
|
||||
fn test_base_dir_failure_propagates_through_run_event_sync() {
|
||||
use crate::managed_agents::{
|
||||
config_sync_readiness::{self, ReadinessState},
|
||||
retention::{get_retained_event, open_retention_db},
|
||||
};
|
||||
use buzz_core_pkg::kind::KIND_PERSONA;
|
||||
|
||||
config_sync_readiness::mark_unready();
|
||||
|
||||
let keys = nostr::Keys::generate();
|
||||
let pubkey = keys.public_key().to_hex();
|
||||
let base = tempfile::tempdir().unwrap();
|
||||
write_base_personas(base.path(), &one_persona());
|
||||
|
||||
// A db_path inside a non-existent directory — SQLite cannot open it.
|
||||
let bad_db_path = base.path().join("nonexistent_subdir").join("retention.db");
|
||||
// A valid db_path for the successful retry.
|
||||
let good_db_path = base.path().join("retention.db");
|
||||
|
||||
// Step 1: claim InProgress, run reconcile with bad db_path — leg fails.
|
||||
// The claim must NOT be completed (scope stays Unready via RAII drop).
|
||||
{
|
||||
let claim = config_sync_readiness::claim_in_progress().expect("must claim from Unready");
|
||||
|
||||
let result = run_event_sync_in_dir(base.path(), &keys, &bad_db_path);
|
||||
assert!(
|
||||
result.is_err(),
|
||||
"run_event_sync_in_dir must return Err when a reconcile leg fails"
|
||||
);
|
||||
|
||||
// Observe the error: claim drops here without complete() → Unready.
|
||||
let _ = result;
|
||||
drop(claim);
|
||||
}
|
||||
assert_eq!(
|
||||
config_sync_readiness::readiness_state(),
|
||||
Some(ReadinessState::Unready),
|
||||
"a failed reconcile leg must leave scope Unready (not Ready)"
|
||||
);
|
||||
|
||||
// Step 2: retry — full transition with valid db_path. The row must be queued.
|
||||
{
|
||||
let claim = config_sync_readiness::claim_in_progress()
|
||||
.expect("Unready scope must allow retry claim");
|
||||
|
||||
run_event_sync_in_dir(base.path(), &keys, &good_db_path)
|
||||
.expect("full retry with valid db_path must succeed");
|
||||
|
||||
claim.complete(good_db_path.clone());
|
||||
}
|
||||
assert!(
|
||||
config_sync_readiness::is_ready_for(&good_db_path),
|
||||
"full retry must certify Ready"
|
||||
);
|
||||
|
||||
let conn = open_retention_db(&good_db_path).expect("open db");
|
||||
let row = get_retained_event(&conn, KIND_PERSONA, &pubkey, "code-reviewer")
|
||||
.expect("db query")
|
||||
.expect("row must exist after successful retry");
|
||||
assert!(
|
||||
row.pending_sync,
|
||||
"disk edit must be pending after successful full retry"
|
||||
);
|
||||
|
||||
config_sync_readiness::mark_unready(); // cleanup
|
||||
}
|
||||
|
||||
@@ -251,9 +251,14 @@ async fn run_boot_barrier_for_scope(
|
||||
// Phase 4 — generation-gate the enforcement phase.
|
||||
//
|
||||
// `enforce_decision_pass` checks `claim.is_current()` while
|
||||
// `managed_agents_store_lock` is held (i.e., right here), so the
|
||||
// generation cannot change between the check and the `run_decision_pass`
|
||||
// write below. A preempted barrier returns `Abandoned` without writing.
|
||||
// `managed_agents_store_lock` is held (i.e., right here). Note: the store
|
||||
// lock does NOT freeze the readiness generation — `force_claim_in_progress`
|
||||
// takes only the separate readiness mutex, so it can advance the generation
|
||||
// during this interval. The real invariant is: a force-claim that preempts
|
||||
// after the check sets `InProgress` on the new generation and will run its
|
||||
// own barrier (or a full flush retry will) to supply the final gate writes.
|
||||
// A preempted barrier returns `Abandoned` without writing, which is safe
|
||||
// because the new owner's barrier is the authoritative one.
|
||||
enforce_decision_pass(claim, &conn, &owner_pubkey, &states)
|
||||
}
|
||||
|
||||
|
||||
@@ -464,4 +464,56 @@ mod tests {
|
||||
drop(force_claim);
|
||||
assert_eq!(readiness_state(), Some(ReadinessState::Unready));
|
||||
}
|
||||
|
||||
// ── (i) force-claim before scope resolution: scope failure → Unready ─────
|
||||
//
|
||||
// Covers the Thufir IMPORTANT#1 fix: `force_claim_in_progress` runs BEFORE
|
||||
// `active_retention_scope`. If scope resolution fails, the claim drops via
|
||||
// RAII and the latch lands `Unready`. Without this fix, the stale
|
||||
// `Ready(old_path)` would survive and `claim_in_progress` would reject every
|
||||
// flush retry for the session (it only accepts from `Unready`).
|
||||
//
|
||||
// Scenario: scope A Ready → swap → scope resolution fails →
|
||||
// Unready (via RAII drop) → retry from flush loop wins claim →
|
||||
// full transition runs for scope B → Ready(B).
|
||||
#[test]
|
||||
fn test_scope_resolution_failure_after_force_claim_leaves_latch_unready_and_retryable() {
|
||||
mark_unready();
|
||||
|
||||
let path_a = PathBuf::from("/scope/a.db");
|
||||
let path_b = PathBuf::from("/scope/b.db");
|
||||
|
||||
// Scope A is Ready before the swap.
|
||||
let claim_a = claim_in_progress().unwrap();
|
||||
claim_a.complete(path_a.clone());
|
||||
assert!(is_ready_for(&path_a), "scope A must be Ready before swap");
|
||||
|
||||
// Force-claim BEFORE scope resolution (the fix). Simulated scope
|
||||
// resolution failure: drop the claim without calling complete().
|
||||
{
|
||||
let _claim = force_claim_in_progress();
|
||||
// Scope resolution fails here — _claim drops via RAII.
|
||||
}
|
||||
assert_eq!(
|
||||
readiness_state(),
|
||||
Some(ReadinessState::Unready),
|
||||
"scope failure after force_claim must leave latch Unready, not Ready(old_path)"
|
||||
);
|
||||
assert!(
|
||||
!is_ready_for(&path_a),
|
||||
"stale Ready(A) must not survive after force-claim + drop"
|
||||
);
|
||||
|
||||
// The flush loop's Unready arm can now claim and run the full
|
||||
// transition for scope B.
|
||||
let retry_claim = claim_in_progress();
|
||||
assert!(
|
||||
retry_claim.is_some(),
|
||||
"Unready latch must accept retry claim for scope B"
|
||||
);
|
||||
retry_claim.unwrap().complete(path_b.clone());
|
||||
assert!(is_ready_for(&path_b), "full retry must certify Ready(B)");
|
||||
|
||||
mark_unready(); // cleanup
|
||||
}
|
||||
}
|
||||
|
||||
@@ -212,8 +212,8 @@ fn test_barrier_failure_resets_to_unready_for_retry() {
|
||||
// Scenario (reproduced deterministically using the process-global latch):
|
||||
//
|
||||
// 1. `spawn_event_sync` claims `InProgress` before reconcile starts.
|
||||
// 2. A flush tick fires and calls `claim_in_progress` (simulating
|
||||
// run_boot_barrier's CAS entry). It is rejected — InProgress is taken.
|
||||
// 2. A flush tick fires and calls `claim_in_progress` (simulating the flush
|
||||
// loop's CAS entry). It is rejected — InProgress is taken.
|
||||
// 3. Reconcile runs and retains a new row (publish_blocked=false).
|
||||
// 4. `run_boot_barrier_after_claim` runs the barrier: gates the stale row,
|
||||
// then marks Ready.
|
||||
|
||||
@@ -36,9 +36,7 @@ pub(crate) fn reconcile_agents_to_events(
|
||||
keys: &nostr::Keys,
|
||||
db_path: &Path,
|
||||
) -> Result<(), String> {
|
||||
let Ok(base_dir) = super::managed_agents_base_dir(app) else {
|
||||
return Ok(());
|
||||
};
|
||||
let base_dir = super::managed_agents_base_dir(app)?;
|
||||
|
||||
match reconcile_agents_in_dir_at(&base_dir, keys, db_path) {
|
||||
Ok(0) => Ok(()),
|
||||
@@ -71,6 +69,17 @@ pub(crate) fn reconcile_agents_in_dir(base_dir: &Path, keys: &nostr::Keys) -> Re
|
||||
reconcile_agents_in_dir_at(base_dir, keys, &base_dir.join("retention.db"))
|
||||
}
|
||||
|
||||
/// Test seam with an explicit `db_path`. Used by `event_sync::run_event_sync_in_dir`
|
||||
/// so tests can pass a controlled (including failing) db_path to all three reconcile legs.
|
||||
#[cfg(test)]
|
||||
pub(crate) fn reconcile_agents_in_dir_at_test(
|
||||
base_dir: &Path,
|
||||
keys: &nostr::Keys,
|
||||
db_path: &Path,
|
||||
) -> Result<u32, String> {
|
||||
reconcile_agents_in_dir_at(base_dir, keys, db_path)
|
||||
}
|
||||
|
||||
fn reconcile_agents_in_dir_at(
|
||||
base_dir: &Path,
|
||||
keys: &nostr::Keys,
|
||||
|
||||
Reference in New Issue
Block a user