From b1da831a6707a048b8acfeba815853f5f584f075 Mon Sep 17 00:00:00 2001
From: Elrond
 <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Date: Wed, 14 Oct 2026 13:49:57 -0400
Subject: [PATCH 1/7] buzz-db: check events created_at floor in the app writers
 (phase 1)

Every channel-bearing INSERT INTO events now checks created_at against
database clock_timestamp() minus CREATED_AT_FLOOR_SECS on the admitted
transaction before inserting, and returns the typed, non-retryable
DbError::CreatedAtFloorRejected. The check samples the clock after BEGIN,
so the replica-fence wall min(oldest_xact_start, S) - floor - margin
still covers it. Migration 0021's deferred trigger and the writer-pool
GUC stay in place as the mixed-version backstop until phase 2.

A source policy requires the check on every channel-bearing events
INSERT and forbids production UPDATEs of events.created_at/channel_id.
The admin seed script now refuses non-local PGHOST.

Refs BUZZ-316.

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
---
 crates/buzz-db/src/error.rs                   |   7 +
 crates/buzz-db/src/runtime/replica_fence.rs   |  49 ++-
 crates/buzz-db/src/runtime/tests.rs           |  74 +++--
 .../tests/thread_window_postgres_tests.rs     |   4 +-
 crates/buzz-db/src/store/channel_members.rs   |   6 +
 crates/buzz-db/src/store/event.rs             |  12 +-
 .../src/store/event_floor_postgres_tests.rs   | 313 ++++++++++++++++++
 crates/buzz-db/src/store/mod.rs               |   3 +
 .../participation_postgres_tests.rs           |   6 +-
 .../src/store/personal_read/postgres_tests.rs |   9 +-
 .../personal_read/threads_postgres_tests.rs   |   4 +-
 crates/buzz-db/src/store/replaceable.rs       |   4 +
 crates/buzz-db/src/test_support.rs            |  51 +++
 crates/buzz-db/tests/observability_source.rs  | 186 +++++++++++
 scripts/seed-admin-dashboard.sh               |   9 +
 15 files changed, 677 insertions(+), 60 deletions(-)
 create mode 100644 crates/buzz-db/src/store/event_floor_postgres_tests.rs

diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs
index 1627521d7e2..a85470e8491 100644
--- a/crates/buzz-db/src/error.rs
+++ b/crates/buzz-db/src/error.rs
@@ -114,6 +114,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
diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs
index 7278fcc3b2b..1426791c953 100644
--- a/crates/buzz-db/src/runtime/replica_fence.rs
+++ b/crates/buzz-db/src/runtime/replica_fence.rs
@@ -3,12 +3,11 @@
 //! 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 before
+//!    INSERT. This is the primary guard. Migration 0021's deferred trigger and
+//!    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 +24,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
@@ -57,9 +56,9 @@
 //! keeps failing, the newest entry ages past [`FENCE_STALENESS`] and the
 //! fence closes.
 //!
-//! 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};
@@ -73,7 +72,7 @@ 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
@@ -81,6 +80,34 @@ use buzz_datastore_tracing::datastore_span;
 /// 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 preceding SELECT keeps the five existing INSERT shapes intact. 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.
 ///
diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs
index 23828f92d15..6fa4a720a76 100644
--- a/crates/buzz-db/src/runtime/tests.rs
+++ b/crates/buzz-db/src/runtime/tests.rs
@@ -913,27 +913,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(
@@ -943,29 +943,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.
diff --git a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs
index b5ef396b746..93e61bde7a6 100644
--- a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs
+++ b/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();
diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs
index e0487877f9b..f52c8b092d8 100644
--- a/crates/buzz-db/src/store/channel_members.rs
+++ b/crates/buzz-db/src/store/channel_members.rs
@@ -503,6 +503,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(superseded_with_mentions_sql!(
             "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"
         ))
diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs
index 586a8514da2..a4be1edb2e0 100644
--- a/crates/buzz-db/src/store/event.rs
+++ b/crates/buzz-db/src/store/event.rs
@@ -498,6 +498,7 @@ async fn insert_event_on(
     let not_before = extract_not_before(event);
     crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
         .await?;
+    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
     let result = sqlx::query(
         r#"
         INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
@@ -1964,6 +1965,7 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
     crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
         .await?;
 
+    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
     let result = sqlx::query(
         r#"
         INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
@@ -5758,7 +5760,15 @@ mod postgres_tests {
     // ─── canvas CAS tests ─────────────────────────────────────────────────────
 
     fn make_canvas_event_at(content: &str, created_at: u64) -> nostr::Event {
-        make_event_at(buzz_core::kind::KIND_CANVAS as u16, content, created_at)
+        // Keep the CAS fixtures' logical seconds and ties, anchored at a fresh
+        // timestamp so they exercise the serving writer's floor admission.
+        static BASE: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
+        let base = BASE.get_or_init(|| nostr::Timestamp::now().as_secs() - 1000);
+        make_event_at(
+            buzz_core::kind::KIND_CANVAS as u16,
+            content,
+            base + created_at,
+        )
     }
 
     /// Build two canvas events sharing `created_at`, returned `(lower, higher)`
diff --git a/crates/buzz-db/src/store/event_floor_postgres_tests.rs b/crates/buzz-db/src/store/event_floor_postgres_tests.rs
new file mode 100644
index 00000000000..857d91ba686
--- /dev/null
+++ b/crates/buzz-db/src/store/event_floor_postgres_tests.rs
@@ -0,0 +1,313 @@
+//! App-floor regressions with every floor-trigger clone disabled in the
+//! nextest-provided per-test database, so the DB backstop cannot mask a bypass.
+
+use buzz_core::CommunityId;
+use chrono::{DateTime, Utc};
+use nostr::{Event, EventBuilder, Keys, Kind, Tag, Timestamp};
+use sqlx::postgres::PgPoolOptions;
+use uuid::Uuid;
+
+use crate::replica_fence::{
+    probe_once, ReplicaFence, CREATED_AT_FLOOR_SECS, FENCE_CLOCK_MARGIN_SECS,
+};
+use crate::{event, AdmittedTx, Db, DbError};
+
+async fn fixture() -> (Db, CommunityId, Uuid, Keys) {
+    let pool = PgPoolOptions::new()
+        .max_connections(4)
+        .connect(&crate::test_support::database_url())
+        .await
+        .expect("isolated test database");
+    sqlx::query("ALTER TABLE events DISABLE TRIGGER events_created_at_floor")
+        .execute(&pool)
+        .await
+        .expect("disable floor trigger recursively on the partitioned table");
+    let (total, enabled): (i64, i64) = sqlx::query_as(
+        "SELECT count(*), count(*) FILTER (WHERE tgenabled <> 'D') FROM pg_trigger \
+         WHERE tgname = 'events_created_at_floor' \
+         AND tgrelid IN (SELECT relid FROM pg_partition_tree('events'))",
+    )
+    .fetch_one(&pool)
+    .await
+    .expect("check every trigger clone");
+    assert!(total > 0, "the test must disable an existing backstop");
+    assert_eq!(enabled, 0, "no partition may mask a missing app check");
+    let community = CommunityId::from_uuid(Uuid::new_v4());
+    sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+        .bind(community.as_uuid())
+        .bind(format!("floor-{}.invalid", community.as_uuid()))
+        .execute(&pool)
+        .await
+        .expect("community");
+    let channel = Uuid::new_v4();
+    let keys = Keys::generate();
+    sqlx::query(
+        "INSERT INTO channels (id, community_id, name, created_by) VALUES ($1,$2,'floor',$3)",
+    )
+    .bind(channel)
+    .bind(community.as_uuid())
+    .bind(keys.public_key().to_bytes().as_slice())
+    .execute(&pool)
+    .await
+    .expect("channel");
+    (Db::from_pool(pool), community, channel, keys)
+}
+
+async fn database_clock(db: &Db) -> DateTime<Utc> {
+    sqlx::query_scalar("SELECT clock_timestamp()")
+        .fetch_one(db.pool())
+        .await
+        .expect("database clock")
+}
+
+fn signed(keys: &Keys, kind: u16, channel: Uuid, time: DateTime<Utc>) -> Event {
+    EventBuilder::new(Kind::Custom(kind), "floor-test")
+        .tags([Tag::parse(["d", &channel.to_string()]).expect("coordinate")])
+        .custom_created_at(Timestamp::from(time.timestamp() as u64))
+        .sign_with_keys(keys)
+        .expect("signed event")
+}
+
+async fn count(db: &Db, community: CommunityId, event: &Event) -> i64 {
+    sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2")
+        .bind(community.as_uuid())
+        .bind(event.id.as_bytes().as_slice())
+        .fetch_one(db.pool())
+        .await
+        .expect("event count")
+}
+
+async fn begin(db: &Db, community: CommunityId) -> AdmittedTx {
+    db.begin_event_write_transaction(community)
+        .await
+        .expect("admission")
+}
+
+fn rejected<T: std::fmt::Debug>(result: crate::Result<T>) {
+    assert!(
+        matches!(result, Err(DbError::CreatedAtFloorRejected)),
+        "expected the typed app-floor rejection, got {result:?}"
+    );
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn insert_event_floor_rejects_old_and_accepts_fresh() {
+    let (db, community, channel, keys) = fixture().await;
+    let now = database_clock(&db).await;
+    let old = signed(
+        &keys,
+        9,
+        channel,
+        now - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60),
+    );
+    rejected(event::insert_event(db.pool(), community, &old, Some(channel)).await);
+    assert_eq!(count(&db, community, &old).await, 0);
+    let fresh = signed(&keys, 9, channel, now);
+    assert!(
+        event::insert_event(db.pool(), community, &fresh, Some(channel))
+            .await
+            .expect("fresh")
+            .1
+    );
+    assert_eq!(count(&db, community, &fresh).await, 1);
+    // Channel-NULL history is structurally outside covered windows.
+    assert!(
+        event::insert_event(db.pool(), community, &old, None)
+            .await
+            .expect("NULL-channel history")
+            .1
+    );
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn thread_metadata_event_floor_rejects_old_and_accepts_fresh() {
+    let (db, community, channel, keys) = fixture().await;
+    let now = database_clock(&db).await;
+    for (time, old) in [
+        (
+            now - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60),
+            true,
+        ),
+        (now, false),
+    ] {
+        let event = signed(&keys, 9, channel, time);
+        let meta = event::ThreadMetadataParams {
+            event_id: event.id.as_bytes().as_slice(),
+            event_created_at: time,
+            channel_id: channel,
+            parent_event_id: None,
+            parent_event_created_at: None,
+            root_event_id: None,
+            root_event_created_at: None,
+            depth: 0,
+            broadcast: false,
+        };
+        let mut tx = begin(&db, community).await;
+        let result =
+            event::insert_event_with_thread_metadata_tx(&mut tx, &event, Some(channel), Some(meta))
+                .await;
+        if old {
+            rejected(result);
+            tx.rollback().await.expect("rollback refusal");
+            assert_eq!(count(&db, community, &event).await, 0);
+        } else {
+            assert!(result.expect("fresh").1);
+            tx.commit().await.expect("commit");
+            assert_eq!(count(&db, community, &event).await, 1);
+        }
+    }
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn parameterized_event_floor_rejects_old_and_accepts_fresh() {
+    let (db, community, channel, keys) = fixture().await;
+    let now = database_clock(&db).await;
+    let old = signed(
+        &keys,
+        30078,
+        channel,
+        now - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60),
+    );
+    rejected(
+        db.replace_parameterized_event(community, &old, &channel.to_string(), Some(channel))
+            .await,
+    );
+    assert_eq!(count(&db, community, &old).await, 0);
+    let fresh = signed(&keys, 30078, channel, now);
+    assert!(
+        db.replace_parameterized_event(community, &fresh, &channel.to_string(), Some(channel))
+            .await
+            .expect("fresh")
+            .1
+    );
+    assert_eq!(count(&db, community, &fresh).await, 1);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn addressable_event_floor_rejects_old_and_accepts_fresh() {
+    let (db, community, channel, keys) = fixture().await;
+    let now = database_clock(&db).await;
+    let old = signed(
+        &keys,
+        39000,
+        channel,
+        now - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60),
+    );
+    rejected(
+        db.replace_addressable_event(community, &old, Some(channel))
+            .await,
+    );
+    assert_eq!(count(&db, community, &old).await, 0);
+    let fresh = signed(&keys, 39000, channel, now);
+    assert!(
+        db.replace_addressable_event(community, &fresh, Some(channel))
+            .await
+            .expect("fresh")
+            .1
+    );
+    assert_eq!(count(&db, community, &fresh).await, 1);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn member_snapshot_event_floor_rejects_old_and_accepts_fresh() {
+    let (db, community, channel, keys) = fixture().await;
+    let now = database_clock(&db).await;
+    let old = signed(
+        &keys,
+        39002,
+        channel,
+        now - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60),
+    );
+    let mut snapshot = db
+        .lock_member_snapshot(community, channel, &keys.public_key().to_bytes())
+        .await
+        .expect("snapshot");
+    rejected(snapshot.replace_member_event(&old).await);
+    // A refusal must leave the held snapshot usable and write no event.
+    assert_eq!(count(&db, community, &old).await, 0);
+    let fresh = signed(&keys, 39002, channel, now);
+    assert!(
+        snapshot
+            .replace_member_event(&fresh)
+            .await
+            .expect("fresh")
+            .1
+    );
+    snapshot.release().await.expect("commit snapshot");
+    assert_eq!(count(&db, community, &fresh).await, 1);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn cluster_global_event_floor_held_transaction_cannot_write_below_probe_wall() {
+    let (db, community, channel, keys) = fixture().await;
+    let mut tx = begin(&db, community).await;
+    let start: DateTime<Utc> = sqlx::query_scalar("SELECT now()")
+        .fetch_one(tx.conn())
+        .await
+        .expect("transaction start");
+    let entry = probe_once(db.pool(), &ReplicaFence::new())
+        .await
+        .expect("heartbeat while writer is open");
+    assert!(
+        entry.fence_wall
+            <= start - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + FENCE_CLOCK_MARGIN_SECS)
+    );
+    let below_wall = signed(
+        &keys,
+        9,
+        channel,
+        entry.fence_wall - chrono::Duration::seconds(1),
+    );
+    rejected(event::insert_event_in_transaction(&mut tx, &below_wall, Some(channel)).await);
+    // Sample just after the probe; a near-floor accepted insert is above its
+    // wall even when this transaction commits after the heartbeat.
+    let fresh = signed(
+        &keys,
+        9,
+        channel,
+        database_clock(&db).await - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS - 5),
+    );
+    let created_at =
+        DateTime::from_timestamp(fresh.created_at.as_secs() as i64, 0).expect("timestamp");
+    assert!(created_at > entry.fence_wall);
+    assert!(
+        event::insert_event_in_transaction(&mut tx, &fresh, Some(channel))
+            .await
+            .expect("near floor")
+            .1
+    );
+    tx.commit().await.expect("commit after probe");
+    assert_eq!(count(&db, community, &below_wall).await, 0);
+    assert_eq!(count(&db, community, &fresh).await, 1);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn event_floor_samples_statement_clock_after_transaction_start() {
+    let (db, community, channel, keys) = fixture().await;
+    let mut tx = begin(&db, community).await;
+    // The smallest integral event timestamp inside the floor at BEGIN.
+    let secs: i64 =
+        sqlx::query_scalar("SELECT ceil(extract(epoch FROM now()) - $1::double precision)::bigint")
+            .bind(CREATED_AT_FLOOR_SECS as f64)
+            .fetch_one(tx.conn())
+            .await
+            .expect("BEGIN boundary");
+    let time = DateTime::from_timestamp(secs, 0).expect("timestamp");
+    // Wait on the DB clock until this exact boundary expires (at most 1.1 s),
+    // not an arbitrary delay and not a Rust clock. now() remains frozen inside.
+    tokio::time::timeout(std::time::Duration::from_secs(3),
+        sqlx::query("SELECT pg_sleep(GREATEST(0, extract(epoch FROM ($1::timestamptz + make_interval(secs => $2::double precision) - clock_timestamp()))) + 0.1)")
+            .bind(time).bind(CREATED_AT_FLOOR_SECS as f64).execute(tx.conn()))
+        .await.expect("bounded boundary wait").expect("database wait");
+    let event = signed(&keys, 9, channel, time);
+    rejected(event::insert_event_in_transaction(&mut tx, &event, Some(channel)).await);
+    tx.rollback().await.expect("rollback");
+    assert_eq!(count(&db, community, &event).await, 0);
+}
diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs
index d1cb647694b..6fad41294c4 100644
--- a/crates/buzz-db/src/store/mod.rs
+++ b/crates/buzz-db/src/store/mod.rs
@@ -41,6 +41,9 @@ pub mod user_deletion_fences;
 #[cfg(test)]
 mod event_follow_up_postgres_tests;
 
+#[cfg(test)]
+mod event_floor_postgres_tests;
+
 /// Home feed queries.
 pub mod feed;
 /// Git repository name registry (NIP-34 kind:30617).
diff --git a/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs
index 5a3d3a0acad..2800b9b1804 100644
--- a/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs
+++ b/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs
@@ -70,10 +70,8 @@ impl World {
             .custom_created_at(Timestamp::from(at))
             .sign_with_keys(author)
             .unwrap();
-        self.db
-            .insert_event(self.community, &event, Some(channel))
-            .await
-            .unwrap();
+        crate::test_support::insert_historical_event(&self.pool, self.community, &event, channel)
+            .await;
         event
     }
 
diff --git a/crates/buzz-db/src/store/personal_read/postgres_tests.rs b/crates/buzz-db/src/store/personal_read/postgres_tests.rs
index 99da1ff848c..81423e445f5 100644
--- a/crates/buzz-db/src/store/personal_read/postgres_tests.rs
+++ b/crates/buzz-db/src/store/personal_read/postgres_tests.rs
@@ -164,8 +164,9 @@ async fn personal_read_sidebar_author_window_excludes_expired_and_late_old_messa
         "an empty author-time window has nothing unread"
     );
     assert_eq!(row.latest_id.as_deref(), Some(latest.id.to_hex().as_str()));
-    // Acceptance time is irrelevant: a message accepted now with an author
-    // time beyond the horizon stays out; one authored now is counted.
+    // Stored historical data with an author time beyond the horizon stays out;
+    // one authored now is counted. The serving writer now rejects the old
+    // timestamp, so seed the historical row through unarmed fixture SQL.
     let now = nostr::Timestamp::now().as_secs();
     for (age, unread, mentions) in [(40 * 86400, false, 0), (0, true, 1)] {
         let event = EventBuilder::new(Kind::Custom(9), "accepted now")
@@ -173,9 +174,7 @@ async fn personal_read_sidebar_author_window_excludes_expired_and_late_old_messa
             .custom_created_at(nostr::Timestamp::from(now - age))
             .sign_with_keys(&Keys::generate())
             .unwrap();
-        db.insert_event(community, &event, Some(channel))
-            .await
-            .unwrap();
+        crate::test_support::insert_historical_event(&pool, community, &event, channel).await;
         let row = &sidebar(&db, community, &actor).await.channels[0];
         assert_eq!((row.unread, row.mentions), (unread, mentions), "age={age}");
     }
diff --git a/crates/buzz-db/src/store/personal_read/threads_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/threads_postgres_tests.rs
index 2bb86aab8da..09786f55856 100644
--- a/crates/buzz-db/src/store/personal_read/threads_postgres_tests.rs
+++ b/crates/buzz-db/src/store/personal_read/threads_postgres_tests.rs
@@ -32,9 +32,7 @@ async fn post_as(
         .custom_created_at(nostr::Timestamp::from(at))
         .sign_with_keys(author)
         .unwrap();
-    db.insert_event(community, &event, Some(channel))
-        .await
-        .unwrap();
+    crate::test_support::insert_historical_event(&db.pool, community, &event, channel).await;
     event
 }
 
diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs
index f6a0368b3d8..e18253b65d6 100644
--- a/crates/buzz-db/src/store/replaceable.rs
+++ b/crates/buzz-db/src/store/replaceable.rs
@@ -251,6 +251,8 @@ async fn replace_parameterized_event_in_transaction_impl(
         ));
     }
 
+    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
+
     // Everything from the supersede through the mention index runs under one
     // savepoint on the admitted transaction. A duplicate insert or any failure
     // (including mention indexing) rolls the whole replacement back and leaves
@@ -478,6 +480,8 @@ pub(crate) async fn replace_addressable_event_under_lock(
     crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
         .await?;
 
+    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
+
     // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety.
     sqlx::query(superseded_with_mentions_sql!(
         "UPDATE events SET deleted_at = NOW() \
diff --git a/crates/buzz-db/src/test_support.rs b/crates/buzz-db/src/test_support.rs
index 9fc8d9fec7c..fb9b0b56e28 100644
--- a/crates/buzz-db/src/test_support.rs
+++ b/crates/buzz-db/src/test_support.rs
@@ -318,6 +318,57 @@ where
     result.unwrap().expect("sweep task")
 }
 
+/// Seed historical read fixtures through raw SQL on an unarmed test pool.
+/// Supported serving APIs reject these old timestamps; this represents data
+/// already in storage, while retaining admission, mentions and event follow-ups.
+#[cfg(test)] // Keep raw fixture SQL visibly test-only to the source-policy scan.
+pub(crate) async fn insert_historical_event(
+    pool: &sqlx::PgPool,
+    community: buzz_core::CommunityId,
+    event: &nostr::Event,
+    channel: uuid::Uuid,
+) {
+    let mut tx = crate::begin_community_write_transaction(
+        pool,
+        community,
+        crate::observability::WriterOperation::EventWrite,
+    )
+    .await
+    .expect("admit historical fixture");
+    sqlx::query(
+        "INSERT INTO events (community_id,id,pubkey,created_at,kind,tags,content,sig,channel_id,d_tag) \
+         VALUES ($1,$2,$3,to_timestamp($4),$5,$6,$7,$8,$9,$10)",
+    )
+    .bind(community.as_uuid())
+    .bind(event.id.as_bytes().as_slice())
+    .bind(event.pubkey.to_bytes().as_slice())
+    .bind(event.created_at.as_secs() as f64)
+    .bind(i32::from(event.kind.as_u16()))
+    .bind(serde_json::to_value(&event.tags).expect("fixture tags"))
+    .bind(&event.content)
+    .bind(event.sig.serialize().as_slice())
+    .bind(channel)
+    .bind(crate::event::extract_d_tag(event))
+    .execute(tx.conn())
+    .await
+    .expect("insert historical fixture");
+    let created_at = chrono::DateTime::from_timestamp(event.created_at.as_secs() as i64, 0)
+        .expect("fixture created_at");
+    crate::store::event_follow_up::after_admitted_insert(
+        &mut tx,
+        event.id.as_bytes(),
+        created_at,
+        i32::from(event.kind.as_u16()),
+        Some(channel),
+    )
+    .await
+    .expect("fixture follow-ups");
+    crate::insert_mentions_in_transaction(&mut tx, event, Some(channel))
+        .await
+        .expect("fixture mentions");
+    tx.commit().await.expect("commit historical fixture");
+}
+
 /// Move a community fixture into a deletion lifecycle state the way the
 /// executor does: the tombstone trigger only accepts lifecycle changes from a
 /// transaction carrying the executor GUCs for that community, including its
diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs
index f8969b0f653..7c1960dd330 100644
--- a/crates/buzz-db/tests/observability_source.rs
+++ b/crates/buzz-db/tests/observability_source.rs
@@ -2229,3 +2229,189 @@ fn every_search_hydrator_preserves_the_exact_hit_timestamp() {
         );
     }
 }
+
+// These startup probes always roll back; they deliberately exercise the DB
+// backstops rather than the app check. The two serving writers are exempt only
+// while their INSERT remains structurally channel-NULL (checked below).
+const EVENT_FLOOR_PROBE_EXCEPTIONS: [&str; 2] = [
+    "pub async fn verify_floor_guard_behavior(",
+    "pub async fn verify_channel_roster_fence_behavior(",
+];
+
+fn is_channel_null_event_writer(function: &str) -> bool {
+    let header = function_header(function);
+    if header.starts_with("pub async fn publish_nip43_membership_locked(") {
+        // Kind 13534: the channel bind is always NULL.
+        return function.contains(".bind::<Option<Uuid>>(None)");
+    }
+    if header.starts_with("pub async fn accept_lease_event(") {
+        // Kind 30350: channel_id is the literal NULL in VALUES.
+        return normalized_sql(function).contains("now(),null,$8)");
+    }
+    false
+}
+
+fn updates_event_floor_columns(function: &str) -> bool {
+    let normalized = normalized_sql(function);
+    ["update events set ", "update public.events set "]
+        .iter()
+        .any(|marker| {
+            normalized.match_indices(marker).any(|(index, marker)| {
+                let assignments = normalized[index + marker.len()..]
+                    .split(" where ")
+                    .next()
+                    .expect("split has first part");
+                assignments.split(',').any(|assignment| {
+                    assignment.split_once('=').is_some_and(|(column, _)| {
+                        matches!(column.trim(), "created_at" | "channel_id")
+                    })
+                })
+            })
+        })
+}
+
+fn event_floor_policy_violations(production: &str) -> Vec<String> {
+    function_slices(production)
+        .into_iter()
+        .filter(|function| {
+            let header = function_header(function);
+            if EVENT_FLOOR_PROBE_EXCEPTIONS
+                .iter()
+                .any(|exception| header.starts_with(exception))
+            {
+                return false;
+            }
+            updates_event_floor_columns(function)
+                || (inserts_events_row(function)
+                    && !is_channel_null_event_writer(function)
+                    && !mask_comments_and_literals(function)
+                        .contains("replica_fence::guard_event_created_at_floor("))
+        })
+        .map(|function| function_header(function).to_owned())
+        .collect()
+}
+
+#[test]
+fn event_floor_policy_rejects_unguarded_inserts_and_floor_column_updates() {
+    let source = r#"
+async fn missing_guard(tx: &mut AdmittedTx) {
+    sqlx::query("INSERT INTO events (id) VALUES ($1)");
+    // replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?;
+}
+async fn guarded(tx: &mut AdmittedTx) {
+    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel).await?;
+    sqlx::query("INSERT INTO public.events (id) VALUES ($1)");
+}
+async fn rewrite_time(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE events SET tags = $1, created_at = $2 WHERE id = $3");
+}
+async fn attach_channel(tx: &mut AdmittedTx) {
+    sqlx::query("uPdAtE public.events\nSET channel_id=$1 WHERE id=$2");
+}
+async fn delete_only(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE events SET deleted_at=now() WHERE channel_id=$1 AND created_at<$2");
+}
+"#;
+    assert_eq!(
+        event_floor_policy_violations(source),
+        [
+            "async fn missing_guard(tx: &mut AdmittedTx) {",
+            "async fn rewrite_time(tx: &mut AdmittedTx) {",
+            "async fn attach_channel(tx: &mut AdmittedTx) {",
+        ]
+    );
+}
+
+#[test]
+fn every_production_channel_event_insert_checks_the_floor_and_never_rewrites_it() {
+    fn collect(dir: &std::path::Path, files: &mut Vec<std::path::PathBuf>) {
+        for entry in std::fs::read_dir(dir).expect("read source directory") {
+            let path = entry.expect("source entry").path();
+            if path.is_dir() {
+                collect(&path, files);
+            } else if should_scan_guarded_write_source_file(&path) {
+                files.push(path);
+            }
+        }
+    }
+    let mut files = Vec::new();
+    collect(
+        &std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src"),
+        &mut files,
+    );
+    let mut guarded_writers = 0;
+    let mut violations = Vec::new();
+    for path in files {
+        let production = strip_cfg_test_items(&std::fs::read_to_string(&path).expect("source"));
+        guarded_writers += function_slices(&production)
+            .into_iter()
+            .filter(|function| {
+                inserts_events_row(function)
+                    && mask_comments_and_literals(function)
+                        .contains("replica_fence::guard_event_created_at_floor(")
+            })
+            .count();
+        violations.extend(
+            event_floor_policy_violations(&production)
+                .into_iter()
+                .map(|header| format!("{}: {header}", path.display())),
+        );
+    }
+    assert!(
+        violations.is_empty(),
+        "event floor policy violations: {violations:?}"
+    );
+    assert!(
+        guarded_writers >= 5,
+        "expected five channel-bearing event writers, found {guarded_writers}"
+    );
+}
+
+#[test]
+fn event_floor_guard_uses_the_database_statement_clock_and_shared_floor() {
+    let source = strip_cfg_test_items(include_str!("../src/runtime/replica_fence.rs"));
+    let guard = function_slices(&source)
+        .into_iter()
+        .find(|function| function_header(function).contains("fn guard_event_created_at_floor("))
+        .expect("app floor guard");
+    assert!(
+        guard.contains("SELECT $1::timestamptz < clock_timestamp()"),
+        "the guard must sample the database statement clock, after BEGIN"
+    );
+    assert!(
+        guard.contains(".bind(CREATED_AT_FLOOR_SECS as f64)"),
+        "app check, GUC and fence wall must share the same floor constant"
+    );
+}
+
+#[cfg(unix)]
+#[test]
+fn admin_seed_refuses_remote_database_before_invoking_psql() {
+    use std::os::unix::fs::PermissionsExt;
+    let root = std::env::temp_dir().join(format!("buzz-seed-policy-{}", uuid::Uuid::new_v4()));
+    let scripts = root.join("scripts");
+    std::fs::create_dir_all(&scripts).expect("scratch scripts directory");
+    let script = scripts.join("seed-admin-dashboard.sh");
+    std::fs::write(
+        &script,
+        include_str!("../../../scripts/seed-admin-dashboard.sh"),
+    )
+    .expect("copy script without developer .env");
+    let psql = root.join("psql");
+    std::fs::write(&psql, "#!/bin/sh\ntouch \"$SEED_PSQL_WITNESS\"\nexit 1\n").expect("stub psql");
+    std::fs::set_permissions(&psql, std::fs::Permissions::from_mode(0o755))
+        .expect("executable stub");
+    let witness = root.join("psql-was-invoked");
+    let output = std::process::Command::new("/bin/bash")
+        .arg(script)
+        .env("PGHOST", "shared-db.example.invalid")
+        .env("PATH", format!("{}:/usr/bin:/bin", root.display()))
+        .env("SEED_PSQL_WITNESS", &witness)
+        .output()
+        .expect("run local seed guard");
+    let invoked = witness.exists();
+    std::fs::remove_dir_all(root).expect("remove scratch fixture");
+    assert!(!invoked, "a non-local target reached psql");
+    assert!(!output.status.success());
+    assert!(String::from_utf8_lossy(&output.stderr).contains("requires a local PostgreSQL target"));
+}
diff --git a/scripts/seed-admin-dashboard.sh b/scripts/seed-admin-dashboard.sh
index 774b9f75128..9caad1f2d20 100755
--- a/scripts/seed-admin-dashboard.sh
+++ b/scripts/seed-admin-dashboard.sh
@@ -19,6 +19,15 @@ export PGUSER="${PGUSER:-buzz}"
 export PGPASSWORD="${PGPASSWORD:-buzz_dev}"
 export PGDATABASE="${PGDATABASE:-buzz}"
 
+# This raw fixture writer is unarmed and must never target a live replica fence.
+case "${PGHOST}" in
+  localhost|127.0.0.1|::1|/*) ;;
+  *)
+    echo "error: admin seed requires a local PostgreSQL target (PGHOST=${PGHOST})" >&2
+    exit 1
+    ;;
+esac
+
 if command -v psql >/dev/null 2>&1; then
   run_psql() {
     PGPASSWORD="${PGPASSWORD}" psql -h "${PGHOST}" -p "${PGPORT}" \

From 4c0224ff26d11939a28aef55c8d4176f41317d5f Mon Sep 17 00:00:00 2001
From: Elrond
 <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Date: Wed, 14 Oct 2026 14:25:17 -0400
Subject: [PATCH 2/7] buzz-db: check the floor only on newly inserted event
 rows

Review rework for the insert-time created_at floor (BUZZ-316 phase 1).

- Ordinary and thread event writers now run the floor check after their
  ON CONFLICT DO NOTHING INSERT, only when a row was inserted and before
  follow-ups. Replaying an already-stored id stays Duplicate even when
  its created_at has aged below the floor, matching the trigger.
- Source policy: UPDATE rule catches ONLY, aliases and row-constructor
  column lists; quoted identifiers are normalized; the floor policy
  scans every crate's src via the shared workspace walk.
- NIP-43 membership insert writes a literal NULL channel_id, pinned
  positionally; the bind-based exemption is gone.
- Canvas CAS fixtures take a per-test base instead of a process-wide
  OnceLock.
- seed-admin-dashboard.sh also refuses PGHOSTADDR, PGSERVICE and
  multi-host PGHOST.

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
---
 crates/buzz-db/src/runtime/replica_fence.rs   |  14 +-
 crates/buzz-db/src/store/event.rs             |  95 +++++---
 .../src/store/event_floor_postgres_tests.rs   |  70 ++++++
 crates/buzz-db/src/store/relay_members.rs     |   3 +-
 crates/buzz-db/tests/observability_source.rs  | 205 ++++++++++++++----
 scripts/seed-admin-dashboard.sh               |   4 +
 6 files changed, 305 insertions(+), 86 deletions(-)

diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs
index 1426791c953..cbe8e1546a5 100644
--- a/crates/buzz-db/src/runtime/replica_fence.rs
+++ b/crates/buzz-db/src/runtime/replica_fence.rs
@@ -4,10 +4,11 @@
 //! contain is provably present on the replica. The proof rests on two parts:
 //!
 //! 1. **Insert-time floor**: supported channel-event writers check `created_at`
-//!    against database `clock_timestamp()` on their admitted transaction before
-//!    INSERT. This is the primary guard. Migration 0021's deferred trigger and
-//!    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.
+//!    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
@@ -82,7 +83,10 @@ pub const CREATED_AT_FLOOR_SECS: i64 = 960;
 
 /// Enforce the channel-event floor on the admitted INSERT connection.
 ///
-/// One preceding SELECT keeps the five existing INSERT shapes intact. Its
+/// 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.
diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs
index a4be1edb2e0..aafee90d9e9 100644
--- a/crates/buzz-db/src/store/event.rs
+++ b/crates/buzz-db/src/store/event.rs
@@ -455,7 +455,8 @@ pub async fn insert_event(
 ///
 /// This is the transaction-composition seam for callers that must keep the
 /// event insert open while performing related work. The caller owns commit or
-/// rollback.
+/// rollback. On error, roll back or drop the transaction: the event row may
+/// already have been inserted before the app floor check rejected it.
 pub async fn insert_event_in_transaction(
     tx: &mut AdmittedTx,
     event: &Event,
@@ -498,7 +499,6 @@ async fn insert_event_on(
     let not_before = extract_not_before(event);
     crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
         .await?;
-    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
     let result = sqlx::query(
         r#"
         INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
@@ -523,6 +523,7 @@ async fn insert_event_on(
 
     let was_inserted = result.rows_affected() > 0;
     if was_inserted {
+        crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
         crate::store::event_follow_up::after_admitted_insert(
             tx,
             id_bytes.as_slice(),
@@ -1965,7 +1966,6 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
     crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
         .await?;
 
-    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
     let result = sqlx::query(
         r#"
         INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
@@ -1991,6 +1991,7 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
     let was_inserted = result.rows_affected() > 0;
 
     if was_inserted {
+        crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
         crate::store::event_follow_up::after_admitted_insert(
             tx,
             id_bytes.as_slice(),
@@ -5759,11 +5760,8 @@ mod postgres_tests {
 
     // ─── canvas CAS tests ─────────────────────────────────────────────────────
 
-    fn make_canvas_event_at(content: &str, created_at: u64) -> nostr::Event {
-        // Keep the CAS fixtures' logical seconds and ties, anchored at a fresh
-        // timestamp so they exercise the serving writer's floor admission.
-        static BASE: std::sync::OnceLock<u64> = std::sync::OnceLock::new();
-        let base = BASE.get_or_init(|| nostr::Timestamp::now().as_secs() - 1000);
+    fn make_canvas_event_at(base: u64, content: &str, created_at: u64) -> nostr::Event {
+        // Each test owns its fresh base; logical seconds and ties stay explicit.
         make_event_at(
             buzz_core::kind::KIND_CANVAS as u16,
             content,
@@ -5774,9 +5772,9 @@ mod postgres_tests {
     /// Build two canvas events sharing `created_at`, returned `(lower, higher)`
     /// by event id. Regenerates until the ids differ (always, since keys differ)
     /// so tests can assert deterministic head ordering on the id tiebreak.
-    fn same_second_ordered_pair(created_at: u64) -> (nostr::Event, nostr::Event) {
-        let a = make_canvas_event_at("# A", created_at);
-        let b = make_canvas_event_at("# B", created_at);
+    fn same_second_ordered_pair(base: u64, created_at: u64) -> (nostr::Event, nostr::Event) {
+        let a = make_canvas_event_at(base, "# A", created_at);
+        let b = make_canvas_event_at(base, "# B", created_at);
         if a.id.as_bytes() <= b.id.as_bytes() {
             (a, b)
         } else {
@@ -5784,6 +5782,20 @@ mod postgres_tests {
         }
     }
 
+    #[test]
+    fn canvas_fixture_base_is_owned_by_each_test() {
+        let first_base = 1_700_000_000;
+        let second_base = first_base + crate::replica_fence::CREATED_AT_FLOOR_SECS as u64 * 2;
+        let first = make_canvas_event_at(first_base, "first test", 1000);
+        let second = make_canvas_event_at(second_base, "later test", 1000);
+        assert_eq!(first.created_at.as_secs(), first_base + 1000);
+        assert_eq!(second.created_at.as_secs(), second_base + 1000);
+        let (lower, higher) = same_second_ordered_pair(second_base, 1000);
+        assert_eq!(lower.created_at, higher.created_at);
+        assert_eq!(lower.created_at.as_secs(), second_base + 1000);
+        assert!(lower.id.as_bytes() < higher.id.as_bytes());
+    }
+
     #[tokio::test]
     async fn canvas_head_checked_rejects_non_canvas_kinds_before_opening_a_transaction() {
         // A lazy pool that is never connected: the kind check must reject
@@ -5811,10 +5823,11 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_expect_no_head_creates_first_canvas() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
-        let event = make_canvas_event_at("# First", 1000);
+        let event = make_canvas_event_at(canvas_base, "# First", 1000);
 
         let (_, status) = insert_canvas_head_checked(
             &pool,
@@ -5831,10 +5844,11 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_expect_no_head_rejects_when_head_exists() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
-        let first = make_canvas_event_at("# First", 1000);
+        let first = make_canvas_event_at(canvas_base, "# First", 1000);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -5845,7 +5859,7 @@ mod postgres_tests {
         .await
         .expect("first canvas");
 
-        let second = make_canvas_event_at("# Racing create", 1001);
+        let second = make_canvas_event_at(canvas_base, "# Racing create", 1001);
         let (_, status) = insert_canvas_head_checked(
             &pool,
             community,
@@ -5871,10 +5885,11 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_expected_head_matches_and_advances() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
-        let first = make_canvas_event_at("# First", 1000);
+        let first = make_canvas_event_at(canvas_base, "# First", 1000);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -5885,7 +5900,7 @@ mod postgres_tests {
         .await
         .expect("first canvas");
 
-        let second = make_canvas_event_at("# Second", 1001);
+        let second = make_canvas_event_at(canvas_base, "# Second", 1001);
         let head = first.id.as_bytes();
         let (_, status) = insert_canvas_head_checked(
             &pool,
@@ -5902,10 +5917,11 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_expected_head_mismatch_rejects() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
-        let first = make_canvas_event_at("# First", 1000);
+        let first = make_canvas_event_at(canvas_base, "# First", 1000);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -5916,7 +5932,7 @@ mod postgres_tests {
         .await
         .expect("first canvas");
 
-        let stale = make_canvas_event_at("# Stale edit", 1002);
+        let stale = make_canvas_event_at(canvas_base, "# Stale edit", 1002);
         let wrong_head = [0u8; 32];
         let (_, status) = insert_canvas_head_checked(
             &pool,
@@ -5933,10 +5949,11 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_expected_head_missing_rejects() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
-        let event = make_canvas_event_at("# Edit with no head", 1000);
+        let event = make_canvas_event_at(canvas_base, "# Edit with no head", 1000);
         let some_head = [1u8; 32];
 
         let (_, status) = insert_canvas_head_checked(
@@ -5954,10 +5971,11 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_replay_of_head_is_duplicate() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
-        let event = make_canvas_event_at("# First", 1000);
+        let event = make_canvas_event_at(canvas_base, "# First", 1000);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -5989,10 +6007,11 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_replay_skips_stale_precondition() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
-        let event = make_canvas_event_at("# First", 1000);
+        let event = make_canvas_event_at(canvas_base, "# First", 1000);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -6024,11 +6043,12 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_rejects_same_second_higher_id() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
 
-        let (lower, higher) = same_second_ordered_pair(1000);
+        let (lower, higher) = same_second_ordered_pair(canvas_base, 1000);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -6067,11 +6087,12 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_accepts_same_second_lower_id() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
 
-        let (lower, higher) = same_second_ordered_pair(1000);
+        let (lower, higher) = same_second_ordered_pair(canvas_base, 1000);
         // Seed the higher-id event as the head first.
         insert_canvas_head_checked(
             &pool,
@@ -6101,10 +6122,11 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_rejects_behind_clock_writer() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
-        let head = make_canvas_event_at("# Head", 2000);
+        let head = make_canvas_event_at(canvas_base, "# Head", 2000);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -6116,7 +6138,7 @@ mod postgres_tests {
         .expect("head at t=2000");
 
         // Writer's clock is behind — created_at earlier than head.
-        let behind = make_canvas_event_at("# Behind clock", 1100);
+        let behind = make_canvas_event_at(canvas_base, "# Behind clock", 1100);
         let (_, status) = insert_canvas_head_checked(
             &pool,
             community,
@@ -6200,12 +6222,13 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_deleted_replay_returns_revision_mismatch() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
 
         // H: seed the initial head.
-        let head = make_canvas_event_at("# Head", 1000);
+        let head = make_canvas_event_at(canvas_base, "# Head", 1000);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -6218,7 +6241,7 @@ mod postgres_tests {
         let head_id = head.id.to_bytes().to_vec();
 
         // A: insert a second revision.
-        let a = make_canvas_event_at("# A", 1001);
+        let a = make_canvas_event_at(canvas_base, "# A", 1001);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -6323,11 +6346,12 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_concurrent_authors_only_one_advances() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
 
-        let base = make_canvas_event_at("# Base", 1000);
+        let base = make_canvas_event_at(canvas_base, "# Base", 1000);
         insert_canvas_head_checked(
             &pool,
             community,
@@ -6352,8 +6376,8 @@ mod postgres_tests {
         // Both edits assert `base` as their head and are timestamped strictly
         // ahead of it (writer discipline), so neither trips SupersedeFailed —
         // only the advisory-lock serialization decides the winner.
-        let a = make_canvas_event_at("# Author A", 1001);
-        let b = make_canvas_event_at("# Author B", 1002);
+        let a = make_canvas_event_at(canvas_base, "# Author A", 1001);
+        let b = make_canvas_event_at(canvas_base, "# Author B", 1002);
 
         let (pool_a, pool_b) = (pool.clone(), pool.clone());
         let (id_a, id_b) = (base_id.clone(), base_id.clone());
@@ -6464,6 +6488,7 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_checked_concurrent_first_create_only_one_wins() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
@@ -6477,8 +6502,8 @@ mod postgres_tests {
             .await
             .expect("blocker acquires key");
 
-        let a = make_canvas_event_at("# First A", 1000);
-        let b = make_canvas_event_at("# First B", 1001);
+        let a = make_canvas_event_at(canvas_base, "# First A", 1000);
+        let b = make_canvas_event_at(canvas_base, "# First B", 1001);
 
         let (pool_a, pool_b) = (pool.clone(), pool.clone());
         let ta = tokio::spawn(async move {
@@ -6585,6 +6610,7 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn db_untagged_canvas_append_serializes_on_advisory_key() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let db = crate::Db::from_pool(pool.clone());
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
@@ -6598,7 +6624,7 @@ mod postgres_tests {
             .await
             .expect("holder acquires key");
 
-        let event = make_canvas_event_at("# Untagged", 1000);
+        let event = make_canvas_event_at(canvas_base, "# Untagged", 1000);
         let write_task = tokio::spawn(async move {
             db.insert_event_with_thread_metadata(community, &event, Some(channel), None)
                 .await
@@ -6640,13 +6666,14 @@ mod postgres_tests {
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn channel_head_canvas_deletion_serializes_on_advisory_key() {
+        let canvas_base = nostr::Timestamp::now().as_secs() - 1000;
         let pool = setup_pool().await;
         let community = CommunityId::from_uuid(make_test_community(&pool).await);
         let channel = make_test_channel(&pool, community.as_uuid().to_owned(), None).await;
         let lock_key = canvas_lock_key(community, channel);
 
         // Insert a canvas event to delete.
-        let event = make_canvas_event_at("# To delete", 1000);
+        let event = make_canvas_event_at(canvas_base, "# To delete", 1000);
         insert_canvas_head_checked(
             &pool,
             community,
diff --git a/crates/buzz-db/src/store/event_floor_postgres_tests.rs b/crates/buzz-db/src/store/event_floor_postgres_tests.rs
index 857d91ba686..741e3b7d52a 100644
--- a/crates/buzz-db/src/store/event_floor_postgres_tests.rs
+++ b/crates/buzz-db/src/store/event_floor_postgres_tests.rs
@@ -120,6 +120,62 @@ async fn insert_event_floor_rejects_old_and_accepts_fresh() {
     );
 }
 
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn insert_event_floor_accepts_expired_duplicate() {
+    let (db, community, channel, keys) = fixture().await;
+    let old = signed(
+        &keys,
+        9,
+        channel,
+        database_clock(&db).await - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60),
+    );
+    crate::test_support::insert_historical_event(db.pool(), community, &old, channel).await;
+    let (_, inserted) = event::insert_event(db.pool(), community, &old, Some(channel))
+        .await
+        .expect("expired duplicate is idempotent");
+    assert!(!inserted);
+    assert_eq!(count(&db, community, &old).await, 1);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn thread_metadata_event_floor_accepts_expired_duplicate() {
+    let (db, community, channel, keys) = fixture().await;
+    let time = database_clock(&db).await - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60);
+    let old = signed(&keys, 9, channel, time);
+    crate::test_support::insert_historical_event(db.pool(), community, &old, channel).await;
+    let mut tx = begin(&db, community).await;
+    let meta = event::ThreadMetadataParams {
+        event_id: old.id.as_bytes(),
+        event_created_at: DateTime::from_timestamp(old.created_at.as_secs() as i64, 0)
+            .expect("timestamp"),
+        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,
+    };
+    let (_, inserted) =
+        event::insert_event_with_thread_metadata_tx(&mut tx, &old, Some(channel), Some(meta))
+            .await
+            .expect("expired duplicate is idempotent");
+    assert!(!inserted);
+    tx.commit().await.expect("commit duplicate result");
+    assert_eq!(count(&db, community, &old).await, 1);
+    let metadata: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM thread_metadata WHERE community_id=$1 AND event_id=$2",
+    )
+    .bind(community.as_uuid())
+    .bind(old.id.as_bytes().as_slice())
+    .fetch_one(db.pool())
+    .await
+    .expect("metadata count");
+    assert_eq!(metadata, 0, "a duplicate must not run thread follow-ups");
+}
+
 #[tokio::test]
 #[ignore = "requires Postgres"]
 async fn thread_metadata_event_floor_rejects_old_and_accepts_fresh() {
@@ -264,7 +320,21 @@ async fn cluster_global_event_floor_held_transaction_cannot_write_below_probe_wa
         channel,
         entry.fence_wall - chrono::Duration::seconds(1),
     );
+    sqlx::query("SAVEPOINT rejected_floor_insert")
+        .execute(tx.conn())
+        .await
+        .expect("isolate rejected insert");
     rejected(event::insert_event_in_transaction(&mut tx, &below_wall, Some(channel)).await);
+    // The check runs after INSERT. Roll back that write before continuing in
+    // the same held transaction, just as a caller must on any insert failure.
+    sqlx::query("ROLLBACK TO SAVEPOINT rejected_floor_insert")
+        .execute(tx.conn())
+        .await
+        .expect("roll back rejected row");
+    sqlx::query("RELEASE SAVEPOINT rejected_floor_insert")
+        .execute(tx.conn())
+        .await
+        .expect("release rejection savepoint");
     // Sample just after the probe; a near-floor accepted insert is above its
     // wall even when this transaction commits after the heartbeat.
     let fresh = signed(
diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs
index ac7f50491d4..4766aeb07e6 100644
--- a/crates/buzz-db/src/store/relay_members.rs
+++ b/crates/buzz-db/src/store/relay_members.rs
@@ -1151,7 +1151,7 @@ pub(crate) async fn write_nip43_membership_snapshot(
     .await?;
     sqlx::query(
         "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \
-         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)",
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NULL, $10)",
     )
     .bind(community_id.as_uuid())
     .bind(event.id.as_bytes().as_slice())
@@ -1162,7 +1162,6 @@ pub(crate) async fn write_nip43_membership_snapshot(
     .bind(&event.content)
     .bind(event.sig.serialize().as_slice())
     .bind(received_at)
-    .bind::<Option<Uuid>>(None)
     .bind(d_tag.as_deref())
     .execute(tx.conn())
     .await?;
diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs
index 7c1960dd330..66c43c7ceab 100644
--- a/crates/buzz-db/tests/observability_source.rs
+++ b/crates/buzz-db/tests/observability_source.rs
@@ -1586,12 +1586,20 @@ fn inserts_events_row(source: &str) -> bool {
     !events_insert_tails(source).is_empty()
 }
 
-/// `source` lowercased, with every run of whitespace, `\` line continuations,
-/// and `\n`/`\t`/`\r` escapes collapsed to one space.
+/// `source` lowercased, with identifier quotes removed and every run of
+/// whitespace, `\` line continuations, and `\n`/`\t`/`\r` escapes collapsed
+/// to one space.
 fn normalized_sql(source: &str) -> String {
     let mut normalized = String::with_capacity(source.len());
     let mut chars = source.chars().peekable();
     while let Some(ch) = chars.next() {
+        if ch == '"' {
+            continue;
+        }
+        if ch == '\\' && chars.peek() == Some(&'"') {
+            chars.next();
+            continue;
+        }
         if ch.is_whitespace() || ch == '\\' {
             if ch == '\\' && matches!(chars.peek(), Some('n' | 't' | 'r')) {
                 chars.next();
@@ -1661,6 +1669,9 @@ pub(crate) async fn unhooked_writer(tx: &mut AdmittedTx) {
         .await
         .expect("write");
 }
+pub(crate) async fn quoted_writer(tx: &mut AdmittedTx) {
+    sqlx::query("INSERT INTO \"public\".\"events\" (id) VALUES ($1)");
+}
 pub(crate) async fn push_only_writer(tx: &mut AdmittedTx) {
     sqlx::query("INSERT INTO events \
                  (community_id, id) VALUES ($1, $2)")
@@ -1825,6 +1836,7 @@ mod tests {
         event_insert_follow_up_violations(&strip_cfg_test_items(source)),
         [
             "pub(crate) async fn unhooked_writer(tx: &mut AdmittedTx) {",
+            "pub(crate) async fn quoted_writer(tx: &mut AdmittedTx) {",
             "pub(crate) async fn push_only_writer(tx: &mut AdmittedTx) {",
             "pub(crate) async fn lowercase_writer(tx: &mut AdmittedTx) {",
             "pub(crate) async fn continued_writer(tx: &mut AdmittedTx) {",
@@ -2240,34 +2252,60 @@ const EVENT_FLOOR_PROBE_EXCEPTIONS: [&str; 2] = [
 
 fn is_channel_null_event_writer(function: &str) -> bool {
     let header = function_header(function);
-    if header.starts_with("pub async fn publish_nip43_membership_locked(") {
-        // Kind 13534: the channel bind is always NULL.
-        return function.contains(".bind::<Option<Uuid>>(None)");
+    let sql = normalized_sql(function).replace(' ', "");
+    if header.starts_with("pub(crate) async fn write_nip43_membership_snapshot(") {
+        // Kind 13534: channel_id is the tenth value, a literal NULL.
+        return sql.contains("values($1,$2,$3,$4,$5,$6,$7,$8,$9,null,$10)");
     }
     if header.starts_with("pub async fn accept_lease_event(") {
-        // Kind 30350: channel_id is the literal NULL in VALUES.
-        return normalized_sql(function).contains("now(),null,$8)");
+        // Kind 30350: channel_id is the literal NULL after received_at.
+        return sql.contains("now(),null,$8)");
     }
     false
 }
 
 fn updates_event_floor_columns(function: &str) -> bool {
     let normalized = normalized_sql(function);
-    ["update events set ", "update public.events set "]
-        .iter()
-        .any(|marker| {
-            normalized.match_indices(marker).any(|(index, marker)| {
-                let assignments = normalized[index + marker.len()..]
-                    .split(" where ")
-                    .next()
-                    .expect("split has first part");
-                assignments.split(',').any(|assignment| {
-                    assignment.split_once('=').is_some_and(|(column, _)| {
-                        matches!(column.trim(), "created_at" | "channel_id")
-                    })
+    normalized.match_indices("update ").any(|(index, marker)| {
+        let Some((target, assignments)) = normalized[index + marker.len()..].split_once(" set ")
+        else {
+            return false;
+        };
+        let mut target = target.split_whitespace();
+        let mut table = target.next();
+        if table == Some("only") {
+            table = target.next();
+        }
+        if !matches!(table, Some("events" | "public.events")) {
+            return false;
+        }
+        // The target may have an alias, with or without AS.
+        let alias = target.collect::<Vec<_>>();
+        if !matches!(alias.as_slice(), [] | [_] | ["as", _]) {
+            return false;
+        }
+        let assignments = assignments.split(" where ").next().expect("first part");
+        let mut depth = 0usize;
+        assignments
+            .split(|ch| {
+                match ch {
+                    '(' => depth += 1,
+                    ')' => depth = depth.saturating_sub(1),
+                    _ => {}
+                }
+                ch == ',' && depth == 0
+            })
+            .any(|assignment| {
+                assignment.split_once('=').is_some_and(|(columns, _)| {
+                    columns
+                        .trim()
+                        .trim_start_matches('(')
+                        .trim_end_matches(')')
+                        .split(',')
+                        .any(|column| matches!(column.trim(), "created_at" | "channel_id"))
                 })
             })
-        })
+    })
 }
 
 fn event_floor_policy_violations(production: &str) -> Vec<String> {
@@ -2308,6 +2346,36 @@ async fn rewrite_time(tx: &mut AdmittedTx) {
 async fn attach_channel(tx: &mut AdmittedTx) {
     sqlx::query("uPdAtE public.events\nSET channel_id=$1 WHERE id=$2");
 }
+async fn alias_time(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE events e SET created_at = $1 WHERE e.id = $2");
+}
+async fn as_alias_time(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE events AS e SET created_at = $1 WHERE e.id = $2");
+}
+async fn only_channel(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE ONLY events SET channel_id = $1 WHERE id = $2");
+}
+async fn only_public_alias(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE ONLY public.events AS e SET channel_id = $1 WHERE e.id = $2");
+}
+async fn row_time(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE events SET (created_at, kind) = ($1, 1) WHERE id = $2");
+}
+async fn row_channel(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE events SET (kind, channel_id) = (1, $1) WHERE id = $2");
+}
+async fn quoted_insert(tx: &mut AdmittedTx) {
+    sqlx::query("INSERT INTO \"events\" (id) VALUES ($1)");
+}
+async fn quoted_schema_insert(tx: &mut AdmittedTx) {
+    sqlx::query("INSERT INTO \"public\".\"events\" (id) VALUES ($1)");
+}
+async fn quoted_update(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE \"events\" SET created_at = $1 WHERE id = $2");
+}
+async fn quoted_schema_update(tx: &mut AdmittedTx) {
+    sqlx::query("UPDATE \"public\".\"events\" SET channel_id = $1 WHERE id = $2");
+}
 async fn delete_only(tx: &mut AdmittedTx) {
     sqlx::query("UPDATE events SET deleted_at=now() WHERE channel_id=$1 AND created_at<$2");
 }
@@ -2318,26 +2386,48 @@ async fn delete_only(tx: &mut AdmittedTx) {
             "async fn missing_guard(tx: &mut AdmittedTx) {",
             "async fn rewrite_time(tx: &mut AdmittedTx) {",
             "async fn attach_channel(tx: &mut AdmittedTx) {",
+            "async fn alias_time(tx: &mut AdmittedTx) {",
+            "async fn as_alias_time(tx: &mut AdmittedTx) {",
+            "async fn only_channel(tx: &mut AdmittedTx) {",
+            "async fn only_public_alias(tx: &mut AdmittedTx) {",
+            "async fn row_time(tx: &mut AdmittedTx) {",
+            "async fn row_channel(tx: &mut AdmittedTx) {",
+            "async fn quoted_insert(tx: &mut AdmittedTx) {",
+            "async fn quoted_schema_insert(tx: &mut AdmittedTx) {",
+            "async fn quoted_update(tx: &mut AdmittedTx) {",
+            "async fn quoted_schema_update(tx: &mut AdmittedTx) {",
         ]
     );
 }
 
+#[test]
+fn event_floor_policy_requires_nip43_channel_null_at_its_insert_position() {
+    let literal_null = r#"
+pub(crate) async fn write_nip43_membership_snapshot(tx: &mut AdmittedTx) {
+    sqlx::query("INSERT INTO events (community_id,id,pubkey,kind,content,tags,sig,created_at,received_at,channel_id,d_tag) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,NULL,$10)");
+}
+"#;
+    assert!(event_floor_policy_violations(literal_null).is_empty());
+    let unrelated_null_bind = literal_null.replace("$9,NULL,$10", "$9,$10,$11").replace(
+        "sqlx::query(",
+        "let unrelated = query.bind::<Option<Uuid>>(None); sqlx::query(",
+    );
+    assert_eq!(
+        event_floor_policy_violations(&unrelated_null_bind).len(),
+        1,
+        "an unrelated NULL bind must not exempt a channel-bearing INSERT"
+    );
+}
+
 #[test]
 fn every_production_channel_event_insert_checks_the_floor_and_never_rewrites_it() {
-    fn collect(dir: &std::path::Path, files: &mut Vec<std::path::PathBuf>) {
-        for entry in std::fs::read_dir(dir).expect("read source directory") {
-            let path = entry.expect("source entry").path();
-            if path.is_dir() {
-                collect(&path, files);
-            } else if should_scan_guarded_write_source_file(&path) {
-                files.push(path);
-            }
-        }
-    }
-    let mut files = Vec::new();
-    collect(
-        &std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src"),
-        &mut files,
+    let files = production_event_writer_source_files();
+    let crates_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+        .parent()
+        .expect("crates root");
+    assert!(
+        files.contains(&crates_root.join("buzz-relay/src/state.rs")),
+        "floor policy must scan the workspace, including relay writers"
     );
     let mut guarded_writers = 0;
     let mut violations = Vec::new();
@@ -2402,16 +2492,41 @@ fn admin_seed_refuses_remote_database_before_invoking_psql() {
     std::fs::set_permissions(&psql, std::fs::Permissions::from_mode(0o755))
         .expect("executable stub");
     let witness = root.join("psql-was-invoked");
-    let output = std::process::Command::new("/bin/bash")
-        .arg(script)
-        .env("PGHOST", "shared-db.example.invalid")
-        .env("PATH", format!("{}:/usr/bin:/bin", root.display()))
-        .env("SEED_PSQL_WITNESS", &witness)
-        .output()
-        .expect("run local seed guard");
-    let invoked = witness.exists();
+    let cases = [
+        ("remote host", "shared-db.example.invalid", "", ""),
+        ("address override", "localhost", "192.0.2.1", ""),
+        ("service override", "localhost", "", "production"),
+        (
+            "socket-prefixed host list",
+            "/tmp,shared-db.example.invalid",
+            "",
+            "",
+        ),
+        ("host list", "localhost,shared-db.example.invalid", "", ""),
+    ];
+    let results = cases.map(|(name, host, address, service)| {
+        let output = std::process::Command::new("/bin/bash")
+            .arg(&script)
+            .env("PGHOST", host)
+            .env("PGHOSTADDR", address)
+            .env("PGSERVICE", service)
+            .env("PATH", format!("{}:/usr/bin:/bin", root.display()))
+            .env("SEED_PSQL_WITNESS", &witness)
+            .output()
+            .expect("run local seed guard");
+        let invoked = witness.exists();
+        if invoked {
+            std::fs::remove_file(&witness).expect("clear witness");
+        }
+        (name, output, invoked)
+    });
     std::fs::remove_dir_all(root).expect("remove scratch fixture");
-    assert!(!invoked, "a non-local target reached psql");
-    assert!(!output.status.success());
-    assert!(String::from_utf8_lossy(&output.stderr).contains("requires a local PostgreSQL target"));
+    for (name, output, invoked) in results {
+        assert!(!invoked, "{name} reached psql");
+        assert!(!output.status.success(), "{name} accepted");
+        assert!(
+            String::from_utf8_lossy(&output.stderr).contains("requires a local PostgreSQL target"),
+            "{name}: missing refusal"
+        );
+    }
 }
diff --git a/scripts/seed-admin-dashboard.sh b/scripts/seed-admin-dashboard.sh
index 9caad1f2d20..75f63c2fd61 100755
--- a/scripts/seed-admin-dashboard.sh
+++ b/scripts/seed-admin-dashboard.sh
@@ -20,6 +20,10 @@ export PGPASSWORD="${PGPASSWORD:-buzz_dev}"
 export PGDATABASE="${PGDATABASE:-buzz}"
 
 # This raw fixture writer is unarmed and must never target a live replica fence.
+if [[ -n "${PGHOSTADDR:-}" || -n "${PGSERVICE:-}" || "${PGHOST}" == *,* ]]; then
+  echo "error: admin seed requires a local PostgreSQL target without PGHOSTADDR, PGSERVICE or multiple hosts" >&2
+  exit 1
+fi
 case "${PGHOST}" in
   localhost|127.0.0.1|::1|/*) ;;
   *)

From bfa04175fe00fa2a5d550d4ce2e204cb105b0e05 Mon Sep 17 00:00:00 2001
From: Elrond
 <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Date: Fri, 16 Oct 2026 09:53:30 -0400
Subject: [PATCH 3/7] fix(seed): refuse connection-string PGDATABASE in
 local-only admin seed guard

psql -d treats URI and key=value values as full connection strings that
override -h, so PGDATABASE could route the unarmed fixture writer to a
shared database. Accept only a plain database name before either psql
path, and extend the refusal test with URI and key/value cases
(host=, hostaddr=, service=) plus a plain-name control that must reach
psql.

Also note in replica_fence that prepared-transaction coverage comes from
the retained deferred trigger, not the insert-time app guard.

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
---
 crates/buzz-db/src/runtime/replica_fence.rs  |  5 +-
 crates/buzz-db/tests/observability_source.rs | 77 ++++++++++++++++----
 scripts/seed-admin-dashboard.sh              |  6 ++
 3 files changed, 74 insertions(+), 14 deletions(-)

diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs
index cbe8e1546a5..16ae8b8457e 100644
--- a/crates/buzz-db/src/runtime/replica_fence.rs
+++ b/crates/buzz-db/src/runtime/replica_fence.rs
@@ -866,7 +866,10 @@ async fn sample_writer(
         //    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
diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs
index 66c43c7ceab..3e711149cc6 100644
--- a/crates/buzz-db/tests/observability_source.rs
+++ b/crates/buzz-db/tests/observability_source.rs
@@ -2492,24 +2492,13 @@ fn admin_seed_refuses_remote_database_before_invoking_psql() {
     std::fs::set_permissions(&psql, std::fs::Permissions::from_mode(0o755))
         .expect("executable stub");
     let witness = root.join("psql-was-invoked");
-    let cases = [
-        ("remote host", "shared-db.example.invalid", "", ""),
-        ("address override", "localhost", "192.0.2.1", ""),
-        ("service override", "localhost", "", "production"),
-        (
-            "socket-prefixed host list",
-            "/tmp,shared-db.example.invalid",
-            "",
-            "",
-        ),
-        ("host list", "localhost,shared-db.example.invalid", "", ""),
-    ];
-    let results = cases.map(|(name, host, address, service)| {
+    let run = |host: &str, address: &str, service: &str, database: &str| {
         let output = std::process::Command::new("/bin/bash")
             .arg(&script)
             .env("PGHOST", host)
             .env("PGHOSTADDR", address)
             .env("PGSERVICE", service)
+            .env("PGDATABASE", database)
             .env("PATH", format!("{}:/usr/bin:/bin", root.display()))
             .env("SEED_PSQL_WITNESS", &witness)
             .output()
@@ -2518,9 +2507,71 @@ fn admin_seed_refuses_remote_database_before_invoking_psql() {
         if invoked {
             std::fs::remove_file(&witness).expect("clear witness");
         }
+        (output, invoked)
+    };
+    // Control: a local host with a plain database name reaches psql, so the
+    // refusals below come from the guard rather than an unrunnable fixture.
+    let (_, control_invoked) = run("localhost", "", "", "buzz");
+    let cases = [
+        ("remote host", "shared-db.example.invalid", "", "", ""),
+        ("address override", "localhost", "192.0.2.1", "", ""),
+        ("service override", "localhost", "", "production", ""),
+        (
+            "socket-prefixed host list",
+            "/tmp,shared-db.example.invalid",
+            "",
+            "",
+            "",
+        ),
+        (
+            "host list",
+            "localhost,shared-db.example.invalid",
+            "",
+            "",
+            "",
+        ),
+        (
+            "database URI",
+            "localhost",
+            "",
+            "",
+            "postgresql://buzz@shared-db.example.invalid/buzz",
+        ),
+        (
+            "short database URI",
+            "localhost",
+            "",
+            "",
+            "postgres://shared-db.example.invalid/buzz",
+        ),
+        (
+            "database key/value host",
+            "localhost",
+            "",
+            "",
+            "host=shared-db.example.invalid dbname=buzz",
+        ),
+        (
+            "database key/value hostaddr",
+            "localhost",
+            "",
+            "",
+            "hostaddr=192.0.2.1 dbname=buzz",
+        ),
+        (
+            "database key/value service",
+            "localhost",
+            "",
+            "",
+            "service=production",
+        ),
+    ];
+    let results = cases.map(|(name, host, address, service, database)| {
+        let (output, invoked) = run(host, address, service, database);
         (name, output, invoked)
     });
     std::fs::remove_dir_all(root).expect("remove scratch fixture");
+    assert!(control_invoked, "plain local target did not reach psql");
     for (name, output, invoked) in results {
         assert!(!invoked, "{name} reached psql");
         assert!(!output.status.success(), "{name} accepted");
diff --git a/scripts/seed-admin-dashboard.sh b/scripts/seed-admin-dashboard.sh
index 75f63c2fd61..e5fd234eeb7 100755
--- a/scripts/seed-admin-dashboard.sh
+++ b/scripts/seed-admin-dashboard.sh
@@ -31,6 +31,12 @@ case "${PGHOST}" in
     exit 1
     ;;
 esac
+# psql -d treats URI and key=value values as full connection strings that
+# override -h, so accept only a plain database name.
+if [[ ! "${PGDATABASE}" =~ ^[A-Za-z0-9_-]+$ ]]; then
+  echo "error: admin seed requires a local PostgreSQL target with a plain PGDATABASE name" >&2
+  exit 1
+fi
 
 if command -v psql >/dev/null 2>&1; then
   run_psql() {

From 9512e8c165abbfd45ebed6a9d815ac4e9be4c78b Mon Sep 17 00:00:00 2001
From: Elrond
 <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Date: Fri, 16 Oct 2026 10:02:38 -0400
Subject: [PATCH 4/7] fix(seed): run psql with -X so psqlrc cannot retarget
 after the local guard

psql -c no longer implies -X, so ~/.psqlrc or PSQLRC (exportable from
.env) ran after the guard and could \connect to another host. Pass -X
on both the local and Docker psql paths and assert the control
invocation carries it.

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
---
 crates/buzz-db/tests/observability_source.rs | 25 +++++++++++++-------
 scripts/seed-admin-dashboard.sh              |  5 ++--
 2 files changed, 20 insertions(+), 10 deletions(-)

diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs
index 3e711149cc6..80dbcb8f048 100644
--- a/crates/buzz-db/tests/observability_source.rs
+++ b/crates/buzz-db/tests/observability_source.rs
@@ -2488,7 +2488,11 @@ fn admin_seed_refuses_remote_database_before_invoking_psql() {
     )
     .expect("copy script without developer .env");
     let psql = root.join("psql");
-    std::fs::write(&psql, "#!/bin/sh\ntouch \"$SEED_PSQL_WITNESS\"\nexit 1\n").expect("stub psql");
+    std::fs::write(
+        &psql,
+        "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$SEED_PSQL_WITNESS\"\nexit 1\n",
+    )
+    .expect("stub psql");
     std::fs::set_permissions(&psql, std::fs::Permissions::from_mode(0o755))
         .expect("executable stub");
     let witness = root.join("psql-was-invoked");
@@ -2503,15 +2507,16 @@ fn admin_seed_refuses_remote_database_before_invoking_psql() {
             .env("SEED_PSQL_WITNESS", &witness)
             .output()
             .expect("run local seed guard");
-        let invoked = witness.exists();
-        if invoked {
+        let args = std::fs::read_to_string(&witness).ok();
+        if args.is_some() {
             std::fs::remove_file(&witness).expect("clear witness");
         }
-        (output, invoked)
+        (output, args)
     };
     // Control: a local host with a plain database name reaches psql, so the
     // refusals below come from the guard rather than an unrunnable fixture.
-    let (_, control_invoked) = run("localhost", "", "", "buzz");
+    // It must also pass -X so a psqlrc cannot \connect after the guard.
+    let (_, control_args) = run("localhost", "", "", "buzz");
     let cases = [
         ("remote host", "shared-db.example.invalid", "", "", ""),
         ("address override", "localhost", "192.0.2.1", "", ""),
@@ -2567,11 +2572,15 @@ fn admin_seed_refuses_remote_database_before_invoking_psql() {
         ),
     ];
     let results = cases.map(|(name, host, address, service, database)| {
-        let (output, invoked) = run(host, address, service, database);
-        (name, output, invoked)
+        let (output, args) = run(host, address, service, database);
+        (name, output, args.is_some())
     });
     std::fs::remove_dir_all(root).expect("remove scratch fixture");
-    assert!(control_invoked, "plain local target did not reach psql");
+    let control_args = control_args.expect("plain local target did not reach psql");
+    assert!(
+        control_args.lines().any(|arg| arg == "-X"),
+        "psql invoked without -X: {control_args:?}"
+    );
     for (name, output, invoked) in results {
         assert!(!invoked, "{name} reached psql");
         assert!(!output.status.success(), "{name} accepted");
diff --git a/scripts/seed-admin-dashboard.sh b/scripts/seed-admin-dashboard.sh
index e5fd234eeb7..f67e971a8ef 100755
--- a/scripts/seed-admin-dashboard.sh
+++ b/scripts/seed-admin-dashboard.sh
@@ -38,15 +38,16 @@ if [[ ! "${PGDATABASE}" =~ ^[A-Za-z0-9_-]+$ ]]; then
   exit 1
 fi
 
+# -X skips psqlrc, which runs after this guard and could \connect elsewhere.
 if command -v psql >/dev/null 2>&1; then
   run_psql() {
-    PGPASSWORD="${PGPASSWORD}" psql -h "${PGHOST}" -p "${PGPORT}" \
+    PGPASSWORD="${PGPASSWORD}" psql -X -h "${PGHOST}" -p "${PGPORT}" \
       -U "${PGUSER}" -d "${PGDATABASE}" "$@"
   }
 elif docker exec buzz-postgres psql --version >/dev/null 2>&1; then
   run_psql() {
     docker exec -i -e PGPASSWORD="${PGPASSWORD}" buzz-postgres \
-      psql -U "${PGUSER}" -d "${PGDATABASE}" "$@"
+      psql -X -U "${PGUSER}" -d "${PGDATABASE}" "$@"
   }
 else
   echo "error: neither psql nor buzz-postgres docker psql is available" >&2

From 1e86b593f53b3d05df30b27b992d54bfa96487d4 Mon Sep 17 00:00:00 2001
From: Luke Tornquist <tornquist@squareup.com>
Date: Tue, 10 Nov 2026 08:30:19 -0800
Subject: [PATCH 5/7] buzz-db: check the replacement floor after the replay
 exit

Replaying a stored but soft-deleted parameterized or addressable id older
than the floor returned CreatedAtFloorRejected, because the guard ran before
the INSERT ... ON CONFLICT DO NOTHING and the early exits only see live
heads. The trigger never fired on a conflict, so the old result was a
duplicate. Both writers now check only when the INSERT stored a row; the
savepoint or the caller's rollback restores the superseded head.

Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Luke Tornquist <tornquist@squareup.com>
---
 crates/buzz-db/src/runtime/replica_fence.rs   |  7 +--
 .../src/store/event_floor_postgres_tests.rs   | 45 +++++++++++++++++++
 crates/buzz-db/src/store/replaceable.rs       | 11 +++--
 3 files changed, 56 insertions(+), 7 deletions(-)

diff --git a/crates/buzz-db/src/runtime/replica_fence.rs b/crates/buzz-db/src/runtime/replica_fence.rs
index 16ae8b8457e..40bb6aa7b77 100644
--- a/crates/buzz-db/src/runtime/replica_fence.rs
+++ b/crates/buzz-db/src/runtime/replica_fence.rs
@@ -83,9 +83,10 @@ 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.
+/// One SELECT keeps the five existing INSERT shapes intact. Every writer
+/// checks after its INSERT, only when it inserted a row, before follow-ups, so
+/// replaying a stored id (even a soft-deleted one) stays a duplicate.
+/// Replacements roll back the supersede with it (savepoint or caller).
 /// 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
diff --git a/crates/buzz-db/src/store/event_floor_postgres_tests.rs b/crates/buzz-db/src/store/event_floor_postgres_tests.rs
index 741e3b7d52a..3d2baef9c01 100644
--- a/crates/buzz-db/src/store/event_floor_postgres_tests.rs
+++ b/crates/buzz-db/src/store/event_floor_postgres_tests.rs
@@ -268,6 +268,51 @@ async fn addressable_event_floor_rejects_old_and_accepts_fresh() {
     assert_eq!(count(&db, community, &fresh).await, 1);
 }
 
+/// Seed an aged, stored and then soft-deleted event, as NIP-09 leaves it.
+async fn soft_deleted_aged(db: &Db, community: CommunityId, channel: Uuid, keys: &Keys, kind: u16) -> Event {
+    let now = database_clock(db).await;
+    let aged = signed(
+        keys,
+        kind,
+        channel,
+        now - chrono::Duration::seconds(CREATED_AT_FLOOR_SECS + 60),
+    );
+    crate::test_support::insert_historical_event(db.pool(), community, &aged, channel).await;
+    sqlx::query("UPDATE events SET deleted_at = now() WHERE community_id = $1 AND id = $2")
+        .bind(community.as_uuid())
+        .bind(aged.id.as_bytes().as_slice())
+        .execute(db.pool())
+        .await
+        .expect("soft-delete");
+    aged
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn parameterized_replay_of_soft_deleted_aged_event_is_a_duplicate() {
+    let (db, community, channel, keys) = fixture().await;
+    let aged = soft_deleted_aged(&db, community, channel, &keys, 30078).await;
+    let (_, stored) = db
+        .replace_parameterized_event(community, &aged, &channel.to_string(), Some(channel))
+        .await
+        .expect("replay of a stored id is a duplicate, not a floor rejection");
+    assert!(!stored);
+    assert_eq!(count(&db, community, &aged).await, 1);
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn addressable_replay_of_soft_deleted_aged_event_is_a_duplicate() {
+    let (db, community, channel, keys) = fixture().await;
+    let aged = soft_deleted_aged(&db, community, channel, &keys, 39000).await;
+    let (_, stored) = db
+        .replace_addressable_event(community, &aged, Some(channel))
+        .await
+        .expect("replay of a stored id is a duplicate, not a floor rejection");
+    assert!(!stored);
+    assert_eq!(count(&db, community, &aged).await, 1);
+}
+
 #[tokio::test]
 #[ignore = "requires Postgres"]
 async fn member_snapshot_event_floor_rejects_old_and_accepts_fresh() {
diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs
index e18253b65d6..971e638bc50 100644
--- a/crates/buzz-db/src/store/replaceable.rs
+++ b/crates/buzz-db/src/store/replaceable.rs
@@ -251,8 +251,6 @@ async fn replace_parameterized_event_in_transaction_impl(
         ));
     }
 
-    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
-
     // Everything from the supersede through the mention index runs under one
     // savepoint on the admitted transaction. A duplicate insert or any failure
     // (including mention indexing) rolls the whole replacement back and leaves
@@ -331,6 +329,10 @@ async fn replace_parameterized_event_in_transaction_impl(
         if insert_result.rows_affected() == 0 {
             return Ok(false);
         }
+        // After the replay exit: a stored id (even soft-deleted) is a
+        // duplicate, not a floor rejection. Failure rolls back to the
+        // savepoint, restoring the previous live head.
+        crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
 
         if is_nip_rs {
             sqlx::query(
@@ -480,8 +482,6 @@ pub(crate) async fn replace_addressable_event_under_lock(
     crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json)
         .await?;
 
-    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
-
     // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety.
     sqlx::query(superseded_with_mentions_sql!(
         "UPDATE events SET deleted_at = NOW() \
@@ -528,6 +528,9 @@ pub(crate) async fn replace_addressable_event_under_lock(
             false,
         ));
     }
+    // After the replay exit, as in the parameterized path; the caller's
+    // rollback restores the previous head on failure.
+    crate::replica_fence::guard_event_created_at_floor(tx, created_at, channel_id).await?;
 
     crate::store::event_follow_up::after_admitted_insert(
         tx,

From 259b664d33d02999b776abbbcf57df7695c43856 Mon Sep 17 00:00:00 2001
From: Luke Tornquist <tornquist@squareup.com>
Date: Tue, 10 Nov 2026 08:35:45 -0800
Subject: [PATCH 6/7] test(buzz-db): seed aged rows for main's newer tests
 through historical fixtures

Tests that main added after the floor guard branched insert aged channel
events through the serving writers, which now reject them with
CreatedAtFloorRejected. Seed those rows as already-stored data through
insert_historical_event, add a historical fixture that also records thread
metadata via the store's insert_thread_metadata, and keep the operations under
test (reads, deletions, replays) on the real paths. Deletion events whose
timestamps must stay older than their targets are stored without a channel,
and two incidentally aged timestamps move inside the floor.

Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Luke Tornquist <tornquist@squareup.com>
---
 crates/buzz-db/src/store/event.rs             |  26 ++---
 .../store/event_deletion/postgres_tests.rs    | 103 +++++++++---------
 .../store/event_hydration/postgres_tests.rs   |   8 +-
 crates/buzz-db/src/store/thread.rs            |  38 ++++++-
 crates/buzz-db/src/test_support.rs            |  30 +++++
 .../src/test_support/recent_pass_tests.rs     |   7 +-
 6 files changed, 131 insertions(+), 81 deletions(-)

diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs
index aafee90d9e9..98124926a9c 100644
--- a/crates/buzz-db/src/store/event.rs
+++ b/crates/buzz-db/src/store/event.rs
@@ -3557,7 +3557,6 @@ mod postgres_tests {
             channels.push(make_test_channel(&pool, community_uuid, None).await);
         }
         let (busy, quiet, empty, outside) = (channels[0], channels[1], channels[2], channels[3]);
-        let db = crate::Db::from_pool(pool.clone());
         let author = Keys::generate();
         let now = Utc::now().timestamp() as u64;
         let floor = crate::store::recent_pass::floor_at(Utc::now()).timestamp() as u64;
@@ -3588,9 +3587,8 @@ mod postgres_tests {
         events.push(insert(busy, 9, floor - 86_400, true));
         events.push(insert(quiet, 9, floor - 2 * 86_400, true));
         for (channel, event) in &events {
-            db.insert_event(community, event, Some(*channel))
-                .await
-                .expect("insert");
+            // Rows already in storage: the floor rejects the aged ones.
+            crate::test_support::insert_historical_event(&pool, community, event, *channel).await;
         }
         matching.sort_by(|a, b| {
             b.0.cmp(&a.0)
@@ -3727,9 +3725,7 @@ mod postgres_tests {
                 .custom_created_at(nostr::Timestamp::from(at.timestamp() as u64))
                 .sign_with_keys(&author)
                 .expect("sign");
-            db.insert_event(community, &event, Some(channel))
-                .await
-                .expect("insert");
+            crate::test_support::insert_historical_event(&pool, community, &event, channel).await;
             let partition: String =
                 sqlx::query_scalar("SELECT tableoid::regclass::text FROM events WHERE id = $1")
                     .bind(event.id.as_bytes().to_vec())
@@ -3799,9 +3795,7 @@ mod postgres_tests {
                 .custom_created_at(nostr::Timestamp::from(at.timestamp() as u64))
                 .sign_with_keys(&author)
                 .expect("sign");
-            db.insert_event(community, &event, Some(channel))
-                .await
-                .expect("insert");
+            crate::test_support::insert_historical_event(&pool, community, &event, channel).await;
             if at >= floor {
                 recent.push(event.id);
                 let partition: String =
@@ -3883,7 +3877,6 @@ mod postgres_tests {
         let community_uuid = make_test_community(&pool).await;
         let community = CommunityId::from_uuid(community_uuid);
         let channel = make_test_channel(&pool, community_uuid, None).await;
-        let db = crate::Db::from_pool(pool.clone());
         let author = Keys::generate();
         let now = Utc::now().timestamp() as u64;
         let floor = crate::store::recent_pass::floor_at(Utc::now()).timestamp() as u64;
@@ -3900,9 +3893,7 @@ mod postgres_tests {
                 .custom_created_at(nostr::Timestamp::from(at))
                 .sign_with_keys(&author)
                 .expect("sign");
-            db.insert_event(community, &event, Some(channel))
-                .await
-                .expect("insert");
+            crate::test_support::insert_historical_event(&pool, community, &event, channel).await;
             newest_first.push(event.id);
         }
 
@@ -3977,9 +3968,7 @@ mod postgres_tests {
             .custom_created_at(nostr::Timestamp::from(1_748_736_000)) // 2025-06-01
             .sign_with_keys(&author)
             .expect("sign");
-        db.insert_event(community, &ancient, Some(channel))
-            .await
-            .expect("insert ancient");
+        crate::test_support::insert_historical_event(&pool, community, &ancient, channel).await;
         let partition_of = |id: nostr::EventId| {
             let pool = pool.clone();
             async move {
@@ -4251,7 +4240,8 @@ mod postgres_tests {
         let community = CommunityId::from_uuid(community_uuid);
         let channel = make_test_channel(&pool, community_uuid, None).await;
         let keys = Keys::generate();
-        let now = nostr::Timestamp::now().as_secs() - 1_000;
+        // Inside the `created_at` floor (960 s) even with the later offsets.
+        let now = nostr::Timestamp::now().as_secs() - 600;
         let h = channel.to_string();
         let message = |kind: u16, tags: Vec<Vec<String>>, at: u64| {
             EventBuilder::new(Kind::Custom(kind), "")
diff --git a/crates/buzz-db/src/store/event_deletion/postgres_tests.rs b/crates/buzz-db/src/store/event_deletion/postgres_tests.rs
index 28cd85cccf8..51317d38818 100644
--- a/crates/buzz-db/src/store/event_deletion/postgres_tests.rs
+++ b/crates/buzz-db/src/store/event_deletion/postgres_tests.rs
@@ -48,6 +48,27 @@ fn event(kind: u16, secs: u64, tags: Vec<Tag>) -> nostr::Event {
         .expect("signed event")
 }
 
+/// Seed `event` as a row already in storage. The serving writers reject a
+/// channel event older than the `created_at` floor, and these tests pin
+/// months far in the past; the deletion paths under test still run for real.
+async fn seed(db: &Db, community: CommunityId, event: &nostr::Event, channel: Uuid) {
+    crate::test_support::insert_historical_event(&db.pool, community, event, channel).await;
+}
+
+/// [`seed`] for a reply or root that also carries thread metadata.
+async fn seed_with_thread(
+    db: &Db,
+    community: CommunityId,
+    event: &nostr::Event,
+    channel: Uuid,
+    meta: ThreadMetadataParams<'_>,
+) {
+    crate::test_support::insert_historical_event_with_thread_metadata(
+        &db.pool, community, event, channel, meta,
+    )
+    .await;
+}
+
 fn timestamp(event: &nostr::Event) -> DateTime<Utc> {
     DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).expect("timestamp")
 }
@@ -84,15 +105,12 @@ async fn known_targets_across_months_are_exact_and_tenant_scoped() {
     let jan = event(7, JAN, vec![]);
     let may = event(7, MAY, vec![]);
     for ev in [&jan, &may] {
-        db.insert_event(community, ev, Some(channel))
-            .await
-            .expect("insert");
-        db.insert_event(other, ev, Some(other_channel))
-            .await
-            .expect("other copy");
+        seed(&db, community, ev, channel).await;
+        seed(&db, other, ev, other_channel).await;
     }
     // Deliberately older than both targets: e-target deletion has no inferred
-    // causal chronology, and the deletion's month is not a target key.
+    // chronology, and the deletion's month is not a target key. It is stored
+    // without a channel, since a channel event this old is below the floor.
     let deletion = event(5, JAN - 86_400, vec![]);
     let missing = event(7, JAN + 1, vec![]);
     let wrong = DeletionTarget {
@@ -103,7 +121,7 @@ async fn known_targets_across_months_are_exact_and_tenant_scoped() {
         .insert_deletion_event(
             community,
             &deletion,
-            Some(channel),
+            None,
             None,
             &[wrong, target(&missing), target(&jan), target(&may)],
         )
@@ -119,7 +137,7 @@ async fn known_targets_across_months_are_exact_and_tenant_scoped() {
         db.insert_deletion_event(
             community,
             &deletion,
-            Some(channel),
+            None,
             None,
             &[target(&jan), target(&may)]
         )
@@ -150,9 +168,7 @@ async fn retention_free_contract() {
     let (other, other_channel) = tenant(&db.pool).await;
     let slot = format!("read-state:{}", "b".repeat(32));
     let marker = event(45011, JAN, vec![]);
-    db.insert_event(community, &marker, Some(channel))
-        .await
-        .expect("undeletable marker");
+    seed(&db, community, &marker, channel).await;
     for (kind, d, extra) in [
         (30078, slot.as_str(), ["t", "read-state"]),
         (
@@ -169,9 +185,9 @@ async fn retention_free_contract() {
             ];
             let ev = event(kind, secs, tags);
             for (tenant, ch) in [(community, channel), (other, other_channel)] {
-                db.replace_parameterized_event(tenant, &ev, d, Some(ch))
-                    .await
-                    .expect("retention-free insert");
+                // Each event has its own author, so nothing is superseded and
+                // storing it is the whole replacement.
+                seed(&db, tenant, &ev, ch).await;
                 assert_eq!(counts(&db, tenant, ev.id.as_bytes()).await, (1, 1, 1));
             }
             let deletion = event(5, JAN - 1, vec![]);
@@ -254,11 +270,12 @@ async fn migration_schema_known_target_retention_free_purge_cleans_mentions() {
 async fn known_target_batch_rollback_then_replay_repairs_and_counts_once() {
     let (db, community, channel) = fixture().await;
     let root = event(9, JAN, vec![]);
-    db.insert_event_with_thread_metadata(
+    seed_with_thread(
+        &db,
         community,
         &root,
-        Some(channel),
-        Some(ThreadMetadataParams {
+        channel,
+        ThreadMetadataParams {
             event_id: root.id.as_bytes(),
             event_created_at: timestamp(&root),
             channel_id: channel,
@@ -268,16 +285,16 @@ async fn known_target_batch_rollback_then_replay_repairs_and_counts_once() {
             root_event_created_at: None,
             depth: 0,
             broadcast: false,
-        }),
+        },
     )
-    .await
-    .expect("root");
+    .await;
     let reply = event(9, MAY, vec![]);
-    db.insert_event_with_thread_metadata(
+    seed_with_thread(
+        &db,
         community,
         &reply,
-        Some(channel),
-        Some(ThreadMetadataParams {
+        channel,
+        ThreadMetadataParams {
             event_id: reply.id.as_bytes(),
             event_created_at: timestamp(&reply),
             channel_id: channel,
@@ -287,10 +304,9 @@ async fn known_target_batch_rollback_then_replay_repairs_and_counts_once() {
             root_event_created_at: Some(timestamp(&root)),
             depth: 1,
             broadcast: false,
-        }),
+        },
     )
-    .await
-    .expect("reply");
+    .await;
     let counter = || async {
         sqlx::query_as::<_, (i32, i32)>("SELECT reply_count, descendant_count FROM thread_metadata WHERE community_id=$1 AND event_id=$2")
             .bind(community.as_uuid()).bind(root.id.as_bytes().as_slice())
@@ -300,20 +316,20 @@ async fn known_target_batch_rollback_then_replay_repairs_and_counts_once() {
     // A forbidden removal marker makes the second target fail after the first
     // was mutated. No test-only triggers or copied deletion implementation.
     let marker = event(45011, JAN, vec![]);
-    db.insert_event(community, &marker, Some(channel))
-        .await
-        .expect("marker");
+    seed(&db, community, &marker, channel).await;
     let reply_target = DeletionTarget {
         parent_event_id: Some(root.id.as_bytes()),
         root_event_id: Some(root.id.as_bytes()),
         ..target(&reply)
     };
+    // Stored without a channel on this attempt: a channel event this old is
+    // below the floor, which would fail the write before the targets run.
     let deletion = event(5, MAY + 1, vec![]);
     assert!(db
         .insert_deletion_event(
             community,
             &deletion,
-            Some(channel),
+            None,
             None,
             &[reply_target, target(&marker)]
         )
@@ -324,9 +340,7 @@ async fn known_target_batch_rollback_then_replay_repairs_and_counts_once() {
     assert_eq!(counter().await, (1, 1));
     // Emulate a historical torn write, then concurrently replay it from two
     // independent connections: exactly one transition decrements counters.
-    db.insert_event(community, &deletion, Some(channel))
-        .await
-        .expect("torn deletion");
+    seed(&db, community, &deletion, channel).await;
     let concurrent = Db::from_pool(
         PgPool::connect(&crate::test_support::database_url())
             .await
@@ -394,9 +408,7 @@ async fn production_known_target_statements_prune_to_one_month() {
             Tag::parse(["t", "read-state"]).expect("t"),
         ],
     );
-    db.replace_parameterized_event(community, &ev, &slot, Some(channel))
-        .await
-        .expect("insert");
+    seed(&db, community, &ev, channel).await;
     let deletion = event(5, MAY, vec![]);
     db.insert_deletion_event(community, &deletion, None, None, &[target(&ev)])
         .await
@@ -482,9 +494,7 @@ async fn production_known_target_deletion_locks_one_partition() {
             Tag::parse(["t", "read-state"]).expect("t"),
         ],
     );
-    db.replace_parameterized_event(community, &ev, &slot, Some(channel))
-        .await
-        .expect("insert");
+    seed(&db, community, &ev, channel).await;
     // The assertion is only meaningful if other partitions exist to be locked.
     let partitions: i64 =
         sqlx::query_scalar("SELECT count(*) FROM pg_inherits WHERE inhparent = 'events'::regclass")
@@ -534,9 +544,7 @@ async fn production_known_target_deletion_locks_one_partition() {
 async fn known_canvas_target_uses_canonical_replacement_lock() {
     let (db, community, channel) = fixture().await;
     let canvas = event(KIND_CANVAS as u16, JAN, vec![]);
-    db.insert_event(community, &canvas, Some(channel))
-        .await
-        .expect("canvas");
+    seed(&db, community, &canvas, channel).await;
     // The fixture's single connection identifies the backend that will delete.
     let deleter_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
         .fetch_one(&db.pool)
@@ -560,15 +568,10 @@ async fn known_canvas_target_uses_canonical_replacement_lock() {
     let delete_db = Db::from_pool(db.pool.clone());
     let id = canvas.id.to_bytes();
     let delete = tokio::spawn(async move {
+        // No channel: a channel event this old is below the floor.
         let deletion = event(5, MAY, vec![]);
         delete_db
-            .insert_deletion_event(
-                community,
-                &deletion,
-                Some(channel),
-                None,
-                &[target(&canvas)],
-            )
+            .insert_deletion_event(community, &deletion, None, None, &[target(&canvas)])
             .await
     });
     // Observe an actual advisory-lock wait, not merely a task that hasn't yet
diff --git a/crates/buzz-db/src/store/event_hydration/postgres_tests.rs b/crates/buzz-db/src/store/event_hydration/postgres_tests.rs
index 1150b2ed869..5e8acf1b407 100644
--- a/crates/buzz-db/src/store/event_hydration/postgres_tests.rs
+++ b/crates/buzz-db/src/store/event_hydration/postgres_tests.rs
@@ -363,8 +363,12 @@ async fn channel_create_resend_requires_exact_timestamp_and_live_community_row()
     let (db, a, b, keys) = fixture().await;
     let channel_id = Uuid::new_v4();
     let creator = keys.public_key().to_bytes();
+    // Created now: the serving writer rejects a channel event older than the
+    // `created_at` floor, and the first create must succeed. The malformed
+    // March row below still differs from it in the partition key.
+    let created_at = Timestamp::now();
     let event = EventBuilder::new(Kind::Custom(9007), "channel create")
-        .custom_created_at(Timestamp::from(JAN as u64))
+        .custom_created_at(created_at)
         .sign_with_keys(&keys)
         .unwrap();
     let channel = || NewChannel {
@@ -395,7 +399,7 @@ async fn channel_create_resend_requires_exact_timestamp_and_live_community_row()
     db.insert_event(b, &event, None).await.unwrap();
     sqlx::query("UPDATE events SET deleted_at = now() WHERE community_id = $1 AND id = $2 AND created_at = $3")
         .bind(a.as_uuid()).bind(event.id.to_bytes().to_vec())
-        .bind(DateTime::from_timestamp(JAN, 0).unwrap())
+        .bind(DateTime::from_timestamp(created_at.as_secs() as i64, 0).unwrap())
         .execute(&db.pool).await.unwrap();
     assert!(matches!(
         db.create_channel_with_event(a, channel_id, channel(), &event)
diff --git a/crates/buzz-db/src/store/thread.rs b/crates/buzz-db/src/store/thread.rs
index 5ae76ec594b..a452f2136f8 100644
--- a/crates/buzz-db/src/store/thread.rs
+++ b/crates/buzz-db/src/store/thread.rs
@@ -2097,6 +2097,34 @@ mod postgres_tests {
         .expect("insert top-level event");
     }
 
+    /// [`insert_root`] for a row already in storage: the serving writer
+    /// rejects an event older than the `created_at` floor.
+    async fn insert_historical_root(
+        pool: &PgPool,
+        community: CommunityId,
+        channel_id: Uuid,
+        event: &nostr::Event,
+    ) {
+        crate::test_support::insert_historical_event_with_thread_metadata(
+            pool,
+            community,
+            event,
+            channel_id,
+            ThreadMetadataParams {
+                event_id: event.id.as_bytes(),
+                event_created_at: event_created_at(event),
+                channel_id,
+                parent_event_id: None,
+                parent_event_created_at: None,
+                root_event_id: None,
+                root_event_created_at: None,
+                depth: 0,
+                broadcast: true,
+            },
+        )
+        .await;
+    }
+
     /// Insert a depth-1 reply under `root`, with the given broadcast flag.
     async fn insert_reply(
         pool: &PgPool,
@@ -2327,7 +2355,7 @@ mod postgres_tests {
                 .custom_created_at(Timestamp::from(at))
                 .sign_with_keys(&author)
                 .expect("sign");
-            insert_root(&pool, community, channel.id, &event).await;
+            insert_historical_root(&pool, community, channel.id, &event).await;
             newest_first.push(event.id);
         }
 
@@ -2437,7 +2465,7 @@ mod postgres_tests {
             .custom_created_at(Timestamp::from(1_748_736_000)) // 2025-06-01
             .sign_with_keys(&author)
             .expect("sign");
-        insert_root(&pool, community, channel.id, &ancient).await;
+        insert_historical_root(&pool, community, channel.id, &ancient).await;
         let partition_of = |id: nostr::EventId| {
             let pool = pool.clone();
             async move {
@@ -2582,16 +2610,14 @@ mod postgres_tests {
         )
         .await
         .expect("create channel");
-        let db = crate::Db::from_pool(pool.clone());
         let mut partitions = Vec::new();
         for at in [Utc::now().timestamp() as u64 - 10, 1_748_736_000] {
             let event = EventBuilder::new(Kind::Custom(9), format!("at {at}"))
                 .custom_created_at(nostr::Timestamp::from(at))
                 .sign_with_keys(&author)
                 .expect("sign");
-            db.insert_event(community, &event, Some(channel.id))
-                .await
-                .expect("insert");
+            crate::test_support::insert_historical_event(&pool, community, &event, channel.id)
+                .await;
             partitions.push(
                 sqlx::query_scalar::<_, String>(
                     "SELECT tableoid::regclass::text FROM events WHERE id = $1",
diff --git a/crates/buzz-db/src/test_support.rs b/crates/buzz-db/src/test_support.rs
index fb9b0b56e28..6fa07632bcb 100644
--- a/crates/buzz-db/src/test_support.rs
+++ b/crates/buzz-db/src/test_support.rs
@@ -369,6 +369,36 @@ pub(crate) async fn insert_historical_event(
     tx.commit().await.expect("commit historical fixture");
 }
 
+/// [`insert_historical_event`] plus the thread metadata the serving writer
+/// records for it, through the same `insert_thread_metadata` store function
+/// (parent and root stubs, reply and descendant counters). For aged rows whose
+/// thread state a test reads or deletes through the production paths.
+#[cfg(test)]
+pub(crate) async fn insert_historical_event_with_thread_metadata(
+    pool: &sqlx::PgPool,
+    community: buzz_core::CommunityId,
+    event: &nostr::Event,
+    channel: uuid::Uuid,
+    meta: crate::event::ThreadMetadataParams<'_>,
+) {
+    insert_historical_event(pool, community, event, channel).await;
+    crate::thread::insert_thread_metadata(
+        pool,
+        community,
+        meta.event_id,
+        meta.event_created_at,
+        meta.channel_id,
+        meta.parent_event_id,
+        meta.parent_event_created_at,
+        meta.root_event_id,
+        meta.root_event_created_at,
+        meta.depth,
+        meta.broadcast,
+    )
+    .await
+    .expect("fixture thread metadata");
+}
+
 /// Move a community fixture into a deletion lifecycle state the way the
 /// executor does: the tombstone trigger only accepts lifecycle changes from a
 /// transaction carrying the executor GUCs for that community, including its
diff --git a/crates/buzz-db/src/test_support/recent_pass_tests.rs b/crates/buzz-db/src/test_support/recent_pass_tests.rs
index db13662866f..2654d2bceec 100644
--- a/crates/buzz-db/src/test_support/recent_pass_tests.rs
+++ b/crates/buzz-db/src/test_support/recent_pass_tests.rs
@@ -27,7 +27,6 @@ where
 {
     use nostr::{EventBuilder, Keys, Kind, Timestamp};
 
-    let db = crate::Db::from_pool(pool.clone());
     let author = Keys::generate();
     let now = chrono::Utc::now();
     let floor = crate::store::recent_pass::floor_at(now).timestamp() as u64;
@@ -37,11 +36,9 @@ where
             .custom_created_at(Timestamp::from(at))
             .sign_with_keys(&author)
             .expect("sign");
-        let db = db.clone();
         async move {
-            db.insert_event(community, &event, Some(channel))
-                .await
-                .expect("insert");
+            // Seeded as stored rows: the serving writers reject the aged ones.
+            crate::test_support::insert_historical_event(pool, community, &event, channel).await;
             event.id
         }
     };

From b046df76d361e65575a7ba1a8c5f30e6d0403b55 Mon Sep 17 00:00:00 2001
From: Luke Tornquist <tornquist@squareup.com>
Date: Tue, 10 Nov 2026 08:38:49 -0800
Subject: [PATCH 7/7] style(buzz-db): rustfmt the soft_deleted_aged fixture
 signature

Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Luke Tornquist <tornquist@squareup.com>
---
 crates/buzz-db/src/store/event_floor_postgres_tests.rs | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/crates/buzz-db/src/store/event_floor_postgres_tests.rs b/crates/buzz-db/src/store/event_floor_postgres_tests.rs
index 3d2baef9c01..24898fa7a1d 100644
--- a/crates/buzz-db/src/store/event_floor_postgres_tests.rs
+++ b/crates/buzz-db/src/store/event_floor_postgres_tests.rs
@@ -269,7 +269,13 @@ async fn addressable_event_floor_rejects_old_and_accepts_fresh() {
 }
 
 /// Seed an aged, stored and then soft-deleted event, as NIP-09 leaves it.
-async fn soft_deleted_aged(db: &Db, community: CommunityId, channel: Uuid, keys: &Keys, kind: u16) -> Event {
+async fn soft_deleted_aged(
+    db: &Db,
+    community: CommunityId,
+    channel: Uuid,
+    keys: &Keys,
+    kind: u16,
+) -> Event {
     let now = database_clock(db).await;
     let aged = signed(
         keys,