relay: gate push enqueue on live leases; batch matcher pipeline (T1b/T1a-repair/T2b)

crates/buzz-db/src/channel.rs

@@ -538,6 +538,33 @@ pub async fn is_member(
     Ok(cnt > 0)
 }
 
+/// Return which of the given (channel, pubkey) combinations are active
+/// memberships, restricted to non-deleted channels — one statement for any
+/// batch size (T2b). Semantics per pair match [`is_member`].
+pub async fn membership_pairs(
+    pool: &PgPool,
+    community_id: CommunityId,
+    channel_ids: &[Uuid],
+    pubkeys: &[Vec<u8>],
+) -> Result<Vec<(Uuid, Vec<u8>)>> {
+    if channel_ids.is_empty() || pubkeys.is_empty() {
+        return Ok(Vec::new());
+    }
+    let rows = sqlx::query(
+        "SELECT cm.channel_id, cm.pubkey FROM channel_members cm \
+         JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \
+         WHERE cm.community_id = $1 AND cm.channel_id = ANY($2) AND cm.pubkey = ANY($3) AND cm.removed_at IS NULL",
+    )
+    .bind(community_id.as_uuid())
+    .bind(channel_ids)
+    .bind(pubkeys)
+    .fetch_all(pool)
+    .await?;
+    rows.into_iter()
+        .map(|row| Ok((row.try_get("channel_id")?, row.try_get("pubkey")?)))
+        .collect()
+}
+
 /// Returns all active members of the given channel.
 ///
 /// Returns an empty list if the channel has been soft-deleted.
