From 1386eb480b09e75339bde650709dcd5b594694a1 Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Mon, 3 Aug 2026 19:21:51 -0400 Subject: [PATCH] fix(serve): drain request body on reject so the client isn't RST mid-read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_serve.py flaked intermittently under full-suite load — a DIFFERENT serve test failed each run with ConnectionResetError [Errno 54] on resp.read(), passing in isolation. Not a port-bind race (the port is already ephemeral and bind+listen happen in the constructor). The real cause is in serve.py's handler: do_POST replies 503 (mesh disabled) and 413 (oversized) WITHOUT reading the posted body; only the 202 and bad-JSON-400 paths read it first. A rejection that leaves bytes in the kernel receive buffer makes the OS send RST instead of FIN on close, so the client — which has a valid response status — gets its connection reset mid-body-read. That exactly matches which tests flaked: is_disabled (503) and rejects_oversized (413) did; rejects_non_json (400) never did, because it reads the body to parse it. This is a real server defect, not a test artifact: a live telemetry producer POSTing to a disabled endpoint hits the same reset, not a clean 503. Fix: - serve.py: a bounded, idempotent `_drain()` discards any unread body before every reject reply (wired through `_reply`, plus the wrong-path 404). Bounded to MESH_MAX_BODY + 64 KiB so a normal/slightly-oversized body drains fully (no RST) while the 413 DoS guard still holds for a pathological body. The path that reads the body sets `_body_consumed` so the drain is a no-op there (never blocks). - test_serve.py: also close the listening socket (`server_close()`) and join the serve_forever thread in finally — each call previously leaked both. Verified: 20/20 consecutive full-suite `pytest tools/tests/` runs clean (was ~1-in-6 failing). At the old rate, 20 clean by luck is ~4%. --- serve.py | 33 ++++++++++++++++++++++++++++++++- tools/tests/test_serve.py | 5 +++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/serve.py b/serve.py index db19e78..587ec63 100644 --- a/serve.py +++ b/serve.py @@ -181,7 +181,37 @@ def do_GET(self): else: self.send_response(404); self.end_headers() + def _drain(self) -> None: + """Discard any unread request body so closing the socket sends FIN, not RST. + + A rejection (503/413/400) that replies WITHOUT reading the posted body leaves bytes in + the kernel receive buffer; on close the OS sends RST instead of FIN, and the client sees + ``ConnectionResetError`` mid-read instead of the response we just sent — intermittently, + depending on delivery timing. A real telemetry producer POSTing to a disabled endpoint + would hit the same reset, not a clean 503. Draining fixes it at the source. + + Bounded to MESH_MAX_BODY + 64 KiB: a normal or slightly-oversized body is drained in full + (no reset), while a pathologically huge body is not read to completion (the 413 DoS guard + still holds — the remainder is discarded on close). Idempotent: a path that already read + the body sets ``_body_consumed`` and this returns immediately, so it never blocks waiting + for bytes that were already consumed. + """ + if getattr(self, "_body_consumed", False): + return + self._body_consumed = True + try: + length = int(self.headers.get("Content-Length", 0) or 0) + except (ValueError, TypeError): + return + remaining = min(max(length, 0), MESH_MAX_BODY + (1 << 16)) + while remaining > 0: + chunk = self.rfile.read(min(remaining, 1 << 16)) + if not chunk: + break + remaining -= len(chunk) + def _reply(self, code: int, obj: dict) -> None: + self._drain() # never RST a client that is still reading the response we send below body = (json.dumps(obj) + "\n").encode() self.send_response(code); self.send_header("Content-Type", "application/json") self.end_headers(); self.wfile.write(body) @@ -191,7 +221,7 @@ def do_POST(self): # it is published onto the in-process bus and drained by the mesh loop under # the same fail-closed consume() contract. Off unless MESH_ENABLED. if self.path != "/mesh/telemetry": - self.send_response(404); self.end_headers(); return + self._drain(); self.send_response(404); self.end_headers(); return if not MESH_ENABLED: self._reply(503, {"error": "mesh disabled"}); return try: @@ -203,6 +233,7 @@ def do_POST(self): if length > MESH_MAX_BODY: self._reply(413, {"error": f"body exceeds {MESH_MAX_BODY} bytes"}); return body = self.rfile.read(length) + self._body_consumed = True # body now read; _reply()'s drain becomes a no-op # Edge guard: reject blatantly non-JSON at ingest (envelope validation still # happens fail-closed at consume time); don't let junk fill the bus. try: diff --git a/tools/tests/test_serve.py b/tools/tests/test_serve.py index 04956c3..e6af7fc 100644 --- a/tools/tests/test_serve.py +++ b/tools/tests/test_serve.py @@ -22,6 +22,7 @@ def _serve_and_post(path, body, *, mesh_enabled, headers=None): port = srv.server_address[1] t = threading.Thread(target=srv.serve_forever, daemon=True) t.start() + conn = None try: conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) payload = body if isinstance(body, (bytes, str)) else json.dumps(body) @@ -29,7 +30,11 @@ def _serve_and_post(path, body, *, mesh_enabled, headers=None): resp = conn.getresponse() return resp.status, resp.read() finally: + if conn is not None: + conn.close() srv.shutdown() + srv.server_close() # close the listening socket — every call leaked one otherwise + t.join(timeout=5) # and its serve_forever thread def test_int_env_falls_back_on_malformed():