diff --git a/Cargo.lock b/Cargo.lock index 94a5ebd1396..ca351aa2b5f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1399,6 +1399,7 @@ dependencies = [ "nostr 0.44.7", "openssl", "p256", + "percent-encoding", "proptest", "rand 0.10.1", "reqwest 0.13.4", diff --git a/crates/buzz-push-gateway/Cargo.toml b/crates/buzz-push-gateway/Cargo.toml index 29dc8d21372..605cdc42521 100644 --- a/crates/buzz-push-gateway/Cargo.toml +++ b/crates/buzz-push-gateway/Cargo.toml @@ -54,5 +54,6 @@ uuid = { workspace = true } [dev-dependencies] openssl = { version = "0.10", features = ["vendored"] } ciborium = "0.2" +percent-encoding = "2.3" proptest = { workspace = true } reqwest = { workspace = true } diff --git a/crates/buzz-push-gateway/schema/schema.sql b/crates/buzz-push-gateway/schema/schema.sql new file mode 100644 index 00000000000..98a698022fa --- /dev/null +++ b/crates/buzz-push-gateway/schema/schema.sql @@ -0,0 +1,37 @@ +-- Push gateway desired-state database manifest. +-- +-- The gateway's authority store is deployment-global and lives in its own +-- dedicated database, outside relay community tenancy. This directory declares +-- its end state; `../migrations/` remains the incremental path the gateway +-- binary applies at startup. +-- `src/postgres/desired_schema_postgres_tests.rs` proves that, applied through +-- `pgschema`, the two produce the same `public` tables, columns, table options, +-- constraints, and indexes. That is not full SchemaBot support: the inline FK +-- in `push_gateway_delegations.sql` must be revisited before SchemaBot adopts +-- this schema. +-- +-- `tables/public/.sql` declares one table per file: its CREATE TABLE +-- and the CREATE INDEX statements on it, the one-table-per-file shape SchemaBot +-- reads. `tables/` is the SchemaBot schema directory and `public` is its +-- namespace, so nothing else may live under it. The gateway has no types, +-- functions, triggers, or seed rows. Its pre-migration bootstrap guard and +-- runtime-role grants are operational steps in `src/postgres.rs`, not schema. +-- It also has nothing pgschema cannot express (no seed DML, no storage +-- parameters), so the relay's `scripts/reconcile-schema-after-pgschema.sql` +-- step (root `AGENTS.md` gotcha #7) does not apply to it. +-- +-- `./bin/pgschema apply --file crates/buzz-push-gateway/schema/schema.sql` +-- resolves the `\i` includes relative to this file. Raw `psql` resolves `\i` +-- against the working directory instead, so run it from here: +-- `cd crates/buzz-push-gateway/schema && psql -f schema.sql`. +-- +-- Adding a table: add `tables/public/
.sql` and include it below after +-- the tables it references. This file holds only `\i` lines. The layout +-- test in `src/postgres/desired_schema_postgres_tests.rs` enforces both. + +\i tables/public/push_gateway_challenges.sql +\i tables/public/push_gateway_installations.sql +\i tables/public/push_gateway_delegations.sql +\i tables/public/push_gateway_endpoint_quotas.sql +\i tables/public/push_gateway_delivery_auth_replays.sql +\i tables/public/push_gateway_delivery_request_replays.sql diff --git a/crates/buzz-push-gateway/schema/tables/public/push_gateway_challenges.sql b/crates/buzz-push-gateway/schema/tables/public/push_gateway_challenges.sql new file mode 100644 index 00000000000..0ada41a5343 --- /dev/null +++ b/crates/buzz-push-gateway/schema/tables/public/push_gateway_challenges.sql @@ -0,0 +1,13 @@ +-- Single-use App Attest enrollment challenges. Only the hash is stored. +CREATE TABLE push_gateway_challenges ( + id UUID PRIMARY KEY, + challenge_hash BYTEA NOT NULL CHECK (length(challenge_hash) = 32), + expires_at TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX push_gateway_challenges_expiry ON push_gateway_challenges (expires_at); + +-- The unauthenticated challenge route applies a deployment-global rolling +-- issuance quota. Keep its count query bounded as challenge volume grows. +CREATE INDEX push_gateway_challenges_created_at ON push_gateway_challenges (created_at); diff --git a/crates/buzz-push-gateway/schema/tables/public/push_gateway_delegations.sql b/crates/buzz-push-gateway/schema/tables/public/push_gateway_delegations.sql new file mode 100644 index 00000000000..0fcfbc0f163 --- /dev/null +++ b/crates/buzz-push-gateway/schema/tables/public/push_gateway_delegations.sql @@ -0,0 +1,17 @@ +-- An installation's delegation to one relay signing key. Installations may +-- authorize multiple relay deployments, one row per relay key. +CREATE TABLE push_gateway_delegations ( + id UUID PRIMARY KEY, + installation_id UUID NOT NULL REFERENCES push_gateway_installations(id), + relay_pubkey BYTEA NOT NULL CHECK (length(relay_pubkey) = 32), + endpoint_epoch BIGINT NOT NULL CHECK (endpoint_epoch > 0), + generation BIGINT NOT NULL CHECK (generation > 0), + not_before TIMESTAMPTZ NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + revoked_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (installation_id, relay_pubkey), + CHECK (not_before < expires_at) +); + +CREATE INDEX push_gateway_delegations_expiry ON push_gateway_delegations (expires_at) WHERE revoked_at IS NULL; diff --git a/crates/buzz-push-gateway/schema/tables/public/push_gateway_delivery_auth_replays.sql b/crates/buzz-push-gateway/schema/tables/public/push_gateway_delivery_auth_replays.sql new file mode 100644 index 00000000000..0a16dbf7214 --- /dev/null +++ b/crates/buzz-push-gateway/schema/tables/public/push_gateway_delivery_auth_replays.sql @@ -0,0 +1,9 @@ +-- Replay ledger for relay delivery auth events, retained until expiry. +CREATE TABLE push_gateway_delivery_auth_replays ( + relay_pubkey BYTEA NOT NULL CHECK (length(relay_pubkey) = 32), + auth_event_id BYTEA NOT NULL CHECK (length(auth_event_id) = 32), + expires_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (relay_pubkey, auth_event_id) +); + +CREATE INDEX push_gateway_delivery_auth_replays_expiry ON push_gateway_delivery_auth_replays (expires_at); diff --git a/crates/buzz-push-gateway/schema/tables/public/push_gateway_delivery_request_replays.sql b/crates/buzz-push-gateway/schema/tables/public/push_gateway_delivery_request_replays.sql new file mode 100644 index 00000000000..101b98a3721 --- /dev/null +++ b/crates/buzz-push-gateway/schema/tables/public/push_gateway_delivery_request_replays.sql @@ -0,0 +1,9 @@ +-- Replay ledger for relay delivery request ids, retained until expiry. +CREATE TABLE push_gateway_delivery_request_replays ( + relay_pubkey BYTEA NOT NULL CHECK (length(relay_pubkey) = 32), + request_id UUID NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (relay_pubkey, request_id) +); + +CREATE INDEX push_gateway_delivery_request_replays_expiry ON push_gateway_delivery_request_replays (expires_at); diff --git a/crates/buzz-push-gateway/schema/tables/public/push_gateway_endpoint_quotas.sql b/crates/buzz-push-gateway/schema/tables/public/push_gateway_endpoint_quotas.sql new file mode 100644 index 00000000000..f1b55551d05 --- /dev/null +++ b/crates/buzz-push-gateway/schema/tables/public/push_gateway_endpoint_quotas.sql @@ -0,0 +1,9 @@ +-- Per-APNs-token delivery admission window. +CREATE TABLE push_gateway_endpoint_quotas ( + token_fingerprint BYTEA PRIMARY KEY CHECK (length(token_fingerprint) = 32), + window_started_at TIMESTAMPTZ NOT NULL, + admitted BIGINT NOT NULL CHECK (admitted >= 0), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX push_gateway_endpoint_quotas_updated ON push_gateway_endpoint_quotas (updated_at); diff --git a/crates/buzz-push-gateway/schema/tables/public/push_gateway_installations.sql b/crates/buzz-push-gateway/schema/tables/public/push_gateway_installations.sql new file mode 100644 index 00000000000..a40c532f542 --- /dev/null +++ b/crates/buzz-push-gateway/schema/tables/public/push_gateway_installations.sql @@ -0,0 +1,23 @@ +-- App Attest-bound device installations and their encrypted APNs tokens. +-- Revoked installations remain as retry tombstones until authority expiry, so +-- ownership uniqueness covers live rows only: a replacement can enroll without +-- deleting the old installation's idempotency state. +CREATE TABLE push_gateway_installations ( + id UUID PRIMARY KEY, + app_attest_key_id BYTEA NOT NULL CHECK (octet_length(app_attest_key_id) BETWEEN 1 AND 128), + app_attest_public_key BYTEA NOT NULL CHECK (octet_length(app_attest_public_key) BETWEEN 33 AND 256), + assertion_counter BIGINT NOT NULL CHECK (assertion_counter BETWEEN 0 AND 4294967295), + token_ciphertext BYTEA NOT NULL CHECK (octet_length(token_ciphertext) BETWEEN 1 AND 2048), + token_fingerprint BYTEA NOT NULL CHECK (length(token_fingerprint) = 32), + endpoint_epoch BIGINT NOT NULL CHECK (endpoint_epoch > 0), + expires_at TIMESTAMPTZ NOT NULL, + revoked_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX push_gateway_installations_expiry ON push_gateway_installations (expires_at) WHERE revoked_at IS NULL; + +CREATE UNIQUE INDEX push_gateway_installations_active_app_attest_key ON push_gateway_installations (app_attest_key_id) WHERE revoked_at IS NULL; + +CREATE UNIQUE INDEX push_gateway_installations_active_token ON push_gateway_installations (token_fingerprint) WHERE revoked_at IS NULL; diff --git a/crates/buzz-push-gateway/src/postgres.rs b/crates/buzz-push-gateway/src/postgres.rs index a58d8c209b4..887dd77d072 100644 --- a/crates/buzz-push-gateway/src/postgres.rs +++ b/crates/buzz-push-gateway/src/postgres.rs @@ -10,8 +10,23 @@ use uuid::Uuid; #[path = "postgres/bootstrap_tests.rs"] mod bootstrap_postgres_tests; +#[cfg(test)] +#[path = "postgres/desired_schema_postgres_tests.rs"] +mod desired_schema_postgres_tests; + static GATEWAY_MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations"); +/// Every gateway authority table. Readiness probes and runtime-role grants +/// cover exactly this set; the desired-state parity test pins it to `schema/`. +const GATEWAY_TABLES: [&str; 6] = [ + "push_gateway_challenges", + "push_gateway_installations", + "push_gateway_delegations", + "push_gateway_endpoint_quotas", + "push_gateway_delivery_auth_replays", + "push_gateway_delivery_request_replays", +]; + #[cfg(test)] async fn migrate_gateway_through( pool: &PgPool, @@ -58,14 +73,8 @@ impl PostgresAuthorityStore { REVOKE CREATE ON SCHEMA public FROM {role}; GRANT CONNECT ON DATABASE {database} TO {role}; GRANT USAGE ON SCHEMA public TO {role}; - GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE - push_gateway_challenges, - push_gateway_installations, - push_gateway_delegations, - push_gateway_endpoint_quotas, - push_gateway_delivery_auth_replays, - push_gateway_delivery_request_replays - TO {role};" + GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE {tables} TO {role};", + tables = GATEWAY_TABLES.join(", "), ); sqlx::raw_sql(AssertSqlSafe(grants)).execute(pool).await?; Ok(()) @@ -87,16 +96,8 @@ fn bytes32(v: Vec) -> Result<[u8; 32], AuthorityError> { #[async_trait] impl AuthorityStore for PostgresAuthorityStore { async fn ready(&self) -> Result<(), AuthorityError> { - const TABLES: [&str; 6] = [ - "push_gateway_challenges", - "push_gateway_installations", - "push_gateway_delegations", - "push_gateway_endpoint_quotas", - "push_gateway_delivery_auth_replays", - "push_gateway_delivery_request_replays", - ]; let mut tx = self.pool.begin().await.map_err(db)?; - for table in TABLES { + for table in GATEWAY_TABLES { let ready: bool = sqlx::query_scalar( "SELECT to_regclass($1) IS NOT NULL AND COALESCE(has_table_privilege(current_user, to_regclass($1), 'SELECT'), false) diff --git a/crates/buzz-push-gateway/src/postgres/desired_schema_postgres_tests.rs b/crates/buzz-push-gateway/src/postgres/desired_schema_postgres_tests.rs new file mode 100644 index 00000000000..c01d322ce56 --- /dev/null +++ b/crates/buzz-push-gateway/src/postgres/desired_schema_postgres_tests.rs @@ -0,0 +1,373 @@ +//! `schema/` is the gateway's desired-state schema in the one-table-per-file +//! shape SchemaBot reads; `migrations/` is the incremental path the binary +//! applies at startup. These tests keep the two from drifting apart. +use super::*; +use sqlx::postgres::PgConnectOptions; +use sqlx::ConnectOptions; +use std::collections::BTreeSet; +use std::path::{Path, PathBuf}; + +fn schema_dir() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")).join("schema") +} + +fn strip_comments(text: &str) -> String { + text.lines() + .map(|line| line.split_once("--").map_or(line, |(code, _)| code)) + .collect::>() + .join("\n") +} + +/// `schema/schema.sql` includes every file under `schema/` exactly once and +/// holds nothing but `\i` lines. `schema/tables/public/
.sql` declares +/// exactly that table plus plain `CREATE [UNIQUE] INDEX ... ON
` +/// statements (SchemaBot refuses `IF NOT EXISTS` and `CONCURRENTLY`). +#[test] +fn desired_state_schema_layout_is_complete_and_schemabot_shaped() { + fn sql_files(dir: &Path, root: &Path, out: &mut BTreeSet) { + for entry in std::fs::read_dir(dir).expect("read schema directory") { + let path = entry.expect("schema directory entry").path(); + if path.is_dir() { + sql_files(&path, root, out); + } else { + let relative = path.strip_prefix(root).expect("schema-relative path"); + out.insert(relative.to_str().expect("utf-8 path").replace('\\', "/")); + } + } + } + + let dir = schema_dir(); + let manifest = std::fs::read_to_string(dir.join("schema.sql")).expect("read schema.sql"); + let mut included = Vec::new(); + for line in strip_comments(&manifest).lines().map(str::trim) { + if line.is_empty() { + continue; + } + let include = line.strip_prefix("\\i ").unwrap_or_else(|| { + panic!( + "schema/schema.sql may only hold \\i includes; \ + put table DDL in tables/public/
.sql: {line:.80}" + ) + }); + included.push(include.trim().to_owned()); + } + let included_set: BTreeSet<_> = included.iter().cloned().collect(); + assert_eq!( + included.len(), + included_set.len(), + "schema.sql includes a file twice" + ); + + let mut on_disk = BTreeSet::new(); + sql_files(&dir, &dir, &mut on_disk); + on_disk.remove("schema.sql"); + assert_eq!( + on_disk, included_set, + "every schema/ file must be included by schema/schema.sql exactly once" + ); + + for file in &included { + let table = file + .strip_prefix("tables/public/") + .and_then(|name| name.strip_suffix(".sql")) + .unwrap_or_else(|| { + panic!("{file}: the gateway schema holds only tables/public/
.sql") + }); + let text = std::fs::read_to_string(dir.join(file)).expect("read table file"); + let mut creates = 0; + for statement in strip_comments(&text) + .split(';') + .map(|s| s.split_whitespace().collect::>().join(" ")) + .filter(|s| !s.is_empty()) + { + if let Some(rest) = statement.strip_prefix("CREATE TABLE ") { + assert!( + rest.starts_with(&format!("{table} (")), + "{file} may only create table {table}: {statement:.80}" + ); + creates += 1; + continue; + } + let index = statement + .strip_prefix("CREATE INDEX ") + .or_else(|| statement.strip_prefix("CREATE UNIQUE INDEX ")) + .unwrap_or_else(|| { + panic!("{file} may only hold CREATE TABLE/INDEX: {statement:.80}") + }); + assert!( + !index.starts_with("IF NOT EXISTS ") && !index.starts_with("CONCURRENTLY "), + "{file} must use a plain CREATE INDEX: {statement:.80}" + ); + assert!( + index.contains(&format!(" ON {table} ")), + "{file} may only index {table}: {statement:.80}" + ); + } + assert_eq!(creates, 1, "{file} must create exactly one table"); + } + assert!( + !included.is_empty(), + "schema/tables/public declares no tables" + ); +} + +/// One probe database is bootstrapped from `schema/schema.sql` through the +/// real `bin/pgschema apply` binary (the SchemaBot-equivalent desired-state +/// path); another runs every gateway migration. Their `public` catalogs must +/// match: the same tables, column shapes (full type with typmod, default, +/// nullability, identity, generation, collation; keyed by name, since +/// migrations drop and append columns), table options, constraints, and index +/// shapes including per-key `indoption`. Neither side may hold functions, +/// triggers, policies, views, sequences, or user types. So an unmirrored +/// migration that changes `public` tables, columns, table options, +/// constraints, or indexes fails here. Data-only migrations and objects +/// outside `public` (other schemas, roles, database settings) are not +/// compared. This proves equivalence under `pgschema`, not full SchemaBot +/// support: the inline FK in `push_gateway_delegations.sql` must be revisited +/// before SchemaBot adopts this schema. +#[tokio::test] +#[ignore = "requires PostgreSQL"] +async fn desired_state_schema_matches_migrations() { + type Rows = Vec>>; + async fn catalog(pool: &PgPool, sql: &str) -> Rows { + sqlx::query(AssertSqlSafe(sql.to_owned())) + .fetch_all(pool) + .await + .expect("read catalog") + .iter() + .map(|row| { + (0..row.len()) + .map(|i| row.get::, _>(i)) + .collect() + }) + .collect() + } + const TABLES: &str = "SELECT tablename::text FROM pg_tables \ + WHERE schemaname = 'public' AND tablename <> '_sqlx_migrations' ORDER BY 1"; + // `pg_attribute` rather than `information_schema.columns`: the latter's + // `data_type` drops typmods and array element types (`varchar(64)` and + // `varchar(128)` both read `character varying`). + const COLUMNS: &str = "SELECT t.relname::text, a.attname::text, \ + format_type(a.atttypid, a.atttypmod), a.attnotnull::text, \ + pg_get_expr(d.adbin, d.adrelid), a.attidentity::text, a.attgenerated::text, \ + a.attcollation::regcollation::text \ + FROM pg_attribute a JOIN pg_class t ON t.oid = a.attrelid \ + JOIN pg_namespace n ON n.oid = t.relnamespace \ + LEFT JOIN pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum \ + WHERE n.nspname = 'public' AND t.relkind = 'r' AND t.relname <> '_sqlx_migrations' \ + AND a.attnum > 0 AND NOT a.attisdropped ORDER BY 1, 2"; + const TABLE_OPTIONS: &str = "SELECT t.relname::text, t.relpersistence::text, \ + t.relrowsecurity::text, t.relforcerowsecurity::text, t.reloptions::text \ + FROM pg_class t JOIN pg_namespace n ON n.oid = t.relnamespace \ + WHERE n.nspname = 'public' AND t.relkind = 'r' AND t.relname <> '_sqlx_migrations' \ + ORDER BY 1"; + const CONSTRAINTS: &str = "SELECT t.relname::text, c.conname::text, c.contype::text, \ + pg_get_constraintdef(c.oid) \ + FROM pg_constraint c JOIN pg_class t ON t.oid = c.conrelid \ + JOIN pg_namespace n ON n.oid = t.relnamespace \ + WHERE n.nspname = 'public' AND t.relname <> '_sqlx_migrations' ORDER BY 1, 2"; + const INDEXES: &str = "SELECT t.relname::text, c.relname::text, \ + pg_get_indexdef(i.indexrelid), i.indoption::int2[]::text \ + FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid \ + JOIN pg_class t ON t.oid = i.indrelid \ + JOIN pg_namespace n ON n.oid = t.relnamespace \ + WHERE n.nspname = 'public' AND t.relname <> '_sqlx_migrations' ORDER BY 1, 2"; + // `schema/` may only declare tables and indexes, so neither path may + // create anything else in `public`: SchemaBot would never reproduce it. + const NON_TABLE_OBJECTS: &str = "SELECT 'function', p.proname::text FROM pg_proc p \ + JOIN pg_namespace n ON n.oid = p.pronamespace WHERE n.nspname = 'public' \ + UNION ALL SELECT 'trigger', tg.tgname::text FROM pg_trigger tg \ + JOIN pg_class t ON t.oid = tg.tgrelid JOIN pg_namespace n ON n.oid = t.relnamespace \ + WHERE n.nspname = 'public' AND NOT tg.tgisinternal \ + UNION ALL SELECT 'policy', pol.polname::text FROM pg_policy pol \ + JOIN pg_class t ON t.oid = pol.polrelid JOIN pg_namespace n ON n.oid = t.relnamespace \ + WHERE n.nspname = 'public' \ + UNION ALL SELECT 'relation', c.relname::text FROM pg_class c \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE n.nspname = 'public' AND c.relkind NOT IN ('r', 'i') \ + UNION ALL SELECT 'type', ty.typname::text FROM pg_type ty \ + JOIN pg_namespace n ON n.oid = ty.typnamespace \ + WHERE n.nspname = 'public' AND ty.typtype IN ('d', 'e', 'r', 'm') \ + ORDER BY 1, 2"; + + // pgschema takes discrete connection flags. `PgConnectOptions` decodes the + // URL the way the gateway does, including a Unix-socket host, which + // pgschema accepts as a socket directory in `--host` like libpq does. + // The password comes from the resolved options, so the userinfo, + // `?password=`, `PGPASSWORD` and `.pgpass` precedence matches SQLx's. + // pgschema 1.7.4 joins these flags into an unquoted keyword DSN, so a + // password or socket path containing whitespace, a quote, `=` or a + // backslash is not supported here. + let base_url = std::env::var("BUZZ_TEST_DATABASE_URL").expect("test database URL"); + let options: PgConnectOptions = base_url.parse().expect("parse test database URL"); + let host = match options.get_socket() { + Some(dir) => dir.to_str().expect("socket directory utf-8").to_owned(), + None => options.get_host().to_owned(), + }; + let password = + percent_encoding::percent_decode_str(options.to_url_lossy().password().unwrap_or_default()) + .decode_utf8() + .expect("database password utf-8") + .into_owned(); + let connection = PgschemaConnection { + host, + port: options.get_port().to_string(), + user: options.get_username().to_owned(), + password, + }; + let admin = PgPool::connect_with(options.clone()) + .await + .expect("connect admin database"); + + let suffix = Uuid::new_v4().simple(); + let probes = [ + format!("push_gateway_desired_{suffix}"), + format!("push_gateway_migrated_{suffix}"), + ]; + // Run the comparison in its own task so a failed assertion, pgschema run + // or probe setup still reaches the cleanup below; the task's panic is + // re-raised unchanged afterwards. + let outcome = tokio::spawn(compare_probes( + admin.clone(), + options, + connection, + probes.clone(), + )) + .await; + // Attempt every drop before reporting any failure, so one failed drop + // never leaks the other probe. + let mut drop_failures = Vec::new(); + for db in &probes { + if let Err(error) = sqlx::query(AssertSqlSafe(format!( + "DROP DATABASE IF EXISTS {db} WITH (FORCE)" + ))) + .execute(&admin) + .await + { + drop_failures.push(format!("drop probe database {db}: {error}")); + } + } + match outcome { + Err(failure) => { + for message in &drop_failures { + eprintln!("{message}"); + } + std::panic::resume_unwind(failure.into_panic()); + } + Ok(()) => assert!(drop_failures.is_empty(), "{}", drop_failures.join("; ")), + } + + struct PgschemaConnection { + host: String, + port: String, + user: String, + password: String, + } + + async fn compare_probes( + admin: PgPool, + options: PgConnectOptions, + connection: PgschemaConnection, + [desired_db, migrated_db]: [String; 2], + ) { + // `template0` so neither probe inherits objects from `template1`. + for db in [&desired_db, &migrated_db] { + sqlx::query(AssertSqlSafe(format!( + "CREATE DATABASE {db} TEMPLATE template0" + ))) + .execute(&admin) + .await + .expect("create probe database"); + } + + let PgschemaConnection { + host, + port, + user, + password, + } = &connection; + let pgschema = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../bin/pgschema"); + let schema_file = schema_dir().join("schema.sql"); + let apply = std::process::Command::new(&pgschema) + .args([ + "apply", + "--auto-approve", + "--file", + schema_file.to_str().expect("schema path utf-8"), + "--host", + host, + "--port", + port, + "--user", + user, + "--password", + password, + "--db", + &desired_db, + "--plan-host", + host, + "--plan-port", + port, + "--plan-user", + user, + "--plan-password", + password, + "--plan-db", + &desired_db, + ]) + .output() + .expect("run bin/pgschema apply (hermit env required)"); + assert!( + apply.status.success(), + "pgschema apply failed: {}\n{}", + String::from_utf8_lossy(&apply.stdout), + String::from_utf8_lossy(&apply.stderr), + ); + + let desired = PgPool::connect_with(options.clone().database(&desired_db)) + .await + .expect("connect desired-state probe"); + let migrated = PgPool::connect_with(options.database(&migrated_db)) + .await + .expect("connect migrated probe"); + GATEWAY_MIGRATOR + .run(&migrated) + .await + .expect("apply gateway migrations"); + + let declared: BTreeSet = catalog(&desired, TABLES) + .await + .into_iter() + .filter_map(|row| row.into_iter().next().flatten()) + .collect(); + let served: BTreeSet = GATEWAY_TABLES.iter().map(|t| t.to_string()).collect(); + assert_eq!( + declared, served, + "schema/ must declare exactly GATEWAY_TABLES, the set readiness probes and \ + runtime grants cover" + ); + for pool in [&desired, &migrated] { + assert_eq!( + catalog(pool, NON_TABLE_OBJECTS).await, + Rows::new(), + "the gateway schema holds only tables and indexes" + ); + } + for (what, sql) in [ + ("table set", TABLES), + ("column", COLUMNS), + ("table option", TABLE_OPTIONS), + ("constraint", CONSTRAINTS), + ("index", INDEXES), + ] { + assert_eq!( + catalog(&desired, sql).await, + catalog(&migrated, sql).await, + "{what} parity mismatch: crates/buzz-push-gateway/schema/ has drifted from \ + migrations/; update the table file to the post-migration end state" + ); + } + desired.close().await; + migrated.close().await; + } +} diff --git a/docs/push-gateway-deployment.md b/docs/push-gateway-deployment.md index 331d49708fd..170122627cc 100644 --- a/docs/push-gateway-deployment.md +++ b/docs/push-gateway-deployment.md @@ -87,7 +87,7 @@ that app identity; omitting it is not a push shutdown mechanism. The gateway's dedicated pool does not consume the relay-oriented `BUZZ_DB_LOCK_TIMEOUT_MS`, `BUZZ_DB_IDLE_TXN_TIMEOUT_MS`, or `BUZZ_DB_STATEMENT_TIMEOUT_MS` settings. Its session-timeout policy remains separate from the `buzz-db` writer policy and must be designed and rolled out independently. -All replicas must share one PostgreSQL database. Delivery authority, replay admission, and endpoint quota reservation are transactional there, so replica count does not multiply the abuse ceiling. The gateway owns a scoped migration history under `crates/buzz-push-gateway/migrations`; it creates only the six `push_gateway_*` authority tables plus SQLx's migration-history table and never runs relay migrations. +All replicas must share one PostgreSQL database. Delivery authority, replay admission, and endpoint quota reservation are transactional there, so replica count does not multiply the abuse ceiling. The gateway owns a scoped migration history under `crates/buzz-push-gateway/migrations`; it creates only the six `push_gateway_*` authority tables plus SQLx's migration-history table and never runs relay migrations. `crates/buzz-push-gateway/schema/` mirrors those migrations' end state as declarative desired-state DDL, one file per table; a parity test keeps their `public` tables, columns, table options, constraints, and indexes identical. The Helm chart runs a single pre-install/pre-upgrade migration Job using `migration.existingSecret`; that secret contains a DDL-capable `DATABASE_URL`. The URL MUST name a dedicated gateway database, not the relay database: SQLx stores its `_sqlx_migrations` history in `public`, so sharing a database would collide with another application's migration history. `migration.runtimeDatabaseRole` names an existing LOGIN role (the default is `buzz_push_gateway_runtime`) used by runtime `DATABASE_URL`. After scoped migrations, the Job revokes database `CREATE` from that role and schema `CREATE` from both `PUBLIC` and the role, then grants only database `CONNECT`, schema `USAGE`, and `SELECT, INSERT, UPDATE, DELETE` on the six gateway tables. The migration role must own the database/schema objects or otherwise be allowed to issue those grants; it is never provided to runtime replicas. Readiness rejects an empty/partial schema, missing DML, or a runtime role that retains database/schema `CREATE`. Helm waits for the migration hook before updating replicas, so rolling deployments never race unconditional startup migration. Readiness must be removed from load-balancer service endpoints before terminating a pod.