diff --git a/.agents/notes/implemented/2026-10-26-a-bot-by-its-name-a-door-that-answers-its-own-failure-an-alarm-that-tries-again.md b/.agents/notes/implemented/2026-10-26-a-bot-by-its-name-a-door-that-answers-its-own-failure-an-alarm-that-tries-again.md new file mode 100644 index 00000000..69525247 --- /dev/null +++ b/.agents/notes/implemented/2026-10-26-a-bot-by-its-name-a-door-that-answers-its-own-failure-an-alarm-that-tries-again.md @@ -0,0 +1,65 @@ +--- +title: A bot by its name, a door that answers its own failure, an alarm that tries again, and a rollback that says what it does +status: implemented +date: 2026-10-26 +scope: stack +--- + +## Decision + +A second deployment running copies of these files met four things in a week +that hold here too. + +1. **The watch scanner recognizes a bot by its name.** `WATCH_BOTS` was the only + way to drop a bot, so a bot nobody listed reached the digest as a message to + answer. `BOT_NAME` drops a sender whose display name has bot or robot as a + word of its own, a CamelCase `…Bot`, or 机器人; bot inside a word (Talbot, + Abbott) is a person. The list stays, for bots whose names say nothing. +2. **A door answers and records a failure nobody planned for.** An exception a + sidecar did not expect ended the handler thread: the caller waited out its + own timeout, the ledger had no line, and the traceback went to stderr with + no time. The shared HTTP handler now answers 502 and calls the door's own + ledger hook (`call.failed` at the gate, `signal.failed` at the signer); the + egress proxy refuses with 502 into its refusal ledger unless the tunnel was + already spliced; every server logs an escaped error through the door's + logger, with the peer. The proxy also writes the request line it forwards + back as Latin-1, the way it read it: UTF-8 turned every raw non-ASCII byte in + a path into two. +3. **An alarm tries a connection that did not happen again, in place.** The + investigator's, the pipe's and the judge's alarms posted once, so a moment + without a resolver dropped the page and only a later failure sent another. + Two more attempts, two seconds apart; an HTTP answer is the channel speaking + and is not retried; a timeout is, at the price of a possible duplicate. +4. **The releasing page says what a rollback does with newer rows,** and a + release that changes a schema says it in its tag note. Read from the code for + 0.8.0 back to 0.7.1: nothing fails to open; a silence scoped to one alert + quiets its whole source until it expires. + +Fixed while verifying: the forwarder's bind test, changed while clearing a +CodeQL alert, clashed with a loopback socket, and a loopback socket does not +stop a wildcard bind with SO_REUSEADDR on macOS — the local gate was red on a +Mac while CI was green. It now clashes two forwarder binds, which fails on both. + +## Why + +Each was observed there, not reasoned: an unlisted bot became a to-do twice; a +query string with raw non-ASCII bytes killed a handler thread with no answer and +no ledger line; ten resolver misses in eleven days on the same kind of container +network, every one answered by a resend half a minute later; and a schema change +whose rollback nobody had written down. The code here had the same shapes. Each +fix has a test that fails without it. + +## Consequences + +- A person whose display name has bot as a word of its own is dropped from the + watch digest. Nobody's is today; if it ever happens, the pattern is the place + to change, not the list. +- `call.failed` rows can carry `reason: unexpected ` with a method and + path instead of a client and tools; `signal.failed` is a new signer event; + `egress.refused` rows can carry code 502 and `unexpected …`. The readers that + filter on `signal.signed` or count refusals are unaffected. +- An alarm whose first attempts miss arrives a few seconds later, and rarely + twice. +- The watch and sidecar changes reach a deployment at its next build from the + checkout; the alarm changes in the three services reach the published images + at the next release. diff --git a/deploy/docker-compose.work.yml b/deploy/docker-compose.work.yml index a300275b..c985f291 100644 --- a/deploy/docker-compose.work.yml +++ b/deploy/docker-compose.work.yml @@ -1011,6 +1011,9 @@ services: WATCH_DOW: ${WATCH_DOW:-1-5} WATCH_EXCLUDE: ${WATCH_EXCLUDE:-} WATCH_ME: ${WATCH_ME:-} + # Only the bots whose names do not give them away: a sender whose display + # name has bot or robot as a word of its own, a CamelCase ...Bot, or 机器人 + # is dropped without being listed (watch_scan.py, BOT_NAME). WATCH_BOTS: ${WATCH_BOTS:-} WATCH_JIRA_SCRIPT: /watch/jira_watch.py # The digest's byte budget, and it must stay well under diff --git a/deploy/sidecars/common.py b/deploy/sidecars/common.py index 402723e0..05a1f37c 100644 --- a/deploy/sidecars/common.py +++ b/deploy/sidecars/common.py @@ -29,7 +29,7 @@ import socketserver import threading import time -from http.server import BaseHTTPRequestHandler +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any @@ -105,11 +105,27 @@ def pump(a: socket.socket, b: socket.socket, idle_seconds: float) -> None: return -class Server(socketserver.ThreadingTCPServer): +class LoggedErrors: + """socketserver's own handle_error prints a bare traceback to stderr: no + time, no door, no peer. A handler that raised then left a line nobody could + place next to the ledger. This sends it through the door's logger instead, + which has the time, and names the peer.""" + + def handle_error(self, request: Any, client_address: Any) -> None: # noqa: ARG002 — socketserver's signature + handler = getattr(self, "RequestHandlerClass", None) + name = getattr(handler, "server_version", "") or getattr(handler, "log_name", "") or "sidecar" + logging.getLogger(name).error("unhandled error serving %s", client_address, exc_info=True) + + +class Server(LoggedErrors, socketserver.ThreadingTCPServer): allow_reuse_address = True daemon_threads = True +class HttpServer(LoggedErrors, ThreadingHTTPServer): + """The HTTP doors' server (the gate, the signer), with the same error log.""" + + _CONTROL_CHARS = getattr(BaseHTTPRequestHandler, "_control_char_table", None) @@ -143,7 +159,40 @@ def handle_one_request(self) -> None: # Waiting for a request line: a timeout from here until one arrives is the # connection idling, not a request going wrong (see log_error) self._between_requests = True - super().handle_one_request() + self._answer_started = False + try: + super().handle_one_request() + except (ConnectionResetError, BrokenPipeError): + raise # nobody to answer; handle() lets these go quietly + except Exception as exc: # noqa: BLE001 — a request that fails is still answered and recorded + # Anything a door did not plan for used to end the thread: no answer, + # so the caller waited out its own timeout, and no ledger line, so the + # record of what was asked had a hole exactly where something odd was + # asked. A second deployment running copies of these files found it + # with a query string carrying raw non-ASCII bytes. + self.close_connection = True + logging.getLogger(self.server_version).error( + "unexpected failure on %s %s", + getattr(self, "command", None) or "-", + plain(getattr(self, "path", "") or "")[:80], + exc_info=True, + ) + with contextlib.suppress(Exception): + self.unexpected(exc) + if not self._answer_started: + with contextlib.suppress(OSError): + body = json.dumps({"error": f"this door failed on the request: {type(exc).__name__}"}) + self._answer(502, body.encode()) + self.wfile.flush() + + def send_response(self, code: int, message: str | None = None) -> None: + self._answer_started = True + super().send_response(code, message) + + def unexpected(self, exc: BaseException) -> None: + """The door's own ledger line for a request that failed in a way nobody + planned. A door with a ledger overrides this; the log line is written + either way.""" def parse_request(self) -> bool: self._between_requests = False diff --git a/deploy/sidecars/gate.py b/deploy/sidecars/gate.py index ab6e5235..10879700 100644 --- a/deploy/sidecars/gate.py +++ b/deploy/sidecars/gate.py @@ -78,10 +78,9 @@ import sys import urllib.error import urllib.request -from http.server import ThreadingHTTPServer from typing import Any -from common import HttpHandler, Ledger, constant_time_eq +from common import HttpHandler, HttpServer, Ledger, constant_time_eq, plain logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") logger = logging.getLogger("mcp-gate") @@ -319,6 +318,14 @@ class Gate(HttpHandler): ledger = Ledger("") # ── who is asking ── + def unexpected(self, exc: BaseException) -> None: + self.ledger.write( + event="call.failed", + method=getattr(self, "command", None) or "-", + path=plain(getattr(self, "path", "") or "")[:80], + reason=f"unexpected {type(exc).__name__}", + ) + def _caller(self) -> Client | None: header = self.headers.get("Authorization", "") token = header[7:].strip() if header[:7].lower() == "bearer " else "" @@ -537,7 +544,7 @@ def redirect_request(self, *args: Any, **kwargs: Any) -> None: # noqa: D102 return None -def build() -> ThreadingHTTPServer: +def build() -> HttpServer: """The gate, configured from the environment and bound, not yet serving: doors.py runs it beside the proxy and the ingress; main() runs it alone.""" environ = dict(os.environ) @@ -564,7 +571,7 @@ def host(url: str) -> str: len(Gate.clients), ", ".join(f"{c.name}({sum(len(t) for t in c.tools.values())})" for c in Gate.clients), ) - return ThreadingHTTPServer(("0.0.0.0", port), Gate) # noqa: S104 — the container's own network only + return HttpServer(("0.0.0.0", port), Gate) # noqa: S104 — the container's own network only def main() -> int: diff --git a/deploy/sidecars/proxy.py b/deploy/sidecars/proxy.py index 64454345..55d3344a 100644 --- a/deploy/sidecars/proxy.py +++ b/deploy/sidecars/proxy.py @@ -222,11 +222,29 @@ class Handler(socketserver.StreamRequestHandler): rbufsize = 0 def handle(self) -> None: + self._target = "" + self._spliced = False + try: + self._handle() + except (ConnectionResetError, BrokenPipeError): + return + except Exception as exc: # noqa: BLE001 — every request leaves a ledger line and gets an answer + # Anything but a Refused used to end the thread: no answer, and no + # line in the refusal ledger for the one request that did something + # odd — a name the resolver cannot even encode, say. Before the + # splice it is a refusal like any other; after it, the tunnel was + # already the answer, and the log line is what is left to say. + logger.error("unexpected failure on %s", self._target or "a request", exc_info=True) + if not self._spliced: + self._refuse(502, f"unexpected {type(exc).__name__}: {exc}"[:200], self._target or "?") + + def _handle(self) -> None: try: line = self.rfile.readline(65536).decode("latin-1").strip() except OSError: return parts = line.split() + self._target = (parts[1] if len(parts) > 1 else line)[:80] if len(parts) < 3: self._refuse(400, "not a request line", line[:40]) return @@ -263,6 +281,7 @@ def handle(self) -> None: more = self.rfile.readline(65536) if more in (b"", b"\r\n", b"\n"): break + self._spliced = True self.wfile.write(b"HTTP/1.1 200 Connection established\r\n\r\n") self.wfile.flush() logger.info("allowed %s", target[:80]) @@ -301,7 +320,10 @@ def _plain(self, parts: list[str]) -> None: # head as the client sent it. Connection: close on both sides — this # forwards one request per connection rather than tracking keep-alive # state, which is the difference between a boundary and a web server. - head = [f"{method} /{path if slash else ''} {parts[2]}".encode(), b"Connection: close"] + # Latin-1 both ways: the line was read as Latin-1 above, so encoding it + # back the same way hands the upstream the bytes the client sent. UTF-8 + # here turned every raw non-ASCII byte in a path into two. + head = [f"{method} /{path if slash else ''} {parts[2]}".encode("latin-1"), b"Connection: close"] while True: raw = self.rfile.readline(65536) if raw in (b"", b"\r\n", b"\n"): @@ -310,6 +332,7 @@ def _plain(self, parts: list[str]) -> None: continue head.append(raw.rstrip(b"\r\n")) logger.info("allowed %s %s:%s", method, host, port) + self._spliced = True with upstream: upstream.sendall(b"\r\n".join(head) + b"\r\n\r\n") pump(self.connection, upstream, IDLE_SECONDS) diff --git a/deploy/sidecars/signer.py b/deploy/sidecars/signer.py index b315c4eb..29c00e1e 100644 --- a/deploy/sidecars/signer.py +++ b/deploy/sidecars/signer.py @@ -70,11 +70,10 @@ import time import urllib.error import urllib.request -from http.server import ThreadingHTTPServer from pathlib import Path from typing import Any -from common import HttpHandler, Ledger, constant_time_eq, sign_timestamped +from common import HttpHandler, HttpServer, Ledger, constant_time_eq, sign_timestamped logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") logger = logging.getLogger("watch-signer") @@ -291,6 +290,9 @@ class Signer(HttpHandler): ledger = Ledger("") counter = Counter() + def unexpected(self, exc: BaseException) -> None: + self.ledger.write(event="signal.failed", why=f"unexpected {type(exc).__name__}") + def _say(self, status: int, payload: dict[str, Any]) -> None: self._answer(status, json.dumps(payload, ensure_ascii=False).encode()) @@ -408,7 +410,7 @@ def main() -> int: Signer.counter = Counter() port = int(os.environ.get("WATCH_SIGNER_PORT", "8099")) logger.info("up on :%d, signing for %s, ceiling %d per round", port, Signer.door, PER_ROUND_MAX) - ThreadingHTTPServer(("0.0.0.0", port), Signer).serve_forever() # noqa: S104 — the container's own network only + HttpServer(("0.0.0.0", port), Signer).serve_forever() # noqa: S104 — the container's own network only return 0 diff --git a/deploy/sidecars/tests/test_allowlist.py b/deploy/sidecars/tests/test_allowlist.py index 335c4322..eb6b6fc7 100644 --- a/deploy/sidecars/tests/test_allowlist.py +++ b/deploy/sidecars/tests/test_allowlist.py @@ -275,3 +275,73 @@ def handle(self) -> None: for server in (upstream, gateway): server.shutdown() server.server_close() + + +def test_a_failure_nobody_planned_for_is_refused_and_recorded(tmp_path, monkeypatch) -> None: + """Anything but a Refused used to end the thread: no answer, no ledger line.""" + import json + import socket + import threading + + ledger = tmp_path / "egress.jsonl" + proxy = _with("example.com", ledger=str(ledger)) + + def boom(host, port): + raise RuntimeError("nobody planned for this") + + monkeypatch.setattr(proxy, "decide", boom) + gateway = proxy.Server(("127.0.0.1", 0), proxy.Handler) + threading.Thread(target=gateway.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start() + try: + with socket.create_connection(("127.0.0.1", gateway.server_address[1]), timeout=5) as client: + client.sendall(b"CONNECT example.com:443 HTTP/1.1\r\nHost: example.com\r\n\r\n") + first = client.recv(200).decode("latin-1").splitlines()[0] + finally: + gateway.shutdown() + gateway.server_close() + assert first.startswith("HTTP/1.1 502") + row = json.loads(ledger.read_text(encoding="utf-8").splitlines()[-1]) + assert row["event"] == "egress.refused" and row["target"] == "example.com:443" and row["code"] == 502 + assert "RuntimeError" in row["why"] + + +def test_a_plain_request_line_reaches_the_peer_byte_for_byte() -> None: + """The line is read as Latin-1, so it must leave as Latin-1: a client that + does not percent-encode sends raw UTF-8, and UTF-8 on the way out doubled it.""" + import socket + import socketserver + import threading + + seen: list[bytes] = [] + + class Capture(socketserver.BaseRequestHandler): + def handle(self) -> None: + self.request.settimeout(5) + data = b"" + while b"\r\n\r\n" not in data: + chunk = self.request.recv(4096) + if not chunk: + break + data += chunk + seen.append(data.split(b"\r\n", 1)[0]) + self.request.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + + upstream = socketserver.TCPServer(("127.0.0.1", 0), Capture) + port = upstream.server_address[1] + proxy = _with("", str(port), internal="127.0.0.1") + gateway = proxy.Server(("127.0.0.1", 0), proxy.Handler) + for server in (upstream, gateway): + threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start() + path = "/trace/告警?q=磁盘".encode() + try: + with socket.create_connection(("127.0.0.1", gateway.server_address[1]), timeout=5) as client: + client.sendall(b"GET http://127.0.0.1:%d%s HTTP/1.1\r\nHost: x\r\n\r\n" % (port, path)) + answer = b"" + while chunk := client.recv(4096): + answer += chunk + finally: + for server in (upstream, gateway): + server.shutdown() + server.server_close() + assert answer.startswith(b"HTTP/1.1 200"), answer[:80] + assert seen == [b"GET " + path + b" HTTP/1.1"] diff --git a/deploy/sidecars/tests/test_common.py b/deploy/sidecars/tests/test_common.py index a6104c3f..de4549c5 100644 --- a/deploy/sidecars/tests/test_common.py +++ b/deploy/sidecars/tests/test_common.py @@ -4,6 +4,7 @@ import logging import socket +import socketserver import sys import threading import time @@ -78,3 +79,77 @@ def test_a_body_that_never_arrives_names_the_request_it_belonged_to(server, capl assert "requests_seen=1" in message assert "'method': 'POST'" in message and "'path': '/v1/thing'" in message assert "'content_length': '10'" in message + + +class Boom(common.HttpHandler): + """A door that raises where nobody planned it, before or after answering.""" + + timeout = 0.3 + seen: list[str] = [] # noqa: RUF012 — test bookkeeping + + def do_GET(self) -> None: # noqa: N802 + if self.path == "/late": + self._answer(200, b"{}") + raise RuntimeError("a path nobody planned for") + + def unexpected(self, exc: BaseException) -> None: + Boom.seen.append(type(exc).__name__) + + +@pytest.fixture +def boom(): + Boom.seen = [] + srv = common.HttpServer(("127.0.0.1", 0), Boom) + threading.Thread(target=srv.serve_forever, daemon=True).start() + yield srv.server_address + srv.shutdown() + srv.server_close() + + +def _get(address, path: str) -> bytes: + chunks = [] + with socket.create_connection(address, timeout=5) as sock: + sock.sendall(f"GET {path} HTTP/1.1\r\nHost: x\r\n\r\n".encode()) + sock.settimeout(2) + try: + while chunk := sock.recv(4096): + chunks.append(chunk) + except TimeoutError: + pass + return b"".join(chunks) + + +def test_a_door_that_raises_still_answers_records_and_logs(boom, caplog): + caplog.set_level(logging.ERROR) + answer = _get(boom, "/x") + assert answer.startswith(b"HTTP/1.1 502"), answer[:80] + assert b"RuntimeError" in answer + assert Boom.seen == ["RuntimeError"], "the door's own ledger hook ran" + record = wait_for(caplog, "unexpected failure on GET /x") + assert record.exc_info, "the traceback travels with the timestamped line" + + +def test_a_door_that_raises_after_answering_sends_nothing_more(boom): + answer = _get(boom, "/late") + assert answer.startswith(b"HTTP/1.1 200") and b"502" not in answer, answer + assert Boom.seen == ["RuntimeError"] + + +def test_an_error_that_escapes_a_handler_is_logged_with_its_peer(caplog): + caplog.set_level(logging.ERROR) + + class Raw(socketserver.BaseRequestHandler): + def handle(self) -> None: + raise RuntimeError("raw handler") + + srv = common.Server(("127.0.0.1", 0), Raw) + threading.Thread(target=srv.serve_forever, daemon=True).start() + try: + with socket.create_connection(srv.server_address, timeout=5) as sock: + sock.sendall(b"x") + record = wait_for(caplog, "unhandled error serving") + finally: + srv.shutdown() + srv.server_close() + assert record.levelno == logging.ERROR and record.exc_info + assert "127.0.0.1" in record.getMessage() diff --git a/deploy/sidecars/tests/test_forward.py b/deploy/sidecars/tests/test_forward.py index f04303a1..c516d8d9 100644 --- a/deploy/sidecars/tests/test_forward.py +++ b/deploy/sidecars/tests/test_forward.py @@ -50,15 +50,16 @@ def test_a_port_that_cannot_be_bound_is_an_error_before_anything_serves(): """Every listener is bound in main() before any serves: the first shape served route one on the main thread and bound the rest in daemon threads, where a taken port was a traceback in the log and a dark console.""" - taken = socket.socket() - taken.bind(("127.0.0.1", 0)) # the same loopback address the forwarder binds, so the clash is real - taken.listen(1) - _, port = taken.getsockname() + # The clash has to be with the forwarder's own kind of bind. The forwarder + # binds the wildcard address with SO_REUSEADDR, and on macOS that bind + # succeeds beside a socket on 127.0.0.1 — so a loopback-only clash passed on + # Linux and failed on a Mac. Two forwarder binds on one port fail on both. + held = forward.bind(0, "127.0.0.1", 1) try: with pytest.raises(OSError): - forward.bind(port, "127.0.0.1", 1) + forward.bind(held.server_address[1], "127.0.0.1", 1) finally: - taken.close() + held.server_close() def test_bytes_go_both_ways_and_the_far_end_closing_ends_it(): diff --git a/deploy/sidecars/tests/test_gate.py b/deploy/sidecars/tests/test_gate.py index 71757f60..62b08fae 100644 --- a/deploy/sidecars/tests/test_gate.py +++ b/deploy/sidecars/tests/test_gate.py @@ -535,3 +535,16 @@ def test_one_upstream_and_several_do_not_mix(monkeypatch): monkeypatch.setenv("MCPGATE_CLIENT_WATCH_TOOLS", "jira:jira.get_issue") with pytest.raises(SystemExit): gate_module.build() + + +def test_a_failure_nobody_planned_for_is_answered_and_recorded(stack, monkeypatch): + def boom(self, tool, route=""): + raise RuntimeError("nobody planned for this") + + monkeypatch.setattr(gate_module.Client, "permits", boom) + status, body, _ = call(stack, _tool("chat.list_mentions")) + assert status == 502 and "RuntimeError" in body["error"] + failed = rows()[-1] + assert failed["event"] == "call.failed" and failed["reason"] == "unexpected RuntimeError" + assert failed["method"] == "POST" + assert not Upstream.seen, "nothing reached the chat server" diff --git a/deploy/sidecars/tests/test_signer.py b/deploy/sidecars/tests/test_signer.py index a20b8d1f..5bf8cb66 100644 --- a/deploy/sidecars/tests/test_signer.py +++ b/deploy/sidecars/tests/test_signer.py @@ -399,3 +399,15 @@ def test_a_config_that_would_sign_anything_stops_the_container(monkeypatch, env, monkeypatch.setenv(name, value) with pytest.raises(SystemExit): signer_module.main() + + +def test_a_failure_nobody_planned_for_is_answered_and_recorded(stack, monkeypatch): + url, _, ledger = stack + + def boom(signal, offer): + raise RuntimeError("nobody planned for this") + + monkeypatch.setattr(signer_module, "check", boom) + status, body = post(url, a_signal()) + assert status == 502 and "RuntimeError" in body["error"] + assert events(ledger)[-1] == "signal.failed" diff --git a/deploy/watch/tests/test_scan.py b/deploy/watch/tests/test_scan.py index 86cd6e2a..e1883192 100644 --- a/deploy/watch/tests/test_scan.py +++ b/deploy/watch/tests/test_scan.py @@ -182,3 +182,53 @@ def test_jira_runs_against_a_staged_copy_of_its_state(state, monkeypatch): assert committed == {"jira_last_check": 4}, "the committed state is untouched" staged = json.loads((state / "jira.staged.json").read_text()) assert staged["from"] == 4, "the copy started from the committed state" + + +@pytest.mark.parametrize( + "name", ["ops_alert_bot", "CI-bot", "Deploy BOT", "AlertBot", "robot", "监控bot", "告警机器人"] +) +def test_a_name_that_says_bot_needs_no_listing(name): + assert watch_scan.BOT_NAME.search(name) + + +@pytest.mark.parametrize("name", ["Talbot", "TALBOT", "Abbott", "Botao Wang", "Bottas", "Robert Chen", "Ada Lovelace"]) +def test_bot_inside_a_person_name_is_not_a_bot(name): + assert watch_scan.BOT_NAME.search(name) is None + + +class _OneFeed: + """A chat service with one group and the messages it answers for it.""" + + def __init__(self, messages): + self.messages = messages + + def call(self, tool, **args): + if tool == "list_folder_feeds": + return { + "feeds": [ + { + "name": "ops room", + "chat_type_label": "group chat", + "last_message_send_at": "2026-10-27T02:05:00Z", + } + ] + } + assert tool == "search_chat_records", tool + return {"messages": self.messages} + + +def test_a_bot_that_says_so_in_its_name_never_reaches_the_digest(monkeypatch): + monkeypatch.setattr(watch_scan, "BOTS", []) + monkeypatch.setattr(watch_scan, "ME", "") + chat = _OneFeed( + [ + {"send_at": "2026-10-27T02:04:00Z", "send_name": "ops_alert_bot", "content": "P1 disk at 91% on db-1"}, + { + "send_at": "2026-10-27T02:05:00Z", + "send_name": "Talbot", + "content": "can someone look at db-1 before lunch?", + }, + ] + ) + found = watch_scan.scan_chat(chat, {}, watch_scan.when("2026-10-27T02:10:00Z"), []) + assert [(name, [who for _, who, _ in msgs]) for name, msgs in found] == [("ops room", ["Talbot"])] diff --git a/deploy/watch/watch_scan.py b/deploy/watch/watch_scan.py index ae0c9d4b..781fbb1b 100644 --- a/deploy/watch/watch_scan.py +++ b/deploy/watch/watch_scan.py @@ -83,6 +83,11 @@ EXCLUDE = [s.strip() for s in (os.environ.get("WATCH_EXCLUDE") or "").split(",") if s.strip()] ME = (os.environ.get("WATCH_ME") or "").strip() BOTS = [s.strip() for s in (os.environ.get("WATCH_BOTS") or "").split(",") if s.strip()] +# 名字自己说了是机器人的，不用登记进 WATCH_BOTS：bot 或 robot 单独成词（前后不挨着字母， +# 如 ops_alert_bot、CI-bot、Deploy BOT）、驼峰结尾的 Bot（AlertBot），或者名字里带「机器人」。 +# 嵌在人名里的不算（Talbot、Abbott、Botao）。名单只留给名字看不出来的机器人；代价是显示名里 +# bot 单独成词的人也会被剔除。 +BOT_NAME = re.compile(r"(?i:(? None: @@ -85,7 +92,14 @@ async def _send(self, client: httpx.AsyncClient, text: str, now: float) -> None: text += f"\n({held} more folded into this one during the quiet window)" # Feishu bot text shape doubles as a readable generic JSON body. try: - await client.post(self._url, json={"msg_type": "text", "content": {"text": text}}, timeout=5.0) + for attempt in range(RETRIES + 1): + try: + await client.post(self._url, json={"msg_type": "text", "content": {"text": text}}, timeout=5.0) + break + except httpx.TransportError: + if attempt == RETRIES: + raise + await asyncio.sleep(RETRY_DELAY_SECONDS) except Exception: # noqa: BLE001 — an alarm must never raise into the worker self._last_sent = 0.0 # let the next dead return try again # The count was cleared on the way in, for a message that never diff --git a/hookjudge/tests/test_alarm_retry.py b/hookjudge/tests/test_alarm_retry.py new file mode 100644 index 00000000..a396c0fa --- /dev/null +++ b/hookjudge/tests/test_alarm_retry.py @@ -0,0 +1,47 @@ +"""The judge's alarm tries a connection that did not happen again, in place. + +An alarm is one message with nothing behind it that resends it. Any answer from +the channel, whatever its status, is not retried — httpx does not raise for one. +""" + +from __future__ import annotations + +import httpx +import pytest + +from hookjudge import alarm as alarm_module +from hookjudge.alarm import SelfAlarm + + +@pytest.fixture(autouse=True) +def _no_pause(monkeypatch): + monkeypatch.setattr(alarm_module, "RETRY_DELAY_SECONDS", 0) + + +class _Client: + def __init__(self, misses: int) -> None: + self.misses = misses + self.posts: list[str] = [] + + async def post(self, url: str, *, json: dict, timeout: float) -> object: + self.posts.append(json["content"]["text"]) + if len(self.posts) <= self.misses: + raise httpx.ConnectError("[Errno -2] Name or service not known") + return object() + + +async def test_a_dead_return_alarm_survives_a_moment_without_a_resolver() -> None: + alarm = SelfAlarm("https://bot.example/hook", min_interval_seconds=600) + client = _Client(misses=2) + await alarm.dead_return(client, title="disk full on db-1", error="pipe answered 503", now=1000.0) + assert len(client.posts) == 3 and "disk full on db-1" in client.posts[-1] + + +async def test_after_its_retries_it_gives_up_owes_the_count_and_the_next_one_tries() -> None: + alarm = SelfAlarm("https://bot.example/hook", min_interval_seconds=600) + down = _Client(misses=99) + await alarm.dead_return(down, title="a", error="e", now=1000.0) # must not raise + assert len(down.posts) == alarm_module.RETRIES + 1 + up = _Client(misses=0) + await alarm.dead_return(up, title="b", error="e", now=1001.0) + assert len(up.posts) == 1 and "1 more folded" in up.posts[0], "the lost message is owed to the next" diff --git a/hookprobe/hookprobe/notify.py b/hookprobe/hookprobe/notify.py index 6c686a58..51e10086 100644 --- a/hookprobe/hookprobe/notify.py +++ b/hookprobe/hookprobe/notify.py @@ -30,6 +30,7 @@ import json import logging import time +import urllib.error import urllib.request from hookprobe import actions, audit, remediation @@ -46,6 +47,15 @@ # can reply in a thread posts THIS under it, so the phone reads the evidence # the card was written from. Pilot zero's cards were read on a phone with no # way to reach the console, and the summary was all they ever carried. +# A connection that did not happen is not the channel refusing. On a desktop +# container network the embedded resolver misses a name for a moment at a time: +# a second deployment running copies of these files counted ten such failures +# in eleven days, every one answered by a resend half a minute later. An alarm +# has nothing behind it that resends, so it tries again in place first. An HTTP +# error is the channel answering, and is not tried again. A timeout is: a page +# that arrives twice costs less than one that never did. +ALARM_RETRIES = 2 +ALARM_RETRY_DELAY_SECONDS = 2.0 _DETAIL_MAX = 60_000 @@ -236,7 +246,17 @@ async def alarm(self, text: str) -> bool: message += f"\n({held} more folded into this one during the quiet window)" body = json.dumps({"msg_type": "text", "content": {"text": message}}, ensure_ascii=False).encode() try: - await asyncio.to_thread(self._post_alarm, body) + for attempt in range(ALARM_RETRIES + 1): + try: + await asyncio.to_thread(self._post_alarm, body) + break + except urllib.error.HTTPError: + raise # the channel answered; asking again will not change its mind + except OSError as exc: # URLError included: a name or a connection that did not happen + if attempt == ALARM_RETRIES: + raise + logger.warning("alarm post failed (%s); trying again in %.0fs", exc, ALARM_RETRY_DELAY_SECONDS) + await asyncio.sleep(ALARM_RETRY_DELAY_SECONDS) except Exception: # noqa: BLE001 — an alarm must never raise into its caller self._alarm_last_sent = 0.0 # let the next failure try again return False diff --git a/hookprobe/tests/test_alarm_retry.py b/hookprobe/tests/test_alarm_retry.py new file mode 100644 index 00000000..4ac4bea5 --- /dev/null +++ b/hookprobe/tests/test_alarm_retry.py @@ -0,0 +1,56 @@ +"""An alarm is one message with nothing behind it that resends it. + +A name that does not resolve for a moment is not the channel refusing; an HTTP +error is. The first is tried again in place, the second is not. +""" + +from __future__ import annotations + +import asyncio +import socket +import urllib.error + +from hookprobe import notify +from hookprobe.runs import RunStore +from tests.helpers import make_settings + + +def _delivery(tmp_path, monkeypatch, outcomes): + monkeypatch.setattr(notify, "ALARM_RETRY_DELAY_SECONDS", 0) + settings = make_settings(tmp_path, workdir=tmp_path, alarm_url="https://alarm.invalid/hook") + delivery = notify.ReturnDelivery(settings, RunStore(tmp_path / "results")) + tries: list[bytes] = [] + + def post(body: bytes) -> None: + tries.append(body) + outcome = outcomes[min(len(tries), len(outcomes)) - 1] + if outcome is not None: + raise outcome + + monkeypatch.setattr(delivery, "_post_alarm", post) + return delivery, tries + + +def _unresolved() -> urllib.error.URLError: + return urllib.error.URLError(socket.gaierror(-2, "Name or service not known")) + + +def test_a_name_that_did_not_resolve_is_tried_again_in_place(tmp_path, monkeypatch) -> None: + delivery, tries = _delivery(tmp_path, monkeypatch, [_unresolved(), _unresolved(), None]) + assert asyncio.run(delivery.alarm("node down")) is True + assert len(tries) == 3 + + +def test_the_channel_refusing_is_not_asked_again(tmp_path, monkeypatch) -> None: + refused = urllib.error.HTTPError("https://alarm.invalid/hook", 400, "Bad Request", None, None) + delivery, tries = _delivery(tmp_path, monkeypatch, [refused]) + assert asyncio.run(delivery.alarm("node down")) is False + assert len(tries) == 1 + + +def test_after_its_retries_it_gives_up_and_the_next_alarm_still_tries(tmp_path, monkeypatch) -> None: + delivery, tries = _delivery(tmp_path, monkeypatch, [_unresolved()]) + assert asyncio.run(delivery.alarm("node down")) is False + assert len(tries) == 3 + assert asyncio.run(delivery.alarm("still down")) is False + assert len(tries) == 6, "a failed alarm does not open the quiet window" diff --git a/hookrelay/hookrelay/alarm.py b/hookrelay/hookrelay/alarm.py index fca83f89..707d2d8c 100644 --- a/hookrelay/hookrelay/alarm.py +++ b/hookrelay/hookrelay/alarm.py @@ -11,12 +11,35 @@ - suppressed count folded into the next message, so throttling never hides the scale of the problem. - never raises: an alarm failure must not become a delivery failure. + +And one more: a connection that did not happen is not the channel refusing. A +desktop container network's resolver misses a name for a moment at a time, and +an alarm has nothing behind it that resends, so it is tried again in place +first. Any answer, whatever its status, is the channel speaking. A timeout is +tried again too: a page that arrives twice costs less than one that never did. """ from __future__ import annotations +import asyncio + import httpx +RETRIES = 2 +RETRY_DELAY_SECONDS = 2.0 + + +async def post_with_retries(client: httpx.AsyncClient, url: str, body: dict) -> None: + """One message, tried again in place while the connection does not happen.""" + for attempt in range(RETRIES + 1): + try: + await client.post(url, json=body, timeout=5.0) + return + except httpx.TransportError: + if attempt == RETRIES: + raise + await asyncio.sleep(RETRY_DELAY_SECONDS) + class SelfAlarm: def __init__(self, url: str, min_interval_seconds: int) -> None: @@ -45,7 +68,7 @@ async def dead_letter( text += f"\n({held} more dead letters folded into this one during the quiet window)" # Feishu bot text shape doubles as a readable generic JSON body. try: - await client.post(self._url, json={"msg_type": "text", "content": {"text": text}}, timeout=5.0) + await post_with_retries(client, self._url, {"msg_type": "text", "content": {"text": text}}) except Exception: # noqa: BLE001 — an alarm must never raise into delivery self._last_sent = 0.0 # let the next dead letter try again # The count was cleared on the way in, for a message that never diff --git a/hookrelay/tests/test_observability.py b/hookrelay/tests/test_observability.py index 65cc08eb..4773cee8 100644 --- a/hookrelay/tests/test_observability.py +++ b/hookrelay/tests/test_observability.py @@ -8,11 +8,18 @@ from __future__ import annotations import httpx +import pytest +from hookrelay import alarm as alarm_module from hookrelay import metrics from hookrelay.alarm import SelfAlarm +@pytest.fixture(autouse=True) +def _no_pause_between_alarm_attempts(monkeypatch): + monkeypatch.setattr(alarm_module, "RETRY_DELAY_SECONDS", 0) + + def test_render_emits_help_type_and_all_consumer_series() -> None: metrics._events.clear() metrics._deliveries.clear() @@ -91,9 +98,10 @@ async def test_alarm_failure_never_raises_and_permits_a_retry() -> None: client = _RecordingClient(fail=True) await alarm.dead_letter(client, channel="c", event_id=1, error="e", now=1000.0) # must not raise - # A failed alarm does not consume the window — the next dead letter tries again. + # A failed alarm does not consume the window — the next dead letter tries again, + # and each one tried in place first. await alarm.dead_letter(client, channel="c", event_id=2, error="e", now=1001.0) - assert len(client.posts) == 2 + assert len(client.posts) == 2 * (alarm_module.RETRIES + 1) async def test_a_failed_alarm_still_owes_the_count_it_swallowed() -> None: @@ -117,6 +125,30 @@ async def test_a_failed_alarm_still_owes_the_count_it_swallowed() -> None: assert "4 more dead letters" in text, f"the lost message and its folded count went missing: {text}" +class _FlakyClient(_RecordingClient): + """A resolver that misses the name for the first `misses` posts.""" + + def __init__(self, misses: int) -> None: + super().__init__() + self.misses = misses + + async def post(self, url: str, *, json: dict, timeout: float) -> object: + self.posts.append({"url": url, "json": json}) + if len(self.posts) <= self.misses: + raise httpx.ConnectError("[Errno -2] Name or service not known") + return object() + + +async def test_a_connection_that_did_not_happen_is_tried_again_in_place() -> None: + alarm = SelfAlarm("https://bot.example/hook", min_interval_seconds=600) + client = _FlakyClient(misses=2) + await alarm.dead_letter(client, channel="c", event_id=1, error="e", now=1000.0) + assert len(client.posts) == 3, "two misses, then the message" + # It arrived, so the window is spent: the next dead letter folds. + await alarm.dead_letter(client, channel="c", event_id=2, error="e", now=1001.0) + assert len(client.posts) == 3 + + async def test_unconfigured_alarm_is_inert() -> None: alarm = SelfAlarm("", min_interval_seconds=600) client = _RecordingClient()