diff --git a/hookdeck/tunnel.py b/hookdeck/tunnel.py index cc10cb4..537be3e 100644 --- a/hookdeck/tunnel.py +++ b/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: diff --git a/tests/test_tunnel.py b/tests/test_tunnel.py index a94c2d7..cf3d18f 100644 --- a/tests/test_tunnel.py +++ b/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"