mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-18 08:03:11 +00:00
fix: keep applied migrations and db settings in data table copies (#11753)
* fix: keep applied migrations and db settings in data table copies Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: scope copied db settings per transaction and send timestamps as iso Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: copy migrations and settings only into a fresh import target Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ci: install the latest pg_dump for backend integration tests Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ci: enable pgdg so the latest pg_dump is installed Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
88c3cc0ced
commit
694ed7a0eb
@@ -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<bool> {
|
||||
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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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<String>,
|
||||
datatables: Vec<String>,
|
||||
versions: Vec<i64>,
|
||||
installed_at: Vec<String>,
|
||||
}
|
||||
|
||||
/// 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<tokio_postgres::Client> {
|
||||
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<AppliedMigrations> {
|
||||
let read_err = |e: tokio_postgres::Error| {
|
||||
Error::internal_err(format!(
|
||||
"Failed to read _wm_migrations: {}",
|
||||
pg_error_message(&e)
|
||||
))
|
||||
};
|
||||
let schema: Option<String> = 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, _))
|
||||
|
||||
@@ -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<String> = 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::<Vec<_>>()
|
||||
.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?;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user