jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 12:40:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

BUZZ-175: Add application-owned writer lock foundations (#7706)

Browse Source 
## Summary

Buzz is moving database invariants from triggers and foreign keys into
application-owned transaction protocols. This PR adds the first
transition foundations while preserving all current database backstops.

- Add a shared community-write lock for single-community writer
transactions, and make deletion lifecycle transitions take the matching
exclusive lock. A supported serving transaction never spans communities.
- Add shared event-writer and exclusive replica-probe lock ordering
around the existing heartbeat publication model.
- Document the supported writer contract, reviewed backfill path, and
transition gates.
- Keep one replica-floor publication concept and remove an unrelated
reconciliation-script edit found during review. This narrows the change
rather than adding parallel protocols.

The blast radius is limited to new opt-in writer helpers plus lock
acquisition in deletion and replica probing. Existing trigger and GUC
enforcement remains authoritative, so callers can migrate in later pull
requests.

### Related issue

[BUZZ-175](https://linear.app/squareup/issue/BUZZ-175). This is the
first implementation slice from the OSC migration plan.

### Testing

No manual testing. CI covers the Rust checks and database tests added
here.

Generated with [Codex](https://openai.com/codex)

---------

Signed-off-by: tornquist <tornquist@squareup.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Luke Tornquist
2026-10-02 16:53:09 -04:00
committed by GitHub
parent ea143ea64b
commit fe9e2409a0
6 changed files with 499 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
ARCHITECTURE.md
View File

@@ -484,6 +484,40 @@ 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
one shared community deletion lock per transaction, while deletion lifecycle
transitions take the matching exclusive lock. A supported serving transaction
never spans communities; 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 sampling `S`, scanning
activity, and committing the heartbeat token last. `probe_once` remains the
sole token/fence-wall publication path; there is no separately persisted active
floor cutoff. 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. Removing those backstops is a separate
gated migration, not part of the lock foundation.
- **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.
---

29
crates/buzz-db/src/runtime/mod.rs
View File

@@ -1237,6 +1237,35 @@ impl Db {
.map_err(Into::into)
}
/// Begin an event-write transaction and guard its community against
/// concurrent deletion.
pub async fn begin_community_write_transaction(
&self,
community: CommunityId,
) -> Result<sqlx::Transaction<'static, sqlx::Postgres>> {
let mut tx = self.begin_event_write_transaction().await?;
self.deletion_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

92
crates/buzz-db/src/runtime/replica_fence.rs
View File

@@ -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,6 +1046,54 @@ 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 mut sampling = tokio::spawn(async move { sample_writer(&sample_pool).await });
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`) —

125
crates/buzz-db/src/runtime/tests.rs
View File

@@ -3118,6 +3118,131 @@ 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: 3,
..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 shared_contender = db.pool.begin().await.expect("begin shared contender");
let shared_taken: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock_shared($1)")
.bind(crate::replica_fence::REPLICA_FLOOR_LOCK_KEY)
.fetch_one(&mut *shared_contender)
.await
.expect("probe shared floor lock");
assert!(
shared_taken,
"compliant writer must allow another shared replica-floor lock holder"
);
shared_contender
.rollback()
.await
.expect("rollback shared contender");
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 mut probing = tokio::spawn(async move {
crate::replica_fence::probe_once(&probe_pool, probe_fence.as_ref()).await
});
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

231
crates/buzz-db/src/store/deletion.rs
View File

@@ -1807,6 +1807,7 @@ impl DeletionStore {
.bind(token.community_id.as_uuid())
.fetch_one(&mut *tx)
.await?;
verify_lease(&mut tx, token, DeletionStage::Approved).await?;
sqlx::query(
"UPDATE community_deletion_requests SET pre_quiesce_archived_at = $2, \
quiescing_started_at = now(), updated_at = now() \
@@ -2665,9 +2666,16 @@ impl DeletionStore {
.await?
.map(CommunityId::from_uuid)
.ok_or_else(|| DbError::NotFound(format!("community deletion {request_id}")))?;
// Every lifecycle transition takes the community lock before any row lock.
// Inverting this order lets abort and the executor deadlock each other.
// Every lifecycle transition takes the community advisory lock before any
// row lock, then the community row before the request row. Inverting
// either order lets abort and the executor deadlock each other.
lock_community_deletion(&mut tx, community_id).await?;
let (old_generation, current_archived_at): (i64, Option<DateTime<Utc>>) = sqlx::query_as(
"SELECT deletion_fence_generation, archived_at FROM communities WHERE id = $1 FOR UPDATE",
)
.bind(community_id.as_uuid())
.fetch_one(&mut *tx)
.await?;
let row = sqlx::query("SELECT * FROM community_deletion_requests WHERE id = $1 FOR UPDATE")
.bind(request_id)
.fetch_optional(&mut *tx)
@@ -2703,12 +2711,6 @@ impl DeletionStore {
"deletion {request_id} cannot abort while {active_writes} serving write lease(s) remain active"
)));
}
let (old_generation, current_archived_at): (i64, Option<DateTime<Utc>>) = sqlx::query_as(
"SELECT deletion_fence_generation, archived_at FROM communities WHERE id = $1 FOR UPDATE",
)
.bind(request.community_id.as_uuid())
.fetch_one(&mut *tx)
.await?;
let new_generation = old_generation.checked_add(1).ok_or_else(|| {
DbError::DeletionSafety("community deletion fence generation overflow".to_string())
})?;
@@ -6488,6 +6490,12 @@ mod postgres_tests {
.execute(&mut *gate)
.await
.expect("hold community lock");
let mut request_gate = db.pool.begin().await.expect("begin request gate");
sqlx::query("SELECT id FROM community_deletion_requests WHERE id = $1 FOR UPDATE")
.bind(request.id)
.execute(&mut *request_gate)
.await
.expect("hold request row gate");
let abort_store = store.clone();
let aborting = tokio::spawn(async move {
@@ -6509,25 +6517,67 @@ mod postgres_tests {
"forward transition must queue on the same lock"
);
gate.commit().await.expect("release lock gate");
tokio::time::sleep(Duration::from_millis(50)).await;
assert!(
!aborting.is_finished(),
"abort must still be queued while the request row stays locked"
);
assert!(
!forwarding.is_finished(),
"quiescing must still be queued while the request row stays locked"
);
request_gate
.commit()
.await
.expect("release request row gate");
let aborted = tokio::time::timeout(Duration::from_secs(5), aborting)
.await
.expect("abort must not deadlock")
.expect("abort task")
.expect("abort wins lock queue");
assert_eq!(aborted.stage, DeletionStage::Aborted);
let forward_error = tokio::time::timeout(Duration::from_secs(5), forwarding)
.expect("abort task");
let forwarding = tokio::time::timeout(Duration::from_secs(5), forwarding)
.await
.expect("forward transition must not deadlock")
.expect("forward task")
.expect_err("post-lock lease verification rejects aborted request");
.expect("forward task");
assert!(
!matches!(
&forward_error,
DbError::Sqlx(sqlx::Error::Database(error)) if error.code().as_deref() == Some("40P01")
&aborted,
Err(DbError::Sqlx(sqlx::Error::Database(error)))
if error.code().as_deref() == Some("40P01")
),
"abort must not report a PostgreSQL deadlock"
);
assert!(
!matches!(
&forwarding,
Err(DbError::Sqlx(sqlx::Error::Database(error)))
if error.code().as_deref() == Some("40P01")
),
"serialization must not report a PostgreSQL deadlock"
);
match (aborted, forwarding) {
(Ok(aborted), Ok(())) => {
assert_eq!(aborted.stage, DeletionStage::Aborted);
let reloaded = store.get(request.id).await.expect("reload request");
assert_eq!(reloaded.stage, DeletionStage::Aborted);
assert!(
reloaded.quiescing_started_at.is_some(),
"successful quiescing should persist its durable intent before abort"
);
}
(Ok(aborted), Err(DbError::AccessDenied(message))) => {
assert_eq!(aborted.stage, DeletionStage::Aborted);
assert!(
message.contains("stale deletion lease"),
"post-lock quiescing must reject a stale aborted lease: {message}"
);
}
(aborted, forwarding) => panic!(
"unexpected abort/quiesce outcome: abort={aborted:?}, forward={forwarding:?}"
),
}
}
#[tokio::test]
@@ -6808,6 +6858,155 @@ mod postgres_tests {
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn compliant_writer_shared_xact_lock_blocks_deletion_exclusive_xact_lock() {
let (db, _) = store().await;
let community = db
.ensure_configured_community(&format!(
"community-lock-contract-{}.example",
Uuid::new_v4().simple()
))
.await
.expect("create community")
.id;
let writer = db
.begin_community_write_transaction(community)
.await
.expect("open writer transaction with community lock");
let mut shared_contender = db.pool.begin().await.expect("begin shared contender");
let shared_taken: bool = sqlx::query_scalar(
"SELECT pg_try_advisory_xact_lock_shared(community_deletion_lock_key($1))",
)
.bind(community.as_uuid())
.fetch_one(&mut *shared_contender)
.await
.expect("try shared deletion lock");
assert!(
shared_taken,
"compliant writer must allow another shared community deletion lock holder"
);
shared_contender
.rollback()
.await
.expect("rollback shared contender");
let mut deleter = db.pool.begin().await.expect("begin deletion contender");
let exclusive_taken: bool =
sqlx::query_scalar("SELECT pg_try_advisory_xact_lock(community_deletion_lock_key($1))")
.bind(community.as_uuid())
.fetch_one(&mut *deleter)
.await
.expect("try deletion exclusive lock");
assert!(
!exclusive_taken,
"compliant writer must hold shared community lock that blocks deletion exclusive lock"
);
deleter
.rollback()
.await
.expect("rollback deletion contender");
writer
.rollback()
.await
.expect("rollback writer transaction");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn community_fence_rejects_fresh_begin_transaction_after_fence() {
let (db, store) = store().await;
let (request, _) = inventoried_request(&db, &store).await;
store
.approve(request.id, "approver", None)
.await
.expect("approve");
let claim = store
.claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION)
.await
.expect("claim")
.expect("won claim");
store.begin_quiescing(&claim.lease).await.expect("quiesce");
store.fence(&claim.lease).await.expect("fence");
let error = db
.begin_community_write_transaction(request.community_id)
.await
.expect_err("fenced community must reject fresh write admission");
assert!(
matches!(
&error,
DbError::AccessDenied(message)
if message
== &format!("community {} is write-fenced (fenced)", request.community_id)
),
"expected write-fenced access denial, got: {error:#}"
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn begin_quiescing_waits_for_open_admitted_writer_note_update() {
let (db, store) = store().await;
let (request, _) = inventoried_request(&db, &store).await;
store
.approve(request.id, "approver", None)
.await
.expect("approve");
let claim = store
.claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION)
.await
.expect("claim")
.expect("won claim");
let pubkey = [7_u8; 32];
let added_by = [8_u8; 32];
assert!(
db.add_to_allowlist(request.community_id, &pubkey, &added_by, Some("before"))
.await
.expect("seed allowlist row"),
"seeded allowlist row must insert"
);
let mut admitted_writer = db
.begin_community_write_transaction(request.community_id)
.await
.expect("open admitted writer");
let updated = sqlx::query(
"UPDATE pubkey_allowlist SET note = $3 WHERE community_id = $1 AND pubkey = $2",
)
.bind(request.community_id.as_uuid())
.bind(pubkey.to_vec())
.bind("during")
.execute(&mut *admitted_writer)
.await
.expect("update allowlist note")
.rows_affected();
assert_eq!(updated, 1, "note-only update must touch the seeded row");
let store_for_quiesce = store.clone();
let lease = claim.lease.clone();
let quiescing =
tokio::spawn(async move { store_for_quiesce.begin_quiescing(&lease).await });
tokio::time::sleep(Duration::from_millis(50)).await;
assert!(
!quiescing.is_finished(),
"begin_quiescing must wait for an admitted writer holding the shared deletion lock"
);
admitted_writer
.commit()
.await
.expect("release admitted writer");
quiescing
.await
.expect("quiesce task")
.expect("quiesce after writer release");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn fence_waits_for_open_write_and_rejects_it_after_transition() {

17
scripts/cutover/README.md
View File

@@ -19,6 +19,23 @@ Only when upgrading a Postgres that already holds **pre-1321 single-community
data** to 1321. A brand-new deployment does **not** run this — it provisions
from `migrations/0001_initial_schema.sql` (or `schema/schema.sql`) directly.
## Supported writer contract during transition
Outside this one-off cutover, direct owner SQL mutation is not a supported
steady-state path unless explicitly reviewed.
- Reviewed backfills and reconciliations should use relay-owned transaction and
lock protocols where they exist (one community fence per transaction and
replica-floor shared/exclusive lock ordering).
- If a workflow cannot use those protocols yet, execute it under an explicitly
reviewed procedure that keeps replica routing fences closed for the run.
- Existing trigger/function backstops remain the commit-time authority during
this transition. Their removal requires separate coverage and fleet gates;
this cutover does not authorize it.
- Role separation is deliberate: schema reconciliation scripts carry schema
convergence only; serving admission/fencing semantics remain in runtime/store
code paths with startup verification and metrics.
## Preconditions
- The DB is pristine pre-1321: no `communities` table, no `community_id`