Add application-owned writer lock foundations

ARCHITECTURE.md

@@ -433,6 +433,35 @@ All database access. Uses `sqlx::query()` (runtime, not compile-time macros) —
 - Approval tokens: `create_approval` receives the raw token and hashes it internally with SHA-256.
 - DDL injection protection in partition manager: allowlist of table names + strict suffix/date validators.
 
+### Supported Writer Contract (transition state)
+
+The repository is moving toward a **supported-writer contract** where relay-owned
+write APIs provide the only supported mutation path for tenant data. The current
+shape is an explicit transition, not a flag day:
+
+- **Trajectory:** converge from trigger/function/FK-enforced fencing toward
+  application-owned transaction protocols, while keeping existing DB backstops
+  in place until coverage and fleet gates are proven.
+- **Community fence (transition foundation):** runtime admission APIs now offer
+  shared community deletion locks, while deletion lifecycle transitions take the
+  matching exclusive lock. Multi-community batches lock in stable UUID order;
+  unmigrated paths still rely on trigger/function backstops.
+- **Replica floor (transition foundation):** runtime now provides a shared
+  replica-floor lock helper for compliant channel-event writers, and the writer
+  probe handshake takes the exclusive counterpart before `S`/activity/token.
+  Commit-time trigger+GUC enforcement remains authoritative for all paths.
+- **Dual enforcement (current):** application-owned lock+precheck paths run in
+  front of the existing trigger/function enforcement; commit-time trigger checks
+  remain authoritative during this phase.
+- **Role separation:** relay/runtime code owns admission and lock protocols;
+  operator maintenance/backfill workflows must either use those protocols or run
+  under explicit reviewed procedures that keep routing fences closed.
+- **Unsupported shape:** direct owner SQL that bypasses supported transaction and
+  lock protocols is not a supported write path.
+- **Fleet gates and reconciliation:** startup/fence probes verify guard catalog +
+  behavior, pgschema bootstraps run reconciliation, and lock/pool metrics remain
+  the operational evidence path for rollout safety.
+
 **Does NOT:** cache queries, implement connection pooling logic (delegated to sqlx), or make network calls outside Postgres.
 
 ---

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

@@ -1143,6 +1143,58 @@ impl Db {
             .map_err(Into::into)
     }
 
+    /// Begin an event-write transaction and guard one community through the
+    /// stable multi-community admission path.
+    pub async fn begin_community_write_transaction(
+        &self,
+        community: CommunityId,
+    ) -> Result<sqlx::Transaction<'static, sqlx::Postgres>> {
+        self.begin_community_write_transaction_batch(&[community])
+            .await
+    }
+
+    /// Begin one event-write transaction and guard communities in deterministic
+    /// UUID order under shared deletion/admission locks.
+    ///
+    /// The batch must include every community the transaction may mutate. The
+    /// sorted lock order prevents opposite-order callers from deadlocking when
+    /// overlapping community sets are locked in one transaction.
+    pub async fn begin_community_write_transaction_batch(
+        &self,
+        communities: &[CommunityId],
+    ) -> Result<sqlx::Transaction<'static, sqlx::Postgres>> {
+        if communities.is_empty() {
+            return Err(DbError::InvalidData(
+                "community write transaction batch requires at least one community".to_string(),
+            ));
+        }
+        let mut tx = self.begin_event_write_transaction().await?;
+        let mut ordered = communities.to_vec();
+        ordered.sort_unstable();
+        ordered.dedup();
+        let store = self.deletion_store();
+        for community in ordered {
+            store.guard_transaction(&mut tx, community).await?;
+        }
+        Ok(tx)
+    }
+
+    /// Begin an event-write transaction that takes the shared replica-floor
+    /// advisory lock.
+    ///
+    /// This is a lock-ordering foundation only. Floor correctness remains
+    /// authoritative at commit time via the existing trigger/GUC contract.
+    pub async fn begin_replica_floor_locked_event_write_transaction(
+        &self,
+    ) -> Result<sqlx::Transaction<'static, sqlx::Postgres>> {
+        let mut tx = self.begin_event_write_transaction().await?;
+        sqlx::query("SELECT pg_advisory_xact_lock_shared($1)")
+            .bind(replica_fence::REPLICA_FLOOR_LOCK_KEY)
+            .execute(&mut *tx)
+            .await?;
+        Ok(tx)
+    }
+
     /// Begin an event-write transaction through the pre-operation API name.
     ///
     /// New callers should use [`Self::begin_event_write_transaction`] so the

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

@@ -10,10 +10,12 @@
 //!    is armed per session via the `buzz.created_at_floor` GUC, which the
 //!    relay's writer pool sets on every connection.
 //! 2. **Ordered heartbeat handshake** (this module): on one pinned writer
