fix(opal-server): reconnect broadcaster on backbone disconnect (PER-15065)

.pre-commit-config.yaml

@@ -1,3 +1,9 @@
+# The pinned formatter versions below (black 23.1.0, isort 5.12.0, docformatter
+# 1.7.x) do not run on Python 3.13+, so pin the hook toolchain to 3.12 (the version
+# CI uses). Without this, hooks fail on machines whose default python is newer.
+default_language_version:
+  python: python3.12
+
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v5.0.0
@@ -18,8 +24,11 @@ repos:
     hooks:
       - id: codespell
         args: [--skip, "*.json,*.lock"]
+  # v1.7.6 dropped the `docformatter-venv` hook whose `language: python_venv` made
+  # prek / pre-commit>=4 reject the whole manifest. Pinned to v1.7.6 (not v1.7.7,
+  # whose docstring wrapping differs) so the bump introduces no reformatting.
   - repo: https://github.com/PyCQA/docformatter
-    rev: v1.7.5
+    rev: v1.7.6
     hooks:
       - id: docformatter
         args: [--in-place]

app-tests/docker-compose-app-tests.yml

@@ -47,6 +47,11 @@ services:
       - POSTGRES_DB=postgres
       - POSTGRES_USER=postgres
       - POSTGRES_PASSWORD=postgres
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres"]
+      interval: 5s
+      timeout: 3s
+      retries: 10
 
   opal_server:
     image: permitio/opal-server:${OPAL_IMAGE_TAG:-latest}

app-tests/run.sh

@@ -238,6 +238,33 @@ function check_no_error {
   fi
 }
 
+function check_servers_logged {
+  echo "- Looking for msg '$1' in server's logs"
+  compose logs opal_server | grep -q "$1"
+}
+
+function check_servers_not_logged {
+  echo "- Ensuring msg '$1' is absent from server's logs"
+  if compose logs opal_server | grep -q "$1"; then
+    echo "- Unexpectedly found '$1' in server logs:"
+    compose logs opal_server | grep "$1"
+    exit 1
+  fi
+}
+
+function wait_for_broadcaster {
+  echo "- Waiting for broadcast_channel to accept connections"
+  for _ in $(seq 1 30); do
+    if compose exec -T broadcast_channel pg_isready -U postgres -q; then
+      echo "  broadcast_channel is ready"
+      return 0
+    fi
+    sleep 1
+  done
+  echo "  broadcast_channel did not become ready in time"
+  exit 1
+}
+
 function clean_up {
     ARG=$?
     # Ensure we're in the script directory for cleanup
@@ -277,11 +304,9 @@ function test_push_policy {
   check_clients_logged "PUT /v1/policies/$regofile -> 200"
 }
 
-function test_data_publish {
-  echo "- Testing data publish for user $1"
+function publish_data {
+  # POST a data update to a single OPAL server (no assertion).
   user=$1
-
-  # Use curl to publish data update via OPAL server API
   curl -s -X POST http://localhost:7002/data/config \
     -H "Authorization: Bearer $OPAL_DATA_SOURCE_TOKEN" \
     -H "Content-Type: application/json" \
@@ -294,9 +319,13 @@ function test_data_publish {
         "save_method": "PUT"
       }]
     }'
+}
 
