starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-12 08:02:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: replace the hub_sync script with an opt-in daily server sync (#11624)

Browse Source 
* feat: sync hub resource types daily from the server

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat: retire the legacy hub_sync script and its schedule

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: sync only from the hub on the daily schedule

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: sync resource types through the endpoint in e2e setup

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: honor ACCEPT_INVALID_CERTS in the hub resource type sync

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat: audit manual hub resource type syncs that change types

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor: show the AI step's resource type sync as a setting card

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: keep the hub_sync opt-out and move sync state to task state

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: hold the daily hub sync until the legacy hub_sync is retired

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: turn off the daily sync when an edited hub_sync is scheduled

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor: make the daily hub resource type sync opt-in

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat: bring back the setup wizard's daily resource type sync toggle

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: apply the image's resource types on every SYNC_CACHED_RT start

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
hugocasa
2026-10-09 18:32:05 +02:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 338c4e921d
commit 2b55a8afbc
17 changed files with 846 additions and 624 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-212bf5b32de102a9c537907aaff04befc3a6596c258086805dd42f513ddb3ead.json
Generated
-20
View File
@@ -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"
}
backend/.sqlx/query-386e14cf7572027f2c4ef313cd7cc5dd6c7b0da4f76131313d3e6d97f36c5c34.json
Generated
-28
View File
@@ -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"
}
backend/src/main.rs
+15 -243
View File
@@ -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<String>) -> 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<String>,
pub app: String,
pub description: Option<String>,
/// Absent from hubs predating the column, and from caches written before it.
#[serde(default)]
pub format_extension: Option<String>,
/// 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<Option<String>>,
}
/// Processed resource type with parsed schema
#[derive(serde::Deserialize, serde::Serialize, Clone)]
pub struct HubResourceType {
pub id: i64,
pub name: String,
pub schema: Option<serde_json::Value>,
pub app: String,
pub description: Option<String>,
/// 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<Option<String>>,
/// 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<Option<String>>,
}
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<HubResourceTypeRaw> = response
.json::<Vec<HubResourceTypeRaw>>()
.await
.with_context(|| "Failed to parse resource types from hub")?;
// Parse schema strings into JSON values
let resource_types: Vec<HubResourceType> = 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<sqlx::Postgres>) -> 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<HubResourceType> =
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<serde_json::Value>,
Option<String>,
Option<String>,
bool,
Option<String>,
)> = 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<serde_json::Value>,
Option<String>,
Option<String>,
bool,
Option<String>,
),
> = 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'"
backend/tests/wm_token_confinement.rs
+1 -1
View File
@@ -319,7 +319,7 @@ async fn test_wm_token_is_confined_to_its_workspace(db: Pool<Postgres>) -> 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(
backend/windmill-api-auth/src/scopes.rs
+2 -1
View File
@@ -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,
backend/windmill-api-scripts/src/scripts.rs
-12
View File
@@ -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),
backend/windmill-api-settings/src/lib.rs
+24 -185
View File
@@ -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<JwksResponse> {
Ok(Json(JwksResponse { keys: vec![] }))
}
#[derive(serde::Deserialize, serde::Serialize)]
struct CachedResourceType {
#[allow(dead_code)]
id: i64,
name: String,
schema: Option<serde_json::Value>,
#[allow(dead_code)]
app: String,
description: Option<String>,
/// 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<Option<String>>,
/// 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<Option<String>>,
}
#[derive(serde::Deserialize)]
struct HubResourceTypeRaw {
id: i64,
name: String,
schema: Option<String>,
app: String,
description: Option<String>,
#[serde(default)]
format_extension: Option<String>,
#[serde(
default,
deserialize_with = "windmill_common::more_serde::double_option"
)]
display_name: Option<Option<String>>,
}
async fn fetch_resource_types_from_hub() -> error::Result<Vec<CachedResourceType>> {
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<HubResourceTypeRaw> = 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<String>,
@@ -2677,118 +2589,45 @@ async fn sync_cached_resource_types(
) -> error::Result<String> {
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::<Vec<CachedResourceType>>(&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<bool> = 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)]
backend/windmill-api/openapi.yaml
+1
View File
@@ -31811,6 +31811,7 @@ components:
- "resource_types.create"
- "resource_types.update"
- "resource_types.delete"
- "resource_types.sync"
- "schedule.create"
- "schedule.setenabled"
- "schedule.edit"
backend/windmill-api/src/live_migrations.rs
+150 -2
View File
@@ -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<i64> = 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<Option<serde_json::Value>> {
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(())
}
}
backend/windmill-common/src/global_settings.rs
+1
View File
@@ -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";
backend/windmill-common/src/hub_resource_types.rs
+578
View File
@@ -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<serde_json::Value>,
pub description: Option<String>,
#[serde(
default,
deserialize_with = "crate::more_serde::double_option",
skip_serializing_if = "Option::is_none"
)]
pub format_extension: Option<Option<String>>,
#[serde(
default,
deserialize_with = "crate::more_serde::double_option",
skip_serializing_if = "Option::is_none"
)]
pub display_name: Option<Option<String>>,
/// 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<bool>,
}
/// An entry of the hub's `/resource_types/list`, whose schema is a JSON string.
#[derive(Deserialize)]
struct HubListing {
name: String,
schema: Option<String>,
description: Option<String>,
/// 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<String>,
#[serde(default, deserialize_with = "crate::more_serde::double_option")]
display_name: Option<Option<String>>,
#[serde(default)]
is_fileset: Option<bool>,
}
/// 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<Vec<HubResourceType>> {
// 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<HubListing> = 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<Vec<HubResourceType>> {
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<serde_json::Value>,
description: Option<String>,
format_extension: Option<String>,
is_fileset: bool,
display_name: Option<String>,
}
/// 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<String>,
/// The types this sync created or updated.
pub changed: Vec<String>,
}
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<SyncOutcome> {
let stored: HashMap<String, ResourceTypeRow> = sqlx::query_as::<
_,
(
String,
Option<serde_json::Value>,
Option<String>,
Option<String>,
bool,
Option<String>,
),
>(
"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<DateTime<Utc>>,
last_hub_sync_at: Option<DateTime<Utc>>,
}
async fn load_state(db: &DB) -> Result<SyncState> {
let value: Option<serde_json::Value> =
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<bool> {
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<SyncOutcome> {
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<SyncOutcome> {
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<Option<SyncOutcome>> {
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<Option<SyncOutcome>> {
sync_from_cache_file(db, &cache_path()).await
}
async fn sync_from_cache_file(db: &DB, path: &str) -> Result<Option<SyncOutcome>> {
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<HubResourceType> = 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<Option<&str>>,
display_name: Option<Option<&str>>,
is_fileset: Option<bool>,
) -> 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(&current));
assert_eq!(kept, current);
let cleared = target_row(&hub_type(Some(None), Some(None), None), Some(&current));
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<String>>(
"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<serde_json::Value> =
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(())
}
}
backend/windmill-common/src/instance_config.rs
+2
View File
@@ -267,6 +267,8 @@ pub struct GlobalSettings {
#[serde(skip_serializing_if = "Option::is_none")]
pub disable_hub: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub sync_hub_resource_types_daily: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub auto_build_binary_on_deploy: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub ai_sessions_instance_storage_fallback: Option<bool>,
backend/windmill-common/src/lib.rs
+1
View File
@@ -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;
frontend/e2e/global-setup.ts
+4 -12
View File
@@ -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' })
frontend/src/lib/components/instanceSettings.ts
+9
View File
@@ -377,6 +377,15 @@ export const settings: Record<string, Setting[]> = {
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: [
frontend/src/lib/components/instanceSettings/InstanceAISettings.svelte
+29 -39
View File
@@ -1,5 +1,5 @@
<script lang="ts">
import { JobService, SettingService, WorkspaceService, type AIConfig } from '$lib/gen'
import { SettingService, WorkspaceService, type AIConfig } from '$lib/gen'
import { setCopilotInfo } from '$lib/aiStore'
import { workspaceStore, userStore } from '$lib/stores'
import { getUserExt } from '$lib/user'
@@ -7,6 +7,7 @@
import { workspaceAIClients } from '../copilot/lib'
import AISettings from '../workspaceSettings/AISettings.svelte'
import { Alert, Button } from '../common'
import SettingCard from './SettingCard.svelte'
interface Props {
hasUnsavedChanges?: boolean
@@ -80,20 +81,15 @@
hubSyncStatus = 'loading'
hubSyncMessage = ''
try {
await JobService.runWaitResultScriptByPath({
workspace: 'admins',
path: 'u/admin/hub_sync',
requestBody: {}
})
const res = await fetch('/api/settings/sync_cached_resource_types', { method: 'POST' })
if (!res.ok) {
const body = await res.text()
throw new Error(body || res.statusText)
}
hubSyncMessage = await res.text()
hubSyncStatus = 'success'
hubSyncMessage = 'Resource types synced from hub successfully'
} catch (e: any) {
hubSyncMessage =
e?.body?.error?.message ||
e?.body?.message ||
(typeof e?.body === 'string' ? e.body : null) ||
e?.message ||
'Failed to sync from hub'
hubSyncMessage = e?.message ?? 'Failed to sync from hub'
hubSyncStatus = 'error'
}
}
@@ -101,37 +97,31 @@
{#if loaded}
{#if showHubSync}
<div
class="p-3 border rounded-md bg-surface-secondary mb-4 mt-4 flex items-center justify-between gap-4"
<SettingCard
label="Resource types"
description="AI providers require their resource types. Sync from the Hub if they are missing."
class="my-4"
>
<div>
<p class="text-xs font-medium text-secondary">Resource types</p>
<p class="text-2xs text-tertiary mt-0.5">
AI providers require their resource types. Sync from the Hub if they are missing.
</p>
</div>
<Button
variant="default"
unifiedSize="sm"
loading={hubSyncStatus === 'loading'}
onClick={syncFromHub}
>
Sync from hub
</Button>
</div>
{#if hubSyncStatus === 'success'}
<div class="mb-4">
<Alert type="success" title="Resource types synced">
{#snippet headerAction()}
<Button
variant="default"
unifiedSize="sm"
loading={hubSyncStatus === 'loading'}
onClick={syncFromHub}
>
Sync from hub
</Button>
{/snippet}
{#if hubSyncStatus === 'success'}
<Alert type="success" title="Resource types synced" class="mt-2">
{hubSyncMessage}
</Alert>
</div>
{:else if hubSyncStatus === 'error'}
<div class="mb-4">
<Alert type="error" title="Sync failed">
{:else if hubSyncStatus === 'error'}
<Alert type="error" title="Sync failed" class="mt-2">
{hubSyncMessage}
</Alert>
</div>
{/if}
{/if}
</SettingCard>
{/if}
<AISettings
frontend/src/routes/(root)/(logged)/user/(user)/instance_settings/+page.svelte
+29 -81
View File
@@ -18,7 +18,7 @@
import { ChevronRight, ArrowLeft } from 'lucide-svelte'
import { superadmin } from '$lib/stores'
import { onDestroy, tick } from 'svelte'
import { UserService, JobService } from '$lib/gen'
import { SettingService, UserService } from '$lib/gen'
import { sendUserToast } from '$lib/toast'
import TextInput from '$lib/components/text_input/TextInput.svelte'
import Toggle from '$lib/components/Toggle.svelte'
@@ -89,7 +89,6 @@
// --- Account step state ---
let newEmail = $state('')
let newPassword = $state('')
let enableHubSync = $state(true)
let accountSubmitting = $state(false)
let accountError = $state('')
let showOssAccountDialog = $state(false)
@@ -99,7 +98,7 @@
let rtSyncStatus: 'idle' | 'loading' | 'success' | 'error' = $state('idle')
let rtSyncMessage = $state('')
async function syncCachedResourceTypes() {
async function syncResourceTypes() {
rtSyncStatus = 'loading'
rtSyncMessage = ''
try {
@@ -116,41 +115,24 @@
}
}
// --- Daily resource type sync (saved on finish) ---
const SYNC_DAILY_SETTING = 'sync_hub_resource_types_daily'
let syncResourceTypesDaily = $state(true)
SettingService.getGlobal({ key: SYNC_DAILY_SETTING })
.then((value) => {
if (typeof value === 'boolean') syncResourceTypesDaily = value
})
.catch(() => {})
$effect(() => {
if (
rtSyncStatus === 'idle' &&
((mode === 'wizard' && !isSettingsStep(wizardStep)) || (mode === 'full' && fullStep === 1))
) {
syncCachedResourceTypes()
syncResourceTypes()
}
})
// --- Live hub sync ---
let hubSyncStatus: 'idle' | 'loading' | 'success' | 'error' = $state('idle')
let hubSyncMessage = $state('')
async function syncFromHub() {
hubSyncStatus = 'loading'
hubSyncMessage = ''
try {
await JobService.runWaitResultScriptByPath({
workspace: 'admins',
path: 'u/admin/hub_sync',
requestBody: {}
})
hubSyncStatus = 'success'
hubSyncMessage = 'Resource types synced from hub successfully'
} catch (e: any) {
hubSyncMessage =
e?.body?.error?.message ||
e?.body?.message ||
(typeof e?.body === 'string' ? e.body : null) ||
e?.message ||
'Failed to sync from hub'
hubSyncStatus = 'error'
}
}
const emailPattern = /^[^\s@]+@[^\s@]+\.[^\s@]{2,}$/
let emailValid = $derived(emailPattern.test(newEmail))
let passwordValid = $derived(newPassword.length >= 2)
@@ -281,6 +263,12 @@
accountError = ''
accountSubmitting = true
try {
// Before the account switch: on CE it fails and setup ends with the default account.
await SettingService.setGlobal({
key: SYNC_DAILY_SETTING,
requestBody: { value: syncResourceTypesDaily }
})
let oldEmail = $superadmin
if (!oldEmail) {
oldEmail = await UserService.getCurrentEmail()
@@ -305,34 +293,6 @@
const { OpenAPI } = await import('$lib/gen')
OpenAPI.TOKEN = token
if (enableHubSync) {
try {
// Use direct fetch with token as query param to avoid the old session cookie
// overriding the Authorization header
const resp = await fetch(
`/api/w/admins/schedules/create?token=${encodeURIComponent(token)}`,
{
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
path: 'g/all/hub_sync',
schedule: '0 0 0 * * *',
script_path: 'u/admin/hub_sync',
is_flow: false,
args: {},
enabled: true,
timezone: 'Etc/UTC'
})
}
)
if (!resp.ok) {
console.warn('Schedule creation failed:', await resp.text())
}
} catch (e: any) {
console.warn('Schedule creation failed:', e?.body ?? e)
}
}
try {
await UserService.globalUserDelete({ email: oldEmail })
} catch (e: any) {
@@ -403,51 +363,39 @@
<SettingCard
label="Resource Types"
description="Resource types bundled with the Docker image are synced automatically. You can also fetch the latest from the hub."
description="Resource types are synced from the Hub when you reach this step. Without access to a Hub, the ones bundled with the Docker image are used."
>
<div class="flex flex-col gap-3 mt-1">
{#if rtSyncStatus === 'loading'}
<Alert type="info" title="Syncing cached resource types..." />
<Alert type="info" title="Syncing resource types..." />
{:else if rtSyncStatus === 'success'}
<Alert type="success" title="Cached resource types synced">
<Alert type="success" title="Resource types synced">
{rtSyncMessage}
</Alert>
{:else if rtSyncStatus === 'error'}
<Alert type="error" title="Cached resource types sync failed">
<Alert type="error" title="Resource type sync failed">
{rtSyncMessage}
</Alert>
{/if}
<div class="flex items-center gap-2">
<div class="flex">
<Button
variant="accent"
variant="default"
unifiedSize="sm"
loading={hubSyncStatus === 'loading'}
onClick={syncFromHub}
loading={rtSyncStatus === 'loading'}
onClick={syncResourceTypes}
>
Sync latest from hub
</Button>
<p class="text-tertiary text-2xs">
Fetches the latest resource types directly from the Windmill Hub (requires internet
access).
</p>
</div>
{#if hubSyncStatus === 'success'}
<Alert type="success" title="Hub sync complete">
{hubSyncMessage}
</Alert>
{:else if hubSyncStatus === 'error'}
<Alert type="error" title="Hub sync failed">
{hubSyncMessage}
</Alert>
{/if}
<Toggle
bind:checked={enableHubSync}
bind:checked={syncResourceTypesDaily}
options={{ right: 'Sync resource types every day' }}
size="xs"
/>
<p class="text-tertiary text-2xs">
The daily schedule synchronizes resource types from the Hub every day at midnight UTC.
The server updates them from the Hub once a day, overwriting local edits to the types the
Hub also defines. You can change this later in the instance settings.
</p>
</div>
</SettingCard>