fix: widen variable.value without breaking live servers' cached queries (#11737)

* fix: widen variable.value only once every process can recover from it

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

* fix: wait out the previous release's servers before widening variable.value

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

* fix: write a v2 secret too long for the narrow column in cbc until it is widened

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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 {

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();

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();
 }
 

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::<sqlx::postgres::PgDatabaseError>()
+            .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()
+}

backend/windmill-common/src/error.rs

@@ -153,6 +153,10 @@ impl From<anyhow::Error> for Error {
 impl From<sqlx::Error> 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()) }
     }
 }

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");

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<bool> {
+    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) {

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);