jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(push): stop backfilling messages on lease activation (#8178)

Browse Source 
Remove the 120-second message scan from push lease activation and
renewal. Subsequent eligible messages continue through the existing
database trigger. The lease protocol, schema, trigger, and advisory-lock
order remain unchanged.

Related enqueue-gating work: #2145. No duplicate open PR or issue found.

### Validation

- All 18 PostgreSQL push persistence tests pass against isolated
desired-state databases after review changes.
- The existing concurrency test verifies activation waits for an
in-flight insert and a subsequent message queues once. The retention
fixture activates its lease before inserting its message.
- Formatting and PostgreSQL test discovery pass. The earlier full local
run passed repository lint/static checks, native desktop pre-push
checks, and every non-ACP integration step.
- Broad local checks hit unrelated ACP idle/keepalive timing failures. A
serial retry passed two affected tests but still failed the keepalive
deadline test. The pre-push hook was skipped with author authorization
after its other checks passed.

---------

Signed-off-by: Tom Brow <tomb@block.xyz>
This commit is contained in:
Tom Brow
2026-10-08 17:34:41 -07:00
committed by GitHub
parent acf1ede202
commit 4db7bb0e7f
1 changed files with 21 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
crates/buzz-db/src/store/push.rs
View File

@@ -49,40 +49,6 @@ async fn acquire_push_gate_lock(tx: &mut sqlx::PgConnection, community: Communit
Ok(())
}
/// Recovery window for the activation backfill: recent events that the gate
/// legitimately skipped (no eligible lease at their commit time) are enqueued
/// when a lease activates, so a user who registers moments after a message
/// still gets woken. Product coverage only — the advisory-lock total order is
/// what makes the gate correct; see `PUSH_GATE_LOCK_NAMESPACE`.
const PUSH_GATE_BACKFILL_SECS: i64 = 120;
/// Enqueue match jobs for recent gate-skipped events. MUST run inside the same
/// transaction that holds the exclusive push-gate lock: after this commit,
/// every event is either backfilled here or ordered after the activation and
/// enqueued by the trigger — running it post-commit would reopen the gap.
/// Keyed on relay `received_at` (not author-controlled `created_at`); the kind
/// list mirrors the trigger allowlist; `ON CONFLICT DO NOTHING` dedups against
/// rows the trigger already enqueued.
async fn backfill_push_match_jobs(
tx: &mut sqlx::PgConnection,
community: CommunityId,
) -> Result<()> {
sqlx::query(
"INSERT INTO push_match_queue (community_id, event_id) \
SELECT community_id, id FROM events \
WHERE community_id = $1 \
AND kind IN (9, 40002, 45001, 45003) \
AND deleted_at IS NULL \
AND received_at > now() - make_interval(secs => $2) \
ON CONFLICT DO NOTHING",
)
.bind(community.as_uuid())
.bind(PUSH_GATE_BACKFILL_SECS as f64)
.execute(&mut *tx)
.await?;
Ok(())
}
/// Maximum claims for a malformed matcher job before it is discarded.
pub const MAX_MATCH_ATTEMPTS: i32 = 8;
@@ -429,9 +395,6 @@ pub async fn accept_lease_event(
}
return Err(error.into());
}
if is_active {
backfill_push_match_jobs(tx.conn(), community).await?;
}
tx.commit().await?;
Ok(AcceptLeaseOutcome::Accepted)
})
@@ -573,9 +536,6 @@ async fn replace_lease(
.await?;
if accepted.is_some() {
if is_active {
backfill_push_match_jobs(&mut tx, community).await?;
}
tx.commit().await?;
return Ok(ReplaceLeaseOutcome::Accepted);
}
@@ -2440,14 +2400,14 @@ mod postgres_tests {
.await
.expect("drain matcher queue");
let community = make_community(&pool).await;
let author = [25; 32];
activate(&pool, community, &author, "install", &[26; 32], 1).await;
let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "poison")
.sign_with_keys(&nostr::Keys::generate())
.expect("sign event");
crate::event::insert_event(&pool, community, &event, None)
.await
.expect("insert event");
let author = [25; 32];
activate(&pool, community, &author, "install", &[26; 32], 1).await;
let wake_id = match enqueue_wake(
&pool,
community,
@@ -2775,38 +2735,14 @@ mod postgres_tests {
);
}
/// T1b lost-wake race, forced (migration 0023): with no eligible lease the
/// events trigger must skip the match enqueue, and a lease activation
/// racing an in-flight event insert must be ordered AFTER it by the gate
/// lock — so the activation's backfill enqueues the event the trigger
/// skipped, exactly once. Without the shared/exclusive advisory pair, the
/// trigger could read "no lease" while the activation commits
/// concurrently, dropping that wake with no retry.
// Lease activation must wait for an event insert holding the shared gate lock.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn gate_orders_lease_activation_after_in_flight_event_and_backfills_it() {
async fn gate_orders_lease_activation_after_in_flight_event() {
let pool = setup_pool().await;
let community = make_community(&pool).await;
// Phase 0: no lease anywhere in this community — a committed gated-kind
// event must not be enqueued.
let skipped = nostr::EventBuilder::new(nostr::Kind::Custom(9), "gate skips me")
.sign_with_keys(&nostr::Keys::generate())
.expect("sign skipped event");
crate::event::insert_event(&pool, community, &skipped, None)
.await
.expect("insert lease-less event");
let queued: i64 =
sqlx::query_scalar("SELECT count(*) FROM push_match_queue WHERE community_id=$1")
.bind(community.as_uuid())
.fetch_one(&pool)
.await
.expect("count queue after lease-less insert");
assert_eq!(queued, 0, "gate must skip enqueue with no eligible lease");
// Phase 1: hold an event-insert transaction open past its INSERT. The
// (non-deferred) trigger has already run: shared gate lock held, EXISTS
// saw no lease, enqueue skipped — the classic lost-wake window.
// Hold the insert transaction open after its trigger takes the shared gate lock.
let raced = nostr::EventBuilder::new(nostr::Kind::Custom(9), "raced wake")
.sign_with_keys(&nostr::Keys::generate())
.expect("sign raced event");
@@ -2823,6 +2759,11 @@ mod postgres_tests {
.await
.expect("insert raced event inside held txn");
let insert_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
.fetch_one(&mut *insert_tx)
.await
.expect("read insert backend pid");
// A concurrent activation must block on the exclusive gate lock until
// the insert transaction resolves.
let activation = {
@@ -2852,8 +2793,10 @@ mod postgres_tests {
let mut parked = false;
for _ in 0..100 {
let waiting: i64 = sqlx::query_scalar(
"SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted",
"SELECT count(*) FROM pg_locks WHERE locktype='advisory' AND NOT granted \
AND $1 = ANY(pg_blocking_pids(pid))",
)
.bind(insert_pid)
.fetch_one(&pool)
.await
.expect("inspect advisory waiters");
@@ -2866,8 +2809,7 @@ mod postgres_tests {
assert!(parked, "activation must block on the exclusive gate lock");
assert!(!activation.is_finished());
// Release the event; the activation acquires the gate, and its
// backfill must enqueue the event the trigger skipped — exactly once.
// Committing the insert releases the shared lock so activation can proceed.
insert_tx.commit().await.expect("commit raced insert");
assert_eq!(
tokio::time::timeout(std::time::Duration::from_secs(10), activation)
@@ -2876,33 +2818,21 @@ mod postgres_tests {
.expect("join activation"),
ReplaceLeaseOutcome::Accepted
);
let backfilled: Vec<Vec<u8>> = sqlx::query_scalar(
"SELECT event_id FROM push_match_queue WHERE community_id=$1 ORDER BY created_at",
)
.bind(community.as_uuid())
.fetch_all(&pool)
.await
.expect("read backfilled queue");
assert!(
backfilled.contains(&raced.id.as_bytes().to_vec()),
"raced event must be recovered by the activation backfill"
);
// Phase 2: with the lease now active, the trigger enqueues directly and
// the backfill's ON CONFLICT dedup keeps it single.
// The active lease makes subsequent messages eligible for enqueueing.
let direct = nostr::EventBuilder::new(nostr::Kind::Custom(9), "direct enqueue")
.sign_with_keys(&nostr::Keys::generate())
.expect("sign direct event");
crate::event::insert_event(&pool, community, &direct, None)
.await
.expect("insert post-activation event");
let per_event: Vec<i64> = sqlx::query_scalar(
"SELECT count(*) FROM push_match_queue WHERE community_id=$1 GROUP BY event_id",
let queued: i64 = sqlx::query_scalar(
"SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2",
)
.bind(community.as_uuid())
.fetch_all(&pool)
.bind(direct.id.as_bytes().as_slice())
.fetch_one(&pool)
.await
.expect("count queue rows per event");
assert!(per_event.iter().all(|count| *count == 1));
.expect("count post-activation message jobs");
assert_eq!(queued, 1);
}
}