Escalate a tunnel failure that retrying cannot fix

hookdeck/tunnel.py

@@ -32,6 +32,48 @@
 # Generous enough that CLI output never kills a working tunnel.
 _STDOUT_LINE_LIMIT = 1024 * 1024
 
+# How many consecutive too-short runs before the restart loop is treated as a
+# standing failure rather than a blip. Two would fire on an ordinary flap; this
+# is small enough to be prompt and large enough not to cry wolf.
+_FAST_FAILURES_BEFORE_ESCALATING = 3
+
+# How much of the CLI's output to keep for diagnosis. The useful line is the
+# last one before it exits.
+_OUTPUT_MEMORY = 12
+
+#: Substrings the CLI prints for failures that retrying cannot fix, paired with
+#: what to do about them. The backoff exists for network blips; these need a
+#: person, and without this they scroll past at warning level for ever.
+_DETERMINISTIC_FAILURES: tuple[tuple[str, str], ...] = (
+    (
+        "no connection found matching filter",
+        "the CLI is forwarding from a different Hookdeck project than the API "
+        "key manages, or `hermes hookdeck setup` has not been run for this "
+        "route. Run `hermes hookdeck doctor` — it compares the two.",
+    ),
+    (
+        "automatically creating source",
+        "the source does not exist in the project the CLI is forwarding from, "
+        "so the CLI has just created a stray one there. That is usually a "
+        "different project than the API key manages. Run `hermes hookdeck "
+        "doctor`.",
+    ),
+    (
+        "authentication failed",
+        "the CLI session is invalid or expired. If the gateway owns its "
+        "config, delete it and restart; otherwise run `hookdeck login`.",
+    ),
+)
+
+
+def diagnose(lines: list[str]) -> str:
+    """A likely cause for the CLI's exit, or "" if none is recognised."""
+    haystack = "\n".join(lines).lower()
+    for marker, explanation in _DETERMINISTIC_FAILURES:
+        if marker in haystack:
+            return explanation
+    return ""
+
 
 def _device_name() -> str:
     """How this gateway's CLI sessions identify themselves to Hookdeck.
@@ -89,6 +131,9 @@ def __init__(
         self._process: asyncio.subprocess.Process | None = None
         self._supervisor: asyncio.Task | None = None
         self._stopping = False
+        #: Tail of the last run's output, kept so a failing restart loop can
+        #: say *why* rather than only that it is looping.
+        self._recent_output: list[str] = []
 
     # ------------------------------------------------------------------
     # Command construction
@@ -222,6 +267,8 @@ async def authenticate(self) -> bool:
 
     async def _supervise(self, binary: str) -> None:
         backoff = _BACKOFF_INITIAL
+        fast_failures = 0
+        escalated = False
         while not self._stopping:
             started = asyncio.get_running_loop().time()
             try:
@@ -237,6 +284,29 @@ async def _supervise(self, binary: str) -> None:
             ran_for = asyncio.get_running_loop().time() - started
             if ran_for >= _HEALTHY_RUN_SECONDS:
                 backoff = _BACKOFF_INITIAL
+                fast_failures = 0
+                escalated = False
+            else:
+                fast_failures += 1
+
+            # A tunnel that never stays up is not retrying its way out of
+            # anything, and at warning level it reads as routine churn — which
+            # is how "the gateway is receiving nothing" stays invisible. Say so
+            # once, loudly, with the cause when the CLI named one, then fall
+            # back to the quiet line so the log does not fill up.
+            if fast_failures >= _FAST_FAILURES_BEFORE_ESCALATING and not escalated:
+                escalated = True
+                cause = diagnose(self._recent_output)
+                logger.error(
+                    "[hookdeck] CLI tunnel has failed to stay up %d times in a "
+                    "row (last run %.0fs). No events are reaching the gateway. "
+                    "%sLast output: %s",
+                    fast_failures,
+                    ran_for,
+                    f"Likely cause: {cause} " if cause else "",
+                    " / ".join(self._recent_output[-3:]) or "(none)",
+                )
+
             logger.warning(
                 "[hookdeck] CLI tunnel exited after %.0fs — restarting in %.0fs",
                 ran_for,
@@ -259,11 +329,17 @@ async def _run_once(self, binary: str) -> None:
             limit=_STDOUT_LINE_LIMIT,
             env={**os.environ, **({CLI_API_KEY_ENV: self._api_key} if self._api_key else {})},
         )
+        # Reset per run: the diagnosis is about why *this* run ended, and
+        # carrying lines over from the previous one would name a cause that has
+        # since been fixed.
+        self._recent_output = []
         assert self._process.stdout is not None
         async for line in self._process.stdout:
             text = line.decode("utf-8", "replace").rstrip()
             if text:
                 logger.info("[hookdeck cli] %s", text)
+                self._recent_output.append(text)
+                del self._recent_output[:-_OUTPUT_MEMORY]
         await self._process.wait()
 
     async def _terminate(self) -> None:

tests/test_tunnel.py

@@ -446,3 +446,133 @@ async def test_the_session_names_itself_so_it_is_findable_in_the_dashboard(
     argv = spawned.commands[0][0]
     assert argv[argv.index("--device-name") + 1].startswith("hermes-")
     assert argv[argv.index("--name") + 1] == "hermes-gateway"
+
+
+# ----------------------------------------------------------------------
+# Escalating a failure that retrying cannot fix (#4)
+# ----------------------------------------------------------------------
+
+#: The output observed in #4, verbatim. The project the CLI forwards from has
+#: no such source, so it invents one and then finds no connection for it.
+PROJECT_MISMATCH_OUTPUT = [
+    b'Source "hermes-livetest" not found.\n',
+    b'Non-interactive mode detected. Automatically creating source "hermes-livetest".\n',
+    b'no connection found matching filter "livetest" for source "hermes-livetest"\n',
+]
+
+
+def test_a_project_mismatch_is_recognised_from_the_cli_output():
+    cause = tunnel_mod.diagnose(
+        [line.decode().strip() for line in PROJECT_MISMATCH_OUTPUT]
+    )
+    assert "different Hookdeck project" in cause
+    assert "hermes hookdeck doctor" in cause
+
+
+def test_an_expired_session_is_recognised():
+    assert "invalid or expired" in tunnel_mod.diagnose(
+        ["Authentication failed: your API key is invalid or expired."]
+    )
+
+
+def test_output_with_no_known_cause_diagnoses_nothing():
+    # Better to say only what is known than to guess a cause and send an
+    # operator after the wrong thing.
+    assert tunnel_mod.diagnose(["some unrecognised failure"]) == ""
+    assert tunnel_mod.diagnose([]) == ""
+
+
+async def test_a_tunnel_stuck_in_a_restart_loop_escalates_and_names_the_cause(
+    spawned, no_waiting, caplog
+):
+    # #4: the gateway logs that it is listening, the tunnel fails identically
+    # every couple of seconds at warning level, and nothing ever says the
+    # gateway is receiving no events.
+    for _ in range(10):
+        spawned.queue.append(FakeProcess(lines=list(PROJECT_MISMATCH_OUTPUT)))
+
+    tunnel = HookdeckTunnel(port=1, path="/x", source="hermes-livetest")
+    with caplog.at_level(logging.INFO):
+        await _supervise_rounds(tunnel, 5, no_waiting)
+
+    errors = [r.getMessage() for r in caplog.records if r.levelno >= logging.ERROR]
+    assert len(errors) == 1, "escalate once per streak, not on every restart"
+    assert "No events are reaching the gateway" in errors[0]
+    assert "different Hookdeck project" in errors[0]
+    # The operator needs the CLI's own words too, not just our interpretation.
+    assert "no connection found matching filter" in errors[0]
+
+
+async def test_escalation_waits_for_a_streak_rather_than_one_bad_start(
+    spawned, no_waiting, caplog
+):
+    # A single fast exit happens on an ordinary flap. Crying wolf there would
+    # make the error level meaningless for the case that matters.
+    for _ in range(10):
+        spawned.queue.append(FakeProcess(lines=list(PROJECT_MISMATCH_OUTPUT)))
+
+    tunnel = HookdeckTunnel(port=1, path="/x", source="s")
+    await _supervise_rounds(
+        tunnel, tunnel_mod._FAST_FAILURES_BEFORE_ESCALATING - 1, no_waiting
+    )
+
+    assert [r for r in caplog.records if r.levelno >= logging.ERROR] == []
+
+
+async def test_a_healthy_run_rearms_the_escalation(
+    spawned, no_waiting, caplog, monkeypatch
+):
+    # A tunnel that recovers and later breaks again deserves to be shouted
+    # about again — otherwise the second outage is silent.
+    clock = {"now": 0.0}
+
+    class FakeLoop:
+        def time(self):
+            return clock["now"]
+
+    monkeypatch.setattr(tunnel_mod.asyncio, "get_running_loop", lambda: FakeLoop())
+    for _ in range(20):
+        spawned.queue.append(FakeProcess(lines=list(PROJECT_MISMATCH_OUTPUT)))
+
+    tunnel = HookdeckTunnel(port=1, path="/x", source="s")
+    runs = {"n": 0}
+    original = tunnel._run_once
+
+    async def timed_run(binary):
+        runs["n"] += 1
+        # Three fast failures, one healthy session, then three more failures.
+        healthy = runs["n"] == 4
+        clock["now"] += tunnel_mod._HEALTHY_RUN_SECONDS * 2 if healthy else 1.0
+        await original(binary)
+
+    tunnel._run_once = timed_run
+    await _supervise_rounds(tunnel, 7, no_waiting)
+
+    errors = [r for r in caplog.records if r.levelno >= logging.ERROR]
+    assert len(errors) == 2, "once before the recovery, once after it breaks again"
+
+
+async def test_the_diagnosis_does_not_outlive_the_run_that_produced_it(spawned):
+    # Carrying output across runs would name a cause that has since been
+    # fixed — the worst kind of wrong, because it looks specific.
+    tunnel = HookdeckTunnel(port=1, path="/x", source="s")
+    spawned.queue.append(FakeProcess(lines=list(PROJECT_MISMATCH_OUTPUT)))
+    await tunnel._run_once("/usr/bin/hookdeck")
+    assert tunnel_mod.diagnose(tunnel._recent_output)
+
+    spawned.queue.append(FakeProcess(lines=[b"Ready!\n"]))
+    await tunnel._run_once("/usr/bin/hookdeck")
+    assert tunnel._recent_output == ["Ready!"]
+    assert tunnel_mod.diagnose(tunnel._recent_output) == ""
+
+
+async def test_the_output_buffer_is_bounded(spawned):
+    # A chatty tunnel runs for days; the tail is what diagnoses, not the whole
+    # session.
+    tunnel = HookdeckTunnel(port=1, path="/x", source="s")
+    spawned.queue.append(
+        FakeProcess(lines=[f"line {i}\n".encode() for i in range(500)])
+    )
+    await tunnel._run_once("/usr/bin/hookdeck")
+    assert len(tunnel._recent_output) == tunnel_mod._OUTPUT_MEMORY
+    assert tunnel._recent_output[-1] == "line 499"