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(relay): enqueue push work after message commit

Browse Source 
Signed-off-by: Tom Brow <tomb@block.xyz>
This commit is contained in:
Tom Brow
2026-10-09 10:06:50 -07:00
parent 6f330328b8
commit 3164b25cde
16 changed files with 523 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -1203,6 +1203,7 @@ dependencies = [
"sqlx",
"thiserror 2.0.18",
"tokio",
"tokio-util",
"tracing",
"uuid",
]

1
crates/buzz-db/Cargo.toml
View File

@@ -12,6 +12,7 @@ buzz-core = { workspace = true }
buzz-datastore-tracing = { workspace = true }
sqlx = { workspace = true }
tokio = { workspace = true }
tokio-util = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
uuid = { workspace = true }

28
crates/buzz-db/src/runtime/admitted_tx.rs
View File

@@ -73,6 +73,8 @@ pub struct AdmittedTx {
community: CommunityId,
/// Channels whose TTL deadline [`AdmittedTx::commit`] refreshes.
ttl_refresh_channels: BTreeSet<Uuid>,
push_enqueue: Option<super::push_enqueue::PushEnqueue>,
push_events: Vec<Vec<u8>>,
}
impl AdmittedTx {
@@ -87,6 +89,8 @@ impl AdmittedTx {
tx,
community,
ttl_refresh_channels: BTreeSet::new(),
push_enqueue: None,
push_events: Vec::new(),
})
}
@@ -104,9 +108,28 @@ impl AdmittedTx {
tx,
community: lease.community_id,
ttl_refresh_channels: BTreeSet::new(),
push_enqueue: None,
push_events: Vec::new(),
})
}
pub(crate) fn set_push_enqueue(&mut self, producer: Option<super::push_enqueue::PushEnqueue>) {
self.push_enqueue = producer;
}
pub(crate) fn record_push_event(&mut self, event_id: &[u8], kind: i32) {
if self.push_enqueue.is_some()
&& crate::store::event_follow_up::PUSH_MATCH_KINDS.contains(&kind)
{
// Bound transaction-local staging as well as the shared worker queue.
if self.push_events.len() < super::push_enqueue::CAPACITY {
self.push_events.push(event_id.to_vec());
} else {
super::push_enqueue::record_drop("transaction_full");
}
}
}
/// The community this transaction was admitted for.
pub fn community(&self) -> CommunityId {
self.community
@@ -145,6 +168,11 @@ impl AdmittedTx {
.await?;
}
self.tx.commit().await?;
if let Some(producer) = self.push_enqueue {
for event_id in self.push_events {
producer.submit(self.community, event_id);
}
}
Ok(())
}

36
crates/buzz-db/src/runtime/mod.rs
View File

