From 21cfd1e9052074b8b931751e61b6906fbb18cdf5 Mon Sep 17 00:00:00 2001 From: garethx Date: Wed, 12 Aug 2026 14:00:58 +0100 Subject: [PATCH 1/2] Wrap bulk retry's filters in the query object the API expects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `POST /bulk/events/retry` takes its filters inside a `query` object. We sent them at the top level, so bulk retry has never worked — not from `hookdeck_bulk_retry`, and not from `hermes hookdeck retry --failed`, which goes through the same client method. Found by asking the agent to triage real failed events. It reasoned correctly, called the tool, and reported back that Hookdeck was down: Retry Attempt: ❌ Hookdeck API is currently down. The retry API endpoint (POST /bulk/events/retry) is returning HTTP 500 FATAL_ERROR. It was not down. The API answers a malformed body with 500 FATAL_ERROR rather than 400, so a wrong shape is indistinguishable from an outage — which is how this survived, and why the wrapper now lives in the client rather than in each caller. Every existing test fakes the client and asserts the dict handed to `bulk_retry_events`, so none of them could see the wire. The new one drives a real `httpx.MockTransport` and pins the encoded body. Verified against the live API: the flat shape 500s, the wrapped shape returns a batch (`estimated_count: 3`), and the agent now completes the same request it previously reported as an outage. Co-Authored-By: Claude Opus 5 --- hookdeck/api.py | 11 ++++++++++- tests/test_api.py | 23 +++++++++++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/hookdeck/api.py b/hookdeck/api.py index 93a4ef0..c68cec4 100644 --- a/hookdeck/api.py +++ b/hookdeck/api.py @@ -198,7 +198,16 @@ async def retry_event(self, event_id: str) -> Any: return await self.request("POST", f"/events/{event_id}/retry") async def bulk_retry_events(self, query: Mapping[str, Any]) -> Any: - return await self.request("POST", "/bulk/events/retry", json=dict(query)) + """POST /bulk/events/retry — redeliver every event matching *query*. + + The filters go inside a ``query`` object. Sent at the top level the API + answers 500 ``FATAL_ERROR`` rather than a 400, so a wrong shape reads + as Hookdeck being down — which is exactly how this went unnoticed, and + why the wrapper is applied here rather than left to each caller. + """ + return await self.request( + "POST", "/bulk/events/retry", json={"query": dict(query)} + ) # ------------------------------------------------------------------ # Observability diff --git a/tests/test_api.py b/tests/test_api.py index 05c1b04..54dfc40 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -110,3 +110,26 @@ def handler(request: httpx.Request) -> httpx.Response: api = HookdeckAPI("key", client=_client(handler)) with pytest.raises(HookdeckAPIError): await api.list_events() + + +async def test_bulk_retry_wraps_its_filters_in_a_query_object(): + """The filters belong inside `query`, not at the top level. + + Sent flat, the API answers 500 FATAL_ERROR rather than 400 — so a wrong + shape is indistinguishable from Hookdeck being down, and the agent + reported exactly that. Every other test fakes the client and asserts the + dict handed to this method, which is why none of them saw the wire. + """ + import json as _json + + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["body"] = _json.loads(request.content) + return httpx.Response(200, json={"id": "bch_1", "estimated_count": 3}) + + api = HookdeckAPI("key", client=_client(handler)) + await api.bulk_retry_events({"status": "FAILED", "webhook_id": "web_1"}) + + assert seen["body"] == {"query": {"status": "FAILED", "webhook_id": "web_1"}} + assert "status" not in seen["body"], "filters must not sit at the top level" From 54347d20b7cd3b6ab59122366b9c0cf275e35870 Mon Sep 17 00:00:00 2001 From: garethx Date: Wed, 12 Aug 2026 14:24:51 +0100 Subject: [PATCH 2/2] Scope bulk retry to this gateway, and report burst headroom MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the query-shape fix, plus what it exposed once the call started working. **Scope.** `{"status": "FAILED"}` matches every failed event in the project, and a project usually holds connections belonging to something else. The tool takes no required arguments, so an agent can call it bare and redeliver somebody else's traffic. It now resolves the configured routes to connection ids and filters on them — the same definition of "ours" the dashboard uses to decide what it may pause — and refuses rather than widening when it owns nothing. An explicit `connection_id` still overrides. **Empty matches.** The API answers a batch that would match nothing with 422, which surfaced as an API error. On a healthy gateway that is the common case, and "nothing to retry" is an answer, not a fault. **The `since` path was never exercised.** The fix was verified with `{status, webhook_id}` only. Checked against the live API: the nested `created_at: {gte: …}` shape parses (it 422s on an empty match rather than 500ing on a bad body), so it was correct — but it was correct untested. **Burst headroom.** A deferred event only gets back in on a retry, so a burst drains at `max_concurrent` per round against the rule's `count` rounds. Their product is the burst that survives, and past it the tail exhausts its retries while waiting — 6 events against `max_concurrent: 1` and a `count: 5` connection lost 2. `doctor` now reports the figure per connection from the rule really on it, and reliability.md explains the sizing. The default stays 2: it pairs with the provisioned count of 10 for ~20, and raising it is a spending decision as much as a throughput one. Co-Authored-By: Claude Opus 5 --- docs/reliability.md | 32 ++++++++++++++++++++ hookdeck/cli.py | 44 +++++++++++++++++++++++++-- hookdeck/tools.py | 50 +++++++++++++++++++++++++++---- tests/test_cli.py | 45 ++++++++++++++++++++++++++++ tests/test_tools.py | 73 +++++++++++++++++++++++++++++++++++++++------ 5 files changed, 227 insertions(+), 17 deletions(-) diff --git a/docs/reliability.md b/docs/reliability.md index 6e2c5ee..c5643ec 100644 --- a/docs/reliability.md +++ b/docs/reliability.md @@ -48,6 +48,38 @@ limit deliveries; only the adapter can see runs. CLI destinations have no `rate_limit` field at all, so in the default transport it is not on the table either way. +**Size it against the retry count, or a big enough burst loses its tail.** A +deferred event is not held anywhere — it gets back in only when Hookdeck +retries it. So a simultaneous burst drains at roughly `max_concurrent` events +per retry round, and each event has the connection rule's `count` rounds before +Hookdeck gives up on it. The product is the burst that survives: + +| `max_concurrent` | retry `count` | burst absorbed | +|---|---|---| +| 2 (default) | 10 (what `setup` provisions) | ~20 | +| 1 | 5 | ~5 | + +Past that, the tail exhausts its retries while still waiting for a slot, and +those events end `FAILED`. Measured, not theorised: a burst of 6 against +`max_concurrent: 1` and a `count: 5` connection lost 2. + +`hermes hookdeck doctor` reports the figure per connection, using the rule +that is really on it rather than the default: + +``` +✓ connection 'github' absorbs a burst of about 20 events (max_concurrent 2 x 10 retries) +``` + +Raise `max_concurrent` to spend more on parallel runs, or the rule's `count` to +wait longer. Two runs is the default because agent runs cost money per +execution, which is a different economy from a webhook handler — it is a +spending decision as much as a throughput one. + +`defer_attempt_limit` (default 2) keeps a *sustained* overload from burning +the budget faster still: past that many deferrals of one event the `Retry-After` +hint is dropped, so Hookdeck falls back to exponential backoff instead of +returning every few seconds. + In `sync` mode this inverts: the delivery stays open for the run's duration, so a destination-level `--rate-limit N --rate-limit-period concurrent` genuinely does cap concurrent runs, and does it better — Hookdeck holds the event without diff --git a/hookdeck/cli.py b/hookdeck/cli.py index a730827..f0d670c 100644 --- a/hookdeck/cli.py +++ b/hookdeck/cli.py @@ -21,6 +21,7 @@ from .constants import ( API_KEY_ENV, CLI_API_KEY_ENV, + DEFAULT_MAX_CONCURRENT, DEFAULT_PATH, DEFAULT_PORT, MODE_ENV, @@ -505,6 +506,44 @@ async def _go() -> str: return "" +def _burst_headroom(connection: dict, retry_rule: dict, extra: dict) -> Check: + """How large a burst this connection absorbs before events start dying. + + An event deferred with 503 gets back in only on a later retry attempt, so + the queue drains at roughly `max_concurrent` events per round and each + event has `count` rounds before Hookdeck gives up. Their product is the + burst that survives; past it the tail exhausts its retries while waiting. + + Reported rather than judged: the number that matters is the burst this + gateway actually sees, and only the operator knows that. + """ + concurrent = int(extra.get("max_concurrent", DEFAULT_MAX_CONCURRENT) or 0) + count = int(retry_rule.get("count") or 0) + name = connection.get("name") + + if not concurrent: + return Check( + True, + f"connection '{name}': max_concurrent is unlimited, so nothing is " + "deferred for capacity", + ) + if not count: + return Check( + True, f"connection '{name}': retry rule has no count to reason about" + ) + return Check( + True, + f"connection '{name}' absorbs a burst of about {concurrent * count} " + f"events (max_concurrent {concurrent} x {count} retries)", + note=( + "A larger simultaneous burst drains at max_concurrent per retry " + "round, and the tail runs out of retries before it is admitted. " + "Raise max_concurrent to spend more on parallel runs, or the " + "rule's count to wait longer." + ), + ) + + def _check_cli_project(extra: dict) -> Check: """The two projects in play must be the same one. @@ -635,7 +674,7 @@ def _report_stranded_runs() -> None: ledger.close() -async def _check_live_connections(routes: dict) -> list[Check]: +async def _check_live_connections(routes: dict, extra: dict) -> list[Check]: """Reachability, plus whether each retry rule covers what the adapter emits. A rule narrower than the emitted statuses is silent data loss — a deferred @@ -670,6 +709,7 @@ async def _check_live_connections(routes: dict) -> list[Check]: "never come back. Re-run `hermes hookdeck setup`.", ) ) + checks.append(_burst_headroom(connection, rule, extra)) checks.append(Check(True, "Hookdeck API reachable and the key is accepted")) return checks @@ -706,7 +746,7 @@ def _cmd_doctor(_args: argparse.Namespace) -> int: if api_key(): print() try: - live = run_sync(_check_live_connections(routes)) + live = run_sync(_check_live_connections(routes, extra)) except HookdeckAPIError as exc: live = [Check(False, f"Hookdeck API check failed: {exc}")] for check in live: diff --git a/hookdeck/tools.py b/hookdeck/tools.py index d6e9618..c8fc7a2 100644 --- a/hookdeck/tools.py +++ b/hookdeck/tools.py @@ -226,21 +226,58 @@ async def _go() -> str: @_guard def hookdeck_bulk_retry(args: dict) -> str: - """Retry every failed event, optionally scoped by time or connection.""" + """Retry failed events for this gateway's connections.""" query: dict[str, Any] = {"status": "FAILED"} if args.get("since"): query["created_at"] = {"gte": args["since"]} - if args.get("connection_id"): - query["webhook_id"] = args["connection_id"] async def _go() -> str: async with HookdeckAPI() as api: - result = await api.bulk_retry_events(query) + if args.get("connection_id"): + query["webhook_id"] = args["connection_id"] + else: + # Scoped to the routes this gateway serves. `status: FAILED` + # alone matches the whole project, and a project usually holds + # connections belonging to something else — redelivering their + # traffic is not this tool's to do, and an agent can call it + # with no arguments at all. + owned = await _owned_connection_ids(api) + if not owned: + return ( + "No connection matches a configured route, so there is " + "nothing this gateway owns to retry. Pass connection_id " + "to act on a specific connection." + ) + query["webhook_id"] = owned + try: + result = await api.bulk_retry_events(query) + except HookdeckAPIError as exc: + if exc.status == 422 and "does not include any events" in exc.body: + # A normal answer, not a fault: the API refuses a batch that + # would match nothing. + return "No failed events matched — nothing to retry." + raise return f"Bulk retry queued: {json.dumps(result)[:1000]}" return _run(_go()) +async def _owned_connection_ids(api: Any) -> list[str]: + """Connection ids for the routes this gateway is configured to serve. + + Resolved by name, one request per route, the same way the dashboard decides + which connections it may pause — so both surfaces agree on what "ours" + means rather than each inventing it. + """ + from .settings import platform_extra + + ids: list[str] = [] + for route_name in platform_extra().get("routes") or {}: + found = await api.list_connections(name=route_name, limit=10) + ids += [c["id"] for c in _models(found) if c.get("id")] + return ids + + #: Longest an agent may pause a connection for. Pausing is safe — events are #: held rather than dropped — but only until someone resumes, and an agent that #: pauses and then fails leaves the queue growing with nobody watching. @@ -360,8 +397,9 @@ def _schema(name: str, description: str, properties: dict, required: list[str]) ), "hookdeck_bulk_retry": _schema( "hookdeck_bulk_retry", - "Retry every failed event, optionally scoped by time or connection. " - "Prefer retrying individually unless the failures share one cause.", + "Retry failed events, scoped to this gateway's own connections unless " + "you name connection_id. Prefer retrying individually unless the " + "failures share one cause.", dict(_SINCE_PROP), [], ), diff --git a/tests/test_cli.py b/tests/test_cli.py index ae61fa4..413d71c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -617,3 +617,48 @@ def test_doctor_reports_an_unreachable_api_as_a_failed_check( _configure(doctor_env, routes={"github": {}}) assert cli.hookdeck_command(_ns("doctor")) == 1 assert "Hookdeck API check failed" in capsys.readouterr().out + + +# ── doctor: burst headroom ────────────────────────────────────────── + + +def test_doctor_reports_how_large_a_burst_survives(doctor_env, fake_api, monkeypatch, capsys): + # An event deferred with 503 only gets back in on a retry, so the queue + # drains at max_concurrent per round and each event has `count` rounds. + # Their product is the burst that survives — a real loss we measured. + from hookdeck.provision import retryable_status_codes + + monkeypatch.setenv("HOOKDECK_API_KEY", "key") + doctor_env.setattr(cli.shutil, "which", lambda _b: "/usr/local/bin/hookdeck") + doctor_env.setattr(cli, "_cli_version", lambda _b: "2.4.0") + doctor_env.setattr(cli, "_other_hookdeck_binaries", lambda _r: []) + fake_api.responses["list_connections"] = { + "models": [{ + "name": "github", "team_id": "tm_1", + "rules": [{"type": "retry", "count": 10, + "response_status_codes": retryable_status_codes()}], + }] + } + _configure(doctor_env, secret="s", max_concurrent=3, + cli_config_path="", routes={"github": {}}) + + cli.hookdeck_command(_ns("doctor")) + out = capsys.readouterr().out + assert "absorbs a burst of about 30 events" in out + assert "max_concurrent 3 x 10 retries" in out + + +def test_unlimited_concurrency_defers_nothing(doctor_env, fake_api, monkeypatch, capsys): + monkeypatch.setenv("HOOKDECK_API_KEY", "key") + doctor_env.setattr(cli.shutil, "which", lambda _b: "/usr/local/bin/hookdeck") + doctor_env.setattr(cli, "_cli_version", lambda _b: "2.4.0") + doctor_env.setattr(cli, "_other_hookdeck_binaries", lambda _r: []) + fake_api.responses["list_connections"] = { + "models": [{"name": "github", "team_id": "tm_1", + "rules": [{"type": "retry", "count": 10}]}] + } + _configure(doctor_env, secret="s", max_concurrent=0, + cli_config_path="", routes={"github": {}}) + + cli.hookdeck_command(_ns("doctor")) + assert "nothing is deferred for capacity" in capsys.readouterr().out diff --git a/tests/test_tools.py b/tests/test_tools.py index 4f5f1a9..2943d8d 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -31,6 +31,9 @@ class FakeAPI: responses: ClassVar[dict] = {} calls: ClassVar[list] = [] raises: ClassVar[Exception | None] = None + #: Raise only for a named method, so a test can fail one call in a + #: sequence rather than all of them. + raises_for: ClassVar[dict] = {} def __init__(self, *_args, **_kwargs) -> None: pass @@ -45,6 +48,9 @@ async def _record(self, _name: str, /, *args, **kwargs): # Positional-only: callers pass through their own kwargs, and # `list_connections(name=…)` would otherwise collide with this one. type(self).calls.append((_name, args, kwargs)) + per_method = type(self).raises_for.get(_name) + if per_method is not None: + raise per_method if type(self).raises is not None: raise type(self).raises return type(self).responses.get(_name, {}) @@ -82,14 +88,12 @@ async def unpause_connection(self, connection_id): @pytest.fixture() def api(monkeypatch): - FakeAPI.responses = {} - FakeAPI.calls = [] - FakeAPI.raises = None + FakeAPI.responses, FakeAPI.calls = {}, [] + FakeAPI.raises, FakeAPI.raises_for = None, {} monkeypatch.setattr(tools, "HookdeckAPI", FakeAPI) yield FakeAPI - FakeAPI.responses = {} - FakeAPI.calls = [] - FakeAPI.raises = None + FakeAPI.responses, FakeAPI.calls = {}, [] + FakeAPI.raises, FakeAPI.raises_for = None, {} @pytest.fixture() @@ -420,10 +424,61 @@ def test_retrying_one_event_names_it_back(api): assert calls_named(api, "retry_event") == [("retry_event", ("evt_1",), {})] -def test_an_unscoped_bulk_retry_is_still_scoped_to_failures(api): - # Without the status filter this would replay successful events too. +def test_bulk_retry_is_scoped_to_this_gateways_connections(api, ledger_at, monkeypatch): + # `status: FAILED` alone matches the whole project, and a project usually + # holds connections belonging to something else. An agent can call this + # with no arguments, so the default must not redeliver their traffic. + monkeypatch.setattr( + "hookdeck.settings.load_hermes_config", + lambda: {"gateway": {"platforms": {"hookdeck": {"extra": { + "routes": {"github": {}, "stripe": {}}}}}}}, + ) + api.responses["list_connections"] = {"models": [{"id": "web_mine"}]} call("hookdeck_bulk_retry") - assert calls_named(api, "bulk_retry_events")[0][1][0] == {"status": "FAILED"} + + sent = calls_named(api, "bulk_retry_events")[0][1][0] + assert sent["status"] == "FAILED" + assert sent["webhook_id"] == ["web_mine", "web_mine"] # one per route + # Resolved by name, the same way the dashboard decides what it may pause. + assert [c[2]["name"] for c in calls_named(api, "list_connections")] == [ + "github", "stripe" + ] + + +def test_bulk_retry_refuses_rather_than_widening_when_it_owns_nothing( + api, ledger_at, monkeypatch +): + # The dangerous fallback would be "no connections resolved, so retry + # everything". Refuse and say so instead. + monkeypatch.setattr( + "hookdeck.settings.load_hermes_config", + lambda: {"gateway": {"platforms": {"hookdeck": {"extra": {"routes": {}}}}}}, + ) + message = call("hookdeck_bulk_retry") + assert "nothing this gateway owns" in message + assert not calls_named(api, "bulk_retry_events") + + +def test_an_explicit_connection_overrides_the_default_scope(api, ledger_at): + call("hookdeck_bulk_retry", {"connection_id": "web_explicit"}) + sent = calls_named(api, "bulk_retry_events")[0][1][0] + assert sent["webhook_id"] == "web_explicit" + assert not calls_named(api, "list_connections"), "no need to resolve ours" + + +def test_matching_no_events_reads_as_an_answer_not_a_failure(api, ledger_at): + # The API refuses an empty batch with 422. Surfacing that as an API error + # tells the model something is broken when the true answer is "nothing to + # do" — and it is the common case on a healthy gateway. + from hookdeck.api import HookdeckAPIError + + api.responses["list_connections"] = {"models": [{"id": "web_1"}]} + api.raises_for = {"bulk_retry_events": HookdeckAPIError( + 422, "POST", "/bulk/events/retry", + "The query filter for the batch operations does not include any events.", + )} + message = call("hookdeck_bulk_retry", {"connection_id": "web_1"}) + assert message == "No failed events matched — nothing to retry." def test_a_bulk_retry_carries_its_scope_into_the_query(api):