diff --git a/.github/workflows/backend-test.yml b/.github/workflows/backend-test.yml index 73f955131e..7a045ad70a 100644 --- a/.github/workflows/backend-test.yml +++ b/.github/workflows/backend-test.yml @@ -83,9 +83,13 @@ jobs: chmod +x "$HOME/.local/bin/wmill" echo "$HOME/.local/bin" >> $GITHUB_PATH working-directory: / - - name: Install PowerShell, mold and clang + - name: Install PowerShell, mold, clang and the latest pg_dump + # PGDG's `postgresql-client` is its newest major, Ubuntu's is not: the `postgres` service + # image floats, and pg_dump refuses a server newer than itself. run: | - sudo apt-get update && sudo apt-get install -y powershell mold clang libcurl4-openssl-dev + sudo /usr/share/postgresql-common/pgdg/apt.postgresql.org.sh -y + sudo apt-get update && sudo apt-get install -y powershell mold clang libcurl4-openssl-dev postgresql-client + pg_dump --version working-directory: / - uses: actions-rust-lang/setup-rust-toolchain@v1 with: diff --git a/backend/tests/datatable_fork_migrations.rs b/backend/tests/datatable_fork_migrations.rs new file mode 100644 index 0000000000..43f43c1cd9 --- /dev/null +++ b/backend/tests/datatable_fork_migrations.rs @@ -0,0 +1,135 @@ +//! A fork's copy of a data table must report the migrations its schema already holds as applied. +//! +//! A schema-only copy restores `_wm_migrations` without its rows, and no copy carries the source's +//! `ALTER DATABASE ... SET search_path`, through which the bookkeeping table is found. Either one +//! alone makes the fork list every applied migration as pending, and running them fails on +//! objects that already exist. The source's `DateStyle` differs from the copy's while the rows +//! are written, so they must not travel in its text form. + +use serde_json::{json, Value}; +use sqlx::{Pool, Postgres}; + +use windmill_test_utils::*; + +const ROLE: &str = "wm_dtfork_test_role"; +const ROLE_PASSWORD: &str = "wm_dtfork_test_pwd"; + +fn authed(b: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + b.header("Authorization", "Bearer DTMIG_ADMIN_TOKEN") +} + +#[sqlx::test(fixtures("datatable_migrations_grants"))] +async fn a_forked_copy_keeps_its_applied_migrations(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let opts = (*db.connect_options()).clone(); + // Databases are cluster-wide, so both are named for this run alone. + let suffix = format!( + "{:x}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)? + .as_nanos() + ); + let source = format!("wm_dtfork_src_{suffix}"); + let fork_id = format!("wm-fork-dtfork-{suffix}"); + let copy = format!("wm_fork_dtfork_{suffix}__main"); + + sqlx::query(&format!( + "DO $$ BEGIN \ + CREATE ROLE {ROLE} LOGIN CREATEDB PASSWORD '{ROLE_PASSWORD}'; \ + EXCEPTION WHEN duplicate_object OR unique_violation THEN NULL; \ + END $$" + )) + .execute(&db) + .await?; + sqlx::query(&format!("CREATE DATABASE {source} OWNER {ROLE}")) + .execute(&db) + .await?; + + let result = async { + let source_pool = sqlx::PgPool::connect_with(opts.clone().database(&source)).await?; + sqlx::raw_sql(&format!( + "SET ROLE {ROLE}; \ + CREATE SCHEMA app; \ + CREATE TABLE app.foo (id int); \ + CREATE TABLE app._wm_migrations (datatable TEXT NOT NULL, version BIGINT NOT NULL, \ + installed_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (datatable, version)); \ + INSERT INTO app._wm_migrations (datatable, version, installed_at) \ + VALUES ('main', 1, '2026-01-31 10:00:00+00'); \ + ALTER DATABASE {source} SET search_path = app; \ + ALTER DATABASE {source} SET datestyle = 'SQL, DMY';" + )) + .execute(&source_pool) + .await?; + source_pool.close().await; + + sqlx::query( + "INSERT INTO resource (workspace_id, path, value, resource_type, created_by) \ + VALUES ('dtmig-ws', 'u/dtmig-admin/pg', $1, 'postgresql', 'dtmig-admin')", + ) + .bind(json!({ + "host": opts.get_host(), + "port": opts.get_port(), + "dbname": source, + "user": ROLE, + "password": ROLE_PASSWORD, + "sslmode": "disable", + })) + .execute(&db) + .await?; + sqlx::query( + "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up) \ + VALUES ('dtmig-ws', 'main', 1, 'create_foo', 'CREATE TABLE foo (id int)')", + ) + .execute(&db) + .await?; + + let server = ApiServer::start(db.clone()).await?; + let base = format!("http://localhost:{}/api", server.addr.port()); + let client = reqwest::Client::new(); + + let resp = authed(client.post(format!("{base}/w/dtmig-ws/workspaces/create_fork"))) + .json(&json!({ + "id": fork_id, + "name": "fork", + "forked_datatables": [ + {"name": "main", "new_dbname": copy, "fork_behavior": "schema_only"} + ], + })) + .send() + .await?; + // `ensure!` rather than `assert!`: a panic would skip dropping the databases below. + let fork_status = resp.status(); + anyhow::ensure!(fork_status == 200, "{}", resp.text().await?); + + let status: Value = authed(client.get(format!( + "{base}/w/{fork_id}/workspaces/datatable_migrations_status/main" + ))) + .send() + .await? + .json() + .await?; + anyhow::ensure!(status["migrations"][0]["status"] == "ran", "{status}"); + + let copied: Vec = sqlx::query_scalar( + "SELECT unnest(setconfig) FROM pg_db_role_setting \ + WHERE setrole = 0 AND setdatabase = (SELECT oid FROM pg_database WHERE datname = $1) \ + ORDER BY 1", + ) + .bind(©) + .fetch_all(&db) + .await?; + anyhow::ensure!( + copied == ["DateStyle=SQL, DMY", "search_path=app"], + "{copied:?}" + ); + anyhow::Ok(()) + } + .await; + + for name in [©, &source] { + sqlx::query(&format!("DROP DATABASE IF EXISTS {name} WITH (FORCE)")) + .execute(&db) + .await?; + } + result +} diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index 0acba27c41..66e9663154 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs @@ -810,7 +810,7 @@ async fn a_data_table_under_roles_is_cloned_only_with_its_grants( // The copy an admin asks for goes ahead in an edition that replays grants, and is refused // before any database exists in one that does not. The fixture's database does not exist, so - // the dump fails either way — and leaves neither a database nor a fork behind. + // reading it fails either way — and leaves neither a database nor a fork behind. let resp = fork( "SECRET_TOKEN", json!({"name": "main", "new_dbname": target, "fork_behavior": "schema_only"}), @@ -820,7 +820,7 @@ async fn a_data_table_under_roles_is_cloned_only_with_its_grants( assert_ne!(resp.status(), 200); let body = resp.text().await?; #[cfg(all(feature = "private", feature = "enterprise"))] - assert!(body.contains("pg_dump"), "{body}"); + assert!(body.contains("\"dt_main\" does not exist"), "{body}"); #[cfg(not(all(feature = "private", feature = "enterprise")))] assert!(body.contains("Enterprise Edition"), "{body}"); assert!(!database_exists(&db, target).await?); diff --git a/backend/windmill-api-workspaces/src/datatable_clone.rs b/backend/windmill-api-workspaces/src/datatable_clone.rs index 6083b7b619..5818419d10 100644 --- a/backend/windmill-api-workspaces/src/datatable_clone.rs +++ b/backend/windmill-api-workspaces/src/datatable_clone.rs @@ -28,10 +28,13 @@ use windmill_common::workspaces::{ use windmill_common::{PgDatabase, DB}; use crate::datatable_acl::connect_with_notices; +use crate::datatable_migrations::{ + lock_migration_runs_on, read_applied_migrations, write_applied_migrations, AppliedMigrations, +}; use crate::datatable_permissions::ensure_reaches_governing_datatable; use crate::workspaces::{ - create_database_on_server, ensure_datatable_is_clonable, pg_dump_database, pg_import_dump, - DumpFile, PgDumpOptions, + copy_database_settings, create_database_on_server, ensure_datatable_is_clonable, + pg_dump_database, pg_import_dump, DumpFile, PgDumpOptions, }; /// A database this request created and filled for one data table of the fork. @@ -336,8 +339,8 @@ async fn drop_copy(db: &DB, source_database: &DataTableDatabase, dbname: &str) - Ok(()) } DataTableCatalogResourceType::Postgresql => { - // On a server of the workspace's own, where a resource edited meanwhile can already name it - // and nothing locks such an edit: dropping it could take someone's data. + // On a server of the workspace's own, where a resource edited meanwhile can already name it + // and nothing locks such an edit: dropping it could take someone's data. Err(Error::BadRequest( "kept on its PostgreSQL server, where it may already be in use; drop it there once \ it is not, before forking the same data table under this id again" @@ -392,20 +395,25 @@ async fn make_copy( (server, Some(snapshot)) }; + let schema_only = request.behavior == DataTableForkBehavior::SchemaOnly; + let migrations_lock = if schema_only { + Some(lock_migration_runs_on(db, &server, request.name).await?) + } else { + None + }; // Dumped before the database exists, so a source that cannot be read leaves nothing behind. // Ownership never carries over: the restore runs as the target's connection user. Grants do, // except on a cluster Windmill manages, which plants its own and where the replay below is what // reproduces the roles'. let dump = pg_dump_database( &server, - PgDumpOptions { - schema_only: request.behavior == DataTableForkBehavior::SchemaOnly, - no_owner: true, - no_acl: managed, - ..Default::default() - }, + PgDumpOptions { schema_only, no_owner: true, no_acl: managed, ..Default::default() }, ) .await?; + let applied = match migrations_lock { + Some(client) => Some(read_applied_migrations(&client).await?), + None => None, + }; match source_database.resource_type { DataTableCatalogResourceType::Instance => { @@ -444,6 +452,7 @@ async fn make_copy( &server, &target, &dump, + applied.as_ref(), ) .await; match filled { @@ -484,8 +493,12 @@ async fn fill( source: &PgDatabase, target: &PgDatabase, dump: &DumpFile, + applied: Option<&AppliedMigrations>, ) -> Result { pg_import_dump(target, dump).await?; + if let Some(applied) = applied { + write_applied_migrations(db, target, applied).await?; + } // Held until the replay commits: a role renamed or dropped meanwhile would change what the // replay names, and a settings save could move the source onto another database or put it under @@ -554,6 +567,7 @@ async fn fill( })?; } tx.commit().await?; + copy_database_settings(db, source, target).await?; Ok(replayed) } diff --git a/backend/windmill-api-workspaces/src/datatable_migrations.rs b/backend/windmill-api-workspaces/src/datatable_migrations.rs index 21e460359e..907861c2b4 100644 --- a/backend/windmill-api-workspaces/src/datatable_migrations.rs +++ b/backend/windmill-api-workspaces/src/datatable_migrations.rs @@ -509,6 +509,127 @@ async fn hold_migration_lock( } } +/// The `_wm_migrations` rows of a database a schema-only copy is being made of, read with +/// [`lock_migration_runs_on`] held so they describe the schema the dump captured. +pub(crate) struct AppliedMigrations { + /// `None` when the source has no `_wm_migrations`. + schema: Option, + datatables: Vec, + versions: Vec, + installed_at: Vec, +} + +/// Hold the migration-run lock on database `pg`, for a copy whose schema has to agree with the +/// `_wm_migrations` rows it carries. The lock is per database, whichever data table names it. +pub(crate) async fn lock_migration_runs_on( + db: &DB, + pg: &PgDatabase, + label: &str, +) -> Result { + let (client, connection) = pg.connect(Some(db)).await?; + tokio::spawn(async move { + if let Err(e) = connection.await { + tracing::error!("Datatable connection error: {}", e); + } + }); + hold_migration_lock(&client, label, MIGRATION_LOCK_WAIT_SECS).await?; + Ok(client) +} + +/// Read every `_wm_migrations` row of the database `client` is connected to, along with the +/// schema the table is in: the copy is written to by a connection whose `search_path` need not +/// resolve the way this one does. +pub(crate) async fn read_applied_migrations( + client: &tokio_postgres::Client, +) -> Result { + let read_err = |e: tokio_postgres::Error| { + Error::internal_err(format!( + "Failed to read _wm_migrations: {}", + pg_error_message(&e) + )) + }; + let schema: Option = client + .query_one( + "SELECT n.nspname::text FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace \ + RIGHT JOIN (SELECT to_regclass('_wm_migrations')::oid AS oid) t ON c.oid = t.oid", + &[], + ) + .await + .map_err(read_err)? + .get(0); + let mut applied = + AppliedMigrations { schema, datatables: vec![], versions: vec![], installed_at: vec![] }; + let Some(schema) = applied.schema.as_deref() else { + return Ok(applied); + }; + let rows = client + .query( + &format!( + // ISO 8601 in UTC: `::text` follows the session's `DateStyle`, which the copy's + // session need not parse the same way. + "SELECT datatable, version, \ + to_char(installed_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') \ + FROM {}._wm_migrations", + render_db_quoted_identifier(schema, DbType::Postgresql) + ), + &[], + ) + .await + .map_err(read_err)?; + for row in rows { + applied.datatables.push(row.get(0)); + applied.versions.push(row.get(1)); + applied.installed_at.push(row.get(2)); + } + Ok(applied) +} + +/// Write the rows [`read_applied_migrations`] read into a copy restored from a schema-only dump, +/// which brings `_wm_migrations` over empty: without them the copy reports every migration its +/// schema already holds as pending, and running them fails on objects that exist. +pub(crate) async fn write_applied_migrations( + db: &DB, + target: &PgDatabase, + applied: &AppliedMigrations, +) -> Result<()> { + let Some(schema) = applied.schema.as_deref() else { + return Ok(()); + }; + if applied.versions.is_empty() { + return Ok(()); + } + let (client, connection) = target.connect(Some(db)).await?; + tokio::spawn(async move { + if let Err(e) = connection.await { + tracing::error!("Datatable connection error: {}", e); + } + }); + client + .execute( + &format!( + "INSERT INTO {}._wm_migrations (datatable, version, installed_at) \ + SELECT d, v, i::timestamp AT TIME ZONE 'UTC' \ + FROM unnest($1::text[], $2::bigint[], $3::text[]) AS t(d, v, i) \ + ON CONFLICT DO NOTHING", + render_db_quoted_identifier(schema, DbType::Postgresql) + ), + &[ + &applied.datatables, + &applied.versions, + &applied.installed_at, + ], + ) + .await + .map_err(|e| { + Error::internal_err(format!( + "Failed to copy _wm_migrations into '{}': {}", + target.dbname, + pg_error_message(&e) + )) + })?; + Ok(()) +} + /// Whether `e` is the lock wait running out, as opposed to the database being unreachable. fn is_migration_lock_busy(e: &Error) -> bool { matches!(e, Error::Generic(StatusCode::CONFLICT, _)) diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index bdb3b59e6e..db9c070ee9 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -3594,6 +3594,97 @@ pub(crate) async fn pg_import_dump(target_db: &PgDatabase, dump_file: &DumpFile) Ok(()) } +/// Carry the source database's own settings (`ALTER DATABASE ... SET`) over to a copy. `pg_dump` +/// only emits them with `--create`, so a copy otherwise resolves unqualified names, its +/// `_wm_migrations` included, through the server's default `search_path` rather than the source's. +/// A setting the copy's connection may not set is skipped with a warning rather than failing it. +/// Called once nothing else writes to the copy, which a copied read-only default would refuse. +pub(crate) async fn copy_database_settings( + db: &DB, + source: &PgDatabase, + target: &PgDatabase, +) -> Result<()> { + let (source_client, connection) = source.connect(Some(db)).await?; + tokio::spawn(async move { + if let Err(e) = connection.await { + tracing::error!("Datatable connection error: {}", e); + } + }); + let settings: Vec = source_client + .query( + "SELECT unnest(setconfig) FROM pg_db_role_setting WHERE setrole = 0 \ + AND setdatabase = (SELECT oid FROM pg_database WHERE datname = current_database())", + &[], + ) + .await + .map_err(|e| { + Error::internal_err(format!( + "Failed to read the settings of database '{}': {}", + source.dbname, + pg_error_message(&e) + )) + })? + .iter() + .map(|row| row.get(0)) + .collect(); + drop(source_client); + if settings.is_empty() { + return Ok(()); + } + + let (mut client, connection) = target.connect(Some(db)).await?; + tokio::spawn(async move { + if let Err(e) = connection.await { + tracing::error!("Datatable connection error: {}", e); + } + }); + let quoted_db = render_db_quoted_identifier(&target.dbname, DbType::Postgresql); + for setting in settings { + let Some((name, value)) = setting.split_once('=') else { + continue; + }; + if name.is_empty() + || !name + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '.') + { + tracing::warn!( + "Not copying database setting '{name}' to '{}'", + target.dbname + ); + continue; + } + let quoted_name = name + .split('.') + .map(|part| render_db_quoted_identifier(part, DbType::Postgresql)) + .collect::>() + .join("."); + // `FROM CURRENT` takes the value in the stored text form, which `SET name TO value` would + // need re-quoted per setting type. Set transaction-locally: a session-wide `role` or + // `default_transaction_read_only` would make every later `ALTER DATABASE` fail. + let applied = async { + let tx = client.transaction().await?; + tx.execute("SELECT set_config($1, $2, true)", &[&name, &value]) + .await?; + tx.batch_execute(&format!( + "ALTER DATABASE {quoted_db} SET {quoted_name} FROM CURRENT" + )) + .await?; + tx.commit().await + } + .await; + if let Err(e) = applied { + tracing::warn!( + "Could not copy setting '{name}' of database '{}' to '{}': {}", + source.dbname, + target.dbname, + pg_error_message(&e) + ); + } + } + Ok(()) +} + #[derive(Deserialize)] struct CreatePgDatabaseRequest { /// The datatable source to determine connection info: 'datatable://name' or '$res:path' @@ -3857,12 +3948,33 @@ async fn import_pg_database( // the ACLs are user intent (`REVOKE ... FROM PUBLIC`) and dropping them widens access. let no_acl = target_kind.is_some() || source_kind.is_some(); + // Only a database made to be a copy takes on the source's bookkeeping and settings: without an + // override the target is a data table's own live database, whose settings are its owner's. + let fresh_copy = req.target_dbname_override.is_some(); + let migrations_lock = if schema_only && fresh_copy { + Some( + crate::datatable_migrations::lock_migration_runs_on(&db, &source_pg, &req.source) + .await?, + ) + } else { + None + }; let dump_file = pg_dump_database( &source_pg, PgDumpOptions { schema_only, no_owner: true, no_acl, ..Default::default() }, ) .await?; + let applied = match migrations_lock { + Some(client) => Some(crate::datatable_migrations::read_applied_migrations(&client).await?), + None => None, + }; pg_import_dump(&target_pg, &dump_file).await?; + if let Some(applied) = applied { + crate::datatable_migrations::write_applied_migrations(&db, &target_pg, &applied).await?; + } + if fresh_copy { + copy_database_settings(&db, &source_pg, &target_pg).await?; + } if let Some(tx) = fork_lock { tx.commit().await?; }