feat(db): advance an empty partition catch-all under bounded locks (#7993)

## Summary

Adds bounded, serialized advancement of an empty right-edge partition
catch-all for `events` and `delivery_log`. The repaired catalogs
currently end in an empty catch-all from 2027-01-01, and the merged
manager (#6515) only audits them. Without this change, January writes
land in the catch-all, after which only an operator can move them.

When `BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED=true`, each relay keeps six
dedicated monthlies after the current month. It does this at startup and
on every audit interval by replacing the empty catch-all with canonical
monthlies and a new canonical `<table>_p_future` after the horizon. From
today's catalogs the first run creates January–March 2027 and moves the
catch-all to April 2027.

## Safety design

- **Refuse before locking.** The following return `operator_required`
without taking any table lock:
  - a populated catch-all, or one whose occupancy is unknown;
- a catch-all with child-only triggers, which the replacements would not
inherit;
  - a misaligned bound;
  - a `DEFAULT`, anomalous or pending-detach child;
  - a name collision.
- **One winner.** A schema-scoped `pg_try_advisory_xact_lock` is taken
before any DDL. Losers report `skipped_locked` and never issue DDL. The
winner re-plans from a fresh audit before taking any table lock and
checks for collisions. A runner that lost a race therefore sees the
committed layout as a no-op, not as colliding names.
- **Writer lock order.** First `ONLY` the parent is locked `ACCESS
EXCLUSIVE` under a 2 s `lock_timeout`. Then each foreign-key counterpart
(today only `communities`) is locked in OID order `SHARE ROW EXCLUSIVE
NOWAIT`, then the catch-all `NOWAIT`.
- The counterparts are read again once the parent is locked, and that
set is the one locked. Any FK DDL on or to the parent conflicts with the
parent lock, so the set can't change after that. An FK committed while
maintenance waited is therefore locked `NOWAIT` too, so attaching never
waits on it under the parent lock. The read before the lock remains as a
cheap early refusal.
- This is the order an insert takes locks (parent, then the FK check on
`communities`), so maintenance cannot deadlock with ingest. The internal
manual runbook used to lock `communities` first; it now uses this order
too.
- `SHARE ROW EXCLUSIVE` is the strongest lock the DDL takes on
`communities`; measured on PG 16 and 17, the catch-all `DROP` takes
none, as long as the catch-all's only foreign keys are those inherited
from the parent. A catch-all with its own foreign key, or one that
another table's foreign key references, is refused: once before any
lock, and again after the catch-all lock, where the check is
authoritative. Readers and other writers' `FOR KEY SHARE` checks
proceed. A concurrent `communities` writer makes the run fail fast with
`lock_timeout`.
  - A `40P01` deadlock victim is also reported as `lock_timeout`.
- Parents that are referenced by a foreign key, or that reference an
inherited table, are refused, including when such a key commits while
maintenance waits on the parent.
- **Re-plan under lock.** The catalog is re-audited under those locks.
The transaction proceeds only if the plan is unchanged.
- **Bounded.** `statement_timeout` 5 s,
`idle_in_transaction_session_timeout` 5 s, and a 10 s client deadline
per table. A deadline-abandoned connection is closed, never returned to
the pool, and the server rolls back anything uncommitted. If the
deadline fires after `COMMIT` was sent, the change may have landed; the
next audit reports the truth. The orphaned backend keeps its locks until
its in-flight statement ends, at most `statement_timeout`.
- **Verify before commit.** The changed catalog is re-audited inside the
transaction for bounds, canonical kinds, trigger parity and catch-all
emptiness, and serving safety must not regress. It also checks that
every new child has valid, ready clones of each parent index and
validated clones of each PK/unique/FK constraint.

- **Audits tolerate a concurrent advance.** Catalog rows come from the
statement snapshot, but `pg_get_expr` renders from current catalog
state. So an audit that overlaps another relay's `DROP` of a catch-all
can see a partition that no longer exists. The read-only audit detects
this (a NULL bound, or `42P01` from the emptiness probe) and retries up
to twice with a fresh snapshot.
- **Fail fast on audit probes.** If another relay's audit is probing the
catch-all's emptiness, the catch-all `NOWAIT` lock fails and the run
reports `lock_timeout`. The parent lock is then released immediately,
and the next runner or interval completes the advance.

Uncovered-month creation (`BUZZ_PARTITION_MANAGER_CREATE_ENABLED`,
default `true`) now runs through the same locked transaction, replacing
the unlocked per-month `CREATE`. That flag remains the kill switch for
all automatic partition DDL: with it off, advancement is disabled too.
With advancement disabled (the default) the periodic loop stays
read-only and startup behaves as before, apart from the locking.

A runner that planned DDL always re-audits before returning, even when a
peer did the work, so it never reports the pre-maintenance layout. A
serialized re-plan that finds only disabled work reports
`skipped_disabled`.

## Behavior changes

- The relay audit horizon and the `buzz-admin partition-audit` default
are now 6 months (`PARTITION_MANAGER_MONTHS_AHEAD`), up from 3. Until
advancement is enabled, months already covered by the catch-all make
`degraded` true. That correctly reports a runway under six months.
- New metric `buzz_partition_maintenance_runs_total{table,outcome}` with
outcomes `noop | skipped_disabled | created | advanced | skipped_locked
| operator_required | lock_timeout | deadline | error`.
`buzz_partition_create_attempts_total` keeps its per-month `created |
skipped_covered | error` labels. `error` there still means a DDL or
collision failure; `lock_timeout` and `deadline` runs add nothing to it,
and only `buzz_partition_maintenance_runs_total` records them.
- `ensure_future_partitions` (free function and `Db` method) is removed;
`maintain_partitions` is the one entry point.
- New advisory-lock label `lock_type="partition_maintenance"`.

## Rollout

Enable `BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED` per environment in BPCI:
`bb-block-staging` first, then production. The staging canary proves a
real advance because the six-month horizon already extends past January.
Fallback: rerun the internal manual runbook, which uses the same
parent-first lock order as of 2026-10-05. It keeps `ACCESS EXCLUSIVE` on
the counterparts, because its detach/attach can drop FK triggers on
them.

## Testing

- `cargo test -p buzz-db --lib partition`: 17 unit tests. The 10 new
ones cover:
- plans: repaired layout, policy gating, no-op, refusal states including
child-only triggers, December→January;
  - the create kill switch also stopping advancement;
  - the month cap at exactly 120;
- per-month metric outcomes for transient failures versus DDL failures;
  - the mid-audit-drop retry classifier.
- `cargo test -p buzz-db --lib partition -- --ignored` (PostgreSQL 17):
48/48 on 9 consecutive runs, including 17 new tests:
  - advance to the canonical layout, with routing and idempotent rerun;
  - replacing a canonically named catch-all in one transaction;
  - populated catch-all refused in under 1 s while the parent is locked;
  - a `communities` writer making the run fail fast and roll back;
  - a `communities` reader not blocking the advance;
- an in-flight insert's FK check, and its write to `communities`, behind
queued maintenance completing without a deadlock;
- a foreign key committed while maintenance waits on the parent, with a
writer on its new counterpart, failing fast as `lock_timeout` on that
table instead of waiting under the parent lock;
- a foreign key referencing `events`, committed while maintenance waits
on the parent, being refused under the parent lock;
- a catch-all's own foreign key, or a foreign key referencing it, being
refused before the parent lock with the key intact, and a catch-all key
committed while maintenance waits on the parent being refused after the
catch-all lock;
  - a `40P01` deadlock victim being classified as `lock_timeout`;
  - parent contention timing out without blocking the other table;
  - held advisory lock skipping with no DDL;
- concurrent runners converging, with at most a transient
`lock_timeout`;
- a stale audit after a concurrent advance being a no-op, and the losing
runner returning the committed layout;
- a re-plan that finds only disabled work reporting `skipped_disabled`;
  - an audit retrying when a listed partition is dropped mid-audit;
  - a catch-all with a child-only trigger being refused;
- client deadline abandoning the attempt and closing the session (scoped
to the test's own blocked backend);
  - catch-all name collision;
  - incoming foreign key refused.
- Mutation checks:
- restoring the old counterpart-first `ACCESS EXCLUSIVE` order fails the
reader test (`lock_timeout`) and the writer test (a ~1 s deadlock wait);
- dropping `NOWAIT` from the counterpart lock, or allowing a populated
catch-all, fails the matching tests;
- removing the parent-locked counterpart re-read fails the committed-FK
tests (the parent is held 2.06 s waiting on the new counterpart), and
ignoring its refusal fails the referencing-FK test;
- locking the counterparts `SHARE ROW EXCLUSIVE` before the parent fails
the writer test;
  - removing either catch-all foreign-key check fails its test.
- Full PostgreSQL lane: 716/718. Both failures are timing-sensitive
tests outside partition code, and each passes 3/3 when run alone:
  - `hanging_redis_peer_times_out_before_the_health_listener_binds`;
-
`readiness_check_cancellation_balances_waiter_and_inflight_connection`.
- `cargo clippy -p buzz-db -p buzz-relay -p buzz-admin --all-targets -D
warnings` and `cargo fmt --check` are clean.

## Related

Refs #2396.

This PR conflicts textually with #7994 (partition audit observability
follow-ups) in `crates/buzz-db/src/store/partition.rs` and
`crates/buzz-relay/src/main.rs`. Whichever PR merges second will be
rebased onto the other.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Ian Oberst <ioberst@block.xyz>
Co-authored-by: Ian Oberst <ioberst@block.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

.env.example

@@ -49,8 +49,13 @@ REDIS_URL=redis://localhost:6379
 # BUZZ_DB_POOL_SIZE=50
 # Read-only partition catalog audit interval, clamped to 60s..24h (default 900s).
 # BUZZ_PARTITION_AUDIT_INTERVAL_SECS=900
-# Emergency kill switch for monthly partition CREATE; audits remain enabled.
+# Emergency kill switch for all automatic partition DDL, including catch-all
+# advancement; audits remain enabled.
 # BUZZ_PARTITION_MANAGER_CREATE_ENABLED=true
+# Opt in to replacing an empty partition catch-all with six months of dedicated
+# monthlies, at startup and on every audit interval (default false). Has no
+# effect while the kill switch above is off.
+# BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED=false
 
 # Writer-session Postgres timeouts for buzz-db-backed pools and the relay audit
 # pool, all in milliseconds; 0 disables. The separately deployed push gateway

ARCHITECTURE.md

@@ -462,7 +462,7 @@ All database access. Uses `sqlx::query()` (runtime, not compile-time macros) —
 | `channel.rs` | Channel CRUD, membership management, role enforcement (transactional) |
 | `feed.rs` | `query_mentions` (INNER JOIN event_mentions), `query_needs_action`, `query_activity` |
 | `workflow.rs` | Full workflow/run/approval CRUD; SHA-256 hashed approval tokens |
-| `partition.rs` | Monthly range partitioning for `events` and `delivery_log` tables |
+| `partition.rs` | Monthly range partitioning for `events` and `delivery_log` tables: read-only catalog audit plus advisory-locked, time-bounded maintenance that creates uncovered months and advances an empty catch-all (`partition/maintenance.rs`) |
 | `dm.rs` | DM channel management |
 | `reaction.rs` | Reaction storage and retrieval |
 | `thread.rs` | Thread/reply tracking |

crates/buzz-admin/src/main.rs

@@ -94,7 +94,7 @@ enum Command {
     /// Run the partition catalog audit using a read-only database session.
     PartitionAudit {
         /// Future months to include in the coverage check.
-        #[arg(long, default_value_t = 3)]
+        #[arg(long, default_value_t = buzz_db::partition::PARTITION_MANAGER_MONTHS_AHEAD)]
         months_ahead: u32,
     },
     /// Inspect deployment-wide Buzz product feedback.

crates/buzz-db/src/error.rs

@@ -45,6 +45,11 @@ pub enum DbError {
     #[error("invalid data: {0}")]
     InvalidData(String),
 
+    /// A partition the catalog audit had already listed was dropped before the
+    /// audit finished. A fresh snapshot sees a consistent catalog.
+    #[error("partition catalog changed mid-audit: {0}")]
+    PartitionDroppedMidAudit(String),
+
     /// A serving write admitted before the lifecycle transition is still live.
     /// This is an ordinary retryable drain condition, not a safety violation.
     #[error(

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

@@ -184,16 +184,18 @@ pub(crate) enum LockType {
     PushGate,
     Deletion,
     MigrationSchemaSafety,
+    PartitionMaintenance,
 }
 
 impl LockType {
     #[cfg(test)]
-    pub(crate) const ALL: [Self; 5] = [
+    pub(crate) const ALL: [Self; 6] = [
         Self::Replacement,
         Self::Membership,
         Self::PushGate,
         Self::Deletion,
         Self::MigrationSchemaSafety,
+        Self::PartitionMaintenance,
     ];
 
     pub(crate) const fn as_str(self) -> &'static str {
@@ -203,6 +205,7 @@ impl LockType {
             Self::PushGate => "push_gate",
             Self::Deletion => "deletion",
             Self::MigrationSchemaSafety => "migration_schema_safety",
+            Self::PartitionMaintenance => "partition_maintenance",
         }
     }
 }
@@ -671,6 +674,7 @@ mod tests {
                 "push_gate",
                 "deletion",
                 "migration_schema_safety",
+                "partition_maintenance",
             ]
         );
         assert_eq!(

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

@@ -24,7 +24,15 @@ impl Fixture {
             crate::migration::run_migrations(&pool).await.unwrap();
         }
         let db = Db::from_pool(pool);
-        db.ensure_future_partitions(1, true).await.unwrap();
+        db.maintain_partitions(
+            1,
+            crate::partition::PartitionMaintenancePolicy {
+                create_enabled: true,
+                advance_enabled: false,
+            },
+        )
+        .await
+        .unwrap();
         let community = CommunityId::from_uuid(Uuid::new_v4());
         sqlx::query("INSERT INTO communities(id,host) VALUES($1,$2)")
             .bind(community.as_uuid())

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

@@ -7,18 +7,24 @@ use std::time::{Duration, Instant};
 use buzz_datastore_tracing::datastore_span;
 use chrono::{DateTime, Datelike, NaiveDate, TimeZone, Timelike, Utc};
 use serde::Serialize;
-use sqlx::{Connection, PgConnection, PgPool, Row};
-use tracing::info;
+use sqlx::{PgConnection, PgPool, Row};
 
 use crate::error::{DbError, Result};
 use crate::Db;
 
+mod maintenance;
+
+pub use maintenance::{
+    maintain_partitions, PartitionMaintenanceOutcome, PartitionMaintenancePolicy,
+    PARTITION_MANAGER_MONTHS_AHEAD,
+};
+
 /// Tables that may be partition-managed. The allowlist prevents DDL injection.
 const PARTITIONED_TABLES: &[&str] = &["events", "delivery_log"];
 
 /// Maximum future-month horizon accepted by catalog audits and creation.
 ///
-/// The relay uses three months. Keeping the public API bounded prevents an
+/// The relay uses [`PARTITION_MANAGER_MONTHS_AHEAD`]. Keeping the public API bounded prevents an
 /// operator-supplied `buzz-admin partition-audit --months-ahead` value from
 /// allocating or iterating an effectively unbounded report.
 pub const MAX_PARTITION_MONTHS_AHEAD: u32 = 120;
@@ -29,6 +35,10 @@ const PARTITION_AUDIT_STATEMENT_TIMEOUT: &str = "5s";
 /// Maximum wall-clock duration for one complete multi-table catalog audit.
 const PARTITION_AUDIT_TOTAL_TIMEOUT: Duration = Duration::from_secs(10);
 
+/// Fresh-snapshot retries after a partition vanishes mid-audit, such as when
+/// another relay's maintenance replaces a catch-all.
+const PARTITION_AUDIT_CONCURRENT_DROP_RETRIES: usize = 2;
+
 fn expected_partition_key(table: &str) -> Option<&'static str> {
     match table {
         "events" => Some("RANGE (created_at)"),
@@ -451,131 +461,24 @@ async fn audit_partition_catalog_report_at_with_timeout(
     }
 }
 
-/// Audit first, then create only months proven to be uncovered.
-///
-/// Covered ranges are never probed with DDL. Creation failures are collected
-/// across all managed parents and months before an aggregate error is returned.
-pub async fn ensure_future_partitions(
-    pool: &PgPool,
-    months_ahead: u32,
-    create_enabled: bool,
-) -> Result<PartitionAudit> {
-    ensure_future_partitions_at(pool, months_ahead, create_enabled, Utc::now()).await
-}
-
+/// Create-only maintenance at a fixed clock, for tests of month creation.
+#[cfg(test)]
 async fn ensure_future_partitions_at(
     pool: &PgPool,
     months_ahead: u32,
     create_enabled: bool,
     now: DateTime<Utc>,
 ) -> Result<PartitionAudit> {
-    let audit = audit_partition_catalog_at(pool, months_ahead, now).await?;
-    let mut errors = Vec::new();
-    let mut created_any = false;
-
-    for table in &audit.tables {
-        for month in &table.months {
-            match month.kind {
-                MonthCoverageKind::CoveredByMonthly
-                | MonthCoverageKind::CoveredByCatchAll
-                | MonthCoverageKind::CoveredByDefault => {
-                    metrics::counter!(
-                        "buzz_partition_create_attempts_total",
-                        "table" => table.table,
-                        "outcome" => "skipped_covered"
-                    )
-                    .increment(1);
-                }
-                MonthCoverageKind::Uncovered if !create_enabled => {}
-                MonthCoverageKind::Uncovered => {
-                    let expected_name = partition_name(table.table, month.start);
-                    let collision = match relation_name_exists(pool, &expected_name).await {
-                        Ok(collision) => collision,
-                        Err(error) => {
-                            metrics::counter!(
-                                "buzz_partition_create_attempts_total",
-                                "table" => table.table,
-                                "outcome" => "error"
-                            )
-                            .increment(1);
-                            errors.push(format!(
-                                "{} {}: failed to check canonical name {expected_name}: {error}",
-                                table.table,
-                                month.start.format("%Y-%m")
-                            ));
-                            continue;
-                        }
-                    };
-                    if collision {
-                        metrics::counter!(
-                            "buzz_partition_create_attempts_total",
-                            "table" => table.table,
-                            "outcome" => "error"
-                        )
-                        .increment(1);
-                        errors.push(format!(
-                            "{} {}: canonical name {expected_name} already exists without the expected attachment and bounds",
-                            table.table,
-                            month.start.format("%Y-%m")
-                        ));
-                        continue;
-                    }
-                    if !table.structurally_safe_for_creation() {
-                        metrics::counter!(
-                            "buzz_partition_create_attempts_total",
-                            "table" => table.table,
-                            "outcome" => "error"
-                        )
-                        .increment(1);
-                        errors.push(format!(
-                            "{} {}: catalog is not structurally safe for automatic partition creation",
-                            table.table,
-                            month.start.format("%Y-%m")
-                        ));
-                        continue;
-                    }
-                    match create_month_partition(pool, table.table, month.start).await {
-                        Ok(name) => {
-                            created_any = true;
-                            metrics::counter!(
-                                "buzz_partition_create_attempts_total",
-                                "table" => table.table,
-                                "outcome" => "created"
-                            )
-                            .increment(1);
-                            info!(table = table.table, partition = name, "added partition");
-                        }
-                        Err(error) => {
-                            metrics::counter!(
-                                "buzz_partition_create_attempts_total",
-                                "table" => table.table,
-                                "outcome" => "error"
-                            )
-                            .increment(1);
-                            errors.push(format!(
-                                "{} {}: {error}",
-                                table.table,
-                                month.start.format("%Y-%m")
-                            ));
-                        }
-                    }
-                }
-            }
-        }
-    }
-
-    if errors.is_empty() {
-        if created_any {
-            audit_partition_catalog_at(pool, months_ahead, now).await
-        } else {
-            Ok(audit)
-        }
-    } else {
-        Err(DbError::InvalidData(format!(
-            "partition creation failed: {}",
-            errors.join("; ")
-        )))
-    }
+    maintenance::maintain_partitions_at(
+        pool,
+        months_ahead,
+        PartitionMaintenancePolicy {
+            create_enabled,
+            advance_enabled: false,
+        },
+        now,
+    )
+    .await
 }
 
 async fn audit_table(
@@ -585,6 +488,26 @@ async fn audit_table(
     now: DateTime<Utc>,
 ) -> Result<PartitionTableAudit> {
     let months_ahead = validated_months_ahead(months_ahead)?;
+    let mut retries = 0;
+    loop {
+        match audit_table_once(pool, table, months_ahead, now).await {
+            Err(error)
+                if retries < PARTITION_AUDIT_CONCURRENT_DROP_RETRIES
+                    && is_concurrent_drop(&error) =>
+            {
+                retries += 1;
+            }
+            result => return result,
+        }
+    }
+}
+
+async fn audit_table_once(
+    pool: &PgPool,
+    table: &'static str,
+    months_ahead: i32,
+    now: DateTime<Utc>,
+) -> Result<PartitionTableAudit> {
     let mut transaction = pool.begin().await?;
     sqlx::query("SET TRANSACTION READ ONLY")
         .execute(&mut *transaction)
@@ -595,6 +518,20 @@ async fn audit_table(
     Ok(audit)
 }
 
+/// Whether an audit failed only because a partition it had already listed was
+/// dropped by a concurrent transaction. A fresh snapshot sees a consistent
+/// catalog, so the audit can be retried.
+fn is_concurrent_drop(error: &DbError) -> bool {
+    match error {
+        DbError::PartitionDroppedMidAudit(_) => true,
+        // `relation_contains_rows` names a leaf that no longer exists.
+        DbError::Sqlx(sqlx::Error::Database(database)) => {
+            database.code().as_deref() == Some("42P01")
+        }
+        _ => false,
+    }
+}
+
 async fn configure_audit_transaction(connection: &mut PgConnection) -> Result<()> {
     pin_catalog_rendering(connection).await?;
     sqlx::query("SELECT pg_catalog.set_config('statement_timeout', $1, true)")
@@ -627,14 +564,14 @@ impl Db {
         audit_partition_catalog_report(&self.pool, months_ahead).await
     }
 
-    /// Ensures monthly partitions exist for the next N months when creation is enabled.
-    #[datastore_span(name = "ensure_future_partitions", system = "postgresql")]
-    pub async fn ensure_future_partitions(
+    /// Issues the bounded partition DDL `policy` permits, then returns a fresh audit.
+    #[datastore_span(name = "maintain_partitions", system = "postgresql")]
+    pub async fn maintain_partitions(
         &self,
         months_ahead: u32,
-        create_enabled: bool,
+        policy: PartitionMaintenancePolicy,
     ) -> Result<PartitionAudit> {
-        ensure_future_partitions(&self.pool, months_ahead, create_enabled).await
+        maintain_partitions(&self.pool, months_ahead, policy).await
     }
 }
 
@@ -758,14 +695,25 @@ async fn audit_table_on(
         let is_leaf: bool = row.try_get("is_leaf")?;
         let bound_partition_key: Option<String> = row.try_get("bound_partition_key")?;
         let detach_pending: bool = row.try_get("detach_pending")?;
-        let sibling_bounds: Vec<String> = row.try_get("sibling_bounds")?;
+        // Catalog rows come from the statement snapshot, but `pg_get_expr`
+        // renders from current catalog state and yields NULL for a relation
+        // dropped since the snapshot was taken.
+        let sibling_bounds = row
+            .try_get::<Vec<Option<String>>, _>("sibling_bounds")?
+            .into_iter()
+            .collect::<Option<Vec<String>>>()
+            .ok_or_else(|| {
+                DbError::PartitionDroppedMidAudit(format!("a sibling of partition {name}"))
+            })?;
         if detach_pending {
             pending_roots.insert(root_child_relation_oid);
         }
         let partition_key_compatible = partition_key_valid
             && bound_partition_key.as_deref() == Some(expected_partition_key)
             && !detach_pending;
-        let expression: String = row.try_get("bound")?;
+        let expression: String = row
+            .try_get::<Option<String>, _>("bound")?
+            .ok_or_else(|| DbError::PartitionDroppedMidAudit(format!("partition {name}")))?;
         let is_default = expression.trim() == "DEFAULT";
         let own_range = parse_range_bounds(&expression);
         let is_catch_all = own_range.as_ref().is_some_and(|(lower, upper)| {
@@ -1536,84 +1484,7 @@ fn parse_timestamp_literal(literal: &str) -> Option<DateTime<Utc>> {
     Some(Utc.from_utc_datetime(&date.and_hms_opt(0, 0, 0)?))
 }
 
-async fn create_month_partition(
-    pool: &PgPool,
-    table: &str,
-    start: DateTime<Utc>,
-) -> Result<String> {
-    if !PARTITIONED_TABLES.contains(&table) {
-        return Err(DbError::InvalidData(format!(
-            "table not in partition allowlist: {table:?}"
-        )));
-    }
-    let (end_year, end_month) = add_months(start.year(), start.month(), 1)?;
-    let end = month_start(end_year, end_month)?;
-    let partition_name = partition_name(table, start);
-    let start_date = start.format("%Y-%m-%d");
-    let end_date = end.format("%Y-%m-%d");
-    let sql = format!(
-        "CREATE TABLE {} PARTITION OF {} \
-         FOR VALUES FROM ('{start_date}') TO ('{end_date}')",
-        quote_identifier(&partition_name),
-        quote_identifier(table)
-    );
-    let mut connection = crate::observability::acquire_writer(
-        pool,
-        crate::observability::WriterOperation::Bootstrap,
-    )
-    .await?;
-    let mut transaction = connection.begin().await?;
-    pin_catalog_rendering(&mut transaction).await?;
-    sqlx::query(sqlx::AssertSqlSafe(sql))
-        .execute(&mut *transaction)
-        .await?;
-    let row = sqlx::query(
-        r#"
-        SELECT child.relkind::text AS relation_kind,
-               inherited.inhdetachpending AS detach_pending,
-               pg_catalog.pg_get_partkeydef(parent.oid) AS partition_key,
-               pg_catalog.pg_get_expr(child.relpartbound, child.oid) AS bound
-        FROM pg_catalog.pg_inherits inherited
-        JOIN pg_catalog.pg_class parent ON parent.oid = inherited.inhparent
-        JOIN pg_catalog.pg_namespace parent_ns ON parent_ns.oid = parent.relnamespace
-        JOIN pg_catalog.pg_class child ON child.oid = inherited.inhrelid
-        JOIN pg_catalog.pg_namespace child_ns ON child_ns.oid = child.relnamespace
-        WHERE parent_ns.nspname = current_schema()
-          AND child_ns.nspname = current_schema()
-          AND parent.relname = $1
-          AND child.relname = $2
-        "#,
-    )
-    .bind(table)
-    .bind(&partition_name)
-    .fetch_optional(&mut *transaction)
-    .await?;
-    let exact = row.is_some_and(|row| {
-        let relation_kind = row.try_get::<String, _>("relation_kind").ok();
-        let detach_pending = row.try_get::<bool, _>("detach_pending").ok();
-        let partition_key = row
-            .try_get::<Option<String>, _>("partition_key")
-            .ok()
-            .flatten();
-        let bounds = row
-            .try_get::<String, _>("bound")
-            .ok()
-            .and_then(|bound| parse_range_bounds(&bound));
-        relation_kind.as_deref() == Some("r")
-            && detach_pending == Some(false)
-            && partition_key.as_deref() == expected_partition_key(table)
-            && bounds == Some((PartitionBound::Finite(start), PartitionBound::Finite(end)))
-    });
-    if !exact {
-        transaction.rollback().await?;
-        return Err(DbError::InvalidData(format!(
-            "created partition {partition_name} failed exact catalog postcondition"
-        )));
-    }
-    transaction.commit().await?;
-    Ok(partition_name)
-}
-
+#[cfg(test)]
 async fn relation_name_exists(pool: &PgPool, relation: &str) -> Result<bool> {
     Ok(sqlx::query_scalar(
         r#"
@@ -1687,6 +1558,20 @@ mod tests {
             .all(|error| error.error.contains("months_ahead must be at most 120")));
     }
 
+    #[test]
+    fn only_a_mid_audit_drop_is_retried() {
+        assert!(is_concurrent_drop(&DbError::PartitionDroppedMidAudit(
+            "partition events_p2026_09".to_string()
+        )));
+        assert!(!is_concurrent_drop(&DbError::NotFound("row".to_string())));
+        assert!(!is_concurrent_drop(&DbError::InvalidData(
+            "bound".to_string()
+        )));
+        assert!(!is_concurrent_drop(&DbError::Sqlx(
+            sqlx::Error::PoolTimedOut
+        )));
+    }
+
     #[test]
     fn parses_pg16_range_bound_formats() {
         let expected = Utc.with_ymd_and_hms(2026, 7, 1, 0, 0, 0).unwrap();
@@ -1903,17 +1788,17 @@ mod tests {
         assert_eq!(extra, Some(1.0));
     }
 
-    mod postgres_tests {
+    pub(super) mod postgres_tests {
         use sqlx::postgres::PgPoolOptions;
         use uuid::Uuid;
 
         use super::*;
 
-        async fn scratch_pool() -> (PgPool, PgPool, String) {
+        pub(in crate::store::partition) async fn scratch_pool() -> (PgPool, PgPool, String) {
             scratch_pool_with_max_connections(4).await
         }
 
-        async fn scratch_pool_with_max_connections(
+        pub(in crate::store::partition) async fn scratch_pool_with_max_connections(
             max_connections: u32,
         ) -> (PgPool, PgPool, String) {
             let url = crate::test_support::database_url();
@@ -1941,7 +1826,7 @@ mod tests {
             (pool, admin, schema)
         }
 
-        async fn drop_schema(admin: &PgPool, schema: &str) {
+        pub(in crate::store::partition) async fn drop_schema(admin: &PgPool, schema: &str) {
             let _ = sqlx::query(sqlx::AssertSqlSafe(format!(
                 "DROP SCHEMA IF EXISTS {schema} CASCADE"
             )))
@@ -1949,7 +1834,7 @@ mod tests {
             .await;
         }
 
-        async fn seed_parents(pool: &PgPool) {
+        pub(in crate::store::partition) async fn seed_parents(pool: &PgPool) {
             seed_parents_with_keys(pool, "created_at", "delivered_at").await;
         }
 
@@ -1984,7 +1869,13 @@ mod tests {
             }
         }
 
-        async fn create_child(pool: &PgPool, table: &str, name: &str, lower: &str, upper: &str) {
+        pub(in crate::store::partition) async fn create_child(
+            pool: &PgPool,
+            table: &str,
+            name: &str,
+            lower: &str,
+            upper: &str,
+        ) {
             let sql = format!(
                 "CREATE TABLE {name} PARTITION OF {table} FOR VALUES FROM ({lower}) TO ({upper})"
             );
@@ -2032,7 +1923,9 @@ mod tests {
                 .expect("create nested child");
         }
 
-        async fn catalog_snapshot(pool: &PgPool) -> Vec<(String, String, String, i64)> {
+        pub(in crate::store::partition) async fn catalog_snapshot(
+            pool: &PgPool,
+        ) -> Vec<(String, String, String, i64)> {
             let mut transaction = pool.begin().await.expect("begin catalog snapshot");
             sqlx::query("SET TRANSACTION READ ONLY")
                 .execute(&mut *transaction)
@@ -2067,11 +1960,11 @@ mod tests {
             snapshot
         }
 
-        fn fixed_now() -> DateTime<Utc> {
+        pub(in crate::store::partition) fn fixed_now() -> DateTime<Utc> {
             Utc.with_ymd_and_hms(2026, 9, 15, 12, 0, 0).unwrap()
         }
 
-        async fn seed_fresh_layout(pool: &PgPool) {
+        pub(in crate::store::partition) async fn seed_fresh_layout(pool: &PgPool) {
             seed_parents(pool).await;
             for table in PARTITIONED_TABLES {
                 create_child(

crates/buzz-relay/src/api/artifact_postgres_tests.rs

@@ -25,7 +25,17 @@ impl Fixture {
             .unwrap();
         let host = format!("artifact-review-{}.local", Uuid::new_v4());
         state.db.ensure_configured_community(&host).await.unwrap();
-        state.db.ensure_future_partitions(1, true).await.unwrap();
+        state
+            .db
+            .maintain_partitions(
+                1,
+                buzz_db::partition::PartitionMaintenancePolicy {
+                    create_enabled: true,
+                    advance_enabled: false,
+                },
+            )
+            .await
+            .unwrap();
         let tenant = crate::tenant::bind_community(&state.db, &host)
             .await
             .unwrap();

crates/buzz-relay/src/config.rs

@@ -206,8 +206,13 @@ pub struct Config {
     pub metrics_port: u16,
     /// Interval between read-only partition catalog audits.
     pub partition_audit_interval: Duration,
-    /// Whether the partition manager may create uncovered monthly partitions.
+    /// Whether the partition manager may issue any partition DDL. Off also
+    /// disables catch-all advancement.
     pub partition_manager_create_enabled: bool,
+    /// Whether the partition manager may replace an empty right-edge catch-all
+    /// with dedicated monthlies. Also enables periodic (not only startup) DDL.
+    /// Requires `partition_manager_create_enabled`.
+    pub partition_manager_advance_enabled: bool,
 
     /// When true, NIP-42 pubkey-only authentication (no API token) is
     /// restricted to pubkeys in the `pubkey_allowlist` table. Users with valid
@@ -938,6 +943,9 @@ impl Config {
         );
         let partition_manager_create_enabled =
             parse_bool("BUZZ_PARTITION_MANAGER_CREATE_ENABLED", true)?;
+        // Off by default: each environment opts in after its canary.
+        let partition_manager_advance_enabled =
+            parse_bool("BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED", false)?;
 
         let s3_addressing_style = match std::env::var("BUZZ_S3_ADDRESSING_STYLE") {
             Ok(value) => value.parse().map_err(ConfigError::InvalidValue)?,
@@ -1392,6 +1400,7 @@ impl Config {
             metrics_port,
             partition_audit_interval,
             partition_manager_create_enabled,
+            partition_manager_advance_enabled,
             pubkey_allowlist_enabled,
             require_relay_membership,
             huddle_audio_available,
@@ -1565,6 +1574,7 @@ mod tests {
         assert_eq!(config.max_frame_bytes, DEFAULT_MAX_FRAME_BYTES);
         assert_eq!(config.partition_audit_interval, Duration::from_secs(900));
         assert!(config.partition_manager_create_enabled);
+        assert!(!config.partition_manager_advance_enabled);
         assert!(config.slow_client_grace_limit > 0);
         assert!(
             !config.pubkey_allowlist_enabled,
@@ -2069,6 +2079,33 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn partition_manager_advance_is_opt_in_and_parses_strictly() {
+        let _guards = env_guards();
+        let previous = std::env::var_os("BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED");
+
+        std::env::remove_var("BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED");
+        let default = Config::from_env().expect("default config");
+        std::env::set_var("BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED", " TRUE ");
+        let enabled = Config::from_env().expect("enabled config");
+        std::env::set_var("BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED", "enable-maybe");
+        let invalid = Config::from_env();
+
+        if let Some(value) = previous {
+            std::env::set_var("BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED", value);
+        } else {
+            std::env::remove_var("BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED");
+        }
+
+        assert!(!default.partition_manager_advance_enabled);
+        assert!(enabled.partition_manager_advance_enabled);
+        assert!(matches!(
+            invalid,
+            Err(ConfigError::InvalidValue(ref message))
+                if message.contains("BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED")
+        ));
+    }
+
     #[test]
     fn s3_addressing_style_env_accepts_virtual_and_rejects_invalid_values() {
         let _guards = env_guards();

crates/buzz-relay/src/main.rs

@@ -12,7 +12,7 @@ use uuid::Uuid;
 use buzz_audit::AuditService;
 use buzz_auth::AuthService;
 use buzz_core::CommunityId;
-use buzz_db::{Db, DbConfig};
+use buzz_db::{partition::PARTITION_MANAGER_MONTHS_AHEAD, Db, DbConfig};
 use buzz_pubsub::PubSubManager;
 use buzz_search::SearchService;
 
@@ -480,16 +480,20 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
         info!("Skipping database migrations because BUZZ_AUTO_MIGRATE is not enabled");
     }
 
+    let partition_policy = buzz_db::partition::PartitionMaintenancePolicy {
+        create_enabled: config.partition_manager_create_enabled,
+        advance_enabled: config.partition_manager_advance_enabled,
+    };
     let mut step = StepTimer::start(StartupStep::PartitionEnsure);
     let startup_partition_audit = match db
-        .ensure_future_partitions(3, config.partition_manager_create_enabled)
+        .maintain_partitions(PARTITION_MANAGER_MONTHS_AHEAD, partition_policy)
         .await
     {
         Ok(audit) => Some(audit),
         Err(error) => {
             step.degrade();
             error!(%error, "Failed to ensure partitions");
-            match db.audit_partitions(3).await {
+            match db.audit_partitions(PARTITION_MANAGER_MONTHS_AHEAD).await {
                 Ok(audit) => Some(audit),
                 Err(error) => {
                     error!(%error, "Initial partition catalog audit failed");
@@ -812,8 +816,9 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
         state.record_partition_audit(audit);
     }
 
-    // The periodic path is deliberately read-only. Partition creation only
-    // occurs during the bounded startup pass.
+    // The periodic path is read-only unless catch-all advancement is enabled.
+    // Maintenance takes no lock when the runway is already complete, so a
+    // periodic attempt usually costs one catalog audit.
     {
         let partition_state = Arc::clone(&state);
         let audit_interval = state.config.partition_audit_interval;
@@ -822,7 +827,21 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
                 PartitionAuditSchedule::new(audit_interval, has_startup_partition_audit);
             loop {
                 schedule.wait().await;
-                let result = partition_state.db.audit_partitions(3).await;
+                let db = &partition_state.db;
+                let result = if partition_policy.advance_enabled {
+                    match db
+                        .maintain_partitions(PARTITION_MANAGER_MONTHS_AHEAD, partition_policy)
+                        .await
+                    {
+                        Ok(audit) => Ok(audit),
+                        Err(error) => {
+                            warn!(%error, "Periodic partition maintenance failed");
+                            db.audit_partitions(PARTITION_MANAGER_MONTHS_AHEAD).await
+                        }
+                    }
+                } else {
+                    db.audit_partitions(PARTITION_MANAGER_MONTHS_AHEAD).await
+                };
                 schedule.record_attempt(result.is_ok());
                 match result {
                     Ok(audit) => partition_state.record_partition_audit(audit),