Files
windmill/backend/migrations/20260203172950_polling_based_events.up.sql
Ruben Fiszel e860847073 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 <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>
2026-02-04 07:35:29 +00:00

136 lines
4.3 KiB
PL/PgSQL

-- 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.