feat(relay): group-commit batching for plain event inserts (F1)

crates/buzz-db/src/error.rs

@@ -48,6 +48,13 @@ pub enum DbError {
     /// A stored timestamp value could not be interpreted.
     #[error("invalid timestamp: {0}")]
     InvalidTimestamp(i64),
+
+    /// A transaction's COMMIT failed in a way that does not prove the
+    /// transaction aborted (e.g. the connection dropped while awaiting the
+    /// server's response). The write may or may not be durable; callers must
+    /// treat the outcome as unknown and retry idempotently.
+    #[error("commit outcome unknown: {0}")]
+    CommitOutcomeUnknown(String),
 }
 
 /// Convenience alias for `Result<T, DbError>`.

crates/buzz-db/src/event.rs

@@ -1000,7 +1000,7 @@ pub struct ThreadMetadataParams<'a> {
     pub broadcast: bool,
 }
 
-async fn insert_event_with_thread_metadata_tx(
+pub(crate) async fn insert_event_with_thread_metadata_tx(
     tx: &mut Transaction<'_, Postgres>,
     community_id: CommunityId,
     event: &Event,

crates/buzz-db/src/lib.rs

@@ -15,6 +15,8 @@ pub mod admin_moderation;
 pub mod api_token;
 /// Relay-scoped archived identity persistence (NIP-IA).
 pub mod archived_identities;
+/// Group-commit batching for plain event inserts.
+pub mod batch;
 /// Channel and membership persistence.
 pub mod channel;
 /// Direct message channel persistence.

crates/buzz-relay/src/config.rs

@@ -66,6 +66,13 @@ pub struct Config {
     pub max_connections: usize,
     /// Maximum number of concurrently executing message handlers.
     pub max_concurrent_handlers: usize,
+    /// Maximum plain event inserts coalesced into one group-commit
+    /// transaction (`BUZZ_WRITE_BATCH_MAX`). `0` disables batching and every
+    /// insert commits individually (the pre-batching behavior). Defaults to
+    /// `0`: benchmarks on the repaired-T1a base showed the single batch lane
+    /// cuts DB commits/msg ~25% but regresses p50/p99 latency at 500-1000 QPS,
+    /// so batching is opt-in until that trade-off is resolved.
+    pub write_batch_max: usize,
     /// Per-connection outbound message buffer size (number of messages).
     pub send_buffer_size: usize,
     /// Maximum inbound WebSocket frame size in bytes.
@@ -429,6 +436,14 @@ impl Config {
             .and_then(|v| v.parse().ok())
             .unwrap_or(1024);
 
+        // Default off (`0`): group-commit batching measurably regressed
+        // p50/p99 latency at 500-1000 QPS on the repaired-T1a base despite
+        // saving ~25% of DB commits. Opt in explicitly per deployment.
+        let write_batch_max = std::env::var("BUZZ_WRITE_BATCH_MAX")
+            .ok()
+            .and_then(|v| v.parse().ok())
+            .unwrap_or(0);
+
         let send_buffer_size = std::env::var("BUZZ_SEND_BUFFER")
             .ok()
             .and_then(|v| v.parse().ok())
@@ -835,6 +850,7 @@ impl Config {
             pairing_relay_url,
             max_connections,
             max_concurrent_handlers,
+            write_batch_max,
             send_buffer_size,
             max_frame_bytes,
             slow_client_grace_limit,

crates/buzz-relay/src/handlers/ingest.rs

@@ -537,6 +537,22 @@ impl ThreadMetadataOwned {
             broadcast: self.broadcast,
         }
     }
+
+    /// Owned clone for the group-commit batcher, which cannot borrow across
+    /// its queue.
+    pub fn to_batch_owned(&self) -> buzz_db::batch::ThreadMetadataOwned {
+        buzz_db::batch::ThreadMetadataOwned {
+            event_id: self.event_id.clone(),
+            event_created_at: self.event_created_at,
+            channel_id: self.channel_id,
+            parent_event_id: Some(self.parent_event_id.clone()),
+            parent_event_created_at: Some(self.parent_event_created_at),
+            root_event_id: Some(self.root_event_id.clone()),
+            root_event_created_at: Some(self.root_event_created_at),
+            depth: self.depth,
+            broadcast: self.broadcast,
+        }
+    }
 }
 
 /// Resolve NIP-10 thread ancestry from e-tags.
@@ -2308,17 +2324,43 @@ async fn ingest_event_inner(
             .await
             .map_err(|e| IngestError::Internal(format!("error: {e}")))?
     } else {
-        let thread_params = thread_meta.as_ref().map(|m| m.as_params());
-        match state
-            .db
-            .insert_event_with_thread_metadata(
-                tenant.community(),
-                &event,
-                channel_id,
-                thread_params,
-            )
-            .await
-        {
+        // Plain persistent event: route through the group-commit batcher when
+        // enabled so concurrent inserts share one COMMIT; semantics match the
+        // direct path exactly (see buzz_db::batch). Kind 9007 (channel
+        // create) stays direct: it is rare, carries the channel-compensation
+        // flow below, and the 0022 TTL trigger special-cases it. `None` from
+        // the batcher (disabled, never constructed, or worker gone) falls
+        // through to the direct path.
+        let mut batched = None;
+        if let Some(batcher) = &state.event_batcher {
+            if pre_created_channel.is_none() {
+                let batch_meta = thread_meta.as_ref().map(|m| m.to_batch_owned());
+                batched = batcher
+                    .insert_event_with_thread_metadata(
+                        tenant.community(),
+                        &event,
+                        channel_id,
+                        batch_meta,
+                    )
+                    .await;
+            }
+        }
+        let insert_result = match batched {
+            Some(result) => result,
+            None => {
+                let thread_params = thread_meta.as_ref().map(|m| m.as_params());
+                state
+                    .db
+                    .insert_event_with_thread_metadata(
+                        tenant.community(),
+                        &event,
+                        channel_id,
+                        thread_params,
+                    )
+                    .await
+            }
+        };
+        match insert_result {
             Ok(result) => result,
             Err(e) => {
                 // Compensate: if we pre-created a channel for kind:9007,

crates/buzz-relay/src/main.rs

@@ -978,6 +978,23 @@ async fn main() -> anyhow::Result<()> {
                     }
                 }
 
+                // Write-batcher coalescing counters (monotonic snapshots →
+                // absolute counters). Attempted vs. committed stay
+                // distinguishable: batch_* count only successful multi-event
+                // commits, aborted batches land in fallback_events.
+                if let Some(batcher) = &pool_state.event_batcher {
+                    let bs = batcher.stats();
+                    metrics::counter!("buzz_write_batch_commits_total").absolute(bs.batch_commits);
+                    metrics::counter!("buzz_write_batch_events_committed_total")
+                        .absolute(bs.batch_events_committed);
+                    metrics::counter!("buzz_write_batch_single_flushes_total")
+                        .absolute(bs.single_flushes);
+                    metrics::counter!("buzz_write_batch_fallback_events_total")
+                        .absolute(bs.fallback_events);
+                    metrics::counter!("buzz_write_batch_indeterminate_total")
+                        .absolute(bs.indeterminate_commits);
+                }
+
                 let rs = pool_state.redis_pool.status();
                 metrics::gauge!("buzz_redis_pool_available").set(rs.available as f64);
                 metrics::gauge!("buzz_redis_pool_size").set(rs.size as f64);

crates/buzz-relay/src/state.rs

@@ -436,6 +436,9 @@ pub struct AppState {
     pub config: Arc<Config>,
     /// Database connection pool.
     pub db: Db,
+    /// Group-commit writer for plain event inserts. `None` when disabled
+    /// (`BUZZ_WRITE_BATCH_MAX=0`); ingest then uses the direct insert path.
+    pub event_batcher: Option<buzz_db::batch::EventWriteBatcher>,
     /// Redis pool for readiness health checks.
     pub redis_pool: deadpool_redis::Pool,
     /// Audit event service, absent when audit logging is disabled.
@@ -635,9 +638,12 @@ impl AppState {
             Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
         let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone()));
         let audit_enabled = audit_arc.is_some();
+        let event_batcher = (config.write_batch_max > 0)
+            .then(|| buzz_db::batch::EventWriteBatcher::spawn(db.clone(), config.write_batch_max));
         let state = Self {
             config: Arc::new(config),
             db,
+            event_batcher,
             redis_pool,
             audit: audit_arc,
             pubsub,