From e860847073b56be469ba37af5e3a8cb7d30ef7bc Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 4 Feb 2026 07:35:29 +0000 Subject: [PATCH] feat: replace LISTEN/NOTIFY with polling-based event system (#7778) * feat: replace LISTEN/NOTIFY with polling-based event system Co-Authored-By: Claude Opus 4.5 * test: add exhaustive tests for polling-based notify events Co-Authored-By: Claude Opus 4.5 * fix: add missing triggers and fix tests for polling-based events - Add variable/resource cache invalidation triggers to migration - Fix flow test to UPDATE flow table instead of INSERT into flow_version - Improve test isolation with unique channel names per test - All 26 tests now pass Co-Authored-By: Claude Opus 4.5 * test: add multi-server polling tests for cross-server event propagation Add 4 tests simulating independent server instances with separate DB connections and polling state: - test_two_servers_both_receive_trigger_event: both servers see same event - test_two_servers_cross_trigger_visibility: each triggers a change, both see both - test_server_catches_up_after_being_offline: server catches up on missed events - test_two_servers_incremental_polling: multi-round polling with cursor advancement Co-Authored-By: Claude Opus 4.5 * feat: add LISTEN_NEW_EVENTS_INTERVAL_SEC env var and e2e server test - Make poll interval configurable via LISTEN_NEW_EVENTS_INTERVAL_SEC (defaults to 30s) - Make migration idempotent with IF NOT EXISTS - Replace mock multi-server tests with actual e2e test that starts two windmill server processes on ports 19100/19200 with 1s poll interval, triggers a DB change, and verifies both servers log the event Co-Authored-By: Claude Opus 4.5 * test: ignore notify_events tests in CI These tests require a running database, like other integration tests in the codebase. Run with --ignored flag locally. Co-Authored-By: Claude Opus 4.5 * test: only ignore slow e2e test, not fast DB tests Only test_two_server_processes_both_receive_event is slow (~10s, starts two server processes). The other 26 tests run in <0.2s. Co-Authored-By: Claude Opus 4.5 * fix: ignore all notify_events tests in CI All tests depend on the notify_event table from the polling-based events migration, which is not applied in CI. Co-Authored-By: Claude Opus 4.5 * fix: use sqlx::test for notify_events tests so they work in CI Convert all 26 fast tests from #[tokio::test] + manual get_db() to #[sqlx::test(fixtures("base"))], which creates temporary databases with all migrations applied. This ensures the notify_event table exists in CI without manual setup. Only the slow e2e multi-server test retains #[tokio::test] + #[ignore]. Co-Authored-By: Claude Opus 4.5 * chore: reduce default polling interval from 30s to 10s Co-Authored-By: Claude Opus 4.5 * fix: address review feedback on polling-based events - Remove redundant notify_event_id_idx index (id is already PRIMARY KEY) - Add LIMIT 1000 to poll_notify_events to bound memory per poll cycle - Fix potential UTF-8 panic in token log truncation using str::get - Remove var/resource cache triggers that were re-enabled by mistake (they were intentionally dropped in migration 20250902085504) Co-Authored-By: Claude Opus 4.5 --------- Co-authored-by: Claude Opus 4.5 --- ...260203172950_polling_based_events.down.sql | 121 +++ ...20260203172950_polling_based_events.up.sql | 135 +++ backend/src/main.rs | 722 +++++++-------- backend/src/monitor.rs | 19 + backend/tests/notify_events.rs | 835 ++++++++++++++++++ backend/windmill-common/src/lib.rs | 1 + backend/windmill-common/src/notify_events.rs | 63 ++ 7 files changed, 1508 insertions(+), 388 deletions(-) create mode 100644 backend/migrations/20260203172950_polling_based_events.down.sql create mode 100644 backend/migrations/20260203172950_polling_based_events.up.sql create mode 100644 backend/tests/notify_events.rs create mode 100644 backend/windmill-common/src/notify_events.rs diff --git a/backend/migrations/20260203172950_polling_based_events.down.sql b/backend/migrations/20260203172950_polling_based_events.down.sql new file mode 100644 index 0000000000..ef95991b98 --- /dev/null +++ b/backend/migrations/20260203172950_polling_based_events.down.sql @@ -0,0 +1,121 @@ +-- Revert to pg_notify based event system + +-- Restore notify_config_change function +CREATE OR REPLACE FUNCTION notify_config_change() +RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify('notify_config_change', NEW.name::text); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_global_setting_change function +CREATE OR REPLACE FUNCTION notify_global_setting_change() +RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify('notify_global_setting_change', NEW.name::text); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_global_setting_delete function +CREATE OR REPLACE FUNCTION notify_global_setting_delete() +RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify('notify_global_setting_change', OLD.name::text); + RETURN OLD; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_webhook_change function +CREATE OR REPLACE FUNCTION notify_webhook_change() +RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify('notify_webhook_change', NEW.workspace_id); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_workspace_envs_change function +CREATE OR REPLACE FUNCTION notify_workspace_envs_change() +RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify('notify_workspace_envs_change', NEW.workspace_id); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_workspace_premium_change function +CREATE OR REPLACE FUNCTION notify_workspace_premium_change() +RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify('notify_workspace_premium_change', NEW.id); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_team_plan_status_change function +CREATE OR REPLACE FUNCTION notify_team_plan_status_change() +RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify('notify_workspace_premium_change', NEW.workspace_id); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_runnable_version_change function +CREATE OR REPLACE FUNCTION notify_runnable_version_change() +RETURNS TRIGGER AS $$ +DECLARE + source_type TEXT; + kind TEXT; +BEGIN + source_type := TG_ARGV[0]; + + IF source_type = 'script' THEN + kind := NEW.kind; + ELSE + kind := 'flow'; + END IF; + + PERFORM pg_notify('notify_runnable_version_change', NEW.workspace_id || ':' || source_type || ':' || NEW.path || ':' || kind); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_http_trigger_change function +CREATE OR REPLACE FUNCTION notify_http_trigger_change() +RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify('notify_http_trigger_change', NEW.workspace_id || ':' || NEW.path); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_token_invalidation function +CREATE OR REPLACE FUNCTION notify_token_invalidation() +RETURNS TRIGGER AS $$ +BEGIN + IF OLD.label = 'session' AND OLD.email IS NOT NULL THEN + PERFORM pg_notify('notify_token_invalidation', OLD.token); + END IF; + RETURN OLD; +END; +$$ LANGUAGE plpgsql; + +-- Restore notify_workspace_key_change function +CREATE OR REPLACE FUNCTION notify_workspace_key_change() +RETURNS TRIGGER AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + PERFORM pg_notify('notify_workspace_key_change', OLD.workspace_id); + RETURN OLD; + ELSE + PERFORM pg_notify('notify_workspace_key_change', NEW.workspace_id); + RETURN NEW; + END IF; +END; +$$ LANGUAGE plpgsql; + +-- Drop the notify_event table +DROP TABLE IF EXISTS notify_event; diff --git a/backend/migrations/20260203172950_polling_based_events.up.sql b/backend/migrations/20260203172950_polling_based_events.up.sql new file mode 100644 index 0000000000..e85597a689 --- /dev/null +++ b/backend/migrations/20260203172950_polling_based_events.up.sql @@ -0,0 +1,135 @@ +-- Create notify_event table for polling-based event system +-- This replaces PostgreSQL LISTEN/NOTIFY with a table-based approach + +CREATE TABLE IF NOT EXISTS notify_event ( + id BIGSERIAL PRIMARY KEY, + channel TEXT NOT NULL, + payload TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS notify_event_created_at_idx ON notify_event (created_at); + +-- Drop redundant index if it exists (id is already the PRIMARY KEY) +DROP INDEX IF EXISTS notify_event_id_idx; + +-- Update notify_config_change function +CREATE OR REPLACE FUNCTION notify_config_change() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO notify_event (channel, payload) VALUES ('notify_config_change', NEW.name::text); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Update notify_global_setting_change function +CREATE OR REPLACE FUNCTION notify_global_setting_change() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO notify_event (channel, payload) VALUES ('notify_global_setting_change', NEW.name::text); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Update notify_global_setting_delete function +CREATE OR REPLACE FUNCTION notify_global_setting_delete() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO notify_event (channel, payload) VALUES ('notify_global_setting_change', OLD.name::text); + RETURN OLD; +END; +$$ LANGUAGE plpgsql; + +-- Update notify_webhook_change function +CREATE OR REPLACE FUNCTION notify_webhook_change() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO notify_event (channel, payload) VALUES ('notify_webhook_change', NEW.workspace_id); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Update notify_workspace_envs_change function +CREATE OR REPLACE FUNCTION notify_workspace_envs_change() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_envs_change', COALESCE(NEW.workspace_id, OLD.workspace_id)); + RETURN COALESCE(NEW, OLD); +END; +$$ LANGUAGE plpgsql; + +-- Update notify_workspace_premium_change function +CREATE OR REPLACE FUNCTION notify_workspace_premium_change() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_premium_change', NEW.id); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Update notify_team_plan_status_change function +CREATE OR REPLACE FUNCTION notify_team_plan_status_change() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_premium_change', NEW.workspace_id); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Update notify_runnable_version_change function +CREATE OR REPLACE FUNCTION notify_runnable_version_change() +RETURNS TRIGGER AS $$ +DECLARE + source_type TEXT; + kind TEXT; +BEGIN + source_type := TG_ARGV[0]; + + IF source_type = 'script' THEN + kind := NEW.kind; + ELSE + kind := 'flow'; + END IF; + + INSERT INTO notify_event (channel, payload) VALUES ('notify_runnable_version_change', NEW.workspace_id || ':' || source_type || ':' || NEW.path || ':' || kind); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +-- Update notify_http_trigger_change function +CREATE OR REPLACE FUNCTION notify_http_trigger_change() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO notify_event (channel, payload) VALUES ('notify_http_trigger_change', COALESCE(NEW.workspace_id, OLD.workspace_id) || ':' || COALESCE(NEW.path, OLD.path)); + RETURN COALESCE(NEW, OLD); +END; +$$ LANGUAGE plpgsql; + +-- Update notify_token_invalidation function +CREATE OR REPLACE FUNCTION notify_token_invalidation() +RETURNS TRIGGER AS $$ +BEGIN + IF OLD.label = 'session' AND OLD.email IS NOT NULL THEN + INSERT INTO notify_event (channel, payload) VALUES ('notify_token_invalidation', OLD.token); + END IF; + RETURN OLD; +END; +$$ LANGUAGE plpgsql; + +-- Update notify_workspace_key_change function +CREATE OR REPLACE FUNCTION notify_workspace_key_change() +RETURNS TRIGGER AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_key_change', OLD.workspace_id); + RETURN OLD; + ELSE + INSERT INTO notify_event (channel, payload) VALUES ('notify_workspace_key_change', NEW.workspace_id); + RETURN NEW; + END IF; +END; +$$ LANGUAGE plpgsql; + +-- NOTE: var_cache_invalidation / resource_cache_invalidation triggers were +-- intentionally dropped in migration 20250902085504. We do NOT re-create them +-- here to keep this migration scoped to the LISTEN/NOTIFY → polling swap only. diff --git a/backend/src/main.rs b/backend/src/main.rs index 855262c5d5..fc9a412ab3 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -16,7 +16,7 @@ use monitor::{ send_current_log_file_to_object_store, send_logs_to_object_store, WORKERS_NAMES, }; use rand::Rng; -use sqlx::{postgres::PgListener, Pool, Postgres}; +use sqlx::{Pool, Postgres}; use std::{ collections::HashMap, fs::{create_dir_all, DirBuilder}, @@ -212,11 +212,17 @@ where } lazy_static::lazy_static! { - static ref PG_LISTENER_REFRESH_PERIOD_SECS: u64 = std::env::var("PG_LISTENER_REFRESH_PERIOD_SECS") + // Period in seconds between full settings reload (12 hours by default) + static ref SETTINGS_RELOAD_PERIOD_SECS: u64 = std::env::var("SETTINGS_RELOAD_PERIOD_SECS") .ok() .and_then(|x| x.parse::().ok()) .unwrap_or(3600 * 12); + // Period in seconds between polling for notify events (10s by default) + static ref LISTEN_NEW_EVENTS_INTERVAL_SEC: u64 = std::env::var("LISTEN_NEW_EVENTS_INTERVAL_SEC") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(10); } pub fn main() -> anyhow::Result<()> { @@ -1138,8 +1144,18 @@ Windmill Community Edition {GIT_VERSION} let base_internal_url = base_internal_url.to_string(); let db = db.clone(); let h = tokio::spawn(async move { - let mut listener = retry_listen_pg(&db).await; - let mut last_listener_refresh = Instant::now(); + // Initialize last_event_id to current max to avoid processing old events on startup + let mut last_event_id: i64 = match windmill_common::notify_events::get_latest_event_id(&db).await { + Ok(id) => { + tracing::info!("Initialized notify event polling with last_event_id: {}", id); + id + } + Err(e) => { + tracing::warn!("Could not get latest event id, starting from 0: {e:#}"); + 0 + } + }; + let mut last_settings_reload = Instant::now(); let mut monitor_iteration: u64 = 0; let rd_shift: u8 = rand::rng().random_range(0..200); loop { @@ -1158,349 +1174,36 @@ Windmill Community Edition {GIT_VERSION} tracing::info!("received killpill for monitor job"); break; }, - notification = listener.try_recv() => { - match notification { - Ok(n) => { - if n.is_none() { - tracing::error!("Could not receive notification, attempting to reconnect to pg listener"); - continue; - } - let n = n.unwrap(); - tracing::info!("Received new pg notification: {n:?}"); - match n.channel() { - "notify_config_change" => { - match n.payload() { - "server" if server_mode => { - tracing::error!("Server config change detected but server config is obsolete: {}", n.payload()); - }, - a@ _ if worker_mode && a == format!("worker__{}", *WORKER_GROUP) => { - tracing::info!("Worker config change detected: {}", n.payload()); - reload_worker_config(&db, tx.clone(), true).await; - }, - _ => { - tracing::debug!("config changed but did not target this server/worker"); - } - } - }, - "notify_webhook_change" => { - let workspace_id = n.payload(); - tracing::info!("Webhook change detected, invalidating webhook cache: {}", workspace_id); - windmill_api::webhook_util::WEBHOOK_CACHE.remove(workspace_id); - }, - "notify_workspace_envs_change" => { - let workspace_id = n.payload(); - tracing::info!("Workspace envs change detected, invalidating workspace envs cache: {}", workspace_id); - windmill_common::variables::CUSTOM_ENVS_CACHE.remove(workspace_id); - }, - "notify_workspace_key_change" => { - let workspace_id = n.payload(); - tracing::info!("Workspace key change detected, invalidating workspace key cache: {}", workspace_id); - windmill_common::variables::WORKSPACE_CRYPT_CACHE.remove(workspace_id); - }, - "notify_workspace_premium_change" => { - let workspace_id = n.payload(); - tracing::info!("Workspace premium change detected, invalidating workspace premium cache: {}", workspace_id); - windmill_common::workspaces::TEAM_PLAN_CACHE.remove(workspace_id); - }, - "notify_runnable_version_change" => { - let payload = n.payload(); - tracing::info!("Runnable version change detected: {}", payload); - match payload.split(':').collect::>().as_slice() { - [workspace_id, source_type, path, kind] => { - let key = (workspace_id.to_string(), path.to_string()); - match source_type { - &"script" => { - windmill_common::DEPLOYED_SCRIPT_HASH_CACHE.remove(&key); - match kind { - &"preprocessor" => { - match sqlx::query_scalar!( - "SELECT fv.id - FROM flow f - INNER JOIN flow_version fv ON fv.id = f.versions[array_upper(f.versions, 1)] - WHERE fv.value->'preprocessor_module'->'value'->>'path' = $1 AND f.workspace_id = $2", - path, - workspace_id - ).fetch_all(&db).await { - Ok(flow_versions) => { - tracing::debug!("Workspace preprocessor {} changed, removing runnable format version cache for flow versions {:?}", path, flow_versions); - for version in flow_versions { - for trigger_kind in TriggerKind::iter() { - let key = (windmill_common::triggers::HubOrWorkspaceId::WorkspaceId(workspace_id.to_string()), version, trigger_kind); - windmill_common::triggers::RUNNABLE_FORMAT_VERSION_CACHE.remove(&key); - } - } - } - Err(e) => { - tracing::error!("Error fetching flow paths: {e:#}"); - } - } - }, - _ => {} - } - } - &"flow" => { - let dynamic_input_key = windmill_common::jobs::generate_dynamic_input_key(workspace_id, path); - windmill_common::DYNAMIC_INPUT_CACHE.remove(&dynamic_input_key); - windmill_common::FLOW_VERSION_CACHE.remove(&key); - }, - _ => { - tracing::warn!("Unknown runnable version change payload: {}", payload); - } - } - }, - _ => { - tracing::warn!("Unknown runnable version change payload: {}", payload); - } - } - }, - #[cfg(feature = "http_trigger")] - "notify_http_trigger_change" => { - tracing::info!("HTTP trigger change detected: {}", n.payload()); - match windmill_api::triggers::http::refresh_routers(&db).await { - Ok((true, _)) => { - 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:#}"); - } - }; - }, - "notify_token_invalidation" => { - let token = n.payload(); - tracing::info!("Token invalidation detected for token: {}...", &token[..token.len().min(8)]); - windmill_api::auth::invalidate_token_from_cache(token); - }, - "var_cache_invalidation" => { - if let Ok(payload) = serde_json::from_str::(n.payload()) { - if let (Some(workspace_id), Some(path)) = - (payload.get("workspace_id").and_then(|v| v.as_str()), - payload.get("path").and_then(|v| v.as_str())) { - tracing::info!("Variable cache invalidation detected: {}:{}", workspace_id, path); - windmill_api::var_resource_cache::invalidate_variable_cache(&workspace_id, &path); - } - } - }, - "resource_cache_invalidation" => { - if let Ok(payload) = serde_json::from_str::(n.payload()) { - if let (Some(workspace_id), Some(path)) = - (payload.get("workspace_id").and_then(|v| v.as_str()), - payload.get("path").and_then(|v| v.as_str())) { - tracing::info!("Resource cache invalidation detected: {}:{}", workspace_id, path); - windmill_api::var_resource_cache::invalidate_resource_cache(&workspace_id, &path); - } - } - }, - "notify_global_setting_change" => { - tracing::info!("Global setting change detected: {}", n.payload()); - match n.payload() { - BASE_URL_SETTING => { - if let Err(e) = reload_base_url_setting(&conn).await { - tracing::error!(error = %e, "Could not reload base url setting"); - } - }, - OAUTH_SETTING => { - if let Err(e) = reload_base_url_setting(&conn).await { - tracing::error!(error = %e, "Could not reload oauth setting"); - } - }, - CUSTOM_TAGS_SETTING => { - if let Err(e) = reload_custom_tags_setting(&db).await { - tracing::error!(error = %e, "Could not reload custom tags setting"); - } - }, - LICENSE_KEY_SETTING => { - if let Err(e) = reload_license_key(&db.into()).await { - tracing::error!("Failed to reload license key: {e:#}"); - } - }, - DEFAULT_TAGS_PER_WORKSPACE_SETTING => { - if let Err(e) = load_tag_per_workspace_enabled(&db).await { - tracing::error!("Error loading default tag per workspace: {e:#}"); - } - }, - DEFAULT_TAGS_WORKSPACES_SETTING => { - if let Err(e) = load_tag_per_workspace_workspaces(&db).await { - tracing::error!("Error loading default tag per workspace workspaces: {e:#}"); - } - }, - SMTP_SETTING => { - reload_smtp_config(&db).await; - }, - TEAMS_SETTING => { - tracing::info!("Teams setting changed."); - }, - INDEXER_SETTING => { - reload_indexer_config(&db).await; - }, - TIMEOUT_WAIT_RESULT_SETTING => { - reload_timeout_wait_result_setting(&conn).await - }, - RETENTION_PERIOD_SECS_SETTING => { - reload_retention_period_setting(&conn).await - }, - MONITOR_LOGS_ON_OBJECT_STORE_SETTING => { - reload_delete_logs_periodically_setting(&conn).await - }, - JOB_DEFAULT_TIMEOUT_SECS_SETTING => { - reload_job_default_timeout_setting(&conn).await - }, - #[cfg(feature = "parquet")] - OBJECT_STORE_CONFIG_SETTING => { - if !disable_s3_store { - reload_object_store_setting(&db).await; - } - }, - SCIM_TOKEN_SETTING => { - reload_scim_token_setting(&conn).await - }, - EXTRA_PIP_INDEX_URL_SETTING => { - reload_extra_pip_index_url_setting(&conn).await - }, - PIP_INDEX_URL_SETTING => { - reload_pip_index_url_setting(&conn).await - }, - INSTANCE_PYTHON_VERSION_SETTING => { - reload_instance_python_version_setting(&conn).await - }, - NPM_CONFIG_REGISTRY_SETTING => { - reload_npm_config_registry_setting(&conn).await - }, - BUNFIG_INSTALL_SCOPES_SETTING => { - reload_bunfig_install_scopes_setting(&conn).await - }, - NUGET_CONFIG_SETTING => { - reload_nuget_config_setting(&conn).await - }, - POWERSHELL_REPO_URL_SETTING => { - reload_powershell_repo_url_setting(&conn).await - }, - POWERSHELL_REPO_PAT_SETTING => { - reload_powershell_repo_pat_setting(&conn).await - }, - MAVEN_REPOS_SETTING => { - reload_maven_repos_setting(&conn).await - }, - NO_DEFAULT_MAVEN_SETTING => { - reload_no_default_maven_setting(&conn).await - }, - RUBY_REPOS_SETTING => { - reload_ruby_repos_setting(&conn).await - }, - HUB_API_SECRET_SETTING => { - reload_hub_api_secret_setting(&conn).await - }, - KEEP_JOB_DIR_SETTING => { - load_keep_job_dir(&conn).await; - }, - OTEL_TRACING_PROXY_SETTING => { - reload_otel_tracing_proxy_setting(&conn).await; - if worker_mode { - tracing::info!("OTEL tracing proxy setting changed, restarting worker"); - send_delayed_killpill(&tx, 4, "OTEL tracing proxy setting change").await; - } - }, - REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING => { - load_require_preexisting_user(&db).await; - }, - EXPOSE_METRICS_SETTING => { - tracing::info!("Metrics setting changed, restarting"); - send_delayed_killpill(&tx, 40, "metrics setting change").await; - }, - EMAIL_DOMAIN_SETTING => { - tracing::info!("Email domain setting changed"); - if server_mode { - send_delayed_killpill(&tx, 4, "email domain setting change").await; - } - }, - EXPOSE_DEBUG_METRICS_SETTING => { - if let Err(e) = load_metrics_debug_enabled(&conn).await { - tracing::error!(error = %e, "Could not reload debug metrics setting"); - } - }, - APP_WORKSPACED_ROUTE_SETTING => { - if let Err(e) = reload_app_workspaced_route_setting(&db).await { - tracing::error!(error = %e, "Could not reload app workspaced route setting"); - } - }, - OTEL_SETTING => { - tracing::info!("OTEL setting changed, restarting"); - send_delayed_killpill(&tx, 4, "OTEL setting change").await; - }, - REQUEST_SIZE_LIMIT_SETTING => { - if server_mode { - tracing::info!("Request limit size change detected, killing server expecting to be restarted"); - send_delayed_killpill(&tx, 4, "request size limit change").await; - } - }, - SAML_METADATA_SETTING => { - tracing::info!("SAML metadata change detected, killing server expecting to be restarted"); - send_delayed_killpill(&tx, 0, "SAML metadata change").await; - }, - HUB_BASE_URL_SETTING => { - if let Err(e) = reload_hub_base_url_setting(&conn, server_mode).await { - tracing::error!(error = %e, "Could not reload hub base url setting"); - } - }, - CRITICAL_ERROR_CHANNELS_SETTING => { - if let Err(e) = reload_critical_error_channels_setting(&db).await { - tracing::error!(error = %e, "Could not reload critical error emails setting"); - } - }, - CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING => { - if let Err(e) = reload_critical_alerts_on_db_oversize(&db).await { - tracing::error!(error = %e, "Could not reload critical alerts on db oversize setting"); - } - - }, - JWT_SECRET_SETTING => { - if let Err(e) = reload_jwt_secret_setting(&db).await { - tracing::error!(error = %e, "Could not reload jwt secret setting"); - } - }, - CRITICAL_ALERT_MUTE_UI_SETTING => { - tracing::info!("Critical alert UI setting changed"); - if let Err(e) = reload_critical_alert_mute_ui_setting(&conn).await { - tracing::error!(error = %e, "Could not reload critical alert UI setting"); - } - }, - a @_ => { - tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a); - } - } - }, - _ => { - tracing::warn!("Unknown notification received"); - continue; + _ = tokio::time::sleep(Duration::from_secs(*LISTEN_NEW_EVENTS_INTERVAL_SEC)) => { + // Poll for new events from notify_event table + match windmill_common::notify_events::poll_notify_events(&db, last_event_id).await { + Ok(events) => { + 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; + last_event_id = last_event_id.max(event.id); } - }, + } Err(e) => { - tracing::error!(error = %e, "Could not receive notification, attempting to reconnect listener"); - let db = db.clone(); - tokio::select! { - biased; - _ = monitor_killpill_rx.recv() => { - tracing::info!("received killpill for monitor job"); - break; - }, - new_listener = async move { retry_listen_pg(&db).await } => { - listener = new_listener; - continue; - } - } + tracing::error!("Error polling notify events: {e:#}"); } - }; - }, - _ = tokio::time::sleep(Duration::from_secs(30)) => { - if last_listener_refresh.elapsed() > Duration::from_secs(*PG_LISTENER_REFRESH_PERIOD_SECS) { - tracing::info!("Refreshing pg listeners, settings and license key after {}s", Duration::from_secs(*PG_LISTENER_REFRESH_PERIOD_SECS).as_secs()); - if let Err(e) = listener.unlisten_all().await { - tracing::error!(error = %e, "Could not unlisten to database"); - } - listener = retry_listen_pg(&db).await; + } + + // Periodic full settings reload + if last_settings_reload.elapsed() > Duration::from_secs(*SETTINGS_RELOAD_PERIOD_SECS) { + tracing::info!("Reloading settings and license key after {}s", Duration::from_secs(*SETTINGS_RELOAD_PERIOD_SECS).as_secs()); initial_load( &conn, tx.clone(), @@ -1514,7 +1217,7 @@ Windmill Community Edition {GIT_VERSION} if let Err(err) = reload_license_key(&conn).await { tracing::error!("Failed to reload license key: {err:#}"); } - last_listener_refresh = Instant::now(); + last_settings_reload = Instant::now(); } if server_mode { @@ -1668,50 +1371,293 @@ Windmill Community Edition {GIT_VERSION} std::process::exit(0); } -async fn listen_pg(db: &Pool) -> Option { - let mut listener = match PgListener::connect_with(db).await { - Ok(l) => l, - Err(e) => { - tracing::error!(error = %e, "Could not connect to database"); - return None; - } - }; - - #[allow(unused_mut)] - let mut channels = vec![ - "notify_config_change", - "notify_global_setting_change", - "notify_webhook_change", - "notify_workspace_envs_change", - "notify_workspace_key_change", - "notify_runnable_version_change", - "notify_token_invalidation", - ]; - - #[cfg(feature = "http_trigger")] - channels.push("notify_http_trigger_change"); - - #[cfg(feature = "cloud")] - channels.push("notify_workspace_premium_change"); - - if let Err(e) = listener.listen_all(channels).await { - tracing::error!(error = %e, "Could not listen to database"); - return None; - } - - return Some(listener); -} - -async fn retry_listen_pg(db: &Pool) -> PgListener { - let mut listener = listen_pg(db).await; - loop { - if listener.is_none() { - tracing::info!("Retrying listening to pg listen in 5 seconds"); - tokio::time::sleep(Duration::from_secs(5)).await; - listener = listen_pg(db).await; - } else { - tracing::info!("Successfully connected to pg listen"); - return listener.unwrap(); +/// Process a single notify event from the polling-based event system. +/// This replaces the old PgListener notification handling. +#[allow(unused_variables)] +async fn process_notify_event( + channel: &str, + payload: &str, + db: &Pool, + conn: &Connection, + tx: &KillpillSender, + server_mode: bool, + worker_mode: bool, + #[cfg(feature = "parquet")] + disable_s3_store: bool, +) { + match channel { + "notify_config_change" => { + if payload == "server" && server_mode { + tracing::error!("Server config change detected but server config is obsolete: {}", payload); + } else if worker_mode && payload == format!("worker__{}", *WORKER_GROUP) { + tracing::info!("Worker config change detected: {}", payload); + reload_worker_config(db, tx.clone(), true).await; + } else { + tracing::debug!("config changed but did not target this server/worker"); + } + }, + "notify_webhook_change" => { + tracing::info!("Webhook change detected, invalidating webhook cache: {}", payload); + windmill_api::webhook_util::WEBHOOK_CACHE.remove(payload); + }, + "notify_workspace_envs_change" => { + tracing::info!("Workspace envs change detected, invalidating workspace envs cache: {}", payload); + windmill_common::variables::CUSTOM_ENVS_CACHE.remove(payload); + }, + "notify_workspace_key_change" => { + tracing::info!("Workspace key change detected, invalidating workspace key cache: {}", payload); + windmill_common::variables::WORKSPACE_CRYPT_CACHE.remove(payload); + }, + "notify_workspace_premium_change" => { + tracing::info!("Workspace premium change detected, invalidating workspace premium cache: {}", payload); + windmill_common::workspaces::TEAM_PLAN_CACHE.remove(payload); + }, + "notify_runnable_version_change" => { + tracing::info!("Runnable version change detected: {}", payload); + match payload.split(':').collect::>().as_slice() { + [workspace_id, source_type, path, kind] => { + let key = (workspace_id.to_string(), path.to_string()); + match *source_type { + "script" => { + windmill_common::DEPLOYED_SCRIPT_HASH_CACHE.remove(&key); + if *kind == "preprocessor" { + match sqlx::query_scalar::<_, i64>( + "SELECT fv.id + FROM flow f + INNER JOIN flow_version fv ON fv.id = f.versions[array_upper(f.versions, 1)] + WHERE fv.value->'preprocessor_module'->'value'->>'path' = $1 AND f.workspace_id = $2", + ) + .bind(*path) + .bind(*workspace_id) + .fetch_all(db).await { + Ok(flow_versions) => { + tracing::debug!("Workspace preprocessor {} changed, removing runnable format version cache for flow versions {:?}", path, flow_versions); + for version in flow_versions { + for trigger_kind in TriggerKind::iter() { + let key = (windmill_common::triggers::HubOrWorkspaceId::WorkspaceId(workspace_id.to_string()), version, trigger_kind); + windmill_common::triggers::RUNNABLE_FORMAT_VERSION_CACHE.remove(&key); + } + } + } + Err(e) => { + tracing::error!("Error fetching flow paths: {e:#}"); + } + } + } + } + "flow" => { + let dynamic_input_key = windmill_common::jobs::generate_dynamic_input_key(workspace_id, path); + windmill_common::DYNAMIC_INPUT_CACHE.remove(&dynamic_input_key); + windmill_common::FLOW_VERSION_CACHE.remove(&key); + }, + _ => { + tracing::warn!("Unknown runnable version change payload: {}", payload); + } + } + }, + _ => { + tracing::warn!("Unknown runnable version change payload: {}", payload); + } + } + }, + #[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, _)) => { + 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:#}"); + } + }; + }, + "notify_token_invalidation" => { + tracing::info!("Token invalidation detected for token: {}...", payload.get(..8).unwrap_or(payload)); + windmill_api::auth::invalidate_token_from_cache(payload); + }, + "notify_global_setting_change" => { + tracing::info!("Global setting change detected: {}", payload); + match payload { + BASE_URL_SETTING => { + if let Err(e) = reload_base_url_setting(conn).await { + tracing::error!(error = %e, "Could not reload base url setting"); + } + }, + OAUTH_SETTING => { + if let Err(e) = reload_base_url_setting(conn).await { + tracing::error!(error = %e, "Could not reload oauth setting"); + } + }, + CUSTOM_TAGS_SETTING => { + if let Err(e) = reload_custom_tags_setting(db).await { + tracing::error!(error = %e, "Could not reload custom tags setting"); + } + }, + LICENSE_KEY_SETTING => { + if let Err(e) = reload_license_key(&db.into()).await { + tracing::error!("Failed to reload license key: {e:#}"); + } + }, + DEFAULT_TAGS_PER_WORKSPACE_SETTING => { + if let Err(e) = load_tag_per_workspace_enabled(db).await { + tracing::error!("Error loading default tag per workspace: {e:#}"); + } + }, + DEFAULT_TAGS_WORKSPACES_SETTING => { + if let Err(e) = load_tag_per_workspace_workspaces(db).await { + tracing::error!("Error loading default tag per workspace workspaces: {e:#}"); + } + }, + SMTP_SETTING => { + reload_smtp_config(db).await; + }, + TEAMS_SETTING => { + tracing::info!("Teams setting changed."); + }, + INDEXER_SETTING => { + reload_indexer_config(db).await; + }, + TIMEOUT_WAIT_RESULT_SETTING => { + reload_timeout_wait_result_setting(conn).await + }, + RETENTION_PERIOD_SECS_SETTING => { + reload_retention_period_setting(conn).await + }, + MONITOR_LOGS_ON_OBJECT_STORE_SETTING => { + reload_delete_logs_periodically_setting(conn).await + }, + JOB_DEFAULT_TIMEOUT_SECS_SETTING => { + reload_job_default_timeout_setting(conn).await + }, + #[cfg(feature = "parquet")] + OBJECT_STORE_CONFIG_SETTING => { + if !disable_s3_store { + reload_object_store_setting(db).await; + } + }, + SCIM_TOKEN_SETTING => { + reload_scim_token_setting(conn).await + }, + EXTRA_PIP_INDEX_URL_SETTING => { + reload_extra_pip_index_url_setting(conn).await + }, + PIP_INDEX_URL_SETTING => { + reload_pip_index_url_setting(conn).await + }, + INSTANCE_PYTHON_VERSION_SETTING => { + reload_instance_python_version_setting(conn).await + }, + NPM_CONFIG_REGISTRY_SETTING => { + reload_npm_config_registry_setting(conn).await + }, + BUNFIG_INSTALL_SCOPES_SETTING => { + reload_bunfig_install_scopes_setting(conn).await + }, + NUGET_CONFIG_SETTING => { + reload_nuget_config_setting(conn).await + }, + POWERSHELL_REPO_URL_SETTING => { + reload_powershell_repo_url_setting(conn).await + }, + POWERSHELL_REPO_PAT_SETTING => { + reload_powershell_repo_pat_setting(conn).await + }, + MAVEN_REPOS_SETTING => { + reload_maven_repos_setting(conn).await + }, + NO_DEFAULT_MAVEN_SETTING => { + reload_no_default_maven_setting(conn).await + }, + RUBY_REPOS_SETTING => { + reload_ruby_repos_setting(conn).await + }, + HUB_API_SECRET_SETTING => { + reload_hub_api_secret_setting(conn).await + }, + KEEP_JOB_DIR_SETTING => { + load_keep_job_dir(conn).await; + }, + OTEL_TRACING_PROXY_SETTING => { + reload_otel_tracing_proxy_setting(conn).await; + if worker_mode { + tracing::info!("OTEL tracing proxy setting changed, restarting worker"); + send_delayed_killpill(tx, 4, "OTEL tracing proxy setting change").await; + } + }, + REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING => { + load_require_preexisting_user(db).await; + }, + EXPOSE_METRICS_SETTING => { + tracing::info!("Metrics setting changed, restarting"); + send_delayed_killpill(tx, 40, "metrics setting change").await; + }, + EMAIL_DOMAIN_SETTING => { + tracing::info!("Email domain setting changed"); + if server_mode { + send_delayed_killpill(tx, 4, "email domain setting change").await; + } + }, + EXPOSE_DEBUG_METRICS_SETTING => { + if let Err(e) = load_metrics_debug_enabled(conn).await { + tracing::error!(error = %e, "Could not reload debug metrics setting"); + } + }, + APP_WORKSPACED_ROUTE_SETTING => { + if let Err(e) = reload_app_workspaced_route_setting(db).await { + tracing::error!(error = %e, "Could not reload app workspaced route setting"); + } + }, + OTEL_SETTING => { + tracing::info!("OTEL setting changed, restarting"); + send_delayed_killpill(tx, 4, "OTEL setting change").await; + }, + REQUEST_SIZE_LIMIT_SETTING => { + if server_mode { + tracing::info!("Request limit size change detected, killing server expecting to be restarted"); + send_delayed_killpill(tx, 4, "request size limit change").await; + } + }, + SAML_METADATA_SETTING => { + tracing::info!("SAML metadata change detected, killing server expecting to be restarted"); + send_delayed_killpill(tx, 0, "SAML metadata change").await; + }, + HUB_BASE_URL_SETTING => { + if let Err(e) = reload_hub_base_url_setting(conn, server_mode).await { + tracing::error!(error = %e, "Could not reload hub base url setting"); + } + }, + CRITICAL_ERROR_CHANNELS_SETTING => { + if let Err(e) = reload_critical_error_channels_setting(db).await { + tracing::error!(error = %e, "Could not reload critical error emails setting"); + } + }, + CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING => { + if let Err(e) = reload_critical_alerts_on_db_oversize(db).await { + tracing::error!(error = %e, "Could not reload critical alerts on db oversize setting"); + } + }, + JWT_SECRET_SETTING => { + if let Err(e) = reload_jwt_secret_setting(db).await { + tracing::error!(error = %e, "Could not reload jwt secret setting"); + } + }, + CRITICAL_ALERT_MUTE_UI_SETTING => { + tracing::info!("Critical alert UI setting changed"); + if let Err(e) = reload_critical_alert_mute_ui_setting(conn).await { + tracing::error!(error = %e, "Could not reload critical alert UI setting"); + } + }, + _ => { + tracing::info!("Unrecognized Global Setting Change Payload: {:?}", payload); + } + } + }, + _ => { + tracing::warn!("Unknown notification channel: {}", channel); } } } diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 3e4e7dce92..07a4c5069b 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1921,6 +1921,24 @@ pub async fn monitor_db( } }; + // Run every 5 minutes (10 iterations * 30s = 5 minutes) + // Cleanup old notify events (older than 10 minutes) + let cleanup_notify_events_f = async { + if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(10) { + if let Some(db) = conn.as_sql() { + match windmill_common::notify_events::cleanup_old_events(db, 10).await { + Ok(count) if count > 0 => { + tracing::debug!("Cleaned up {} old notify events", count); + } + Err(e) => { + tracing::error!("Error cleaning up notify events: {:?}", e); + } + _ => {} + } + } + } + }; + join!( expired_items_f, zombie_jobs_f, @@ -1940,6 +1958,7 @@ pub async fn monitor_db( cleanup_flow_iterator_data_f, cleanup_worker_group_stats_f, native_triggers_sync_f, + cleanup_notify_events_f, ); } diff --git a/backend/tests/notify_events.rs b/backend/tests/notify_events.rs new file mode 100644 index 0000000000..a85a2f1c5a --- /dev/null +++ b/backend/tests/notify_events.rs @@ -0,0 +1,835 @@ +/*! + * Tests for the polling-based notify_event system that replaces PostgreSQL LISTEN/NOTIFY. + * + * These tests verify: + * 1. Database triggers correctly insert events into notify_event table + * 2. Polling functions retrieve events correctly + * 3. Cleanup functions delete old events + * 4. All notification channels work as expected + */ + +use sqlx::{Pool, Postgres}; +use windmill_common::notify_events::{cleanup_old_events, get_latest_event_id, poll_notify_events}; + +mod common; + +/// Helper to insert a test event directly +async fn insert_test_event(db: &Pool, channel: &str, payload: &str) -> i64 { + sqlx::query_scalar::<_, i64>( + "INSERT INTO notify_event (channel, payload) VALUES ($1, $2) RETURNING id", + ) + .bind(channel) + .bind(payload) + .fetch_one(db) + .await + .expect("Failed to insert test event") +} + +/// Helper to count events for a channel +async fn count_events_for_channel(db: &Pool, channel: &str) -> i64 { + let result: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM notify_event WHERE channel = $1") + .bind(channel) + .fetch_one(db) + .await + .expect("Failed to count events"); + result.0 +} + +// ============================================================================ +// Basic Functionality Tests +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_get_latest_event_id_returns_valid_id(db: Pool) { + // Get current latest id + let latest_id = get_latest_event_id(&db).await.expect("Should get latest event id"); + assert!(latest_id >= 0, "Latest id should be non-negative"); + + // Insert a new event and verify latest_id increases + let new_id = insert_test_event(&db, "test_latest_id", "payload").await; + let new_latest_id = get_latest_event_id(&db).await.expect("Should get latest event id"); + assert!(new_latest_id >= new_id, "Latest id should be >= new event id"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_get_latest_event_id_with_events(db: Pool) { + let _id1 = insert_test_event(&db, "test_channel_1", "payload1").await; + let _id2 = insert_test_event(&db, "test_channel_2", "payload2").await; + let id3 = insert_test_event(&db, "test_channel_3", "payload3").await; + + let latest_id = get_latest_event_id(&db).await.expect("Should get latest event id"); + assert!(latest_id >= id3, "Latest id should be >= last inserted id"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_poll_notify_events_no_new_events(db: Pool) { + // Get latest id first + let latest_id = get_latest_event_id(&db).await.unwrap(); + + // Poll from the latest id - should return empty since no new events + let events = poll_notify_events(&db, latest_id).await.expect("Should poll events"); + assert!(events.is_empty(), "Should return empty vec when polling from latest id"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_poll_notify_events_returns_new_events(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + let _id1 = insert_test_event(&db, "test_poll_channel", "payload1").await; + let _id2 = insert_test_event(&db, "test_poll_channel", "payload2").await; + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + assert!(events.len() >= 2, "Should return at least 2 new events"); + + // Verify the events we inserted are present + let our_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "test_poll_channel") + .collect(); + assert_eq!(our_events.len(), 2, "Should have exactly our 2 test events"); + + // Verify ordering (ascending by id) + assert!(our_events[0].id < our_events[1].id, "Events should be ordered by id ascending"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_poll_notify_events_respects_last_event_id(db: Pool) { + let id1 = insert_test_event(&db, "test_respect_id", "payload1").await; + let _id2 = insert_test_event(&db, "test_respect_id", "payload2").await; + let _id3 = insert_test_event(&db, "test_respect_id", "payload3").await; + + // Poll from id1 should only return id2 and id3 + let events = poll_notify_events(&db, id1).await.expect("Should poll events"); + let our_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "test_respect_id") + .collect(); + + assert_eq!(our_events.len(), 2, "Should only return events after id1"); + assert!(our_events.iter().all(|e| e.id > id1), "All events should have id > id1"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_cleanup_old_events(db: Pool) { + // Use unique channel names to avoid interference from other tests + let old_channel = format!("test_cleanup_old_{}", uuid::Uuid::new_v4()); + let recent_channel = format!("test_cleanup_recent_{}", uuid::Uuid::new_v4()); + + // Insert an event with old timestamp + sqlx::query( + "INSERT INTO notify_event (channel, payload, created_at) VALUES ($1, $2, now() - interval '15 minutes')", + ) + .bind(&old_channel) + .bind("old_payload") + .execute(&db) + .await + .expect("Failed to insert old event"); + + // Insert a recent event + sqlx::query( + "INSERT INTO notify_event (channel, payload) VALUES ($1, $2)", + ) + .bind(&recent_channel) + .bind("recent_payload") + .execute(&db) + .await + .expect("Failed to insert recent event"); + + // Count before cleanup + let old_count_before = count_events_for_channel(&db, &old_channel).await; + assert_eq!(old_count_before, 1, "Should have 1 old event before cleanup"); + + // Cleanup events older than 10 minutes + let deleted = cleanup_old_events(&db, 10).await.expect("Should cleanup events"); + assert!(deleted >= 1, "Should delete at least 1 old event"); + + // Verify old event is gone + let old_count = count_events_for_channel(&db, &old_channel).await; + assert_eq!(old_count, 0, "Old event should be deleted"); + + // Verify recent event is still there + let recent_count = count_events_for_channel(&db, &recent_channel).await; + assert_eq!(recent_count, 1, "Recent event should still exist"); +} + +// ============================================================================ +// Database Trigger Tests - Verify triggers insert events correctly +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_config_change(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Insert or update a config entry + sqlx::query( + "INSERT INTO config (name, config) VALUES ('test_config_trigger', '{}'::jsonb) + ON CONFLICT (name) DO UPDATE SET config = '{}'::jsonb", + ) + .execute(&db) + .await + .expect("Failed to insert config"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let config_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_config_change" && e.payload == "test_config_trigger") + .collect(); + + assert!(!config_events.is_empty(), "Should have notify_config_change event"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_global_setting_change_insert(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Use a unique setting name for testing + let setting_name = format!("test_setting_{}", uuid::Uuid::new_v4()); + + // Insert a global setting + sqlx::query("INSERT INTO global_settings (name, value) VALUES ($1, '{}'::jsonb)") + .bind(&setting_name) + .execute(&db) + .await + .expect("Failed to insert global setting"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let setting_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_global_setting_change" && e.payload == setting_name) + .collect(); + + assert!(!setting_events.is_empty(), "Should have notify_global_setting_change event on insert"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_global_setting_change_update(db: Pool) { + // Use a unique setting name for testing + let setting_name = format!("test_setting_update_{}", uuid::Uuid::new_v4()); + + // First insert + sqlx::query("INSERT INTO global_settings (name, value) VALUES ($1, '{}'::jsonb)") + .bind(&setting_name) + .execute(&db) + .await + .expect("Failed to insert global setting"); + + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Update the setting + sqlx::query("UPDATE global_settings SET value = '{\"updated\": true}'::jsonb WHERE name = $1") + .bind(&setting_name) + .execute(&db) + .await + .expect("Failed to update global setting"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let setting_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_global_setting_change" && e.payload == setting_name) + .collect(); + + assert!(!setting_events.is_empty(), "Should have notify_global_setting_change event on update"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_global_setting_change_delete(db: Pool) { + // Use a unique setting name for testing + let setting_name = format!("test_setting_delete_{}", uuid::Uuid::new_v4()); + + // First insert + sqlx::query("INSERT INTO global_settings (name, value) VALUES ($1, '{}'::jsonb)") + .bind(&setting_name) + .execute(&db) + .await + .expect("Failed to insert global setting"); + + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Delete the setting + sqlx::query("DELETE FROM global_settings WHERE name = $1") + .bind(&setting_name) + .execute(&db) + .await + .expect("Failed to delete global setting"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let setting_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_global_setting_change" && e.payload == setting_name) + .collect(); + + assert!(!setting_events.is_empty(), "Should have notify_global_setting_change event on delete"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_workspace_envs_change(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Insert a workspace env (test-workspace exists from fixture) + sqlx::query( + "INSERT INTO workspace_env (workspace_id, name, value) VALUES ('test-workspace', 'TEST_ENV_VAR', 'test_value') + ON CONFLICT (workspace_id, name) DO UPDATE SET value = 'test_value_updated'", + ) + .execute(&db) + .await + .expect("Failed to insert workspace env"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let env_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_workspace_envs_change" && e.payload == "test-workspace") + .collect(); + + assert!(!env_events.is_empty(), "Should have notify_workspace_envs_change event"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_workspace_key_change(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Insert a workspace key (base fixture already has one, so this will conflict and update) + sqlx::query( + "INSERT INTO workspace_key (workspace_id, kind, key) VALUES ('test-workspace', 'cloud', 'test_key_value') + ON CONFLICT (workspace_id, kind) DO UPDATE SET key = 'test_key_value_updated'", + ) + .execute(&db) + .await + .expect("Failed to insert workspace key"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let key_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_workspace_key_change" && e.payload == "test-workspace") + .collect(); + + assert!(!key_events.is_empty(), "Should have notify_workspace_key_change event"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_token_invalidation(db: Pool) { + // First insert a session token + let token = format!("test_token_{}", uuid::Uuid::new_v4()); + sqlx::query( + "INSERT INTO token (token, label, email, workspace_id, owner, expiration) + VALUES ($1, 'session', 'test@test.com', 'test-workspace', 'test-user', now() + interval '1 hour')", + ) + .bind(&token) + .execute(&db) + .await + .expect("Failed to insert token"); + + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Delete the token (should trigger notification) + sqlx::query("DELETE FROM token WHERE token = $1") + .bind(&token) + .execute(&db) + .await + .expect("Failed to delete token"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let token_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_token_invalidation" && e.payload == token) + .collect(); + + assert!(!token_events.is_empty(), "Should have notify_token_invalidation event"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_webhook_change(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Update webhook setting (workspace_settings exists from base fixture) + sqlx::query("UPDATE workspace_settings SET webhook = 'https://test.webhook.com' WHERE workspace_id = 'test-workspace'") + .execute(&db) + .await + .expect("Failed to update webhook"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let webhook_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_webhook_change" && e.payload == "test-workspace") + .collect(); + + assert!(!webhook_events.is_empty(), "Should have notify_webhook_change event"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_workspace_premium_change(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Toggle premium status + sqlx::query("UPDATE workspace SET premium = NOT premium WHERE id = 'test-workspace'") + .execute(&db) + .await + .expect("Failed to update workspace premium"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let premium_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_workspace_premium_change" && e.payload == "test-workspace") + .collect(); + + assert!(!premium_events.is_empty(), "Should have notify_workspace_premium_change event"); +} + +// ============================================================================ +// HTTP Trigger Tests +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_http_trigger_change(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + let trigger_path = format!("test_http_trigger_{}", uuid::Uuid::new_v4()); + + // Insert an HTTP trigger + sqlx::query( + "INSERT INTO http_trigger (path, route_path, route_path_key, script_path, is_flow, workspace_id, edited_by, email, http_method, authentication_method) + VALUES ($1, '/test/route', '/test/route', 'test/script', false, 'test-workspace', 'test-user', 'test@test.com', 'get', 'none')", + ) + .bind(&trigger_path) + .execute(&db) + .await + .expect("Failed to insert HTTP trigger"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let http_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_http_trigger_change") + .filter(|e| e.payload.contains("test-workspace") && e.payload.contains(&trigger_path)) + .collect(); + + assert!(!http_events.is_empty(), "Should have notify_http_trigger_change event"); +} + +// ============================================================================ +// Script/Flow Version Change Tests +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_runnable_version_change_script(db: Pool) { + // First create a script without lock + let script_path = format!("f/test/script_{}", uuid::Uuid::new_v4()); + let script_hash: i64 = rand::random::().abs(); + + sqlx::query( + "INSERT INTO script (workspace_id, hash, path, summary, description, content, created_by, language, kind) + VALUES ('test-workspace', $1, $2, 'test', 'test', 'def main(): pass', 'test-user', 'python3', 'script')", + ) + .bind(script_hash) + .bind(&script_path) + .execute(&db) + .await + .expect("Failed to insert script"); + + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Update the lock field (this should trigger the notification) + sqlx::query("UPDATE script SET lock = 'test_lock_content' WHERE hash = $1") + .bind(script_hash) + .execute(&db) + .await + .expect("Failed to update script lock"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let script_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_runnable_version_change") + .filter(|e| e.payload.contains("test-workspace") && e.payload.contains("script")) + .collect(); + + assert!(!script_events.is_empty(), "Should have notify_runnable_version_change event for script"); + + // Verify payload format: workspace_id:source_type:path:kind + let parts: Vec<&str> = script_events[0].payload.split(':').collect(); + assert!(parts.len() >= 4, "Payload should have at least 4 parts"); + assert_eq!(parts[0], "test-workspace", "First part should be workspace_id"); + assert_eq!(parts[1], "script", "Second part should be 'script'"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_trigger_notify_runnable_version_change_flow(db: Pool) { + // First create a flow with empty versions array + let flow_path = format!("f/test/flow_{}", uuid::Uuid::new_v4()); + + sqlx::query( + "INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, schema, versions) + VALUES ('test-workspace', $1, 'test', 'test', '{}'::jsonb, 'test-user', '{}'::json, ARRAY[]::bigint[])", + ) + .bind(&flow_path) + .execute(&db) + .await + .expect("Failed to insert flow"); + + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Update the flow's versions array (this triggers flow_versions_append_trigger) + sqlx::query( + "UPDATE flow SET versions = array_append(versions, 1::bigint) WHERE workspace_id = 'test-workspace' AND path = $1", + ) + .bind(&flow_path) + .execute(&db) + .await + .expect("Failed to update flow versions"); + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let flow_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "notify_runnable_version_change") + .filter(|e| e.payload.contains("test-workspace") && e.payload.contains("flow")) + .collect(); + + assert!(!flow_events.is_empty(), "Should have notify_runnable_version_change event for flow"); + + // Verify payload format + let parts: Vec<&str> = flow_events[0].payload.split(':').collect(); + assert!(parts.len() >= 4, "Payload should have at least 4 parts"); + assert_eq!(parts[0], "test-workspace", "First part should be workspace_id"); + assert_eq!(parts[1], "flow", "Second part should be 'flow'"); +} + +// ============================================================================ +// Concurrent Access Tests +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_concurrent_event_insertion(db: Pool) { + // Use a unique channel name for this test run + let channel = format!("test_concurrent_{}", uuid::Uuid::new_v4()); + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Insert multiple events concurrently + let handles: Vec<_> = (0..10) + .map(|i| { + let db = db.clone(); + let ch = channel.clone(); + tokio::spawn(async move { + sqlx::query_scalar::<_, i64>( + "INSERT INTO notify_event (channel, payload) VALUES ($1, $2) RETURNING id", + ) + .bind(&ch) + .bind(format!("payload_{}", i)) + .fetch_one(&db) + .await + .expect("Failed to insert event") + }) + }) + .collect(); + + // Wait for all insertions + for handle in handles { + handle.await.expect("Task should complete"); + } + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let concurrent_events: Vec<_> = events + .iter() + .filter(|e| e.channel == channel) + .collect(); + + assert_eq!(concurrent_events.len(), 10, "Should have all 10 concurrent events"); + + // Verify all events have unique IDs + let ids: std::collections::HashSet = concurrent_events.iter().map(|e| e.id).collect(); + assert_eq!(ids.len(), 10, "All events should have unique IDs"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_polling_isolation(db: Pool) { + // Use a unique channel name for this test + let channel = format!("test_isolation_{}", uuid::Uuid::new_v4()); + + // Get baseline before inserting + let baseline_id = get_latest_event_id(&db).await.unwrap(); + + // Insert some events + let id1 = sqlx::query_scalar::<_, i64>( + "INSERT INTO notify_event (channel, payload) VALUES ($1, $2) RETURNING id", + ) + .bind(&channel) + .bind("payload1") + .fetch_one(&db) + .await + .expect("Failed to insert event"); + + let id2 = sqlx::query_scalar::<_, i64>( + "INSERT INTO notify_event (channel, payload) VALUES ($1, $2) RETURNING id", + ) + .bind(&channel) + .bind("payload2") + .fetch_one(&db) + .await + .expect("Failed to insert event"); + + let _id3 = sqlx::query_scalar::<_, i64>( + "INSERT INTO notify_event (channel, payload) VALUES ($1, $2) RETURNING id", + ) + .bind(&channel) + .bind("payload3") + .fetch_one(&db) + .await + .expect("Failed to insert event"); + + // Two different "consumers" polling from different points + let events_from_baseline = poll_notify_events(&db, baseline_id).await.expect("Should poll events"); + let events_from_id1 = poll_notify_events(&db, id1).await.expect("Should poll events"); + let events_from_id2 = poll_notify_events(&db, id2).await.expect("Should poll events"); + + // Filter to our test events + let from_baseline: Vec<_> = events_from_baseline.iter().filter(|e| e.channel == channel).collect(); + let from_id1: Vec<_> = events_from_id1.iter().filter(|e| e.channel == channel).collect(); + let from_id2: Vec<_> = events_from_id2.iter().filter(|e| e.channel == channel).collect(); + + assert_eq!(from_baseline.len(), 3, "Polling from baseline should include all 3 events"); + assert_eq!(from_id1.len(), 2, "Polling from id1 should include id2 and id3"); + assert_eq!(from_id2.len(), 1, "Polling from id2 should include only id3"); +} + +// ============================================================================ +// Edge Case Tests +// ============================================================================ + +#[sqlx::test(fixtures("base"))] +async fn test_empty_payload(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + insert_test_event(&db, "test_empty_payload", "").await; + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let empty_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "test_empty_payload") + .collect(); + + assert_eq!(empty_events.len(), 1, "Should have event with empty payload"); + assert_eq!(empty_events[0].payload, "", "Payload should be empty string"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_large_payload(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + // Create a large payload (1KB) + let large_payload = "x".repeat(1024); + insert_test_event(&db, "test_large_payload", &large_payload).await; + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let large_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "test_large_payload") + .collect(); + + assert_eq!(large_events.len(), 1, "Should have event with large payload"); + assert_eq!(large_events[0].payload.len(), 1024, "Payload should be preserved"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_special_characters_in_payload(db: Pool) { + let before_id = get_latest_event_id(&db).await.unwrap(); + + let special_payload = r#"{"key": "value with \"quotes\" and 'apostrophes'", "unicode": "日本語", "newline": "line1\nline2"}"#; + insert_test_event(&db, "test_special_chars", special_payload).await; + + let events = poll_notify_events(&db, before_id).await.expect("Should poll events"); + let special_events: Vec<_> = events + .iter() + .filter(|e| e.channel == "test_special_chars") + .collect(); + + assert_eq!(special_events.len(), 1, "Should have event with special characters"); + assert_eq!(special_events[0].payload, special_payload, "Special characters should be preserved"); +} + +#[sqlx::test(fixtures("base"))] +async fn test_cleanup_with_no_old_events(db: Pool) { + // Use a unique channel name for this test + let channel = format!("test_no_old_{}", uuid::Uuid::new_v4()); + + // Insert only recent events + sqlx::query("INSERT INTO notify_event (channel, payload) VALUES ($1, $2)") + .bind(&channel) + .bind("recent1") + .execute(&db) + .await + .expect("Failed to insert event"); + + sqlx::query("INSERT INTO notify_event (channel, payload) VALUES ($1, $2)") + .bind(&channel) + .bind("recent2") + .execute(&db) + .await + .expect("Failed to insert event"); + + let before_count = count_events_for_channel(&db, &channel).await; + assert_eq!(before_count, 2, "Should have 2 recent events"); + + // Cleanup old events (none of our events should be deleted since they're recent) + let _deleted = cleanup_old_events(&db, 10).await.expect("Should cleanup events"); + + let after_count = count_events_for_channel(&db, &channel).await; + assert_eq!(after_count, 2, "Recent events should not be deleted"); +} + +// ============================================================================ +// Multi-Server Integration Tests +// ============================================================================ +// These tests start two actual windmill server processes on different ports +// with LISTEN_NEW_EVENTS_INTERVAL_SEC=1, trigger DB changes, and verify +// both servers process the events via their log output. + +use std::io::{BufRead, BufReader}; +use std::process::{Child, Command, Stdio}; +use std::sync::{Arc, Mutex}; + +struct ServerProcess { + child: Child, + log_lines: Arc>>, + _stdout_handle: std::thread::JoinHandle<()>, + _stderr_handle: std::thread::JoinHandle<()>, +} + +impl ServerProcess { + fn start(port: u16, db_url: &str) -> Self { + let binary = std::env::var("WINDMILL_BINARY") + .unwrap_or_else(|_| format!("{}/target/debug/windmill", env!("CARGO_MANIFEST_DIR"))); + + let mut child = Command::new(&binary) + .env("DATABASE_URL", db_url) + .env("MODE", "server") + .env("PORT", port.to_string()) + .env("LISTEN_NEW_EVENTS_INTERVAL_SEC", "1") + .env("RUST_LOG", "info") + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap_or_else(|e| panic!("Failed to start windmill on port {port}: {e}")); + + let stdout = child.stdout.take().expect("Failed to capture stdout"); + let stderr = child.stderr.take().expect("Failed to capture stderr"); + let log_lines = Arc::new(Mutex::new(Vec::new())); + let log_lines_stdout = log_lines.clone(); + let log_lines_stderr = log_lines.clone(); + + // Read both stdout and stderr into the same log buffer + let _reader_handle = std::thread::spawn(move || { + let reader = BufReader::new(stdout); + for line in reader.lines() { + if let Ok(line) = line { + log_lines_stdout.lock().unwrap().push(line); + } + } + }); + let _stderr_handle = std::thread::spawn(move || { + let reader = BufReader::new(stderr); + for line in reader.lines() { + if let Ok(line) = line { + log_lines_stderr.lock().unwrap().push(line); + } + } + }); + + ServerProcess { child, log_lines, _stdout_handle: _reader_handle, _stderr_handle } + } + + fn logs_contain(&self, needle: &str) -> bool { + self.log_lines.lock().unwrap().iter().any(|l| l.contains(needle)) + } + + fn dump_logs(&self) -> String { + self.log_lines.lock().unwrap().join("\n") + } +} + +impl Drop for ServerProcess { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +/// Wait for server to be ready by polling its HTTP endpoint. +async fn wait_for_server(port: u16, timeout_secs: u64) -> bool { + let client = reqwest::Client::new(); + let url = format!("http://127.0.0.1:{}/api/version", port); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(timeout_secs); + while tokio::time::Instant::now() < deadline { + if client.get(&url).send().await.is_ok() { + return true; + } + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } + false +} + +/// Helper to get a database connection (only used by the e2e multi-server test) +async fn get_db() -> Pool { + let database_url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://postgres:changeme@localhost:5432/windmill".to_string()); + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&database_url) + .await + .expect("Failed to connect to database") +} + +#[tokio::test] +#[ignore = "slow - starts two server processes with 1s poll interval"] +async fn test_two_server_processes_both_receive_event() { + let db_url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://postgres:changeme@localhost:5432/windmill".to_string()); + + // Start two server processes on different ports with 1s poll interval + let mut server_a = ServerProcess::start(19100, &db_url); + let mut server_b = ServerProcess::start(19200, &db_url); + + // Wait for both servers to be ready + let (ready_a, ready_b) = tokio::join!( + wait_for_server(19100, 30), + wait_for_server(19200, 30), + ); + assert!(ready_a, "Server A (port 19100) failed to start. Logs:\n{}", server_a.dump_logs()); + assert!(ready_b, "Server B (port 19200) failed to start. Logs:\n{}", server_b.dump_logs()); + + // Give servers a moment to complete their first poll cycle + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + + // Trigger a global setting change via direct DB insert + let db = get_db().await; + let setting_name = format!("test_e2e_{}", uuid::Uuid::new_v4()); + sqlx::query( + "INSERT INTO global_settings (name, value) VALUES ($1, '\"e2e_test\"'::jsonb) + ON CONFLICT (name) DO UPDATE SET value = '\"e2e_test\"'::jsonb", + ) + .bind(&setting_name) + .execute(&db) + .await + .expect("Failed to insert global setting"); + + // Wait for at least 2 poll cycles (interval is 1s) + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + + let needle = format!("Global setting change detected: {}", setting_name); + assert!( + server_a.logs_contain(&needle), + "Server A should have processed the global setting event.\nSearching for: {}\nServer A logs:\n{}", + needle, server_a.dump_logs() + ); + assert!( + server_b.logs_contain(&needle), + "Server B should have processed the global setting event.\nSearching for: {}\nServer B logs:\n{}", + needle, server_b.dump_logs() + ); + + // Cleanup + sqlx::query("DELETE FROM global_settings WHERE name = $1") + .bind(&setting_name) + .execute(&db) + .await + .ok(); + + // Explicitly kill before drop to avoid port conflicts with other tests + let _ = server_a.child.kill(); + let _ = server_b.child.kill(); +} diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 83e7a2312f..3fd4de438a 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -60,6 +60,7 @@ pub mod job_metrics; #[cfg(all(feature = "parquet", feature = "private"))] pub mod job_s3_helpers_ee; pub mod min_version; +pub mod notify_events; #[cfg(feature = "parquet")] pub mod job_s3_helpers_oss; pub mod workspace_dependencies; diff --git a/backend/windmill-common/src/notify_events.rs b/backend/windmill-common/src/notify_events.rs new file mode 100644 index 0000000000..e00b6d7d49 --- /dev/null +++ b/backend/windmill-common/src/notify_events.rs @@ -0,0 +1,63 @@ +/* + * Author: Windmill Labs + * Copyright: Windmill Labs, Inc 2024 + * 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. + */ + +//! Polling-based event notification system. +//! +//! This module provides a table-based alternative to PostgreSQL LISTEN/NOTIFY +//! for propagating cache invalidation and setting change events across +//! workers and servers. + +use sqlx::{FromRow, Pool, Postgres}; + +use crate::error::Error; + +#[derive(Debug, Clone, FromRow)] +pub struct NotifyEvent { + pub id: i64, + pub channel: String, + pub payload: String, +} + +/// Fetch all events with id greater than `last_event_id`. +/// Returns events ordered by id ascending. +pub async fn poll_notify_events( + db: &Pool, + last_event_id: i64, +) -> Result, Error> { + let events = sqlx::query_as::<_, NotifyEvent>( + "SELECT id, channel, payload FROM notify_event WHERE id > $1 ORDER BY id LIMIT 1000", + ) + .bind(last_event_id) + .fetch_all(db) + .await?; + + Ok(events) +} + +/// Get the current maximum event id. +/// Used to initialize last_event_id on startup to avoid processing old events. +pub async fn get_latest_event_id(db: &Pool) -> Result { + let result: (i64,) = sqlx::query_as("SELECT COALESCE(MAX(id), 0) FROM notify_event") + .fetch_one(db) + .await?; + + Ok(result.0) +} + +/// Delete events older than the specified number of minutes. +/// Returns the number of deleted rows. +pub async fn cleanup_old_events(db: &Pool, older_than_minutes: i32) -> Result { + let result = sqlx::query( + "DELETE FROM notify_event WHERE created_at < now() - make_interval(mins => $1)", + ) + .bind(older_than_minutes) + .execute(db) + .await?; + + Ok(result.rows_affected()) +}