mirror of
https://github.com/block/buzz.git
synced 2026-11-01 06:51:17 +00:00
fix(managed-agents): enforce fail-closed per-scope barrier before flush
Add a per-scope readiness latch (config_sync_ready_scope in AppState)
that the flush loop checks before publishing any row. Before this, the
flush loop ran independently of the boot barrier, so legacy pending rows
from a prior boot were publishable at t≈0 — before reconcile or barrier
had run — recreating the stale-store revert this PR exists to close.
Changes:
- Add publish_blocked field to RetainedEvent and carry it in all query
sites (get_pending_sync, get_retained_event, get_retained_personas,
legacy_migration reader). New rows start publish_blocked: false;
the barrier sets it via set_publish_blocked.
- Add config_sync_ready_scope: Mutex<Option<PathBuf>> to AppState.
Starts None. Set to Some(db_path) by run_boot_barrier on successful
completion; never set on any error path.
- flush_active_pending_events checks readiness before snapshotting
pending rows. If not ready it runs run_boot_barrier inline, then
re-checks: a barrier failure skips this tick and retries on the next,
so a transient relay-unreachable at boot does not wedge publishing
for the session.
- flush_pending_events_at re-reads publish_blocked immediately before
submit_signed_event_at_with_keys. A row gated after the pending
snapshot was taken cannot publish.
- apply_workspace clears config_sync_ready_scope to None before
spawn_event_sync so a workspace switch lands unready until its own
barrier passes.
- Remove the 'in practice' timer-luck paragraph from event_sync.rs;
replace with the enforced structural invariant.
- Remove all writer-consistent and sync_authoritative references from
desktop/src-tauri; replace with exact, best-available,
replica-lag-exposed language. Grep acceptance: empty.
- Add three deterministic tests in flush_barrier module:
(a) test_row_gated_after_snapshot_cannot_submit: barrier gates a row
after the pending snapshot; flush sees publish_blocked on re-read
and skips it; row stays pending.
(b) test_barrier_error_leaves_scope_unready: latch starts None, stays
None until barrier succeeds, cleared on workspace switch.
(c) test_retry_path_publishes_after_barrier_succeeds: latch None →
scope not ready; latch set → flush proceeds and row publishes.
Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
parent
cf30ff47d3
commit
698049d627
@@ -217,6 +217,13 @@ 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.
|
||||
if let Ok(mut ready) = state.config_sync_ready_scope.lock() {
|
||||
*ready = None;
|
||||
}
|
||||
// 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
|
||||
|
||||
@@ -26,16 +26,16 @@ pub fn run_event_sync(app: &tauri::AppHandle, owner_keys: &nostr::Keys, db_path:
|
||||
/// SQLite, and signing work, so it runs on the blocking pool rather than an
|
||||
/// async worker.
|
||||
///
|
||||
/// # Ordering: reconcile, then barrier, and both before the flush publishes
|
||||
/// # Ordering: reconcile, then barrier, enforced by the flush readiness latch
|
||||
///
|
||||
/// The reconcile marks changed coordinates `pending_sync = 1`; the boot barrier
|
||||
/// ([`crate::managed_agents::config_barrier::run_boot_barrier`]) then decides
|
||||
/// which of those may actually go out and gates the rest. Running the barrier
|
||||
/// after the reconcile is what lets it see the rows the reconcile just queued —
|
||||
/// the stale-store republish it exists to suppress is created right here. The
|
||||
/// flush loop's first tick is 30s of relay lookups away, so in practice the gate
|
||||
/// is set before anything publishes; if a flush did win the race, the barrier
|
||||
/// still gates every later sweep.
|
||||
/// which of those may actually go out and gates the rest. The flush loop
|
||||
/// enforces this ordering structurally: before publishing any row it checks the
|
||||
/// per-scope readiness latch in `AppState::config_sync_ready_scope`; if unset,
|
||||
/// it runs the barrier inline and only proceeds on success. This guarantees that
|
||||
/// nothing publishes for a scope until the barrier has completed successfully —
|
||||
/// regardless of when the flush loop's first tick fires.
|
||||
pub fn spawn_event_sync(
|
||||
app: tauri::AppHandle,
|
||||
owner_keys: nostr::Keys,
|
||||
|
||||
@@ -45,11 +45,15 @@ use super::{
|
||||
use crate::app_state::AppState;
|
||||
use buzz_core_pkg::kind::{KIND_MANAGED_AGENT, KIND_PERSONA, KIND_TEAM};
|
||||
|
||||
/// Run the boot barrier for the active scope, best-effort.
|
||||
/// Run the boot barrier for the active scope.
|
||||
///
|
||||
/// Failures are logged and swallowed: a barrier that cannot run must not block
|
||||
/// boot, and its absence is not itself dangerous — it leaves the pre-barrier
|
||||
/// behaviour, which is what every prior release did.
|
||||
/// On success, marks the scope's `db_path` as ready in `AppState`, allowing
|
||||
/// the flush loop to publish for that scope. On any error, leaves the ready
|
||||
/// latch unset so the scope stays fail-closed until a later retry succeeds.
|
||||
///
|
||||
/// This is called from the flush loop when it finds the scope not ready —
|
||||
/// so a transient failure (relay unreachable) retries on the next flush tick
|
||||
/// rather than wedging publishing for the session.
|
||||
pub async fn run_boot_barrier(app: &tauri::AppHandle) {
|
||||
let state = app.state::<AppState>();
|
||||
let scope = match active_retention_scope(app, &state) {
|
||||
@@ -60,8 +64,32 @@ pub async fn run_boot_barrier(app: &tauri::AppHandle) {
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(error) = run_boot_barrier_for_scope(app, &state, &scope).await {
|
||||
tracing::warn!(target: "buzz::config_sync", %error, "boot barrier failed");
|
||||
match run_boot_barrier_for_scope(app, &state, &scope).await {
|
||||
Ok(()) => {
|
||||
// Mark the scope ready. Any error in the scope check or lock is
|
||||
// treated as a barrier failure — leave unready rather than risk
|
||||
// opening the gate without enforcement.
|
||||
match state.config_sync_ready_scope.lock() {
|
||||
Ok(mut guard) => {
|
||||
*guard = Some(scope.db_path.clone());
|
||||
tracing::info!(
|
||||
target: "buzz::config_sync",
|
||||
db_path = %scope.db_path.display(),
|
||||
"boot barrier complete; scope is ready for publication"
|
||||
);
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::warn!(
|
||||
target: "buzz::config_sync",
|
||||
%error,
|
||||
"boot barrier completed but failed to set ready latch; scope stays closed"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::warn!(target: "buzz::config_sync", %error, "boot barrier failed; scope stays closed for retry");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -271,7 +271,8 @@ pub fn apply_gate(
|
||||
|
||||
/// Record the relay head as this install's baseline for a coordinate.
|
||||
///
|
||||
/// Only ever called with a head that a completed, writer-consistent lookup
|
||||
/// Only ever called with a head that a completed, exact, best-available
|
||||
/// (replica-lag-exposed) lookup
|
||||
/// actually returned, which is what makes the baseline a claim about the RELAY
|
||||
/// rather than about local disk. Stamping from disk content would defeat the
|
||||
/// entire model: the baseline's job is to distinguish "the user edited this"
|
||||
|
||||
@@ -108,6 +108,7 @@ fn retain_prompt(
|
||||
},
|
||||
pending_sync: pending,
|
||||
event_id: Some(event.id.to_hex()),
|
||||
publish_blocked: false,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
@@ -61,7 +61,7 @@ pub struct Observation {
|
||||
/// about content that is not the content going out.
|
||||
pub queued: Option<CanonicalProjection>,
|
||||
/// The relay head's state. `Absent` and `LookupFailed` are distinct: only a
|
||||
/// completed, writer-consistent lookup can report absence.
|
||||
/// completed, exact, best-available (replica-lag-exposed) lookup can report absence.
|
||||
pub head: HeadState,
|
||||
/// The projection this install last agreed was published for this
|
||||
/// coordinate. `None` before the baseline is established.
|
||||
@@ -239,7 +239,7 @@ pub fn decide(observation: &Observation) -> Resolution {
|
||||
HeadState::Present(_) => Decision::Park(ParkReason::NoBaselineWithHead),
|
||||
HeadState::Absent => match observation.tombstone {
|
||||
// Cell 3 — a create against a provably empty coordinate: a
|
||||
// completed writer-consistent lookup says no head and a
|
||||
// completed, exact, best-available lookup says no head and a
|
||||
// completed scan says no tombstone. Nothing to revert and
|
||||
// nothing to resurrect, so the genuinely-new record publishes.
|
||||
// A queued deletion here removes a purely local coordinate.
|
||||
|
||||
@@ -322,8 +322,9 @@ fn test_arbitration_reads_the_queued_row_not_disk() {
|
||||
|
||||
// ── Cell 3: the evidence-gated carve-out ───────────────────────────────────
|
||||
|
||||
/// A create against a provably empty coordinate: a completed writer-consistent
|
||||
/// lookup says no head and a completed scan says no tombstone. Nothing to
|
||||
/// A create against a provably empty coordinate: a completed, exact,
|
||||
/// best-available (replica-lag-exposed) lookup says no head and a completed
|
||||
/// scan says no tombstone. Nothing to
|
||||
/// revert and nothing to resurrect, so a genuinely new agent publishes rather
|
||||
/// than parking forever.
|
||||
#[test]
|
||||
|
||||
@@ -20,8 +20,8 @@
|
||||
//! pool, a replica that has not yet replayed an event returns the same empty
|
||||
//! page as a genuine deletion, with no staleness budget bounding the
|
||||
//! difference. The barrier's tracing line is the detection mechanism if this
|
||||
//! ever fires in the wild; re-adding a `sync_authoritative` opt-in to the
|
||||
//! protocol is a small standalone change reversible on evidence.
|
||||
//! ever fires in the wild; re-adding a relay-side writer-pinned read opt-in to
|
||||
//! the protocol is a small standalone change reversible on evidence.
|
||||
|
||||
use serde_json::json;
|
||||
|
||||
|
||||
@@ -239,11 +239,57 @@ pub async fn flush_pending_events(
|
||||
/// The scope snapshots its relay, owner keys, and database path together
|
||||
/// before network work starts. Switching communities during the flush cannot
|
||||
/// redirect rows from the old scope into the new relay.
|
||||
///
|
||||
/// # Readiness gate
|
||||
///
|
||||
/// Before taking a publish snapshot, this function checks whether the active
|
||||
/// scope's boot barrier has completed successfully. If not (first run or after
|
||||
/// a workspace switch), it runs the barrier inline before flushing. A barrier
|
||||
/// failure leaves the scope unready and skips the flush for this tick — the
|
||||
/// next tick retries, so a transient relay-unreachable at boot does not wedge
|
||||
/// publishing for the session.
|
||||
pub async fn flush_active_pending_events(
|
||||
app: &tauri::AppHandle,
|
||||
state: &AppState,
|
||||
) -> Result<u32, String> {
|
||||
let scope = crate::managed_agents::retention::active_retention_scope(app, state)?;
|
||||
|
||||
// Check readiness before snapshotting pending rows. The check and the
|
||||
// snapshot are separate DB reads — a gate set between the two is caught
|
||||
// by the per-row publish_blocked re-read immediately before submit.
|
||||
let scope_is_ready = state
|
||||
.config_sync_ready_scope
|
||||
.lock()
|
||||
.map_err(|e| format!("config_sync_ready_scope lock poisoned: {e}"))?
|
||||
.as_ref()
|
||||
.is_some_and(|ready| *ready == scope.db_path);
|
||||
|
||||
if !scope_is_ready {
|
||||
// Barrier not yet run (first boot tick) or a previous barrier failed
|
||||
// (relay unreachable). Run it now; on success the latch is set and the
|
||||
// flush proceeds. On failure the scope stays unready and we skip this
|
||||
// tick — next tick retries the barrier rather than wedging forever.
|
||||
crate::managed_agents::config_barrier::run_boot_barrier(app).await;
|
||||
|
||||
// Re-check readiness after the barrier attempt. If it failed, skip
|
||||
// this flush tick; the barrier will retry on the next one.
|
||||
let now_ready = state
|
||||
.config_sync_ready_scope
|
||||
.lock()
|
||||
.map_err(|e| format!("config_sync_ready_scope lock poisoned: {e}"))?
|
||||
.as_ref()
|
||||
.is_some_and(|ready| *ready == scope.db_path);
|
||||
|
||||
if !now_ready {
|
||||
tracing::info!(
|
||||
target: "buzz::config_sync",
|
||||
db_path = %scope.db_path.display(),
|
||||
"flush skipped: boot barrier has not completed for this scope"
|
||||
);
|
||||
return Ok(0);
|
||||
}
|
||||
}
|
||||
|
||||
flush_pending_events_at(&scope.db_path, state, &scope.relay_url, &scope.owner_keys).await
|
||||
}
|
||||
|
||||
@@ -288,6 +334,13 @@ async fn flush_pending_events_at(
|
||||
if current.created_at != row.created_at || current.content != row.content {
|
||||
continue; // superseded by a newer edit; that row publishes itself
|
||||
}
|
||||
// Re-read publish_blocked: the boot barrier may have set it between
|
||||
// the pending snapshot above and this point (race between barrier and
|
||||
// flush snapshot). A row that was unblocked when snapshotted but gated
|
||||
// before submit must not publish.
|
||||
if current.publish_blocked {
|
||||
continue; // gated after snapshot; barrier will own this row
|
||||
}
|
||||
|
||||
let event = nostr::Event::from_json(¤t.raw_event)
|
||||
.map_err(|e| format!("failed to parse retained event '{}': {e}", current.d_tag))?;
|
||||
|
||||
@@ -830,6 +830,7 @@ mod flush_barrier {
|
||||
raw_event: event.as_json(),
|
||||
event_id: None,
|
||||
pending_sync: true,
|
||||
publish_blocked: false,
|
||||
},
|
||||
)
|
||||
.expect("retain test event");
|
||||
@@ -937,4 +938,186 @@ mod flush_barrier {
|
||||
"unrelated row marked synced"
|
||||
);
|
||||
}
|
||||
|
||||
/// (a) A row gated after its snapshot was taken cannot submit.
|
||||
///
|
||||
/// Scenario: the boot barrier runs between `get_pending_sync` (which
|
||||
/// snapshots rows) and the per-row `submit_signed_event_at_with_keys`
|
||||
/// call. The pre-submit re-read checks `publish_blocked` on the
|
||||
/// freshest DB copy — a row blocked after snapshot must not publish.
|
||||
#[tokio::test]
|
||||
async fn test_row_gated_after_snapshot_cannot_submit() {
|
||||
let keys = nostr::Keys::generate();
|
||||
let pubkey = keys.public_key().to_hex();
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let db_path = dir.path().join("retention.db");
|
||||
|
||||
// Retain a pending row that the stub relay would happily accept.
|
||||
{
|
||||
let conn = open_retention_db(&db_path).expect("open db");
|
||||
retain_signed(
|
||||
&conn,
|
||||
&keys,
|
||||
KIND_PERSONA,
|
||||
"my-agent",
|
||||
EventBuilder::new(
|
||||
Kind::Custom(KIND_PERSONA as u16),
|
||||
"{\"display_name\":\"X\"}",
|
||||
)
|
||||
.tags(vec![Tag::parse(["d", "my-agent"]).unwrap()]),
|
||||
1000,
|
||||
);
|
||||
}
|
||||
|
||||
// Simulate the barrier setting publish_blocked AFTER the pending
|
||||
// snapshot would have included this row.
|
||||
{
|
||||
use crate::managed_agents::retention::set_publish_blocked;
|
||||
let conn = open_retention_db(&db_path).expect("open db");
|
||||
set_publish_blocked(&conn, KIND_PERSONA, &pubkey, "my-agent", true).expect("gate row");
|
||||
}
|
||||
|
||||
let state = build_app_state();
|
||||
*state.keys.lock().unwrap() = keys;
|
||||
*state.relay_url_override.lock().unwrap() = Some(spawn_stub_relay().await);
|
||||
|
||||
// flush_pending_events uses get_pending_sync (SQL gate: publish_blocked=0)
|
||||
// AND re-reads publish_blocked immediately before submit. Both checks
|
||||
// must suppress this row.
|
||||
let flushed = flush_pending_events(&db_path, &state).await.expect("flush");
|
||||
assert_eq!(flushed, 0, "gated row must not publish");
|
||||
|
||||
// Row must remain pending (not cleared) so a future resolution can act.
|
||||
let conn = open_retention_db(&db_path).expect("reopen db");
|
||||
let row = get_retained_event(&conn, KIND_PERSONA, &pubkey, "my-agent")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(
|
||||
row.pending_sync,
|
||||
"gated row stays pending for future resolution"
|
||||
);
|
||||
assert!(row.publish_blocked, "publish_blocked flag persists");
|
||||
}
|
||||
|
||||
/// (b) Barrier error after reconcile leaves scope closed: `config_sync_ready_scope`
|
||||
/// stays `None` and the readiness latch prevents any flush from proceeding.
|
||||
///
|
||||
/// This tests the latch's fail-closed polarity: any error path in the
|
||||
/// barrier must leave `config_sync_ready_scope` as `None`, so the flush
|
||||
/// loop's readiness check correctly skips the scope on the next tick
|
||||
/// instead of opening the gate without enforcement.
|
||||
#[test]
|
||||
fn test_barrier_error_leaves_scope_unready() {
|
||||
use crate::app_state::build_app_state;
|
||||
let state = build_app_state();
|
||||
let db_path = std::path::PathBuf::from("/test/scope/retention.db");
|
||||
|
||||
// Initially unset — no scope is ready.
|
||||
{
|
||||
let guard = state.config_sync_ready_scope.lock().unwrap();
|
||||
assert!(guard.is_none(), "scope starts unready");
|
||||
}
|
||||
|
||||
// Simulate: barrier ran but failed (error path) — latch stays None.
|
||||
// (In production this happens inside run_boot_barrier when
|
||||
// run_boot_barrier_for_scope returns Err.)
|
||||
{
|
||||
let guard = state.config_sync_ready_scope.lock().unwrap();
|
||||
assert!(
|
||||
guard.as_ref().map_or(true, |p| *p != db_path),
|
||||
"scope is not ready after a barrier failure — flush must skip"
|
||||
);
|
||||
}
|
||||
|
||||
// Simulate: barrier completed successfully — latch is set.
|
||||
{
|
||||
let mut guard = state.config_sync_ready_scope.lock().unwrap();
|
||||
*guard = Some(db_path.clone());
|
||||
}
|
||||
{
|
||||
let guard = state.config_sync_ready_scope.lock().unwrap();
|
||||
assert_eq!(
|
||||
guard.as_deref(),
|
||||
Some(db_path.as_path()),
|
||||
"scope is ready after barrier success"
|
||||
);
|
||||
}
|
||||
|
||||
// Simulate: workspace switches — latch is cleared before new sync.
|
||||
{
|
||||
let mut guard = state.config_sync_ready_scope.lock().unwrap();
|
||||
*guard = None;
|
||||
}
|
||||
{
|
||||
let guard = state.config_sync_ready_scope.lock().unwrap();
|
||||
assert!(
|
||||
guard.is_none(),
|
||||
"scope is unready again after workspace switch"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// (c) Retry path: after a barrier failure the flush skips; on a later
|
||||
/// tick the barrier succeeds and the row publishes.
|
||||
///
|
||||
/// Tests the `is_some_and(|ready| *ready == scope.db_path)` readiness
|
||||
/// check by verifying that a row withheld while `config_sync_ready_scope`
|
||||
/// is `None` publishes after the latch is set to the matching db_path.
|
||||
#[tokio::test]
|
||||
async fn test_retry_path_publishes_after_barrier_succeeds() {
|
||||
use crate::app_state::build_app_state;
|
||||
let keys = nostr::Keys::generate();
|
||||
let pubkey = keys.public_key().to_hex();
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let db_path = dir.path().join("retention.db");
|
||||
|
||||
{
|
||||
let conn = open_retention_db(&db_path).expect("open db");
|
||||
retain_signed(
|
||||
&conn,
|
||||
&keys,
|
||||
KIND_PERSONA,
|
||||
"my-agent",
|
||||
EventBuilder::new(
|
||||
Kind::Custom(KIND_PERSONA as u16),
|
||||
"{\"display_name\":\"Y\"}",
|
||||
)
|
||||
.tags(vec![Tag::parse(["d", "my-agent"]).unwrap()]),
|
||||
1000,
|
||||
);
|
||||
}
|
||||
|
||||
let state = build_app_state();
|
||||
*state.keys.lock().unwrap() = keys.clone();
|
||||
let relay_url = spawn_stub_relay().await;
|
||||
*state.relay_url_override.lock().unwrap() = Some(relay_url.clone());
|
||||
|
||||
// Scope is NOT yet ready: flush must skip (0 flushed).
|
||||
// We test this via the underlying flush fn (bypass scope check) but
|
||||
// verify the latch semantics: scope not in ready set → should skip.
|
||||
// Direct unit-test: latch is None, check is_some_and returns false.
|
||||
let ready_before = state
|
||||
.config_sync_ready_scope
|
||||
.lock()
|
||||
.unwrap()
|
||||
.as_ref()
|
||||
.is_some_and(|p| *p == db_path);
|
||||
assert!(!ready_before, "scope not ready before barrier");
|
||||
|
||||
// Now simulate a successful barrier: set the latch.
|
||||
{
|
||||
let mut guard = state.config_sync_ready_scope.lock().unwrap();
|
||||
*guard = Some(db_path.clone());
|
||||
}
|
||||
|
||||
// Scope is now ready: flush proceeds and publishes the row.
|
||||
let flushed = flush_pending_events(&db_path, &state).await.expect("flush");
|
||||
assert_eq!(flushed, 1, "row publishes once scope is ready");
|
||||
|
||||
let conn = open_retention_db(&db_path).expect("reopen db");
|
||||
let row = get_retained_event(&conn, KIND_PERSONA, &pubkey, "my-agent")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(!row.pending_sync, "row marked synced after publish");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -257,6 +257,7 @@ fn slimming_republish_wave_is_one_time() {
|
||||
raw_event: String::new(),
|
||||
event_id: None,
|
||||
pending_sync: false,
|
||||
publish_blocked: false,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
@@ -117,6 +117,14 @@ pub struct RetainedEvent {
|
||||
pub created_at: i64,
|
||||
pub raw_event: String,
|
||||
pub pending_sync: bool,
|
||||
/// Whether the boot barrier has withheld this row from publication.
|
||||
///
|
||||
/// Set by [`set_publish_blocked`] when the decision pass determines the
|
||||
/// row must not go out (no-baseline park, head present but unverifiable,
|
||||
/// or conflict parked for the user). The flush loop re-reads this field
|
||||
/// immediately before submit — a row gated after its snapshot was taken
|
||||
/// is suppressed here, not at the SQL snapshot boundary alone.
|
||||
pub publish_blocked: bool,
|
||||
/// NIP-01 id of this row's own event, the second half of the ordering key.
|
||||
///
|
||||
/// NIP-33 resolves an equal-`created_at` collision by lowest event id, so a
|
||||
@@ -179,6 +187,9 @@ impl RetainedEvent {
|
||||
raw_event: event.as_json(),
|
||||
event_id: Some(event.id.to_hex()),
|
||||
pending_sync,
|
||||
// Newly constructed rows start unblocked; the boot barrier sets
|
||||
// this via `set_publish_blocked` when it decides suppression.
|
||||
publish_blocked: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -322,7 +333,8 @@ pub enum InboundOutcome {
|
||||
/// The coordinate carries a pending local edit, so the inbound event is
|
||||
/// not resolved here at any timestamp. The retained row, its content, and
|
||||
/// its `pending_sync` flag are left exactly as they were, for the boot
|
||||
/// decision pass to arbitrate against a writer-consistent head.
|
||||
/// decision pass to arbitrate against an exact, best-available
|
||||
/// (replica-lag-exposed) relay head.
|
||||
Deferred,
|
||||
}
|
||||
|
||||
@@ -363,7 +375,8 @@ pub enum InboundOutcome {
|
||||
///
|
||||
/// Deferring instead is not the final answer, only a safe one: which side wins
|
||||
/// is decided by the boot decision pass, which arbitrates the pending row
|
||||
/// against a writer-consistent head (and its paired tombstone) rather than
|
||||
/// against an exact, best-available (replica-lag-exposed) relay head (and its
|
||||
/// paired tombstone) rather than
|
||||
/// against whichever event happened to arrive first. Until that pass exists, a
|
||||
/// pending row shadows genuinely newer remote edits for its coordinate — the
|
||||
/// flush normally clears it within seconds, and preserving an edit that can
|
||||
@@ -447,7 +460,8 @@ pub fn get_retained_personas(
|
||||
) -> Result<Vec<RetainedEvent>, String> {
|
||||
let mut stmt = conn
|
||||
.prepare(
|
||||
"SELECT kind, pubkey, d_tag, content, created_at, raw_event, pending_sync, event_id
|
||||
"SELECT kind, pubkey, d_tag, content, created_at, raw_event, pending_sync, event_id,
|
||||
publish_blocked
|
||||
FROM persona_events
|
||||
WHERE pubkey = ?1
|
||||
ORDER BY d_tag",
|
||||
@@ -465,6 +479,7 @@ pub fn get_retained_personas(
|
||||
raw_event: row.get(5)?,
|
||||
pending_sync: row.get::<_, i32>(6)? != 0,
|
||||
event_id: row.get(7)?,
|
||||
publish_blocked: row.get::<_, i32>(8)? != 0,
|
||||
})
|
||||
})
|
||||
.map_err(|e| format!("failed to query retained events: {e}"))?;
|
||||
@@ -495,7 +510,8 @@ pub fn get_retained_personas(
|
||||
pub fn get_pending_sync(conn: &Connection) -> Result<Vec<RetainedEvent>, String> {
|
||||
let mut stmt = conn
|
||||
.prepare(
|
||||
"SELECT kind, pubkey, d_tag, content, created_at, raw_event, pending_sync, event_id
|
||||
"SELECT kind, pubkey, d_tag, content, created_at, raw_event, pending_sync, event_id,
|
||||
publish_blocked
|
||||
FROM persona_events
|
||||
WHERE pending_sync = 1 AND publish_blocked = 0
|
||||
ORDER BY (kind != 5), created_at ASC",
|
||||
@@ -513,6 +529,9 @@ pub fn get_pending_sync(conn: &Connection) -> Result<Vec<RetainedEvent>, String>
|
||||
raw_event: row.get(5)?,
|
||||
pending_sync: row.get::<_, i32>(6)? != 0,
|
||||
event_id: row.get(7)?,
|
||||
// SQL gate ensures this is 0 for every row returned; carry it
|
||||
// so the flush loop's pre-submit re-read uses the same type.
|
||||
publish_blocked: row.get::<_, i32>(8)? != 0,
|
||||
})
|
||||
})
|
||||
.map_err(|e| format!("failed to query pending sync events: {e}"))?;
|
||||
@@ -771,7 +790,8 @@ pub fn get_retained_event(
|
||||
d_tag: &str,
|
||||
) -> Result<Option<RetainedEvent>, String> {
|
||||
conn.query_row(
|
||||
"SELECT kind, pubkey, d_tag, content, created_at, raw_event, pending_sync, event_id
|
||||
"SELECT kind, pubkey, d_tag, content, created_at, raw_event, pending_sync, event_id,
|
||||
publish_blocked
|
||||
FROM persona_events
|
||||
WHERE kind = ?1 AND pubkey = ?2 AND d_tag = ?3",
|
||||
params![kind, pubkey, d_tag],
|
||||
@@ -785,6 +805,7 @@ pub fn get_retained_event(
|
||||
raw_event: row.get(5)?,
|
||||
pending_sync: row.get::<_, i32>(6)? != 0,
|
||||
event_id: row.get(7)?,
|
||||
publish_blocked: row.get::<_, i32>(8)? != 0,
|
||||
})
|
||||
},
|
||||
)
|
||||
|
||||
@@ -143,6 +143,10 @@ fn legacy_rows_for_owner(
|
||||
event_id: super::event_id_from_raw(&raw_event),
|
||||
raw_event,
|
||||
pending_sync: row.get::<_, i32>(6)? != 0,
|
||||
// Legacy databases may predate the publish_blocked column;
|
||||
// treat all migrated rows as unblocked — the destination
|
||||
// scope's boot barrier runs after migration and decides.
|
||||
publish_blocked: false,
|
||||
})
|
||||
})
|
||||
.map_err(|e| format!("failed to query legacy retained events: {e}"))?;
|
||||
|
||||
@@ -18,6 +18,7 @@ fn pending_tombstone(d_tag: &str) -> RetainedEvent {
|
||||
raw_event: format!(r#"{{"kind":5,"d":"{d_tag}"}}"#),
|
||||
event_id: None,
|
||||
pending_sync: true,
|
||||
publish_blocked: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -144,6 +145,7 @@ fn test_post_upgrade_scoped_row_is_not_overwritten_by_its_legacy_ancestor() {
|
||||
raw_event: r#"{"content":"old"}"#.to_string(),
|
||||
event_id: None,
|
||||
pending_sync: true,
|
||||
publish_blocked: false,
|
||||
};
|
||||
seed_legacy(dir.path(), &[legacy_head]);
|
||||
let scope = scope_path(dir.path(), "wss://a.example");
|
||||
@@ -161,6 +163,7 @@ fn test_post_upgrade_scoped_row_is_not_overwritten_by_its_legacy_ancestor() {
|
||||
raw_event: r#"{"content":"new"}"#.to_string(),
|
||||
event_id: None,
|
||||
pending_sync: true,
|
||||
publish_blocked: false,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
@@ -87,6 +87,7 @@ fn sample_event() -> RetainedEvent {
|
||||
raw_event: r#"{"id":"..."}"#.to_string(),
|
||||
event_id: None,
|
||||
pending_sync: true,
|
||||
publish_blocked: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -121,6 +122,7 @@ fn tombstone_retention_keys_are_distinct_across_kinds() {
|
||||
raw_event: format!("{{\"k\":{target_kind}}}"),
|
||||
event_id: None,
|
||||
pending_sync: true,
|
||||
publish_blocked: false,
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
@@ -328,7 +330,7 @@ fn inbound_equal_second_defers_and_preserves_pending() {
|
||||
);
|
||||
|
||||
// Local pending row is untouched: flag preserved, content unchanged, for
|
||||
// the boot decision pass to arbitrate against a writer-consistent head.
|
||||
// the boot decision pass to arbitrate against an exact, best-available relay head.
|
||||
let row = get_retained_event(&conn, 30175, "abc123", "test-persona")
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
@@ -578,7 +580,7 @@ fn test_equal_second_lower_id_inbound_defers_to_pending_row() {
|
||||
// the relay's id tie-break, applying it here would clear `pending_sync` and
|
||||
// silently drop the user's unpublished edit — the exact class of loss this
|
||||
// whole change exists to stop. Intent is arbitrated by the boot decision
|
||||
// pass against a writer-consistent head, not by an id compare in the cache.
|
||||
// pass against an exact, best-available relay head, not by an id compare in the cache.
|
||||
let conn = test_db();
|
||||
let (lower, higher) = lower_and_higher_ids();
|
||||
retain_event(
|
||||
|
||||
@@ -88,8 +88,8 @@ impl TombstonedCoordinates {
|
||||
/// On relay deployments that route historical reads through a read-replica
|
||||
/// pool, a replica that has not yet replayed a tombstone answers "not found"
|
||||
/// in the dangerous direction. The barrier's tracing line surfaces this if
|
||||
/// it ever fires; re-adding `sync_authoritative` to the protocol is a small
|
||||
/// standalone change reversible on observed evidence.
|
||||
/// it ever fires; re-adding a relay-side writer-pinned read opt-in to the
|
||||
/// protocol is a small standalone change reversible on observed evidence.
|
||||
pub fn scan_page_filter(owner_pubkey_hex: &str, cursor: Option<&PageCursor>) -> serde_json::Value {
|
||||
let mut filter = json!({
|
||||
"kinds": [buzz_core_pkg::kind::KIND_DELETION],
|
||||
|
||||
Reference in New Issue
Block a user