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/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/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_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" 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):