buzz-db: check events created_at floor in app writers (phase 1)

crates/buzz-db/src/error.rs

@@ -81,6 +81,13 @@ pub enum DbError {
     #[error("invalid timestamp: {0}")]
     InvalidTimestamp(i64),
 
+    /// A channel event is too old for replica-fence completeness. Retrying the
+    /// same signed event cannot make its timestamp newer. Like the mixed-version
+    /// trigger's SQLx error, this remains a database error at the relay boundary
+    /// (WS OK false / HTTP 500), rather than a different ingest rejection class.
+    #[error("database error: events.created_at is below the replica-fence floor")]
+    CreatedAtFloorRejected,
+
     /// A roster mutation would remove the last effective relay Operator,
     /// leaving no one able to administer the deployment through the API.
     /// The transaction is rolled back and the caller must add a replacement

crates/buzz-db/src/runtime/replica_fence.rs

@@ -3,12 +3,12 @@
 //! A read replica may serve a cursor page only when every row the page could
 //! contain is provably present on the replica. The proof rests on two parts:
 //!
-//! 1. **Commit-time floor** (migration 0021): a deferred constraint trigger
-//!    aborts, at COMMIT, any transaction inserting a channel-bearing `events`
-//!    row with `created_at` more than `floor` seconds before commit time
-//!    (`clock_timestamp()`, evaluated inside commit processing). Enforcement
-//!    is armed per session via the `buzz.created_at_floor` GUC, which the
-//!    relay's writer pool sets on every connection.
+//! 1. **Insert-time floor**: supported channel-event writers check `created_at`
+//!    against database `clock_timestamp()` on their admitted transaction for
+//!    each new row. Duplicate INSERTs need no floor check. This is the primary
+//!    guard. Migration 0021's deferred trigger and the writer-pool
+//!    `buzz.created_at_floor` GUC remain the mixed-version backstop until phase 2;
+//!    they also reject rows that age below the floor before COMMIT.
 //! 2. **Ordered heartbeat handshake** (this module): on one pinned writer
 //!    connection, one transaction first takes the exclusive replica-floor
 //!    advisory lock (draining compliant shared-lock channel writers), then
@@ -25,7 +25,7 @@
 //!    commit, so the replica session has replayed it;
 //!    (b) open at the activity scan — represented by `xact_start`, so it is
 //!    bounded by the `oldest_xact_start` term;
-//!    (c) started after the activity scan — its deferred floor guard runs
+//!    (c) started after the activity scan — its insert-time app guard runs
 //!    after `S`, so it cannot commit a row with
 //!    `created_at < S - floor`.
 //!    There is no fourth bucket. Each committed token `M` therefore proves a
@@ -53,9 +53,9 @@
 //! mismatch, or an observed token below every retained entry all route the
 //! request back to the writer — degraded capacity, never holes.
 //!
-//! Operational bypasses (sessions without the GUC, `session_replication_role
-//! = replica` restores) are outside the proof by design and require holding
-//! the fence closed for their duration; see `migrations/0021`.
+//! Raw SQL without the armed GUC and `session_replication_role = replica`
+//! restores bypass the supported app writers. These operational bypasses are
+//! outside the proof and require holding the fence closed; see `migrations/0021`.
 
 use std::collections::VecDeque;
 use std::sync::{Arc, Mutex};
@@ -67,14 +67,45 @@ use uuid::Uuid;
 
 use buzz_datastore_tracing::datastore_span;
 
-/// Seconds of `created_at` history the commit-time floor guard tolerates.
+/// Seconds of `created_at` history the app and mixed-version trigger tolerate.
 ///
 /// Must exceed the relay's ingest envelope (±900 s) by enough slack that a
 /// legitimately accepted event still commits within the floor even under
 /// slow validation/lock waits. The writer pool arms the guard with this value
 /// and the fence subtracts it; the two uses must never diverge.
 pub const CREATED_AT_FLOOR_SECS: i64 = 960;
 
+/// Enforce the channel-event floor on the admitted INSERT connection.
+///
+/// One SELECT keeps the five existing INSERT shapes intact. Ordinary/thread
+/// writers check after INSERT only when it inserted a row, before follow-ups;
+/// replacements check after replay exits, before superseding the old head.
+/// The caller must propagate failure and roll back the write transaction. Its
+/// database clock is sampled after BEGIN: transactions open at the probe scan
+/// are bounded by xact_start; transactions starting later sample after S. No
+/// commit hook, candidate tracking or ordering against TTL refresh is needed.
+/// NULL-channel rows never enter covered channel windows and need no query.
+pub(crate) async fn guard_event_created_at_floor(
+    tx: &mut crate::AdmittedTx,
+    created_at: DateTime<Utc>,
+    channel_id: Option<Uuid>,
+) -> crate::Result<()> {
+    if channel_id.is_none() {
+        return Ok(());
+    }
+    let below_floor: bool = sqlx::query_scalar(
+        "SELECT $1::timestamptz < clock_timestamp() - make_interval(secs => $2::double precision)",
+    )
+    .bind(created_at)
+    .bind(CREATED_AT_FLOOR_SECS as f64)
+    .fetch_one(tx.conn())
+    .await?;
+    if below_floor {
+        return Err(crate::DbError::CreatedAtFloorRejected);
+    }
+    Ok(())
+}
+
 /// Deployment-global advisory-lock key ordering compliant channel-event write
 /// transactions against the writer probe handshake.
 ///
@@ -608,7 +639,10 @@ async fn sample_writer(writer: &PgPool) -> Result<WriterSample, ProbeError> {
     //    prepared they have left `pg_stat_activity` but can still commit
     //    after the token. Their deferred floor guard already ran at PREPARE,
     //    so `pg_prepared_xacts.prepared` bounds their rows exactly like
-    //    `xact_start`; fold it into the same minimum.
+    //    `xact_start`; fold it into the same minimum. Only the retained
+    //    deferred trigger provides that PREPARE-time check: the insert-time
+    //    app guard runs earlier, and no app writer issues PREPARE TRANSACTION.
+    //    Dropping the trigger must keep this case covered.
     let row = sqlx::query(
         r#"
         SELECT

crates/buzz-db/src/runtime/tests.rs

@@ -948,27 +948,27 @@ fn signed_event_at(keys: &nostr::Keys, content: &str, secs: u64) -> nostr::Event
         .expect("sign event")
 }
 
+// Routing tests intentionally model stored history and divergent replicas.
+// Raw fixture SQL keeps them independent of serving-time floor admission.
 async fn insert_top_level(pool: &PgPool, community: Uuid, channel: Uuid, ev: &nostr::Event) {
-    let ts = chrono::DateTime::from_timestamp(ev.created_at.as_secs() as i64, 0).expect("valid ts");
-    event::insert_event_with_thread_metadata(
+    crate::test_support::insert_historical_event(
         pool,
         CommunityId::from_uuid(community),
         ev,
-        Some(channel),
-        Some(event::ThreadMetadataParams {
-            event_id: ev.id.as_bytes(),
-            event_created_at: ts,
-            channel_id: channel,
-            parent_event_id: None,
-            parent_event_created_at: None,
-            root_event_id: None,
-            root_event_created_at: None,
-            depth: 0,
-            broadcast: true,
-        }),
+        channel,
+    )
+    .await;
+    sqlx::query(
+        "INSERT INTO thread_metadata (community_id,event_id,event_created_at,channel_id,depth,broadcast) \
+         VALUES ($1,$2,to_timestamp($3),$4,0,true)",
     )
+    .bind(community)
+    .bind(ev.id.as_bytes().as_slice())
+    .bind(ev.created_at.as_secs() as f64)
+    .bind(channel)
+    .execute(pool)
     .await
-    .expect("insert top-level event");
+    .expect("seed top-level metadata");
 }
 
 async fn insert_thread_reply(
@@ -978,29 +978,37 @@ async fn insert_thread_reply(
     root: &nostr::Event,
     reply: &nostr::Event,
 ) {
-    let reply_ts =
-        chrono::DateTime::from_timestamp(reply.created_at.as_secs() as i64, 0).expect("valid ts");
-    let root_ts =
-        chrono::DateTime::from_timestamp(root.created_at.as_secs() as i64, 0).expect("valid ts");
-    event::insert_event_with_thread_metadata(
+    crate::test_support::insert_historical_event(
         pool,
         CommunityId::from_uuid(community),
         reply,
-        Some(channel),
-        Some(event::ThreadMetadataParams {
-            event_id: reply.id.as_bytes(),
-            event_created_at: reply_ts,
-            channel_id: channel,
-            parent_event_id: Some(root.id.as_bytes()),
-            parent_event_created_at: Some(root_ts),
-            root_event_id: Some(root.id.as_bytes()),
-            root_event_created_at: Some(root_ts),
-            depth: 1,
-            broadcast: false,
-        }),
+        channel,
+    )
+    .await;
+    sqlx::query(
+        "INSERT INTO thread_metadata \
+         (community_id,event_id,event_created_at,channel_id,parent_event_id,parent_event_created_at, \
+          root_event_id,root_event_created_at,depth,broadcast) \
+         VALUES ($1,$2,to_timestamp($3),$4,$5,to_timestamp($6),$5,to_timestamp($6),1,false)",
     )
+    .bind(community)
+    .bind(reply.id.as_bytes().as_slice())
+    .bind(reply.created_at.as_secs() as f64)
+    .bind(channel)
+    .bind(root.id.as_bytes().as_slice())
+    .bind(root.created_at.as_secs() as f64)
+    .execute(pool)
+    .await
+    .expect("seed reply metadata");
+    sqlx::query(
+        "UPDATE thread_metadata SET reply_count=reply_count+1,descendant_count=descendant_count+1, \
+         last_reply_at=now() WHERE community_id=$1 AND event_id=$2",
+    )
+    .bind(community)
+    .bind(root.id.as_bytes().as_slice())
+    .execute(pool)
     .await
-    .expect("insert reply");
+    .expect("seed root counters");
 }
 
 /// Composite thread cursor: 8-byte BE seconds + raw event id.

crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs

@@ -87,9 +87,7 @@ async fn thread_window_upper_fence_terminal_snapshot_and_fallback() {
         .sign_with_keys(&keys)
         .unwrap();
     for pool in [&writer, &replica] {
-        event::insert_event(pool, community, &edit, Some(channel))
-            .await
-            .unwrap();
+        crate::test_support::insert_historical_event(pool, community, &edit, channel).await;
         event::insert_event(pool, community, &deletion, None)
             .await
             .unwrap();

crates/buzz-db/src/store/channel_members.rs

@@ -488,6 +488,12 @@ impl LockedMemberSnapshot {
         }
         let tags = serde_json::to_value(&event.tags)?;
         validate_channel_roster_snapshot(&mut self.tx, channel_id, &tags).await?;
+        crate::replica_fence::guard_event_created_at_floor(
+            &mut self.tx,
+            created_at,
+            Some(channel_id),
+        )
+        .await?;
         sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id = $4 AND deleted_at IS NULL")
             .bind(community_id.as_uuid()).bind(kind).bind(pubkey.as_slice()).bind(channel_id)
             .execute(self.tx.conn()).await?;