diff --git a/backend/src/main.rs b/backend/src/main.rs index 10d1434f98..f8b150383a 100644 --- a/backend/src/main.rs +++ b/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]) { diff --git a/backend/windmill-api/src/triggers/http/handler.rs b/backend/windmill-api/src/triggers/http/handler.rs index 37442c8111..e58aab44a2 100644 --- a/backend/windmill-api/src/triggers/http/handler.rs +++ b/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 diff --git a/backend/windmill-trigger-http/src/lib.rs b/backend/windmill-trigger-http/src/lib.rs index f89247dd16..b78276c28d 100644 --- a/backend/windmill-trigger-http/src/lib.rs +++ b/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 = 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>, 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"); } diff --git a/backend/windmill-trigger-http/tests/refresh_routers.rs b/backend/windmill-trigger-http/tests/refresh_routers.rs new file mode 100644 index 0000000000..f47da2ee0e --- /dev/null +++ b/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, 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) { + 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"); +}