fix(edge): retry+degrade side-effect POSTs; finish ACTIVITY-WP-0027
Consistency sweep and recently-on-scope retry transient 502/503/504 then return a degraded context snapshot instead of thrashing Temporal. Document edge-relay resilience. Retire Binky dual-clock host timers after claim-loop smoke.
This commit is contained in:
parent
2b3fc6a556
commit
4e605a6839
6 changed files with 334 additions and 20 deletions
77
docs/edge-relay-resilience.md
Normal file
77
docs/edge-relay-resilience.md
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
# Edge relay resilience (ACTIVITY-WP-0027-T06)
|
||||
|
||||
**Audience:** operators and agents debugging high-frequency activity-core jobs
|
||||
**Scope:** railiance01 `actcore-statehub-edge-relay` + worker `STATE_HUB_URL`
|
||||
|
||||
## Topology
|
||||
|
||||
```text
|
||||
actcore-worker
|
||||
STATE_HUB_URL=http://actcore-statehub-edge-relay:8000
|
||||
│
|
||||
▼
|
||||
edge relay (cluster)
|
||||
· GET allowlist → upstream or stale read-cache
|
||||
· POST side-effects → upstream (or 503 if busy/unreachable)
|
||||
│
|
||||
▼
|
||||
state-hub service (cluster DNS state-hub.state-hub.svc …)
|
||||
· may depend on tunnel / central hub posture
|
||||
```
|
||||
|
||||
Host rein-aharness should prefer the **same edge ClusterIP** (or `k8s://…edge-relay:8000`)
|
||||
so completion events outbox when central hub is flaky — see
|
||||
`docs/llm-connect-host-access.md`.
|
||||
|
||||
## Resolver classes
|
||||
|
||||
| Class | Examples | Hard-fail? | Notes |
|
||||
| ----- | -------- | ---------- | ----- |
|
||||
| **GET reads** | `fi_brief_status`, `domain_summary`, `state_summary` | Soft by default (`{}` / empty) unless `required: true` | Edge may serve **stale cache** (`X-StateHub-Edge-Cache: stale`) |
|
||||
| **Side-effect POSTs** | `consistency_sweep_remote_all`, `recently_on_scope_hourly` | After retries: **degrade** by default | See env below |
|
||||
| **Required pure reads** | `todo_md_staleness` when enabled | Yes if `required: true` | Prefer not to make hub-only reads required without cache |
|
||||
|
||||
## Side-effect POST behaviour (worker)
|
||||
|
||||
Implemented in `context_resolvers/state_hub.py` (`_post_json`):
|
||||
|
||||
1. **Retry** transient `502/503/504` and network/timeout errors
|
||||
(`STATE_HUB_POST_RETRIES`, default **3**; backoff
|
||||
`STATE_HUB_POST_RETRY_BACKOFF_SECONDS`, default **2s** × attempt).
|
||||
2. If still failing and degrade is on (`STATE_HUB_SIDE_EFFECT_DEGRADE` default
|
||||
**true**, or per-source `params.degrade_on_unavailable`):
|
||||
- **consistency_sweep:** returns
|
||||
`{exit_code: 75, lock_skipped: true, repos_processed: [], degraded: true, …}`
|
||||
so the workflow **completes** without Temporal ApplicationError storms.
|
||||
- **recently_on_scope:** returns
|
||||
`{generated: [], failed: [{reason: edge_unavailable}], degraded: true, …}`.
|
||||
3. Set `degrade_on_unavailable: false` on a definition (or env
|
||||
`STATE_HUB_SIDE_EFFECT_DEGRADE=false`) to keep hard-fail + Temporal retry
|
||||
after in-resolver attempts.
|
||||
|
||||
## Operator checks
|
||||
|
||||
```bash
|
||||
# Edge health (from host with ClusterIP or kubectl exec)
|
||||
curl -sS http://<edge-cluster-ip>:8000/edge/health | python3 -m json.tool
|
||||
|
||||
# Worker still pointing at edge
|
||||
kubectl -n activity-core exec deploy/actcore-worker -- env | grep STATE_HUB
|
||||
|
||||
# Recent consistency fires should COMPLETE even when edge was briefly 503
|
||||
# (look for degraded:true in context_snapshot if all retries failed)
|
||||
```
|
||||
|
||||
## What edge 503 usually means
|
||||
|
||||
- Upstream hub slow / lock held / temporary overload on
|
||||
`/consistency/sweep/remote-all` (large fleet).
|
||||
- Upstream unreachable; edge refuses to invent a false success for POSTs.
|
||||
- Outbox backlog (see `edge/health` → `outbox.pending_count`) is **write**
|
||||
backlog for queued progress — separate from sweep POST 503.
|
||||
|
||||
## Non-goals
|
||||
|
||||
- Making the edge invent successful sweep results when nothing ran.
|
||||
- Public exposure of State Hub or edge.
|
||||
- Replacing Temporal schedules with workstation heartbeats.
|
||||
|
|
@ -50,7 +50,9 @@ activity-core also schedules it — dual cadence and forgotten design.
|
|||
**Primary (Binky / FI):** activity-core schedule → `ops_run` → rein-aharness
|
||||
**claim loop** on railiance01. Host user-systemd timers are **break-glass only**
|
||||
(not cadence authority). FI dual-clock timer was disabled 2026-08-05 after the
|
||||
ops_run per-fire idempotency fix landed; re-enable only if claim loop is down.
|
||||
ops_run per-fire idempotency fix; **Binky dual-clock timers** (daily/mail/review)
|
||||
were disabled 2026-08-05 after claim-loop smoke (ACTIVITY-WP-0027-T07). Re-enable
|
||||
only if claim loop is down.
|
||||
|
||||
---
|
||||
|
||||
|
|
|
|||
|
|
@ -736,6 +736,11 @@ unreachable, which keeps daily triage context resolution alive during outages.
|
|||
> receipts as successful sink delivery pending replay, and consumes stale cached
|
||||
> reads transparently.
|
||||
|
||||
Side-effect POSTs (`consistency_sweep_remote_all`, `recently_on_scope_hourly`)
|
||||
retry transient 502/503/504 then **degrade** by default so required workflows do
|
||||
not thrash Temporal when the edge is briefly unavailable. Details:
|
||||
`docs/edge-relay-resilience.md` (ACTIVITY-WP-0027-T06).
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Worker fails to start: "ACTCORE_DB_URL is required"
|
||||
|
|
|
|||
|
|
@ -26,13 +26,21 @@ When STATE_HUB_URL points at the state-hub edge relay, allowlisted GET reads may
|
|||
be served from a stale local cache during upstream outages (`X-StateHub-Edge-Cache:
|
||||
stale`). activity-core treats those as ordinary successful reads so workflows can
|
||||
continue with last-known hub state.
|
||||
|
||||
Side-effect POSTs (`consistency_sweep_remote_all`, `recently_on_scope_hourly`)
|
||||
retry transient 502/503/504 and network errors, then optionally **degrade**
|
||||
instead of hard-failing the workflow (ACTIVITY-WP-0027-T06). See
|
||||
`docs/edge-relay-resilience.md`.
|
||||
|
||||
Config: STATE_HUB_URL env var (default: http://127.0.0.1:8000).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
|
|
@ -40,6 +48,8 @@ import httpx
|
|||
|
||||
from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY, ContextResolver
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000"
|
||||
_TIMEOUT_SECONDS = 10.0
|
||||
_SWEEP_TIMEOUT_SECONDS = 330.0
|
||||
|
|
@ -50,6 +60,9 @@ _OPEN_TASK_STATUSES = {"wait", "todo", "progress"}
|
|||
# forcing the rule expression to special-case None.
|
||||
_NEVER_SCANNED_AGE_DAYS = 99999
|
||||
|
||||
# Transient HTTP statuses worth retrying against the edge relay / hub.
|
||||
_TRANSIENT_HTTP = frozenset({502, 503, 504})
|
||||
|
||||
|
||||
def _base_url() -> str:
|
||||
return os.environ.get("STATE_HUB_URL", _DEFAULT_STATE_HUB_URL).rstrip("/")
|
||||
|
|
@ -65,11 +78,92 @@ def _fetch_json(path: str, params: dict[str, Any] | None = None) -> Any:
|
|||
return {}
|
||||
|
||||
|
||||
def _post_json(path: str, payload: dict[str, Any], *, timeout: float = _TIMEOUT_SECONDS) -> Any:
|
||||
def _post_retry_attempts() -> int:
|
||||
try:
|
||||
return max(1, int(os.environ.get("STATE_HUB_POST_RETRIES", "3")))
|
||||
except ValueError:
|
||||
return 3
|
||||
|
||||
|
||||
def _post_retry_backoff_seconds() -> float:
|
||||
try:
|
||||
return max(0.0, float(os.environ.get("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "2")))
|
||||
except ValueError:
|
||||
return 2.0
|
||||
|
||||
|
||||
def _side_effect_degrade_default() -> bool:
|
||||
"""Default degrade-on-unavailable for side-effect POSTs after retries."""
|
||||
raw = (os.environ.get("STATE_HUB_SIDE_EFFECT_DEGRADE") or "true").strip().lower()
|
||||
return raw in {"1", "true", "yes", "on", ""}
|
||||
|
||||
|
||||
def _post_json(
|
||||
path: str,
|
||||
payload: dict[str, Any],
|
||||
*,
|
||||
timeout: float = _TIMEOUT_SECONDS,
|
||||
retries: int | None = None,
|
||||
) -> Any:
|
||||
"""POST JSON with retries on transient edge/hub failures (ACTIVITY-WP-0027)."""
|
||||
url = f"{_base_url()}{path}"
|
||||
resp = httpx.post(url, json=payload, timeout=timeout)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
attempts = _post_retry_attempts() if retries is None else max(1, retries)
|
||||
backoff = _post_retry_backoff_seconds()
|
||||
last_exc: Exception | None = None
|
||||
|
||||
for attempt in range(attempts):
|
||||
try:
|
||||
resp = httpx.post(url, json=payload, timeout=timeout)
|
||||
status = getattr(resp, "status_code", None)
|
||||
if status in _TRANSIENT_HTTP and attempt + 1 < attempts:
|
||||
logger.warning(
|
||||
"state-hub POST %s returned %s (attempt %s/%s); retrying",
|
||||
path,
|
||||
status,
|
||||
attempt + 1,
|
||||
attempts,
|
||||
)
|
||||
if backoff:
|
||||
time.sleep(backoff * (attempt + 1))
|
||||
continue
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
except (
|
||||
httpx.TimeoutException,
|
||||
httpx.NetworkError,
|
||||
httpx.RemoteProtocolError,
|
||||
) as exc:
|
||||
last_exc = exc
|
||||
if attempt + 1 >= attempts:
|
||||
raise
|
||||
logger.warning(
|
||||
"state-hub POST %s network error %s (attempt %s/%s); retrying",
|
||||
path,
|
||||
exc,
|
||||
attempt + 1,
|
||||
attempts,
|
||||
)
|
||||
if backoff:
|
||||
time.sleep(backoff * (attempt + 1))
|
||||
except httpx.HTTPStatusError as exc:
|
||||
# Non-transient HTTP errors fail immediately.
|
||||
code = exc.response.status_code if exc.response is not None else None
|
||||
if code in _TRANSIENT_HTTP and attempt + 1 < attempts:
|
||||
last_exc = exc
|
||||
if backoff:
|
||||
time.sleep(backoff * (attempt + 1))
|
||||
continue
|
||||
raise
|
||||
|
||||
if last_exc is not None:
|
||||
raise last_exc
|
||||
raise RuntimeError(f"state-hub POST {path} failed without exception")
|
||||
|
||||
|
||||
def _want_degrade(params: dict[str, Any]) -> bool:
|
||||
if "degrade_on_unavailable" in params:
|
||||
return bool(params.get("degrade_on_unavailable"))
|
||||
return _side_effect_degrade_default()
|
||||
|
||||
|
||||
def _validate_consistency_sweep_remote_all(result: Any) -> dict[str, Any]:
|
||||
|
|
@ -138,21 +232,58 @@ class StateHubContextResolver(ContextResolver):
|
|||
payload = {
|
||||
key: value
|
||||
for key, value in params.items()
|
||||
if key not in {"required"}
|
||||
if key not in {"required", "degrade_on_unavailable"}
|
||||
}
|
||||
result = _post_json("/recently-on-scope/hourly", payload)
|
||||
try:
|
||||
result = _post_json("/recently-on-scope/hourly", payload)
|
||||
except (httpx.HTTPError, ValueError) as exc:
|
||||
if not _want_degrade(params):
|
||||
raise
|
||||
logger.warning(
|
||||
"recently_on_scope_hourly degraded after retries: %s", exc
|
||||
)
|
||||
return {
|
||||
"generated": [],
|
||||
"skipped": [],
|
||||
"failed": [
|
||||
{
|
||||
"reason": "edge_unavailable",
|
||||
"detail": str(exc)[:300],
|
||||
}
|
||||
],
|
||||
"degraded": True,
|
||||
"degraded_reason": str(exc)[:300],
|
||||
}
|
||||
return _validate_recently_on_scope_hourly(result)
|
||||
if query == "consistency_sweep_remote_all":
|
||||
payload = {
|
||||
key: value
|
||||
for key, value in params.items()
|
||||
if key not in {"required"}
|
||||
if key not in {"required", "degrade_on_unavailable"}
|
||||
}
|
||||
result = _post_json(
|
||||
"/consistency/sweep/remote-all",
|
||||
payload,
|
||||
timeout=_SWEEP_TIMEOUT_SECONDS,
|
||||
)
|
||||
try:
|
||||
result = _post_json(
|
||||
"/consistency/sweep/remote-all",
|
||||
payload,
|
||||
timeout=_SWEEP_TIMEOUT_SECONDS,
|
||||
)
|
||||
except (httpx.HTTPError, ValueError) as exc:
|
||||
if not _want_degrade(params):
|
||||
raise
|
||||
logger.warning(
|
||||
"consistency_sweep_remote_all degraded after retries: %s", exc
|
||||
)
|
||||
# Shape matches validator; exit_code 75 = temporary failure.
|
||||
return {
|
||||
"exit_code": 75,
|
||||
"lock_skipped": True,
|
||||
"repos_processed": [],
|
||||
"skipped_clean": [],
|
||||
"skipped_missing": [],
|
||||
"skipped_budget": [],
|
||||
"degraded": True,
|
||||
"degraded_reason": str(exc)[:300],
|
||||
}
|
||||
return _validate_consistency_sweep_remote_all(result)
|
||||
if query == "phase5_stabilization_check":
|
||||
return _phase5_stabilization_check(params)
|
||||
|
|
|
|||
|
|
@ -9,13 +9,25 @@ from activity_core.context_resolvers.state_hub import StateHubContextResolver
|
|||
|
||||
|
||||
class DummyResponse:
|
||||
def __init__(self, payload: Any, status_error: Exception | None = None) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
payload: Any,
|
||||
status_error: Exception | None = None,
|
||||
status_code: int = 200,
|
||||
) -> None:
|
||||
self.payload = payload
|
||||
self.status_error = status_error
|
||||
self.status_code = status_code
|
||||
|
||||
def raise_for_status(self) -> None:
|
||||
if self.status_error is not None:
|
||||
raise self.status_error
|
||||
if self.status_code >= 400:
|
||||
req = httpx.Request("POST", "http://dummy")
|
||||
resp = httpx.Response(self.status_code, request=req)
|
||||
raise httpx.HTTPStatusError(
|
||||
f"{self.status_code}", request=req, response=resp
|
||||
)
|
||||
|
||||
def json(self) -> Any:
|
||||
return self.payload
|
||||
|
|
@ -498,9 +510,30 @@ def test_recently_on_scope_hourly_failure_bubbles(monkeypatch) -> None:
|
|||
raise httpx.ConnectError("offline")
|
||||
|
||||
monkeypatch.setattr(httpx, "post", fake_post)
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRIES", "1")
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0")
|
||||
|
||||
with pytest.raises(httpx.ConnectError):
|
||||
StateHubContextResolver().resolve("recently_on_scope_hourly", None, {"range": "1h"})
|
||||
StateHubContextResolver().resolve(
|
||||
"recently_on_scope_hourly",
|
||||
None,
|
||||
{"range": "1h", "degrade_on_unavailable": False},
|
||||
)
|
||||
|
||||
|
||||
def test_recently_on_scope_hourly_degrades(monkeypatch) -> None:
|
||||
def fake_post(url: str, **kwargs: Any) -> DummyResponse:
|
||||
raise httpx.ConnectError("offline")
|
||||
|
||||
monkeypatch.setattr(httpx, "post", fake_post)
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRIES", "2")
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0")
|
||||
|
||||
result = StateHubContextResolver().resolve(
|
||||
"recently_on_scope_hourly", None, {"range": "1h"}
|
||||
)
|
||||
assert result["degraded"] is True
|
||||
assert result["failed"][0]["reason"] == "edge_unavailable"
|
||||
|
||||
|
||||
def test_consistency_sweep_remote_all_posts_batch(monkeypatch) -> None:
|
||||
|
|
@ -544,15 +577,69 @@ def test_consistency_sweep_remote_all_failure_bubbles(monkeypatch) -> None:
|
|||
raise httpx.ConnectError("offline")
|
||||
|
||||
monkeypatch.setattr(httpx, "post", fake_post)
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRIES", "1")
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0")
|
||||
|
||||
with pytest.raises(httpx.ConnectError):
|
||||
StateHubContextResolver().resolve(
|
||||
"consistency_sweep_remote_all",
|
||||
None,
|
||||
{"max_seconds": 300},
|
||||
{"max_seconds": 300, "degrade_on_unavailable": False},
|
||||
)
|
||||
|
||||
|
||||
def test_consistency_sweep_degrades_after_retries(monkeypatch) -> None:
|
||||
calls: list[int] = []
|
||||
|
||||
def fake_post(url: str, **kwargs: Any) -> DummyResponse:
|
||||
calls.append(1)
|
||||
raise httpx.ConnectError("offline")
|
||||
|
||||
monkeypatch.setattr(httpx, "post", fake_post)
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRIES", "3")
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0")
|
||||
monkeypatch.setenv("STATE_HUB_SIDE_EFFECT_DEGRADE", "true")
|
||||
|
||||
result = StateHubContextResolver().resolve(
|
||||
"consistency_sweep_remote_all",
|
||||
None,
|
||||
{"max_seconds": 300},
|
||||
)
|
||||
assert result["degraded"] is True
|
||||
assert result["exit_code"] == 75
|
||||
assert result["lock_skipped"] is True
|
||||
assert len(calls) == 3
|
||||
|
||||
|
||||
def test_consistency_sweep_retries_then_succeeds(monkeypatch) -> None:
|
||||
calls: list[int] = []
|
||||
|
||||
def fake_post(url: str, **kwargs: Any) -> DummyResponse:
|
||||
calls.append(1)
|
||||
if len(calls) < 2:
|
||||
return DummyResponse({}, status_code=503)
|
||||
return DummyResponse(
|
||||
{
|
||||
"exit_code": 0,
|
||||
"lock_skipped": False,
|
||||
"repos_processed": [],
|
||||
},
|
||||
status_code=200,
|
||||
)
|
||||
|
||||
monkeypatch.setattr(httpx, "post", fake_post)
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRIES", "3")
|
||||
monkeypatch.setenv("STATE_HUB_POST_RETRY_BACKOFF_SECONDS", "0")
|
||||
|
||||
result = StateHubContextResolver().resolve(
|
||||
"consistency_sweep_remote_all",
|
||||
None,
|
||||
{"max_seconds": 300},
|
||||
)
|
||||
assert result["exit_code"] == 0
|
||||
assert len(calls) == 2
|
||||
|
||||
|
||||
def test_consistency_sweep_remote_all_rejects_empty_response(monkeypatch) -> None:
|
||||
def fake_post(url: str, **kwargs: Any) -> DummyResponse:
|
||||
return DummyResponse({})
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ type: workplan
|
|||
title: "Hardened llm-connect access + run→artifact links in ops UI"
|
||||
domain: infotech
|
||||
repo: activity-core
|
||||
status: active
|
||||
status: finished
|
||||
owner: grok
|
||||
topic_slug: activity-core
|
||||
priority: high
|
||||
|
|
@ -276,7 +276,7 @@ links in ops UI without hand-editing DB rows.
|
|||
|
||||
```task
|
||||
id: ACTIVITY-WP-0027-T06
|
||||
status: todo
|
||||
status: done
|
||||
priority: medium
|
||||
state_hub_task_id: "52f1db5e-23db-4dcd-aaba-71c11deaf09a"
|
||||
```
|
||||
|
|
@ -296,7 +296,7 @@ fixed or explicitly accepted with a follow-up id.
|
|||
|
||||
```task
|
||||
id: ACTIVITY-WP-0027-T07
|
||||
status: todo
|
||||
status: done
|
||||
priority: low
|
||||
state_hub_task_id: "245955d4-83bd-4145-b71e-2e6449754e31"
|
||||
```
|
||||
|
|
@ -336,7 +336,7 @@ Depends on T02 + T04 (minimum).
|
|||
- [x] rein-aharness on railiance01 does not depend on ad-hoc `kubectl port-forward` for llm-connect
|
||||
- [x] `https://activity.coulomb.social/ops/ui` shows clickable artefact links for recent FI/Binky successes (API verified; SSO UI serves same payload)
|
||||
- [x] Contract tests cover join + URL builder; no secrets in responses
|
||||
- [ ] Remaining caveats (edge 503, Binky timers) either fixed or explicitly deferred with owner
|
||||
- [x] Remaining caveats (edge 503, Binky timers) either fixed or explicitly deferred with owner
|
||||
|
||||
## References
|
||||
|
||||
|
|
@ -371,3 +371,15 @@ Port-forward deprecated for claim-loop production.
|
|||
### Remaining
|
||||
- T06 edge-relay soft-fail for required resolvers
|
||||
- T07 Binky dual-clock timer retirement after clean claim days
|
||||
|
||||
### T06 (2026-08-05)
|
||||
- `_post_json` retries 502/503/504 + network errors (default 3×)
|
||||
- Side-effect POSTs degrade after retries (`STATE_HUB_SIDE_EFFECT_DEGRADE=true`)
|
||||
- Doc: `docs/edge-relay-resilience.md` + runbook pointer
|
||||
- Tests: degrade + retry success paths
|
||||
|
||||
### T07 (2026-08-05)
|
||||
- Binky claim smoke: ops_run `eeccd98a-…` succeeded via claim loop (manual trigger)
|
||||
- Prior scheduled claim success: Aug 4 `13fcbed2-…`
|
||||
- Disabled user timers: `binky-rhythm-daily`, `binky-rhythm-mail`, `binky-rhythm-review`
|
||||
(units remain on disk for break-glass)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue