diff --git a/backend/src/db_connect.rs b/backend/src/db_connect.rs index d4dee19f82..ab12ae6f6e 100644 --- a/backend/src/db_connect.rs +++ b/backend/src/db_connect.rs @@ -153,6 +153,9 @@ pub async fn connect( pool_options = pool_options.idle_timeout(Duration::from_secs(60)); } pool_options + .before_acquire(|_, meta| { + Box::pin(async move { Ok(!windmill_common::db::connection_is_retired(meta.age)) }) + }) .after_connect(move |conn, _| { if worker_mode { Box::pin(async move { diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 23ab2eaed1..4fae6f34ab 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -4469,12 +4469,14 @@ pub async fn monitor_db( // Re-encrypt the legacy-format secret variables once new values are written in AES-256-GCM. // Every ~5 min until one pass has gone through every workspace; spawned so a large instance - // does not hold up the join below. + // does not hold up the join below. The column widening it starts with is tried every + // iteration: a v2 value too long for the narrow column fails to save until it is done. let legacy_secret_upgrade_f = async { if server_mode && iteration.is_some() - && iteration.as_ref().unwrap().should_run(30) - && windmill_common::variables::legacy_secret_upgrade_pending() + && (windmill_common::variables::variable_value_widen_pending() + || (iteration.as_ref().unwrap().should_run(30) + && windmill_common::variables::legacy_secret_upgrade_pending())) { if let Some(db) = conn.as_sql() { let db = db.clone(); diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 21b738891e..db99a07c92 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -113,6 +113,12 @@ lazy_static::lazy_static! { (20260909163047, include_str!( "../../migrations/20260909163047_workspace_delete_cascade_indexes.up.sql" ).replace("CREATE INDEX", "CREATE INDEX CONCURRENTLY").replace("DROP INDEX", "DROP INDEX CONCURRENTLY")), + // The widening fails every query the previous release's still-running + // servers prepared on the column; `variables::widen_variable_value` does it + // once they are gone. + (20261012213357, include_str!( + "../../migrations/20261012213357_widen_variable_value.up.sql" + ).replace("ALTER TABLE variable ALTER COLUMN value TYPE VARCHAR(15040);", "")), ].into_iter().collect(); } diff --git a/backend/windmill-common/src/db.rs b/backend/windmill-common/src/db.rs index 0048abf438..38b89ad3b6 100644 --- a/backend/windmill-common/src/db.rs +++ b/backend/windmill-common/src/db.rs @@ -290,3 +290,45 @@ impl<'b> DbExecutor<'b> for &'b mut PgConnection { &mut **self } } + +lazy_static::lazy_static! { + static ref PROCESS_START: std::time::Instant = std::time::Instant::now(); +} + +/// Milliseconds after [`PROCESS_START`] before which every pooled connection is retired; +/// 0 retires none. +static RETIRE_CONNECTIONS_OPENED_BEFORE_MS: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); + +/// Postgres refuses a statement prepared before its result columns changed type (even a +/// varchar length) with this error, on every execution, for as long as the connection lives. +/// The routine is matched as well as the message, which `lc_messages` may translate. +pub fn is_stale_cached_plan(e: &sqlx::Error) -> bool { + let sqlx::Error::Database(d) = e else { + return false; + }; + d.code().as_deref() == Some("0A000") + && (d + .try_downcast_ref::() + .is_some_and(|pg| pg.routine() == Some("RevalidateCachedQuery")) + || d.message() + .contains("cached plan must not change result type")) +} + +/// Retire every pooled connection opened up to now, so a schema change that poisoned their +/// prepared statements costs one failed query instead of failing until they expire. +pub fn retire_connections_opened_before_now() { + let now = PROCESS_START.elapsed().as_millis() as u64 + 1; + RETIRE_CONNECTIONS_OPENED_BEFORE_MS.fetch_max(now, std::sync::atomic::Ordering::Relaxed); +} + +/// For the pool's `before_acquire`: whether a connection of this age must be closed. +pub fn connection_is_retired(age: std::time::Duration) -> bool { + let cutoff = RETIRE_CONNECTIONS_OPENED_BEFORE_MS.load(std::sync::atomic::Ordering::Relaxed); + cutoff != 0 && (PROCESS_START.elapsed().saturating_sub(age).as_millis() as u64) < cutoff +} + +/// Time since this process first used its database pool. +pub fn process_uptime() -> std::time::Duration { + PROCESS_START.elapsed() +} diff --git a/backend/windmill-common/src/error.rs b/backend/windmill-common/src/error.rs index c06defaab5..a54e9eb399 100644 --- a/backend/windmill-common/src/error.rs +++ b/backend/windmill-common/src/error.rs @@ -153,6 +153,10 @@ impl From for Error { impl From for Error { #[track_caller] fn from(e: sqlx::Error) -> Self { + if crate::db::is_stale_cached_plan(&e) { + tracing::warn!("a schema change invalidated cached statements, reopening the database connections: {e}"); + crate::db::retire_connections_opened_before_now(); + } Self::SqlErr { error: e, location: prettify_location(std::panic::Location::caller()) } } } diff --git a/backend/windmill-common/src/min_version.rs b/backend/windmill-common/src/min_version.rs index 05b664b23d..70878e3713 100644 --- a/backend/windmill-common/src/min_version.rs +++ b/backend/windmill-common/src/min_version.rs @@ -27,6 +27,8 @@ pub const MIN_VERSION_SUPPORTS_JOB_TOKEN_SCOPES: VC = vc(1, 822, 0, "Restricted // stored secret variables. Reading it ships one release earlier (1.831), so a rollback to // that release still reads every secret; below it, secrets written after the switch stay // unreadable until rolled forward. Only worker versions are seen here, not servers. +// Also gates widening `variable.value`, which needs every process to retire the connections +// the widening poisons (`db::retire_connections_opened_before_now`): that also ships in 1.832. pub const MIN_VERSION_SUPPORTS_AEAD_SECRETS: VC = vc(1, 832, 0, "AES-GCM secret encryption"); pub const MIN_VERSION_SUPPORTS_NODE_DEBOUNCING: VC = vc(1, 658, 0, "Flow node debouncing"); pub const MIN_VERSION_SUPPORTS_TOKEN_HASH: VC = vc(1, 659, 0, "Token hash storage"); diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index 58766cf25b..f357298243 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -400,7 +400,66 @@ pub fn legacy_secret_upgrade_pending() -> bool { && !LEGACY_SECRET_UPGRADE_DONE.load(Ordering::Relaxed) } -/// Re-encrypt the secret variables still in the legacy format in v2, workspace by workspace. +/// AES-256-GCM ciphertext is 32 bytes (magic, nonce, tag) longer than legacy CBC before base64, +/// so the longest CBC value that fits 15000 chars is exactly 15040 chars once re-encrypted. +const VARIABLE_VALUE_WIDTH: i32 = 15040; +static VARIABLE_VALUE_WIDENED: AtomicBool = AtomicBool::new(false); + +/// Whether this process has seen `variable.value` wide enough for every v2 value. Only servers +/// look, so it stays false in a worker. +pub fn variable_value_widened() -> bool { + VARIABLE_VALUE_WIDENED.load(Ordering::Relaxed) +} + +/// Whether `variable.value` may still have to be widened to fit v2 values. +pub fn variable_value_widen_pending() -> bool { + CipherFormat::current() == CipherFormat::V2 && !VARIABLE_VALUE_WIDENED.load(Ordering::Relaxed) +} + +/// Servers report no version, so the worker version gate alone cannot see a server of the +/// previous release; its rollout is assumed done once this server has been up this long. +const VARIABLE_VALUE_WIDEN_MIN_UPTIME: std::time::Duration = + std::time::Duration::from_secs(30 * 60); + +/// Widen `variable.value` to [`VARIABLE_VALUE_WIDTH`], or `false` when it is too early to. +/// Not a migration: changing a column's type, even its length, fails every statement a live +/// process prepared on it until its connection closes. It waits for v2 writes, gated on a +/// worker version that retires such connections on the first failure, so that no process too +/// old to recover is still running. +async fn widen_variable_value(db: &DB) -> error::Result { + if VARIABLE_VALUE_WIDENED.load(Ordering::Relaxed) { + return Ok(true); + } + // varchar's typmod is its length + 4, -1 when unbounded. + let typmod: i32 = sqlx::query_scalar( + "SELECT atttypmod FROM pg_attribute WHERE attrelid = 'variable'::regclass AND attname = 'value'", + ) + .fetch_one(db) + .await?; + if typmod >= 0 && typmod - 4 < VARIABLE_VALUE_WIDTH { + if crate::db::process_uptime() < VARIABLE_VALUE_WIDEN_MIN_UPTIME { + return Ok(false); + } + tracing::info!("widening variable.value to VARCHAR({VARIABLE_VALUE_WIDTH})"); + let mut tx = db.begin().await?; + // Queued behind a long transaction, the ALTER would block every variable read meanwhile. + sqlx::query("SET LOCAL lock_timeout = '3s'") + .execute(&mut *tx) + .await?; + sqlx::query(&format!( + "ALTER TABLE variable ALTER COLUMN value TYPE VARCHAR({VARIABLE_VALUE_WIDTH})" + )) + .execute(&mut *tx) + .await?; + tx.commit().await?; + crate::db::retire_connections_opened_before_now(); + } + VARIABLE_VALUE_WIDENED.store(true, Ordering::Relaxed); + Ok(true) +} + +/// Widen `variable.value` for v2 values, then re-encrypt the secret variables still in the +/// legacy format in v2, workspace by workspace. /// Idempotent, and emits no git-sync event: the plaintexts do not change. pub async fn upgrade_legacy_secret_variables(db: &DB) { struct Running; @@ -409,14 +468,26 @@ pub async fn upgrade_legacy_secret_variables(db: &DB) { LEGACY_SECRET_UPGRADE_RUNNING.store(false, Ordering::Release); } } - if !legacy_secret_upgrade_pending() + if !(variable_value_widen_pending() || legacy_secret_upgrade_pending()) || LEGACY_SECRET_UPGRADE_RUNNING.swap(true, Ordering::AcqRel) { return; } // Cleared on drop, so a pass that panics does not stop every later one in this process. let _running = Running; - upgrade_legacy_secret_variables_pass(db).await; + match widen_variable_value(db).await { + Ok(true) => {} + Ok(false) => return, + Err(e) => { + tracing::warn!( + "secret encryption upgrade: widening variable.value failed, retrying later: {e:#}" + ); + return; + } + } + if legacy_secret_upgrade_pending() { + upgrade_legacy_secret_variables_pass(db).await; + } } async fn upgrade_legacy_secret_variables_pass(db: &DB) { diff --git a/backend/windmill-common/src/workspace_crypt.rs b/backend/windmill-common/src/workspace_crypt.rs index a561060e32..26f38e123b 100644 --- a/backend/windmill-common/src/workspace_crypt.rs +++ b/backend/windmill-common/src/workspace_crypt.rs @@ -30,6 +30,7 @@ pub const V2_MAGIC: [u8; 4] = *b"wmA2"; const NONCE_LEN: usize = 12; const TAG_LEN: usize = 16; const BLOCK: usize = 16; +const NARROW_VARIABLE_VALUE_WIDTH: usize = 15000; const AEAD_KEY_INFO: &[u8] = b"windmill workspace secret aes-256-gcm v1"; @@ -102,7 +103,23 @@ impl WorkspaceCrypt { } pub fn encrypt_str(&self, plaintext: &str) -> String { - self.encrypt_str_in(plaintext, CipherFormat::current()) + self.encrypt_str_fitting( + plaintext, + CipherFormat::current(), + crate::variables::variable_value_widened(), + ) + } + + /// A v2 value is up to 44 chars longer than CBC: until `variable.value` is known widened + /// for it, one past the narrow width is written in CBC, which fits. + fn encrypt_str_fitting(&self, plaintext: &str, format: CipherFormat, widened: bool) -> String { + if format == CipherFormat::V2 { + let v2 = self.encrypt_str_in(plaintext, CipherFormat::V2); + if widened || v2.len() <= NARROW_VARIABLE_VALUE_WIDTH { + return v2; + } + } + self.encrypt_str_in(plaintext, CipherFormat::LegacyCbc) } pub fn encrypt_str_in(&self, plaintext: &str, format: CipherFormat) -> String { @@ -242,6 +259,18 @@ mod tests { ); } + #[test] + fn v2_too_long_for_a_narrow_column_is_written_in_cbc() { + let wc = WorkspaceCrypt::new(KEY); + // The longest plaintext whose CBC value fits 15000 chars. + let long = "x".repeat(11231); + let narrow = wc.encrypt_str_fitting(&long, CipherFormat::V2, false); + assert!(!wc.is_v2(&narrow) && narrow.len() <= NARROW_VARIABLE_VALUE_WIDTH); + assert_eq!(wc.decrypt_str(&narrow).unwrap(), long); + assert!(wc.is_v2(&wc.encrypt_str_fitting(&long, CipherFormat::V2, true))); + assert!(wc.is_v2(&wc.encrypt_str_fitting("short", CipherFormat::V2, false))); + } + #[test] fn v2_passes_the_block_multiple_ciphertext_shape_check() { let wc = WorkspaceCrypt::new(KEY);