fix(opal-server): freeze publishes during broadcaster backbone gaps to keep the fleet consistent

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))
@@ -223,10 +226,17 @@ function compose {
   docker compose -f ./docker-compose-app-tests.yml --env-file .env "$@"
 }
 
+# The positive assertions must exit explicitly on a miss: `main` runs on the
+# left of `&&` (see the retry loop), which suppresses `set -e` throughout it,
+# so a helper that merely returns grep's status can never fail the run.
 function check_clients_logged {
   echo "- Looking for msg '$1' in client's logs"
-  compose logs --index 1 opal_client | grep -q "$1"
-  compose logs --index 2 opal_client | grep -q "$1"
+  for index in 1 2; do
+    if ! compose logs --index "$index" opal_client | grep -q "$1"; then
+      echo "- '$1' not found in client $index logs"
+      exit 1
+    fi
+  done
 }
 
 function check_no_error {
@@ -240,18 +250,85 @@ function check_no_error {
 
 function check_servers_logged {
   echo "- Looking for msg '$1' in server's logs"
-  compose logs opal_server | grep -q "$1"
+  if ! compose logs opal_server | grep -q "$1"; then
+    echo "- '$1' not found in server logs"
+    exit 1
+  fi
 }
 
+# The negative assertions capture the logs first: piped directly into grep, a
+# failing `compose logs` (daemon hiccup, renamed service) is indistinguishable
+# from "message absent" and the check silently passes. (pipefail wouldn't help —
+# the `if pipeline; then fail; fi` shape falls through on ANY pipeline failure.)
 function check_servers_not_logged {
   echo "- Ensuring msg '$1' is absent from server's logs"
-  if compose logs opal_server | grep -q "$1"; then
+  local logs
+  logs=$(compose logs opal_server)
+  if [[ -z "$logs" ]]; then
+    echo "- Could not retrieve any server logs"
+    exit 1
+  fi
+  if grep -q "$1" <<< "$logs"; then
     echo "- Unexpectedly found '$1' in server logs:"
-    compose logs opal_server | grep "$1"
+    grep "$1" <<< "$logs"
     exit 1
   fi
 }
 
+function check_clients_not_logged {
+  echo "- Ensuring msg '$1' is absent from client's logs"
+  local logs
+  logs=$(compose logs opal_client)
+  if [[ -z "$logs" ]]; then
+    echo "- Could not retrieve any client logs"
+    exit 1
+  fi
+  if grep -q "$1" <<< "$logs"; then
+    echo "- Unexpectedly found '$1' in client logs:"
+    grep "$1" <<< "$logs"
+    exit 1
+  fi
+}
+
+function wait_for_servers_logged {
+  # Poll (up to $2 seconds) until a server logs $1 — for assertions whose
+  # timing depends on periodic tasks rather than the just-issued request.
+  echo "- Waiting (up to ${2}s) for msg '$1' in server's logs"
+  for _ in $(seq 1 "$2"); do
+    if compose logs opal_server 2>/dev/null | grep -q "$1"; then
+      return 0
+    fi
+    sleep 1
+  done
+  echo "- Timed out waiting for '$1' in server logs"
+  exit 1
+}
+
+function count_backbone_drops {
+  # Lines the reconnecting reader logs when an established backbone subscription
+  # ends (clean close or error) — i.e. the moments the publish-freeze gate closes.
+  compose logs opal_server 2>/dev/null \
+    | grep -cE "Broadcast subscriber ended|Broadcaster listener error" || true
+}
+
+function wait_for_backbone_drop {
+  # Wait until a server worker OBSERVES the backbone drop (a drop-log line past
+  # the pre-kill baseline in $1): the freeze only engages once the reader's read
+  # cycle exits and clears its connected flag, so publishing after a fixed sleep
+  # races that observation.
+  echo "- Waiting for a server to observe the backbone drop"
+  local baseline=$1
+  for _ in $(seq 1 30); do
+    if (( $(count_backbone_drops) > baseline )); then
+      echo "  backbone drop observed"
+      return 0
+    fi
+    sleep 1
+  done
+  echo "  no server observed the backbone drop in time"
+  exit 1
+}
+
 function wait_for_broadcaster {
   echo "- Waiting for broadcast_channel to accept connections"
   for _ in $(seq 1 30); do
@@ -402,24 +479,43 @@ 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"
+  drops_before=$(count_backbone_drops)
   compose kill broadcast_channel
-  sleep 3
+  wait_for_backbone_drop "$drops_before"
   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 freeze-episode summary is logged by the WORKER that froze, on its next
+  # delivered publish — and the re-publish above lands on 1 of N workers. In
+  # practice the exempt statistics keepalive (~10s, exercising every worker)
+  # delivers it; poll rather than assume one fixed delay covers that coupling.
+  wait_for_servers_logged "publish(es) during the gap" 30
   check_clients_logged "PUT /v1/data/users/consistency_user/location -> 204"
   # TODO: Test statistics feature again after broadcaster restart (should first fix statistics bug)
 }
@@ -430,7 +526,10 @@ RETRY_COUNT=0
 
 while [ $RETRY_COUNT -lt $MAX_RETRIES ]; do
   echo "Running test (attempt $((RETRY_COUNT+1)) of $MAX_RETRIES)..."
-  main && break
+  # Run main in a subshell: the assertion helpers fail via `exit 1` (see
+  # check_clients_logged), which would otherwise terminate the whole script
+  # through the EXIT trap and skip these retries entirely.
+  (main) && break
   RETRY_COUNT=$((RETRY_COUNT + 1))
   echo "Test failed, retrying..."
   # Tear the stack down before retrying so the next attempt starts clean:

packages/opal-server/opal_server/config.py

@@ -103,6 +103,24 @@ 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. Engages only with BROADCAST_RECONNECT_ENABLED (without the reconnecting "
+        "broadcaster there is no gap signal, so the flag is a no-op) and requires "
+        "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(

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,56 @@ 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.
+        # Both guardrails apply only where a freeze can actually engage, i.e. when the
+        # reconnecting broadcaster (the gap signal) was built above — warning a single-
+        # worker or reconnect-disabled deployment about the resync would be misdirection.
+        freeze_on_disconnect = opal_server_config.BROADCAST_FREEZE_ON_DISCONNECT
+        if freeze_on_disconnect and isinstance(
+            self.broadcaster, ReconnectingBroadcaster
+        ):
+            if 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
+        elif freeze_on_disconnect and self.broadcaster is not None:
+            logger.info(
+                "BROADCAST_FREEZE_ON_DISCONNECT is enabled but BROADCAST_RECONNECT_ENABLED "
+                "is disabled — the freeze engages only on the reconnecting broadcaster's "
+                "backbone-gap signal, so it is a no-op with the stock broadcaster."
+            )
+        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,
+            # Exempt: the git-webhook trigger (targets the server-side policy watcher, not
+            # clients — freezing it would drop repo-pull triggers with nothing to replay
+            # them) and the server-to-server coordination channels (statistics + broadcaster
+            # keepalive — dropping those corrupts state no resync rebuilds). The coordination
+            # channels are exempted by their CONFIGURED names: the endpoint's own "__" prefix
+            # rule covers only the defaults, and every one of these is operator-overridable.
+            freeze_exempt_topics=[
+                opal_server_config.POLICY_REPO_WEBHOOK_TOPIC,
+                opal_server_config.BROADCAST_KEEPALIVE_TOPIC,
+                opal_server_config.STATISTICS_WAKEUP_CHANNEL,
+                opal_server_config.STATISTICS_STATE_SYNC_CHANNEL,
+                opal_server_config.STATISTICS_SERVER_KEEPALIVE_CHANNEL,
+                opal_common_config.STATISTICS_ADD_CLIENT_CHANNEL,
+                opal_common_config.STATISTICS_REMOVE_CLIENT_CHANNEL,
+            ],
         )
         # fastapi_websocket_rpc's ConnectionManager.disconnect is not idempotent: the RPC
         # endpoint can call it twice for one socket (handle_disconnect plus the outer