From 8e7aafb72e76cf483cc3bae9e1917c64580eb7ad Mon Sep 17 00:00:00 2001 From: A Guhan Date: Fri, 23 Oct 2026 16:24:09 +0530 Subject: [PATCH 1/3] Set up Streamgres schema at startup --- .env.example | 23 +- Cargo.lock | 68 +-- README.md | 24 +- docs/deploy.md | 47 +-- scripts/sql/ddl-triggers.sql | 17 +- src/client/config.rs | 44 +- src/client/ddl_triggers.rs | 81 +++- src/client/groups.rs | 3 + src/client/mod.rs | 16 +- src/client/setup.rs | 785 +++++++++++++++++++++++++++++++++++ src/sync/pg/threads.rs | 2 +- 11 files changed, 969 insertions(+), 141 deletions(-) create mode 100644 src/client/setup.rs diff --git a/.env.example b/.env.example index af59a06..7cb3b3d 100644 --- a/.env.example +++ b/.env.example @@ -9,22 +9,17 @@ STREAMGRES_BASE_PATH=/sync # The database: the user needs replication rights and read access to every # published table. STREAMGRES_PG_DSN=postgresql://postgres:postgres@localhost:5432/xyne_dev_db +# Writable primary for schema, publication and DDL trigger setup at startup. +STREAMGRES_MASTER_PG_DSN=postgresql://postgres:postgres@localhost:5432/xyne_dev_db # Schemas whose tables the catalog carries; default: public and _. -# STREAMGRES_SCHEMAS=public,xyne_0 -# The publication the change feed streams; the default is retained for -# compatibility. It must already exist: create it on the primary before -# starting Streamgres. +# STREAMGRES_SCHEMAS=public,streamgres_0 +# The publication the change feed streams. Startup creates it for every +# STREAMGRES_SCHEMAS schema if absent and validates it if present. # The replication slot is not configured: each server process creates its # own using the existing compatible naming scheme. -STREAMGRES_PUBLICATION=xyne_sync_pub -# Schema changes are heard through the schema-change event trigger on -# `ddl_command_end` (`_ddl_end_`), whose logical messages carry -# the prefix `//ddl`; the server refuses to start without the -# trigger. A table added, or a column added with no default or a constant -# one, is followed while the server runs; any other change to a served table -# stops the server, which restarts on the new schema (docs/deploy.md ยง4). -# STREAMGRES_DDL_TRIGGER=xyne_ddl_end_0 -# STREAMGRES_DDL_PREFIX=xyne/0/ddl +STREAMGRES_PUBLICATION=streamgres_sync_pub +# Startup creates or validates the DDL triggers; their names and WAL message +# prefix come from STREAMGRES_APP_ID and STREAMGRES_SHARD. # How often a fresh exported snapshot (a temporary slot) is minted for reads. STREAMGRES_SNAPSHOT_ROTATION_MS=1000 # Drop ended processes' inactive permanent feed slots after this age. 0 @@ -106,7 +101,7 @@ STREAMGRES_FORWARD_COOKIES=true STREAMGRES_BACKEND_TIMEOUT_MS=30000 # The application id and shard: together they name the schema the # application server records mutations in (_.clients). -STREAMGRES_APP_ID=xyne +STREAMGRES_APP_ID=streamgres STREAMGRES_SHARD=0 # The connection protocol. A `pong` answers every `ping`; when nothing has diff --git a/Cargo.lock b/Cargo.lock index d2b9643..f6d4298 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2139,6 +2139,40 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "streamgres" +version = "0.1.0" +dependencies = [ + "arc-swap", + "axum", + "base64 0.22.1", + "bytes", + "chrono", + "fallible-iterator 0.2.0", + "futures-util", + "jemalloc_pprof", + "libc", + "log", + "memchr", + "once_cell", + "percent-encoding", + "pgoutput", + "pgwire-replication", + "postgres-protocol", + "pyroscope", + "reqwest 0.12.28", + "rustc-hash", + "rustls", + "serde", + "serde_json", + "socket2", + "tikv-jemallocator", + "tokio", + "tokio-postgres", + "tokio-tungstenite", + "uuid", +] + [[package]] name = "stringprep" version = "0.1.5" @@ -2941,40 +2975,6 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" -[[package]] -name = "streamgres" -version = "0.1.0" -dependencies = [ - "arc-swap", - "axum", - "base64 0.22.1", - "bytes", - "chrono", - "fallible-iterator 0.2.0", - "futures-util", - "jemalloc_pprof", - "libc", - "log", - "memchr", - "once_cell", - "percent-encoding", - "pgoutput", - "pgwire-replication", - "postgres-protocol", - "pyroscope", - "reqwest 0.12.28", - "rustc-hash", - "rustls", - "serde", - "serde_json", - "socket2", - "tikv-jemallocator", - "tokio", - "tokio-postgres", - "tokio-tungstenite", - "uuid", -] - [[package]] name = "yoke" version = "0.8.3" diff --git a/README.md b/README.md index 7c1f4c2..989a4a4 100644 --- a/README.md +++ b/README.md @@ -43,8 +43,9 @@ clients Streamgres PostgreSQL PostgreSQL's write-ahead log. Initial reads use snapshots that are never ahead of it, and writes delivered while a read is in flight are applied to its result before it lands. A client never sees a row go backwards. -- **Writes nothing to your database.** It follows a primary, a logical replica - or a physical standby over logical replication. +- **The change feed is read-only.** Startup setup creates or validates the + mutation schema, publication and DDL triggers on the writable primary; the + feed then follows a primary, logical replica or physical standby. - **Live schema changes.** New tables and new columns with a constant default are picked up while running. Any other change to a served table stops the server with a message naming the change. Once restarted, by your @@ -89,21 +90,10 @@ Run `cargo run --release --bin bench` to measure on your own hardware. ### 1. Prepare PostgreSQL -```sql --- the publication the change feed streams (FOR ALL TABLES needs a superuser) -CREATE PUBLICATION xyne_sync_pub FOR ALL TABLES; -``` - -Install the schema-change event trigger the server listens to. It refuses to -start without it: - -```bash -psql "$DATABASE_URL" < scripts/sql/ddl-triggers.sql -``` - -The checked-in stack is for app `xyne`, shard 0 and publication -`xyne_sync_pub`. For another app, shard or publication, generate the stack -with `trigger_stack_sql` in `src/client/ddl_triggers.rs`. +Set `STREAMGRES_MASTER_PG_DSN` to a writable PostgreSQL superuser connection. +At startup, Streamgres creates or validates the mutation schema, publication +and DDL trigger stack. It refuses to start if an existing object differs from +the configured app ID, shard or publication. ### 2. Configure and run diff --git a/docs/deploy.md b/docs/deploy.md index 8ab2ca8..9b8051c 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -77,43 +77,27 @@ branch name and the short commit SHA. - **A direct connection, not PgBouncer.** Logical replication does not pass through PgBouncer. -**The server writes nothing to the database it follows.** It reads rows and -the replication stream. Its position in the stream comes from each commit and -from the keepalives PostgreSQL sends anyway. So `STREAMGRES_PG_DSN` can point -at: +The change feed writes nothing to the database it follows. Its position comes +from commits and PostgreSQL keepalives. `STREAMGRES_PG_DSN` can point at: - the primary, - a logical replica, or - a physical standby (PostgreSQL 16 or later, the first version that can do logical decoding on a standby). -### One-time setup +Startup setup uses `STREAMGRES_MASTER_PG_DSN` to connect to the writable primary. +It creates the `_` mutation schema, publication, DDL functions, and +event triggers when absent. Existing objects are checked without replacement; +a mismatch stops startup. The master role must be a PostgreSQL superuser to +create the publication and event triggers. -1. **Create the publication** on the primary. The server checks for it at - startup and exits if it is missing: +### Startup setup - ```sql - CREATE PUBLICATION xyne_sync_pub FOR ALL TABLES; -- needs a superuser - ``` - - The name is set by `STREAMGRES_PUBLICATION`. The publication must include - your application's mutation-tracking tables (section 4). - -2. **Install the schema-change event trigger** on the primary. The server - refuses to start without it: - - ```bash - psql "$DATABASE_URL" < scripts/sql/ddl-triggers.sql - ``` - - The checked-in stack is for app `xyne`, shard 0 and publication - `xyne_sync_pub`. For another app, shard or publication, generate the - stack with `trigger_stack_sql` in `src/client/ddl_triggers.rs`. - - This creates the event trigger `_ddl_end_`. During every - migration, inside the migration's own transaction, it records the published - schema before and after into the WAL. The server therefore sees each schema - change at its commit, on any topology. Use the same app id and shard you set - in `STREAMGRES_APP_ID` and `STREAMGRES_SHARD`. +Choose `STREAMGRES_PUBLICATION` and `STREAMGRES_SCHEMAS`. On first startup, +Streamgres creates the publication and the schema-change trigger stack on the +primary. Existing objects must match the configured app ID, shard, publication, +and expected definitions. The trigger names and WAL prefix are derived from +`STREAMGRES_APP_ID` and `STREAMGRES_SHARD`. During a migration, the triggers +record the published schema before and after the DDL in that transaction's WAL. ### Replication slots @@ -210,6 +194,7 @@ directory is read first, and variables already set take precedence. | variable | meaning | |---|---| | `STREAMGRES_PG_DSN` | the PostgreSQL to follow, reached directly (section 3) | +| `STREAMGRES_MASTER_PG_DSN` | writable primary used during startup setup | | `STREAMGRES_QUERY_URL` | the application server's query endpoint | | `STREAMGRES_MUTATE_URL` | the application server's mutate endpoint | @@ -244,8 +229,6 @@ depend on what the defaults happen to be: | `STREAMGRES_ADDR` | `0.0.0.0:4848` | listen address | | `STREAMGRES_SCHEMAS` | `public,_` | schemas whose tables are served | | `STREAMGRES_PUBLICATION` | `xyne_sync_pub` | the publication the change feed streams | -| `STREAMGRES_DDL_TRIGGER` | `_ddl_end_` | the schema-change event trigger the server requires | -| `STREAMGRES_DDL_PREFIX` | `//ddl` | the prefix of that trigger's messages | | `STREAMGRES_FORWARD_COOKIES` | `true` | forward the connection's cookies to the application server | | `STREAMGRES_SNAPSHOT_ROTATION_MS` | `1000` | how often a fresh read snapshot is created | | `STREAMGRES_SLOT_CLEANUP_AGE_MS` | `0` (off) | at startup, drop inactive slots of ended processes older than this (PostgreSQL 17+) | diff --git a/scripts/sql/ddl-triggers.sql b/scripts/sql/ddl-triggers.sql index e89ebd9..02a07e1 100644 --- a/scripts/sql/ddl-triggers.sql +++ b/scripts/sql/ddl-triggers.sql @@ -83,21 +83,21 @@ tables AS (SELECT json_build_object( ELSE "maxLen" END, 'notNull', "notNull", 'dflt', "dflt" - ) + ) ORDER BY col ), 'primaryKey', ARRAY( SELECT json_object_keys( json_strip_nulls( json_object_agg( - DISTINCT "col", "keyPos" ORDER BY "keyPos" + DISTINCT "col", "keyPos" ORDER BY "keyPos", "col" ) ) )), 'publications', json_object_agg( DISTINCT "publication", - jsonb_build_object('rowFilter', "rowFilter") + jsonb_build_object('rowFilter', "rowFilter") ORDER BY "publication" ) -) AS "table" FROM published_columns +) AS "table", "schema", "name" FROM published_columns GROUP BY "schema", "schemaOID", "name", "oid", "replicaIdentity"), indexed_columns AS (SELECT @@ -105,6 +105,7 @@ tables AS (SELECT json_build_object( pg_indexes.tablename as "tableName", pg_indexes.indexname as "name", index_column.name as "col", + index_column.pos as "pos", CASE WHEN pg_index.indoption[index_column.pos-1] & 1 = 1 THEN 'DESC' ELSE 'ASC' END as "dir", (pg_index.indisunique AND pg_index.indpred IS NULL) as "unique", pg_index.indisprimary as "isPrimaryKey", @@ -153,14 +154,14 @@ tables AS (SELECT json_build_object( 'isReplicaIdentity', "isReplicaIdentity", 'isImmediate', "isImmediate", 'predicateSQL', "predicateSQL", - 'columns', json_object_agg("col", "dir") - ) AS index FROM indexed_columns + 'columns', json_object_agg("col", "dir" ORDER BY "pos") + ) AS index, "schema", "tableName", "name" FROM indexed_columns GROUP BY "schema", "tableName", "name", "unique", "isPrimaryKey", "isReplicaIdentity", "isImmediate", "predicateSQL") SELECT json_build_object( - 'tables', COALESCE((SELECT json_agg("table") FROM tables), '[]'::json), - 'indexes', COALESCE((SELECT json_agg("index") FROM indexes), '[]'::json) + 'tables', COALESCE((SELECT json_agg("table" ORDER BY "schema", "name") FROM tables), '[]'::json), + 'indexes', COALESCE((SELECT json_agg("index" ORDER BY "schema", "tableName", "name") FROM indexes), '[]'::json) ) as "publishedSchema" $$ LANGUAGE sql; diff --git a/src/client/config.rs b/src/client/config.rs index 851d4eb..1fe75d8 100644 --- a/src/client/config.rs +++ b/src/client/config.rs @@ -24,6 +24,8 @@ pub struct Config { /// `STREAMGRES_PG_DSN` (or `ZERO_UPSTREAM_DB`): the database, with the /// user and password in the URL. pub dsn: String, + /// `STREAMGRES_MASTER_PG_DSN`: writable primary used for startup setup. + pub master_dsn: String, /// `STREAMGRES_SCHEMAS`: the schemas whose tables the catalog carries /// (`public` plus the app's `zero_` schema by default). pub schemas: Vec, @@ -36,15 +38,11 @@ pub struct Config { /// because that is where `inactive_since` is available. pub slot_cleanup_age: Duration, /// `STREAMGRES_PUBLICATION`: the publication the change feed streams; - /// it must exist before the server starts. + /// startup creates it when absent and validates it when present. pub publication: String, - /// `STREAMGRES_DDL_TRIGGER`: the event trigger on `ddl_command_end` - /// through which the server hears of schema changes, zero-cache's - /// (`_ddl_end_`, so `zero_ddl_end_0`); the server refuses - /// to start without it. + /// The event trigger on `ddl_command_end`, derived from app ID and shard. pub ddl_trigger: String, - /// `STREAMGRES_DDL_PREFIX`: the prefix of that trigger's logical - /// messages (`//ddl`, so `zero/0/ddl`). + /// The logical message prefix, derived from app ID and shard. pub ddl_prefix: String, /// `STREAMGRES_SNAPSHOT_ROTATION_MS`: how often a fresh exported /// snapshot (a temporary replication slot) is minted for reads (1000). @@ -223,6 +221,8 @@ impl Config { }; let dsn = first(&["STREAMGRES_PG_DSN", "ZERO_UPSTREAM_DB"]) .ok_or("STREAMGRES_PG_DSN (or ZERO_UPSTREAM_DB) must name the database")?; + let master_dsn = first(&["STREAMGRES_MASTER_PG_DSN"]) + .ok_or("STREAMGRES_MASTER_PG_DSN must name the writable primary database")?; let query_url = first(&["STREAMGRES_QUERY_URL", "ZERO_QUERY_URL"]) .ok_or("STREAMGRES_QUERY_URL (or ZERO_QUERY_URL) must name the query endpoint")?; let mutate_url = first(&["STREAMGRES_MUTATE_URL", "ZERO_MUTATE_URL"]) @@ -235,10 +235,8 @@ impl Config { .map_err(|_| format!("STREAMGRES_SHARD must be a number, got `{text}`"))?, None => 0, }; - let ddl_trigger = first(&["STREAMGRES_DDL_TRIGGER"]) - .unwrap_or_else(|| format!("{app_id}_ddl_end_{shard}")); - let ddl_prefix = - first(&["STREAMGRES_DDL_PREFIX"]).unwrap_or_else(|| format!("{app_id}/{shard}/ddl")); + let ddl_trigger = format!("{app_id}_ddl_end_{shard}"); + let ddl_prefix = format!("{app_id}/{shard}/ddl"); let schemas = match first(&["STREAMGRES_SCHEMAS"]) { Some(text) => text .split(',') @@ -313,6 +311,7 @@ impl Config { bind: first(&["STREAMGRES_ADDR"]).unwrap_or_else(|| "0.0.0.0:4848".to_owned()), base_path, dsn, + master_dsn, schemas, slot: threads::slot_name(), slot_cleanup_age: millis("STREAMGRES_SLOT_CLEANUP_AGE_MS", 0)?, @@ -498,6 +497,7 @@ mod tests { fn config(vars: &[(&str, &str)]) -> Result { Config::from_lookup(|name| match name { "STREAMGRES_PG_DSN" => Some("postgresql://u:p@db/app".to_owned()), + "STREAMGRES_MASTER_PG_DSN" => Some("postgresql://admin:p@db/app".to_owned()), "STREAMGRES_QUERY_URL" => Some("http://app/query".to_owned()), "STREAMGRES_MUTATE_URL" => Some("http://app/push".to_owned()), _ => vars @@ -507,6 +507,18 @@ mod tests { }) } + #[test] + fn primary_database_is_required() { + let error = Config::from_lookup(|name| match name { + "STREAMGRES_PG_DSN" => Some("postgresql://db/app".to_owned()), + "STREAMGRES_QUERY_URL" => Some("http://app/query".to_owned()), + "STREAMGRES_MUTATE_URL" => Some("http://app/push".to_owned()), + _ => None, + }) + .unwrap_err(); + assert!(error.contains("STREAMGRES_MASTER_PG_DSN")); + } + /// The DDL trigger and its message prefix follow zero-cache's naming /// from the app id and the shard unless named outright. #[test] @@ -573,20 +585,14 @@ mod tests { } #[test] - fn the_ddl_trigger_follows_the_app_and_shard() { + fn ddl_trigger_and_prefix_follow_the_app_and_shard() { let config = config(&[]).unwrap(); assert_eq!(config.ddl_trigger, "zero_ddl_end_0"); assert_eq!(config.ddl_prefix, "zero/0/ddl"); let config = config_with(&[("STREAMGRES_APP_ID", "zero02"), ("STREAMGRES_SHARD", "3")]); assert_eq!(config.ddl_trigger, "zero02_ddl_end_3"); assert_eq!(config.ddl_prefix, "zero02/3/ddl"); - let config = config_with(&[ - ("STREAMGRES_DDL_TRIGGER", "my_trigger"), - ("STREAMGRES_DDL_PREFIX", "mine/ddl"), - ]); - assert_eq!(config.ddl_trigger, "my_trigger"); - assert_eq!(config.ddl_prefix, "mine/ddl"); - assert_eq!(config.engine_settings().ddl_trigger, "my_trigger"); + assert_eq!(config.engine_settings().ddl_trigger, "zero02_ddl_end_3"); } /// [`config`], expected to parse. diff --git a/src/client/ddl_triggers.rs b/src/client/ddl_triggers.rs index ecc0427..d9e19b7 100644 --- a/src/client/ddl_triggers.rs +++ b/src/client/ddl_triggers.rs @@ -10,7 +10,7 @@ use crate::sync::pg::sql::{quote_ident, quote_literal}; /// The advisory lock every DDL statement takes under zero-cache's start /// trigger, so that the schemas its messages carry diff in order. -const DDL_LOCK: i64 = 0x3c6b_8468_f1ba_c0b0; +pub(crate) const DDL_LOCK: i64 = 0x3c6b_8468_f1ba_c0b0; /// The command tags zero-cache's triggers fire on. const DDL_TAGS: [&str; 7] = [ @@ -37,16 +37,73 @@ pub fn trigger_stack_sql(app: &str, shard: u32, publications: &[&str]) -> String .iter() .map(|name| quote_literal(name)) .collect(); + let app_literal = app.replace('\'', "''"); TRIGGER_STACK .replace("@SCHEMA@", "e_ident(&format!("{app}_{shard}"))) .replace("@START@", "e_ident(&format!("{app}_ddl_start_{shard}"))) .replace("@END@", "e_ident(&format!("{app}_ddl_end_{shard}"))) - .replace("@APP@", app) .replace("@SHARD@", &shard.to_string()) .replace("@PUBLICATIONS@", &publications.join(", ")) .replace("@TAGS_AND_COMMENT@", &format!("{tags}, 'COMMENT'")) .replace("@TAGS@", &tags) .replace("@LOCK@", &DDL_LOCK.to_string()) + .replace("@APP@", &app_literal) +} + +pub(crate) struct FunctionSpec { + pub name: String, + pub args: Vec, + pub body: String, + pub language: String, +} + +pub(crate) fn function_specs(app: &str, shard: u32, publications: &[&str]) -> Vec { + let sql = trigger_stack_sql(app, shard, publications); + let mut specs = Vec::new(); + let mut rest = sql.as_str(); + while let Some(start) = rest.find("FUNCTION ") { + rest = &rest[start + "FUNCTION ".len()..]; + let Some(header_end) = rest.find("\nRETURNS ") else { + break; + }; + let header = &rest[..header_end]; + let Some((_, signature)) = header.rsplit_once('.') else { + break; + }; + let Some(open) = signature.find('(') else { + break; + }; + let Some(close) = signature.rfind(')') else { + break; + }; + let args = signature[open + 1..close] + .split(',') + .filter_map(|arg| arg.split_whitespace().last()) + .map(str::to_ascii_lowercase) + .collect(); + let Some(body_start) = rest[header_end..].find("AS $$") else { + break; + }; + let body_start = header_end + body_start + "AS $$".len(); + let Some(body_end) = rest[body_start..].find("$$ LANGUAGE ") else { + break; + }; + let body = rest[body_start..body_start + body_end].to_owned(); + let language_start = body_start + body_end + "$$ LANGUAGE ".len(); + let Some(language_end) = rest[language_start..].find(';') else { + break; + }; + specs.push(FunctionSpec { + name: signature[..open].trim_matches('"').to_owned(), + args, + body, + language: rest[language_start..language_start + language_end] + .trim() + .to_ascii_lowercase(), + }); + rest = &rest[language_start + language_end + 1..]; + } + specs } /// [`trigger_stack_sql`]'s text, with the names to fill in marked. @@ -135,21 +192,21 @@ tables AS (SELECT json_build_object( ELSE "maxLen" END, 'notNull', "notNull", 'dflt', "dflt" - ) + ) ORDER BY col ), 'primaryKey', ARRAY( SELECT json_object_keys( json_strip_nulls( json_object_agg( - DISTINCT "col", "keyPos" ORDER BY "keyPos" + DISTINCT "col", "keyPos" ORDER BY "keyPos", "col" ) ) )), 'publications', json_object_agg( DISTINCT "publication", - jsonb_build_object('rowFilter', "rowFilter") + jsonb_build_object('rowFilter', "rowFilter") ORDER BY "publication" ) -) AS "table" FROM published_columns +) AS "table", "schema", "name" FROM published_columns GROUP BY "schema", "schemaOID", "name", "oid", "replicaIdentity"), indexed_columns AS (SELECT @@ -157,6 +214,7 @@ tables AS (SELECT json_build_object( pg_indexes.tablename as "tableName", pg_indexes.indexname as "name", index_column.name as "col", + index_column.pos as "pos", CASE WHEN pg_index.indoption[index_column.pos-1] & 1 = 1 THEN 'DESC' ELSE 'ASC' END as "dir", (pg_index.indisunique AND pg_index.indpred IS NULL) as "unique", pg_index.indisprimary as "isPrimaryKey", @@ -205,14 +263,14 @@ tables AS (SELECT json_build_object( 'isReplicaIdentity', "isReplicaIdentity", 'isImmediate', "isImmediate", 'predicateSQL', "predicateSQL", - 'columns', json_object_agg("col", "dir") - ) AS index FROM indexed_columns + 'columns', json_object_agg("col", "dir" ORDER BY "pos") + ) AS index, "schema", "tableName", "name" FROM indexed_columns GROUP BY "schema", "tableName", "name", "unique", "isPrimaryKey", "isReplicaIdentity", "isImmediate", "predicateSQL") SELECT json_build_object( - 'tables', COALESCE((SELECT json_agg("table") FROM tables), '[]'::json), - 'indexes', COALESCE((SELECT json_agg("index") FROM indexes), '[]'::json) + 'tables', COALESCE((SELECT json_agg("table" ORDER BY "schema", "name") FROM tables), '[]'::json), + 'indexes', COALESCE((SELECT json_agg("index" ORDER BY "schema", "tableName", "name") FROM indexes), '[]'::json) ) as "publishedSchema" $$ LANGUAGE sql; @@ -400,6 +458,9 @@ mod tests { assert!(sql.contains("\"zero_0\".schema_specs()")); assert!(sql.contains("pb.pubname IN ('zero_public_0', 'zero_meta')")); assert!(sql.contains("ARRAY['zero_public_0', 'zero_meta']")); + assert!(sql.contains("json_agg(\"table\" ORDER BY \"schema\", \"name\")")); + assert!(sql.contains("json_agg(\"index\" ORDER BY \"schema\", \"tableName\", \"name\")")); + assert!(sql.contains("json_object_agg(\"col\", \"dir\" ORDER BY \"pos\")")); let unfilled = sql .split('@') .skip(1) diff --git a/src/client/groups.rs b/src/client/groups.rs index 436673d..019c825 100644 --- a/src/client/groups.rs +++ b/src/client/groups.rs @@ -1949,6 +1949,7 @@ mod tests { fn new(log_bytes: &str) -> Self { let vars: HashMap<&str, &str> = HashMap::from([ ("STREAMGRES_PG_DSN", "postgresql://none/none"), + ("STREAMGRES_MASTER_PG_DSN", "postgresql://none/none"), ("STREAMGRES_QUERY_URL", "http://none/query"), ("STREAMGRES_MUTATE_URL", "http://none/push"), ("STREAMGRES_GROUP_LOG_BYTES", log_bytes), @@ -3299,6 +3300,7 @@ mod tests { let vars: HashMap<&str, &str> = HashMap::from([ ("STREAMGRES_PG_DSN", "postgresql://none/none"), + ("STREAMGRES_MASTER_PG_DSN", "postgresql://none/none"), ("STREAMGRES_QUERY_URL", "http://none/query"), ("STREAMGRES_MUTATE_URL", "http://none/push"), ]); @@ -3981,6 +3983,7 @@ mod tests { let vars: HashMap<&str, &str> = HashMap::from([ ("STREAMGRES_PG_DSN", "postgresql://none/none"), + ("STREAMGRES_MASTER_PG_DSN", "postgresql://none/none"), ("STREAMGRES_QUERY_URL", "http://none/query"), ("STREAMGRES_MUTATE_URL", "http://none/push"), ]); diff --git a/src/client/mod.rs b/src/client/mod.rs index d72a9f1..6b4e70b 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -59,6 +59,7 @@ pub mod plan; pub mod protocol; pub mod sampler; pub mod schema; +mod setup; pub mod transform; pub mod warm; pub mod wire; @@ -111,19 +112,22 @@ async fn join_worker( /// Run the server with `config` until ctrl-c; returns the reason it could /// not start. -pub fn serve(config: Config) -> Result<(), String> { +pub fn serve(mut config: Config) -> Result<(), String> { log::set_level(config.log); log::set_format(config.log_format); - let _profiler = crate::profile::start(crate::profile::Config::from_env()?)?; - let config = Arc::new(config); - let shutdown = Shutdown::new(); - let settings = config.engine_settings(); - let cleanup_settings = settings.clone(); + let master_dsn = std::mem::take(&mut config.master_dsn); let server = tokio::runtime::Builder::new_multi_thread() .enable_all() .thread_name("streamgres-server") .build() .map_err(|error| format!("tokio: {error}"))?; + server.block_on(setup::ensure(&master_dsn, &config))?; + drop(master_dsn); + let config = Arc::new(config); + let shutdown = Shutdown::new(); + let settings = config.engine_settings(); + let cleanup_settings = settings.clone(); + let _profiler = crate::profile::start(crate::profile::Config::from_env()?)?; let catalog = Arc::new(CatalogHandle::new( server.block_on(threads::load_catalog_at(&settings.dsn, &settings.schemas))?, )); diff --git a/src/client/setup.rs b/src/client/setup.rs new file mode 100644 index 0000000..f5e58f8 --- /dev/null +++ b/src/client/setup.rs @@ -0,0 +1,785 @@ +use std::collections::BTreeMap; + +use serde::de::{self, MapAccess, Visitor}; +use serde::{Deserialize, Deserializer}; +use serde_json::Value; +use tokio_postgres::{NoTls, Transaction}; + +use super::Config; +use super::ddl_triggers::{self, FunctionSpec}; +use crate::sync::pg::sql::quote_ident; + +const SETUP_LOCK: i64 = 0x5354_524d_4752_4553; + +pub async fn ensure(master_dsn: &str, config: &Config) -> Result<(), String> { + let schema = config.upstream_schema(); + if !config.schemas.contains(&schema) { + return Err(format!( + "STREAMGRES_SCHEMAS must include mutation schema `{schema}`" + )); + } + + let (mut client, connection) = tokio_postgres::connect(master_dsn, NoTls) + .await + .map_err(|error| format!("connecting to primary database: {error}"))?; + let connection = tokio::spawn(async move { connection.await }); + let result = async { + let tx = client + .transaction() + .await + .map_err(|error| format!("starting database setup: {error}"))?; + tx.batch_execute("SET LOCAL statement_timeout = '60s'; SET LOCAL lock_timeout = '15s'") + .await + .map_err(|error| format!("setting database setup timeouts: {error}"))?; + tx.query_one("SELECT pg_advisory_xact_lock($1)", &[&SETUP_LOCK]) + .await + .map_err(|error| format!("waiting for database setup lock: {error}"))?; + tx.query_one( + "SELECT pg_advisory_xact_lock($1)", + &[&ddl_triggers::DDL_LOCK], + ) + .await + .map_err(|error| format!("waiting for DDL setup lock: {error}"))?; + ensure_locked(&tx, config, &schema).await?; + tx.commit() + .await + .map_err(|error| format!("committing database setup: {error}")) + } + .await; + drop(client); + let _ = connection.await; + result +} + +async fn ensure_locked(tx: &Transaction<'_>, config: &Config, schema: &str) -> Result<(), String> { + let exists = tx + .query_opt("SELECT 1 FROM pg_namespace WHERE nspname = $1", &[&schema]) + .await + .map_err(|error| format!("checking schema `{schema}`: {error}"))? + .is_some(); + if exists { + for table in ["clients", "mutations", "publishedSchema"] { + validate_table(tx, schema, table).await?; + } + let tables = tx + .query( + "SELECT relname FROM pg_class WHERE relnamespace = + (SELECT oid FROM pg_namespace WHERE nspname = $1) AND relkind IN ('r', 'p')", + &[&schema], + ) + .await + .map_err(|error| format!("checking schema `{schema}` tables: {error}"))? + .into_iter() + .map(|row| row.get::<_, String>(0)) + .collect::>(); + if tables + != ["clients", "mutations", "publishedSchema"] + .map(str::to_owned) + .into() + { + return Err(format!( + "schema `{schema}` has unexpected tables: {tables:?}" + )); + } + } else { + let schema_name = quote_ident(schema); + tx.batch_execute(&format!( + "CREATE SCHEMA {schema_name}; + CREATE TABLE {schema_name}.\"clients\" ( + \"clientGroupID\" text NOT NULL, + \"clientID\" text NOT NULL, + \"lastMutationID\" bigint NOT NULL, + \"userID\" text, + PRIMARY KEY (\"clientGroupID\", \"clientID\") + ); + CREATE TABLE {schema_name}.\"mutations\" ( + \"clientGroupID\" text NOT NULL, + \"clientID\" text NOT NULL, + \"mutationID\" bigint NOT NULL, + \"result\" json NOT NULL, + PRIMARY KEY (\"clientGroupID\", \"clientID\", \"mutationID\") + ); + CREATE TABLE {schema_name}.\"publishedSchema\" ( + current json, + exists boolean NOT NULL DEFAULT true PRIMARY KEY CHECK (exists) + )" + )) + .await + .map_err(|error| format!("creating schema `{schema}`: {error}"))?; + } + + for served in &config.schemas { + if tx + .query_opt("SELECT 1 FROM pg_namespace WHERE nspname = $1", &[served]) + .await + .map_err(|error| format!("checking served schema `{served}`: {error}"))? + .is_none() + { + return Err(format!("served schema `{served}` does not exist")); + } + } + + let publication = &config.publication; + let existing = tx + .query_opt( + "SELECT puballtables, pubinsert, pubupdate, pubdelete, pubtruncate, pubviaroot + FROM pg_publication WHERE pubname = $1", + &[publication], + ) + .await + .map_err(|error| format!("checking publication `{publication}`: {error}"))?; + if let Some(row) = existing { + let flags = [ + row.get::<_, bool>(0), + row.get::<_, bool>(1), + row.get::<_, bool>(2), + row.get::<_, bool>(3), + row.get::<_, bool>(4), + row.get::<_, bool>(5), + ]; + let actual = tx + .query( + "SELECT n.nspname FROM pg_publication_namespace pn + JOIN pg_namespace n ON n.oid = pn.pnnspid + WHERE pn.pnpubid = (SELECT oid FROM pg_publication WHERE pubname = $1)", + &[publication], + ) + .await + .map_err(|error| format!("checking publication `{publication}` schemas: {error}"))? + .into_iter() + .map(|row| row.get::<_, String>(0)) + .collect::>(); + let wanted = config.schemas.iter().cloned().collect(); + let explicit: i64 = tx + .query_one( + "SELECT count(*) FROM pg_publication_rel + WHERE prpubid = (SELECT oid FROM pg_publication WHERE pubname = $1)", + &[publication], + ) + .await + .map_err(|error| format!("checking publication `{publication}` tables: {error}"))? + .get(0); + if flags != [false, true, true, true, true, false] || actual != wanted || explicit != 0 { + return Err(format!( + "publication `{publication}` differs: expected schema scopes {wanted:?}, all change types, no explicit tables; found scopes {actual:?}, flags {flags:?}, {explicit} explicit tables" + )); + } + } else { + let schemas = config + .schemas + .iter() + .map(|schema| quote_ident(schema)) + .collect::>() + .join(", "); + tx.batch_execute(&format!( + "CREATE PUBLICATION {} FOR TABLES IN SCHEMA {schemas}", + quote_ident(publication) + )) + .await + .map_err(|error| format!("creating publication `{publication}`: {error}"))?; + } + + ensure_triggers(tx, config, schema).await?; + Ok(()) +} + +async fn ensure_triggers( + tx: &Transaction<'_>, + config: &Config, + schema: &str, +) -> Result<(), String> { + let specs = ddl_triggers::function_specs(&config.app_id, config.shard, &[&config.publication]); + if specs.len() != 7 { + return Err(format!( + "internal error: expected seven DDL trigger functions, found {}", + specs.len() + )); + } + let start = format!("{}_ddl_start_{}", config.app_id, config.shard); + let end = format!("{}_ddl_end_{}", config.app_id, config.shard); + let mut present = false; + for spec in &specs { + let signature = function_signature(schema, spec); + let exists: bool = tx + .query_one("SELECT to_regprocedure($1) IS NOT NULL", &[&signature]) + .await + .map_err(|error| format!("checking DDL function `{signature}`: {error}"))? + .get(0); + present |= exists; + } + for trigger in [&start, &end] { + let exists = tx + .query_opt( + "SELECT 1 FROM pg_event_trigger WHERE evtname = $1", + &[trigger], + ) + .await + .map_err(|error| format!("checking DDL event trigger `{trigger}`: {error}"))? + .is_some(); + present |= exists; + } + + if present { + validate_triggers(tx, schema, &specs, &start, &end).await?; + } else { + let snapshot_populated: bool = tx + .query_opt( + &format!( + "SELECT current IS NOT NULL FROM {}.\"publishedSchema\"", + quote_ident(schema) + ), + &[], + ) + .await + .map_err(|error| format!("checking published schema baseline: {error}"))? + .is_some_and(|row| row.get(0)); + if snapshot_populated { + return Err(format!( + "schema `{schema}` has a published schema snapshot but no DDL trigger stack; refusing to replace the snapshot" + )); + } + tx.batch_execute(&ddl_triggers::trigger_stack_sql( + &config.app_id, + config.shard, + &[&config.publication], + )) + .await + .map_err(|error| format!("installing DDL event triggers: {error}"))?; + validate_triggers(tx, schema, &specs, &start, &end).await?; + } + Ok(()) +} + +fn function_signature(schema: &str, spec: &FunctionSpec) -> String { + format!( + "{}.{}({})", + quote_ident(schema), + quote_ident(&spec.name), + spec.args.join(", ") + ) +} + +async fn validate_triggers( + tx: &Transaction<'_>, + schema: &str, + specs: &[FunctionSpec], + start: &str, + end: &str, +) -> Result<(), String> { + for spec in specs { + let signature = function_signature(schema, spec); + let row = tx + .query_opt( + "SELECT p.prosrc, p.provolatile::text, l.lanname, + pg_get_function_result(p.oid) + FROM pg_proc p JOIN pg_language l ON l.oid = p.prolang + WHERE p.oid = to_regprocedure($1)", + &[&signature], + ) + .await + .map_err(|error| format!("validating DDL function `{signature}`: {error}"))? + .ok_or_else(|| format!("DDL trigger stack is incomplete: missing `{signature}`"))?; + let result = match spec.name.as_str() { + "get_trigger_context" => "record", + "notice_ignore" | "update_schemas" => "void", + "schema_specs" => "json", + "emit_ddl_start" | "emit_ddl_end" => "event_trigger", + _ => unreachable!("function specs come from the trigger template"), + }; + let volatility = if spec.name == "schema_specs" { + "s" + } else { + "v" + }; + if row.get::<_, String>(0) != spec.body + || row.get::<_, String>(1) != volatility + || row.get::<_, String>(2) != spec.language + || row.get::<_, String>(3) != result + { + return Err(format!( + "DDL function `{signature}` differs from the expected Streamgres definition" + )); + } + } + + let start_tags = [ + "ALTER PUBLICATION", + "ALTER SCHEMA", + "ALTER TABLE", + "CREATE INDEX", + "CREATE TABLE", + "DROP INDEX", + "DROP TABLE", + ]; + let end_tags = [ + "ALTER PUBLICATION", + "ALTER SCHEMA", + "ALTER TABLE", + "COMMENT", + "CREATE INDEX", + "CREATE TABLE", + "DROP INDEX", + "DROP TABLE", + ]; + for (name, event, function, expected_tags) in [ + ( + start, + "ddl_command_start", + "emit_ddl_start", + &start_tags[..], + ), + (end, "ddl_command_end", "emit_ddl_end", &end_tags[..]), + ] { + let row = tx + .query_opt( + "SELECT t.evtevent, t.evtenabled::text, t.evttags, + n.nspname, p.proname, + p.oid = to_regprocedure($2) AS expected_function + FROM pg_event_trigger t + JOIN pg_proc p ON p.oid = t.evtfoid + JOIN pg_namespace n ON n.oid = p.pronamespace + WHERE t.evtname = $1", + &[ + &name, + &format!("{}.{}()", quote_ident(schema), quote_ident(function)), + ], + ) + .await + .map_err(|error| format!("validating DDL event trigger `{name}`: {error}"))? + .ok_or_else(|| format!("DDL trigger stack is incomplete: missing `{name}`"))?; + let mut actual_tags = row.get::<_, Option>>(2).unwrap_or_default(); + actual_tags.sort(); + if row.get::<_, String>(0) != event + || row.get::<_, String>(1) != "O" + || actual_tags + != expected_tags + .iter() + .map(|tag| (*tag).to_owned()) + .collect::>() + || row.get::<_, String>(3) != schema + || row.get::<_, String>(4) != function + || !row.get::<_, bool>(5) + { + return Err(format!( + "DDL event trigger `{name}` differs from the expected Streamgres definition" + )); + } + } + let baseline = tx + .query_opt( + &format!( + "SELECT current::text, {}.schema_specs()::text + FROM {}.\"publishedSchema\"", + quote_ident(schema), + quote_ident(schema) + ), + &[], + ) + .await + .map_err(|error| format!("validating published schema baseline: {error}"))?; + let matches = match baseline { + Some(row) => { + let current: Option = row.get(0); + current + .map(|current| { + schema_snapshots_match(¤t, &row.get::<_, String>(1)) + .map_err(|error| format!("comparing published schema baseline: {error}")) + }) + .transpose()? + .unwrap_or(false) + } + None => false, + }; + if !matches { + return Err(format!( + "schema `{schema}` has a missing or stale published schema snapshot" + )); + } + Ok(()) +} + +#[derive(Deserialize, PartialEq)] +struct SchemaSnapshot { + tables: Vec, + indexes: Vec, + #[serde(flatten)] + fields: BTreeMap, +} + +#[derive(Deserialize, PartialEq)] +struct TableSnapshot { + schema: String, + name: String, + #[serde(flatten)] + fields: BTreeMap, +} + +#[derive(Deserialize, PartialEq)] +struct IndexSnapshot { + schema: String, + #[serde(rename = "tableName")] + table_name: String, + name: String, + columns: OrderedObject, + #[serde(flatten)] + fields: BTreeMap, +} + +#[derive(PartialEq)] +struct OrderedObject(Vec<(String, Value)>); + +impl<'de> Deserialize<'de> for OrderedObject { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + struct OrderedObjectVisitor; + + impl<'de> Visitor<'de> for OrderedObjectVisitor { + type Value = OrderedObject; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("an object with unique keys") + } + + fn visit_map(self, mut map: M) -> Result + where + M: MapAccess<'de>, + { + let mut entries = Vec::new(); + while let Some((key, value)) = map.next_entry::()? { + if entries.iter().any(|(existing, _)| existing == &key) { + return Err(de::Error::custom(format!("duplicate key `{key}`"))); + } + entries.push((key, value)); + } + Ok(OrderedObject(entries)) + } + } + + deserializer.deserialize_map(OrderedObjectVisitor) + } +} + +fn schema_snapshots_match(current: &str, expected: &str) -> Result { + let mut current: SchemaSnapshot = serde_json::from_str(current) + .map_err(|error| format!("unreadable stored schema snapshot: {error}"))?; + let mut expected: SchemaSnapshot = serde_json::from_str(expected) + .map_err(|error| format!("unreadable current schema snapshot: {error}"))?; + current.tables.sort_by(|a, b| (&a.schema, &a.name).cmp(&(&b.schema, &b.name))); + expected.tables.sort_by(|a, b| (&a.schema, &a.name).cmp(&(&b.schema, &b.name))); + current.indexes.sort_by(|a, b| { + (&a.schema, &a.table_name, &a.name).cmp(&(&b.schema, &b.table_name, &b.name)) + }); + expected.indexes.sort_by(|a, b| { + (&a.schema, &a.table_name, &a.name).cmp(&(&b.schema, &b.table_name, &b.name)) + }); + if current + .tables + .windows(2) + .any(|pair| pair[0].schema == pair[1].schema && pair[0].name == pair[1].name) + || expected + .tables + .windows(2) + .any(|pair| pair[0].schema == pair[1].schema && pair[0].name == pair[1].name) + || current.indexes.windows(2).any(|pair| { + pair[0].schema == pair[1].schema + && pair[0].table_name == pair[1].table_name + && pair[0].name == pair[1].name + }) + || expected.indexes.windows(2).any(|pair| { + pair[0].schema == pair[1].schema + && pair[0].table_name == pair[1].table_name + && pair[0].name == pair[1].name + }) + { + return Err("duplicate table or index identity".to_owned()); + } + Ok(current == expected) +} + +async fn validate_table(tx: &Transaction<'_>, schema: &str, table: &str) -> Result<(), String> { + let name = format!("{}.{}", quote_ident(schema), quote_ident(table)); + let rows = tx + .query( + "SELECT a.attname, format_type(a.atttypid, a.atttypmod), a.attnotnull, + pg_get_expr(d.adbin, d.adrelid) + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum > 0 AND NOT a.attisdropped + LEFT JOIN pg_attrdef d ON d.adrelid = c.oid AND d.adnum = a.attnum + WHERE n.nspname = $1 AND c.relname = $2 AND c.relkind = 'r' + AND c.relpersistence = 'p' + ORDER BY a.attnum", + &[&schema, &table], + ) + .await + .map_err(|error| format!("checking table {name}: {error}"))?; + let actual = rows + .iter() + .map(|row| { + ( + row.get::<_, String>(0), + row.get::<_, String>(1), + row.get::<_, bool>(2), + row.get::<_, Option>(3), + ) + }) + .collect::>(); + let expected = match table { + "clients" => vec![ + ("clientGroupID", "text", true, None), + ("clientID", "text", true, None), + ("lastMutationID", "bigint", true, None), + ("userID", "text", false, None), + ], + "mutations" => vec![ + ("clientGroupID", "text", true, None), + ("clientID", "text", true, None), + ("mutationID", "bigint", true, None), + ("result", "json", true, None), + ], + _ => vec![ + ("current", "json", false, None), + ("exists", "boolean", true, Some("true")), + ], + }; + if actual + != expected + .iter() + .map(|(name, kind, required, default)| { + ( + (*name).to_owned(), + (*kind).to_owned(), + *required, + default.map(str::to_owned), + ) + }) + .collect::>() + { + return Err(format!( + "table {name} has incompatible columns: expected {expected:?}, found {actual:?}" + )); + } + + let constraints = tx + .query( + "SELECT contype::text, pg_get_constraintdef(oid) + FROM pg_constraint WHERE conrelid = to_regclass($1) AND contype IN ('p', 'c')", + &[&name], + ) + .await + .map_err(|error| format!("checking table {name} constraints: {error}"))? + .into_iter() + .map(|row| (row.get::<_, String>(0), row.get::<_, String>(1))) + .collect::>(); + let key = match table { + "clients" => "PRIMARY KEY (\"clientGroupID\", \"clientID\")", + "mutations" => "PRIMARY KEY (\"clientGroupID\", \"clientID\", \"mutationID\")", + _ => "PRIMARY KEY (\"exists\")", + }; + if constraints.len() != if table == "publishedSchema" { 2 } else { 1 } + || !constraints + .iter() + .any(|(kind, definition)| kind == "p" && definition == key) + || (table == "publishedSchema" + && !constraints + .iter() + .any(|(kind, definition)| kind == "c" && definition == "CHECK (\"exists\")")) + { + return Err(format!( + "table {name} has incompatible constraints: expected primary key {key}, found {constraints:?}" + )); + } + let indexes: i64 = tx + .query_one( + "SELECT count(*) FROM pg_index WHERE indrelid = to_regclass($1)", + &[&name], + ) + .await + .map_err(|error| format!("checking table {name} indexes: {error}"))? + .get(0); + if indexes != 1 { + return Err(format!( + "table {name} has {indexes} indexes; expected only its primary key" + )); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn schema_snapshot_comparison_ignores_object_and_collection_order() { + let stored = r#"{"indexes":[{"schema":"public","tableName":"b","name":"b_idx","unique":false,"columns":{"id":"ASC"}},{"columns":{"id":"ASC"},"name":"a_idx","tableName":"a","schema":"public","unique":false}],"tables":[{"schema":"public","name":"b","primaryKey":["id"],"columns":{"id":{"pos":1,"dataType":"int8"}}},{"columns":{"id":{"dataType":"int8","pos":1}},"primaryKey":["id"],"name":"a","schema":"public"}]}"#; + let actual = r#"{"tables":[{"name":"a","schema":"public","columns":{"id":{"pos":1,"dataType":"int8"}},"primaryKey":["id"]},{"columns":{"id":{"dataType":"int8","pos":1}},"name":"b","primaryKey":["id"],"schema":"public"}],"indexes":[{"unique":false,"name":"a_idx","columns":{"id":"ASC"},"tableName":"a","schema":"public"},{"name":"b_idx","schema":"public","columns":{"id":"ASC"},"unique":false,"tableName":"b"}]}"#; + assert!(schema_snapshots_match(stored, actual).unwrap()); + } + + #[test] + fn schema_snapshot_comparison_preserves_key_and_index_column_order() { + let stored = r#"{"tables":[{"schema":"public","name":"items","primaryKey":["tenantId","id"],"columns":{}}],"indexes":[{"schema":"public","tableName":"items","name":"items_idx","columns":{"tenantId":"ASC","id":"DESC"}}]}"#; + let reordered_primary_key = r#"{"tables":[{"schema":"public","name":"items","primaryKey":["id","tenantId"],"columns":{}}],"indexes":[{"schema":"public","tableName":"items","name":"items_idx","columns":{"tenantId":"ASC","id":"DESC"}}]}"#; + let reordered_index_columns = r#"{"tables":[{"schema":"public","name":"items","primaryKey":["tenantId","id"],"columns":{}}],"indexes":[{"schema":"public","tableName":"items","name":"items_idx","columns":{"id":"DESC","tenantId":"ASC"}}]}"#; + assert!(!schema_snapshots_match(stored, reordered_primary_key).unwrap()); + assert!(!schema_snapshots_match(stored, reordered_index_columns).unwrap()); + } + + #[tokio::test] + async fn fresh_setup_is_repeatable_and_existing_tables_are_validated() { + let Ok(dsn) = std::env::var("STREAMGRES_SETUP_TEST_DSN") else { + return; + }; + let app = format!("sgtest{}", uuid::Uuid::new_v4().simple()); + let schema = format!("{app}_0"); + let publication = format!("{app}_pub"); + let schemas = format!("public,{schema}"); + let config = Config::from_lookup(|name| match name { + "STREAMGRES_PG_DSN" | "STREAMGRES_MASTER_PG_DSN" => Some(dsn.clone()), + "STREAMGRES_QUERY_URL" => Some("http://localhost/query".to_owned()), + "STREAMGRES_MUTATE_URL" => Some("http://localhost/push".to_owned()), + "STREAMGRES_APP_ID" => Some(app.clone()), + "STREAMGRES_SCHEMAS" => Some(schemas.clone()), + "STREAMGRES_PUBLICATION" => Some(publication.clone()), + _ => None, + }) + .unwrap(); + let (first, second) = tokio::join!(ensure(&dsn, &config), ensure(&dsn, &config)); + first.unwrap(); + second.unwrap(); + ensure(&dsn, &config).await.unwrap(); + + let (client, connection) = tokio_postgres::connect(&dsn, NoTls).await.unwrap(); + let connection = tokio::spawn(async move { connection.await }); + client + .batch_execute(&format!( + "UPDATE {}.\"publishedSchema\" SET current = json_build_object( + 'indexes', (SELECT COALESCE(json_agg(item ORDER BY item->>'schema' DESC, item->>'tableName' DESC, item->>'name' DESC), '[]'::json) + FROM json_array_elements(current->'indexes') AS i(item)), + 'tables', (SELECT COALESCE(json_agg(item ORDER BY item->>'schema' DESC, item->>'name' DESC), '[]'::json) + FROM json_array_elements(current->'tables') AS t(item)) + )", + quote_ident(&schema) + )) + .await + .unwrap(); + ensure(&dsn, &config).await.unwrap(); + + let probe = format!("{}.ddl_probe", quote_ident(&schema)); + client + .batch_execute(&format!( + "CREATE TABLE {probe} (id bigint PRIMARY KEY, value text); + ALTER TABLE {probe} ADD COLUMN probe_added text DEFAULT 'ok';" + )) + .await + .unwrap(); + let added: bool = client + .query_one( + &format!( + "SELECT EXISTS ( + SELECT 1 FROM {}.\"publishedSchema\", + LATERAL jsonb_array_elements(current::jsonb->'tables') AS t(item) + WHERE item->>'schema' = $1 AND item->>'name' = 'ddl_probe' + AND item->'columns' ? 'probe_added' + )", + quote_ident(&schema) + ), + &[&schema], + ) + .await + .unwrap() + .get(0); + assert!(added); + let index = format!("{}.ddl_probe_idx", quote_ident(&schema)); + client + .batch_execute(&format!("CREATE INDEX ddl_probe_idx ON {probe} (probe_added)")) + .await + .unwrap(); + let index_added: bool = client + .query_one( + &format!( + "SELECT EXISTS ( + SELECT 1 FROM {}.\"publishedSchema\", + LATERAL jsonb_array_elements(current::jsonb->'indexes') AS i(item) + WHERE item->>'schema' = $1 AND item->>'name' = 'ddl_probe_idx' + )", + quote_ident(&schema) + ), + &[&schema], + ) + .await + .unwrap() + .get(0); + assert!(index_added); + client + .batch_execute(&format!("DROP INDEX {index}; DROP TABLE {probe}")) + .await + .unwrap(); + + client + .batch_execute(&format!( + "ALTER PUBLICATION {} SET (publish = 'insert')", + quote_ident(&publication) + )) + .await + .unwrap(); + assert!( + ensure(&dsn, &config) + .await + .unwrap_err() + .contains("publication") + ); + client + .batch_execute(&format!( + "ALTER PUBLICATION {} SET (publish = 'insert, update, delete, truncate')", + quote_ident(&publication) + )) + .await + .unwrap(); + let mut conflicting = config.clone(); + conflicting.app_id.push('x'); + let other_schema = conflicting.upstream_schema(); + conflicting.schemas = vec!["public".to_owned(), other_schema.clone()]; + assert!( + ensure(&dsn, &conflicting) + .await + .unwrap_err() + .contains("publication") + ); + assert!( + client + .query_opt( + "SELECT 1 FROM pg_namespace WHERE nspname = $1", + &[&other_schema] + ) + .await + .unwrap() + .is_none() + ); + client + .batch_execute(&format!( + "ALTER TABLE {}.\"clients\" ADD COLUMN extra text", + quote_ident(&schema) + )) + .await + .unwrap(); + assert!( + ensure(&dsn, &config) + .await + .unwrap_err() + .contains("incompatible columns") + ); + client + .batch_execute(&format!( + "DROP PUBLICATION {}; DROP SCHEMA {} CASCADE", + quote_ident(&publication), + quote_ident(&schema) + )) + .await + .unwrap(); + drop(client); + connection.await.unwrap().unwrap(); + } +} diff --git a/src/sync/pg/threads.rs b/src/sync/pg/threads.rs index e765e0c..4e33be0 100644 --- a/src/sync/pg/threads.rs +++ b/src/sync/pg/threads.rs @@ -176,7 +176,7 @@ pub async fn require_ddl_trigger(dsn: &str, name: &str) -> Result<(), String> { Ok(()) } else { Err(format!( - "the event trigger `{name}` (STREAMGRES_DDL_TRIGGER) is not on this database or is disabled; the server hears of schema changes through it and will not serve without it" + "the event trigger `{name}` is missing or disabled; the server hears of schema changes through it and will not serve without it" )) } } From 5d3f7c024fbcf70fce8083aefdb2f38b10eb13f8 Mon Sep 17 00:00:00 2001 From: A Guhan Date: Fri, 23 Oct 2026 16:39:58 +0530 Subject: [PATCH 2/3] Fix Clippy warnings --- src/bin/bench.rs | 6 +++--- src/bin/load.rs | 17 +++++++---------- src/client/sampler.rs | 2 +- src/client/schema.rs | 4 ++-- src/client/setup.rs | 4 ++-- src/client/warm.rs | 4 ++-- src/stats.rs | 4 ++-- 7 files changed, 19 insertions(+), 22 deletions(-) diff --git a/src/bin/bench.rs b/src/bin/bench.rs index 9659437..8d9c5c5 100644 --- a/src/bin/bench.rs +++ b/src/bin/bench.rs @@ -2535,12 +2535,12 @@ fn production_query_workload() { stats.disjuncts_registered, stats.conditions_indexed, stats.snapshots_shared ); - let mut message = XY_CHANNELS * XY_CONVERSATIONS_PER_CHANNEL * XY_MESSAGES_PER_CONVERSATION; let mut inserts = Vec::with_capacity(XY_MESSAGE_INSERTS); - for _ in 0..XY_MESSAGE_INSERTS { + for message in (XY_CHANNELS * XY_CONVERSATIONS_PER_CHANNEL * XY_MESSAGES_PER_CONVERSATION..) + .take(XY_MESSAGE_INSERTS) + { let conversation = rng.below(XY_CHANNELS * XY_CONVERSATIONS_PER_CHANNEL); inserts.push(xy_message(&mut rng, message, conversation)); - message += 1; } let mut churn = Vec::with_capacity(XY_MEMBERSHIP_CHURN); for step in 0..XY_MEMBERSHIP_CHURN { diff --git a/src/bin/load.rs b/src/bin/load.rs index e245796..29cf464 100644 --- a/src/bin/load.rs +++ b/src/bin/load.rs @@ -1744,8 +1744,8 @@ impl Conn { counter: u64, owed_from_ns: Option, ) -> Result, String> { - let client_id = format!("lc-{counter}-{}", &name); - let group_id = format!("lg-{counter}-{}", &name); + let client_id = format!("lc-{counter}-{name}"); + let group_id = format!("lg-{counter}-{name}"); let patch = profile.patch(&shared.world); let wanted: HashSet = patch .iter() @@ -1902,14 +1902,11 @@ impl Conn { .push(format!("transformError: {}", preview(&body.to_string()))); self.changed.notify_waiters(); } - "pushResponse" => { - if body.get("error").is_some() { - self.errors - .lock() - .unwrap() - .push(format!("pushResponse: {}", preview(&body.to_string()))); - } - } + "pushResponse" if body.get("error").is_some() => self + .errors + .lock() + .unwrap() + .push(format!("pushResponse: {}", preview(&body.to_string()))), _ => {} } } diff --git a/src/client/sampler.rs b/src/client/sampler.rs index 1711826..7bfe484 100644 --- a/src/client/sampler.rs +++ b/src/client/sampler.rs @@ -58,7 +58,7 @@ fn run(state: Arc, interval: Duration) { stats.publish_thread_core_cpu(attributed.thread_cores); stats.publish_thread_core(attributed.placement); stats.publish_thread_migrations(attributed.migrations); - if ticks % 6 == 0 { + if ticks.is_multiple_of(6) { let (message, fields) = stats.summary_line(&previous, last_summary.elapsed()); crate::log::event(Level::Info, message, fields); previous = stats.snapshot(); diff --git a/src/client/schema.rs b/src/client/schema.rs index 1183fea..febb443 100644 --- a/src/client/schema.rs +++ b/src/client/schema.rs @@ -47,7 +47,7 @@ pub fn client_type(declared: &ValueType) -> &'static str { pub fn mismatches(catalog: &Catalog, schema: &ClientSchema) -> Vec { let mut found = Vec::new(); let mut tables: Vec<_> = schema.tables.iter().collect(); - tables.sort_by(|(a, _), (b, _)| a.cmp(b)); + tables.sort_by_key(|(name, _)| *name); for (name, spec) in tables { let Some(table) = catalog.table(name) else { found.push(format!( @@ -57,7 +57,7 @@ pub fn mismatches(catalog: &Catalog, schema: &ClientSchema) -> Vec { continue; }; let mut columns: Vec<_> = spec.columns.iter().collect(); - columns.sort_by(|(a, _), (b, _)| a.cmp(b)); + columns.sort_by_key(|(name, _)| *name); for (column, declared) in columns { let Some(known) = table.column(column) else { found.push(format!( diff --git a/src/client/setup.rs b/src/client/setup.rs index f5e58f8..c0eace6 100644 --- a/src/client/setup.rs +++ b/src/client/setup.rs @@ -22,7 +22,7 @@ pub async fn ensure(master_dsn: &str, config: &Config) -> Result<(), String> { let (mut client, connection) = tokio_postgres::connect(master_dsn, NoTls) .await .map_err(|error| format!("connecting to primary database: {error}"))?; - let connection = tokio::spawn(async move { connection.await }); + let connection = tokio::spawn(connection); let result = async { let tx = client .transaction() @@ -652,7 +652,7 @@ mod tests { ensure(&dsn, &config).await.unwrap(); let (client, connection) = tokio_postgres::connect(&dsn, NoTls).await.unwrap(); - let connection = tokio::spawn(async move { connection.await }); + let connection = tokio::spawn(connection); client .batch_execute(&format!( "UPDATE {}.\"publishedSchema\" SET current = json_build_object( diff --git a/src/client/warm.rs b/src/client/warm.rs index bf19deb..90ac170 100644 --- a/src/client/warm.rs +++ b/src/client/warm.rs @@ -93,7 +93,7 @@ impl WarmStart { return Vec::new(); } }; - shapes.sort_by(|a, b| b.last.cmp(&a.last)); + shapes.sort_by_key(|shape| std::cmp::Reverse(shape.last)); shapes.truncate(self.capacity); let mut seen = self.lock(); for shape in &shapes { @@ -187,7 +187,7 @@ impl WarmStart { let started = Instant::now(); let total = shapes.len(); let mut outcome = Replayed::default(); - let mut work = stream::iter(shapes.into_iter()) + let mut work = stream::iter(shapes) .map(|shape| async move { let ast: Ast = match serde_json::from_str(shape.ast.get()) { Ok(ast) => ast, diff --git a/src/stats.rs b/src/stats.rs index 4a56823..8cd5fae 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -1139,7 +1139,7 @@ impl Stats { "name": name, "table": entry.table, "rows": entry.rows, - "percent_of_limit": if limit == 0 { 0 } else { entry.rows.saturating_mul(100) / limit }, + "percent_of_limit": entry.rows.saturating_mul(100).checked_div(limit).unwrap_or(0), "count": entry.count, "last_ms": entry.last_ms, }) @@ -1842,7 +1842,7 @@ mod tests { assert_eq!(bucket_of(5), 10); assert_eq!(bucket_of(7), 12); assert_eq!(bucket_of(8), 13); - let used = |index: usize| (index - 1) / 4 >= 2 || (index - 1) % 4 == 0; + let used = |index: usize| (index - 1) / 4 >= 2 || (index - 1).is_multiple_of(4); let mut previous = 0; for index in (1..BUCKETS).filter(|index| used(*index)) { assert!(lower_bound(index) > previous, "at {index}"); From c5f7b48498402dee34a8868135983e47f066d970 Mon Sep 17 00:00:00 2001 From: A Guhan Date: Fri, 23 Oct 2026 17:14:13 +0530 Subject: [PATCH 3/3] Fix image CI startup configuration --- .github/workflows/ci-pr.yml | 8 +------- scripts/smoke.mjs | 5 +---- 2 files changed, 2 insertions(+), 11 deletions(-) diff --git a/.github/workflows/ci-pr.yml b/.github/workflows/ci-pr.yml index c570de2..1d4c0c6 100644 --- a/.github/workflows/ci-pr.yml +++ b/.github/workflows/ci-pr.yml @@ -178,18 +178,12 @@ jobs: otel/opentelemetry-collector-contrib:0.146.1 until docker exec pg pg_isready -U postgres; do sleep 1; done node scripts/smoke.mjs --prepare - # The server hears of schema changes through the schema-change - # event trigger and refuses to start without it: install the - # stack for app `xyne`, shard 0, over the publication the server uses. - docker exec pg psql -U postgres -d postgres -v ON_ERROR_STOP=1 \ - -c 'CREATE PUBLICATION xyne_sync_pub FOR ALL TABLES' - docker exec -i pg psql -U postgres -d postgres -v ON_ERROR_STOP=1 -q \ - < scripts/sql/ddl-triggers.sql - name: Start the server from the image run: | docker run -d --name server --network host \ -e STREAMGRES_PG_DSN=postgresql://postgres:postgres@localhost:5432/postgres \ + -e STREAMGRES_MASTER_PG_DSN=postgresql://postgres:postgres@localhost:5432/postgres \ -e STREAMGRES_BASE_PATH=/sync \ -e STREAMGRES_APP_ID=xyne \ -e STREAMGRES_QUERY_URL=http://127.0.0.1:4849/query \ diff --git a/scripts/smoke.mjs b/scripts/smoke.mjs index 497659c..b3cb86c 100755 --- a/scripts/smoke.mjs +++ b/scripts/smoke.mjs @@ -33,7 +33,6 @@ import { createServer } from 'node:http'; const PG = process.env.SMOKE_PG ?? 'postgresql://postgres:postgres@localhost:5432/postgres'; const GATEWAY = process.env.SMOKE_GATEWAY ?? 'ws://localhost:4848/sync'; -const SHARD_SCHEMA = `${process.env.STREAMGRES_APP_ID ?? 'xyne'}_${process.env.STREAMGRES_SHARD ?? '0'}`; const METRICS_PREFIX = process.env.STREAMGRES_METRICS_PREFIX ?? 'xyne_sync'; const metric = (name) => `${METRICS_PREFIX}_${name}`; const HTTP = process.env.SMOKE_HTTP ?? 'http://localhost:4848'; @@ -46,9 +45,7 @@ const fail = (why) => { console.error('FAIL:', why); process.exit(1); }; const sql = (text) => execFileSync('psql', [PG, '-v', 'ON_ERROR_STOP=1', '-Atc', text]).toString().trim(); if (process.argv.includes('--prepare')) { - sql(`CREATE SCHEMA IF NOT EXISTS ${SHARD_SCHEMA}; - CREATE TABLE IF NOT EXISTS ${SHARD_SCHEMA}.clients ("clientGroupID" text, "clientID" text, "lastMutationID" bigint, "userID" text, PRIMARY KEY ("clientGroupID", "clientID")); - DROP TABLE IF EXISTS smoke_items; + sql(`DROP TABLE IF EXISTS smoke_items; CREATE TABLE smoke_items (id text PRIMARY KEY, status text NOT NULL, points integer NOT NULL, meta jsonb, tag jsonb, score jsonb, loose json, bucket text, "createdAt" timestamptz NOT NULL DEFAULT now()); INSERT INTO smoke_items (id, status, points, meta, tag, score, loose, bucket) SELECT 'seed-' || n, 'OPEN', n, jsonb_build_object('n', n), to_jsonb('t' || (n % 5)), (CASE WHEN n % 5 = 0 THEN '1.50' ELSE '2' END)::jsonb, (CASE WHEN n % 5 = 0 THEN '{ "k":1.50, "a":[1e1], "k": 1.50 }' ELSE '{"k":2}' END)::json, 'b' || (n % 10) FROM generate_series(1, ${SEEDED}) n; INSERT INTO smoke_items (id, status, points) VALUES ('closed-1', 'DONE', 0);`);