starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: replace LISTEN/NOTIFY with polling-based event system (#7778)

Browse Source 
* feat: replace LISTEN/NOTIFY with polling-based event system

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

* test: add exhaustive tests for polling-based notify events

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

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* chore: reduce default polling interval from 30s to 10s

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

* 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-02-04 07:35:29 +00:00
committed by GitHub
parent b807e0f5fc
commit e860847073
7 changed files with 1508 additions and 388 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20260203172950_polling_based_events.down.sql
+121
View File
@@ -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;
backend/migrations/20260203172950_polling_based_events.up.sql
+135
View File
@@ -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.
backend/src/main.rs
+334 -388
View File
@@ -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::<u64>().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::<u64>().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::<Vec<&str>>().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::<serde_json::Value>(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::<serde_json::Value>(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<Postgres>) -> Option<PgListener> {
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<Postgres>) -> 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<Postgres>,
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::<Vec<&str>>().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);
}
}
}
backend/src/monitor.rs
+19
View File
@@ -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,
);
}
backend/tests/notify_events.rs
+835
View File
@@ -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<Postgres>, 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<Postgres>, 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<Postgres>) {
// 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<Postgres>) {
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<Postgres>) {
// 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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
// 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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
// 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<Postgres>) {
// 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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
// 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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
// First create a script without lock
let script_path = format!("f/test/script_{}", uuid::Uuid::new_v4());
let script_hash: i64 = rand::random::<i64>().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<Postgres>) {
// 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<Postgres>) {
// 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<i64> = 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<Postgres>) {
// 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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
// 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<Mutex<Vec<String>>>,
_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<Postgres> {
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();
}
backend/windmill-common/src/lib.rs
+1
View File
@@ -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;
backend/windmill-common/src/notify_events.rs
+63
View File
@@ -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<Postgres>,
last_event_id: i64,
) -> Result<Vec<NotifyEvent>, 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<Postgres>) -> Result<i64, Error> {
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<Postgres>, older_than_minutes: i32) -> Result<u64, Error> {
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())
}