From 660c687359cd04d826df9c71a1121dbaabb9d6cc Mon Sep 17 00:00:00 2001 From: zivxx Date: Thu, 2 Jul 2026 17:19:00 +0300 Subject: [PATCH 1/5] fix(opal-server): freeze client publishes while broadcaster backbone is down MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit During a broadcaster-backbone outage, a publish reaching one worker was still delivered to that worker's own clients (local in-process notifier) while peers' clients stayed stale — a transient fleet split lasting the whole outage (observed 3/9 PDPs diverging in a 3-server/9-PDP staging test). Gate the whole publish on live backbone connectivity: add is_backbone_connected() to ReconnectingBroadcaster (True only while actively subscribed — unlike is_reader_healthy(), which deliberately stays healthy across a transient reconnect) and a FreezablePubSubEndpoint that skips publish entirely while disconnected. Clients stay connected but frozen on the last consistent state; the write still lands in its datasource and the existing reconnect resync makes every client refetch it, so the fleet converges together. Skipping the whole publish also keeps the replay buffer out of the recovery path (no partial replayed state racing the resync refetch). Configurable via OPAL_BROADCAST_FREEZE_ON_DISCONNECT (default true). Scope: designed for source-based updates (entry carries a URL clients refetch). Inline updates issued during an outage are dropped rather than deferred — accepted, inline is legacy. Validated on a 3-server/9-PDP staging stack against a real RDS stop/start: during-outage divergence 3/9 -> 0/9, zero client disconnects and zero worker restarts across ~20-minute outages, and a source-based write made during the gap converged 9/9 via the resync refetch alone (its notification provably frozen, nothing replayed). Co-Authored-By: Claude Fable 5 --- packages/opal-server/opal_server/config.py | 11 +++ packages/opal-server/opal_server/pubsub.py | 12 ++- .../opal_server/pubsub_resilience.py | 90 ++++++++++++++++- .../tests/reconnecting_broadcaster_test.py | 96 ++++++++++++++++++- 4 files changed, 205 insertions(+), 4 deletions(-) diff --git a/packages/opal-server/opal_server/config.py b/packages/opal-server/opal_server/config.py index 58341ff37..1ebdbe370 100644 --- a/packages/opal-server/opal_server/config.py +++ b/packages/opal-server/opal_server/config.py @@ -103,6 +103,17 @@ 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="While the broadcaster backbone is disconnected, 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). The write still lands in its datasource and is reconciled by the " + "reconnect resync. Applies to source-based updates and requires " + "BROADCAST_RECONNECT_ENABLED. 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..df878516e 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,17 @@ 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( + 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 while the backbone is down 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=opal_server_config.BROADCAST_FREEZE_ON_DISCONNECT, ) # 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..29912b2bf 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 — True only while actively subscribed, so a + # publish right now would actually reach peer workers. Deliberately distinct from + # is_reader_healthy() (which stays True across a transient reconnect so the k8s probe + # doesn't flap): this flips False the instant the subscription drops. Read by + # FreezablePubSubEndpoint to freeze client-facing publishes during a gap, so a write + # that can't fan out to the fleet is never applied on just one worker (fleet + # consistency). Instance attr (not task-local): the endpoint reads it from outside + # the reader task. + self._backbone_connected = False def set_reconnect_callback(self, callback: Optional[ReconnectCallback]): """Register an ``async () -> None`` callback fired once after each gap @@ -221,6 +230,22 @@ 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. ``FreezablePubSubEndpoint`` reads this to freeze + client-facing publishes while the backbone is down, so a write that cannot fan out + to the whole fleet is never applied on a single worker (fleet consistency; the + write still lands in the source of truth and is picked up by the reconnect resync). + """ + return self._backbone_connected + async def __broadcast_notifications__(self, subscription: Subscription, data): """Share a local notification with the backbone; buffer it on failure. @@ -277,6 +302,9 @@ 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 # 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 +345,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 +564,59 @@ 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 while the broadcaster + backbone is disconnected, to keep a multi-worker fleet consistent during 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, when the broadcaster is a ``ReconnectingBroadcaster`` that is + currently disconnected (``is_backbone_connected()`` is False), ``publish`` is skipped + entirely: neither local clients nor the outbound buffer see it. The write still lands in + the source of truth (the datasource the update points at), and the reconnect *resync* + makes every worker refetch it on recovery — so the whole fleet moves together instead of + diverging mid-outage. + + Skipping the *whole* publish (not just local delivery) also means nothing is buffered for + replay during the freeze, so recovery converges purely via the resync refetch with no + partial replayed state racing ahead of it. + + Delegates straight to the base (no freeze) when: freeze is disabled; there is no + broadcaster (single worker — no fleet to keep consistent); or the broadcaster is the + stock ``EventBroadcaster`` (reconnect disabled — the legacy drop-on-disconnect path). + + Scope: designed for *source-based* updates (the entry carries a URL the client refetches, + which is how permit's data flows). An *inline* update (data embedded in the broadcast) has + no source to refetch, so an inline update issued during a freeze is dropped rather than + deferred — accepted, since inline is legacy/unused. + """ + + def __init__(self, *args, freeze_on_disconnect: bool = True, **kwargs): + super().__init__(*args, **kwargs) + self._freeze_on_disconnect = freeze_on_disconnect + + def _should_freeze(self) -> bool: + broadcaster = self.broadcaster + return ( + self._freeze_on_disconnect + and isinstance(broadcaster, ReconnectingBroadcaster) + and not broadcaster.is_backbone_connected() + ) + + async def publish(self, topics, data=None): + if self._should_freeze(): + logger.warning( + "Broadcaster backbone disconnected; freezing publish to preserve fleet " + "consistency (not delivered to clients; reconciled via resync on reconnect). " + "topics={topics}", + topics=topics, + ) + return + return await super().publish(topics, data) 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..ea40c0924 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,10 @@ 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" + class FakeBus: """A controllable in-memory broadcast backbone for a single channel.""" @@ -831,3 +838,90 @@ 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 — the exact signal the publish gate needs.""" + bus = FakeBus() + broadcaster = _reconnecting(bus) + assert not broadcaster.is_backbone_connected() # not started -> disconnected + task = await broadcaster.start_reader_task() + try: + await _wait_for(lambda: bus.subscribes >= 1) + await _wait_for(broadcaster.is_backbone_connected) + assert 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 + + +def _endpoint(broadcaster, notifier, freeze=True): + return FreezablePubSubEndpoint( + broadcaster=broadcaster, notifier=notifier, freeze_on_disconnect=freeze + ) + + +@pytest.mark.asyncio +async def test_should_freeze_matrix(): + """The gate freezes only when: freeze enabled AND a ReconnectingBroadcaster AND the + backbone is currently disconnected. Every other combination delegates normally.""" + bus = FakeBus() + b = _reconnecting(bus) # not started -> is_backbone_connected() is False + + # disconnected reconnecting broadcaster + freeze on -> FREEZE + assert _endpoint(b, FakeNotifier(), freeze=True)._should_freeze() is True + # freeze disabled -> never + assert _endpoint(b, FakeNotifier(), freeze=False)._should_freeze() is False + # connected -> never + b._backbone_connected = True + assert _endpoint(b, FakeNotifier(), freeze=True)._should_freeze() is False + b._backbone_connected = False + # no broadcaster (single worker) -> never + assert _endpoint(None, FakeNotifier(), freeze=True)._should_freeze() 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(), freeze=True)._should_freeze() is False + + +@pytest.mark.asyncio +async def test_publish_is_suppressed_while_backbone_down(): + """When frozen, publish must not deliver to clients at all (notifier untouched).""" + notifier = FakeNotifier() + b = _reconnecting(bus := FakeBus()) # disconnected + endpoint = _endpoint(b, notifier, freeze=True) + await endpoint.publish(["policy_data"], {"x": 1}) + assert notifier.notified == [] # frozen: nothing reached the clients + + +@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, freeze=True) + await endpoint.publish(["policy_data"], {"x": 1}) + assert notifier.notified and notifier.notified[0][0] == ["policy_data"] From dfd53ada5a54610508a363fb5aee898ce38c1a15 Mon Sep 17 00:00:00 2001 From: zivxx Date: Thu, 2 Jul 2026 17:44:02 +0300 Subject: [PATCH 2/5] =?UTF-8?q?fix(opal-server):=20harden=20the=20publish?= =?UTF-8?q?=20freeze=20=E2=80=94=20gap-only=20gating,=20topic=20exemptions?= =?UTF-8?q?,=20alias=20bypass?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review findings on the freeze-on-disconnect gate, all fixed here: - Gate on a real backbone GAP (had a session, lost it, reader retrying) via is_in_backbone_gap(), not on mere "not subscribed". Freezing while the reader was never started (healthy idle worker: the global listening context is only pinned when statistics are enabled, and upstream cancels the reader when the last listener leaves) or never yet connected (boot) silently dropped publishes fleet-wide with no resync ever firing to reconcile them. - Exempt internal coordination topics from the freeze: "__"-prefixed channels (statistics protocol, broadcaster keepalive) and the git-webhook trigger topic. These target server-side subscribers, not clients — dropping them broke statistics state (ghost clients, workers that never sync) and lost repo-pull triggers that no resync re-issues (webhooks are the only pull trigger with the default POLICY_REPO_POLLING_INTERVAL=0). Exempt topics keep the pre-freeze deliver-locally + buffer-for-replay behavior. - Re-bind the library's class-level `notify = publish` alias, which bound the BASE publish and let endpoint.notify(...) bypass the gate entirely. - Refuse BROADCAST_FREEZE_ON_DISCONNECT with BROADCAST_RESYNC_ON_RECONNECT disabled (warn + disable freeze): the resync is the freeze's only recovery path, so that combination silently lost every update published during gaps. - Rate-limit the freeze log: WARNING once per gap episode, DEBUG afterwards, and a suppressed-count summary on recovery (a long outage previously emitted an unbounded WARNING per frozen stats keepalive). - Document recovery scope honestly (configured-source refetch; one-off URL / inline updates are dropped by a freeze) and the residual windows that keep pre-freeze behavior (outbound-failure while subscribed, connect flap, RPC client-originated publishes). Co-Authored-By: Claude Fable 5 --- packages/opal-server/opal_server/config.py | 19 +- packages/opal-server/opal_server/pubsub.py | 25 ++- .../opal_server/pubsub_resilience.py | 165 ++++++++++++----- .../tests/reconnecting_broadcaster_test.py | 167 +++++++++++++++--- 4 files changed, 299 insertions(+), 77 deletions(-) diff --git a/packages/opal-server/opal_server/config.py b/packages/opal-server/opal_server/config.py index 1ebdbe370..fdd3ea534 100644 --- a/packages/opal-server/opal_server/config.py +++ b/packages/opal-server/opal_server/config.py @@ -106,13 +106,18 @@ class OpalServerConfig(Confi): BROADCAST_FREEZE_ON_DISCONNECT = confi.bool( "BROADCAST_FREEZE_ON_DISCONNECT", True, - description="While the broadcaster backbone is disconnected, 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). The write still lands in its datasource and is reconciled by the " - "reconnect resync. Applies to source-based updates and requires " - "BROADCAST_RECONNECT_ENABLED. Set to False for the previous behavior, where the " - "receiving worker's own clients update immediately and peers only after reconnect.", + 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 diff --git a/packages/opal-server/opal_server/pubsub.py b/packages/opal-server/opal_server/pubsub.py index df878516e..9d57a8ccd 100644 --- a/packages/opal-server/opal_server/pubsub.py +++ b/packages/opal-server/opal_server/pubsub.py @@ -185,6 +185,21 @@ 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.) + # 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, @@ -192,10 +207,14 @@ def __init__(self, signer: JWTSigner, broadcaster_uri: str = None): ignore_broadcaster_disconnected=not isinstance( self.broadcaster, ReconnectingBroadcaster ), - # Freeze client-facing publishes while the backbone is down so a write that - # cannot reach the whole fleet is not applied on a single worker (see + # 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=opal_server_config.BROADCAST_FREEZE_ON_DISCONNECT, + 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 29912b2bf..ba9fd62e5 100644 --- a/packages/opal-server/opal_server/pubsub_resilience.py +++ b/packages/opal-server/opal_server/pubsub_resilience.py @@ -154,15 +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 — True only while actively subscribed, so a - # publish right now would actually reach peer workers. Deliberately distinct from - # is_reader_healthy() (which stays True across a transient reconnect so the k8s probe - # doesn't flap): this flips False the instant the subscription drops. Read by - # FreezablePubSubEndpoint to freeze client-facing publishes during a gap, so a write - # that can't fan out to the fleet is never applied on just one worker (fleet - # consistency). Instance attr (not task-local): the endpoint reads it from outside + # 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 @@ -192,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 @@ -239,13 +246,36 @@ def is_backbone_connected(self) -> bool: 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. ``FreezablePubSubEndpoint`` reads this to freeze - client-facing publishes while the backbone is down, so a write that cannot fan out - to the whole fleet is never applied on a single worker (fleet consistency; the - write still lands in the source of truth and is picked up by the reconnect resync). + 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. @@ -305,6 +335,7 @@ async def __read_notifications__(self): # 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: @@ -567,8 +598,8 @@ def _backoff_seconds(self, attempt: int) -> float: class FreezablePubSubEndpoint(PubSubEndpoint): - """A ``PubSubEndpoint`` that *freezes* client-facing publishes while the broadcaster - backbone is disconnected, to keep a multi-worker fleet consistent during an outage. + """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 @@ -577,46 +608,100 @@ class FreezablePubSubEndpoint(PubSubEndpoint): worker's clients but not the fleet — a transient split (some PDPs new, others old) that lasts the whole outage. - With freeze enabled, when the broadcaster is a ``ReconnectingBroadcaster`` that is - currently disconnected (``is_backbone_connected()`` is False), ``publish`` is skipped - entirely: neither local clients nor the outbound buffer see it. The write still lands in - the source of truth (the datasource the update points at), and the reconnect *resync* - makes every worker refetch it on recovery — so the whole fleet moves together instead of - diverging mid-outage. - - Skipping the *whole* publish (not just local delivery) also means nothing is buffered for - replay during the freeze, so recovery converges purely via the resync refetch with no - partial replayed state racing ahead of it. - - Delegates straight to the base (no freeze) when: freeze is disabled; there is no - broadcaster (single worker — no fleet to keep consistent); or the broadcaster is the - stock ``EventBroadcaster`` (reconnect disabled — the legacy drop-on-disconnect path). - - Scope: designed for *source-based* updates (the entry carries a URL the client refetches, - which is how permit's data flows). An *inline* update (data embedded in the broadcast) has - no source to refetch, so an inline update issued during a freeze is dropped rather than - deferred — accepted, since inline is legacy/unused. + 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, **kwargs): + 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) -> bool: + def _should_freeze(self, topics) -> bool: broadcaster = self.broadcaster return ( self._freeze_on_disconnect and isinstance(broadcaster, ReconnectingBroadcaster) - and not broadcaster.is_backbone_connected() + and broadcaster.is_in_backbone_gap() + and not self._is_exempt(topics) ) async def publish(self, topics, data=None): - if self._should_freeze(): - logger.warning( - "Broadcaster backbone disconnected; freezing publish to preserve fleet " - "consistency (not delivered to clients; reconciled via resync on reconnect). " - "topics={topics}", + if self._should_freeze(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 + if self._frozen_in_episode: + 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 ea40c0924..2c477a6d4 100644 --- a/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py +++ b/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py @@ -49,6 +49,11 @@ 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.""" @@ -859,15 +864,13 @@ def _reconnecting(bus, notifier=None): @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 — the exact signal the publish gate needs.""" + 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(lambda: bus.subscribes >= 1) - await _wait_for(broadcaster.is_backbone_connected) - assert broadcaster.is_backbone_connected() # subscribed -> connected + await _wait_for(broadcaster.is_backbone_connected) # subscribed -> connected finally: task.cancel() with pytest.raises(asyncio.CancelledError): @@ -875,29 +878,120 @@ async def test_is_backbone_connected_tracks_subscription(): assert not broadcaster.is_backbone_connected() # reader gone -> disconnected -def _endpoint(broadcaster, notifier, freeze=True): +@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 + 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(): - """The gate freezes only when: freeze enabled AND a ReconnectingBroadcaster AND the - backbone is currently disconnected. Every other combination delegates normally.""" + """Freeze only when: enabled AND ReconnectingBroadcaster AND mid-GAP AND a + non-exempt topic. Everything else delegates normally.""" bus = FakeBus() - b = _reconnecting(bus) # not started -> is_backbone_connected() is False - - # disconnected reconnecting broadcaster + freeze on -> FREEZE - assert _endpoint(b, FakeNotifier(), freeze=True)._should_freeze() is True - # freeze disabled -> never - assert _endpoint(b, FakeNotifier(), freeze=False)._should_freeze() is False - # connected -> never - b._backbone_connected = True - assert _endpoint(b, FakeNotifier(), freeze=True)._should_freeze() is False - b._backbone_connected = False + 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(), freeze=True)._should_freeze() is False + assert _endpoint(None, FakeNotifier())._should_freeze(topics) is False # stock (non-reconnecting) broadcaster -> never (legacy drop-on-disconnect path) stock = EventBroadcaster( "memory://", @@ -905,23 +999,42 @@ async def test_should_freeze_matrix(): channel="test", broadcast_type=bus.channel_factory, ) - assert _endpoint(stock, FakeNotifier(), freeze=True)._should_freeze() is False + assert _endpoint(stock, FakeNotifier())._should_freeze(topics) is False @pytest.mark.asyncio -async def test_publish_is_suppressed_while_backbone_down(): - """When frozen, publish must not deliver to clients at all (notifier untouched).""" +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(bus := FakeBus()) # disconnected - endpoint = _endpoint(b, notifier, freeze=True) - await endpoint.publish(["policy_data"], {"x": 1}) - assert notifier.notified == [] # frozen: nothing reached the clients + 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_publish_delivers_when_not_freezing(): """With no broadcaster (nothing to freeze), publish delegates and delivers.""" notifier = FakeNotifier() - endpoint = _endpoint(None, notifier, freeze=True) + 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 From 274d829f7a5785b69c7eadd2d1205e7971c9fb6d Mon Sep 17 00:00:00 2001 From: zivxx Date: Thu, 2 Jul 2026 18:41:35 +0300 Subject: [PATCH 3/5] style(opal-server): apply docformatter to freeze-gate docstrings CI fixes: - pre-commit / docformatter: new docstrings were not wrapped per the pinned docformatter 1.7.5; re-wrapped in place (no code changes), verified idempotent alongside black/isort and the test suite. Co-Authored-By: Claude Fable 5 --- .../opal_server/pubsub_resilience.py | 5 ++-- .../tests/reconnecting_broadcaster_test.py | 24 +++++++++++-------- 2 files changed, 17 insertions(+), 12 deletions(-) diff --git a/packages/opal-server/opal_server/pubsub_resilience.py b/packages/opal-server/opal_server/pubsub_resilience.py index ba9fd62e5..1b8bd0815 100644 --- a/packages/opal-server/opal_server/pubsub_resilience.py +++ b/packages/opal-server/opal_server/pubsub_resilience.py @@ -598,8 +598,9 @@ def _backoff_seconds(self, attempt: int) -> float: 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. + """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 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 2c477a6d4..4b48dc8a6 100644 --- a/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py +++ b/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py @@ -863,8 +863,8 @@ def _reconnecting(bus, notifier=None): @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.""" + """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 @@ -943,8 +943,10 @@ def _endpoint(broadcaster, notifier, freeze=True, 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.""" + 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 @@ -1004,8 +1006,9 @@ async def test_should_freeze_matrix(): @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.""" + """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) @@ -1026,7 +1029,8 @@ async def test_publish_is_suppressed_during_gap(): @pytest.mark.asyncio async def test_publish_delivers_when_not_freezing(): - """With no broadcaster (nothing to freeze), publish delegates and delivers.""" + """With no broadcaster (nothing to freeze), publish delegates and + delivers.""" notifier = FakeNotifier() endpoint = _endpoint(None, notifier) await endpoint.publish(["policy_data"], {"x": 1}) @@ -1034,7 +1038,7 @@ async def test_publish_delivers_when_not_freezing(): 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.""" + """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 From 4d8367e6bd68382a3ce595035ceb0b7e71fde563 Mon Sep 17 00:00:00 2001 From: zivxx Date: Sun, 5 Jul 2026 13:29:56 +0300 Subject: [PATCH 4/5] fix(opal-server): only log the freeze-episode summary once the gap is over An exempt (internal-topic) publish arriving mid-gap reaches the delivery path and was resetting the freeze-episode counter and logging "Backbone recovered" while the backbone was still down. Emit the summary only when the gap has actually ended; exempt deliveries mid-gap no longer end the episode. Co-Authored-By: Claude Fable 5 --- .../opal_server/pubsub_resilience.py | 11 ++++++--- .../tests/reconnecting_broadcaster_test.py | 24 +++++++++++++++++++ 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/packages/opal-server/opal_server/pubsub_resilience.py b/packages/opal-server/opal_server/pubsub_resilience.py index 1b8bd0815..d41201f51 100644 --- a/packages/opal-server/opal_server/pubsub_resilience.py +++ b/packages/opal-server/opal_server/pubsub_resilience.py @@ -673,16 +673,19 @@ def _is_exempt(self, topics) -> bool: ) 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() - and not self._is_exempt(topics) ) async def publish(self, topics, data=None): - if self._should_freeze(topics): + 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( @@ -693,7 +696,9 @@ async def publish(self, topics, data=None): count=self._frozen_in_episode, ) return - if self._frozen_in_episode: + # 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 " 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 4b48dc8a6..0c2fbe544 100644 --- a/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py +++ b/packages/opal-server/opal_server/tests/reconnecting_broadcaster_test.py @@ -1027,6 +1027,30 @@ async def test_publish_is_suppressed_during_gap(): 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 From 3934a45a2d9bf734ef0f4073a9e5d235f7491a11 Mon Sep 17 00:00:00 2001 From: zivxx Date: Sun, 5 Jul 2026 13:29:56 +0300 Subject: [PATCH 5/5] test(app-tests): align cross-instance consistency E2E with freeze semantics CI fixes: - E2E Tests / App Tests: the cross-instance consistency test asserted the pre-freeze behavior (a one-off update published during a backbone outage reaches clients via buffer+replay). With BROADCAST_FREEZE_ON_DISCONNECT (default on) that publish is frozen so the fleet never splits. The test now asserts the new semantics: the publish is frozen at the gate, exempt internal topics still ride buffer+replay, recovery resyncs clients, the frozen update reached NO client (new check_clients_not_logged helper), and a post-recovery re-publish converges both clients together and logs the freeze-episode summary. Also raise the gitea startup wait budget (120s -> 300s): slow bind-mount environments need minutes to bring the web listener up; fast environments exit the poll early and pay nothing. Co-Authored-By: Claude Fable 5 --- app-tests/run.sh | 45 ++++++++++++++++++++++++++++++++++++--------- 1 file changed, 36 insertions(+), 9 deletions(-) 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) }