diff --git a/app-tests/run.sh b/app-tests/run.sh index 6c4bf5657..f28df05cf 100755 --- a/app-tests/run.sh +++ b/app-tests/run.sh @@ -117,7 +117,10 @@ function prepare_policy_repo { # Wait for Gitea to be ready and initialized echo " Waiting for Gitea to be ready..." - timeout=120 + # Generous budget: gitea can take minutes to bring the web listener up on slow + # bind-mount I/O (e.g. Docker Desktop) — the loop exits as soon as it is ready, + # so fast environments (CI) pay nothing for the headroom. + timeout=300 counter=0 while ! curl -sf http://localhost:3000 > /dev/null 2>&1; do counter=$((counter + 1)) @@ -252,6 +255,15 @@ function check_servers_not_logged { fi } +function check_clients_not_logged { + echo "- Ensuring msg '$1' is absent from client's logs" + if compose logs opal_client | grep -q "$1"; then + echo "- Unexpectedly found '$1' in client logs:" + compose logs opal_client | grep "$1" + exit 1 + fi +} + function wait_for_broadcaster { echo "- Waiting for broadcast_channel to accept connections" for _ in $(seq 1 30); do @@ -402,24 +414,39 @@ function main { check_servers_not_logged "list.remove(x): x not in list" # Cross-instance consistency: publish an update WHILE the backbone is down, then - # recover. The two clients connect to different server replicas via the service VIP, - # so for BOTH to end up with the value the missed cross-server update must converge - # after recovery (via the replay buffer and/or the resync-on-reconnect path). + # recover. The two clients connect to different server replicas via the service VIP. + # With BROADCAST_FREEZE_ON_DISCONNECT (the default), a client-facing publish that + # cannot fan out to the whole fleet is FROZEN — applied by NO client — so the fleet + # never splits (one replica's clients seeing the update while the other's don't). + # One-off updates like this one are not part of the clients' configured data + # sources, so the freeze DROPS them (documented trade); freshness is restored by + # re-publishing after recovery, and the fleet converges together. echo "- Testing cross-instance consistency across a backbone outage" compose kill broadcast_channel sleep 3 publish_data "consistency_user" sleep 2 + # The receiving server must have frozen the publish at the gate... + check_servers_logged "freezing publish to preserve fleet consistency" compose up -d broadcast_channel wait_for_broadcaster - # allow buffered replay + (if needed) client resync + full refetch to settle + # allow recovery: exempt-topic replay + client resync + full refetch to settle sleep 15 - # The server that received the publish while the backbone was down must have - # buffered it and replayed it on recovery (proves the replay path actually ran, - # not just a client refetch). + # Internal (exempt) topics still ride the pre-freeze buffer+replay path during the + # gap — these lines prove that path stayed intact alongside the freeze. check_servers_logged "buffered for replay" check_servers_logged "Replaying" - # BOTH clients (on different replicas via the VIP) must end up with the value. + # Recovery resynced this worker's clients (the freeze's convergence path). + check_servers_logged "resyncing this worker's clients" + # THE consistency assertion: the frozen update reached NO client — neither during + # the gap nor via replay after it. No fleet split. + check_clients_not_logged "PUT /v1/data/users/consistency_user/location -> 204" + # After recovery the fleet is fully functional: re-publish and BOTH clients + # (on different replicas via the VIP) converge on the value together. + publish_data "consistency_user" + sleep 5 + # The freezing server's first post-gap delivery logs the freeze-episode summary. + check_servers_logged "publish(es) during the gap" check_clients_logged "PUT /v1/data/users/consistency_user/location -> 204" # TODO: Test statistics feature again after broadcaster restart (should first fix statistics bug) } diff --git a/packages/opal-server/opal_server/config.py b/packages/opal-server/opal_server/config.py index 58341ff37..fdd3ea534 100644 --- a/packages/opal-server/opal_server/config.py +++ b/packages/opal-server/opal_server/config.py @@ -103,6 +103,22 @@ class OpalServerConfig(Confi): "reader is wedged while clients depend on it. Set to False to revert " "/healthcheck to always returning ok.", ) + BROADCAST_FREEZE_ON_DISCONNECT = confi.bool( + "BROADCAST_FREEZE_ON_DISCONNECT", + True, + description="During a broadcaster backbone gap, freeze client-facing publishes on " + "every worker instead of applying them locally — so a write that cannot fan out to " + "the whole fleet is never served by just one worker (fleet consistency over " + "freshness). Recovery is the reconnect resync: clients re-fetch their configured " + "data sources and policy, so updates covered by those are reconciled; one-off " + "updates outside them (inline data payloads, ad-hoc fetch URLs) are DROPPED by a " + "freeze, not deferred. Internal coordination topics (statistics, keepalive, the git " + "webhook trigger) are exempt and keep the deliver-locally + buffer-for-replay " + "behavior. Requires BROADCAST_RECONNECT_ENABLED and BROADCAST_RESYNC_ON_RECONNECT " + "(if the resync is disabled, freezing is refused with a warning). Set to False for " + "the previous behavior, where the receiving worker's own clients update immediately " + "and peers only after reconnect.", + ) # server security AUTH_PRIVATE_KEY_FORMAT = confi.enum( diff --git a/packages/opal-server/opal_server/pubsub.py b/packages/opal-server/opal_server/pubsub.py index 8f8a0a0d9..9d57a8ccd 100644 --- a/packages/opal-server/opal_server/pubsub.py +++ b/packages/opal-server/opal_server/pubsub.py @@ -31,7 +31,11 @@ from opal_common.config import opal_common_config from opal_common.logger import logger from opal_server.config import opal_server_config -from opal_server.pubsub_resilience import ReconnectingBroadcaster, SafeConnectionManager +from opal_server.pubsub_resilience import ( + FreezablePubSubEndpoint, + ReconnectingBroadcaster, + SafeConnectionManager, +) from pydantic import BaseModel from starlette.datastructures import QueryParams @@ -181,13 +185,36 @@ def __init__(self, signer: JWTSigner, broadcaster_uri: str = None): # we keep the library-safe default (True) to degrade to "stale but connected" # rather than the fleet-wide drop storm. (Replaces an earlier experimental # broadcast-connection-loss flag.) - self.endpoint = PubSubEndpoint( + # The reconnect resync is the freeze's ONLY recovery path (frozen publishes are + # dropped, not buffered) — with the resync disabled the combination would silently + # lose every update published during every gap, so refuse it rather than honor it. + freeze_on_disconnect = opal_server_config.BROADCAST_FREEZE_ON_DISCONNECT + if ( + freeze_on_disconnect + and not opal_server_config.BROADCAST_RESYNC_ON_RECONNECT + ): + logger.warning( + "BROADCAST_FREEZE_ON_DISCONNECT is enabled but BROADCAST_RESYNC_ON_RECONNECT " + "is disabled — the resync is the freeze's only recovery path, so freezing is " + "DISABLED to avoid silently losing updates published during a backbone gap. " + "Re-enable BROADCAST_RESYNC_ON_RECONNECT to get the fleet-consistency freeze." + ) + freeze_on_disconnect = False + self.endpoint = FreezablePubSubEndpoint( broadcaster=self.broadcaster, notifier=self.notifier, rpc_channel_get_remote_id=opal_common_config.STATISTICS_ENABLED, ignore_broadcaster_disconnected=not isinstance( self.broadcaster, ReconnectingBroadcaster ), + # Freeze client-facing publishes during a backbone gap so a write that cannot + # reach the whole fleet is not applied on a single worker (see + # FreezablePubSubEndpoint). No-op unless the reconnecting broadcaster is in use. + freeze_on_disconnect=freeze_on_disconnect, + # The git-webhook trigger targets the server-side policy watcher, not clients — + # freezing it would drop repo-pull triggers with nothing to replay them (topics + # prefixed "__", i.e. statistics/keepalive, are exempted by the endpoint itself). + freeze_exempt_topics=[opal_server_config.POLICY_REPO_WEBHOOK_TOPIC], ) # fastapi_websocket_rpc's ConnectionManager.disconnect is not idempotent: the RPC # endpoint can call it twice for one socket (handle_disconnect plus the outer diff --git a/packages/opal-server/opal_server/pubsub_resilience.py b/packages/opal-server/opal_server/pubsub_resilience.py index 2c5095be5..d41201f51 100644 --- a/packages/opal-server/opal_server/pubsub_resilience.py +++ b/packages/opal-server/opal_server/pubsub_resilience.py @@ -42,7 +42,7 @@ from typing import Awaitable, Callable, Optional from fastapi import WebSocket -from fastapi_websocket_pubsub import EventBroadcaster +from fastapi_websocket_pubsub import EventBroadcaster, PubSubEndpoint from fastapi_websocket_pubsub.event_broadcaster import BroadcastNotification from fastapi_websocket_pubsub.event_notifier import Subscription from fastapi_websocket_pubsub.util import pydantic_serialize @@ -154,6 +154,15 @@ def __init__( # Fired once if the reader gives up (exhausts reconnect retries) and returns, # so OPAL can graceful-restart the worker even with statistics disabled. self._on_give_up: Optional[ReconnectCallback] = None + # Live backbone-subscription state; see is_backbone_connected() / is_in_backbone_gap(). + # Instance attrs (not task-local): FreezablePubSubEndpoint reads them from outside + # the reader task. + self._backbone_connected = False + # Whether this broadcaster ever held a backbone subscription — distinguishes a real + # GAP (had a session, lost it) from "never connected yet" (boot, or backbone down + # from the start), where freezing would be wrong: no resync fires on a FIRST + # connect, so anything frozen before it would be lost, not deferred. + self._had_backbone_connection = False def set_reconnect_callback(self, callback: Optional[ReconnectCallback]): """Register an ``async () -> None`` callback fired once after each gap @@ -183,6 +192,13 @@ async def start_reader_task(self): logger.debug("No need for listen task, already started") return self._subscription_task logger.debug("Spawning reconnecting broadcast listen task") + # Scope gap detection to THIS reader task: a stale True from a previous task + # (reader cancelled when the last listener left, then restarted later) would make + # is_in_backbone_gap() freeze publishes before the new task's FIRST subscribe — + # but a first connect fires no resync, so those publishes would be lost, not + # deferred. Same task-scoping rationale as ``had_prior_connection`` in + # ``__read_notifications__``. + self._had_backbone_connection = False self._subscription_task = asyncio.create_task(self.__read_notifications__()) return self._subscription_task @@ -221,6 +237,45 @@ def is_reader_healthy(self) -> bool: self._subscription_task is not None and not self._subscription_task.done() ) + def is_backbone_connected(self) -> bool: + """Whether the reader currently holds a live backbone subscription. + + True only while actively subscribed — i.e. a publish right now would actually + reach peer workers. Flips False the instant the subscription drops and back to + True only once re-subscribed. + + This is intentionally NOT ``is_reader_healthy()``: that one stays True across a + transient reconnect (so the k8s probe does not flap the pod), which is exactly the + wrong signal for gating delivery. + """ + return self._backbone_connected + + def is_in_backbone_gap(self) -> bool: + """Whether the broadcaster is mid-GAP: it *had* a live backbone subscription, + lost it, and the reader is still trying to get it back. + + This — not mere "not connected" — is the publish-freeze condition used by + ``FreezablePubSubEndpoint``, because only a real gap has the recovery path the + freeze relies on (the ``on_reconnect`` resync fires exclusively for reconnects + that follow an established session). The two excluded states must NOT freeze: + + * **Reader not running** (no listeners yet / worker idle / last client left and + the upstream cancelled the reader): the backbone may be perfectly healthy — + freezing here would silently drop publishes fleet-wide with nothing to ever + reconcile them. Delegating preserves the pre-freeze behavior (share-context + broadcast + local delivery). + * **Never connected in this reader's lifetime** (boot, or backbone already down + at startup): a FIRST successful connect fires no gap recovery, so a publish + frozen in this window would be lost, not deferred. Pre-freeze behavior + (deliver locally, buffer outbound for replay) is strictly better here. + """ + return ( + self._subscription_task is not None + and not self._subscription_task.done() + and self._had_backbone_connection + and not self._backbone_connected + ) + async def __broadcast_notifications__(self, subscription: Subscription, data): """Share a local notification with the backbone; buffer it on failure. @@ -277,6 +332,10 @@ async def __read_notifications__(self): f"Broadcaster listener connected to channel '{self._channel}'" ) async with channel.subscribe(channel=self._channel) as subscriber: + # Subscribed: the backbone is reachable, so publishes will fan out to + # peers again — reopen the publish gate (see FreezablePubSubEndpoint). + self._backbone_connected = True + self._had_backbone_connection = True # We are subscribed again; recover concurrently so we keep reading # (and can receive peers' replays) during the settle window. if had_prior_connection: @@ -317,6 +376,10 @@ async def __read_notifications__(self): await self._fire_give_up() return finally: + # Any exit from the read cycle (backbone closed, error, or cancel) means we + # are no longer subscribed — close the publish gate until we re-subscribe, so + # a write during the gap is not applied on this worker alone. + self._backbone_connected = False await self._safe_disconnect_channel() await asyncio.sleep(self._backoff_seconds(attempt)) @@ -532,3 +595,119 @@ def _backoff_seconds(self, attempt: int) -> float: base = min(base, self._reconnect_backoff_max) # Equal jitter, so a fleet of pods does not reconnect to the backbone in lockstep. return base / 2 + random.uniform(0, base / 2) + + +class FreezablePubSubEndpoint(PubSubEndpoint): + """A ``PubSubEndpoint`` that *freezes* client-facing publishes during a + broadcaster backbone GAP, to keep a multi-worker fleet consistent through + an outage. + + The problem: a server-side ``publish`` fans out two independent ways — local in-process + delivery to *this* worker's own clients, and (via the broadcaster) to peer workers. Only + the outbound path is buffered when the backbone is down; local delivery still fires. So a + data/policy update that reaches one worker during a backbone gap is applied to that + worker's clients but not the fleet — a transient split (some PDPs new, others old) that + lasts the whole outage. + + With freeze enabled, while the ``ReconnectingBroadcaster`` reports a real gap + (``is_in_backbone_gap()`` — an established backbone session was lost and is being + re-acquired; see its docstring for why "never connected" and "reader not running" must + NOT freeze), ``publish`` is skipped entirely: neither local clients nor the outbound + buffer see it. The write still lands in the source of truth, and the reconnect *resync* + makes every worker's clients refetch on recovery — the whole fleet moves together. + Skipping the whole publish also means nothing is buffered for replay during the freeze, + so recovery converges purely via the resync refetch. + + **Exempt topics** keep the pre-freeze behavior (local delivery + outbound replay buffer) + even mid-gap: topics prefixed ``__`` (the statistics protocol and the broadcaster + keepalive — dropping those corrupts server-to-server state that no resync rebuilds: + ghost clients, workers that never stat-sync) and any topic in ``freeze_exempt_topics`` + (OPAL passes the git-webhook trigger topic: it targets the server-side policy watcher, + not clients, and a dropped trigger means the repo pull it requests simply never happens + — the resync would then refetch from a clone that was never advanced). + + Delegates straight to the base when: freeze is disabled; there is no broadcaster + (single worker); or the broadcaster is the stock ``EventBroadcaster``. + + **Recovery scope** (what "reconciled by the resync" actually covers): data the clients + re-fetch on reconnect, i.e. their configured data sources (``OPAL_DATA_CONFIG_SOURCES`` + or scope config) and the policy bundle. One-off updates outside that set — an inline + ``data`` payload, or a fetch URL that is not part of the configured sources — are + dropped by a freeze, not deferred. Accepted trade: consistency over freshness, and such + updates are the legacy path. + + **Known limitations** (all degrade to the PRE-freeze behavior, never worse): + * If the reader's subscription is alive but an individual outbound broadcast fails + (separate per-publish channel), the gate does not engage — that publish is delivered + locally and buffered for replay, the pre-existing split-until-replay behavior. + * The gate reopens when the subscription is re-established, before the session proves + "sustained" — during a rare connect-then-instant-close flap a publish can slip + through (deliver locally + buffer). Gating on sustained instead would wrongly freeze + quiet channels forever (a session only proves sustained on its first inbound event). + * Client-originated RPC publishes (``RpcEventServerMethods.publish``) notify the local + subscribers directly, bypassing this override. + """ + + def __init__( + self, + *args, + freeze_on_disconnect: bool = True, + freeze_exempt_topics=(), + **kwargs, + ): + super().__init__(*args, **kwargs) + self._freeze_on_disconnect = freeze_on_disconnect + self._freeze_exempt_topics = frozenset(freeze_exempt_topics) + # Publishes suppressed in the current freeze episode — first one logs at WARNING, + # the rest at DEBUG (a long outage would otherwise emit an unbounded WARNING per + # frozen stats keepalive), and the first delivered publish afterwards logs a + # summary count. + self._frozen_in_episode = 0 + + def _is_exempt(self, topics) -> bool: + if isinstance(topics, str): + topics = [topics] + return all( + topic.startswith("__") or topic in self._freeze_exempt_topics + for topic in topics + ) + + def _should_freeze(self, topics) -> bool: + return self._in_frozen_gap() and not self._is_exempt(topics) + + def _in_frozen_gap(self) -> bool: + broadcaster = self.broadcaster + return ( + self._freeze_on_disconnect + and isinstance(broadcaster, ReconnectingBroadcaster) + and broadcaster.is_in_backbone_gap() + ) + + async def publish(self, topics, data=None): + in_gap = self._in_frozen_gap() + if in_gap and not self._is_exempt(topics): + self._frozen_in_episode += 1 + log = logger.warning if self._frozen_in_episode == 1 else logger.debug + log( + "Broadcaster backbone gap; freezing publish to preserve fleet consistency " + "(not delivered to clients; reconciled via resync on reconnect). " + "topics={topics} (suppressed {count} so far this gap)", + topics=topics, + count=self._frozen_in_episode, + ) + return + # Emit the episode summary only once the gap is actually over — an EXEMPT publish + # mid-gap also reaches this point and must not reset the counter or claim recovery. + if self._frozen_in_episode and not in_gap: + count, self._frozen_in_episode = self._frozen_in_episode, 0 + logger.warning( + "Backbone recovered; froze {count} publish(es) during the gap — clients " + "reconcile via the reconnect resync", + count=count, + ) + return await super().publish(topics, data) + + # The library aliases ``notify = publish`` at class level (backward-compat canonical + # name), which binds the BASE publish — re-bind it here or ``endpoint.notify(...)`` + # would silently bypass the freeze gate. + notify = publish diff --git a/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py b/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py index c90d771cd..0c2fbe544 100644 --- a/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py +++ b/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py @@ -13,7 +13,10 @@ import pytest from fastapi_websocket_pubsub import EventBroadcaster from fastapi_websocket_pubsub.event_broadcaster import BroadcastNotification -from opal_server.pubsub_resilience import ReconnectingBroadcaster +from opal_server.pubsub_resilience import ( + FreezablePubSubEndpoint, + ReconnectingBroadcaster, +) _END = object() @@ -42,6 +45,15 @@ def __init__(self): async def notify(self, topics, data, notifier_id=None): self.notified.append((list(topics), data, notifier_id)) + def gen_subscriber_id(self): + # PubSubEndpoint.__init__ mints a server id from the notifier. + return "fake-subscriber-id" + + async def subscribe(self, *args, **kwargs): + # The broadcaster's sharing context registers itself on the notifier + # (EventBroadcaster._subscribe_to_all_topics); a no-op suffices here. + pass + class FakeBus: """A controllable in-memory broadcast backbone for a single channel.""" @@ -831,3 +843,226 @@ async def test_partial_replay_requeue_preserves_drop_oldest(): "refill-2", ] # The OLDEST (unsent-1) was dropped from the front, NOT the newest refill. + + +# --------------------------------------------------------------------------- +# Fleet-consistency freeze (is_backbone_connected + FreezablePubSubEndpoint) +# --------------------------------------------------------------------------- + + +def _reconnecting(bus, notifier=None): + return ReconnectingBroadcaster( + "memory://", + notifier=notifier or FakeNotifier(), + channel="test", + broadcast_type=bus.channel_factory, + reconnect_backoff_min=0, + reconnect_backoff_max=0, + ) + + +@pytest.mark.asyncio +async def test_is_backbone_connected_tracks_subscription(): + """The flag is False before the reader subscribes, True while subscribed, + and False again once the reader stops.""" + bus = FakeBus() + broadcaster = _reconnecting(bus) + assert not broadcaster.is_backbone_connected() # not started -> disconnected + task = await broadcaster.start_reader_task() + try: + await _wait_for(broadcaster.is_backbone_connected) # subscribed -> connected + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert not broadcaster.is_backbone_connected() # reader gone -> disconnected + + +@pytest.mark.asyncio +async def test_is_in_backbone_gap_lifecycle(): + """A GAP is: had a live subscription, lost it, reader still retrying. Neither + 'never started' nor 'never yet connected' nor 'reconnected' is a gap.""" + bus = FakeBus() + broadcaster = _reconnecting(bus) + assert not broadcaster.is_in_backbone_gap() # never started -> not a gap + task = await broadcaster.start_reader_task() + try: + await _wait_for(broadcaster.is_backbone_connected) + assert not broadcaster.is_in_backbone_gap() # subscribed -> not a gap + + # Backbone drops and stays unavailable: block re-subscribes, then close the read. + bus.fail_subscribe_times = 10_000 + await bus.drop() + await _wait_for(broadcaster.is_in_backbone_gap) # GAP: had one, lost it + + # Backbone returns: allow re-subscribes -> gap ends. + bus.fail_subscribe_times = 0 + await _wait_for(lambda: not broadcaster.is_in_backbone_gap()) + assert broadcaster.is_backbone_connected() + finally: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +@pytest.mark.asyncio +async def test_reader_restart_does_not_inherit_gap_state(): + """A RESTARTED reader (previous one cancelled when the last listener left) must + start with a clean slate: its pre-first-subscribe window is 'never connected', + not a gap — a first connect fires no resync, so freezing there would LOSE + publishes rather than defer them.""" + bus = FakeBus() + broadcaster = _reconnecting(bus) + task = await broadcaster.start_reader_task() + await _wait_for(broadcaster.is_backbone_connected) # session established + # Last listener leaves: upstream cancels the reader and clears the task slot. + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + broadcaster._subscription_task = None + # Backbone is down when a new listener restarts the reader. + bus.fail_subscribe_times = 10_000 + task2 = await broadcaster.start_reader_task() + try: + await _wait_for(lambda: bus.subscribes >= 2) # new reader is retrying + assert not broadcaster.is_in_backbone_gap() # clean slate: NOT a gap + finally: + task2.cancel() + with pytest.raises(asyncio.CancelledError): + await task2 + + +def _endpoint(broadcaster, notifier, freeze=True, exempt=()): + return FreezablePubSubEndpoint( + broadcaster=broadcaster, + notifier=notifier, + freeze_on_disconnect=freeze, + freeze_exempt_topics=exempt, + ) + + +def _fabricate_gap(broadcaster): + """Put a broadcaster into the mid-gap state (had a session, lost it, reader + pending) without driving a real backbone. + + Returns the dummy reader task — cancel it in the test's cleanup. + """ + dummy = asyncio.get_event_loop().create_task(asyncio.sleep(60)) + broadcaster._subscription_task = dummy + broadcaster._had_backbone_connection = True + broadcaster._backbone_connected = False + return dummy + + +@pytest.mark.asyncio +async def test_should_freeze_matrix(): + """Freeze only when: enabled AND ReconnectingBroadcaster AND mid-GAP AND a + non-exempt topic. Everything else delegates normally.""" + bus = FakeBus() + topics = ["policy_data"] + + # never-started reader -> NOT a gap -> no freeze (healthy idle worker must not drop) + b = _reconnecting(bus) + assert _endpoint(b, FakeNotifier())._should_freeze(topics) is False + # reader running but never yet connected (boot / backbone down from the start) -> no + # freeze: no resync fires on a FIRST connect, so a frozen publish would be lost. + dummy = _fabricate_gap(b) + b._had_backbone_connection = False + try: + assert _endpoint(b, FakeNotifier())._should_freeze(topics) is False + # a real gap (had a session, lost it) -> FREEZE + b._had_backbone_connection = True + endpoint = _endpoint(b, FakeNotifier()) + assert endpoint._should_freeze(topics) is True + # exempt topics keep flowing mid-gap: "__" internals and the explicit exempt set + assert endpoint._should_freeze(["__opal_stats_server_keepalive"]) is False + assert ( + _endpoint(b, FakeNotifier(), exempt=["webhook"])._should_freeze(["webhook"]) + is False + ) + # a mixed list containing any non-exempt topic still freezes (consistency wins) + assert endpoint._should_freeze(["__opal_stats_wakeup", "policy_data"]) is True + # freeze disabled -> never + assert ( + _endpoint(b, FakeNotifier(), freeze=False)._should_freeze(topics) is False + ) + # reconnected -> never + b._backbone_connected = True + assert _endpoint(b, FakeNotifier())._should_freeze(topics) is False + b._backbone_connected = False + finally: + dummy.cancel() + # no broadcaster (single worker) -> never + assert _endpoint(None, FakeNotifier())._should_freeze(topics) is False + # stock (non-reconnecting) broadcaster -> never (legacy drop-on-disconnect path) + stock = EventBroadcaster( + "memory://", + notifier=FakeNotifier(), + channel="test", + broadcast_type=bus.channel_factory, + ) + assert _endpoint(stock, FakeNotifier())._should_freeze(topics) is False + + +@pytest.mark.asyncio +async def test_publish_is_suppressed_during_gap(): + """Mid-gap, publish must not deliver to clients at all (notifier + untouched); after the gap the next publish delivers and resets the episode + counter.""" + notifier = FakeNotifier() + b = _reconnecting(FakeBus()) + dummy = _fabricate_gap(b) + endpoint = _endpoint(b, notifier) + try: + await endpoint.publish(["policy_data"], {"x": 1}) + await endpoint.publish(["policy_data"], {"x": 2}) + assert notifier.notified == [] # frozen: nothing reached the clients + assert endpoint._frozen_in_episode == 2 + # gap ends -> publish flows again and the episode counter resets + b._backbone_connected = True + await endpoint.publish(["policy_data"], {"x": 3}) + assert [d for _, d, _ in notifier.notified] == [{"x": 3}] + assert endpoint._frozen_in_episode == 0 + finally: + dummy.cancel() + + +@pytest.mark.asyncio +async def test_exempt_publish_mid_gap_delivers_without_ending_episode(): + """An exempt (internal/webhook) publish during the gap is delivered but + must NOT reset the freeze-episode counter or log the 'recovered' summary — + the gap is still open.""" + notifier = FakeNotifier() + b = _reconnecting(FakeBus()) + dummy = _fabricate_gap(b) + endpoint = _endpoint(b, notifier, exempt=["webhook"]) + try: + await endpoint.publish(["policy_data"], {"x": 1}) # frozen + assert endpoint._frozen_in_episode == 1 + await endpoint.publish(["webhook"], {"w": 1}) # exempt -> delivered + await endpoint.publish(["__opal_stats_wakeup"], {"s": 1}) # exempt + assert [t for t, _, _ in notifier.notified] == [ + ["webhook"], + ["__opal_stats_wakeup"], + ] + # episode still open: exempt deliveries mid-gap must not end it + assert endpoint._frozen_in_episode == 1 + finally: + dummy.cancel() + + +@pytest.mark.asyncio +async def test_publish_delivers_when_not_freezing(): + """With no broadcaster (nothing to freeze), publish delegates and + delivers.""" + notifier = FakeNotifier() + endpoint = _endpoint(None, notifier) + await endpoint.publish(["policy_data"], {"x": 1}) + assert notifier.notified and notifier.notified[0][0] == ["policy_data"] + + +def test_notify_alias_is_frozen(): + """The library aliases ``notify = publish`` at class level (binding the + BASE publish); the subclass must re-bind it or ``endpoint.notify(...)`` + bypasses the freeze gate.""" + assert FreezablePubSubEndpoint.notify is FreezablePubSubEndpoint.publish