+function test_data_publish {
+  echo "- Testing data publish for user $1"
+  publish_data "$1"
   sleep 5
-  check_clients_logged "PUT /v1/data/users/$user/location -> 204"
+  check_clients_logged "PUT /v1/data/users/$1/location -> 204"
 }
 
 function test_statistics {
@@ -344,15 +373,54 @@ function main {
   test_push_policy "something"
   test_statistics
 
-  echo "- Testing broadcast channel disconnection"
+  echo "- Testing broadcast channel disconnection (graceful restart)"
   compose restart broadcast_channel
-  sleep 10
+  wait_for_broadcaster
+  # Give the servers' reconnecting broadcaster a moment to re-establish the backbone
+  sleep 5
 
   test_data_publish "alice"
   test_push_policy "another"
+
+  echo "- Testing broadcast channel disconnection (ungraceful kill)"
+  compose kill broadcast_channel
+  sleep 3
+  compose up -d broadcast_channel
+  wait_for_broadcaster
+  sleep 5
+
   test_data_publish "sunil"
   test_data_publish "eve"
   test_push_policy "best_one_yet"
+
+  # Regression guards for the broadcaster-disconnect storm (see pubsub_resilience.py):
+  # the servers must have reconnected to the backbone (this line is logged on every
+  # (re)connect, so it fires on both the graceful-restart and ungraceful-kill paths),
+  # and must NOT have spewed the non-idempotent-disconnect ValueError that drove the
+  # fleet-wide drop storm.
+  check_servers_logged "Broadcaster listener connected to channel"
+  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).
+  echo "- Testing cross-instance consistency across a backbone outage"
+  compose kill broadcast_channel
+  sleep 3
+  publish_data "consistency_user"
+  sleep 2
+  compose up -d broadcast_channel
+  wait_for_broadcaster
+  # allow buffered replay + (if needed) 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).
+  check_servers_logged "buffered for replay"
+  check_servers_logged "Replaying"
+  # BOTH clients (on different replicas via the VIP) must end up with the value.
+  check_clients_logged "PUT /v1/data/users/consistency_user/location -> 204"
   # TODO: Test statistics feature again after broadcaster restart (should first fix statistics bug)
 }
 

documentation/docs/getting-started/running-opal/run-opal-server/broadcast-interface.mdx

@@ -59,3 +59,22 @@ This is how you define the number of workers (pay attention: this env var is not
 | Env Var Name        | Function                                                         |
 | :------------------ | :--------------------------------------------------------------- |
 | UVICORN_NUM_WORKERS | the number of workers in a single container (example value: `4`) |
+
+#### 4) Broadcaster reconnection (resilience)
+
+If the broadcast backbone (Postgres/Redis/Kafka) briefly drops — for example during a managed-database failover or restart — OPAL servers reconnect to it automatically with bounded exponential backoff, instead of dropping their connected clients. This is enabled by default; the following **server-side** env vars (all prefixed with `OPAL_`) tune it:
+
+| Env Var Name                                 | Default | Function                                                                                                                                          |
+| :------------------------------------------- | :------ | :----------------------------------------------------------------------------------------------------------------------------------------------- |
+| OPAL_BROADCAST_RECONNECT_ENABLED             | `true`  | Reconnect the broadcaster reader on a backbone disconnect instead of dropping all client connections. Set to `false` for the legacy behavior.     |
+| OPAL_BROADCAST_RECONNECT_MAX_RETRIES         | `0`     | Maximum consecutive reconnect attempts before giving up and letting the worker restart. `0` means retry forever.                                  |
+| OPAL_BROADCAST_RECONNECT_BACKOFF_MIN_SECONDS | `0.5`   | Minimum backoff (seconds) between reconnect attempts.                                                                                             |
+| OPAL_BROADCAST_RECONNECT_BACKOFF_MAX_SECONDS | `30`    | Maximum backoff (seconds) between reconnect attempts.                                                                                             |
+| OPAL_BROADCAST_REPLAY_BUFFER_SIZE            | `10000` | Max number of outbound broadcasts buffered while the backbone is down and replayed on reconnect (`0` disables buffering). On overflow the oldest are dropped. |
+| OPAL_BROADCAST_RESYNC_ON_RECONNECT           | `true`  | After a backbone gap, force this worker's clients to reconnect so they re-fetch full policy + data state. Set to `false` to rely only on best-effort replay. |
+| OPAL_BROADCAST_RESYNC_SETTLE_SECONDS         | `2`     | Grace period after a reconnect before replaying buffered broadcasts and resyncing clients, to let peer servers re-subscribe.                      |
+
+Consistency across the outage is handled in two layers. While the backbone is unreachable, client websocket connections are kept alive but cross-server fan-out is paused. On reconnect:
+
+- **Replay buffer** — broadcasts that failed to reach the backbone during the outage are replayed, so peer servers that have re-subscribed catch up without a client refetch. This is best-effort: the backbone keeps no replay of its own, so a peer that is slow to re-subscribe may miss a replayed message.
+- **Resync** (the guarantee) — each server forces its own clients to reconnect and re-fetch the full policy/data state. Because every server experienced the same gap, every server reconciles its own clients and the fleet converges to current truth. Updates missed during the gap are therefore reconciled even if the replay did not reach a peer in time.

packages/opal-server/opal_server/config.py

@@ -57,6 +57,57 @@ class OpalServerConfig(Confi):
         True,
         description="Enable experimental bugfix for broadcast connection loss",
     )
