From 2b55a8afbc8a86c6cebe97b1cab02c313b986f22 Mon Sep 17 00:00:00 2001 From: hugocasa Date: Fri, 9 Oct 2026 18:32:05 +0200 Subject: [PATCH] feat: replace the hub_sync script with an opt-in daily server sync (#11624) * feat: sync hub resource types daily from the server Co-Authored-By: Claude Opus 5.5 * feat: retire the legacy hub_sync script and its schedule Co-Authored-By: Claude Opus 5.5 * fix: sync only from the hub on the daily schedule Co-Authored-By: Claude Opus 5.5 * test: sync resource types through the endpoint in e2e setup Co-Authored-By: Claude Opus 5.5 * fix: honor ACCEPT_INVALID_CERTS in the hub resource type sync Co-Authored-By: Claude Opus 5.5 * feat: audit manual hub resource type syncs that change types Co-Authored-By: Claude Opus 5.5 * refactor: show the AI step's resource type sync as a setting card Co-Authored-By: Claude Opus 5.5 * fix: keep the hub_sync opt-out and move sync state to task state Co-Authored-By: Claude Opus 5.5 * fix: hold the daily hub sync until the legacy hub_sync is retired Co-Authored-By: Claude Opus 5.5 * fix: turn off the daily sync when an edited hub_sync is scheduled Co-Authored-By: Claude Opus 5.5 * refactor: make the daily hub resource type sync opt-in Co-Authored-By: Claude Opus 5.5 * feat: bring back the setup wizard's daily resource type sync toggle Co-Authored-By: Claude Opus 5.5 * fix: apply the image's resource types on every SYNC_CACHED_RT start Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- ...04befc3a6596c258086805dd42f513ddb3ead.json | 20 - ...cc5dd6c7b0da4f76131313d3e6d97f36c5c34.json | 28 - backend/src/main.rs | 258 +------- backend/tests/wm_token_confinement.rs | 2 +- backend/windmill-api-auth/src/scopes.rs | 3 +- backend/windmill-api-scripts/src/scripts.rs | 12 - backend/windmill-api-settings/src/lib.rs | 209 +------ backend/windmill-api/openapi.yaml | 1 + backend/windmill-api/src/live_migrations.rs | 152 ++++- .../windmill-common/src/global_settings.rs | 1 + .../windmill-common/src/hub_resource_types.rs | 578 ++++++++++++++++++ .../windmill-common/src/instance_config.rs | 2 + backend/windmill-common/src/lib.rs | 1 + frontend/e2e/global-setup.ts | 16 +- .../src/lib/components/instanceSettings.ts | 9 + .../InstanceAISettings.svelte | 68 +-- .../(user)/instance_settings/+page.svelte | 110 +--- 17 files changed, 846 insertions(+), 624 deletions(-) delete mode 100644 backend/.sqlx/query-212bf5b32de102a9c537907aaff04befc3a6596c258086805dd42f513ddb3ead.json delete mode 100644 backend/.sqlx/query-386e14cf7572027f2c4ef313cd7cc5dd6c7b0da4f76131313d3e6d97f36c5c34.json create mode 100644 backend/windmill-common/src/hub_resource_types.rs diff --git a/backend/.sqlx/query-212bf5b32de102a9c537907aaff04befc3a6596c258086805dd42f513ddb3ead.json b/backend/.sqlx/query-212bf5b32de102a9c537907aaff04befc3a6596c258086805dd42f513ddb3ead.json deleted file mode 100644 index 04b23832d9..0000000000 --- a/backend/.sqlx/query-212bf5b32de102a9c537907aaff04befc3a6596c258086805dd42f513ddb3ead.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO resource_type (workspace_id, name, schema, description, format_extension, display_name, edited_at)\n VALUES ('admins', $1, $2, $3, $4, $6, now())\n ON CONFLICT (workspace_id, name) DO UPDATE\n SET schema = EXCLUDED.schema, description = EXCLUDED.description,\n -- A fileset is a set of files, so it cannot also be one file.\n -- Create and update reject the pair; this writer bypasses both, so\n -- it declines the extension rather than persisting the forbidden\n -- combination onto a same-named local fileset.\n format_extension = CASE\n WHEN resource_type.is_fileset THEN NULL\n WHEN $5 THEN EXCLUDED.format_extension\n ELSE resource_type.format_extension END,\n display_name = CASE WHEN $7 THEN EXCLUDED.display_name ELSE resource_type.display_name END,\n edited_at = now()", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Jsonb", - "Text", - "Varchar", - "Bool", - "Varchar", - "Bool" - ] - }, - "nullable": [] - }, - "hash": "212bf5b32de102a9c537907aaff04befc3a6596c258086805dd42f513ddb3ead" -} diff --git a/backend/.sqlx/query-386e14cf7572027f2c4ef313cd7cc5dd6c7b0da4f76131313d3e6d97f36c5c34.json b/backend/.sqlx/query-386e14cf7572027f2c4ef313cd7cc5dd6c7b0da4f76131313d3e6d97f36c5c34.json deleted file mode 100644 index 2f794e6be0..0000000000 --- a/backend/.sqlx/query-386e14cf7572027f2c4ef313cd7cc5dd6c7b0da4f76131313d3e6d97f36c5c34.json +++ /dev/null @@ -1,28 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT EXISTS(SELECT 1 FROM resource_type WHERE workspace_id = 'admins' AND name = $1 AND schema IS NOT DISTINCT FROM $2 AND description IS NOT DISTINCT FROM $3 AND ($5 IS NOT TRUE OR format_extension IS NOT DISTINCT FROM $4) AND ($7 IS NOT TRUE OR display_name IS NOT DISTINCT FROM $6))", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "exists", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Text", - "Jsonb", - "Text", - "Text", - "Bool", - "Text", - "Bool" - ] - }, - "nullable": [ - null - ] - }, - "hash": "386e14cf7572027f2c4ef313cd7cc5dd6c7b0da4f76131313d3e6d97f36c5c34" -} diff --git a/backend/src/main.rs b/backend/src/main.rs index 81104d6764..66ddce60e2 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -63,12 +63,14 @@ use windmill_common::{ SANDBOX_IMAGE_CACHE_MAX_MB_SETTING, SANDBOX_IMAGE_DEFAULT_REGISTRY_SETTING, SANDBOX_IMAGE_MAX_SIZE_MB_SETTING, SANDBOX_IMAGE_PULL_POLICY_SETTING, SANDBOX_REGISTRY_AUTH_SETTING, SCIM_TOKEN_SETTING, SERVICE_LOG_RETENTION_SECS_SETTING, - SMTP_SETTING, STORE_AUDIT_LOGS_S3_SETTING, TEAMS_SETTING, TIMEOUT_WAIT_RESULT_SETTING, - UV_EXCLUDE_NEWER_SETTING, UV_INDEX_STRATEGY_SETTING, UV_PYTHON_INSTALL_MIRROR_SETTING, + SMTP_SETTING, STORE_AUDIT_LOGS_S3_SETTING, SYNC_HUB_RESOURCE_TYPES_DAILY_SETTING, + TEAMS_SETTING, TIMEOUT_WAIT_RESULT_SETTING, UV_EXCLUDE_NEWER_SETTING, + UV_INDEX_STRATEGY_SETTING, UV_PYTHON_INSTALL_MIRROR_SETTING, WORKSPACE_FAIRNESS_DURATION_SECS_SETTING, WORKSPACE_FAIRNESS_ENABLED_SETTING, WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING, WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING, WORKSPACE_MAX_QUEUED_JOBS_SETTING, WORKSPACE_REGISTRIES_SETTING, }, + hub_resource_types, scripts::ScriptLang, stats_oss::schedule_stats, triggers::TriggerKind, @@ -401,111 +403,16 @@ async fn cache_hub_scripts(file_path: Option) -> anyhow::Result<()> { Ok(()) } -/// Raw resource type from hub API (schema is a JSON string) -#[derive(serde::Deserialize)] -struct HubResourceTypeRaw { - pub id: i64, - pub name: String, - pub schema: Option, - pub app: String, - pub description: Option, - /// Absent from hubs predating the column, and from caches written before it. - #[serde(default)] - pub format_extension: Option, - /// Doubly optional, so a hub predating the field (no key) is told apart from a type the - /// hub leaves unnamed (null). - #[serde( - default, - deserialize_with = "windmill_common::more_serde::double_option" - )] - pub display_name: Option>, -} - -/// Processed resource type with parsed schema -#[derive(serde::Deserialize, serde::Serialize, Clone)] -pub struct HubResourceType { - pub id: i64, - pub name: String, - pub schema: Option, - pub app: String, - pub description: Option, - /// Doubly optional on purpose. A cache written before this column has no key at - /// all (`None`) and must leave the stored extension alone; one written since - /// always writes the key, so an explicit null (`Some(None)`) is the hub genuinely - /// dropping it and must clear. A single `Option` conflates the two, and picking - /// either meaning breaks the other — as does plain serde, which folds `null` - /// into the outer `None`, hence the wrapping deserializer. - #[serde( - default, - deserialize_with = "windmill_common::more_serde::double_option", - skip_serializing_if = "Option::is_none" - )] - pub format_extension: Option>, - /// Doubly optional like `format_extension`: a cache written before the field leaves the - /// stored name alone, while a null from the hub clears it. - #[serde( - default, - deserialize_with = "windmill_common::more_serde::double_option", - skip_serializing_if = "Option::is_none" - )] - pub display_name: Option>, -} - -const HUB_RT_CACHE_FILE: &str = "resource_types.json"; - async fn cache_hub_resource_types() -> anyhow::Result<()> { println!("Caching resource types from hub..."); - let response = HTTP_CLIENT - .get(format!("{}/resource_types/list", DEFAULT_HUB_BASE_URL)) - .header("Accept", "application/json") - .send() - .await - .with_context(|| "Failed to fetch resource types from hub")?; - - if !response.status().is_success() { - anyhow::bail!( - "Failed to fetch resource types from hub: {}", - response.status() - ); - } - - let raw_types: Vec = response - .json::>() - .await - .with_context(|| "Failed to parse resource types from hub")?; - - // Parse schema strings into JSON values - let resource_types: Vec = raw_types - .into_iter() - .filter_map(|rt| { - let schema = match rt.schema { - Some(s) => match serde_json::from_str(&s) { - Ok(v) => Some(v), - Err(e) => { - println!("Warning: failed to parse schema for {}: {}", rt.name, e); - return None; - } - }, - None => None, - }; - Some(HubResourceType { - id: rt.id, - name: rt.name, - schema, - app: rt.app, - description: rt.description, - format_extension: Some(rt.format_extension), - display_name: rt.display_name, - }) - }) - .collect(); + let resource_types = hub_resource_types::fetch(DEFAULT_HUB_BASE_URL, None, None).await?; println!("Fetched {} resource types from hub", resource_types.len()); create_dir_all(&*HUB_RT_CACHE_DIR)?; - let cache_path = format!("{}/{}", *HUB_RT_CACHE_DIR, HUB_RT_CACHE_FILE); + let cache_path = hub_resource_types::cache_path(); let content = serde_json::to_string_pretty(&resource_types) .with_context(|| "Failed to serialize resource types")?; @@ -516,147 +423,6 @@ async fn cache_hub_resource_types() -> anyhow::Result<()> { Ok(()) } -pub async fn sync_cached_resource_types(db: &sqlx::Pool) -> anyhow::Result<()> { - let cache_path = format!("{}/{}", *HUB_RT_CACHE_DIR, HUB_RT_CACHE_FILE); - - if tokio::fs::metadata(&cache_path).await.is_err() { - tracing::info!( - "No cached resource types found at {}, skipping sync", - cache_path - ); - return Ok(()); - } - - tracing::info!("Syncing cached resource types to admins workspace..."); - - let content = tokio::fs::read_to_string(&cache_path) - .await - .with_context(|| format!("Failed to read cache file from {}", cache_path))?; - - let cached_types: Vec = - serde_json::from_str(&content).with_context(|| "Failed to parse cached resource types")?; - - tracing::info!("Found {} cached resource types", cached_types.len()); - - // Get existing resource types in admins workspace. `format_extension` is part of - // the comparison below, so a type whose only change is gaining or losing it is - // not mistaken for unchanged; `is_fileset` decides whether it may take one. - let existing_types: Vec<( - String, - Option, - Option, - Option, - bool, - Option, - )> = sqlx::query_as( - "SELECT name, schema, description, format_extension, is_fileset, display_name FROM resource_type WHERE workspace_id = 'admins'", - ) - .fetch_all(db) - .await - .with_context(|| "Failed to fetch existing resource types")?; - - let existing_map: std::collections::HashMap< - String, - ( - Option, - Option, - Option, - bool, - Option, - ), - > = existing_types - .into_iter() - .map( - |(name, schema, desc, format_extension, is_fileset, display_name)| { - ( - name, - (schema, desc, format_extension, is_fileset, display_name), - ) - }, - ) - .collect(); - - let mut synced_count = 0; - let mut skipped_count = 0; - - for rt in cached_types { - let existing = existing_map.get(&rt.name); - let is_fileset = existing.map(|(_, _, _, f, _)| *f).unwrap_or(false); - let stored_extension = existing.and_then(|(_, _, e, _, _)| e.clone()); - let stored_display_name = existing.and_then(|(_, _, _, _, n)| n.clone()); - // A fileset is a set of files, so it cannot also be one file. Create, update - // and the manual sync all reject the pair; this writer would otherwise - // persist it onto a same-named local fileset. - // - // A cache with no key at all leaves the stored value alone, so the target is - // what is already there — which is also what makes the comparison below - // agree with the write instead of re-upserting the row on every boot. - let format_extension = if is_fileset { - None - } else { - match &rt.format_extension { - Some(from_cache) => from_cache.clone(), - None => stored_extension.clone(), - } - }; - // No key in the cache leaves the stored name alone, as for the extension. So does a name - // too long for the column: one bad entry must not fail the upsert and end the sync. - let display_name = match &rt.display_name { - Some(Some(name)) if name.chars().count() > 100 => { - tracing::warn!( - "Ignoring the display_name of resource type {}: longer than 100 characters", - rt.name - ); - stored_display_name.clone() - } - Some(from_cache) => from_cache.clone(), - None => stored_display_name.clone(), - }; - - if let Some((existing_schema, existing_desc, _, _, _)) = existing { - if existing_schema == &rt.schema - && existing_desc == &rt.description - && stored_extension == format_extension - && stored_display_name == display_name - { - skipped_count += 1; - continue; - } - } - - // Insert or update resource type - sqlx::query( - // `format_extension` is resolved above rather than coalesced here: a - // COALESCE could never clear one, so a hub that dropped an extension - // would leave the stale value behind forever. - "INSERT INTO resource_type (workspace_id, name, schema, description, format_extension, display_name, edited_at) - VALUES ('admins', $1, $2, $3, $4, $5, now()) - ON CONFLICT (workspace_id, name) DO UPDATE - SET schema = EXCLUDED.schema, description = EXCLUDED.description, - format_extension = EXCLUDED.format_extension, - display_name = EXCLUDED.display_name, edited_at = now()", - ) - .bind(&rt.name) - .bind(&rt.schema) - .bind(&rt.description) - .bind(&format_extension) - .bind(&display_name) - .execute(db) - .await - .with_context(|| format!("Failed to upsert resource type {}", rt.name))?; - - synced_count += 1; - } - - tracing::info!( - "Synced {} resource types to admins workspace ({} skipped as unchanged)", - synced_count, - skipped_count - ); - - Ok(()) -} - fn print_help() { println!("Windmill - a fast, open-source workflow engine and job runner."); println!(); @@ -696,7 +462,7 @@ fn print_help() { println!(" LICENSE_KEY = None (EE only) Enterprise license key (workers require valid key)"); println!(" RUN_UPDATE_CA_CERTIFICATE_AT_START = false Run system CA update at startup"); println!(" RUN_UPDATE_CA_CERTIFICATE_PATH = /usr/sbin/update-ca-certificates Path to CA update tool"); - println!(" SYNC_CACHED_RT = false Sync cached resource types to admins workspace on server start"); + println!(" SYNC_CACHED_RT = false Sync the image's cached resource types to the admins workspace on server start"); println!(" HUB_BASE_URL = https://hub.windmill.dev Hub to fetch scripts from in `cache` mode (server/worker use the DB setting instead)"); println!(); println!("Notes:"); @@ -1015,8 +781,10 @@ async fn windmill_main() -> anyhow::Result<()> { .map(|v| v.to_lowercase() == "true" || v == "1") .unwrap_or(false) { - if let Err(e) = sync_cached_resource_types(db).await { - tracing::warn!("Failed to sync cached resource types: {:#}", e); + match hub_resource_types::sync_from_image_cache(db).await { + Ok(Some(outcome)) => tracing::info!("{}", outcome.summary()), + Ok(None) => {} + Err(e) => tracing::warn!("Failed to sync cached resource types: {:#}", e), } } } @@ -1748,6 +1516,7 @@ Windmill Community Edition {GIT_VERSION} if server_mode { if let Some(db) = conn.as_sql() { schedule_stats(&db, &HTTP_CLIENT).await; + hub_resource_types::spawn_daily_sync(db.clone(), killpill_tx.subscribe()); } } @@ -2322,6 +2091,9 @@ async fn process_notify_event( RESTART_COORDINATION_SETTING => { // Internal coordination key for staggered restarts, no action needed } + SYNC_HUB_RESOURCE_TYPES_DAILY_SETTING => { + // Read by the daily resource type sync each time it runs + } "plain_emails_telemetry" => { let enabled = sqlx::query_scalar!( "SELECT value FROM global_settings WHERE name = 'plain_emails_telemetry'" diff --git a/backend/tests/wm_token_confinement.rs b/backend/tests/wm_token_confinement.rs index 23d88301b7..e75170ec6f 100644 --- a/backend/tests/wm_token_confinement.rs +++ b/backend/tests/wm_token_confinement.rs @@ -319,7 +319,7 @@ async fn test_wm_token_is_confined_to_its_workspace(db: Pool) -> anyho ); } // ...and the `settings/global` keys on the allowlist, which the CLI reads from a job: on a - // git-sync push, and in `u/admin/hub_sync`. `ws_base_url` is the control: the handler + // git-sync push, and in `wmill hub pull`. `ws_base_url` is the control: the handler // leaves it as ungated as these, so only the allowlist stops it. for key in ["automate_username_creation", "uid", "hub_base_url"] { let resp = authed( diff --git a/backend/windmill-api-auth/src/scopes.rs b/backend/windmill-api-auth/src/scopes.rs index b636e11a1c..d586b6ba63 100644 --- a/backend/windmill-api-auth/src/scopes.rs +++ b/backend/windmill-api-auth/src/scopes.rs @@ -862,7 +862,8 @@ fn scope_grants_access( /// from its own super-admin gate, these among them, so each is already readable by every /// authenticated user; each is here because the CLI reads it from a job: /// `automate_username_creation` before creating a user during a git-sync push, `uid` and -/// `hub_base_url` when `u/admin/hub_sync` pulls resource types from the Hub. The other +/// `hub_base_url` when a job runs `wmill hub pull`, as an edited copy of the retired +/// `u/admin/hub_sync` still does (servers sync resource types themselves). The other /// ungated keys have no such caller, so they stay confined — being ungated earns a key /// nothing on its own. Listing a gated key earns it nothing either: `require_super_admin` /// refuses every job token, so `hub_api_secret`, which that pull reads for a private Hub, diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index f622c4bb91..4f4c962759 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -4266,12 +4266,6 @@ async fn delete_script_by_path( check_scopes(&authed, || format!("scripts:write:{}", path))?; - if path == "u/admin/hub_sync" && w_id == "admins" { - return Err(Error::BadRequest( - "Cannot delete the global setup app".to_string(), - )); - } - if let RuleCheckResult::Blocked(msg) = check_deploy_rules( &w_id, AuditAuthorable::username(&authed), @@ -4445,12 +4439,6 @@ async fn delete_scripts_bulk( require_admin(authed.is_admin, &authed.username)?; - if request.paths.contains(&"u/admin/hub_sync".to_string()) && w_id == "admins" { - return Err(Error::BadRequest( - "Cannot delete the global setup app".to_string(), - )); - } - if let RuleCheckResult::Blocked(msg) = check_deploy_rules( &w_id, AuditAuthorable::username(&authed), diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index f9f0a993fa..116ff73457 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -30,6 +30,7 @@ mod storage_usage; use windmill_api_auth::{require_devops_role, require_super_admin, ApiAuthed}; use windmill_audit::{audit_oss::audit_log, ActionKind}; +use windmill_common::hub_resource_types::{self, SyncSource}; use windmill_common::utils::HTTP_CLIENT_PERMISSIVE as HTTP_CLIENT; use windmill_common::DB; @@ -2576,95 +2577,6 @@ pub async fn get_jwks() -> JsonResult { Ok(Json(JwksResponse { keys: vec![] })) } -#[derive(serde::Deserialize, serde::Serialize)] -struct CachedResourceType { - #[allow(dead_code)] - id: i64, - name: String, - schema: Option, - #[allow(dead_code)] - app: String, - description: Option, - /// Doubly optional, and read through a wrapping deserializer: this struct also - /// decodes the on-disk cache, where an absent key means "written before the - /// column, leave the stored extension alone" and an explicit null means the hub - /// dropped it. Plain serde folds both into `None`. - #[serde( - default, - deserialize_with = "windmill_common::more_serde::double_option" - )] - format_extension: Option>, - /// Doubly optional like `format_extension`: no key leaves the stored name alone, an explicit - /// null (the hub naming nothing) clears it. - #[serde( - default, - deserialize_with = "windmill_common::more_serde::double_option" - )] - display_name: Option>, -} - -#[derive(serde::Deserialize)] -struct HubResourceTypeRaw { - id: i64, - name: String, - schema: Option, - app: String, - description: Option, - #[serde(default)] - format_extension: Option, - #[serde( - default, - deserialize_with = "windmill_common::more_serde::double_option" - )] - display_name: Option>, -} - -async fn fetch_resource_types_from_hub() -> error::Result> { - let response = HTTP_CLIENT - .get(format!( - "{}/resource_types/list", - windmill_common::DEFAULT_HUB_BASE_URL - )) - .header("Accept", "application/json") - .send() - .await - .map_err(|e| error::Error::InternalErr(format!("Failed to fetch from hub: {}", e)))?; - - if !response.status().is_success() { - return Err(error::Error::InternalErr(format!( - "Hub returned status {}", - response.status() - ))); - } - - let raw_types: Vec = response - .json() - .await - .map_err(|e| error::Error::InternalErr(format!("Failed to parse hub response: {}", e)))?; - - Ok(raw_types - .into_iter() - .filter_map(|rt| { - let schema = match rt.schema { - Some(s) => match serde_json::from_str(&s) { - Ok(v) => Some(v), - Err(_) => return None, - }, - None => None, - }; - Some(CachedResourceType { - id: rt.id, - name: rt.name, - schema, - app: rt.app, - description: rt.description, - format_extension: Some(rt.format_extension), - display_name: rt.display_name, - }) - }) - .collect()) -} - #[derive(serde::Deserialize)] struct SyncResourceTypesQuery { name: Option, @@ -2677,118 +2589,45 @@ async fn sync_cached_resource_types( ) -> error::Result { require_super_admin(&db, &authed).await?; - use windmill_common::worker::HUB_RT_CACHE_DIR; - let cache_path = format!("{}/resource_types.json", *HUB_RT_CACHE_DIR); + let outcome = hub_resource_types::sync(&db).await?; - // Manual sync is hub-first so it lands newly-published hub types on demand. The - // on-disk cache is only a fallback for when the hub is unreachable (airgapped - // installs / network error); refreshing it is left to the daily cache-rt cron and - // the startup sync in main.rs, which own the offline path. - let (resource_types, from_hub) = match fetch_resource_types_from_hub().await { - Ok(types) => { - tracing::info!("Fetched {} resource types live from the hub", types.len()); - (types, true) - } - Err(hub_err) => { - tracing::warn!( - "Live hub fetch failed ({hub_err}), falling back to on-disk cache at {cache_path}" - ); - match tokio::fs::read_to_string(&cache_path).await { - Ok(content) => { - let parsed = serde_json::from_str::>(&content) - .map_err(|e| { - error::Error::InternalErr(format!( - "Failed to parse cached resource types: {}", - e - )) - })?; - (parsed, false) - } - Err(_) => return Err(hub_err), - } - } - }; - - let mut synced_count = 0; - - for rt in &resource_types { - // A name too long for the column counts as absent, leaving the stored one alone: one bad - // entry must not fail the upsert and end the rest of the sync. - let display_name = match &rt.display_name { - Some(Some(name)) if name.chars().count() > 100 => None, - other => other.clone(), + if !outcome.changed.is_empty() { + let source = match outcome.source { + SyncSource::Hub => "hub", + SyncSource::ImageCache => "image_cache", }; - let exists: Option = sqlx::query_scalar!( - "SELECT EXISTS(SELECT 1 FROM resource_type WHERE workspace_id = 'admins' AND name = $1 AND schema IS NOT DISTINCT FROM $2 AND description IS NOT DISTINCT FROM $3 AND ($5 IS NOT TRUE OR format_extension IS NOT DISTINCT FROM $4) AND ($7 IS NOT TRUE OR display_name IS NOT DISTINCT FROM $6))", - &rt.name, - rt.schema.as_ref(), - rt.description.as_deref(), - rt.format_extension.clone().flatten(), - rt.format_extension.is_some(), - display_name.clone().flatten(), - display_name.is_some(), + let changed = outcome.changed.join(", "); + audit_log( + &db, + &authed, + "resource_types.sync", + ActionKind::Update, + "admins", + None, + Some([("source", source), ("changed", changed.as_str())].into()), ) - .fetch_one(&db) .await?; - - if exists.unwrap_or(false) { - continue; - } - - sqlx::query!( - // Whether the payload carried the key at all is what decides: present - // (even as null) is authoritative and may clear, absent means a cache - // written before the column and must leave the stored value alone. - "INSERT INTO resource_type (workspace_id, name, schema, description, format_extension, display_name, edited_at) - VALUES ('admins', $1, $2, $3, $4, $6, now()) - ON CONFLICT (workspace_id, name) DO UPDATE - SET schema = EXCLUDED.schema, description = EXCLUDED.description, - -- A fileset is a set of files, so it cannot also be one file. - -- Create and update reject the pair; this writer bypasses both, so - -- it declines the extension rather than persisting the forbidden - -- combination onto a same-named local fileset. - format_extension = CASE - WHEN resource_type.is_fileset THEN NULL - WHEN $5 THEN EXCLUDED.format_extension - ELSE resource_type.format_extension END, - display_name = CASE WHEN $7 THEN EXCLUDED.display_name ELSE resource_type.display_name END, - edited_at = now()", - &rt.name, - rt.schema.as_ref(), - rt.description.as_deref(), - rt.format_extension.clone().flatten(), - rt.format_extension.is_some(), - display_name.clone().flatten(), - display_name.is_some(), - ) - .execute(&db) - .await?; - - synced_count += 1; } // If a specific type was requested and is still absent after syncing, surface an // explicit not-found instead of a silent "Synced 0". Word it by source so the // cache-fallback path does not claim it checked the hub. if let Some(name) = name.as_deref() { - if !resource_types.iter().any(|rt| rt.name == name) { - let source = if from_hub { - "on the hub" - } else { - "in the cached resource types (hub unreachable)" + if !outcome.listed.iter().any(|listed| listed == name) { + let source = match outcome.source { + SyncSource::Hub => "on the hub", + SyncSource::ImageCache => "in the cached resource types (hub unavailable)", }; return Err(error::Error::NotFound(format!( - "resource type '{}' not found {}", - name, source + "resource type '{name}' not found {source}" ))); } } - Ok(format!( - "Synced {} resource types ({} unchanged)", - synced_count, - resource_types.len() - synced_count - )) + Ok(match outcome.source { + SyncSource::Hub => outcome.summary(), + SyncSource::ImageCache => format!("{}, the hub being unavailable", outcome.summary()), + }) } #[cfg(test)] diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index da193c2f66..f9586529ac 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -31811,6 +31811,7 @@ components: - "resource_types.create" - "resource_types.update" - "resource_types.delete" + - "resource_types.sync" - "schedule.create" - "schedule.setenabled" - "schedule.edit" diff --git a/backend/windmill-api/src/live_migrations.rs b/backend/windmill-api/src/live_migrations.rs index 334ee340a8..a620807902 100644 --- a/backend/windmill-api/src/live_migrations.rs +++ b/backend/windmill-api/src/live_migrations.rs @@ -8,7 +8,9 @@ use sqlx::{PgConnection, Postgres}; use tokio::task::JoinHandle; -use windmill_common::{db::DB, error::Error}; +use windmill_common::{ + db::DB, error::Error, global_settings::SYNC_HUB_RESOURCE_TYPES_DAILY_SETTING, +}; use crate::db::CustomMigrator; use sqlx::migrate::Migrate; @@ -160,7 +162,11 @@ async fn run_background_migrations(db: &DB) -> Result<(), Error> { return Ok(()); } - for step in [AUDIT_OPERATION_INDEX, RETIRE_LEGACY_AUDIT] { + for step in [ + RETIRE_LEGACY_HUB_SYNC, + AUDIT_OPERATION_INDEX, + RETIRE_LEGACY_AUDIT, + ] { if background_migration_done(&mut conn, step).await? { continue; } @@ -171,6 +177,7 @@ async fn run_background_migrations(db: &DB) -> Result<(), Error> { let run = match step { AUDIT_OPERATION_INDEX => create_audit_operation_index(&mut conn).await, RETIRE_LEGACY_AUDIT => retire_legacy_audit_table(&mut conn).await, + RETIRE_LEGACY_HUB_SYNC => retire_legacy_hub_sync(&mut conn).await, _ => unreachable!("background migration {step} has no step function"), }; match run { @@ -192,6 +199,7 @@ async fn run_background_migrations(db: &DB) -> Result<(), Error> { const AUDIT_OPERATION_INDEX: &str = "audit_partitioned_workspace_operation_index"; const RETIRE_LEGACY_AUDIT: &str = "retire_legacy_audit_table"; +const RETIRE_LEGACY_HUB_SYNC: &str = "retire_legacy_hub_sync_script"; // Short on purpose: a statement waiting for its lock is also a wait for every audit insert // queued behind it. @@ -371,6 +379,70 @@ async fn retire_legacy_audit_table(conn: &mut PgConnection) -> Result<(), Error> Ok(()) } +/// The hash the migrations gave `u/admin/hub_sync`; they rewrote its content in place. +const LEGACY_HUB_SYNC_HASH: i64 = -28028598712388162; + +/// `u/admin/hub_sync` and the schedules running it synced resource types before servers did +/// it themselves (`windmill_common::hub_resource_types`). Retired only while the script is +/// still the version Windmill created: a copy someone edited is theirs to keep running. +async fn retire_legacy_hub_sync(conn: &mut PgConnection) -> Result<(), Error> { + let mut tx = conn.begin().await?; + let latest: Option = sqlx::query_scalar( + "SELECT hash FROM script + WHERE workspace_id = 'admins' AND path = 'u/admin/hub_sync' AND archived = false + ORDER BY created_at DESC LIMIT 1", + ) + .fetch_optional(&mut *tx) + .await?; + if latest != Some(LEGACY_HUB_SYNC_HASH) { + return Ok(()); + } + + let schedules: Vec<(String, bool, serde_json::Value)> = sqlx::query_as( + "SELECT path, enabled, jsonb_build_object('row', to_jsonb(s)) FROM schedule s + WHERE workspace_id = 'admins' AND script_path = 'u/admin/hub_sync' AND NOT is_flow", + ) + .fetch_all(&mut *tx) + .await?; + // An enabled schedule hands over to the server's daily sync in the transaction that deletes + // it, so the instance never has both syncs running, nor neither. + if schedules.iter().any(|(_, enabled, _)| *enabled) { + let inserted = sqlx::query( + "INSERT INTO global_settings (name, value) VALUES ($1, 'true'::jsonb) + ON CONFLICT (name) DO NOTHING", + ) + .bind(SYNC_HUB_RESOURCE_TYPES_DAILY_SETTING) + .execute(&mut *tx) + .await? + .rows_affected(); + if inserted > 0 { + tracing::info!( + "u/admin/hub_sync ran on an enabled schedule: turned on {SYNC_HUB_RESOURCE_TYPES_DAILY_SETTING}" + ); + } + } + for (path, _, row) in schedules { + windmill_queue::schedule::clear_schedule(&mut tx, &path, "admins").await?; + sqlx::query("DELETE FROM schedule WHERE workspace_id = 'admins' AND path = $1") + .bind(&path) + .execute(&mut *tx) + .await?; + windmill_common::trashbin::move_to_trash( + &mut *tx, "admins", "schedule", &path, row, "system", + ) + .await?; + tracing::info!("Deleted schedule {path}, which ran u/admin/hub_sync"); + } + sqlx::query( + "UPDATE script SET archived = true WHERE workspace_id = 'admins' AND path = 'u/admin/hub_sync'", + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; + tracing::info!("Archived u/admin/hub_sync: servers sync resource types themselves"); + Ok(()) +} + #[cfg(test)] mod tests { use super::*; @@ -412,4 +484,80 @@ mod tests { assert_eq!(seen, 2); Ok(()) } + + async fn insert_legacy_hub_sync_schedule(db: &DB, enabled: bool) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO schedule (workspace_id, path, edited_by, email, permissioned_as, schedule, timezone, script_path, is_flow, enabled) + VALUES ('admins', 'g/all/hub_sync', 'admin', 'admin@windmill.dev', 'u/admin', '0 0 0 * * *', 'Etc/UTC', 'u/admin/hub_sync', false, $1)", + ) + .bind(enabled) + .execute(db) + .await?; + Ok(()) + } + + async fn schedules_and_live_versions(db: &DB) -> anyhow::Result<(i64, i64)> { + Ok(sqlx::query_as( + "SELECT (SELECT count(*) FROM schedule WHERE workspace_id = 'admins'), + (SELECT count(*) FROM script WHERE path = 'u/admin/hub_sync' AND NOT archived)", + ) + .fetch_one(db) + .await?) + } + + async fn daily_sync_setting(db: &DB) -> anyhow::Result> { + Ok( + sqlx::query_scalar("SELECT value FROM global_settings WHERE name = $1") + .bind(SYNC_HUB_RESOURCE_TYPES_DAILY_SETTING) + .fetch_optional(db) + .await?, + ) + } + + #[sqlx::test(migrations = "../migrations")] + async fn retire_legacy_hub_sync_spares_an_edited_script(db: DB) -> anyhow::Result<()> { + insert_legacy_hub_sync_schedule(&db, true).await?; + sqlx::query( + "INSERT INTO script (workspace_id, hash, path, content, created_by, language, summary, description, schema, lock, created_at) + SELECT workspace_id, 42, path, content || '// edited', 'admin', language, summary, description, schema, lock, now() + interval '1 minute' + FROM script WHERE workspace_id = 'admins' AND path = 'u/admin/hub_sync'", + ) + .execute(&db) + .await?; + + retire_legacy_hub_sync(&mut *db.acquire().await?).await?; + assert_eq!(schedules_and_live_versions(&db).await?, (1, 2)); + assert_eq!(daily_sync_setting(&db).await?, None); + + // Unedited, the enabled schedule hands over to the server's daily sync. + sqlx::query("DELETE FROM script WHERE hash = 42") + .execute(&db) + .await?; + retire_legacy_hub_sync(&mut *db.acquire().await?).await?; + assert_eq!(schedules_and_live_versions(&db).await?, (0, 0)); + assert_eq!( + daily_sync_setting(&db).await?, + Some(serde_json::json!(true)) + ); + Ok(()) + } + + #[sqlx::test(migrations = "../migrations")] + async fn retire_legacy_hub_sync_leaves_the_daily_sync_off_without_an_enabled_schedule( + db: DB, + ) -> anyhow::Result<()> { + for disabled_schedule in [true, false] { + sqlx::query("UPDATE script SET archived = false WHERE path = 'u/admin/hub_sync'") + .execute(&db) + .await?; + if disabled_schedule { + insert_legacy_hub_sync_schedule(&db, false).await?; + } + + retire_legacy_hub_sync(&mut *db.acquire().await?).await?; + assert_eq!(schedules_and_live_versions(&db).await?, (0, 0)); + assert_eq!(daily_sync_setting(&db).await?, None); + } + Ok(()) + } } diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index 1cd3609820..2bc303f5c2 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -158,6 +158,7 @@ pub const SSO_GROUPS_CLAIM_SETTING: &str = "sso_groups_claim"; pub const HUB_BASE_URL_SETTING: &str = "hub_base_url"; pub const HUB_ACCESSIBLE_URL_SETTING: &str = "hub_accessible_url"; pub const DISABLE_HUB_SETTING: &str = "disable_hub"; +pub const SYNC_HUB_RESOURCE_TYPES_DAILY_SETTING: &str = "sync_hub_resource_types_daily"; pub const CRITICAL_ERROR_CHANNELS_SETTING: &str = "critical_error_channels"; pub const CRITICAL_ALERT_MUTE_UI_SETTING: &str = "critical_alert_mute_ui"; pub const CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING: &str = "critical_alerts_on_db_oversize"; diff --git a/backend/windmill-common/src/hub_resource_types.rs b/backend/windmill-common/src/hub_resource_types.rs new file mode 100644 index 0000000000..88de520094 --- /dev/null +++ b/backend/windmill-common/src/hub_resource_types.rs @@ -0,0 +1,578 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2022 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +//! Keeps the `admins` workspace's resource types, which every workspace falls back to, in +//! step with the hub. Servers run [`spawn_daily_sync`] once a superadmin turns it on, +//! superadmins run [`sync`] on demand, and `windmill cache-rt` bakes the public hub's listing +//! into the image for instances that cannot reach a hub. + +use std::collections::HashMap; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::json; + +use crate::{ + error::{Error, Result}, + global_settings::{ + load_value_from_global_settings, DISABLE_HUB_SETTING, SYNC_HUB_RESOURCE_TYPES_DAILY_SETTING, + }, + utils::{get_license_id_or_uid, HTTP_CLIENT_PERMISSIVE, HUB_API_SECRET}, + worker::HUB_RT_CACHE_DIR, + DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, +}; + +const SYNC_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::hours(24); +const RETRY_INTERVAL: chrono::TimeDelta = chrono::TimeDelta::hours(1); +const CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(600); +const SYNC_LOCK_ID: i64 = 737_483_925; +/// `background_task_state.name` of the sync's last attempt and last hub sync. +const SYNC_STATE_TASK: &str = "hub_resource_type_sync"; + +pub fn cache_path() -> String { + format!("{}/resource_types.json", *HUB_RT_CACHE_DIR) +} + +/// A hub resource type with its schema parsed, as synced and as `cache-rt` stores it. +/// +/// `format_extension` and `display_name` are doubly optional: no key (a cache written before +/// the field) leaves the stored value alone, while an explicit null clears it. Plain serde +/// folds both into `None`, hence the wrapping deserializer. +#[derive(Deserialize, Serialize, Clone)] +pub struct HubResourceType { + pub name: String, + pub schema: Option, + pub description: Option, + #[serde( + default, + deserialize_with = "crate::more_serde::double_option", + skip_serializing_if = "Option::is_none" + )] + pub format_extension: Option>, + #[serde( + default, + deserialize_with = "crate::more_serde::double_option", + skip_serializing_if = "Option::is_none" + )] + pub display_name: Option>, + /// Absent from hubs that do not publish filesets, which leaves the stored flag alone. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub is_fileset: Option, +} + +/// An entry of the hub's `/resource_types/list`, whose schema is a JSON string. +#[derive(Deserialize)] +struct HubListing { + name: String, + schema: Option, + description: Option, + /// Absent from hubs predating the column, which do not publish file types either, so + /// the hub is authoritative for it whether or not it sends the key. + #[serde(default)] + format_extension: Option, + #[serde(default, deserialize_with = "crate::more_serde::double_option")] + display_name: Option>, + #[serde(default)] + is_fileset: Option, +} + +/// One hub's listing. A type whose schema does not parse is left out. +pub async fn fetch( + hub_base_url: &str, + api_secret: Option<&str>, + uid: Option<&str>, +) -> Result> { + // Permissive, so `ACCEPT_INVALID_CERTS` lets the sync reach a private hub behind a + // self-signed certificate. + let mut request = HTTP_CLIENT_PERMISSIVE + .get(format!("{hub_base_url}/resource_types/list")) + .header("Accept", "application/json"); + if let Some(uid) = uid { + request = request.header("X-uid", uid); + } + if let Some(secret) = api_secret { + request = request.header("X-api-secret", secret); + } + let response = request.send().await.map_err(|e| { + Error::InternalErr(format!("Failed to reach the hub at {hub_base_url}: {e}")) + })?; + let status = response.status(); + if status == reqwest::StatusCode::UNAUTHORIZED { + return Err(Error::InternalErr(format!( + "The hub at {hub_base_url} refused the request: check the Private Hub API secret" + ))); + } + if !status.is_success() { + return Err(Error::InternalErr(format!( + "The hub at {hub_base_url} answered {status}" + ))); + } + let listing: Vec = response.json().await.map_err(|e| { + Error::InternalErr(format!( + "Failed to parse the resource types listed by {hub_base_url}: {e}" + )) + })?; + + Ok(listing + .into_iter() + .filter_map(|rt| { + let schema = match rt.schema.as_deref().map(serde_json::from_str).transpose() { + Ok(schema) => schema, + Err(e) => { + tracing::warn!( + "Skipping hub resource type {}: invalid schema: {e}", + rt.name + ); + return None; + } + }; + Some(HubResourceType { + name: rt.name, + schema, + description: rt.description, + format_extension: Some(rt.format_extension), + display_name: rt.display_name, + is_fileset: rt.is_fileset, + }) + }) + .collect()) +} + +/// The listing of the hub this instance is configured with. +async fn fetch_configured_hub(db: &DB) -> Result> { + let hub = (**HUB_BASE_URL.load()).clone(); + let uid = get_license_id_or_uid(db).await.ok(); + if hub == DEFAULT_HUB_BASE_URL { + return fetch(&hub, None, uid.as_deref()).await; + } + let secret = (**HUB_API_SECRET.load()).clone(); + let types = fetch(&hub, secret.as_deref(), uid.as_deref()).await?; + if !types.is_empty() { + return Ok(types); + } + // A private hub that publishes no resource types of its own defers to the public one, + // which must not be sent the private hub's secret. + tracing::info!("The hub at {hub} lists no resource types, syncing the public hub's"); + fetch(DEFAULT_HUB_BASE_URL, None, uid.as_deref()).await +} + +#[derive(PartialEq, Debug)] +struct ResourceTypeRow { + schema: Option, + description: Option, + format_extension: Option, + is_fileset: bool, + display_name: Option, +} + +/// The row a synced type should end up as, given what `admins` already stores for it. +fn target_row(rt: &HubResourceType, stored: Option<&ResourceTypeRow>) -> ResourceTypeRow { + let is_fileset = rt + .is_fileset + .unwrap_or_else(|| stored.is_some_and(|s| s.is_fileset)); + // A fileset is a set of files, so it cannot also be one file. Create and update reject + // the pair; this writer bypasses both. + let format_extension = if is_fileset { + None + } else { + match &rt.format_extension { + Some(extension) => extension.clone(), + None => stored.and_then(|s| s.format_extension.clone()), + } + }; + // A name too long for the column counts as absent: one bad entry must not fail the + // upsert and end the rest of the sync. + let display_name = match &rt.display_name { + Some(Some(name)) if name.chars().count() > 100 => { + tracing::warn!( + "Ignoring the display_name of resource type {}: longer than 100 characters", + rt.name + ); + stored.and_then(|s| s.display_name.clone()) + } + Some(name) => name.clone(), + None => stored.and_then(|s| s.display_name.clone()), + }; + ResourceTypeRow { + schema: rt.schema.clone(), + description: rt.description.clone(), + format_extension, + is_fileset, + display_name, + } +} + +#[derive(Clone, Copy, PartialEq, Debug)] +pub enum SyncSource { + Hub, + ImageCache, +} + +pub struct SyncOutcome { + pub source: SyncSource, + /// Every type the source listed, changed or not. + pub listed: Vec, + /// The types this sync created or updated. + pub changed: Vec, +} + +impl SyncOutcome { + pub fn summary(&self) -> String { + let synced = self.changed.len(); + let unchanged = self.listed.len() - synced; + match self.source { + SyncSource::Hub => { + format!("Synced {synced} resource types from the hub ({unchanged} unchanged)") + } + SyncSource::ImageCache => format!( + "Synced {synced} resource types from the image's cache ({unchanged} unchanged)" + ), + } + } +} + +async fn apply(db: &DB, source: SyncSource, types: &[HubResourceType]) -> Result { + let stored: HashMap = sqlx::query_as::< + _, + ( + String, + Option, + Option, + Option, + bool, + Option, + ), + >( + "SELECT name, schema, description, format_extension, is_fileset, display_name + FROM resource_type WHERE workspace_id = 'admins'", + ) + .fetch_all(db) + .await? + .into_iter() + .map( + |(name, schema, description, format_extension, is_fileset, display_name)| { + ( + name, + ResourceTypeRow { schema, description, format_extension, is_fileset, display_name }, + ) + }, + ) + .collect(); + + let mut changed = vec![]; + for rt in types { + let current = stored.get(&rt.name); + let target = target_row(rt, current); + if current == Some(&target) { + continue; + } + sqlx::query( + "INSERT INTO resource_type + (workspace_id, name, schema, description, format_extension, is_fileset, display_name, edited_at) + VALUES ('admins', $1, $2, $3, $4, $5, $6, now()) + ON CONFLICT (workspace_id, name) DO UPDATE + SET schema = EXCLUDED.schema, description = EXCLUDED.description, + format_extension = EXCLUDED.format_extension, is_fileset = EXCLUDED.is_fileset, + display_name = EXCLUDED.display_name, edited_at = now()", + ) + .bind(&rt.name) + .bind(&target.schema) + .bind(&target.description) + .bind(&target.format_extension) + .bind(target.is_fileset) + .bind(&target.display_name) + .execute(db) + .await?; + changed.push(rt.name.clone()); + } + + Ok(SyncOutcome { source, listed: types.iter().map(|rt| rt.name.clone()).collect(), changed }) +} + +#[derive(Deserialize, Default)] +struct SyncState { + last_attempt_at: Option>, + last_hub_sync_at: Option>, +} + +async fn load_state(db: &DB) -> Result { + let value: Option = + sqlx::query_scalar("SELECT value FROM background_task_state WHERE name = $1") + .bind(SYNC_STATE_TASK) + .fetch_optional(db) + .await?; + Ok(value + .and_then(|v| serde_json::from_value(v).ok()) + .unwrap_or_default()) +} + +/// Merged into the stored state, so a writer only ever sets the keys it names. +async fn record_state(db: &DB, patch: serde_json::Value) -> Result<()> { + sqlx::query( + "INSERT INTO background_task_state (name, value) VALUES ($1, $2) + ON CONFLICT (name) DO UPDATE + SET value = background_task_state.value || EXCLUDED.value, updated_at = now()", + ) + .bind(SYNC_STATE_TASK) + .bind(patch) + .execute(db) + .await?; + Ok(()) +} + +async fn setting_enabled(db: &DB, name: &str) -> Result { + Ok(load_value_from_global_settings(db, name) + .await? + .and_then(|v| v.as_bool()) + .unwrap_or(false)) +} + +/// Syncs from the configured hub and records the attempt, successful or not. +async fn sync_from_hub(db: &DB) -> Result { + let attempted_at = Utc::now(); + let result = match fetch_configured_hub(db).await { + Ok(types) => apply(db, SyncSource::Hub, &types).await, + Err(e) => Err(e), + }; + let state = match &result { + Ok(_) => { + json!({ "last_attempt_at": attempted_at, "last_hub_sync_at": attempted_at, "last_error": null }) + } + Err(e) => json!({ "last_attempt_at": attempted_at, "last_error": e.to_string() }), + }; + record_state(db, state).await?; + result +} + +/// Syncs from the configured hub, falling back to the image's cache when the hub is disabled +/// or unreachable and the instance has never synced from a hub. +pub async fn sync(db: &DB) -> Result { + let hub_err = if setting_enabled(db, DISABLE_HUB_SETTING).await? { + Error::BadRequest("The hub is disabled on this instance".to_string()) + } else { + match sync_from_hub(db).await { + Ok(outcome) => return Ok(outcome), + Err(e) => e, + } + }; + match fall_back_to_image_cache(db, &cache_path()).await? { + Some(outcome) => { + tracing::warn!("Hub resource type sync failed, used the image's cache: {hub_err}"); + Ok(outcome) + } + None => Err(hub_err), + } +} + +/// The cache is the listing as of the image's build, so standing in for the hub on an +/// instance that has synced from one would revert every type the hub has updated since. +async fn fall_back_to_image_cache(db: &DB, path: &str) -> Result> { + if let Some(at) = load_state(db).await?.last_hub_sync_at { + tracing::info!( + "Not applying the image's cached resource types: synced from the hub at {at}" + ); + return Ok(None); + } + sync_from_cache_file(db, path).await +} + +/// Applies the listing baked into the image (`SYNC_CACHED_RT`), on every start and whether or +/// not the instance has synced from a hub: it is how an instance that cannot reach the public +/// hub, a private hub listing only its own types included, takes each new image's public types. +/// `None` when the image carries no cache. +pub async fn sync_from_image_cache(db: &DB) -> Result> { + sync_from_cache_file(db, &cache_path()).await +} + +async fn sync_from_cache_file(db: &DB, path: &str) -> Result> { + let content = match tokio::fs::read_to_string(path).await { + Ok(content) => content, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + tracing::info!("No cached resource types at {path}"); + return Ok(None); + } + Err(e) => { + return Err(Error::InternalErr(format!( + "Failed to read cached resource types at {path}: {e}" + ))) + } + }; + let types: Vec = serde_json::from_str(&content).map_err(|e| { + Error::InternalErr(format!( + "Failed to parse cached resource types at {path}: {e}" + )) + })?; + Ok(Some(apply(db, SyncSource::ImageCache, &types).await?)) +} + +/// While `sync_hub_resource_types_daily` is on, syncs once a day, from whichever server first +/// finds the sync due. Off by default: each sync overwrites local edits to the `admins` types +/// the hub also defines. The state it reads lives in the database, so servers starting or +/// restarting do not each sync again. +pub fn spawn_daily_sync(db: DB, mut killpill_rx: tokio::sync::broadcast::Receiver<()>) { + tokio::spawn(async move { + // Leaves time for the hub URL and secret settings to load, and spreads out servers + // that start together. + let mut wait = std::time::Duration::from_secs(rand::random_range(30..90)); + loop { + tokio::select! { + _ = tokio::time::sleep(wait) => {} + _ = killpill_rx.recv() => return, + } + if let Err(e) = sync_if_due(&db).await { + tracing::error!("Hub resource type sync: {e:#}"); + } + wait = CHECK_INTERVAL; + } + }); +} + +async fn sync_if_due(db: &DB) -> Result<()> { + if !setting_enabled(db, SYNC_HUB_RESOURCE_TYPES_DAILY_SETTING).await? + || setting_enabled(db, DISABLE_HUB_SETTING).await? + { + return Ok(()); + } + // A transaction-scoped lock: a dropped transaction rolls back and releases it, so a + // pass cut short cannot leave it held on a pooled connection. The transaction only + // owns the lock; the sync runs on other connections. + let mut lock_tx = db.begin().await?; + let locked: bool = sqlx::query_scalar("SELECT pg_try_advisory_xact_lock($1)") + .bind(SYNC_LOCK_ID) + .fetch_one(&mut *lock_tx) + .await?; + if !locked { + return Ok(()); + } + let state = load_state(db).await?; + let now = Utc::now(); + let due = state + .last_hub_sync_at + .is_none_or(|at| now - at >= SYNC_INTERVAL) + && state + .last_attempt_at + .is_none_or(|at| now - at >= RETRY_INTERVAL); + // Hub only. Falling back to the image's cache here would re-apply the same listing every + // hour on an instance that never reaches a hub, over any edit made since; the cache is + // applied at boot (`SYNC_CACHED_RT`) or on demand instead. + if due { + match sync_from_hub(db).await { + Ok(outcome) if outcome.changed.is_empty() => tracing::info!("{}", outcome.summary()), + Ok(outcome) => { + tracing::info!("{}: {}", outcome.summary(), outcome.changed.join(", ")) + } + Err(e) => tracing::warn!("Hub resource type sync failed, retrying in an hour: {e}"), + } + } + lock_tx.rollback().await?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn hub_type( + format_extension: Option>, + display_name: Option>, + is_fileset: Option, + ) -> HubResourceType { + HubResourceType { + name: "t".to_string(), + schema: None, + description: None, + format_extension: format_extension.map(|e| e.map(str::to_string)), + display_name: display_name.map(|n| n.map(str::to_string)), + is_fileset, + } + } + + fn stored( + format_extension: Option<&str>, + display_name: Option<&str>, + is_fileset: bool, + ) -> ResourceTypeRow { + ResourceTypeRow { + schema: None, + description: None, + format_extension: format_extension.map(str::to_string), + is_fileset, + display_name: display_name.map(str::to_string), + } + } + + #[test] + fn absent_keys_keep_the_stored_values_and_nulls_clear_them() { + let current = stored(Some("csv"), Some("T"), false); + let kept = target_row(&hub_type(None, None, None), Some(¤t)); + assert_eq!(kept, current); + let cleared = target_row(&hub_type(Some(None), Some(None), None), Some(¤t)); + assert_eq!(cleared, stored(None, None, false)); + } + + #[test] + fn a_fileset_never_takes_a_format_extension() { + let local_fileset = stored(None, None, true); + let row = target_row( + &hub_type(Some(Some("csv")), None, None), + Some(&local_fileset), + ); + assert_eq!(row, stored(None, None, true)); + let row = target_row(&hub_type(Some(Some("csv")), None, Some(true)), None); + assert_eq!(row, stored(None, None, true)); + } + + #[sqlx::test(migrations = "../migrations")] + async fn only_sync_cached_rt_applies_the_image_cache_after_a_hub_sync( + db: DB, + ) -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let path = dir.path().join("resource_types.json"); + std::fs::write( + &path, + json!([{ "name": "cached_type", "schema": null, "description": "image copy" }]) + .to_string(), + )?; + let path = path.to_str().unwrap(); + let description = || { + sqlx::query_scalar::<_, Option>( + "SELECT description FROM resource_type WHERE workspace_id = 'admins' AND name = 'cached_type'", + ) + .fetch_one(&db) + }; + + assert!(fall_back_to_image_cache(&db, path).await?.is_some()); + assert_eq!(description().await?.as_deref(), Some("image copy")); + + record_state(&db, json!({ "last_hub_sync_at": Utc::now() })).await?; + sqlx::query( + "UPDATE resource_type SET description = 'hub copy' WHERE workspace_id = 'admins' AND name = 'cached_type'", + ) + .execute(&db) + .await?; + assert!(fall_back_to_image_cache(&db, path).await?.is_none()); + assert_eq!(description().await?.as_deref(), Some("hub copy")); + + assert!(sync_from_cache_file(&db, path).await?.is_some()); + assert_eq!(description().await?.as_deref(), Some("image copy")); + Ok(()) + } + + #[sqlx::test(migrations = "../migrations")] + async fn no_daily_sync_unless_turned_on(db: DB) -> anyhow::Result<()> { + sync_if_due(&db).await?; + + let state: Option = + sqlx::query_scalar("SELECT value FROM background_task_state WHERE name = $1") + .bind(SYNC_STATE_TASK) + .fetch_optional(&db) + .await?; + assert_eq!(state, None); + Ok(()) + } +} diff --git a/backend/windmill-common/src/instance_config.rs b/backend/windmill-common/src/instance_config.rs index bd7c28628c..3d50558c7f 100644 --- a/backend/windmill-common/src/instance_config.rs +++ b/backend/windmill-common/src/instance_config.rs @@ -267,6 +267,8 @@ pub struct GlobalSettings { #[serde(skip_serializing_if = "Option::is_none")] pub disable_hub: Option, #[serde(skip_serializing_if = "Option::is_none")] + pub sync_hub_resource_types_daily: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub auto_build_binary_on_deploy: Option, #[serde(skip_serializing_if = "Option::is_none")] pub ai_sessions_instance_storage_fallback: Option, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 2c94d95618..5ccc8beff1 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -75,6 +75,7 @@ pub mod flow_status; pub mod flows; pub mod folders; pub mod global_settings; +pub mod hub_resource_types; pub mod guest_jwt; pub mod indexer; pub mod instance_config; diff --git a/frontend/e2e/global-setup.ts b/frontend/e2e/global-setup.ts index db27741a5a..6d10c53d92 100644 --- a/frontend/e2e/global-setup.ts +++ b/frontend/e2e/global-setup.ts @@ -47,18 +47,10 @@ async function globalSetup(config: FullConfig) { await page.waitForURL('**/') if (!process.env.SKIP_HUB_SYNC) { - await page.locator('#home-search-input').waitFor({ state: 'visible' }) - await page.locator('#home-search-input').fill('u/admin/hub_sync') - - const hubSyncBtn = page.locator('text=u/admin/hub_sync').first() - await hubSyncBtn.waitFor({ state: 'visible' }) - await hubSyncBtn.click() - - const runBtn = page.locator('#run-form-run-button') - await runBtn.waitFor({ state: 'visible' }) - await runBtn.click() - await page.waitForURL(/\/run\/.+/) - await page.locator('text=Success').waitFor({ timeout: 30000, state: 'visible' }) + const res = await page.request.post(`${baseURL}/api/settings/sync_cached_resource_types`) + if (!res.ok()) { + throw new Error(`Resource type sync failed: ${await res.text()}`) + } } // Save the authenticated state await context.storageState({ path: './e2e/auth.json' }) diff --git a/frontend/src/lib/components/instanceSettings.ts b/frontend/src/lib/components/instanceSettings.ts index 75dada92ab..867570e1a1 100644 --- a/frontend/src/lib/components/instanceSettings.ts +++ b/frontend/src/lib/components/instanceSettings.ts @@ -377,6 +377,15 @@ export const settings: Record = { value === null || value === '' || parseAccentColor(value) !== undefined + }, + { + label: 'Sync resource types from the hub daily', + description: + 'Every day, the server updates the resource types of the admins workspace from the Hub, overwriting local edits to the types the Hub also defines. Superadmins can also sync them on demand with "Sync resource types with Hub".', + key: 'sync_hub_resource_types_daily', + fieldType: 'boolean', + storage: 'setting', + hideInQuickSetup: true } ], Jobs: [ diff --git a/frontend/src/lib/components/instanceSettings/InstanceAISettings.svelte b/frontend/src/lib/components/instanceSettings/InstanceAISettings.svelte index 8d7d107940..2f2145da9b 100644 --- a/frontend/src/lib/components/instanceSettings/InstanceAISettings.svelte +++ b/frontend/src/lib/components/instanceSettings/InstanceAISettings.svelte @@ -1,5 +1,5 @@