@@ -1076,9 +1103,32 @@ pub async fn update_channel(
     q = q.bind(community_id.as_uuid());
     q = q.bind(channel_id);
 
-    let result = q.execute(pool).await?;
-    if result.rows_affected() == 0 {
-        return Err(DbError::ChannelNotFound(channel_id));
+    // T1a repair: a TTL change can flip this channel's event-trigger fast
+    // path (migration 0024 reads ttl_seconds under a SHARED per-channel
+    // advisory lock). Take the same key EXCLUSIVE before the UPDATE so a
+    // concurrent event either sees the committed TTL or strictly precedes
+    // this transition — whose own deadline reset is then the latest word.
+    // Non-TTL updates don't touch the fast path and skip the lock.
+    if updates.ttl_seconds.is_some() {
+        let mut tx = pool.begin().await?;
+        sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+            .bind(format!(
+                "buzz_channel_ttl:{}:{}",
+                community_id.as_uuid(),
+                channel_id
+            ))
+            .execute(&mut *tx)
+            .await?;
+        let result = q.execute(&mut *tx).await?;
+        if result.rows_affected() == 0 {
+            return Err(DbError::ChannelNotFound(channel_id));
+        }
+        tx.commit().await?;
+    } else {
+        let result = q.execute(pool).await?;
+        if result.rows_affected() == 0 {
+            return Err(DbError::ChannelNotFound(channel_id));
+        }
     }
 
     get_channel(pool, community_id, channel_id).await

crates/buzz-db/src/event.rs

@@ -1561,6 +1561,14 @@ mod tests {
         assert_eq!(stale_ttl, None);
 
         let mut activation = pool.begin().await.expect("begin TTL activation");
+        // Model the repaired update_channel protocol (migration 0024): the
+        // TTL transition holds the per-channel advisory key EXCLUSIVE, which
+        // is what the event trigger's shared acquisition now waits on.
+        sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+            .bind(format!("buzz_channel_ttl:{community_uuid}:{racing}"))
+            .execute(&mut *activation)
+            .await
+            .expect("acquire exclusive channel TTL key");
         let activation_deadline: DateTime<Utc> = sqlx::query_scalar(
             "UPDATE channels \
              SET ttl_seconds = 60, ttl_deadline = clock_timestamp() + interval '60 seconds' \
@@ -1605,6 +1613,99 @@ mod tests {
         );
     }
 
+    /// T1a repair regression test (migration 0024): permanent-channel event
+    /// commits must not serialize on the channel row. The 0022 trigger took
+    /// `FOR UPDATE` on the channel tuple before testing `ttl_seconds`, so
+    /// concurrent commits into one hot permanent channel queued at commit
+    /// time (deferred trigger) — invisible to any single-connection test.
+    /// This holds N insert transactions at a barrier past their INSERTs,
+    /// then proves (a) while all N sit pre-commit, no transaction holds a
+    /// row-level lock on the channel tuple, and (b) all N commits succeed
+    /// with the channel row untouched (permanent ⇒ no deadline write).
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn permanent_channel_event_commits_do_not_lock_the_channel_row() {
+        const N: usize = 8;
+        // setup_pool's default cap (10) covers N held transactions plus the
+        // pg_locks inspector connection.
+        let pool = setup_pool().await;
+        let community_uuid = make_test_community(&pool).await;
+        let channel = make_test_channel(&pool, community_uuid, None).await;
+
+        // Open N transactions, run the full event INSERT in each (the deferred
+        // trigger fires at COMMIT), and park them at a barrier.
+        let mut txs = Vec::new();
+        for i in 0..N {
+            let mut tx = pool.begin().await.expect("begin insert txn");
+            let event = make_text_event(&format!("hot channel event {i}"));
+            sqlx::query(
+                "INSERT INTO events (community_id,id,pubkey,created_at,kind,tags,content,sig,received_at,channel_id) \
+                 VALUES ($1,$2,$3,$4,9,$5,$6,$7,now(),$8)",
+            )
+            .bind(community_uuid)
+            .bind(event.id.as_bytes().as_slice())
+            .bind(event.pubkey.as_bytes().as_slice())
+            .bind(DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).unwrap())
+            .bind(serde_json::to_value(&event.tags).unwrap())
+            .bind(&event.content)
+            .bind(event.sig.serialize().as_slice())
+            .bind(channel)
+            .execute(&mut *tx)
+            .await
+            .expect("insert event inside held txn");
+            txs.push(tx);
+        }
+
+        // With all N transactions holding completed INSERTs, none may hold a
+        // row-level lock on the channels tuple. (The 0022 trigger would not
+        // have taken it yet either — it locks at COMMIT — so also verify the
+        // commit phase below completes without mutual blocking.)
+        let tuple_locks: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM pg_locks l \
+             JOIN pg_class c ON c.oid = l.relation \
+             WHERE c.relname = 'channels' AND l.locktype = 'tuple'",
+        )
+        .fetch_one(&pool)
+        .await
+        .expect("inspect pg_locks");
+        assert_eq!(tuple_locks, 0, "no channel tuple locks while txns are held");
+
+        // Release all commits concurrently. Under 0022 these serialized on the
+        // channel row (each holding it across its WAL flush); under 0024 the
+        // shared advisory key admits them all. Join with a timeout so a
+        // regression fails fast instead of hanging the suite.
+        let commits = txs
+            .into_iter()
+            .map(|tx| tokio::spawn(async move { tx.commit().await }))
+            .collect::<Vec<_>>();
+        for c in commits {
+            tokio::time::timeout(std::time::Duration::from_secs(10), c)
+                .await
+                .expect("concurrent permanent-channel commits must not block")
+                .expect("join commit task")
+                .expect("commit succeeds");
+        }
+
+        let deadline: Option<DateTime<Utc>> = sqlx::query_scalar(
+            "SELECT ttl_deadline FROM channels WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community_uuid)
+        .bind(channel)
+        .fetch_one(&pool)
+        .await
+        .expect("read deadline after commits");
+        assert_eq!(deadline, None, "permanent channel must remain untouched");
+        let stored: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM events WHERE community_id = $1 AND channel_id = $2",
+        )
+        .bind(community_uuid)
+        .bind(channel)
+        .fetch_one(&pool)
+        .await
+        .expect("count stored events");
+        assert_eq!(stored as usize, N);
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn get_event_by_id_is_scoped_when_event_id_collides_across_communities() {

crates/buzz-db/src/lib.rs

@@ -1220,12 +1220,13 @@ impl Db {
         event::get_events_by_ids(&self.pool, community_id, ids).await
     }
 
-    /// Exclusively claim the next due event-to-push matcher job.
-    pub async fn claim_due_push_match(
+    /// Exclusively claim a batch of due matcher jobs from one community.
+    pub async fn claim_due_push_match_batch(
         &self,
+        limit: i64,
         lease_until: DateTime<Utc>,
-    ) -> Result<Option<push::ClaimedMatch>> {
-        push::claim_due_match(&self.pool, lease_until).await
+    ) -> Result<Option<push::ClaimedMatchBatch>> {
+        push::claim_due_match_batch(&self.pool, limit, lease_until).await
     }
 
     /// Load active endpoint-enabled leases eligible for push matching.
@@ -1236,18 +1237,30 @@ impl Db {
         push::active_match_leases(&self.pool, community).await
     }
 
-    /// Complete a matcher job if its claim fence is still held.
-    pub async fn complete_push_match(&self, job: &push::ClaimedMatch) -> Result<bool> {
-        push::complete_match(&self.pool, job).await
+    /// Complete matcher jobs from one claimed batch while the fence holds.
+    pub async fn complete_push_match_batch(
+        &self,
+        community: CommunityId,
+        claim_id: uuid::Uuid,
+        event_ids: &[Vec<u8>],
+    ) -> Result<u64> {
+        push::complete_match_batch(&self.pool, community, claim_id, event_ids).await
     }
 
-    /// Release a matcher claim for retry at the supplied time.
-    pub async fn retry_push_match(
+    /// Release fenced matcher claims from one batch for retry.
+    pub async fn retry_push_match_batch(
         &self,
-        job: &push::ClaimedMatch,
+        community: CommunityId,
+        claim_id: uuid::Uuid,
+        event_ids: &[Vec<u8>],
         next: DateTime<Utc>,
-    ) -> Result<bool> {
-        push::retry_match(&self.pool, job, next).await
+    ) -> Result<u64> {
+        push::retry_match_batch(&self.pool, community, claim_id, event_ids, next).await
+    }
+
+    /// Delete exhausted matcher jobs (periodic sweep, off the claim path).
+    pub async fn reap_exhausted_push_matches(&self) -> Result<u64> {
+        push::reap_exhausted_matches(&self.pool).await
     }
 
     /// Idempotently enqueue a wake for a matched lease and event.
@@ -1261,6 +1274,15 @@ impl Db {
         push::enqueue_wake(&self.pool, community, author, installation_id, wake).await
     }
 
+    /// Set-wise [`Self::enqueue_push_wake`]: one transaction per batch.
+    pub async fn enqueue_push_wakes(
+        &self,
+        community: CommunityId,
+        requests: &[push::WakeRequest],
+    ) -> Result<Vec<push::EnqueueWakeOutcome>> {
+        push::enqueue_wakes(&self.pool, community, requests).await
+    }
+
     /// Exclusively claim due wake jobs for one community.
     pub async fn claim_due_push_wakes(
         &self,
@@ -1534,6 +1556,17 @@ impl Db {
         channel::is_member(&self.pool, community_id, channel_id, pubkey).await
     }
 
+    /// Return the active (channel, pubkey) membership pairs among the given
+    /// sets, in one statement.
+    pub async fn membership_pairs(
+        &self,
+        community_id: CommunityId,
+        channel_ids: &[Uuid],
+        pubkeys: &[Vec<u8>],
+    ) -> Result<Vec<(Uuid, Vec<u8>)>> {
+        channel::membership_pairs(&self.pool, community_id, channel_ids, pubkeys).await
+    }
+
     /// Returns all active members of a channel.
     pub async fn get_members(
         &self,
@@ -4618,9 +4651,11 @@ mod tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn test_usage_metrics_lock_has_single_owner_and_releases_on_drop() {
+        let database_url =
+            std::env::var("TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.into());
         let pool = PgPoolOptions::new()
             .max_connections(2)
-            .connect(TEST_DB_URL)
+            .connect(&database_url)
             .await
             .expect("connect to test DB");
         let first = Db::from_pool(pool.clone());

crates/buzz-db/src/migration.rs

@@ -560,7 +560,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 22);
+        assert_eq!(migrations.len(), 24);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -852,6 +852,33 @@ mod tests {
         assert!(ttl_refresh.contains("DEFERRABLE INITIALLY DEFERRED"));
         assert!(ttl_refresh.contains("clock_timestamp()"));
         assert!(ttl_refresh.contains("NEW.kind <> 9007"));
+
+        // T1b push gate: the match-queue trigger only enqueues when the
+        // community has an eligible lease, ordered against lease activations
+        // through the shared/exclusive per-community advisory lock.
+        assert_eq!(migrations[22].version, 23);
+        let push_gate = migrations[22].sql.as_str();
+        assert!(push_gate.contains("CREATE OR REPLACE FUNCTION enqueue_push_match_job"));
+        assert!(push_gate.contains("pg_advisory_xact_lock_shared"));
+        assert!(push_gate.contains("'buzz_push_gate:' || NEW.community_id::text"));
+        assert!(push_gate.contains("endpoint_enabled"));
+
+        // T1a repair: the TTL refresh trigger synchronizes on a shared
+        // per-channel advisory lock instead of FOR UPDATE on the channel row,
+        // so permanent-channel commits no longer serialize.
+        assert_eq!(migrations[23].version, 24);
+        let ttl_shared = migrations[23].sql.as_str();
+        assert!(ttl_shared
+            .contains("CREATE OR REPLACE FUNCTION refresh_channel_ttl_after_event_insert"));
+        assert!(ttl_shared.contains("pg_advisory_xact_lock_shared"));
+        assert!(ttl_shared.contains("'buzz_channel_ttl:' || NEW.community_id::text"));
+        // The row read must be a bare SELECT (comments describe the removed
+        // FOR UPDATE; the executable body must not reintroduce it).
+        assert!(ttl_shared.contains("SELECT ttl_seconds INTO channel_ttl"));
+        assert!(!strip_sql_comments(ttl_shared)
+            .to_lowercase()
+            .contains("for update"));
+        assert!(ttl_shared.contains("NEW.kind <> 9007"));
     }
 
     #[test]
@@ -1094,7 +1121,7 @@ mod tests {
         run_migrations(&pool)
             .await
             .expect("retry succeeds after operator repair");
-        assert_eq!(applied_versions(&pool).await.last().copied(), Some(22));
+        assert_eq!(applied_versions(&pool).await.last().copied(), Some(24));
     }
 
     #[tokio::test]