-//!    connection, separately-awaited statements sample
-//!    `S = clock_timestamp()`, then scan `pg_stat_activity` for the oldest
-//!    open transaction, then — **last** — commit heartbeat token `M` via a
-//!    single-row `UPDATE replica_heartbeat ... RETURNING token, epoch`
+//!    connection, one transaction first takes the exclusive replica-floor
+//!    advisory lock (draining compliant shared-lock channel writers), then
+//!    separately-awaited statements sample `S = clock_timestamp()`, scan
+//!    `pg_stat_activity` for the oldest open transaction, and — **last** —
+//!    commit heartbeat token `M` via a single-row
+//!    `UPDATE replica_heartbeat ... RETURNING token, epoch`
 //!    (migration 0026). Because the single-row UPDATE serializes all pods'
 //!    probes, tokens are globally commit-ordered. A reader **session** that
 //!    observes `token >= M` on its own connection has, by WAL/storage replay
@@ -73,6 +75,14 @@ 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;
 
+/// Deployment-global advisory-lock key ordering compliant channel-event write
+/// transactions against the writer probe handshake.
+///
+/// Shared holders are channel-event write transactions validating `created_at`
+/// under the armed floor contract; the probe takes the exclusive lock before
+/// sampling and publishing its heartbeat token.
+pub const REPLICA_FLOOR_LOCK_KEY: i64 = 0x62757a7a666c6f72;
+
 /// Safety margin subtracted from the fence on top of the floor.
 ///
 /// All proof timestamps (`clock_timestamp()`, `xact_start`, the guard's
@@ -550,22 +560,28 @@ pub enum ProbeError {
     HeartbeatRowMissing,
 }
 
-/// Take one ordered writer sample: S, then activity scan, then commit the
-/// heartbeat token **last**.
+/// Take one ordered writer sample: under the exclusive replica-floor lock,
+/// sample S, then activity scan, then commit the heartbeat token **last**.
 ///
