Merge remote-tracking branch 'origin/datatable-external-instance-4' into datatable-external-instance-ui

backend/ee-repo-ref.txt

@@ -1 +1 @@
-cbacb629a604f8d15b0b3a526266eda82df880a8
+151be033ea2bf2f768f381a8cd3b888bdd366cf9

backend/windmill-api-settings/src/lib.rs

@@ -1678,6 +1678,8 @@ struct CustomInstanceDb {
     tag: Option<String>,
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     used_by_workspaces: Vec<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    workspace_id: Option<String>,
 }
 
 #[derive(Deserialize, Debug, Serialize, Default)]
@@ -1914,12 +1916,31 @@ async fn setup_custom_instance_pg_database(
     Path(dbname): Path<String>,
     Json(body): Json<SetupCustomInstanceDbBody>,
 ) -> JsonResult<CustomInstanceDb> {
+    // 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<String>>(
+        "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!(

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)

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!(

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

backend/windmill-common/src/external_instance_pg.rs

@@ -41,6 +41,21 @@ pub struct ExternalInstancePgState {
     pub databases: BTreeMap<String, CustomInstanceDb>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub last_setup: Option<ExternalInstancePgSetupReport>,
+    /// 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<String>,
+}
+
+/// 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<ExternalInstancePgSt
 }
 
 /// The databases Windmill created on the external cluster, without the passwords kept beside them.
+///
+/// Authorization: names every database across all workspaces, and the workspace each fork copy is
+/// reserved for, and checks nothing. Callers MUST be superadmin or an internal authorization or
+/// lifecycle path that does not return the names to a workspace caller.
 pub async fn external_instance_databases(db: &DB) -> Result<BTreeMap<String, CustomInstanceDb>> {
     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::<ExternalInstancePg>(value.clone()) else {
         return Ok(());
     };
-    let address = |c: &ExternalInstancePg| (c.host.trim().to_lowercase(), c.port.unwrap_or(5432));
-    if address(&current) == address(&desired) {
+    if external_instance_pg_address(&current) == external_instance_pg_address(&desired) {
         return Ok(());
     }
     ensure_external_instance_pg_unused(

backend/windmill-common/src/lib.rs

@@ -1690,6 +1690,10 @@ pub fn system_ca_bundle() -> Option<std::path::PathBuf> {
 /// 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,

backend/windmill-worker/src/duckdb_executor.rs

@@ -2265,20 +2265,55 @@ fn pg_attach_verification(res: &PgDatabase) -> Result<Option<(&str, std::path::P
     }
     let roots = format!("{bundle}\n{pem}\n");
     use sha2::Digest;
-    let path = std::env::temp_dir().join(format!(
-        "windmill-pg-roots-{}.pem",
+    let dir = std::env::temp_dir().join("windmill-pg-roots");
+    let path = dir.join(format!(
+        "{}.pem",
         hex::encode(&sha2::Sha256::digest(roots.as_bytes())[..8])
     ));
-    if !path.is_file() {
+    let write_err = |e: std::io::Error| {
+        Error::ExecutionErr(format!("Failed to write root certificates: {e}"))
+    };
+    if path.is_file() {
+        // Marks it recently used, so pruning takes the others first.
+        let _ = std::fs::File::options()
+            .append(true)
+            .open(&path)
+            .and_then(|f| f.set_modified(std::time::SystemTime::now()));
+    } else {
+        std::fs::create_dir_all(&dir).map_err(write_err)?;
         // Renamed into place: a job attaching concurrently must never read a half-written file.
         let partial = path.with_extension(format!("{}.partial", Uuid::new_v4()));
         std::fs::write(&partial, &roots)
             .and_then(|()| std::fs::rename(&partial, &path))
-            .map_err(|e| Error::ExecutionErr(format!("Failed to write root certificates: {e}")))?;
+            .map_err(write_err)?;
+        prune_pg_roots(&dir, &path);
     }
     Ok(Some((mode, path)))
 }
 
+/// Root files outlive the job: a resource's certificate is workspace-controlled, so each distinct
+/// one would otherwise add a file forever. Keeps the most recently used ones.
+const PG_ROOTS_KEPT: usize = 32;
+
+fn prune_pg_roots(dir: &std::path::Path, keep: &std::path::Path) {
+    let Ok(entries) = std::fs::read_dir(dir) else {
+        return;
+    };
+    let mut files: Vec<(std::time::SystemTime, std::path::PathBuf)> = 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<String> {
     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];