fix: force HTTP router rebuild on trigger-change notification (#10849)

* fix: force HTTP router rebuild on trigger-change notification

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

* fix: coalesce http trigger change events into one forced rebuild

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

* fix: retry the coalesced http router rebuild when it fails

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

* fix: mark http routers stale when a forced rebuild fails

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

* fix: keep the router invalidation across an in-flight rebuild

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

---------

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

backend/src/main.rs

@@ -1434,21 +1434,30 @@ Windmill Community Edition {GIT_VERSION}
                                     // Poll for new events from notify_event table
                                     match windmill_common::notify_events::poll_notify_events(&db, last_event_id).await {
                                         Ok(events) => {
+                                            let mut http_trigger_change_handled = false;
                                             for event in events {
                                                 if !*windmill_common::QUIET_LOGS {
                                                     tracing::info!("Processing notify event: channel={}, payload={}", event.channel, event.payload);
                                                 }
-                                                process_notify_event(
-                                                    &event.channel,
-                                                    &event.payload,
-                                                    &db,
-                                                    &conn,
-                                                    &tx,
-                                                    server_mode,
-                                                    worker_mode,
-                                                    #[cfg(feature = "parquet")]
-                                                    disable_s3_store,
-                                                ).await;
+                                                let is_http_trigger_change = event.channel == "notify_http_trigger_change";
+                                                // Every changed http_trigger row emits its own event and each one forces
+                                                // a full router rebuild, but the batch's first successful rebuild already
+                                                // read every row the batch committed. A failed rebuild leaves the flag
+                                                // clear so the next event in the batch retries it.
+                                                if !(is_http_trigger_change && http_trigger_change_handled) {
+                                                    let handled = process_notify_event(
+                                                        &event.channel,
+                                                        &event.payload,
+                                                        &db,
+                                                        &conn,
+                                                        &tx,
+                                                        server_mode,
+                                                        worker_mode,
+                                                        #[cfg(feature = "parquet")]
+                                                        disable_s3_store,
+                                                    ).await;
+                                                    http_trigger_change_handled |= is_http_trigger_change && handled;
+                                                }
                                                 last_event_id = last_event_id.max(event.id);
                                             }
                                         }
@@ -1670,6 +1679,9 @@ Windmill Community Edition {GIT_VERSION}
 
 /// Process a single notify event from the polling-based event system.
 /// This replaces the old PgListener notification handling.
+///
+/// Returns `false` when the event still needs handling. Only the HTTP router rebuild reports
+/// that, because the poll loop coalesces those events and must not swallow the retry.
 #[allow(unused_variables)]
 async fn process_notify_event(
     channel: &str,
@@ -1680,7 +1692,7 @@ async fn process_notify_event(
     server_mode: bool,
     worker_mode: bool,
     #[cfg(feature = "parquet")] disable_s3_store: bool,
-) {
+) -> bool {
     match channel {
         "notify_config_change" => {
             if payload == "server" && server_mode {
@@ -1825,17 +1837,14 @@ async fn process_notify_event(
         #[cfg(feature = "http_trigger")]
         "notify_http_trigger_change" => {
             tracing::info!("HTTP trigger change detected: {}", payload);
-            match windmill_api::triggers::http::refresh_routers(db).await {
-                Ok((true, _)) => {
+            match windmill_api::triggers::http::refresh_routers(db, true).await {
+                Ok(_) => {
                     tracing::info!("Refreshed HTTP routers (trigger change)");
                 }
-                Ok((false, _)) => {
-                    tracing::warn!(
-                        "Should have refreshed HTTP routers (trigger change) but did not"
-                    );
-                }
                 Err(err) => {
                     tracing::error!("Error refreshing HTTP routers (trigger change): {err:#}");
+                    windmill_api::triggers::http::invalidate_routers();
+                    return false;
                 }
             };
         }
@@ -2059,7 +2068,7 @@ async fn process_notify_event(
                         tracing::error!(error = %e, "Could not reload http route workspaced route setting");
                     }
                     #[cfg(feature = "http_trigger")]
-                    match windmill_api::triggers::http::refresh_routers(db).await {
+                    match windmill_api::triggers::http::refresh_routers(db, false).await {
                         Ok((true, _)) => {
                             tracing::info!(
                                 "Refreshed HTTP routers (http workspaced route setting change)"
@@ -2179,6 +2188,7 @@ async fn process_notify_event(
             tracing::warn!("Unknown notification channel: {}", channel);
         }
     }
+    true
 }
 
 fn display_config(envs: &[&str]) {

backend/windmill-api/src/triggers/http/handler.rs

@@ -123,7 +123,9 @@ async fn get_http_route_trigger(
 
     let routers_cache = if routers_cache.routers.is_empty() {
         tracing::warn!("HTTP routers are not loaded, loading from db");
-        let (_, routers_cache) = refresh_routers(db).await?;
+        // refresh_routers takes the write lock, so holding this read guard across it deadlocks.
+        drop(routers_cache);
+        let (_, routers_cache) = refresh_routers(db, false).await?;
         routers_cache
     } else {
         routers_cache

backend/windmill-trigger-http/src/lib.rs

@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
 
 use quick_cache::sync::Cache;
 use serde::{Deserialize, Serialize};
@@ -27,9 +28,12 @@ lazy_static::lazy_static! {
     pub static ref HTTP_ROUTERS_CACHE: RwLock<RoutersCache> = RwLock::new(RoutersCache {
         routers: HashMap::new(),
         version: 0,
+        invalidations: 0,
     });
 }
 
+static HTTP_ROUTERS_INVALIDATIONS: AtomicU64 = AtomicU64::new(0);
+
 #[derive(Debug, Deserialize, Clone)]
 pub struct TriggerRoute {
     pub path: String,
@@ -56,6 +60,10 @@ pub struct TriggerRoute {
 pub struct RoutersCache {
     pub routers: HashMap<HttpMethod, matchit::Router<TriggerRoute>>,
     pub version: i64,
+    /// `HTTP_ROUTERS_INVALIDATIONS` as of the moment these rows were read. A rebuild that
+    /// started before an invalidation publishes a count behind the current one, which is what
+    /// stops it from passing its own stale rows off as covering that invalidation.
+    invalidations: u64,
 }
 
 #[derive(Serialize, Deserialize, sqlx::Type, Debug, Clone, Copy, Hash, Eq, PartialEq)]
@@ -223,12 +231,24 @@ pub fn validate_authentication_method(
     }
 }
 
-pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, RoutersCache>)> {
+/// `force` rebuilds unconditionally. `nextval` on `http_trigger_version_seq` runs inside the
+/// writing transaction and sequences are non-transactional, so another session can cache the
+/// bumped version against still-uncommitted rows, after which every version-gated refresh is a
+/// no-op. Force when reacting to a bump that could have been observed before its own rows were.
+pub async fn refresh_routers(
+    db: &DB,
+    force: bool,
+) -> Result<(bool, RwLockReadGuard<'_, RoutersCache>)> {
+    let invalidations = HTTP_ROUTERS_INVALIDATIONS.load(Ordering::Relaxed);
     let version = sqlx::query_scalar!("SELECT last_value FROM http_trigger_version_seq",)
         .fetch_one(db)
         .await?;
     let routers_cache = HTTP_ROUTERS_CACHE.read().await;
-    if routers_cache.version == 0 || version > routers_cache.version {
+    if force
+        || routers_cache.version == 0
+        || version > routers_cache.version
+        || invalidations != routers_cache.invalidations
+    {
         drop(routers_cache);
         let mut routers = HashMap::new();
 
@@ -274,7 +294,8 @@ pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, Route
             .await?;
 
             let mut router = matchit::Router::new();
-            let http_route_workspaced = HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed);
+            let http_route_workspaced =
+                HTTP_ROUTE_WORKSPACED_ROUTE.load(std::sync::atomic::Ordering::Relaxed);
 
             for trigger in triggers {
                 let full_path =
@@ -306,7 +327,7 @@ pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, Route
         }
 
         let mut routers_cache = HTTP_ROUTERS_CACHE.write().await;
-        *routers_cache = RoutersCache { routers, version };
+        *routers_cache = RoutersCache { routers, version, invalidations };
 
         Ok((true, routers_cache.downgrade()))
     } else {
@@ -315,11 +336,19 @@ pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, Route
     }
 }
 
+/// Record that the cache no longer covers everything committed, so the next refresh rebuilds
+/// whatever the version says. The routes already loaded keep being served in the meantime. Use
+/// after a forced refresh fails: its change is inside the cached version, so nothing else would
+/// retry it.
+pub fn invalidate_routers() {
+    HTTP_ROUTERS_INVALIDATIONS.fetch_add(1, Ordering::Relaxed);
+}
+
 pub async fn refresh_routers_loop(
     db: &DB,
     mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
 ) -> () {
-    match refresh_routers(db).await {
+    match refresh_routers(db, false).await {
         Ok(_) => {
             tracing::info!("Loaded HTTP routers");
         }
@@ -335,7 +364,7 @@ pub async fn refresh_routers_loop(
                     break;
                 }
                 _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {
-                    match refresh_routers(&db).await {
+                    match refresh_routers(&db, false).await {
                         Ok((true, _)) => {
                             tracing::info!("Refreshed HTTP routers");
                         }

backend/windmill-trigger-http/tests/refresh_routers.rs

@@ -0,0 +1,62 @@
+use sqlx::{Pool, Postgres};
+use windmill_trigger_http::{invalidate_routers, refresh_routers, HttpMethod, RoutersCache};
+
+async fn insert_trigger(db: &Pool<Postgres>, path: &str, route_path: &str) {
+    sqlx::query(
+        "INSERT INTO http_trigger (
+            path, route_path, route_path_key, script_path, is_flow, workspace_id, edited_by,
+            permissioned_as, http_method, authentication_method, request_type, is_static_website,
+            workspaced_route, wrap_body, raw_string, mode
+        ) VALUES ($1, $2, $2, 'f/test/handler', false, 'test-workspace', 'test-user',
+            'u/test-user', 'get', 'none', 'async', false, false, false, false, 'enabled')",
+    )
+    .bind(path)
+    .bind(route_path)
+    .execute(db)
+    .await
+    .expect("insert http_trigger");
+}
+
+fn routes(cache: &RoutersCache, path: &str) -> bool {
+    cache.routers[&HttpMethod::Get].at(path).is_ok()
+}
+
+// A trigger row can commit without advancing http_trigger_version_seq past what the cache
+// already holds, because `nextval` runs ahead of the commit it belongs to. The version gate
+// cannot see such a row; only forcing, or an invalidation, recovers the route.
+#[sqlx::test(migrations = "../migrations")]
+async fn rebuilds_a_change_the_cached_version_does_not_cover(db: Pool<Postgres>) {
+    insert_trigger(&db, "f/test/first", "first").await;
+    let (rebuilt, cache) = refresh_routers(&db, false).await.unwrap();
+    assert!(rebuilt);
+    assert!(routes(&cache, "/first"));
+    drop(cache);
+
+    insert_trigger(&db, "f/test/second", "second").await;
+
+    let (rebuilt, cache) = refresh_routers(&db, false).await.unwrap();
+    assert!(!rebuilt, "an unchanged version must not rebuild");
+    assert!(!routes(&cache, "/second"));
+    drop(cache);
+
+    let (rebuilt, cache) = refresh_routers(&db, true).await.unwrap();
+    assert!(rebuilt, "force must rebuild whatever the version says");
+    assert!(routes(&cache, "/second"));
+    drop(cache);
+
+    // A forced refresh that failed leaves its change inside the cached version, so the periodic
+    // version-gated refresh has to rebuild on the invalidation alone.
+    insert_trigger(&db, "f/test/third", "third").await;
+    invalidate_routers();
+
+    let (rebuilt, cache) = refresh_routers(&db, false).await.unwrap();
+    assert!(
+        rebuilt,
+        "an invalidation must rebuild through the version gate"
+    );
+    assert!(routes(&cache, "/third"));
+    drop(cache);
+
+    let (rebuilt, _) = refresh_routers(&db, false).await.unwrap();
+    assert!(!rebuilt, "a served invalidation must not rebuild forever");
+}