diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 0d9639c6c0..7bd84dfca3 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -131,9 +131,11 @@ minimal explicit set for dev. ## Workspace object storage in dev — use the local filesystem For a dev workspace you don't need MinIO/S3: use the built-in **`FilesystemStorage`** large-file -storage (a root path on local disk). It is intentionally hidden from the settings-UI storage -dropdown (dev-only), so set it via the API. Requires the backend built with `parquet` (+ `private` -for the real S3 helpers, + `enterprise` if you want advanced permission rules enforced): +storage (a root path on local disk). It is a **debug-build affordance only** — every site that +builds a filesystem object store calls `ensure_filesystem_storage_allowed`, so release builds +refuse it, and the settings UI never offers it — so set it via the API on a `cargo run`/`cargo +test` binary. Requires the backend built with `parquet` (+ `private` for the real S3 helpers, ++ `enterprise` if you want advanced permission rules enforced): ```bash curl -X POST "$BASE/api/w//workspaces/edit_large_file_storage_config" \ diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index e84533c5a7..d67f24bedb 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -4657e39ba368be10ae54325314a26048c42828d2 \ No newline at end of file +3bdee7852787fe9034d7029b58291c939ef080b8 \ No newline at end of file diff --git a/backend/migrations/20260826202939_queue_suspended_resume_at_index.down.sql b/backend/migrations/20260826202939_queue_suspended_resume_at_index.down.sql new file mode 100644 index 0000000000..fe727f6761 --- /dev/null +++ b/backend/migrations/20260826202939_queue_suspended_resume_at_index.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS queue_suspended_v2; diff --git a/backend/migrations/20260826202939_queue_suspended_resume_at_index.up.sql b/backend/migrations/20260826202939_queue_suspended_resume_at_index.up.sql new file mode 100644 index 0000000000..6d91a64d5f --- /dev/null +++ b/backend/migrations/20260826202939_queue_suspended_resume_at_index.up.sql @@ -0,0 +1,20 @@ +-- Serves the suspended-job pull in windmill-common/src/worker.rs, whose resume test is the +-- indexed CASE expression. Two things about the shape are load-bearing: +-- * (priority DESC NULLS LAST, created_at) leads, so the scan yields that query's ORDER BY +-- and stops at the first match rather than sorting. +-- * the index is dropped before it is built rather than relying on IF NOT EXISTS. The +-- OVERRIDDEN_MIGRATIONS rewrite in windmill-api/src/db.rs runs these CONCURRENTLY, and an +-- interrupted concurrent build leaves the index present but invalid, which IF NOT EXISTS +-- would then skip rebuilding. Retiring the index this replaces is left to the migration +-- that follows, so this one can only ever be replayed while that index is still there to +-- cover the rebuild. +DROP INDEX IF EXISTS queue_suspended_v2; + +CREATE INDEX IF NOT EXISTS queue_suspended_v2 + ON v2_job_queue ( + priority DESC NULLS LAST, + created_at, + (CASE WHEN suspend <= 0 THEN '-infinity'::timestamptz ELSE suspend_until END), + tag + ) + WHERE suspend_until IS NOT NULL; diff --git a/backend/migrations/20260826214706_queue_suspended_drop_legacy_index.down.sql b/backend/migrations/20260826214706_queue_suspended_drop_legacy_index.down.sql new file mode 100644 index 0000000000..011105927a --- /dev/null +++ b/backend/migrations/20260826214706_queue_suspended_drop_legacy_index.down.sql @@ -0,0 +1,3 @@ +CREATE INDEX IF NOT EXISTS queue_suspended + ON v2_job_queue (priority DESC NULLS LAST, created_at, suspend_until, suspend, tag) + WHERE suspend_until IS NOT NULL; diff --git a/backend/migrations/20260826214706_queue_suspended_drop_legacy_index.up.sql b/backend/migrations/20260826214706_queue_suspended_drop_legacy_index.up.sql new file mode 100644 index 0000000000..b8851b99e0 --- /dev/null +++ b/backend/migrations/20260826214706_queue_suspended_drop_legacy_index.up.sql @@ -0,0 +1,6 @@ +-- Retires the index queue_suspended_v2 replaces. Separate from the migration that builds it +-- so that one is only ever replayed while this index still exists: sqlx records a migration +-- only after all its statements run, so a process that dies before the record is written +-- replays the build, and its leading DROP would otherwise be destroying the sole usable +-- index rather than an interrupted build. +DROP INDEX IF EXISTS queue_suspended; diff --git a/backend/tests/suspended_pull_index.rs b/backend/tests/suspended_pull_index.rs new file mode 100644 index 0000000000..4b31f4d45e --- /dev/null +++ b/backend/tests/suspended_pull_index.rs @@ -0,0 +1,77 @@ +//! Pins the plan of the suspended-job pull. Its resume test degrades silently: once the +//! query expression and `queue_suspended_v2` stop matching, Postgres still returns the right +//! job, just by falling back to a heap filter and fetching one tuple per suspended row on +//! every worker poll. No functional test can see that, so assert on the plan instead. + +use serde_json::Value; +use sqlx::{Pool, Postgres}; +use windmill_common::worker::make_suspended_pull_query; + +/// Depth-first walk of an `EXPLAIN (FORMAT JSON)` plan tree. +fn nodes(plan: &Value, out: &mut Vec) { + out.push(plan.clone()); + for child in plan["Plans"].as_array().unwrap_or(&vec![]) { + nodes(child, out); + } +} + +#[sqlx::test(fixtures("base"))] +async fn suspended_pull_tests_resume_time_inside_the_index( + db: Pool, +) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO v2_job_queue (id, workspace_id, created_at, scheduled_for, running, suspend, suspend_until, tag) + SELECT gen_random_uuid(), 'test-workspace', now() - make_interval(secs => i), + now(), true, 1 + (i % 3), now() + interval '7 day', 'flow' + FROM generate_series(1, 2000) i", + ) + .execute(&db) + .await?; + sqlx::query("ANALYZE v2_job_queue").execute(&db).await?; + + // Both plans are cheap on a 2000-row table, and which one wins there says nothing + // about a queue with a large suspended backlog. Force the index path, which is the + // one production takes, and assert on how it evaluates the resume test. + let mut conn = db.acquire().await?; + sqlx::query("SET enable_seqscan = off") + .execute(&mut *conn) + .await?; + let version: String = sqlx::query_scalar("SELECT version()") + .fetch_one(&mut *conn) + .await?; + // FORMAT JSON rather than the default: `Index Cond` and `Filter` are separate keys on the + // node, so this does not ride on EXPLAIN's line layout staying put across a major bump. + let explained: Value = sqlx::query_scalar(&format!( + "EXPLAIN (FORMAT JSON) {}", + make_suspended_pull_query(&["flow".to_string()]) + )) + .bind("test-worker") + .fetch_one(&mut *conn) + .await?; + + let mut all = vec![]; + nodes(&explained[0]["Plan"], &mut all); + let pretty = serde_json::to_string_pretty(&explained)?; + let scan = all + .iter() + .find(|n| n["Index Name"] == "queue_suspended_v2") + .unwrap_or_else(|| { + panic!("suspended pull did not scan queue_suspended_v2 on {version}:\n{pretty}") + }); + // Only `Index Cond` is checked against the index tuple, so that is where the resume test + // has to land — as a `Filter` it would cost a heap fetch per suspended row. The residual + // `suspend_until IS NOT NULL` filter is not that: it is always true for rows the partial + // index holds, and only ever runs on the row LIMIT 1 already fetched. + let cond = scan["Index Cond"].as_str().unwrap_or_else(|| { + panic!("no Index Cond on the suspended pull scan on {version}:\n{pretty}") + }); + assert!( + cond.contains("CASE WHEN"), + "resume test is not an index condition on {version}:\n{pretty}" + ); + assert!( + !scan["Filter"].as_str().unwrap_or("").contains("CASE WHEN"), + "resume test fell back to a heap filter on {version}:\n{pretty}" + ); + Ok(()) +} diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index e04e1c1ab9..3b3ea2713c 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -1970,6 +1970,23 @@ async fn edit_large_file_storage_config( ))); } + if !windmill_common::workspaces::filesystem_storage_allowed() { + let named = std::iter::once(("primary storage", &lfs_config.large_file_storage)).chain( + lfs_config + .secondary_storage + .iter() + .map(|(name, storage)| (name.as_str(), storage)), + ); + for (name, storage) in named { + if matches!(storage, LargeFileStorage::FilesystemStorage(_)) { + return Err(Error::BadRequest(format!( + "{name}: {}", + windmill_common::workspaces::FILESYSTEM_STORAGE_DEV_ONLY_MSG + ))); + } + } + } + let serialized_lfs_config = serde_json::to_value::(lfs_config) .map_err(|err| Error::internal_err(err.to_string()))?; diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 04d21c7b53..d781e229d2 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -102,6 +102,12 @@ lazy_static::lazy_static! { (20260727151319, include_str!( "../../migrations/20260727151319_draft_only_listing_indexes.up.sql" ).replace("CREATE INDEX", "CREATE INDEX CONCURRENTLY")), + (20260826202939, include_str!( + "../../migrations/20260826202939_queue_suspended_resume_at_index.up.sql" + ).replace("CREATE INDEX", "CREATE INDEX CONCURRENTLY").replace("DROP INDEX", "DROP INDEX CONCURRENTLY")), + (20260826214706, include_str!( + "../../migrations/20260826214706_queue_suspended_drop_legacy_index.up.sql" + ).replace("DROP INDEX", "DROP INDEX CONCURRENTLY")), ].into_iter().collect(); } @@ -228,6 +234,8 @@ impl Migrate for CustomMigrator { // CONCURRENTLY operations cannot run inside a transaction block // or a multi-statement query (PostgreSQL requires top-level execution). // Split into individual statements and execute each separately. + // The split is naive, so a `;` anywhere in an overridden migration — + // inside a comment or a string literal included — splits mid-statement. for stmt in migration_sql.split(';') { let stmt = stmt.trim(); if !stmt.is_empty() diff --git a/backend/windmill-common/src/db_params.rs b/backend/windmill-common/src/db_params.rs deleted file mode 100644 index d52700376d..0000000000 --- a/backend/windmill-common/src/db_params.rs +++ /dev/null @@ -1,45 +0,0 @@ -use anyhow::Result; - -/// Parsed database connection parameters, shared across DB auth providers (IAM RDS, Entra ID, etc.) -#[derive(Debug, Clone)] -pub struct DatabaseParams { - pub hostname: String, - pub port: u64, - pub username: String, - pub database: String, -} - -/// Extract database connection parameters from a PostgreSQL URL -pub fn extract_database_params(database_url: &str) -> Result { - let url = url::Url::parse(database_url) - .map_err(|e| anyhow::anyhow!("Failed to parse database URL: {}", e))?; - - let hostname = url - .host_str() - .ok_or_else(|| anyhow::anyhow!("Database URL missing hostname"))? - .to_string(); - - let port = url.port().unwrap_or(5432) as u64; - - let username = if url.username().is_empty() { - return Err(anyhow::anyhow!("Database URL missing username")); - } else { - urlencoding::decode(url.username())?.to_string() - }; - - let database = url - .path() - .trim_start_matches('/') - .split('/') - .next() - .filter(|s| !s.is_empty()) - .ok_or_else(|| anyhow::anyhow!("Database URL missing database name"))? - .to_string(); - - Ok(DatabaseParams { - hostname, - port, - username, - database: urlencoding::decode(&database)?.to_string(), - }) -} diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 89db0ee769..cf0be2bbd8 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -42,7 +42,6 @@ pub mod db; mod db_entra_ee; #[cfg(all(feature = "enterprise", feature = "private"))] mod db_iam_ee; -pub mod db_params; pub mod dbt_manifest; pub mod deploy_origin; #[cfg(feature = "private")] @@ -1479,6 +1478,17 @@ pub async fn create_custom_instance_database( Ok(()) } +/// Connection options parsed from a database URL. +/// +/// The only place a database URL becomes `PgConnectOptions`. Providers that mint the password +/// themselves override it on these and keep the rest: options assembled field by field instead +/// would drop every query parameter, `sslmode` and `sslrootcert` above all, leaving the +/// connection on sqlx's default TLS policy rather than the operator's. +pub fn base_connect_options(database_url: &str) -> Result { + sqlx::postgres::PgConnectOptions::from_str(database_url) + .map_err(|e| Error::InternalErr(format!("Failed to parse database URL: {}", e))) +} + #[derive(Clone)] pub enum DatabaseUrl { #[cfg(all(feature = "enterprise", feature = "private"))] @@ -1509,8 +1519,8 @@ impl DatabaseUrl { } /// Get PgConnectOptions for this database URL. - /// For token-based auth (IAM RDS, Entra ID), this returns options built directly from the - /// token to avoid double-encoding issues with temporary credentials. + /// For token-based auth (IAM RDS, Entra ID), this returns options carrying the current + /// token, set on the builder to avoid double-encoding temporary credentials. /// For static URLs, this parses the URL string. pub async fn connect_options(&self) -> Result { match self { @@ -1524,8 +1534,7 @@ impl DatabaseUrl { let guard = entra_url.read().await; Ok(guard.connect_options()) } - DatabaseUrl::Static(url) => sqlx::postgres::PgConnectOptions::from_str(url) - .map_err(|e| Error::InternalErr(format!("Failed to parse database URL: {}", e))), + DatabaseUrl::Static(url) => base_connect_options(url), } } diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 7945e73b45..84284759c6 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -734,11 +734,17 @@ fn format_pull_query(peek: String) -> String { r } +// The `CASE` is `suspend <= 0 OR suspend_until <= now()` written as one indexable +// expression, equivalent only under the `suspend_until IS NOT NULL` guard. It must stay in +// sync with `queue_suspended_v2` (migration 20260826202939): if it no longer matches, the +// test silently reverts to a heap filter over every suspended row on every worker poll. pub fn make_suspended_pull_query(tags: &[String]) -> String { format_pull_query(format!( "SELECT id FROM v2_job_queue - WHERE suspend_until IS NOT NULL AND (suspend <= 0 OR suspend_until <= now()) AND tag IN ({}) + WHERE suspend_until IS NOT NULL + AND (CASE WHEN suspend <= 0 THEN '-infinity'::timestamptz ELSE suspend_until END) <= now() + AND tag IN ({}) ORDER BY priority DESC NULLS LAST, created_at FOR UPDATE SKIP LOCKED LIMIT 1", diff --git a/backend/windmill-common/src/workspaces.rs b/backend/windmill-common/src/workspaces.rs index 5f558d325a..ae3e6546cd 100644 --- a/backend/windmill-common/src/workspaces.rs +++ b/backend/windmill-common/src/workspaces.rs @@ -2193,6 +2193,32 @@ pub fn lfs_entry_storage_ref(entry: &serde_json::Value) -> Option { Some(format!("{typ}:{path}")) } +pub const FILESYSTEM_STORAGE_DEV_ONLY_MSG: &str = + "Filesystem storage is only available in development builds of Windmill: it points the \ + workspace at a directory on the server's own disk rather than at a resource. Use an S3, \ + Azure Blob or Google Cloud Storage backend instead."; + +/// A filesystem workspace storage names a directory on the server's own disk, so it hands whoever +/// configures it — a workspace admin, or any member who can write a `filesystem` resource — +/// whatever the server process can reach, and it only resolves when server and workers share that +/// disk. It is there so local development can skip MinIO, hence debug builds only. Instance object +/// storage on local disk is a separate, superadmin-only setting and stays allowed everywhere. +pub fn filesystem_storage_allowed() -> bool { + cfg!(debug_assertions) +} + +/// Guards every site that builds an `ObjectStoreResource::Filesystem`, so nothing downstream can +/// reach a local-disk store: a stored config outlives the build that accepted it, and the resource +/// route never passes through the workspace-storage settings at all. +pub fn ensure_filesystem_storage_allowed() -> Result<()> { + if !filesystem_storage_allowed() { + return Err(Error::BadRequest( + FILESYSTEM_STORAGE_DEV_ONLY_MSG.to_string(), + )); + } + Ok(()) +} + /// Resolve a `$res:`/`$var:` reference tree to its concrete value (recursively, secrets /// decrypted). No permission checks — trusted server-side callers only; never echo the result /// to a user. diff --git a/backend/windmill-object-store/src/lib.rs b/backend/windmill-object-store/src/lib.rs index 0f919f1048..cc588d485f 100644 --- a/backend/windmill-object-store/src/lib.rs +++ b/backend/windmill-object-store/src/lib.rs @@ -1171,6 +1171,7 @@ pub fn lfs_to_object_store_resource( Ok(ObjectStoreResource::Gcs(gcs_resource)) } LargeFileStorage::FilesystemStorage(fs) => { + windmill_common::workspaces::ensure_filesystem_storage_allowed()?; Ok(ObjectStoreResource::Filesystem(FilesystemSettings { root_path: fs.root_path.clone(), })) diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 6ee22fd882..f6391eab07 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1640,6 +1640,7 @@ pub(crate) async fn get_workspace_s3_resource_path( ) } Some(LargeFileStorage::FilesystemStorage(fs)) => { + windmill_common::workspaces::ensure_filesystem_storage_allowed()?; return Ok(Some( windmill_object_store::ObjectStoreResource::Filesystem( windmill_object_store::FilesystemSettings { root_path: fs.root_path.clone() }, diff --git a/docker-compose.yml b/docker-compose.yml index a801a0ce7c..bd1e060f22 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,15 +8,28 @@ x-logging: &default-logging compress: "true" services: + ## UPGRADING FROM POSTGRES 16: db_data holds a cluster 18 cannot read, so the + ## container exits with an explanatory error rather than coming up blank. Migrating + ## means dumping the WHOLE cluster (pg_dumpall), never just the windmill database: + ## Windmill keeps datatable, DuckLake and wm_fork_* databases beside it and grants + ## its RLS policies to cluster-level roles, and a single-database dump loses both + ## silently. Full procedure, and why 16 is still a valid choice until Nov 2028: + ## https://www.windmill.dev/docs/advanced/self_host#upgrade-postgresql-to-18 db: deploy: # To use an external database, set replicas to 0 and set DATABASE_URL to the external database url in the .env file replicas: 1 - image: postgres:16 + image: postgres:18 shm_size: 1g restart: unless-stopped volumes: - - db_data:/var/lib/postgresql/data + # From 18 on the official image keeps the cluster in a major-version + # subdirectory (/var/lib/postgresql/18/docker), so the mount has to be the + # parent directory: that is what lets pg_upgrade see an old and a new + # cluster inside a single mount point. Mounting the pre-18 .../data path + # instead makes the image exit rather than start, which is what turns a + # stale 16 cluster into a loud failure instead of an empty instance. + - db_data:/var/lib/postgresql expose: - 5432 environment: diff --git a/frontend/src/lib/components/workspaceSettings/StorageSettings.svelte b/frontend/src/lib/components/workspaceSettings/StorageSettings.svelte index 1e6c990819..a2ccf0b958 100644 --- a/frontend/src/lib/components/workspaceSettings/StorageSettings.svelte +++ b/frontend/src/lib/components/workspaceSettings/StorageSettings.svelte @@ -44,6 +44,14 @@ onDiscard?: () => void } = $props() + const creatableStorageTypes = [ + { value: 's3', label: 'S3' }, + { value: 'azure_blob', label: 'Azure Blob' }, + { value: 's3_aws_oidc', label: 'AWS OIDC' }, + { value: 'azure_workload_identity', label: 'Azure Workload Identity' }, + { value: 'gcloud_storage', label: 'Google Cloud Storage' } + ] + let advancedPermissionModalState: | { open: false } | { open: true; storage: S3ResourceSettingsItem } = $state({ open: false }) @@ -294,31 +302,38 @@
- {#if tableRow[1].resourceType === 'filesystem'} - +
+ - {/if} + {#if tableRow[1].resourceType === 'filesystem'} + + Filesystem storage points the workspace at a directory on the server's own + disk. Only development builds of Windmill accept it — switch this storage to + S3, Azure Blob or Google Cloud Storage to configure it here. + + {/if} +
{#if tableRow[1].resourceType === 'filesystem'}