+    BROADCAST_RECONNECT_ENABLED = confi.bool(
+        "BROADCAST_RECONNECT_ENABLED",
+        True,
+        description="Reconnect the broadcaster reader on a backbone disconnect instead "
+        "of dropping all client connections. Set to False to revert to the legacy "
+        "(non-reconnecting) broadcaster.",
+    )
+    BROADCAST_RECONNECT_MAX_RETRIES = confi.int(
+        "BROADCAST_RECONNECT_MAX_RETRIES",
+        0,
+        description="Maximum consecutive broadcaster reconnect attempts before giving "
+        "up and letting the worker restart (0 = retry forever).",
+    )
+    BROADCAST_RECONNECT_BACKOFF_MIN_SECONDS = confi.float(
+        "BROADCAST_RECONNECT_BACKOFF_MIN_SECONDS",
+        0.5,
+        description="Minimum backoff in seconds between broadcaster reconnect attempts.",
+    )
+    BROADCAST_RECONNECT_BACKOFF_MAX_SECONDS = confi.float(
+        "BROADCAST_RECONNECT_BACKOFF_MAX_SECONDS",
+        30.0,
+        description="Maximum backoff in seconds between broadcaster reconnect attempts.",
+    )
+    BROADCAST_REPLAY_BUFFER_SIZE = confi.int(
+        "BROADCAST_REPLAY_BUFFER_SIZE",
+        10000,
+        description="Max number of outbound broadcasts buffered while the backbone is "
+        "down and replayed on reconnect (0 disables buffering). On overflow the oldest "
+        "are dropped and clients are resynced instead.",
+    )
+    BROADCAST_RESYNC_ON_RECONNECT = confi.bool(
+        "BROADCAST_RESYNC_ON_RECONNECT",
+        True,
+        description="After a backbone gap that may have lost updates, force this "
+        "worker's connected clients to reconnect so they re-fetch full policy + data "
+        "state (guarantees cross-instance consistency).",
+    )
+    BROADCAST_RESYNC_SETTLE_SECONDS = confi.float(
+        "BROADCAST_RESYNC_SETTLE_SECONDS",
+        2.0,
+        description="Grace period after a broadcaster reconnect before replaying "
+        "buffered broadcasts and resyncing clients, to let peer servers re-subscribe.",
+    )
+    BROADCAST_HEALTHCHECK_ENABLED = confi.bool(
+        "BROADCAST_HEALTHCHECK_ENABLED",
+        True,
+        description="Make /healthcheck reflect the broadcaster reader's health so a "
+        "k8s readiness/liveness probe can route away from or restart a worker whose "
+        "reader is wedged while clients depend on it. Set to False to revert "
+        "/healthcheck to always returning ok.",
+    )
 
     # server security
     AUTH_PRIVATE_KEY_FORMAT = confi.enum(

packages/opal-server/opal_server/pubsub.py

@@ -1,3 +1,5 @@
+import asyncio
+import random
 import time
 from contextlib import contextmanager
 from contextvars import ContextVar
@@ -29,6 +31,7 @@
 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 pydantic import BaseModel
 from starlette.datastructures import QueryParams
 
@@ -141,12 +144,29 @@ def __init__(self, signer: JWTSigner, broadcaster_uri: str = None):
 
         self.broadcaster = None
         if broadcaster_uri is not None:
-            logger.info(f"Initializing broadcaster for server<->server communication")
-            self.broadcaster = EventBroadcaster(
-                broadcaster_uri,
-                notifier=self.notifier,
-                channel=opal_server_config.BROADCAST_CHANNEL_NAME,
-            )
+            if opal_server_config.BROADCAST_RECONNECT_ENABLED:
+                logger.info(
+                    "Initializing reconnecting broadcaster for server<->server communication"
+                )
+                self.broadcaster = ReconnectingBroadcaster(
+                    broadcaster_uri,
+                    notifier=self.notifier,
+                    channel=opal_server_config.BROADCAST_CHANNEL_NAME,
+                    reconnect_max_retries=opal_server_config.BROADCAST_RECONNECT_MAX_RETRIES,
+                    reconnect_backoff_min=opal_server_config.BROADCAST_RECONNECT_BACKOFF_MIN_SECONDS,
+                    reconnect_backoff_max=opal_server_config.BROADCAST_RECONNECT_BACKOFF_MAX_SECONDS,
+                    replay_buffer_size=opal_server_config.BROADCAST_REPLAY_BUFFER_SIZE,
+                    resync_settle_seconds=opal_server_config.BROADCAST_RESYNC_SETTLE_SECONDS,
+                )
+            else:
+                logger.info(
+                    "Initializing broadcaster for server<->server communication"
+                )
+                self.broadcaster = EventBroadcaster(
+                    broadcaster_uri,
+                    notifier=self.notifier,
+                    channel=opal_server_config.BROADCAST_CHANNEL_NAME,
+                )
         else:
             logger.info("Pub/Sub broadcaster is off")
 
@@ -159,6 +179,27 @@ def __init__(self, signer: JWTSigner, broadcaster_uri: str = None):
                 not opal_server_config.BROADCAST_CONN_LOSS_BUGFIX_EXPERIMENT_ENABLED
             ),
         )
