mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-25 16:02:11 +00:00
feat(datatables): add a down migration from the migration viewer (#10812)
* feat(datatables): add a down migration from the migration viewer Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ha889ogajX9jbqmwTaF8kD * fix(datatables): refuse an empty down migration and keep the saved one visible Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ha889ogajX9jbqmwTaF8kD * refactor(datatables): use unifiedSize on the new buttons and fix the lock comment Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ha889ogajX9jbqmwTaF8kD * fix(datatables): make the add-down exemption atomic against concurrent additions Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ha889ogajX9jbqmwTaF8kD * fix(datatables): re-test the whole observed row when an upsert skips the lock Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ha889ogajX9jbqmwTaF8kD * fix(datatables): re-test the observed row even when none was read, and sync the down draft on the leading change Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Ha889ogajX9jbqmwTaF8kD --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
GitHub
+6
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (workspace_id, datatable, timestamp) DO UPDATE SET name = EXCLUDED.name, code_up = EXCLUDED.code_up, code_down = EXCLUDED.code_down",
|
||||
"query": "INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (workspace_id, datatable, timestamp) DO UPDATE SET name = EXCLUDED.name, code_up = EXCLUDED.code_up, code_down = EXCLUDED.code_down WHERE NOT $7 OR (datatable_migrations.name = $8::text AND datatable_migrations.code_up = $9::text AND datatable_migrations.code_down IS NOT DISTINCT FROM $10::text)",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
@@ -10,10 +10,14 @@
|
||||
"Int8",
|
||||
"Varchar",
|
||||
"Text",
|
||||
"Text",
|
||||
"Bool",
|
||||
"Text",
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "c5639d48c92d6863f16f97de91eae590d0ed2f4b7831956bb429a47c478f7c1f"
|
||||
"hash": "2b93351b9c2c91272c3f69c9c8eba192d2e4d087c7d7db8d3b257bb864c8265d"
|
||||
}
|
||||
@@ -0,0 +1,144 @@
|
||||
//! A migration that has already run may still gain the down it was missing.
|
||||
//!
|
||||
//! Rewriting an applied migration is refused because its `_wm_migrations` record
|
||||
//! would no longer match its SQL. Filling in an absent `code_down` is the one
|
||||
//! edit that keeps that record true — and the only way to make an already-run
|
||||
//! migration revertable — so it must stay allowed while every other edit stays
|
||||
//! refused.
|
||||
|
||||
use serde_json::json;
|
||||
use sqlx::{Pool, Postgres};
|
||||
|
||||
use windmill_test_utils::*;
|
||||
|
||||
const ROLE: &str = "wm_dtmig_down_role";
|
||||
const ROLE_PASSWORD: &str = "wm_dtmig_down_pwd";
|
||||
const VERSION: i64 = 20260101000000;
|
||||
const CODE_UP: &str = "CREATE TABLE widgets (id int);";
|
||||
const CODE_DOWN: &str = "DROP TABLE widgets;";
|
||||
|
||||
fn authed(b: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
|
||||
b.header("Authorization", "Bearer DTMIG_ADMIN_TOKEN")
|
||||
}
|
||||
|
||||
/// Point the fixture's data table at this test's own database and put it in the
|
||||
/// state that matters: one migration defined without a down, recorded as applied.
|
||||
async fn setup_applied_migration_without_down(db: &Pool<Postgres>) -> anyhow::Result<()> {
|
||||
let opts = (*db.connect_options()).clone();
|
||||
let dbname = opts.get_database().expect("test database name").to_string();
|
||||
|
||||
sqlx::query(&format!(
|
||||
// Roles are cluster objects, so a leftover role or a parallel test
|
||||
// session reaching here at the same time must not fail the setup.
|
||||
"DO $$ BEGIN \
|
||||
CREATE ROLE {ROLE} LOGIN PASSWORD '{ROLE_PASSWORD}'; \
|
||||
EXCEPTION WHEN duplicate_object OR unique_violation THEN NULL; \
|
||||
END $$"
|
||||
))
|
||||
.execute(db)
|
||||
.await?;
|
||||
sqlx::raw_sql(&format!(
|
||||
"GRANT CONNECT ON DATABASE \"{dbname}\" TO {ROLE}; \
|
||||
GRANT USAGE ON SCHEMA public TO {ROLE}; \
|
||||
CREATE TABLE _wm_migrations ( \
|
||||
datatable TEXT NOT NULL, \
|
||||
version BIGINT NOT NULL, \
|
||||
installed_at TIMESTAMPTZ NOT NULL DEFAULT now(), \
|
||||
PRIMARY KEY (datatable, version)); \
|
||||
GRANT SELECT ON _wm_migrations TO {ROLE}; \
|
||||
INSERT INTO _wm_migrations (datatable, version) VALUES ('main', {VERSION});"
|
||||
))
|
||||
.execute(db)
|
||||
.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": dbname,
|
||||
"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_widgets', $2)",
|
||||
)
|
||||
.bind(VERSION)
|
||||
.bind(CODE_UP)
|
||||
.execute(db)
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("datatable_migrations_grants"))]
|
||||
async fn test_add_down_to_an_applied_migration(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
initialize_tracing().await;
|
||||
setup_applied_migration_without_down(&db).await?;
|
||||
|
||||
let server = ApiServer::start(db.clone()).await?;
|
||||
let port = server.addr.port();
|
||||
let url = format!(
|
||||
"http://localhost:{port}/api/w/dtmig-ws/workspaces/upsert_datatable_migration/main"
|
||||
);
|
||||
let upsert = |code_up: &str, code_down: &str| {
|
||||
authed(reqwest::Client::new().post(&url)).json(&json!({
|
||||
"timestamp": VERSION,
|
||||
"name": "create_widgets",
|
||||
"code_up": code_up,
|
||||
"code_down": code_down,
|
||||
}))
|
||||
};
|
||||
|
||||
let resp = upsert(CODE_UP, CODE_DOWN).send().await?;
|
||||
let status = resp.status();
|
||||
let body = resp.text().await?;
|
||||
assert_eq!(
|
||||
status, 200,
|
||||
"adding a missing down to an applied migration should be allowed: {body}"
|
||||
);
|
||||
let stored = sqlx::query_scalar::<_, Option<String>>(
|
||||
"SELECT code_down FROM datatable_migrations \
|
||||
WHERE workspace_id = 'dtmig-ws' AND datatable = 'main' AND timestamp = $1",
|
||||
)
|
||||
.bind(VERSION)
|
||||
.fetch_one(&db)
|
||||
.await?;
|
||||
assert_eq!(stored.as_deref(), Some(CODE_DOWN));
|
||||
|
||||
// The up it ran is what the `_wm_migrations` record stands for: still frozen.
|
||||
let resp = upsert("CREATE TABLE gadgets (id int);", CODE_DOWN)
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(resp.status(), 400);
|
||||
assert!(
|
||||
resp.text().await?.contains("has already been applied"),
|
||||
"rewriting the up of an applied migration must stay refused"
|
||||
);
|
||||
|
||||
// And so is a down that has already been recorded — only the absent-to-present
|
||||
// step is exempt, in that direction alone.
|
||||
let resp = upsert(CODE_UP, "DROP TABLE widgets CASCADE;")
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(resp.status(), 400);
|
||||
|
||||
let resp = authed(reqwest::Client::new().post(&url))
|
||||
.json(&json!({ "timestamp": VERSION, "name": "create_widgets", "code_up": CODE_UP }))
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(
|
||||
resp.status(),
|
||||
400,
|
||||
"dropping the down of an applied migration must stay refused"
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -1244,7 +1244,9 @@ async fn upsert_datatable_migration(
|
||||
// its SQL, so a later `migrate up` would skip it and a rollback would run a
|
||||
// `down` that doesn't correspond to what was applied. Only an actual change
|
||||
// to an existing migration is guarded; unchanged re-pushes (e.g.
|
||||
// `wmill sync push`) always proceed.
|
||||
// `wmill sync push`) always proceed, and so does filling in a down migration
|
||||
// that was missing — the up that ran is untouched, and that is the only way
|
||||
// to make an already-applied migration revertable.
|
||||
let existing = sqlx::query!(
|
||||
"SELECT name, code_up, code_down FROM datatable_migrations \
|
||||
WHERE workspace_id = $1 AND datatable = $2 AND timestamp = $3",
|
||||
@@ -1254,16 +1256,26 @@ async fn upsert_datatable_migration(
|
||||
)
|
||||
.fetch_optional(&db)
|
||||
.await?;
|
||||
// When modifying an existing definition, hold the run-serialization lock
|
||||
// across the applied-check and the write below so an in-flight run can't
|
||||
// record a version for the SQL we're about to overwrite. Held until the end
|
||||
// of the handler (well past the write); a new/unchanged upsert needs no lock.
|
||||
let _run_lock = match existing {
|
||||
Some(existing)
|
||||
if !(existing.name == payload.name
|
||||
&& existing.code_up == payload.code_up
|
||||
&& existing.code_down == payload.code_down) =>
|
||||
{
|
||||
// When overwriting the SQL of an existing definition, hold the
|
||||
// run-serialization lock across the applied-check and the write below so an
|
||||
// in-flight run can't record a version for the SQL we're about to overwrite.
|
||||
// Held until the end of the handler (well past the write). The exempt
|
||||
// upserts need no lock: a new or unchanged one overwrites nothing, and one
|
||||
// that only adds a down leaves the `code_up` a concurrent run is recording
|
||||
// a version for untouched.
|
||||
let only_adds_down = existing.as_ref().is_some_and(|existing| {
|
||||
existing.name == payload.name
|
||||
&& existing.code_up == payload.code_up
|
||||
&& existing.code_down.is_none()
|
||||
&& payload.code_down.is_some()
|
||||
});
|
||||
let unchanged = existing.as_ref().is_some_and(|existing| {
|
||||
existing.name == payload.name
|
||||
&& existing.code_up == payload.code_up
|
||||
&& existing.code_down == payload.code_down
|
||||
});
|
||||
let _run_lock = match existing.as_ref() {
|
||||
Some(_) if !only_adds_down && !unchanged => {
|
||||
// Fail closed: if we can't lock/read the applied set (e.g. the
|
||||
// data-table database is temporarily unreachable), refuse the change
|
||||
// rather than risk overwriting a migration that has already run.
|
||||
@@ -1293,20 +1305,44 @@ async fn upsert_datatable_migration(
|
||||
_ => None,
|
||||
};
|
||||
|
||||
sqlx::query!(
|
||||
// An exempt upsert judged the row from an unlocked read and then writes
|
||||
// without the lock, so that whole read is re-tested here, where `ON CONFLICT
|
||||
// DO UPDATE` re-reads the row under a row lock. Otherwise a request working
|
||||
// from a stale definition silently reverts whatever changed in between — a
|
||||
// second addition's down, or a locked rewrite of the up whose new SQL a run
|
||||
// may already have recorded a version for. Reading no row at all is part of
|
||||
// the premise: the equality is NULL when `$8` is, so a version created in the
|
||||
// meantime is refused rather than overwritten.
|
||||
let recheck_observed = _run_lock.is_none();
|
||||
let written = sqlx::query!(
|
||||
"INSERT INTO datatable_migrations (workspace_id, datatable, timestamp, name, code_up, code_down) \
|
||||
VALUES ($1, $2, $3, $4, $5, $6) \
|
||||
ON CONFLICT (workspace_id, datatable, timestamp) DO UPDATE \
|
||||
SET name = EXCLUDED.name, code_up = EXCLUDED.code_up, code_down = EXCLUDED.code_down",
|
||||
SET name = EXCLUDED.name, code_up = EXCLUDED.code_up, code_down = EXCLUDED.code_down \
|
||||
WHERE NOT $7 \
|
||||
OR (datatable_migrations.name = $8::text \
|
||||
AND datatable_migrations.code_up = $9::text \
|
||||
AND datatable_migrations.code_down IS NOT DISTINCT FROM $10::text)",
|
||||
&w_id,
|
||||
&datatable_name,
|
||||
payload.timestamp,
|
||||
&payload.name,
|
||||
&payload.code_up,
|
||||
payload.code_down.as_deref(),
|
||||
recheck_observed,
|
||||
existing.as_ref().map(|e| e.name.as_str()),
|
||||
existing.as_ref().map(|e| e.code_up.as_str()),
|
||||
existing.as_ref().and_then(|e| e.code_down.as_deref()),
|
||||
)
|
||||
.execute(&db)
|
||||
.await?;
|
||||
if written.rows_affected() == 0 {
|
||||
return Err(Error::BadRequest(format!(
|
||||
"Migration {} on data table '{}' changed while this change was being saved. \
|
||||
Reload it before editing.",
|
||||
payload.timestamp, datatable_name
|
||||
)));
|
||||
}
|
||||
// The definition is written; runs may resume (audit/deploy metadata below
|
||||
// don't need the lock).
|
||||
drop(_run_lock);
|
||||
|
||||
@@ -67,12 +67,73 @@
|
||||
let deleteOpen = $state(false)
|
||||
let deleteTarget = $state<DatatableMigrationWithStatus | undefined>(undefined)
|
||||
|
||||
let addingDown = $state(false)
|
||||
let downDraft = $state('')
|
||||
let savingDown = $state(false)
|
||||
|
||||
function openView(m: DatatableMigrationWithStatus) {
|
||||
viewMigration = m
|
||||
viewTab = 'up'
|
||||
addingDown = false
|
||||
viewOpen = true
|
||||
}
|
||||
|
||||
const DOWN_TEMPLATE = 'BEGIN;\n\n-- Add your down migration here\n\nEND;'
|
||||
|
||||
function startAddDownMigration() {
|
||||
// Same transaction frame the new-migration modal starts from, so the down
|
||||
// applies atomically.
|
||||
downDraft = DOWN_TEMPLATE
|
||||
addingDown = true
|
||||
}
|
||||
|
||||
// Saving the bare template would attach a down that reverts nothing, and a
|
||||
// down can only be attached once — the upsert refuses to correct it later on a
|
||||
// migration that has already run. So require an actual statement inside the
|
||||
// transaction frame, not just comments.
|
||||
function hasStatement(sql: string): boolean {
|
||||
return (
|
||||
sql
|
||||
.replace(/--[^\n]*/g, '')
|
||||
.replace(/\/\*[\s\S]*?\*\//g, '')
|
||||
.replace(/\b(BEGIN|START\s+TRANSACTION|COMMIT|END)\b/gi, '')
|
||||
.replace(/;/g, '')
|
||||
.trim() !== ''
|
||||
)
|
||||
}
|
||||
|
||||
// Attach a down migration to one that has none. Only the down is sent back;
|
||||
// the name and up SQL are echoed unchanged so the upsert doesn't count as a
|
||||
// rewrite of a migration that may already be applied.
|
||||
async function saveDownMigration() {
|
||||
const m = viewMigration
|
||||
if (!m) return
|
||||
savingDown = true
|
||||
try {
|
||||
await WorkspaceService.upsertDatatableMigration({
|
||||
workspace,
|
||||
datatableName: datatable,
|
||||
requestBody: {
|
||||
timestamp: m.timestamp,
|
||||
name: m.name,
|
||||
code_up: m.code_up,
|
||||
code_down: downDraft
|
||||
}
|
||||
})
|
||||
sendUserToast('Down migration saved')
|
||||
addingDown = false
|
||||
// Show the saved down straight away: `loadMigrations` swallows its own
|
||||
// errors, so re-reading the migration from the refreshed list could hand
|
||||
// back the pre-save entry and bounce the tab to its empty state.
|
||||
viewMigration = { ...m, code_down: downDraft }
|
||||
await loadMigrations()
|
||||
} catch (e: any) {
|
||||
sendUserToast(`Failed to save down migration: ${e?.body ?? e?.message ?? e}`, true)
|
||||
} finally {
|
||||
savingDown = false
|
||||
}
|
||||
}
|
||||
|
||||
const confirmationModal = createAsyncConfirmationModal()
|
||||
|
||||
const hasPending = $derived(migrations.some((m) => m.status !== 'ran'))
|
||||
@@ -540,11 +601,47 @@
|
||||
<TabContent value="up" class="h-80 border rounded-md overflow-hidden">
|
||||
<SimpleEditor class="h-full" lang="sql" code={viewMigration?.code_up ?? ''} readOnly />
|
||||
</TabContent>
|
||||
<TabContent value="down" class="h-80 border rounded-md overflow-hidden">
|
||||
<TabContent value="down" class="flex flex-col gap-2 h-80">
|
||||
{#if viewMigration?.code_down}
|
||||
<SimpleEditor class="h-full" lang="sql" code={viewMigration.code_down} readOnly />
|
||||
<div class="grow min-h-0 border rounded-md overflow-hidden">
|
||||
<SimpleEditor class="h-full" lang="sql" code={viewMigration.code_down} readOnly />
|
||||
</div>
|
||||
{:else if addingDown}
|
||||
<div class="grow min-h-0 border rounded-md overflow-hidden">
|
||||
<SimpleEditor class="h-full" lang="sql" bind:code={downDraft} leadingChangeSync />
|
||||
</div>
|
||||
<div class="flex justify-end gap-2">
|
||||
<Button
|
||||
variant="default"
|
||||
unifiedSize="sm"
|
||||
disabled={savingDown}
|
||||
on:click={() => (addingDown = false)}
|
||||
>
|
||||
Cancel
|
||||
</Button>
|
||||
<Button
|
||||
variant="accent"
|
||||
unifiedSize="sm"
|
||||
disabled={savingDown || !hasStatement(downDraft)}
|
||||
on:click={saveDownMigration}
|
||||
>
|
||||
Save
|
||||
</Button>
|
||||
</div>
|
||||
{:else}
|
||||
<div class="p-6 text-center text-sm text-tertiary">No down migration</div>
|
||||
<div
|
||||
class="flex flex-col items-center justify-center gap-3 grow border rounded-md text-sm text-tertiary"
|
||||
>
|
||||
<span>No down migration</span>
|
||||
<Button
|
||||
variant="subtle"
|
||||
unifiedSize="xs"
|
||||
startIcon={{ icon: Plus }}
|
||||
on:click={startAddDownMigration}
|
||||
>
|
||||
Add a down migration
|
||||
</Button>
|
||||
</div>
|
||||
{/if}
|
||||
</TabContent>
|
||||
{/snippet}
|
||||
|
||||
Reference in New Issue
Block a user