Wrap bulk retry's filters in the query object the API expects

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

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

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:

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),
         [],
     ),

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"

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

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):