-/// The statements are separately awaited on a single pinned connection;
+/// The statements are separately awaited on one pinned writer transaction;
 /// a single SELECT would not guarantee evaluation order across the
 /// subexpressions, reopening the race this ordering exists to close.
 async fn sample_writer(writer: &PgPool) -> Result<WriterSample, ProbeError> {
-    let mut conn = crate::observability::acquire_writer(
+    let connection = crate::observability::acquire_writer(
         writer,
         crate::observability::WriterOperation::Maintenance,
     )
     .await?;
+    let mut tx = sqlx::Transaction::begin(connection, None).await?;
+
+    sqlx::query("SELECT pg_advisory_xact_lock($1)")
+        .bind(REPLICA_FLOOR_LOCK_KEY)
+        .execute(&mut *tx)
+        .await?;
 
     // 1. S first.
     let sampled_at: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
-        .fetch_one(&mut *conn)
+        .fetch_one(&mut *tx)
         .await?;
 
     // 2. Activity scan. Classification (fail closed on anything unknown):
@@ -612,7 +628,7 @@ async fn sample_writer(writer: &PgPool) -> Result<WriterSample, ProbeError> {
             ) AS masked
         "#,
     )
-    .fetch_one(&mut *conn)
+    .fetch_one(&mut *tx)
     .await?;
     let masked: i64 = row.get("masked");
     if masked > 0 {
@@ -629,17 +645,19 @@ async fn sample_writer(writer: &PgPool) -> Result<WriterSample, ProbeError> {
     let row = sqlx::query(
         "UPDATE replica_heartbeat SET token = token + 1 WHERE id = 1 RETURNING token, epoch",
     )
-    .fetch_optional(&mut *conn)
+    .fetch_optional(&mut *tx)
     .await?
     .ok_or(ProbeError::HeartbeatRowMissing)?;
 
-    Ok(WriterSample {
+    let sample = WriterSample {
         sampled_at,
         oldest_xact_start,
         token: row.get("token"),
         epoch: row.get("epoch"),
         committed_at,
-    })
+    };
+    tx.commit().await?;
+    Ok(sample)
 }
 
 /// The fence wall proved by one handshake:
@@ -1028,10 +1046,60 @@ mod postgres_tests {
         tx.rollback().await.expect("rollback");
     }
 
+    /// The sample must not capture `S` while a compliant shared writer lock is
+    /// still held. It must block until release, then sample.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn sample_writer_waits_for_shared_floor_lock_before_sampling_time() {
+        let (admin, pool, name) = scratch_db().await;
+
+        let mut blocker = pool.begin().await.expect("begin shared-lock blocker");
+        sqlx::query("SELECT pg_advisory_xact_lock_shared($1)")
+            .bind(REPLICA_FLOOR_LOCK_KEY)
+            .execute(&mut *blocker)
+            .await
+            .expect("hold shared floor lock");
+
+        let sample_pool = pool.clone();
+        let sampling = tokio::spawn(async move { sample_writer(&sample_pool).await });
+        let mut sampling = sampling;
+        assert!(
+            tokio::time::timeout(Duration::from_millis(100), &mut sampling)
+                .await
+                .is_err(),
+            "sample_writer should block while a shared floor lock is held"
+        );
+
+        let marker_before_release: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
+            .fetch_one(&pool)
+            .await
+            .expect("capture marker while lock held");
+
+        blocker
+            .rollback()
+            .await
+            .expect("release shared-lock blocker");
+
+        let sample = tokio::time::timeout(Duration::from_secs(5), sampling)
+            .await
+            .expect("sample should complete after shared lock release")
+            .expect("sampling task")
+            .expect("sample writer");
+        assert!(
+            sample.sampled_at >= marker_before_release,
+            "sample time {:?} must be at or after marker {:?}",
+            sample.sampled_at,
+            marker_before_release
+        );
+
+        drop_scratch_db(&admin, pool, &name).await;
+    }
+
     /// An unprivileged probe role sees NULL `state`/`xact_start` for other
     /// sessions' rows in `pg_stat_activity`. The oldest-xact term is then
     /// untrustworthy and the sample must fail closed (`MaskedActivity`) —
     /// never silently `MIN()` the hidden row away.
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn cluster_global_sample_writer_fails_closed_when_activity_is_masked() {

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

@@ -2825,6 +2825,117 @@ async fn armed_pool_rejects_old_channel_inserts_through_public_api() {
     db.pool.close().await;
 }
 
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn replica_floor_writer_transaction_holds_shared_lock() {
+    let admin = PgPool::connect(&admin_url().await)
+        .await
+        .expect("connect admin");
+    let (seed_pool, name) = create_scratch_db(&admin, "floor_writer_foundation").await;
+
+    let base = admin_url().await;
+    let idx = base.rfind('/').expect("db url has a path segment");
+    let scratch_url = format!("{}/{}", &base[..idx], name);
+    let db = Db::new(&DbConfig {
+        database_url: scratch_url,
+        max_connections: 2,
+        ..DbConfig::default()
+    })
+    .await
+    .expect("connect armed Db");
+
+    let writer = db
+        .begin_replica_floor_locked_event_write_transaction()
+        .await
+        .expect("open compliant floor-guarded writer tx");
+
+    let mut contender = db.pool.begin().await.expect("begin exclusive contender");
+    let exclusive_taken: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)")
+        .bind(crate::replica_fence::REPLICA_FLOOR_LOCK_KEY)
+        .fetch_one(&mut *contender)
+        .await
+        .expect("probe exclusive floor lock");
+    assert!(
+        !exclusive_taken,
+        "compliant writer must hold the shared replica-floor advisory lock"
+    );
+
+    contender
+        .rollback()
+        .await
+        .expect("rollback exclusive contender");
+    writer.rollback().await.expect("rollback writer tx");
+    db.pool.close().await;
+    drop_scratch_db(&admin, seed_pool, &name).await;
+}
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn replica_floor_probe_waits_for_shared_writer_and_records_after_release() {
+    let admin = PgPool::connect(&admin_url().await)
+        .await
+        .expect("connect admin");
+    let (seed_pool, name) = create_scratch_db(&admin, "floor_probe_foundation").await;
+
+    let base = admin_url().await;
+    let idx = base.rfind('/').expect("db url has a path segment");
+    let scratch_url = format!("{}/{}", &base[..idx], name);
+    let db = Db::new(&DbConfig {
+        database_url: scratch_url,
+        max_connections: 2,
+        ..DbConfig::default()
+    })
+    .await
+    .expect("connect armed Db");
+
+    let token_before: i64 = sqlx::query_scalar("SELECT token FROM replica_heartbeat WHERE id = 1")
+        .fetch_one(&db.pool)
+        .await
+        .expect("read token before probe");
+
+    let writer = db
+        .begin_replica_floor_locked_event_write_transaction()
+        .await
+        .expect("open compliant floor-guarded writer tx");
+
+    let probe_pool = db.pool.clone();
+    let probe_fence = std::sync::Arc::clone(db.fence());
+    let probing = tokio::spawn(async move {
+        crate::replica_fence::probe_once(&probe_pool, probe_fence.as_ref()).await
+    });
+    let mut probing = probing;
+    assert!(
+        tokio::time::timeout(std::time::Duration::from_millis(100), &mut probing)
+            .await
+            .is_err(),
+        "probe must wait for the exclusive floor lock while compliant writer is open"
+    );
+
+    writer
+        .rollback()
+        .await
+        .expect("release shared floor writer");
+    let entry = tokio::time::timeout(std::time::Duration::from_secs(5), probing)
+        .await
+        .expect("probe must complete after writer release")
+        .expect("probe task")
+        .expect("probe succeeds");
+
+    assert_eq!(
+        entry.token,
+        token_before + 1,
+        "existing handshake must publish one token via probe_once"
+    );
+    assert_eq!(
+        db.fence().verified_through(),
+        Some(entry.fence_wall),
+        "probe entry must be retained in the in-memory fence ring"
+    );
+
+    db.pool.close().await;
+    drop_scratch_db(&admin, seed_pool, &name).await;
+}
+
 /// `spawn_fence_probe` must verify the floor guard before letting the
 /// probe run — catalog shape AND observed behavior — and refuse on
 /// sabotage. This is the production gate for a relay running with