diff --git a/.github/workflows/test-integration.yml b/.github/workflows/test-integration.yml index 07c82fbdb7..bdf4a8026c 100644 --- a/.github/workflows/test-integration.yml +++ b/.github/workflows/test-integration.yml @@ -143,10 +143,12 @@ jobs: && docker save egg-orchestrator:latest | sudo k3s ctr images import - \ && docker save egg-sandbox:latest | sudo k3s ctr images import - \ && docker save egg-litellm:latest | sudo k3s ctr images import - \ + && docker save egg-redis:latest | sudo k3s ctr images import - \ && docker save egg-gateway:"$EGG_IMAGE_TAG" | sudo k3s ctr images import - \ && docker save egg-orchestrator:"$EGG_IMAGE_TAG" | sudo k3s ctr images import - \ && docker save egg-sandbox:"$EGG_IMAGE_TAG" | sudo k3s ctr images import - \ - && docker save egg-litellm:"$EGG_IMAGE_TAG" | sudo k3s ctr images import -; then + && docker save egg-litellm:"$EGG_IMAGE_TAG" | sudo k3s ctr images import - \ + && docker save egg-redis:"$EGG_IMAGE_TAG" | sudo k3s ctr images import -; then echo "::endgroup::" echo "Image import succeeded on attempt ${attempt}" exit 0 @@ -175,7 +177,7 @@ jobs: run: | set -x docker images --format '{{.Repository}}:{{.Tag}}' \ - | grep -E '^egg-(gateway|orchestrator|sandbox|litellm):' \ + | grep -E '^egg-(gateway|orchestrator|sandbox|litellm|redis):' \ | xargs -r docker rmi -f || true docker builder prune -af || true docker image prune -af || true diff --git a/Makefile b/Makefile index de4cae7b38..e2e5c1b9c2 100644 --- a/Makefile +++ b/Makefile @@ -49,7 +49,7 @@ EGG_IMAGE_PREFIX := $(if $(EGG_IMAGE_REGISTRY),$(EGG_IMAGE_REGISTRY)/,) # registry entirely, remove it from EGG_REGISTRY_IMAGES; excluded images # publish through the save+import path instead (slower, but entirely # store-to-store on this host, no registry involved). -EGG_ALL_IMAGES := egg-gateway egg-orchestrator egg-sandbox egg-litellm +EGG_ALL_IMAGES := egg-gateway egg-orchestrator egg-sandbox egg-litellm egg-redis EGG_REGISTRY_IMAGES ?= $(EGG_ALL_IMAGES) # Images the registry path does NOT cover (imported via k3s-import instead). EGG_IMPORT_IMAGES := $(filter-out $(EGG_REGISTRY_IMAGES),$(EGG_ALL_IMAGES)) @@ -518,7 +518,7 @@ build: sync-venv-if-uv @$(PYTHON) scripts/prepare-sandbox-build-context.py repo-deps @echo "==> Building images with tag $(EGG_IMAGE_TAG) ($(BUILD_JOBS) parallel jobs)..." @$(MAKE) --no-print-directory -j$(BUILD_JOBS) $(BUILD_OUTPUT_SYNC) \ - build-gateway build-orchestrator build-sandbox build-litellm + build-gateway build-orchestrator build-sandbox build-litellm build-redis # Per-image sub-targets so `build` can run them under -j. They assume the # repo-deps/ build context has been prepared (the `build` target does that @@ -544,6 +544,9 @@ build-sandbox: build-litellm: DOCKER_BUILDKIT=1 docker build $(call image_tags,egg-litellm) -f config/litellm/Dockerfile config/litellm +build-redis: + DOCKER_BUILDKIT=1 docker build $(call image_tags,egg-redis) -f config/redis/Dockerfile config/redis + # ============================================================================ # Kubernetes (k3s) targets # ============================================================================ @@ -702,7 +705,8 @@ deploy: sudo-keepalive check-egg-images-present ## Deploy egg to k3s sed -e "s|egg-orchestrator:latest|$(call reg_prefix,egg-orchestrator)egg-orchestrator:$(EGG_IMAGE_TAG)|g" \ -e "s|egg-gateway:latest|$(call reg_prefix,egg-gateway)egg-gateway:$(EGG_IMAGE_TAG)|g" \ -e "s|egg-sandbox:latest|$(call reg_prefix,egg-sandbox)egg-sandbox:$(EGG_IMAGE_TAG)|g" \ - -e "s|egg-litellm:latest|$(call reg_prefix,egg-litellm)egg-litellm:$(EGG_IMAGE_TAG)|g" | \ + -e "s|egg-litellm:latest|$(call reg_prefix,egg-litellm)egg-litellm:$(EGG_IMAGE_TAG)|g" \ + -e "s|egg-redis:latest|$(call reg_prefix,egg-redis)egg-redis:$(EGG_IMAGE_TAG)|g" | \ kubectl apply -f - && \ scripts/clear-stuck-egg-pods.sh && \ scripts/await-egg-deploy.sh "$(EGG_IMAGE_TAG)" diff --git a/config/redis/Dockerfile b/config/redis/Dockerfile new file mode 100644 index 0000000000..913aec5215 --- /dev/null +++ b/config/redis/Dockerfile @@ -0,0 +1,25 @@ +# egg-redis — stock Redis, repackaged as a locally built/published image. +# +# Backs the orchestrator's Redis Streams message store +# (orchestrator/redis_message_store.py), the production backend for +# inter-agent messages (#2662). Durability target: the BRC transcript and +# consensus-tracker replay source must survive a mid-phase orchestrator +# restart (#3076) — so messages live here, not in the orchestrator process. +# +# All egg images are built locally and published to the cluster through the +# loopback registry / save+import flow (`make redeploy`, #2999); the cluster +# never pulls from public registries. This thin wrapper keeps Redis on that +# same supply chain instead of introducing the first public image reference +# in the manifests. +# +# Pinned to a specific upstream release; bump deliberately. The orchestrator +# uses Streams primitives only (XADD / XREAD BLOCK / XRANGE / XLEN / DEL) — +# stable since Redis 5 — so patch/minor bumps are low-risk, but keep the +# fakeredis-backed unit suite (orchestrator/tests/test_redis_message_store.py) +# and `make test-integration` green across any bump. +# +# No USER directive: the Deployment's securityContext +# (runAsUser/runAsGroup/fsGroup 1000, runAsNonRoot) governs the runtime UID, +# and the stock entrypoint works under an arbitrary non-root UID as long as +# /data is writable (fsGroup handles that). +FROM redis:8.6.4-alpine diff --git a/docs/architecture/coordination-state.md b/docs/architecture/coordination-state.md index 71ecf5be81..c96a6d43f7 100644 --- a/docs/architecture/coordination-state.md +++ b/docs/architecture/coordination-state.md @@ -181,7 +181,13 @@ and sets a health-visible degraded flag. Explicit level with no degraded flag. The `auto` selection semantics — Redis when available, in-memory fallback — are unchanged; deeper durability work stays in the [#3070](https://github.com/jwbron/egg/issues/3070) -lineage. The Redis path's restart semantics are pinned by +lineage. Since [#2662](https://github.com/jwbron/egg/issues/2662) the +k8s manifests deploy Redis (`k8s/base/redis-deployment.yaml`) and pin +the orchestrator to `EGG_MESSAGE_STORE_BACKEND=redis`, so the +accidental-loss row above cannot occur in-cluster: explicit `redis` +mode raises on an unreachable Redis instead of falling back. The +fallback path (and its fail-loud signal) remains for `auto`-configured +local/dev contexts. The Redis path's restart semantics are pinned by `orchestrator/tests/test_redis_message_store.py`: mid-phase messages survive a store re-instantiation against the same Redis (simulated orchestrator restart), while the designed `_clear_concurrent_state()` diff --git a/docs/guides/concurrent-execution.md b/docs/guides/concurrent-execution.md index 833d73e319..57eb6e1f27 100644 --- a/docs/guides/concurrent-execution.md +++ b/docs/guides/concurrent-execution.md @@ -100,7 +100,7 @@ All concurrent agent containers are wrapped with a shell script defined in `orch ## Message Bus -Agents communicate with each other during concurrent execution via the orchestrator message bus (`orchestrator/message_store.py`). In production, messages are stored in Redis Streams, surviving orchestrator restarts. Messages are cleared at phase transition. In test environments, an in-memory fallback is used when Redis is not available. +Agents communicate with each other during concurrent execution via the orchestrator message bus (`orchestrator/message_store.py`). In production, messages are stored in Redis Streams, surviving orchestrator restarts — the k8s manifests deploy a dedicated Redis (`k8s/base/redis-deployment.yaml`) and pin the orchestrator to it via `EGG_MESSAGE_STORE_BACKEND=redis` ([#2662](https://github.com/jwbron/egg/issues/2662)). Messages are cleared at phase transition. In test environments, an in-memory fallback is used when Redis is not available. ### How to Wait @@ -202,7 +202,7 @@ Returns total message count and a breakdown by message type. ### Message Store Backend -The message store uses Redis Streams when Redis is available, falling back to an in-memory store for tests or unconfigured environments. The backend is selected via the `EGG_MESSAGE_STORE_BACKEND` environment variable (`"auto"` by default, `"redis"` to require Redis, `"memory"` to force in-memory). +The message store uses Redis Streams when Redis is available, falling back to an in-memory store for tests or unconfigured environments. The backend is selected via the `EGG_MESSAGE_STORE_BACKEND` environment variable (`"auto"` by default, `"redis"` to require Redis, `"memory"` to force in-memory). The k8s deployment sets `"redis"` explicitly ([#2662](https://github.com/jwbron/egg/issues/2662)): in-cluster, a Redis outage fails loudly rather than silently degrading to the in-memory store (the `auto`→memory fallback is the mid-phase-restart loss risk that the [#3077](https://github.com/jwbron/egg/issues/3077) slice-6 health flag exists to catch). `integration_tests/test_message_store_backend.py` pins the deployed backend choice and round-trips a message through the live Redis. **Long-poll semantics (both backends):** `GET /messages/wait?for=&timeout=` blocks on both backends until a matching message arrives or the timeout elapses. The in-memory store implements blocking via a per-pipeline `threading.Condition`; the Redis backend uses `XREAD BLOCK` with a server-side type-filter loop. The silent non-blocking fallback that previously lived in `routes/messages.py` was removed in [#1897](https://github.com/jwbron/egg/issues/1897) so backend misconfiguration fails loudly in CI instead of returning empty results. See [Agent Wait Patterns](../reference/agent-wait-patterns.md#3-exit-code-contract-for-egg-orch-message-wait) for the full exit-code contract and the `EGG_MESSAGE_POLL_MAX_WAIT` cap. diff --git a/integration_tests/test_message_store_backend.py b/integration_tests/test_message_store_backend.py new file mode 100644 index 0000000000..6f46a8bd81 --- /dev/null +++ b/integration_tests/test_message_store_backend.py @@ -0,0 +1,126 @@ +"""End-to-end coverage for the deployed message-store backend (#2662). + +The k8s manifests pin the orchestrator to the Redis Streams backend +(``EGG_MESSAGE_STORE_BACKEND=redis`` + the ``redis`` Deployment/Service in +``k8s/base/``). Before #2662 the cluster deployed no Redis, so production +silently ran the in-memory ``MessageStore`` and ``RedisMessageStore`` had +zero live coverage — everything above unit tier went through ``fakeredis``. + +These tests run against the live cluster ``make test-integration`` deploys: + +* the deployed orchestrator explicitly selects the ``redis`` backend (the + silent ``auto``→memory fallback is the #3076 mid-phase-restart loss risk + that #3077 slice-6's degraded health flag exists to catch); +* ``/api/v1/health`` does not report the fallback marker; and +* a real XADD/XRANGE/DEL round-trip works from inside the orchestrator + container, through the production selection logic in + ``message_store._create_message_store()`` — exercising the env wiring, + the in-image redis client, Service DNS, and the live Redis in one path. + +The Streams semantics themselves (filters, blocking reads, cursor +staleness) are pinned by ``orchestrator/tests/test_redis_message_store.py``. +""" + +import subprocess +import uuid + +import requests + +from integration_tests.conftest import EggStack + +_NS = "egg-system" + + +def _kubectl(*args: str, timeout: int = 30) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["kubectl", "-n", _NS, *args], + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + + +class TestMessageStoreBackend: + def test_deployed_backend_is_explicit_redis(self, egg_stack: EggStack) -> None: + """The orchestrator Deployment pins EGG_MESSAGE_STORE_BACKEND=redis. + + Explicit ``redis`` mode cannot silently fall back to in-memory: + creation raises instead. A manifest regression to ``auto`` (or + dropping the env var, whose default is ``auto``) fails here. The + real end-to-end coverage that the pinned backend actually works + lives in ``test_redis_streams_roundtrip_from_orchestrator_pod``. + """ + result = _kubectl( + "get", + "deployment", + "orchestrator", + "-o", + "jsonpath={.spec.template.spec.containers[0].env" + "[?(@.name=='EGG_MESSAGE_STORE_BACKEND')].value}", + ) + assert result.returncode == 0, result.stderr + assert result.stdout.strip() == "redis", ( + "Deployed orchestrator must pin EGG_MESSAGE_STORE_BACKEND=redis; " + f"got {result.stdout.strip()!r}. See #2662." + ) + + def test_health_reports_message_store_ok(self, orchestrator_url: str) -> None: + """/api/v1/health must not carry the auto→memory fallback marker. + + Scope caveat: the marker (``is_memory_fallback_degraded()``) is + only ever set on the *auto*→memory path, so in the explicit + ``redis`` mode this deployment pins it is structurally always + absent — a healthy ``{"status": "ok"}`` here does NOT prove Redis + is reachable (an unreachable Redis 500s ops without flipping this + flag). This is a cheap guard that the orchestrator came up without + degrading; the round-trip test below carries the real proof that + the orchestrator→Redis path works. + """ + resp = requests.get(f"{orchestrator_url}/api/v1/health", timeout=10) + body = resp.json() + component = body.get("components", {}).get("message_store") + assert component == {"status": "ok"}, ( + f"message_store health component is {component!r} — the deployed " + "orchestrator fell back to the in-memory backend (#3077 slice-6 " + "marker). Check the redis Deployment and REDIS_HOST wiring." + ) + + def test_redis_streams_roundtrip_from_orchestrator_pod(self, egg_stack: EggStack) -> None: + """Live XADD/XRANGE/DEL through the production selection path. + + Runs a fresh Python process inside the orchestrator container so + ``get_message_store()`` re-runs backend selection with the pod's + real env. In explicit ``redis`` mode any failure (missing client + package, DNS, connection, AUTH) raises rather than falling back, + so a passing run proves the orchestrator→Redis path end to end. + """ + pipeline_id = f"itest-redis-{uuid.uuid4().hex[:8]}" + script = ( + "from message_store import Message, get_message_store\n" + "store = get_message_store()\n" + f"pid = {pipeline_id!r}\n" + "store.add_message(Message(pipeline_id=pid, from_role='itest'," + " message_type='PROGRESS', body='redis-e2e'))\n" + "msgs = store.get_messages(pid)\n" + "assert [m.body for m in msgs] == ['redis-e2e'], msgs\n" + "cleared = store.clear(pid)\n" + "assert cleared == 1, cleared\n" + "print('BACKEND', type(store).__name__)\n" + ) + result = _kubectl( + "exec", + "deploy/orchestrator", + "--", + "python", + "-c", + script, + timeout=60, + ) + assert result.returncode == 0, ( + f"in-pod round-trip failed (rc={result.returncode}):\n" + f"stdout: {result.stdout}\nstderr: {result.stderr}" + ) + assert "BACKEND RedisMessageStore" in result.stdout, ( + f"expected the RedisMessageStore backend, got: {result.stdout!r}" + ) diff --git a/k8s/base/kustomization.yaml b/k8s/base/kustomization.yaml index d3012c44da..58c0add42c 100644 --- a/k8s/base/kustomization.yaml +++ b/k8s/base/kustomization.yaml @@ -13,4 +13,8 @@ resources: - litellm-configmap.yaml - litellm-deployment.yaml - litellm-service.yaml + # Redis Streams message-store backend (issue #2662) — only the + # orchestrator talks to it (EGG_MESSAGE_STORE_BACKEND=redis). + - redis-deployment.yaml + - redis-service.yaml - network-policies.yaml diff --git a/k8s/base/orchestrator-deployment.yaml b/k8s/base/orchestrator-deployment.yaml index 33f5748ded..79b676ac94 100644 --- a/k8s/base/orchestrator-deployment.yaml +++ b/k8s/base/orchestrator-deployment.yaml @@ -102,6 +102,20 @@ spec: # (`egg-sandbox:latest`), and nothing exists at docker.io/egg. - name: EGG_SANDBOX_IMAGE value: "egg-sandbox:latest" + # Redis Streams message store (#2662). Explicitly `redis`, not + # `auto`: the silent auto→memory fallback is the #3076 + # mid-phase-restart message-loss risk that #3077 slice-6's + # degraded health flag exists to catch — in-cluster we want a + # Redis outage to fail loudly instead. The store singleton is + # created lazily and a failed creation leaves it unset, so the + # orchestrator naturally retries until Redis is reachable (no + # startup-ordering dependency on the redis Deployment). + - name: EGG_MESSAGE_STORE_BACKEND + value: "redis" + - name: REDIS_HOST + value: "redis.egg-system.svc.cluster.local" + - name: REDIS_PORT + value: "6379" # #2528: read per-repo role_patterns from the same # repositories.yaml the gateway uses so plan-time validation # and the planner-prompt boundaries match push-time diff --git a/k8s/base/redis-deployment.yaml b/k8s/base/redis-deployment.yaml new file mode 100644 index 0000000000..de5718eabc --- /dev/null +++ b/k8s/base/redis-deployment.yaml @@ -0,0 +1,154 @@ +# Redis backing the orchestrator's Streams message store (#2662). +# +# Durability scope: inter-agent messages (the BRC transcript and the +# consensus tracker's replay source) must survive a mid-phase orchestrator +# restart (#3076). The orchestrator selects this backend explicitly via +# EGG_MESSAGE_STORE_BACKEND=redis on orchestrator-deployment.yaml — the +# silent auto→memory fallback is exactly what #3077 slice-6's degraded +# health flag exists to catch. +# +# Only the orchestrator talks to Redis. Agent pods cannot reach it: the +# egg-agents namespace is default-deny egress (network-policies.yaml) with +# no allow rule for Redis, and nothing else in egg-system uses it. +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: redis-data + namespace: egg-system + labels: + app.kubernetes.io/name: redis + app.kubernetes.io/component: redis + app.kubernetes.io/part-of: egg +spec: + accessModes: + - ReadWriteOnce + # Deliberately small: streams are cleared at phase boundaries + # (_clear_concurrent_state), so the steady-state footprint is one + # phase's transcript per running pipeline. Bounded on purpose — this + # cluster has a btrfs chunk-allocation history (#2999); see also the + # maxmemory cap on the container args below. + resources: + requests: + storage: 1Gi +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: redis + namespace: egg-system + labels: + app.kubernetes.io/name: redis + app.kubernetes.io/component: redis + app.kubernetes.io/part-of: egg +spec: + # Single replica: the message store is per-pipeline scratch space with + # phase-scoped lifetime, not a shared datastore — one instance bounded + # by orchestrator request volume is sufficient, and the RWO PVC pins it + # to one pod anyway. + replicas: 1 + # Recreate so the replacement pod never deadlocks waiting on the RWO + # volume held by its predecessor. + strategy: + type: Recreate + selector: + matchLabels: + app.kubernetes.io/name: redis + app.kubernetes.io/component: redis + template: + metadata: + labels: + app.kubernetes.io/name: redis + app.kubernetes.io/component: redis + app.kubernetes.io/part-of: egg + spec: + enableServiceLinks: false + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + fsGroup: 1000 + containers: + - name: redis + # egg-redis: stock Redis pinned in config/redis/Dockerfile, built + # and published locally with the other egg images (the deploy + # target rewrites :latest -> :$(EGG_IMAGE_TAG)). Kept on the + # local supply chain rather than referencing a public image — + # see the Dockerfile header. + image: egg-redis:latest + imagePullPolicy: IfNotPresent + # args replace the image CMD, so redis-server must be restated. + # appendonly: messages must survive a Redis pod restart, not + # just an orchestrator restart. noeviction: under memory + # pressure, writes fail loudly (orchestrator surfaces a + # RedisError) instead of silently dropping stream entries — + # losing messages is the failure mode this deployment exists + # to remove. + # + # protected-mode no: stock Redis defaults to protected-mode yes, + # which rejects every non-loopback connection from the default + # (no-password) user (src/networking.c — there is no + # bind-address exemption in Redis 8). The orchestrator connects + # from a different pod via the ClusterIP service with no + # password (get_redis_message_store passes none), so without + # this every message op would be DENIED. The isolation + # protected mode would provide is already enforced by + # network-policies.yaml: egg-agents is default-deny egress with + # no Redis allow rule, and only the orchestrator (in egg-system, + # which has no NetworkPolicy) can reach the service. + args: + - redis-server + - --protected-mode + - "no" + - --appendonly + - "yes" + - --dir + - /data + - --maxmemory + - 256mb + - --maxmemory-policy + - noeviction + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + readOnlyRootFilesystem: true + ports: + - name: redis + containerPort: 6379 + protocol: TCP + livenessProbe: + exec: + command: + - redis-cli + - ping + initialDelaySeconds: 10 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 6 + readinessProbe: + exec: + command: + - redis-cli + - ping + initialDelaySeconds: 2 + periodSeconds: 5 + timeoutSeconds: 3 + failureThreshold: 6 + resources: + requests: + cpu: 50m + memory: 64Mi + limits: + cpu: 500m + # Headroom above the 256mb maxmemory cap: AOF rewrite + # forks and stream/jemalloc overhead live outside the + # maxmemory accounting. + memory: 512Mi + volumeMounts: + - name: data + mountPath: /data + volumes: + - name: data + persistentVolumeClaim: + claimName: redis-data diff --git a/k8s/base/redis-service.yaml b/k8s/base/redis-service.yaml new file mode 100644 index 0000000000..0fc11b37a7 --- /dev/null +++ b/k8s/base/redis-service.yaml @@ -0,0 +1,19 @@ +apiVersion: v1 +kind: Service +metadata: + name: redis + namespace: egg-system + labels: + app.kubernetes.io/name: redis + app.kubernetes.io/component: redis + app.kubernetes.io/part-of: egg +spec: + type: ClusterIP + selector: + app.kubernetes.io/name: redis + app.kubernetes.io/component: redis + ports: + - name: redis + port: 6379 + targetPort: 6379 + protocol: TCP diff --git a/k8s/overlays/local/kustomization.yaml b/k8s/overlays/local/kustomization.yaml index fc79542637..eb2563c372 100644 --- a/k8s/overlays/local/kustomization.yaml +++ b/k8s/overlays/local/kustomization.yaml @@ -23,3 +23,6 @@ images: - name: egg-litellm newName: egg-litellm newTag: latest + - name: egg-redis + newName: egg-redis + newTag: latest diff --git a/orchestrator/redis_message_store.py b/orchestrator/redis_message_store.py index 61bc4615d1..bfa98415be 100644 --- a/orchestrator/redis_message_store.py +++ b/orchestrator/redis_message_store.py @@ -34,6 +34,29 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] logger = get_logger("orchestrator.redis_message_store") +# Connection-pool socket timeout, in seconds. redis-py enforces this on +# the blocked read itself, so a single XREAD BLOCK >= this dies with +# "Timeout reading from socket" before the server can answer. Named so +# ``_MAX_BLOCK_MS`` and its guard test derive from the one real value +# rather than duplicating the literal. +_SOCKET_TIMEOUT_SEC = 5 + +# Upper bound for a single XREAD BLOCK, in milliseconds. MUST stay safely +# below the connection pool's ``socket_timeout`` (``_SOCKET_TIMEOUT_SEC``, +# applied in ``get_redis_message_store``): a single BLOCK >= socket_timeout +# dies with "Timeout reading from socket" before the server can answer — +# every agent long-poll (25-60 s) 500'd at the 5 s mark. Long waits are +# therefore chunked into BLOCK slices of at most this length; XREAD +# returns immediately when data arrives, so chunking costs one extra +# round-trip per idle slice, not delivery latency. Caught live by the +# first deployed canary pipeline for #2662 — fakeredis has no sockets, +# so the unit tier structurally cannot regress-test the timeout itself; +# the slice-cap contract is pinned in test_redis_message_store.py +# instead. The 1 s margin below the socket timeout absorbs round-trip +# and scheduling slack so the slice returns before redis-py trips. +_MAX_BLOCK_MS = (_SOCKET_TIMEOUT_SEC - 1) * 1000 + + def _stream_key(pipeline_id: str) -> str: """Get the Redis Stream key for a pipeline.""" return f"pipeline:{pipeline_id}:messages" @@ -256,13 +279,23 @@ def get_messages_with_meta( start_id = "0-0" since_id_stale = False if from_tip and not since_id and wait > 0: - # Redis XREAD treats ``$`` as "only entries with an ID greater - # than the greatest ID in the stream at call time" — i.e., a - # true event wait. Only safe on the blocking path (XREAD); - # XRANGE does not accept ``$``. Inner wait_for_types loop - # replaces ``$`` with a concrete ``last_sid`` after the first - # read, so subsequent cursor advancement uses real IDs. - start_id = "$" + # from_tip = "deliver only entries added after this call + # begins". Resolve the current tip to a CONCRETE stream id + # once, here, rather than passing Redis's ``$`` sentinel into + # the (chunked) blocking read below. ``$`` is re-resolved + # server-side to the *live* tip on every XREAD re-issue, so + # across idle BLOCK slices a message XADDed in the gap between + # one slice returning empty and the next being issued would be + # skipped — the next ``$`` starts after it — a silent drop in + # the consensus path. A fixed concrete id never advances on + # its own, so re-blocking from it re-scans that gap and cannot + # drop a mid-wait arrival. This mirrors the in-memory backend, + # which snapshots the tip once under its lock at call entry + # (message_store.py) so from_tip is race-free against + # concurrent add_message. An empty/missing stream resolves to + # "0-0" (read everything > 0-0), which still catches the first + # arrival. + start_id = self._resolve_tip_stream_id(pipeline_id) elif since_id: stream_id = self._resolve_stream_id(pipeline_id, since_id) if stream_id: @@ -353,6 +386,28 @@ def _read_once( if result_entries else [] ) + except redis.TimeoutError as e: + if block_ms is not None: + # A blocked read outlived the client socket timeout. + # _MAX_BLOCK_MS is sized to prevent this; if it fires + # anyway (e.g. an operator lowered socket_timeout), + # treat it as an idle slice — the caller's deadline + # loop bounds the retries — rather than 500ing the + # whole long-poll. + logger.warning( + "Blocking Redis Stream read hit the client socket " + "timeout; treating as an empty slice", + pipeline_id=pipeline_id, + block_ms=block_ms, + error=str(e), + ) + return [], None + logger.error( + "Failed to read from Redis Stream", + pipeline_id=pipeline_id, + error=str(e), + ) + raise except redis.RedisError as e: logger.error( "Failed to read from Redis Stream", @@ -379,7 +434,23 @@ def _read_once( # No type filter: preserve the original behaviour (fast path). if want_types is None: - messages, _ = _read_once(start_id, wait * 1000 if wait > 0 else None) + if wait > 0: + # Chunked blocking read (see _MAX_BLOCK_MS): re-issue + # XREAD BLOCK in slices until rows arrive or the wait + # budget elapses. Semantics match the former single + # XREAD BLOCK — the wait ends at the first batch of rows + # whether or not they survive the filters below. + fast_deadline = time.monotonic() + float(wait) + messages = [] + while True: + remaining_ms = int((fast_deadline - time.monotonic()) * 1000) + if remaining_ms <= 0: + break + messages, _ = _read_once(start_id, min(remaining_ms, _MAX_BLOCK_MS)) + if messages: + break + else: + messages, _ = _read_once(start_id, None) if role: messages = [m for m in messages if m.to_role == role or m.to_role == "all"] if from_role: @@ -401,7 +472,12 @@ def _read_once( block_ms: int | None if wait > 0: - block_ms = max(int(remaining * 1000), 1) + # Slice the remaining budget (see _MAX_BLOCK_MS). An idle + # slice reads nothing, leaves the cursor in place, and + # loops back here; the deadline check above terminates + # the wait. The inner-loop cap is no risk: 100 idle + # slices x 4 s far exceeds any wait budget. + block_ms = min(max(int(remaining * 1000), 1), _MAX_BLOCK_MS) else: block_ms = None @@ -423,8 +499,11 @@ def _read_once( return [], meta if last_sid is not None: - # Advance exclusively past the last sid so we don't re-read - # the same rows. + # Advance past the last sid so we don't re-read the same + # rows. On a pure-idle slice last_sid is None, so + # current_start stays the concrete tip id resolved at + # entry — re-blocking from it re-scans the gap, so a + # mid-wait arrival is never dropped. current_start = last_sid inner_loops += 1 @@ -437,6 +516,41 @@ def _read_once( ) return [], meta + def _resolve_tip_stream_id(self, pipeline_id: str) -> str: + """Snapshot the current stream tip as a concrete id for from_tip waits. + + ``XREVRANGE … COUNT 1`` returns the greatest stream id present + *now*; an ``XREAD`` started from it (exclusive) delivers only + later arrivals — the from_tip contract — without ever re-resolving + Redis's ``$`` sentinel mid-wait (see the call site for why that + matters). Returns ``"0-0"`` for an empty/missing stream, which + reads everything ``> 0-0`` and so still catches the first arrival. + A ``RedisError`` degrades to ``"0-0"`` for the same reason: the + caller's deadline loop bounds the read, and starting from the + head of a (typically empty) from_tip stream loses nothing. + """ + key = _stream_key(pipeline_id) + try: + entries = self._redis.xrevrange(key, count=1) + except redis.RedisError as exc: + # Mirror the since_id transient-degradation path above: log + # before degrading so the rare event is visible. On a + # non-empty stream this re-delivers pre-existing history as if + # new (at-least-once), the safe direction vs. dropping a + # message; the caller's deadline loop still bounds the read. + logger.warning( + "from_tip tip resolution failed transiently; degrading to 0-0", + pipeline_id=pipeline_id, + error=str(exc), + ) + return "0-0" + if entries: + stream_id = entries[0][0] + if isinstance(stream_id, bytes): + stream_id = stream_id.decode("utf-8") + return stream_id + return "0-0" + def get_latest_id(self, pipeline_id: str) -> str | None: """Return the ID of the most recent message for *pipeline_id*, or ``None``. @@ -581,8 +695,8 @@ def get_redis_message_store( db=db, decode_responses=False, # Handle decoding ourselves max_connections=20, - socket_timeout=5, - socket_connect_timeout=5, + socket_timeout=_SOCKET_TIMEOUT_SEC, + socket_connect_timeout=_SOCKET_TIMEOUT_SEC, ) client = redis.Redis(connection_pool=pool) # Test connection diff --git a/orchestrator/requirements.txt b/orchestrator/requirements.txt index de437277f3..5d45467d2a 100644 --- a/orchestrator/requirements.txt +++ b/orchestrator/requirements.txt @@ -20,5 +20,11 @@ requests>=2.31.0 # JSON Schema validation (used by egg_anchor for anchor validation) jsonschema>=4.20.0 +# Redis client for the Streams-backed message store (redis_message_store.py). +# Required in the image: the k8s manifests set EGG_MESSAGE_STORE_BACKEND=redis, +# and without this package the `import redis` inside _create_message_store() +# would fail before the backend is ever probed (#2662). +redis>=5.0,<7.0 + # MCP SDK for Streamable HTTP transport (pipeline management MCP server) mcp[cli]>=1.20.0,<2.0.0 diff --git a/orchestrator/tests/test_redis_message_store.py b/orchestrator/tests/test_redis_message_store.py index afc88bfa57..e8781227d5 100644 --- a/orchestrator/tests/test_redis_message_store.py +++ b/orchestrator/tests/test_redis_message_store.py @@ -774,12 +774,69 @@ def test_wait_zero_with_filter_returns_empty(self, store): class TestRedisFromTipSemantics: - """``from_tip=True`` uses Redis ``$`` so XREAD only matches entries - added after the call starts. + """``from_tip=True`` snapshots the stream tip to a concrete id once at + call entry, so XREAD only matches entries added after the call starts. - Backs the ``/messages/wait`` endpoint fix for issue #1925. + The concrete id (rather than Redis's ``$`` sentinel) is what keeps the + chunked blocking read from dropping a message XADDed between idle + slices — ``$`` re-resolves to the live tip on every re-issue, a fixed + id does not. Backs the ``/messages/wait`` endpoint fix for issue #1925. """ + def test_resolve_tip_stream_id_returns_concrete_id(self, store): + """A non-empty stream resolves to its greatest concrete stream id.""" + store.add_message( + Message( + pipeline_id="tip-pipeline", + from_role="coder", + message_type=MessageType.PROGRESS, + subject="first", + ) + ) + tip = store._resolve_tip_stream_id("tip-pipeline") + assert tip != "$" + assert "-" in tip # concrete Redis stream id, e.g. "1700000000000-0" + + def test_resolve_tip_stream_id_empty_stream_is_zero(self, store): + """An empty/missing stream resolves to ``0-0`` (read everything).""" + assert store._resolve_tip_stream_id("never-seen-pipeline") == "0-0" + + def test_from_tip_never_passes_dollar_to_xread(self, redis_client, monkeypatch): + """The from_tip blocking read must issue a CONCRETE start id, not ``$``. + + Pins the BLOCKING-2 fix: ``$`` re-resolves server-side on every + slice and would skip a message added between idle slices. + """ + redis_client.xadd( + _stream_key("tip-pipeline"), + { + "id": "x", + "pipeline_id": "tip-pipeline", + "from_role": "coder", + "to_role": "all", + "message_type": "PROGRESS", + "subject": "pre", + "body": "", + "metadata": "{}", + "timestamp": "", + "phase": "", + }, + ) + captured: list[str] = [] + + def capturing_xread(streams, count=None, block=None): + captured.append(next(iter(streams.values()))) + raise RuntimeError("stop") + + monkeypatch.setattr(redis_client, "xread", capturing_xread) + store = RedisMessageStore(redis_client) + with pytest.raises(RuntimeError): + store.get_messages("tip-pipeline", wait=1, from_tip=True) + + assert captured, "from_tip never issued an XREAD" + assert captured[0] != "$" + assert "-" in captured[0] + def test_pre_existing_match_ignored_with_from_tip(self, store): """A matching pre-existing entry must NOT satisfy a from_tip wait.""" store.add_message( @@ -792,10 +849,11 @@ def test_pre_existing_match_ignored_with_from_tip(self, store): ) ) - # fakeredis's XREAD with $ is a no-op on streams with data — it - # returns empty immediately because no "later" entry exists. This - # is the correct behaviour contract even though real Redis would - # actually block for the timeout. + # Production resolves the tip to a concrete id before XREAD (see + # _resolve_tip_stream_id), so the read starts from that id + # exclusively and returns empty — the pre-existing match is never + # delivered. Regressing _resolve_tip_stream_id to always-"0-0" + # would surface the pre-existing match and fail this assertion. start = time.monotonic() messages = store.get_messages( "test-pipeline", @@ -1294,3 +1352,120 @@ def test_restart_after_phase_boundary_wipe_stays_clean( post_store = RedisMessageStore(redis_client) assert post_store.get_messages(pipeline_id, limit=100) == [] assert post_store.get_status(pipeline_id) == {"total": 0, "by_type": {}} + + +class TestBlockingChunkCap: + """Live-canary regression for #2662: XREAD BLOCK vs client socket_timeout. + + The production connection pool (``get_redis_message_store``) sets + ``socket_timeout=5``. redis-py enforces that timeout on the blocked + read itself, so a single ``XREAD BLOCK`` longer than the socket + timeout dies with ``redis.TimeoutError`` before the server can + answer — on the first deployed pipeline every agent long-poll + (``wait=25``) errored at the 5 s mark. fakeredis has no sockets, so + the timeout itself cannot be reproduced at unit tier; these tests + pin the two halves of the fix instead: + + * no single blocking read ever requests more than ``_MAX_BLOCK_MS``; + * a ``redis.TimeoutError`` on a blocking slice degrades to an idle + slice instead of killing the whole wait (non-blocking reads keep + raising). + """ + + class _BlockCaptured(Exception): + """Sentinel to stop the store after the first blocking read.""" + + def _capture_first_block(self, redis_client, monkeypatch): + captured: list[int | None] = [] + + def capturing_xread(streams, count=None, block=None): + captured.append(block) + raise self._BlockCaptured() + + monkeypatch.setattr(redis_client, "xread", capturing_xread) + return captured + + def test_cap_stays_below_pool_socket_timeout(self): + import redis_message_store + + # Derive the bound from the *same* constant the pool applies + # (_SOCKET_TIMEOUT_SEC), not a duplicated literal — lowering the + # socket timeout then regresses here instead of silently in prod. + socket_timeout_ms = redis_message_store._SOCKET_TIMEOUT_SEC * 1000 + assert redis_message_store._MAX_BLOCK_MS < socket_timeout_ms + + def test_fast_path_block_is_capped(self, redis_client, monkeypatch): + import redis_message_store + + monkeypatch.setattr(redis_message_store, "_MAX_BLOCK_MS", 50) + captured = self._capture_first_block(redis_client, monkeypatch) + store = RedisMessageStore(redis_client) + + with pytest.raises(self._BlockCaptured): + store.get_messages("cap-pipeline", wait=10) + + # Pre-fix this was wait * 1000 == 10000 in a single XREAD. + assert captured == [50] + + def test_wait_for_types_block_is_capped(self, redis_client, monkeypatch): + import redis_message_store + + monkeypatch.setattr(redis_message_store, "_MAX_BLOCK_MS", 50) + captured = self._capture_first_block(redis_client, monkeypatch) + store = RedisMessageStore(redis_client) + + with pytest.raises(self._BlockCaptured): + store.get_messages( + "cap-pipeline", + wait=10, + wait_for_types=[MessageType.CONSENSUS_CONFIRMED], + ) + + assert captured == [50] + + def test_blocking_timeout_degrades_to_idle_slice(self, redis_client, monkeypatch): + import redis + + store = RedisMessageStore(redis_client) + store.add_message( + Message( + pipeline_id="timeout-pipeline", + from_role="coder", + to_role="all", + message_type=MessageType.PROGRESS, + subject="survives the flaky slice", + ) + ) + + real_xread = redis_client.xread + calls = {"n": 0} + + def flaky_xread(streams, count=None, block=None): + calls["n"] += 1 + if calls["n"] == 1: + raise redis.TimeoutError("Timeout reading from socket") + return real_xread(streams, count=count, block=block) + + monkeypatch.setattr(redis_client, "xread", flaky_xread) + + # Pre-fix the TimeoutError propagated and the route 500'd; now + # the first slice is treated as idle and the retry delivers. + messages = store.get_messages("timeout-pipeline", wait=2) + + assert calls["n"] >= 2 + assert [m.subject for m in messages] == ["survives the flaky slice"] + + def test_nonblocking_timeout_still_raises(self, redis_client, monkeypatch): + import redis + + def timeout_xrange(*args, **kwargs): + raise redis.TimeoutError("Timeout reading from socket") + + monkeypatch.setattr(redis_client, "xrange", timeout_xrange) + store = RedisMessageStore(redis_client) + + # The idle-slice degradation is scoped to blocking reads only — + # a timeout on a non-blocking read is a real error and must + # propagate, not silently return []. + with pytest.raises(redis.TimeoutError): + store.get_messages("timeout-pipeline", wait=0) diff --git a/pyproject.toml b/pyproject.toml index c44f1f6477..9fb2382656 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,11 @@ dependencies = [ "httpx>=0.27.0,<1.0.0", "pydantic>=2.0.0,<3.0.0", "packaging>=21.0", + # Redis client for the orchestrator's Streams-backed message store + # (#2662). Also a fakeredis dependency, but declared directly so the + # production import in redis_message_store.py doesn't ride on a + # dev-dep's transitive closure. + "redis>=5.0,<7.0", ] [project.optional-dependencies] diff --git a/scripts/await-egg-deploy.sh b/scripts/await-egg-deploy.sh index 3f84cff891..576e06a9e4 100755 --- a/scripts/await-egg-deploy.sh +++ b/scripts/await-egg-deploy.sh @@ -21,7 +21,7 @@ NS="egg-system" : "${1:?usage: $0 [timeout-seconds]}" TAG="$1" TIMEOUT="${2:-180}" -DEPLOYMENTS=(orchestrator gateway litellm) +DEPLOYMENTS=(orchestrator gateway litellm redis) # Keep kubectl stderr off of stdout so the success-path jsonpath value in # $out is strictly equal to the queried field. If a future cluster ever diff --git a/uv.lock b/uv.lock index cca3db0344..cde641e139 100644 --- a/uv.lock +++ b/uv.lock @@ -256,6 +256,7 @@ dependencies = [ { name = "pydantic" }, { name = "pyjwt" }, { name = "pyyaml" }, + { name = "redis" }, { name = "requests" }, { name = "waitress" }, ] @@ -301,6 +302,7 @@ requires-dist = [ { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=4.1.0" }, { name = "pytest-timeout", marker = "extra == 'dev'", specifier = ">=2.2.0" }, { name = "pyyaml", specifier = ">=6.0,<7.0" }, + { name = "redis", specifier = ">=5.0,<7.0" }, { name = "requests", specifier = ">=2.31.0,<3.0.0" }, { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.15.12,<0.16" }, { name = "starlette", marker = "extra == 'dev'", specifier = ">=0.36.0,<1.0.0" }, @@ -995,11 +997,11 @@ wheels = [ [[package]] name = "redis" -version = "7.3.0" +version = "6.4.0" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/da/82/4d1a5279f6c1251d3d2a603a798a1137c657de9b12cfc1fba4858232c4d2/redis-7.3.0.tar.gz", hash = "sha256:4d1b768aafcf41b01022410b3cc4f15a07d9b3d6fe0c66fc967da2c88e551034", size = 4928081, upload-time = "2026-03-06T18:18:16.287Z" } +sdist = { url = "https://files.pythonhosted.org/packages/0d/d6/e8b92798a5bd67d659d51a18170e91c16ac3b59738d91894651ee255ed49/redis-6.4.0.tar.gz", hash = "sha256:b01bc7282b8444e28ec36b261df5375183bb47a07eb9c603f284e89cbc5ef010", size = 4647399, upload-time = "2025-08-07T08:10:11.441Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/f0/28/84e57fce7819e81ec5aa1bd31c42b89607241f4fb1a3ea5b0d2dbeaea26c/redis-7.3.0-py3-none-any.whl", hash = "sha256:9d4fcb002a12a5e3c3fbe005d59c48a2cc231f87fbb2f6b70c2d89bb64fec364", size = 404379, upload-time = "2026-03-06T18:18:14.583Z" }, + { url = "https://files.pythonhosted.org/packages/e8/02/89e2ed7e85db6c93dfa9e8f691c5087df4e3551ab39081a4d7c6d1f90e05/redis-6.4.0-py3-none-any.whl", hash = "sha256:f0544fa9604264e9464cdf4814e7d4830f74b165d52f2a330a760a88dd248b7f", size = 279847, upload-time = "2025-08-07T08:10:09.84Z" }, ] [[package]]