From 14b7516b3f1e9a64b7955f9a300a4a2fbec42b55 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Mon, 11 May 2026 01:22:24 -0700 Subject: [PATCH] fix(notify): unbreak PG test backend on the notify dispatcher suite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI's postgres-backend run failed 11 of the new notify tests from #505. Three independent issues: 1. Migration 053's ``services_notify`` trigger lives only in the alembic chain, but the test fixture in conftest.py calls ``init_storage(..., run_migrations=False)`` for speed. That path skips migrations and relies on ``metadata.create_all`` for the table tree. Previous alembic-only DDL (migrations 041 / 048 ``CREATE INDEX CONCURRENTLY`` on workstreams) is performance-only, so tests never depended on it. 053's trigger is the first behaviorally-required alembic-only DDL in the project — without it ``register_service`` doesn't fire NOTIFY and the trigger-filter tests time out. Fix: declare the trigger function + trigger in ``_schema.py`` and attach them via ``sa.event.listen(services, "after_create", ...)`` DDL events, gated on ``dialect == "postgresql"``. The same SQL constants are imported by migration 053 so there's a single source of truth. Test fixture stays unchanged — ``create_all`` now installs the trigger on fresh PG test DBs. Migration covers the upgrade-on-existing-DB path; the two are mutually exclusive given ``create_tables = not run_migrations`` in ``init_storage``. 2. NotifyDispatcher tests fired ``storage.notify(...)`` immediately after ``d.start()`` and hit a race: the listener thread is concurrently calling ``psycopg.connect(listen_url)`` + ``LISTEN `` over the network, so the notify can land before any session is listening on the channel and PG drops it (pg_notify only routes to sessions LISTEN'ing at COMMIT time). Fix: dispatcher gains a ``_listener_ready: threading.Event`` set inside ``_listener_loop`` after each successful ``storage.listen`` open and cleared on disconnect, plus a public ``wait_until_ready(timeout)`` method. Tests use a new ``_start_ready(d)`` helper that calls ``start()`` + asserts ready. Production callers don't need this (real reactive traffic arrives well after startup), but it's the right primitive for any future "start dispatcher, immediately send" call site too. 3. ``TestSqliteNotify`` is misnamed — its tests run against whichever backend the ``storage`` fixture provides (PG by default in CI). Two of its assertions were SQLite-specific: ``assert got.pid == 0`` only holds for the synthetic in-process path (PG carries real backend PIDs), and ``test_synthetic_sweep_emits_after_interval`` is fundamentally SQLite-only (no sweep on the PG path). Fix: drop the pid assertion (channel + payload are the backend-agnostic invariants), add an ``_is_sqlite`` fixture mirror of ``_is_postgres``, and gate the sweep test on it. The sweep test also moves from monkey-patching ``stream._sweep_interval`` to passing the ``sweep_interval`` kwarg that ``SQLiteBackend.listen`` now accepts (from the earlier Copilot review fix). Validated locally against a fresh ``turnstone_test`` PG DB: 263 storage + console + notify tests pass on PG, 257 on SQLite, mypy + ruff clean. --- tests/test_notify_dispatcher.py | 26 +++++-- tests/test_storage_notify.py | 24 +++++-- turnstone/console/notify_dispatcher.py | 36 ++++++++++ turnstone/core/storage/_schema.py | 70 +++++++++++++++++++ .../versions/053_services_notify_trigger.py | 55 +++------------ 5 files changed, 154 insertions(+), 57 deletions(-) diff --git a/tests/test_notify_dispatcher.py b/tests/test_notify_dispatcher.py index 59c41c3a..e385de2d 100644 --- a/tests/test_notify_dispatcher.py +++ b/tests/test_notify_dispatcher.py @@ -43,12 +43,26 @@ def _wait_for(predicate, deadline_sec: float = 3.0) -> bool: return False +def _start_ready(d, *, timeout: float = 5.0) -> None: + """``d.start()`` + assert the listener is actually listening. + + Closes the start-vs-notify race for backends where ``storage.listen`` + blocks on the network (Postgres ``LISTEN`` over a fresh psycopg + connection): without the sync, a same-thread ``storage.notify`` can + fire before the LISTEN registers and the notification is lost. + """ + d.start() + if not d.wait_until_ready(timeout=timeout): + msg = f"dispatcher listener did not open within {timeout}s" + raise AssertionError(msg) + + class TestSubscribe: def test_subscribe_registers_handler(self, dispatcher_factory, storage): d = dispatcher_factory(channels=["alpha"]) seen: list = [] d.subscribe("alpha", lambda n: seen.append(n)) - d.start() + _start_ready(d) # Fire a notify via the storage layer — dispatcher delivers to handler. storage.notify("alpha", "hello") assert _wait_for(lambda: any(n.payload == "hello" for n in seen)) @@ -62,7 +76,7 @@ class TestSubscribe: d = dispatcher_factory(channels=["alpha"]) seen: list = [] unsub = d.subscribe("alpha", lambda n: seen.append(n)) - d.start() + _start_ready(d) storage.notify("alpha", "first") assert _wait_for(lambda: any(n.payload == "first" for n in seen)) unsub() @@ -92,7 +106,7 @@ class TestDispatch: seen_b: list = [] d.subscribe("alpha", lambda n: seen_a.append(n)) d.subscribe("alpha", lambda n: seen_b.append(n)) - d.start() + _start_ready(d) storage.notify("alpha", "shared") assert _wait_for(lambda: seen_a and seen_b) assert seen_a[0].payload == "shared" @@ -108,7 +122,7 @@ class TestDispatch: d.subscribe("alpha", _broken) d.subscribe("alpha", lambda n: survived.append(n)) - d.start() + _start_ready(d) storage.notify("alpha", "after_broken") # The second handler runs even though the first raised. assert _wait_for(lambda: any(n.payload == "after_broken" for n in survived)) @@ -119,7 +133,7 @@ class TestDispatch: seen_b: list = [] d.subscribe("alpha", lambda n: seen_a.append(n)) d.subscribe("beta", lambda n: seen_b.append(n)) - d.start() + _start_ready(d) storage.notify("alpha", "for_a") storage.notify("beta", "for_b") assert _wait_for(lambda: seen_a and seen_b) @@ -320,7 +334,7 @@ class TestCoalescing: coalesce_gate.wait(0.05) d.subscribe("alpha", _slow_handler) - d.start() + _start_ready(d) # Burst of 10 notifies on the same channel — should coalesce # down to many fewer handler invocations. for i in range(10): diff --git a/tests/test_storage_notify.py b/tests/test_storage_notify.py index a59332cd..07d2a02b 100644 --- a/tests/test_storage_notify.py +++ b/tests/test_storage_notify.py @@ -42,7 +42,9 @@ class TestSqliteNotify: storage.notify("services", '{"op": "INSERT"}') got = _drain_until(stream, lambda n: n.payload == '{"op": "INSERT"}') assert got.channel == "services" - assert got.pid == 0 + # ``pid`` is 0 on the SQLite synthetic path and the sending + # backend's PID on Postgres — both are valid notify shapes, + # so don't assert on the value here. def test_notify_filters_by_channel(self, storage): with storage.listen(["services"]) as stream: @@ -68,12 +70,12 @@ class TestSqliteNotify: assert stream.poll(0.05) == [] stream.close() - def test_synthetic_sweep_emits_after_interval(self, storage): - # Force a short sweep interval via direct attribute override — - # the production default (``_SQLITE_NOTIFY_SWEEP_INTERVAL``) is - # tuned for an idle dev backstop and is way too long for a test. - with storage.listen(["services"]) as stream: - stream._sweep_interval = 0.1 + def test_synthetic_sweep_emits_after_interval(self, storage, _is_sqlite): + # Synthetic sweep is fundamentally SQLite-specific — the PG path + # uses real ``LISTEN``/``NOTIFY`` and has no sweep tick. Gate + # so the test doesn't false-fail by waiting for a "sweep" notify + # that the PG stream will never produce. + with storage.listen(["services"], sweep_interval=0.1) as stream: # First poll: not yet at the interval, so likely empty. stream.poll(0.05) # Wait past the interval, then poll again — should emit a @@ -102,6 +104,14 @@ def _is_postgres(storage): return True +@pytest.fixture +def _is_sqlite(storage): + """Skip the wrapped test when the active backend isn't SQLite.""" + if storage.__class__.__name__ != "SQLiteBackend": + pytest.skip("SQLite-specific test") + return True + + class TestPostgresNotify: def test_round_trip(self, storage, _is_postgres): # Open a listener, fire a notify on a regular pooled connection, diff --git a/turnstone/console/notify_dispatcher.py b/turnstone/console/notify_dispatcher.py index 3a713393..48c31725 100644 --- a/turnstone/console/notify_dispatcher.py +++ b/turnstone/console/notify_dispatcher.py @@ -85,6 +85,14 @@ class NotifyDispatcher: self._dispatch_thread: threading.Thread | None = None self._dispatch_queue: queue.Queue[Notify | None] = queue.Queue(maxsize=_DISPATCH_QUEUE_MAX) self._drop_count = 0 + # Set inside :meth:`_listener_loop` after each successful + # ``storage.listen`` open; cleared on disconnect. Callers use + # :meth:`wait_until_ready` after :meth:`start` to block until the + # listener is actually listening (matters when the next caller + # action is a ``notify`` whose delivery requires the LISTEN to + # already be in place — e.g. tests, or any startup-path traffic + # that should be reactive from the first event). + self._listener_ready = threading.Event() @property def channels(self) -> list[str]: @@ -127,6 +135,9 @@ class NotifyDispatcher: return self._started = True self._stopping.clear() + # Clear ready so a stop/start cycle's wait_until_ready only + # returns True after the new listener has actually opened. + self._listener_ready.clear() self._listener_thread = threading.Thread( target=self._listener_loop, name="notify-dispatcher-listener", @@ -144,6 +155,23 @@ class NotifyDispatcher: channels=self._channels, ) + def wait_until_ready(self, timeout: float = 5.0) -> bool: + """Block until the listener has opened its stream, or ``timeout`` elapses. + + Returns ``True`` when the listener is ready (``LISTEN`` issued + for every declared channel on PG; subscriber queues registered + on SQLite), ``False`` on timeout. Cleared automatically on + disconnect — call again after a reconnect to wait for the next + successful reopen. + + Doesn't replace :meth:`start` — call ``start()`` first, then + ``wait_until_ready()`` for the explicit sync point. Production + startup typically doesn't need this (the first real event tends + to arrive well after the listener is up); tests use it to close + the start-vs-notify race window. + """ + return self._listener_ready.wait(timeout=timeout) + def stop(self, timeout: float = 5.0) -> None: """Signal shutdown and join the worker threads. @@ -206,6 +234,12 @@ class NotifyDispatcher: if reconcile_pending: self._synthesize_reconcile() reconcile_pending = False + # Signal ``wait_until_ready`` callers that LISTEN is + # in place (PG) / subscriber queues are bound + # (SQLite). Must come AFTER the synthesize so any + # post-reconnect reconcile reaches handlers before + # the caller assumes "fresh notifies will deliver". + self._listener_ready.set() while not self._stopping.is_set(): batch = stream.poll(_LISTENER_POLL_TIMEOUT) for n in batch: @@ -213,6 +247,7 @@ class NotifyDispatcher: except NotifyConnectionError as exc: if self._stopping.is_set(): return + self._listener_ready.clear() log.warning( "notify_dispatcher.connection_lost", error=str(exc), @@ -225,6 +260,7 @@ class NotifyDispatcher: except Exception: if self._stopping.is_set(): return + self._listener_ready.clear() log.exception("notify_dispatcher.listener_unexpected_error") reconcile_pending = True if self._stopping.wait(backoff): diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py index 1045bf4c..18ef1410 100644 --- a/turnstone/core/storage/_schema.py +++ b/turnstone/core/storage/_schema.py @@ -251,6 +251,76 @@ services = sa.Table( sa.Index("idx_services_type_heartbeat", services.c.service_type, services.c.last_heartbeat) + +# -- Postgres NOTIFY trigger on services ----------------------------------- +# +# Producer side of the ``services`` channel that the console +# ``NotifyDispatcher`` listens on for reactive node discovery. Fires on +# real registry changes (INSERT, DELETE, UPDATE that changes ``url`` or +# ``metadata``) and stays quiet on heartbeat-only UPDATEs so the 30s × N +# nodes heartbeat tick doesn't flood the channel. +# +# Declared in the schema (not just in migration 053) so the ``after_create`` +# DDL event installs the trigger any time ``metadata.create_all`` builds +# the ``services`` table — covering fresh dev databases and the test +# fixture path (``run_migrations=False``). Migration 053 covers the +# upgrade-on-existing-DB path; the two are mutually exclusive given the +# ``create_tables = not run_migrations`` switch in ``init_storage``, so +# neither double-installs. SQLite has no equivalent — the in-process +# notify fan-out and synthetic-sweep covers the dev path consumer-side. + +SERVICES_NOTIFY_TRIGGER_FN_NAME = "turnstone_notify_services" +SERVICES_NOTIFY_TRIGGER_NAME = "services_notify" + +SERVICES_NOTIFY_TRIGGER_FN_SQL = f""" +CREATE OR REPLACE FUNCTION {SERVICES_NOTIFY_TRIGGER_FN_NAME}() RETURNS trigger AS $$ +BEGIN + -- Skip heartbeat-only UPDATEs: same url and metadata, only + -- ``last_heartbeat`` changed. ``register_service`` is an UPSERT + -- (on_conflict_do_update), so node restarts that change url or + -- metadata MUST still fire — only no-op heartbeat ticks stay + -- quiet. IS NOT DISTINCT FROM treats NULLs as equal so a row + -- with NULL metadata before/after doesn't trip the diff. + IF TG_OP = 'UPDATE' + AND OLD.url IS NOT DISTINCT FROM NEW.url + AND OLD.metadata IS NOT DISTINCT FROM NEW.metadata THEN + RETURN NULL; + END IF; + + PERFORM pg_notify( + 'services', + json_build_object( + 'service_type', COALESCE(NEW.service_type, OLD.service_type), + 'service_id', COALESCE(NEW.service_id, OLD.service_id), + 'op', TG_OP + )::text + ); + RETURN NULL; +END; +$$ LANGUAGE plpgsql; +""" + +SERVICES_NOTIFY_TRIGGER_SQL = f""" +CREATE TRIGGER {SERVICES_NOTIFY_TRIGGER_NAME} +AFTER INSERT OR UPDATE OR DELETE ON services +FOR EACH ROW EXECUTE FUNCTION {SERVICES_NOTIFY_TRIGGER_FN_NAME}(); +""" + +sa.event.listen( + services, + "after_create", + sa.DDL(SERVICES_NOTIFY_TRIGGER_FN_SQL).execute_if( # type: ignore[no-untyped-call] + dialect="postgresql" + ), +) +sa.event.listen( + services, + "after_create", + sa.DDL(SERVICES_NOTIFY_TRIGGER_SQL).execute_if( # type: ignore[no-untyped-call] + dialect="postgresql" + ), +) + # --------------------------------------------------------------------------- # Node metadata (per-node key/value with source tracking) # --------------------------------------------------------------------------- diff --git a/turnstone/core/storage/migrations/versions/053_services_notify_trigger.py b/turnstone/core/storage/migrations/versions/053_services_notify_trigger.py index c3644d3f..4c3c9b63 100644 --- a/turnstone/core/storage/migrations/versions/053_services_notify_trigger.py +++ b/turnstone/core/storage/migrations/versions/053_services_notify_trigger.py @@ -33,63 +33,30 @@ Create Date: 2026-05-10 import sqlalchemy as sa from alembic import op +from turnstone.core.storage._schema import ( + SERVICES_NOTIFY_TRIGGER_FN_NAME, + SERVICES_NOTIFY_TRIGGER_FN_SQL, + SERVICES_NOTIFY_TRIGGER_NAME, + SERVICES_NOTIFY_TRIGGER_SQL, +) + revision = "053" down_revision = "052" branch_labels = None depends_on = None -_TRIGGER_FN_NAME = "turnstone_notify_services" -_TRIGGER_NAME = "services_notify" - - -_TRIGGER_FN_SQL = f""" -CREATE OR REPLACE FUNCTION {_TRIGGER_FN_NAME}() RETURNS trigger AS $$ -BEGIN - -- Skip heartbeat-only UPDATEs: same url and metadata, only - -- ``last_heartbeat`` changed. ``register_service`` is an UPSERT - -- (on_conflict_do_update), so node restarts that change url or - -- metadata MUST still fire — only no-op heartbeat ticks stay - -- quiet. IS NOT DISTINCT FROM treats NULLs as equal so a row - -- with NULL metadata before/after doesn't trip the diff. - IF TG_OP = 'UPDATE' - AND OLD.url IS NOT DISTINCT FROM NEW.url - AND OLD.metadata IS NOT DISTINCT FROM NEW.metadata THEN - RETURN NULL; - END IF; - - PERFORM pg_notify( - 'services', - json_build_object( - 'service_type', COALESCE(NEW.service_type, OLD.service_type), - 'service_id', COALESCE(NEW.service_id, OLD.service_id), - 'op', TG_OP - )::text - ); - RETURN NULL; -END; -$$ LANGUAGE plpgsql; -""" - - -_TRIGGER_SQL = f""" -CREATE TRIGGER {_TRIGGER_NAME} -AFTER INSERT OR UPDATE OR DELETE ON services -FOR EACH ROW EXECUTE FUNCTION {_TRIGGER_FN_NAME}(); -""" - - def upgrade() -> None: bind = op.get_bind() if bind.dialect.name != "postgresql": return - op.execute(sa.text(_TRIGGER_FN_SQL)) - op.execute(sa.text(_TRIGGER_SQL)) + op.execute(sa.text(SERVICES_NOTIFY_TRIGGER_FN_SQL)) + op.execute(sa.text(SERVICES_NOTIFY_TRIGGER_SQL)) def downgrade() -> None: bind = op.get_bind() if bind.dialect.name != "postgresql": return - op.execute(sa.text(f"DROP TRIGGER IF EXISTS {_TRIGGER_NAME} ON services")) - op.execute(sa.text(f"DROP FUNCTION IF EXISTS {_TRIGGER_FN_NAME}()")) + op.execute(sa.text(f"DROP TRIGGER IF EXISTS {SERVICES_NOTIFY_TRIGGER_NAME} ON services")) + op.execute(sa.text(f"DROP FUNCTION IF EXISTS {SERVICES_NOTIFY_TRIGGER_FN_NAME}()"))