@@ -3,6 +3,7 @@ mod cold_start;
mod connection_observability;
pub mod migration;
pub(crate) mod observability;
pub(crate) mod push_enqueue;
pub mod replica_fence;
pub use admitted_tx::AdmittedTx;
@@ -189,6 +190,8 @@ pub(crate) async fn begin_community_event_write_transaction_with_legacy_metrics(
#[derive(Clone, Debug)]
pub struct Db {
pub(crate) pool: PgPool,
/// Optional bounded post-commit producer, shared across handle clones.
pub(crate) push_enqueue: Option<push_enqueue::PushEnqueue>,
/// Maximum connections configured for this pool (from [`DbConfig::max_connections`]).
pub(crate) max_connections: u32,
/// Optional read-replica pool (from [`DbConfig::read_database_url`]).
@@ -705,7 +708,11 @@ impl Db {
let replica_read_max_age = read_budget_from_ms(config.replica_read_max_age_ms);
let usage_metrics_replica_max_age =
read_budget_from_ms(config.usage_metrics_replica_max_age_ms);
let push_enqueue = config
.push_enabled
.then(|| push_enqueue::PushEnqueue::new(&pool));
Ok(Self {
push_enqueue,
pool,
max_connections: config.max_connections,
read_pool,
@@ -781,7 +788,8 @@ impl Db {
set_config('idle_in_transaction_session_timeout', $2, false), \
set_config('statement_timeout', $3, false), \
set_config('default_transaction_read_only', $4, false), \
set_config('buzz.push_enabled', $5, false)",
set_config('buzz.push_enabled', $5, false), \
set_config('buzz.push_async_enqueue', 'on', false)",
)
.bind(lock_timeout_ms.to_string())
.bind(idle_txn_timeout_ms.to_string())
@@ -942,8 +950,12 @@ impl Db {
}
/// Creates a `Db` from an existing `PgPool` (useful in tests).
///
/// Does not start a push producer or configure sessions. Serving writers
/// that need push must use [`Db::new`] with `push_enabled` instead.
pub fn from_pool(pool: PgPool) -> Self {
Self {
push_enqueue: None,
max_connections: pool.options().get_max_connections(),
read_max_connections: pool.options().get_max_connections(),
pool,
@@ -964,6 +976,7 @@ impl Db {
/// [`Db::fence`]).
pub fn from_pools(pool: PgPool, read_pool: PgPool) -> Self {
Self {
push_enqueue: None,
max_connections: pool.options().get_max_connections(),
read_max_connections: read_pool.options().get_max_connections(),
pool,
@@ -1324,12 +1337,28 @@ impl Db {
&self,
community: CommunityId,
) -> Result<AdmittedTx> {
begin_community_event_write_transaction_with_legacy_metrics(
let mut tx = begin_community_event_write_transaction_with_legacy_metrics(
&self.pool,
community,
observability::WriterOperation::EventWrite,
)
.await
.await?;
tx.set_push_enqueue(self.push_enqueue.clone());
Ok(tx)
}
/// Stop accepting best-effort push enqueue work and cancel in-flight work.
pub fn cancel_push_enqueue(&self) {
if let Some(producer) = &self.push_enqueue {
producer.cancel();
}
}
/// Wait for the cancelled producer to release its dedicated connection.
pub async fn join_push_enqueue(&self) {
if let Some(producer) = &self.push_enqueue {
producer.join().await;
}
}
/// Insert an event while holding and validating an admitted serving-write
@@ -1359,6 +1388,7 @@ impl Db {
let tx = sqlx::Transaction::begin(connection, None).await?;
let mut tx =
AdmittedTx::admit_with_serving_lease(tx, &self.deletion_store(), lease).await?;
tx.set_push_enqueue(self.push_enqueue.clone());
event::acquire_canvas_event_write_lock_if_needed(&mut tx, event, channel_id).await?;
let result =
event::insert_event_with_thread_metadata_tx(&mut tx, event, channel_id, None).await?;

206
crates/buzz-db/src/runtime/push_enqueue.rs Normal file
View File

@@ -0,0 +1,206 @@
//! Bounded best-effort push production, strictly after event commit.
use std::sync::Arc;
use std::time::Duration;
use buzz_core::CommunityId;
use sqlx::{postgres::PgPoolOptions, PgPool};
use tokio::sync::{mpsc, Mutex};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
pub(crate) const CAPACITY: usize = 256;
const ENQUEUE_TIMEOUT: Duration = Duration::from_secs(2);
#[derive(Clone, Debug)]
pub(crate) struct PushEnqueue {
sender: mpsc::Sender<Job>,
cancel: CancellationToken,
task: Arc<Mutex<Option<JoinHandle<()>>>>,
}
#[derive(Debug)]
struct Job {
community: CommunityId,
event_id: Vec<u8>,
#[cfg(test)]
barrier: Option<tokio::sync::oneshot::Sender<()>>,
}
impl PushEnqueue {
pub(crate) fn new(writer: &PgPool) -> Self {
// No connection from the serving pool is ever consumed by push work.
// This lazy pool cannot make relay startup depend on push availability.
let pool = PgPoolOptions::new()
.max_connections(1)
.min_connections(0)
.acquire_timeout(ENQUEUE_TIMEOUT)
.connect_lazy_with(
(*writer.connect_options())
.clone()
.application_name("buzz-push-enqueue"),
);
let (sender, mut receiver) = mpsc::channel::<Job>(CAPACITY);
let cancel = CancellationToken::new();
let stopped = cancel.clone();
let task = tokio::spawn(async move {
loop {
let job = tokio::select! {
biased;
_ = stopped.cancelled() => break,
job = receiver.recv() => match job {
Some(job) => job,
None => break,
},
};
#[cfg(test)]
if let Some(barrier) = job.barrier {
let _ = barrier.send(());
continue;
}
let result = tokio::select! {
biased;
_ = stopped.cancelled() => {
record_drop("shutdown");
break;
},
result = tokio::time::timeout(ENQUEUE_TIMEOUT, enqueue(&pool, &job)) => result,
};
match result {
Ok(Ok(())) => {
metrics::counter!("buzz_push_enqueue_total", "result" => "completed")
.increment(1);
}
Ok(Err(error)) => {
record_drop("error");
tracing::warn!(%error, community = %job.community, "push enqueue failed after message commit");
}
Err(_) => record_drop("timeout"),
}
}
receiver.close();
while receiver.try_recv().is_ok() {
record_drop("shutdown");
}
if tokio::time::timeout(ENQUEUE_TIMEOUT, pool.close())
.await
.is_err()
{
tracing::warn!("push enqueue pool close exceeded shutdown deadline");
}
});
Self {
sender,
cancel,
task: Arc::new(Mutex::new(Some(task))),
}
}
pub(crate) fn submit(&self, community: CommunityId, event_id: Vec<u8>) {
if self.cancel.is_cancelled() {
record_drop("shutdown");
return;
}
if let Err(error) = self.sender.try_send(Job {
community,
event_id,
#[cfg(test)]
barrier: None,
}) {
record_drop(match error {
mpsc::error::TrySendError::Full(_) => "full",
mpsc::error::TrySendError::Closed(_) => "closed",
});
}
}
#[cfg(test)]
pub(crate) async fn flush(&self) {
let (sender, receiver) = tokio::sync::oneshot::channel();
self.sender
.send(Job {
community: CommunityId::from_uuid(uuid::Uuid::nil()),
event_id: Vec::new(),
barrier: Some(sender),
})
.await
.expect("producer open");
receiver.await.expect("producer completed preceding work");
}
pub(crate) fn cancel(&self) {
self.cancel.cancel();
}
pub(crate) async fn join(&self) {
if let Some(task) = self.task.lock().await.take() {
if let Err(error) = task.await {
tracing::warn!(%error, "push enqueue worker failed");
}
}
}
}
pub(crate) fn record_drop(reason: &'static str) {
metrics::counter!("buzz_push_enqueue_total", "result" => reason).increment(1);
}
async fn enqueue(pool: &PgPool, job: &Job) -> crate::Result<()> {
let mut tx = pool.begin().await?;
// Server-side deadlines also bound work if the client disconnects or the
// task is cancelled. These apply only to the dedicated push connection.
sqlx::query("SELECT set_config('lock_timeout', '500ms', true), set_config('statement_timeout', '1500ms', true)")
.execute(&mut *tx).await?;
crate::deletion::DeletionStore::new(pool.clone())
.guard_transaction(&mut tx, job.community)
.await?;
sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))")
.bind(crate::push::push_gate_lock_key(job.community))
.execute(&mut *tx)
.await?;
// Do not enqueue solely because a lease first appeared after the
// message arrived. Updates racing this check may drop a wake, which is
// permitted by the best-effort contract; message storage is authoritative.
sqlx::query(
"INSERT INTO push_match_queue (community_id, event_id) \
SELECT $1, $2 FROM events e \
WHERE e.community_id = $1 AND e.id = $2 AND e.deleted_at IS NULL \
AND community_write_allowed($1) \
AND EXISTS (SELECT 1 FROM push_leases l \
WHERE l.community_id = $1 AND l.active AND l.endpoint_enabled \
AND l.updated_at <= e.received_at \
AND l.expires_at > EXTRACT(EPOCH FROM now())::bigint) \
ON CONFLICT DO NOTHING",
)
.bind(job.community.as_uuid())
.bind(&job.event_id)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn saturation_and_shutdown_never_wait_for_a_receiver() {
let (sender, mut receiver) = mpsc::channel(CAPACITY);
let producer = PushEnqueue {
sender,
cancel: CancellationToken::new(),
task: Arc::new(Mutex::new(None)),
};
let community = CommunityId::from_uuid(uuid::Uuid::new_v4());
for _ in 0..CAPACITY + 1 {
producer.submit(community, vec![1; 32]);
}
assert_eq!(receiver.len(), CAPACITY);
receiver.try_recv().unwrap();
producer.cancel();
producer.submit(community, vec![2; 32]);
assert_eq!(receiver.len(), CAPACITY - 1);
producer.join().await;
}
}

5
crates/buzz-db/src/store/event.rs
View File

@@ -413,6 +413,9 @@ pub async fn huddle_started_link_exists_in_transaction(
/// Insert a Nostr event. Rejects AUTH and ephemeral kinds.
///
/// This raw-pool helper does not start application push production. Serving
/// callers should use [`Db::insert_event`] with a configured [`Db`].
///
/// Returns `(StoredEvent, was_inserted)` — `was_inserted` is `false` on duplicate.
pub async fn insert_event(
pool: &PgPool,
@@ -2116,6 +2119,7 @@ impl Db {
crate::observability::WriterOperation::EventWrite,
)
.await?;
tx.set_push_enqueue(self.push_enqueue.clone());
let result = crate::event::insert_event_in_transaction(&mut tx, event, channel_id).await?;
if result.1 {
crate::insert_mentions_in_transaction(&mut tx, event, channel_id).await?;
@@ -2641,6 +2645,7 @@ impl Db {
crate::observability::WriterOperation::EventWrite,
)
.await?;
tx.set_push_enqueue(self.push_enqueue.clone());
crate::event::acquire_canvas_event_write_lock_if_needed(&mut tx, event, channel_id).await?;
let result = crate::event::insert_event_with_thread_metadata_tx(
&mut tx,

79
crates/buzz-db/src/store/event_follow_up.rs
View File

@@ -2,9 +2,9 @@
//!
//! Two pieces of work follow every durable event insert:
//!
//! - **Push match enqueue.** Push-eligible kinds get a `push_match_queue` row
//! when the community has an eligible lease. This runs right after the
//! INSERT, in the same transaction, and any failure rejects the event.
//! - **Push match enqueue.** Eligible event IDs are staged in memory. Commit
//! submits them to a bounded best-effort worker using a dedicated connection.
//! Push errors, delays, overload, and shutdown can drop wakes, never messages.
//! - **Channel TTL refresh.** A channel-scoped event pushes its ephemeral
//! channel's `ttl_deadline` forward. [`AdmittedTx`] records the channel and
//! runs the refresh as the last statement before COMMIT (see
@@ -13,10 +13,9 @@
//!
//! This is the application-side replacement for the `events_enqueue_push_match`
//! and `events_refresh_channel_ttl` triggers (migrations 0023, 0024, 0040).
//! Both run during the migration window; the push enqueue is idempotent
//! (`ON CONFLICT DO NOTHING`) and a second TTL refresh is harmless. Every
//! production `INSERT INTO events` must call this module, which
//! `tests/observability_source.rs` enforces.
//! Updated writer connections suppress the surviving push trigger. Legacy
//! writers retain transactional enqueue until replaced. Every production
//! `INSERT INTO events` must call this module, enforced by source-policy tests.
use std::collections::BTreeSet;
@@ -38,79 +37,19 @@ const SQLSTATE_LOCK_NOT_AVAILABLE: &str = "55P03";
/// Follow-up for an event row inserted in an admitted transaction.
///
/// Call only when the INSERT affected a row. Enqueues the push match job now
/// and records the channel for the pre-commit TTL refresh.
///
/// If the insert ran inside a savepoint that may still roll back, call
/// [`enqueue_push_match`] inside the savepoint and
/// [`AdmittedTx::record_channel_event`] after it is released instead.
/// Call only when the INSERT affected a row, after releasing any savepoint
/// that may roll it back. Stages push work for after commit and records TTL work.
pub(crate) async fn after_admitted_insert(
tx: &mut AdmittedTx,
event_id: &[u8],
kind: i32,
channel_id: Option<Uuid>,
) -> Result<()> {
let community = tx.community();
enqueue_push_match(tx.conn(), community, event_id, kind).await?;
tx.record_push_event(event_id, kind);
tx.record_channel_event(channel_id, kind);
Ok(())
}
/// Enqueue a push match job for a just-inserted event, if its kind is
/// push-eligible and the community has an eligible lease.
///
/// Takes the push gate SHARED and holds it to transaction end. Lease
/// activations take it EXCLUSIVE (`acquire_push_gate_lock` in `push.rs`), so
/// an event either sees the committed lease or strictly precedes the
/// activation, in which case no wake was owed. The lock must be its own
/// statement: under READ COMMITTED the eligibility check needs a snapshot
/// taken after the lock is granted.
///
/// Errors propagate and reject the event, as the trigger did.
pub(crate) async fn enqueue_push_match(
conn: &mut PgConnection,
community: CommunityId,
event_id: &[u8],
kind: i32,
) -> Result<()> {
if !PUSH_MATCH_KINDS.contains(&kind) {
return Ok(());
}
// Missing means a legacy writer during rollout overlap. New writer pools
// always set this explicitly, including replacement connections.
let enabled: bool = sqlx::query_scalar(
"SELECT COALESCE(NULLIF(current_setting('buzz.push_enabled', true), '')::boolean, true)",
)
.fetch_one(&mut *conn)
.await?;
if !enabled {
return Ok(());
}
crate::observability::observe_advisory_lock(
crate::observability::LockType::PushGate,
sqlx::query("SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))")
.bind(crate::push::push_gate_lock_key(community))
.execute(&mut *conn),
)
.await?;
sqlx::query(
"INSERT INTO push_match_queue (community_id, event_id) \
SELECT $1, $2 \
WHERE EXISTS ( \
SELECT 1 FROM push_leases \
WHERE community_id = $1 \
AND active \
AND endpoint_enabled \
AND expires_at > EXTRACT(EPOCH FROM now())::bigint) \
ON CONFLICT DO NOTHING",
)
.bind(community.as_uuid())
.bind(event_id)
.execute(&mut *conn)
.await?;
Ok(())
}
/// The channel whose TTL an event of `kind` in `channel_id` refreshes, if any.
pub(crate) fn ttl_refresh_channel(channel_id: Option<Uuid>, kind: i32) -> Option<Uuid> {
channel_id.filter(|_| kind != KIND_CHANNEL_CREATE)

227
crates/buzz-db/src/store/event_follow_up_postgres_tests.rs
View File

@@ -86,7 +86,11 @@ fn signed_event(keys: &Keys, kind: u16, content: &str) -> Event {
.expect("sign test event")
}
async fn match_count(pool: &PgPool, community: CommunityId, event: &Event) -> i64 {
async fn match_count(db: &Db, community: CommunityId, event: &Event) -> i64 {
if let Some(producer) = &db.push_enqueue {
producer.flush().await;
}
let pool = db.pool();
sqlx::query_scalar(
"SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2",
)
@@ -285,6 +289,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
activate_lease(pool, community, &keys, 11, now + 3600).await;
assert_rollback(&db, community, channel, &keys).await;
assert_async_isolation(&db, community, &keys).await;
for kind in [9, 40002, 45001, 45003] {
let deadline_before_first_insert = reset_deadline(pool, community, channel).await;
@@ -296,7 +301,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.1
);
assert_eq!(event_count(pool, community, &message).await, 1);
assert_eq!(match_count(pool, community, &message).await, 1);
assert_eq!(match_count(&db, community, &message).await, 1);
let deadline_after_first_insert = deadline(pool, community, channel).await;
assert!(
deadline_after_first_insert > deadline_before_first_insert,
@@ -308,7 +313,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.expect("deduplicate event")
.1
);
assert_eq!(match_count(pool, community, &message).await, 1);
assert_eq!(match_count(&db, community, &message).await, 1);
assert_eq!(
deadline(pool, community, channel).await,
deadline_after_first_insert,
@@ -323,7 +328,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.expect("insert reaction")
.1
);
assert_eq!(match_count(pool, community, &reaction).await, 0);
assert_eq!(match_count(&db, community, &reaction).await, 0);
let no_lease_community = create_community(pool).await;
let no_lease_message = signed_event(&keys, 9, "no-lease");
@@ -334,7 +339,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.1
);
assert_eq!(
match_count(pool, no_lease_community, &no_lease_message).await,
match_count(&db, no_lease_community, &no_lease_message).await,
0
);
@@ -386,7 +391,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.1
);
assert_eq!(event_count(pool, predicate_community, &message).await, 1);
assert_eq!(match_count(pool, predicate_community, &message).await, 0);
assert_eq!(match_count(&db, predicate_community, &message).await, 0);
}
let mut push_gate_holder = pool.begin().await.expect("begin push-gate lock holder");
@@ -401,16 +406,16 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.execute(gated_tx.conn())
.await
.expect("bound push-gate lock wait");
let gated_error = event::insert_event_in_transaction(&mut gated_tx, &gated, None)
.await
.expect_err("push-eligible insert must fail closed when its gate cannot be acquired");
assert_sqlstate(&gated_error, "55P03");
gated_tx
.rollback()
.await
.expect("rollback failed push-gate transaction");
assert_eq!(event_count(pool, community, &gated).await, 0);
assert_eq!(match_count(pool, community, &gated).await, 0);
timeout(Duration::from_secs(5), async {
event::insert_event_in_transaction(&mut gated_tx, &gated, None)
.await
.unwrap();
gated_tx.commit().await.unwrap();
})
.await
.expect("push contention must not delay message commit");
assert_eq!(event_count(pool, community, &gated).await, 1);
assert_eq!(match_count(&db, community, &gated).await, 0);
let ungated = signed_event(&keys, 7, "push-gate-non-eligible-control");
let mut ungated_tx = begin_caller_owned_event_transaction(&db, community).await;
@@ -429,7 +434,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.await
.expect("commit non-push control while gate held");
assert_eq!(event_count(pool, community, &ungated).await, 1);
assert_eq!(match_count(pool, community, &ungated).await, 0);
assert_eq!(match_count(&db, community, &ungated).await, 0);
push_gate_holder
.rollback()
.await
@@ -442,7 +447,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.1
);
assert_eq!(event_count(pool, community, &gated_control).await, 1);
assert_eq!(match_count(pool, community, &gated_control).await, 1);
assert_eq!(match_count(&db, community, &gated_control).await, 1);
let rollback_deadline = reset_deadline(pool, community, channel).await;
let rolled_back = signed_event(&keys, 9, "rolled-back");
@@ -461,13 +466,13 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.fetch_one(rollback_tx.conn())
.await
.expect("read uncommitted follow-up");
assert_eq!(in_transaction, 1);
assert_eq!(in_transaction, 0, "push work starts only after commit");
rollback_tx
.rollback()
.await
.expect("rollback event and follow-up");
assert_eq!(event_count(pool, community, &rolled_back).await, 0);
assert_eq!(match_count(pool, community, &rolled_back).await, 0);
assert_eq!(match_count(&db, community, &rolled_back).await, 0);
assert_eq!(deadline(pool, community, channel).await, rollback_deadline);
let before_commit = reset_deadline(pool, community, channel).await;
@@ -500,7 +505,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
>= database_time_before_commit + chrono::Duration::seconds(60),
"the deferred trigger must refresh TTL at commit"
);
assert_eq!(match_count(pool, community, &committed).await, 1);
assert_eq!(match_count(&db, community, &committed).await, 1);
let before_skips = reset_deadline(pool, community, channel).await;
let channel_create = signed_event(&keys, 9007, "channel-create");
@@ -511,7 +516,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.1
);
assert_eq!(deadline(pool, community, channel).await, before_skips);
assert_eq!(match_count(pool, community, &channel_create).await, 0);
assert_eq!(match_count(&db, community, &channel_create).await, 0);
let channelless = signed_event(&keys, 9, "channel-null");
assert!(
db.insert_event(community, &channelless, None)
@@ -519,7 +524,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.expect("insert channel-less event")
.1
);
assert_eq!(match_count(pool, community, &channelless).await, 1);
assert_eq!(match_count(&db, community, &channelless).await, 1);
let no_ttl_channel =
channel_with_ttl(pool, community, &keys.public_key().to_bytes(), None).await;
@@ -630,7 +635,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.expect_err("a cancelled TTL refresh must reject the event");
assert_sqlstate(&cancel_error, "57014");
assert_eq!(event_count(pool, community, &cancelled_event).await, 0);
assert_eq!(match_count(pool, community, &cancelled_event).await, 0);
assert_eq!(match_count(&db, community, &cancelled_event).await, 0);
assert_eq!(
deadline(pool, community, channel).await,
before_lock_timeout
@@ -640,7 +645,7 @@ async fn assert_contract(migration_schema: bool, arm: Arm) {
.await
.expect("release exclusive TTL lock");
assert_eq!(event_count(pool, community, &locked_event).await, 1);
assert_eq!(match_count(pool, community, &locked_event).await, 1);
assert_eq!(match_count(&db, community, &locked_event).await, 1);
assert_entry_points(&db, community, channel, &keys).await;
}
@@ -660,9 +665,25 @@ async fn assert_entry_points(db: &Db, community: CommunityId, channel: Uuid, key
.expect("insert through the thread-metadata writer")
.1
);
assert_eq!(match_count(pool, community, &threaded).await, 1);
assert_eq!(match_count(db, community, &threaded).await, 1);
assert!(deadline(pool, community, channel).await > before);
// Internal/external-effect completion uses the same post-commit producer.
let lease = db
.deletion_store()
.acquire_serving_write_lease(community, "push-contract", "test", Duration::from_secs(60))
.await
.unwrap();
let guarded = signed_event(keys, 40002, "serving-lease-writer");
db.insert_event_with_serving_write_guard(&lease, &guarded, Some(channel))
.await
.unwrap();
assert_eq!(match_count(db, community, &guarded).await, 1);
db.deletion_store()
.release_serving_write_lease(&lease)
.await
.unwrap();
// Replaceable writer (NIP-16 and NIP-29 discovery state).
let before = reset_deadline(pool, community, channel).await;
let replaceable = signed_event(keys, 39000, "replaceable-writer");
@@ -672,7 +693,7 @@ async fn assert_entry_points(db: &Db, community: CommunityId, channel: Uuid, key
.expect("insert through the replaceable writer")
.1
);
assert_eq!(match_count(pool, community, &replaceable).await, 0);
assert_eq!(match_count(db, community, &replaceable).await, 0);
assert!(deadline(pool, community, channel).await > before);
// Parameterized writer: the insert runs inside a savepoint, and the
@@ -801,7 +822,7 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &
db.insert_event(community, &retained, Some(channel))
.await
.unwrap();
assert_eq!(match_count(pool, community, &retained).await, 1);
assert_eq!(match_count(db, community, &retained).await, 1);
let lease_before: serde_json::Value =
sqlx::query_scalar("SELECT to_jsonb(p) FROM push_leases p WHERE community_id=$1")
.bind(community.as_uuid())
@@ -840,6 +861,7 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &
push_enabled: false,
max_connections: 1,
min_connections: 0,
lock_timeout_ms: 100,
..DbConfig::default()
})
.await
@@ -864,18 +886,18 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &
.unwrap();
let message = signed_event(keys, kind, &format!("disabled-{kind}"));
timeout(
Duration::from_secs(2),
Duration::from_secs(5),
disabled.insert_event(community, &message, Some(channel)),
)
.await
.expect("disabled write must not wait for the push gate")
.unwrap();
assert_eq!(event_count(pool, community, &message).await, 1);
assert_eq!(match_count(pool, community, &message).await, 0);
assert_eq!(match_count(db, community, &message).await, 0);
disabled_events.push(message);
}
holder.rollback().await.unwrap();
assert_eq!(match_count(pool, community, &retained).await, 1);
assert_eq!(match_count(db, community, &retained).await, 1);
let lease_after: serde_json::Value =
sqlx::query_scalar("SELECT to_jsonb(p) FROM push_leases p WHERE community_id=$1")
.bind(community.as_uuid())
@@ -912,7 +934,16 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &
.insert_event(community, &message, Some(channel))
.await
.unwrap();
assert_eq!(match_count(pool, community, &message).await, 1);
let trigger_exists: bool = sqlx::query_scalar(
"SELECT EXISTS (SELECT 1 FROM pg_trigger WHERE tgname='events_enqueue_push_match')",
)
.fetch_one(pool)
.await
.unwrap();
assert_eq!(
match_count(db, community, &message).await,
i64::from(trigger_exists)
);
legacy.pool().close().await;
// Reactivation is an operator decision about retained queues. Reopening
@@ -921,9 +952,137 @@ async fn assert_rollback(db: &Db, community: CommunityId, channel: Uuid, keys: &
db.insert_event(community, &message, Some(channel))
.await
.unwrap();
assert_eq!(match_count(pool, community, &message).await, 1);
assert_eq!(match_count(db, community, &message).await, 1);
for message in disabled_events {
assert_eq!(match_count(pool, community, &message).await, 0);
assert_eq!(match_count(db, community, &message).await, 0);
}
disabled.pool().close().await;
}
async fn assert_async_isolation(db: &Db, community: CommunityId, keys: &Keys) {
// A one-connection serving pool proves a stalled push job cannot monopolize
// its only connection. The lock holder uses the independent fixture pool.
let isolated = Db::new(&DbConfig {
database_url: std::env::var("BUZZ_TEST_DATABASE_URL").unwrap(),
max_connections: 1,
min_connections: 0,
lock_timeout_ms: 100,
push_enabled: true,
..DbConfig::default()
})
.await
.unwrap();
let mut holder = db.pool().begin().await.unwrap();
sqlx::query("LOCK TABLE push_match_queue IN ACCESS EXCLUSIVE MODE")
.execute(&mut *holder)
.await
.unwrap();
let message = signed_event(keys, 9, "async-queue-lock");
timeout(
Duration::from_secs(5),
isolated.insert_event(community, &message, None),
)
.await
.expect("push table lock must not delay storage")
.unwrap();
assert_eq!(event_count(isolated.pool(), community, &message).await, 1);
isolated.push_enqueue.as_ref().unwrap().flush().await;
// The background statement failed its lock timeout; the message survived.
holder.rollback().await.unwrap();
assert_eq!(match_count(&isolated, community, &message).await, 0);
assert_eq!(event_count(isolated.pool(), community, &message).await, 1);
let healthy = signed_event(keys, 9, "async-after-error");
isolated
.insert_event(community, &healthy, None)
.await
.unwrap();
assert_eq!(match_count(&isolated, community, &healthy).await, 1);
// If activation finishes while a post-commit job waits, it must not create
// a wake solely for a lease that appeared after the message was received.
let later = create_community(db.pool()).await;
let mut activation = db.pool().begin().await.unwrap();
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
.bind(push::push_gate_lock_key(later))
.execute(&mut *activation)
.await
.unwrap();
let before_enrollment = signed_event(keys, 9, "before-async-enrollment");
isolated
.insert_event(later, &before_enrollment, None)
.await
.unwrap();
sqlx::query(
"INSERT INTO push_leases (community_id, author, installation_id, source_event_id, \
source_created_at, generation, active, endpoint_enabled, app_profile, endpoint_hash, \
endpoint_grant, max_class, subscriptions, expires_at, updated_at) \
SELECT $1, author, installation_id, source_event_id, source_created_at, generation, \
active, endpoint_enabled, app_profile, endpoint_hash, endpoint_grant, max_class, \
subscriptions, expires_at, clock_timestamp() FROM push_leases WHERE community_id=$2",
)
.bind(later.as_uuid())
.bind(community.as_uuid())
.execute(&mut *activation)
.await
.unwrap();
activation.commit().await.unwrap();
assert_eq!(match_count(&isolated, later, &before_enrollment).await, 0);
assert_eq!(
event_count(isolated.pool(), later, &before_enrollment).await,
1
);
let mut holder = db.pool().begin().await.unwrap();
sqlx::query("LOCK TABLE push_match_queue IN ACCESS EXCLUSIVE MODE")
.execute(&mut *holder)
.await
.unwrap();
let cancelled = signed_event(keys, 9, "async-cancel-inflight");
isolated
.insert_event(community, &cancelled, None)
.await
.unwrap();
timeout(Duration::from_secs(2), async {
loop {
let waiting: bool = sqlx::query_scalar(
"SELECT EXISTS (SELECT 1 FROM pg_stat_activity \
WHERE datname=current_database() AND application_name='buzz-push-enqueue' \
AND wait_event_type='Lock' AND query LIKE 'INSERT INTO push_match_queue%')",
)
.fetch_one(db.pool())
.await
.unwrap();
if waiting {
break;
}
tokio::time::sleep(Duration::from_millis(5)).await;
}
})
.await
.expect("observe actual enqueue blocked on the table");
isolated.cancel_push_enqueue();
timeout(Duration::from_secs(5), isolated.join_push_enqueue())
.await
.expect("shutdown cancels in-flight enqueue without waiting for lock timeout");
holder.rollback().await.unwrap();
assert_eq!(event_count(isolated.pool(), community, &cancelled).await, 1);
let queued: i64 = sqlx::query_scalar(
"SELECT count(*) FROM push_match_queue WHERE community_id=$1 AND event_id=$2",
)
.bind(community.as_uuid())
.bind(cancelled.id.as_bytes().as_slice())
.fetch_one(db.pool())
.await
.unwrap();
assert_eq!(queued, 0);
let after_stop = signed_event(keys, 9, "async-message-after-stop");
isolated
.insert_event(community, &after_stop, None)
.await
.unwrap();
assert_eq!(
event_count(isolated.pool(), community, &after_stop).await,
1
);
isolated.pool().close().await;
}

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

@@ -29,14 +29,11 @@ async fn begin_operation_transaction(
Ok(sqlx::Transaction::begin(connection, None).await?)
}
/// Namespace for the per-community push-gate advisory lock. Event inserts
/// take it SHARED in `event_follow_up::enqueue_push_match` (and in the
/// `enqueue_push_match_job` trigger from migration 0023 until it is retired);
/// every lease transition that can make match eligibility true takes it
/// EXCLUSIVE here, forcing a total order so a concurrent event insert either
/// sees the committed lease or strictly precedes the activation (in which case
/// no wake was owed). Distinct key domain from the audit lock and the lease
/// address/author locks.
/// Namespace for the per-community push-gate advisory lock. The post-commit
/// enqueue worker takes it SHARED, while lease transitions take it EXCLUSIVE.
/// The worker checks receipt-time eligibility after obtaining the lock. Legacy
/// trigger writers still hold the shared lock in their event transaction.
/// Distinct key domain from the audit and lease address/author locks.
const PUSH_GATE_LOCK_NAMESPACE: &str = "buzz_push_gate:";
/// The push-gate advisory lock key for `community`. Both sides of the lock

14
crates/buzz-db/src/store/replaceable.rs
View File

@@ -333,15 +333,6 @@ async fn replace_parameterized_event_in_transaction_impl(
if insert_result.rows_affected() == 0 {
return Ok(false);
}
// Inside the savepoint, so a later failure here also drops the job.
// The TTL refresh is recorded only once the savepoint is released.
crate::store::event_follow_up::enqueue_push_match(
tx.conn(),
community_id,
incoming_id,
kind_i32,
)
.await?;
if is_nip_rs {
sqlx::query(
@@ -390,7 +381,8 @@ async fn replace_parameterized_event_in_transaction_impl(
.execute(tx.conn())
.await?;
if status == ParameterizedReplaceStatus::Inserted {
tx.record_channel_event(channel_id, kind_i32);
crate::store::event_follow_up::after_admitted_insert(tx, incoming_id, kind_i32, channel_id)
.await?;
}
Ok(ParameterizedReplaceResult::new(
@@ -436,6 +428,7 @@ impl Db {
observability::WriterOperation::EventWrite,
)
.await?;
tx.set_push_enqueue(self.push_enqueue.clone());
let transaction_timer = observability::TransactionTimer::start(
observability::TransactionOperation::ReplaceAddressableEvent,
);
@@ -610,6 +603,7 @@ impl Db {
observability::WriterOperation::EventWrite,
)
.await?;
tx.set_push_enqueue(self.push_enqueue.clone());
let transaction_timer =
observability::TransactionTimer::start(TransactionOperation::ReplaceParameterizedEvent);
transaction_timer

2
crates/buzz-relay/src/main.rs
View File

@@ -1672,6 +1672,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
)
.await?;
state.push_cancel.cancel();
state.db.cancel_push_enqueue();
state.db.join_push_enqueue().await;
if let Some((matcher, delivery)) = push_workers {
for result in [matcher.await, delivery.await] {
if let Err(error) = result {

1
crates/buzz-relay/src/state.rs
View File

@@ -1589,6 +1589,7 @@ impl AppState {
pub fn begin_shutdown(&self) {
self.shutting_down.store(true, Ordering::Release);
self.push_cancel.cancel();
self.db.cancel_push_enqueue();
}
#[cfg(test)]

29
docs/push-gateway-deployment.md
View File

@@ -182,6 +182,7 @@ installation, event, or request identifier as a label.
| Metric | Type | Labels | Meaning |
|---|---|---|---|
| `buzz_push_enabled` | gauge | — | `1` only when the deployment opt-in is active. |
| `buzz_push_enqueue_total` | counter | `result` = `completed` \| `full` \| `transaction_full` \| `closed` \| `error` \| `timeout` \| `shutdown` | Best-effort post-commit enqueue outcomes. |
| `buzz_push_match_jobs_total` | counter | `result` = `matched` \| `unmatched` \| `error` \| `context_error` | Accepted message events evaluated by the matcher. |
| `buzz_push_match_queue_seconds` | histogram | — | Relay receipt to matcher evaluation latency. |
| `buzz_push_wakes_total` | counter | `result` = `enqueued` \| `duplicate` \| `inactive_lease` | Durable wake-enqueue outcomes. |
@@ -260,15 +261,17 @@ gateway grants unchanged. Do not stop a shared gateway as part of relay rollback
The relay copies the flag into `DbConfig::push_enabled`. Every physical writer
connection, including the audit pool and replacement connections, sets the
`buzz.push_enabled` PostgreSQL session setting. Other `DbConfig` callers default
to disabled. Both the application producer and the surviving overlap trigger
return before push locks, lease scans or queue writes when this setting is off.
to disabled. Updated writers also set `buzz.push_async_enqueue=on`, which makes
the surviving trigger skip enqueueing regardless of the enabled flag. The
application producer is created only when enabled. Disabled writes perform no
push locks, lease scans or queue writes.
The overlap migration updates only an existing trigger function; it does not
recreate a retired trigger or function. Deploy that migration before relying on
the flag with a surviving trigger. Older images and raw SQL writers with a
missing setting retain legacy enqueue behavior; inventory and replace or
explicitly disable all such connections before declaring rollback complete.
On SIGTERM/Ctrl-C, the relay cancels matcher and delivery futures immediately,
On SIGTERM/Ctrl-C, the relay cancels enqueue, matcher and delivery futures immediately,
before its listener drain. Uncommitted transactions roll back; committed claims
remain leased until their normal expiry. No cancellation cleanup deletes or
acknowledges queue entries. An HTTP request already accepted by the gateway may
@@ -279,9 +282,23 @@ Verify all old writer sessions and workers have exited, eligible messages still
store and arrive in chat, no new matcher jobs appear, and gateway attempts stop
once in-flight work settles. Inspect enqueue activity as well as queue counts:
a stable count alone can conceal additions and removals. Ordinary reads, writes,
authentication and reconnects must remain healthy. Enabled enqueue still runs
inside the message transaction and propagates errors; this change does not
isolate enabled message storage from push failures.
authentication and reconnects must remain healthy.
When enabled, updated application writers stage at most 256 event IDs per
transaction and submit them only after successful commit using a non-waiting
send. One background worker has a 256-item queue and a dedicated one-connection
pool; it never borrows a message-write connection. Each attempt has a two-second
deadline, a 500 ms lock timeout and a 1.5 second statement timeout. Saturation,
errors, deadlines, process loss and shutdown can drop notifications. They cannot
reject or roll back a committed message, and there is no replay/backfill of
in-memory work. Monitor `buzz_push_enqueue_total` by result (`completed`, `full`,
`transaction_full`, `closed`, `error`, `timeout`, `shutdown`). `completed` includes
jobs with no eligible lease. Database enqueue is conditional on a live event and
an eligible lease last updated no later than event receipt. A concurrent lease
update can suppress a notification under this best-effort contract.
This isolation requires the overlap migration and updated writers. Old images
and raw SQL writers can still run the legacy transactional trigger until replaced.
Pending matcher jobs and pending/claimed delivery jobs remain dormant. Do not
process or delete them during urgent rollback. **Before re-enabling**, inspect

6
migrations/0059_push_rollback_gate.sql
View File

@@ -14,12 +14,14 @@ BEGIN
-- Legacy connections without the setting preserve their existing behavior.
-- Updated writer pools always set it, including on reconnect. Check before
-- taking push locks or touching leases/queues so disabled chat stays healthy.
IF COALESCE(current_setting('buzz.push_async_enqueue', true), '') = 'on' THEN
RETURN NEW;
END IF;
IF NOT COALESCE(NULLIF(current_setting('buzz.push_enabled', true), '')::boolean, true) THEN
RETURN NEW;
END IF;
-- Keep this allowlist identical to the relay's validated NIP-PL descriptor.
-- Centralizing it on the events table covers every durable producer,
-- including internal paths that bypass live dispatch.
-- This compatibility path covers legacy writers until they are replaced.
IF NEW.kind IN (9, 40002, 45001, 45003) THEN
PERFORM pg_advisory_xact_lock_shared(
hashtextextended('buzz_push_gate:' || NEW.community_id::text, 0));

6
schema/functions/enqueue_push_match_job.sql
View File

@@ -8,12 +8,14 @@ BEGIN
-- Legacy connections without the setting preserve their existing behavior.
-- Updated writer pools always set it, including on reconnect. Check before
-- taking push locks or touching leases/queues so disabled chat stays healthy.
IF COALESCE(current_setting('buzz.push_async_enqueue', true), '') = 'on' THEN
RETURN NEW;
END IF;
IF NOT COALESCE(NULLIF(current_setting('buzz.push_enabled', true), '')::boolean, true) THEN
RETURN NEW;
END IF;
-- Keep this allowlist identical to the relay's validated NIP-PL descriptor.
-- Centralizing it on the events table covers every durable producer,
-- including internal paths that bypass live dispatch.
-- This compatibility path covers legacy writers until they are replaced.
IF NEW.kind IN (9, 40002, 45001, 45003) THEN
PERFORM pg_advisory_xact_lock_shared(
hashtextextended('buzz_push_gate:' || NEW.community_id::text, 0));

8
schema/tables/public/push_match_queue.sql
View File

@@ -1,7 +1,7 @@
-- Durable event-to-push matching follower. The trigger runs in the event insert
-- transaction, so every accepted persistent event has a crash-safe match job and
-- rejected/rolled-back events never do. Processing is idempotent through the
-- push_wake_outbox endpoint/event unique key.
-- Durable jobs after best-effort post-commit enqueue. Updated application
-- writers may drop push work on overload/failure/shutdown; messages remain
-- authoritative. Legacy writers may still enqueue in their event transaction.
-- Processing is idempotent through the outbox endpoint/event unique key.
CREATE TABLE push_match_queue (
community_id UUID NOT NULL REFERENCES communities(id),
event_id BYTEA NOT NULL CHECK (length(event_id) = 32),