+        # fastapi_websocket_rpc's ConnectionManager.disconnect is not idempotent: the RPC
+        # endpoint can call it twice for one socket (handle_disconnect plus the outer
+        # except in WebsocketRPCEndpoint.main_loop), raising
+        # ValueError('list.remove(x): x not in list'). Swap in an idempotent manager
+        # before any connection is served. Reaching into the wrapped endpoint is the
+        # only injection point the library offers (PubSubEndpoint takes no manager),
+        # so fail loudly if its internals ever move (a bare assert would be stripped
+        # under python -O, silently restoring the storm).
+        if not (
+            hasattr(self.endpoint, "endpoint")
+            and hasattr(self.endpoint.endpoint, "manager")
+        ):
+            raise RuntimeError(
+                "Unexpected fastapi_websocket_pubsub internals: cannot install "
+                "SafeConnectionManager (endpoint.endpoint.manager not found)"
+            )
+        self.endpoint.endpoint.manager = SafeConnectionManager()
+
+        if isinstance(self.broadcaster, ReconnectingBroadcaster):
+            self._wire_broadcaster_resync()
+
         authenticator = WebsocketJWTAuthenticator(signer)
 
         @self.api_router.get(
@@ -202,6 +243,44 @@ async def websocket_rpc_endpoint(
             finally:
                 await websocket.close()
 
+    def _wire_broadcaster_resync(self):
+        """Register the post-gap resync.
+
+        After any backbone gap, force this worker's clients to reconnect
+        so they re-run their full (scope-aware) policy + data
+        reconciliation. Every worker hit the same gap, so each
+        reconciles its own clients and the fleet converges — this is the
+        consistency guarantee; the broadcaster's replay buffer only
+        narrows the staleness window.
+        """
+        manager = self.endpoint.endpoint.manager
+        broadcaster = self.broadcaster
+        resync_enabled = opal_server_config.BROADCAST_RESYNC_ON_RECONNECT
+        settle = opal_server_config.BROADCAST_RESYNC_SETTLE_SECONDS
+
+        async def _on_broadcaster_reconnect():
+            if not resync_enabled:
+                logger.info("Broadcaster recovered after a gap; client resync disabled")
+                return
+            # Every worker hit the same gap; add a per-worker random delay so the
+            # fleet does not recycle its clients in lockstep.
+            if settle > 0:
+                await asyncio.sleep(random.uniform(0, settle))
+            logger.warning(
+                "Broadcaster recovered after a gap; resyncing this worker's clients "
+                "so they re-fetch current policy + data state"
+            )
+            # Closing every client would drive the broadcaster's listener count to 0
+            # and cancel the reconnecting reader task. Pin a listening context so the
+            # reader survives the recycle, and hold it briefly so reconnecting clients
+            # re-establish the count before we release.
+            async with broadcaster.get_listening_context():
+                await manager.close_all_staggered()
+                if settle > 0:
+                    await asyncio.sleep(settle)
+
+        broadcaster.set_reconnect_callback(_on_broadcaster_reconnect)
+
     @staticmethod
     async def _verify_permitted_topics(
         topics: Union[TopicList, ALL_TOPICS], channel: RpcChannel