fix(db): preserve partition reader grants (#8485)

## Summary

PostgreSQL does not copy parent table grants to new partitions. A reader
can query the `events` parent but fail direct partition reads after the
relay creates another month or replaces its catch-all.

Copy each managed parent's explicit table-level `SELECT` grants to its
new children inside the existing maintenance transaction. The parent ACL
supplies the policy; there is no new setting or deployment-specific role
name. Writes, column-only grants, owner entries, and grant options are
excluded. A grant failure rolls back the table's partition changes.

Existing children and manually created or attached partitions still need
grants from database provisioning. Apply the [companion initializer
fix](https://github.com/squareup/bb-utility/pull/2253) for existing
partitions before relying on automatic grants for future ones.

### Related issue

No duplicate found. The closest related PR is #8464, which enables
catch-all advancement by default; this change independently preserves
reader access during that maintenance.

### Testing

No manual testing. Production rollout and the Fivetran setup check
remain pending.

Generated with Codex

Signed-off-by: coder2 <4acc49cb93576448ba82e631688c51a40bf03d6a889d78fddf57917d198de10f@buzz.block.builderlab.xyz>
Co-authored-by: coder2 <4acc49cb93576448ba82e631688c51a40bf03d6a889d78fddf57917d198de10f@buzz.block.builderlab.xyz>

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

@@ -11,7 +11,8 @@
 //!    lock, then the catch-all, in the order writers take them, waiting only on
 //!    the parent and failing fast on any other contention;
 //! 3. re-audits under those locks and refuses if the plan changed; and
-//! 4. re-audits the changed catalog and verifies index and constraint
+//! 4. copies the parent's explicit table-level SELECT grants to new children;
+//! 5. re-audits the changed catalog and verifies index and constraint
 //!    inheritance before commit.
 //!
 //! A populated catch-all is always refused before any lock is taken: moving
@@ -732,10 +733,12 @@ async fn apply_change(
         .await?;
     }
 
+    let new_children = expected.target_names(table);
+    grant_parent_readers(&mut transaction, parent.oid, &parent.schema, &new_children).await?;
+
     let after = audit_table_on(&mut transaction, table, months_ahead, now).await?;
-    let verified = verify_applied(&locked, &after, &expected).and(
-        verify_inheritance(&mut transaction, parent.oid, &expected.target_names(table)).await?,
-    );
+    let verified = verify_applied(&locked, &after, &expected)
+        .and(verify_inheritance(&mut transaction, parent.oid, &new_children).await?);
     if let Err(reason) = verified {
         transaction.rollback().await?;
         return Err(DbError::InvalidData(format!(
@@ -746,6 +749,37 @@ async fn apply_change(
     Ok(LockedAttempt::Committed(expected))
 }
 
+/// The parent ACL is the reader policy; never copy writes or grant options.
+/// `quote_ident` handles role names containing quotes.
+/// PUBLIC is an ACL grantee too, but must remain an unquoted SQL keyword.
+async fn grant_parent_readers(
+    connection: &mut PgConnection,
+    parent_oid: i64,
+    schema: &str,
+    children: &[String],
+) -> Result<()> {
+    let grants: Vec<String> = sqlx::query_scalar(
+        "SELECT DISTINCT pg_catalog.format(\
+             'GRANT SELECT ON TABLE %I.%I TO %s', $2::text, child, \
+             CASE WHEN acl.grantee = 0 THEN 'PUBLIC' \
+                  ELSE pg_catalog.quote_ident(pg_catalog.pg_get_userbyid(acl.grantee)) END) \
+         FROM pg_catalog.pg_class parent \
+         CROSS JOIN LATERAL pg_catalog.aclexplode(parent.relacl) acl \
+         CROSS JOIN pg_catalog.unnest($3::text[]) child \
+         WHERE parent.oid = $1::bigint::oid \
+           AND acl.privilege_type = 'SELECT' AND acl.grantee <> parent.relowner",
+    )
+    .bind(parent_oid)
+    .bind(schema)
+    .bind(children)
+    .fetch_all(&mut *connection)
+    .await?;
+    for grant in grants {
+        execute_ddl(connection, grant).await?;
+    }
+    Ok(())
+}
+
 async fn configure_maintenance_transaction(connection: &mut PgConnection) -> Result<()> {
     pin_catalog_rendering(connection).await?;
     for (setting, value) in [

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

@@ -585,6 +585,184 @@ mod postgres_tests {
             .any(|child| child.name == format!("{}_p2026_09", table.table))));
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn cluster_global_parent_readers_follow_monthly_creation_and_catch_all_advancement() {
+        use crate::store::partition::quote_identifier;
+
+        let (pool, scratch) = scratch_pool().await;
+        seed_parents(&pool).await;
+        let reader = format!("{}_r\"x", scratch.name);
+        let writer = format!("{}_writer", scratch.name);
+        for role in [&reader, &writer] {
+            execute_ddl(
+                &mut pool.acquire().await.unwrap(),
+                format!("CREATE ROLE {}", quote_identifier(role)),
+            )
+            .await
+            .unwrap();
+        }
+        for statement in [
+            format!(
+                "GRANT USAGE ON SCHEMA {} TO {}, {}",
+                scratch.name,
+                quote_identifier(&reader),
+                quote_identifier(&writer)
+            ),
+            format!(
+                "GRANT SELECT ON events TO {} WITH GRANT OPTION",
+                quote_identifier(&reader)
+            ),
+            format!(
+                "GRANT INSERT, UPDATE ON events TO {}",
+                quote_identifier(&reader)
+            ),
+            format!(
+                "GRANT INSERT, SELECT (id) ON events TO {}",
+                quote_identifier(&writer)
+            ),
+            "GRANT SELECT ON delivery_log TO PUBLIC".to_string(),
+            "CREATE TABLE unrelated (id BIGINT)".to_string(),
+        ] {
+            execute_ddl(&mut pool.acquire().await.unwrap(), statement)
+                .await
+                .unwrap();
+        }
+        for table in PARTITIONED_TABLES {
+            create_child(
+                &pool,
+                table,
+                &format!("{table}_p_past"),
+                "MINVALUE",
+                "'2026-09-01'",
+            )
+            .await;
+        }
+
+        maintain_partitions_at(&pool, 0, CREATE_ONLY, fixed_now())
+            .await
+            .expect("create uncovered September");
+        let can = |role: String, relation: String, privilege: &'static str| {
+            let pool = pool.clone();
+            async move {
+                sqlx::query_scalar::<_, bool>("SELECT has_table_privilege($1, $2, $3)")
+                    .bind(role)
+                    .bind(relation)
+                    .bind(privilege)
+                    .fetch_one(&pool)
+                    .await
+                    .unwrap()
+            }
+        };
+        let mut direct_read = pool.begin().await.unwrap();
+        execute_ddl(
+            &mut direct_read,
+            format!("SET LOCAL ROLE {}", quote_identifier(&reader)),
+        )
+        .await
+        .unwrap();
+        sqlx::query("SELECT * FROM events_p2026_09 LIMIT 0")
+            .execute(&mut *direct_read)
+            .await
+            .expect("parent reader can query the new child directly");
+        direct_read.rollback().await.unwrap();
+        assert!(can(reader.clone(), "events_p2026_09".into(), "SELECT").await);
+        for privilege in ["INSERT", "UPDATE", "SELECT WITH GRANT OPTION"] {
+            assert!(!can(reader.clone(), "events_p2026_09".into(), privilege).await);
+        }
+        for relation in [
+            "events_p2026_09",
+            "events_p_past",
+            "delivery_log_p_past",
+            "unrelated",
+        ] {
+            assert!(!can(writer.clone(), relation.into(), "SELECT").await);
+        }
+        assert!(!can(reader.clone(), "events_p_past".into(), "SELECT").await);
+        assert!(!can(reader.clone(), "unrelated".into(), "SELECT").await);
+        assert!(can(writer.clone(), "delivery_log_p2026_09".into(), "SELECT").await);
+
+        // The next rotation uses the current parent ACL, not a cached reader list.
+        for statement in [
+            format!("REVOKE SELECT ON events FROM {}", quote_identifier(&reader)),
+            format!("GRANT SELECT ON events TO {}", quote_identifier(&writer)),
+        ] {
+            execute_ddl(&mut pool.acquire().await.unwrap(), statement)
+                .await
+                .unwrap();
+        }
+        for table in PARTITIONED_TABLES {
+            create_child(
+                &pool,
+                table,
+                &format!("{table}_p_future"),
+                "'2026-10-01'",
+                "MAXVALUE",
+            )
+            .await;
+        }
+        maintain_partitions_at(&pool, 1, ADVANCE, fixed_now())
+            .await
+            .expect("replace catch-all and grant new readers");
+        for relation in ["events_p2026_10", "events_p_future"] {
+            assert!(can(writer.clone(), relation.into(), "SELECT").await);
+            assert!(!can(writer.clone(), relation.into(), "INSERT").await);
+            assert!(!can(reader.clone(), relation.into(), "SELECT").await);
+        }
+        for relation in ["delivery_log_p2026_10", "delivery_log_p_future"] {
+            assert!(can(reader.clone(), relation.into(), "SELECT").await);
+            assert!(!can(reader.clone(), relation.into(), "INSERT").await);
+        }
+        let before = catalog_snapshot(&pool).await;
+        maintain_partitions_at(&pool, 1, ADVANCE, fixed_now())
+            .await
+            .unwrap();
+        assert_eq!(catalog_snapshot(&pool).await, before);
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn cluster_global_reader_grant_failure_rolls_back_partition_replacement() {
+        let (pool, scratch) = scratch_pool().await;
+        seed_repaired_layout(&pool, false).await;
+        sqlx::query("GRANT SELECT ON events TO PUBLIC")
+            .execute(&pool)
+            .await
+            .unwrap();
+        let before = catalog_snapshot(&pool).await;
+        sqlx::query(
+            "CREATE FUNCTION reject_partition_grant() RETURNS event_trigger \
+             LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'injected grant failure'; END $$",
+        )
+        .execute(&pool)
+        .await
+        .unwrap();
+        execute_ddl(
+            &mut pool.acquire().await.unwrap(),
+            format!(
+                "CREATE EVENT TRIGGER {}_grant_failure ON ddl_command_start \
+                 WHEN TAG IN ('GRANT') EXECUTE FUNCTION reject_partition_grant()",
+                scratch.name
+            ),
+        )
+        .await
+        .unwrap();
+
+        let result = maintain(&pool).await;
+        assert!(
+            matches!(result, Err(DbError::InvalidData(ref message))
+                if message.contains("events error") && message.contains("injected grant failure")),
+            "{result:?}"
+        );
+        let events = |snapshot: Vec<(String, String, String, i64)>| {
+            snapshot
+                .into_iter()
+                .filter(|(name, ..)| name.starts_with("events_"))
+                .collect::<Vec<_>>()
+        };
+        assert_eq!(events(catalog_snapshot(&pool).await), events(before));
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn populated_catch_all_is_refused_before_any_lock() {

docs/partition-catalog-monitoring.md

@@ -30,6 +30,23 @@ later. Once its upper bound is at or before the start of the current month, it i
 historical. It remains in the per-child report as `legacy_leaf` but no longer
 degrades the table.
 
+## Reader grants on new partitions
+
+When maintenance creates a monthly partition or replaces the empty catch-all,
+it copies explicit table-level `SELECT` grants from that parent to the new
+children before committing. Group roles and explicit `PUBLIC` grants are
+included; the owner's ACL entry, column-only grants, writes, and grant options
+are not copied. No additional relay setting is needed. A grant failure rolls
+back that table's entire maintenance transaction.
+
+Existing children and manually created or attached partitions still need their
+own grants. Maintenance does not reconcile grants on existing tables or revoke
+old grants when the parent ACL changes. Replacing the catch-all drops and
+recreates it, so grants added directly to the old catch-all are lost; only the
+parent's `SELECT` grants carry over. Keep those permissions aligned through
+the database's provisioning process. Initial schema bootstrap runs before
+deployment-specific reader grants are provisioned.
+
 ## Metrics
 
 Per-table gauges are set only by a successful audit of that table: