diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 6d78150bbb..d8d84e2b6a 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -cbacb629a604f8d15b0b3a526266eda82df880a8 \ No newline at end of file +151be033ea2bf2f768f381a8cd3b888bdd366cf9 \ No newline at end of file diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index de170b48f4..386923a8ee 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -1678,6 +1678,8 @@ struct CustomInstanceDb { tag: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] used_by_workspaces: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + workspace_id: Option, } #[derive(Deserialize, Debug, Serialize, Default)] @@ -1914,12 +1916,31 @@ async fn setup_custom_instance_pg_database( Path(dbname): Path, Json(body): Json, ) -> JsonResult { + // Before anything is recorded: the status written below replaces the registry entry, and with it + // the workspace a fork copy is reserved for. + require_super_admin(&db, &authed).await?; + // A re-run keeps the fork reservation: without it, the workspace the copy was made for could no + // longer import into it or finish its fork. + let workspace_id = sqlx::query_scalar::<_, Option>( + "SELECT value->'databases'->$1->>'workspace_id' FROM global_settings + WHERE name = 'custom_instance_pg_databases'", + ) + .bind(&dbname) + .fetch_optional(&db) + .await? + .flatten(); let mut logs = CustomInstanceDbLogs::default(); let result = setup_custom_instance_pg_database_inner(authed, &db, &dbname, &mut logs).await; let success = result.is_ok(); let error = result.err().map(|e| e.to_string()); - let status = - CustomInstanceDb { logs, success, error, tag: body.tag, used_by_workspaces: vec![] }; + let status = CustomInstanceDb { + logs, + success, + error, + tag: body.tag, + used_by_workspaces: vec![], + workspace_id, + }; let status_json = serde_json::to_value(&status).map_err(to_anyhow)?; // Save that the database was setup successfully sqlx::query!( diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 395f2945d2..9f26862c1c 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -8762,6 +8762,11 @@ async fn create_workspace_fork( } let mut tx: Transaction<'_, Postgres> = db.begin().await?; + // Before the settings clone reads the parent's data tables: a pointer this fork ends up with + // must not be written after cleanup of the parent decided that nothing points at its copies. + // Also before the external cluster's lifecycle lock, which finalizing an external copy takes: + // fork cleanup takes the two in this order. + windmill_common::workspaces::lock_fork_datatables(&mut tx, &parent_workspace_id).await?; if nw.is_dev_workspace { // The checks above ran outside a transaction, so the parent's eligibility and the chain's @@ -8874,9 +8879,6 @@ async fn create_workspace_fork( // re-enables in the fork, with parent-conflict warnings on enable. clone_triggers_and_schedules(&mut tx, &parent_workspace_id, &forked_id).await?; - // Before the external cluster's lifecycle lock, which finalizing an external copy takes: fork - // cleanup takes the two in this order. - windmill_common::workspaces::lock_fork_datatables(&mut tx, &parent_workspace_id).await?; // Update forked datatable settings to point to new databases for fdt in &nw.forked_datatables { apply_forked_datatable(&db, &mut tx, &authed, &parent_workspace_id, &forked_id, fdt) diff --git a/backend/windmill-api-workspaces/src/workspaces_extra.rs b/backend/windmill-api-workspaces/src/workspaces_extra.rs index f82f487bf7..5a16aab58b 100644 --- a/backend/windmill-api-workspaces/src/workspaces_extra.rs +++ b/backend/windmill-api-workspaces/src/workspaces_extra.rs @@ -110,6 +110,21 @@ pub(crate) async fn change_workspace_id( .execute(&mut *tx) .await?; + // A fork copy reserved for the old id would otherwise be unreachable: its creator cannot + // import into it or finish its fork under the new id, and nothing else would ever drop it. + sqlx::query( + r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}', ( + SELECT COALESCE(jsonb_object_agg(k, CASE WHEN v->>'workspace_id' = $1 + THEN jsonb_set(v, '{workspace_id}', to_jsonb($2::text)) ELSE v END), '{}'::jsonb) + FROM jsonb_each(COALESCE(value->'databases', '{}'::jsonb)) AS e(k, v) + )) + WHERE name = 'custom_instance_pg_databases'"#, + ) + .bind(&old_id) + .bind(&rw.new_id) + .execute(&mut *tx) + .await?; + // Duplicate workspace settings (keep copy in old workspace for reference) info!("Duplicating workspace_settings table"); sqlx::query!( diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 8e27601115..8b95fdc6f1 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -34110,6 +34110,9 @@ components: items: type: string description: Workspaces that reference this database via a ducklake catalog or datatable database with resource_type 'instance'. Computed at request time, not persisted. + workspace_id: + type: string + description: The workspace a member created this database for as a fork copy. Only that workspace can import into it or point a fork at it. NewSqsTrigger: type: object diff --git a/backend/windmill-common/src/external_instance_pg.rs b/backend/windmill-common/src/external_instance_pg.rs index 0ef45dc9f7..f02f77c260 100644 --- a/backend/windmill-common/src/external_instance_pg.rs +++ b/backend/windmill-common/src/external_instance_pg.rs @@ -41,6 +41,21 @@ pub struct ExternalInstancePgState { pub databases: BTreeMap, #[serde(default, skip_serializing_if = "Option::is_none")] pub last_setup: Option, + /// The cluster ([`external_instance_pg_address`]) the last successful setup converged. Databases + /// are only created on a cluster setup succeeded on: the passwords above exist as soon as setup + /// first runs, whether or not the cluster accepted them. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub set_up_for: Option, +} + +/// What identifies the cluster a configuration points at. Other fields (admin login, sslmode) can +/// change without it becoming another cluster. +pub fn external_instance_pg_address(config: &ExternalInstancePg) -> String { + format!( + "{}:{}", + config.host.trim().to_lowercase(), + config.port.unwrap_or(5432) + ) } #[derive(Serialize, Deserialize, Clone, Debug)] @@ -125,6 +140,10 @@ pub async fn external_instance_pg_status(db: &DB) -> Result Result> { Ok(read_external_instance_pg_state(db).await?.databases) } @@ -394,8 +413,7 @@ async fn ensure_external_instance_pg_not_repointed( let Ok(desired) = serde_json::from_value::(value.clone()) else { return Ok(()); }; - let address = |c: &ExternalInstancePg| (c.host.trim().to_lowercase(), c.port.unwrap_or(5432)); - if address(¤t) == address(&desired) { + if external_instance_pg_address(¤t) == external_instance_pg_address(&desired) { return Ok(()); } ensure_external_instance_pg_unused( diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 324a5a2c9e..592762e2f6 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -1690,6 +1690,10 @@ pub fn system_ca_bundle() -> Option { /// its external instance counterpart) and nothing uses it yet. The `wm_fork_` prefix is no /// authorization: every database of a cluster answers to the same `custom_instance_user`, so a name /// is all it takes to reach another workspace's copy. +/// +/// Authorization: reads the global registries and every workspace's settings, and names other +/// workspaces in its refusal. Callers MUST have authorized `w_id` for the caller first — a member +/// of it forking or importing there — and MUST NOT call it on a workspace the caller is not in. pub async fn ensure_fork_database_available_to( db: &DB, kind: workspaces::DataTableCatalogResourceType, diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index 1e42a6f3e2..9a9b1da028 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -2265,20 +2265,55 @@ fn pg_attach_verification(res: &PgDatabase) -> Result = entries + .filter_map(|e| e.ok()) + .map(|e| e.path()) + .filter(|p| p.extension().is_some_and(|x| x == "pem") && p != keep) + .filter_map(|p| Some((std::fs::metadata(&p).ok()?.modified().ok()?, p))) + .collect(); + if files.len() < PG_ROOTS_KEPT { + return; + } + files.sort(); + for (_, p) in &files[..=files.len() - PG_ROOTS_KEPT] { + let _ = std::fs::remove_file(p); + } +} + fn pg_attach_uri(res: &PgDatabase) -> Result { let uri = res.to_uri(); let Some((mode, roots)) = pg_attach_verification(res)? else { @@ -2881,6 +2916,16 @@ mod tests { assert!(uri.contains("?sslmode=verify-full&sslrootcert="), "{uri}"); let root = urlencoding::decode(uri.split("sslrootcert=").nth(1).unwrap()).unwrap(); let roots = std::fs::read_to_string(root.as_ref()).unwrap(); + for i in 0..(PG_ROOTS_KEPT + 5) { + let mut other = pg("verify-full", Some(false)); + other.root_certificate_pem = Some(format!("-----BEGIN CERTIFICATE-----{i}")); + pg_attach_uri(&other).unwrap(); + } + let kept = std::fs::read_dir(std::env::temp_dir().join("windmill-pg-roots")) + .unwrap() + .filter(|e| e.as_ref().unwrap().path().extension().is_some_and(|x| x == "pem")) + .count(); + assert!(kept <= PG_ROOTS_KEPT, "{kept} root files kept"); assert!(roots.contains("-----BEGIN CERTIFICATE-----test")); let external = serde_json::to_value(pg("verify-full", Some(false))).unwrap(); let attach = &pg_secret_attach_statements(external, "dt").unwrap()[3];