diff --git a/scripts/hil_monitor.py b/scripts/hil_monitor.py index 3a106c0..690b125 100644 --- a/scripts/hil_monitor.py +++ b/scripts/hil_monitor.py @@ -37,48 +37,93 @@ IP_PATTERN = re.compile(r"IP address:\s*(\d+\.\d+\.\d+\.\d+)") JSON_CHECK_DELAY_SECS = 15 # let the web server settle after boot -JSON_CHECK_WORKERS = 3 -JSON_CHECK_REQUESTS = 10 # sequential requests per worker, on one kept-alive connection +JSON_CHECK_WORKERS = 3 # concurrent connections — enough to reliably collide on serveJson +JSON_CHECK_REQUESTS = 12 # sequential requests per worker, on one kept-alive connection -def json_endpoint_check(ip, failure): - """Hammer /json/devices concurrently and verify every response is well-formed. +def json_endpoint_check(ip, bug): + """Detect the serveJson() double-send under concurrent load. serveJson() serialises itself with a `servingJson` flag and returns 429 when busy. - A missing early-return there sends two responses down one socket. The extra one is - left in the buffer, so every later response on that connection is off by one. - - That shift is invisible if you only check "is this valid JSON" — the stale response - is a perfectly good 200. So each worker reuses one keep-alive connection (to expose - the shift at all) and alternates two endpoints with different shapes: /json has no - "devices" key, /json/devices does. A shifted stream answers the wrong question. + A missing early-return there sends TWO responses down one socket for one request. + The extra one is left in the buffer, so every later response on that connection is + off by one — and each shifted response is still a valid 200. That is the only thing + this check is looking for, and it is the only thing that fails the build. + + To surface the shift, each worker reuses one keep-alive connection and alternates two + endpoints with different shapes: /json has no "devices" key, /json/devices does. A + shifted stream answers the wrong question, or hands back the 429 text as a 200 body. + + The rule is simple: a 200 must be a complete, correct JSON object for the URL that + asked for it. Refusing to serve under pressure is fine — a 429 (busy or low heap) is + the designed refusal and is silently accepted; anything odder that still isn't the bug + (a dropped TCP connection, an unexpected non-200/429 status) is counted into the + "shed under load" tally, logged, and reconnected past, never failed on. But a *200* + that is null, truncated, unparseable, or the wrong document is the server lying about + success — the double-send, or the low-heap null-body path (fixed in firmware by + refusing with 429) — and it fails the build. """ - - def worker(n): - conn = http.client.HTTPConnection(ip, timeout=10) - for i in range(JSON_CHECK_REQUESTS): - path = "/json/devices" if i % 2 else "/json" - conn.request("GET", path) - resp = conn.getresponse() - body = resp.read() - if resp.status == 429: - continue - if resp.status != 200: - raise AssertionError(f"worker {n} req {i}: GET {path} -> HTTP {resp.status}") - try: - doc = json.loads(body) - except ValueError as e: - raise AssertionError( - f"worker {n} req {i}: GET {path} -> body is not valid JSON ({e}); " - f"{len(body)} bytes, starts {body[:80]!r}" - ) - if ("devices" in doc) != path.endswith("/devices"): - raise AssertionError( - f"worker {n} req {i}: GET {path} answered with the wrong document " - f"(keys {sorted(doc)}) — a stale response is queued on this " - f"connection, i.e. something sent two responses to one request" - ) - conn.close() + TOO_MANY = b"Too Many Requests" + + def worker(n, drops): + conn = None + try: + for i in range(JSON_CHECK_REQUESTS): + path = "/json/devices" if i % 2 else "/json" + try: + if conn is None: + conn = http.client.HTTPConnection(ip, timeout=10) + conn.request("GET", path) + resp = conn.getresponse() + body = resp.read() + except (OSError, http.client.HTTPException) as e: + # Connection-level hiccup: the node shed load. Not the bug. Reconnect. + drops.append(str(e)) + if conn is not None: + conn.close() + conn = None + continue + + # The firmware refuses with 429 (busy or low-heap) — the one accepted + # non-200, and silently accepted (not counted as shedding). A 503 means + # wrong/old firmware: the low-heap path was deliberately changed from 503 + # to 429 (ESPresense#2428), so a 503 on a direct connection is a contract + # violation, not shedding. + if resp.status != 200: + if resp.status == 503: + raise _Bug(f"GET {path} returned 503 — firmware must refuse with 429, not 503") + if resp.status != 429: + drops.append(f"HTTP {resp.status} on {path}") + continue + + # From here a 200 must be a complete, correct object — anything less is a lie. + + # A 200 carrying the 429 text is the double-send caught red-handed. + if TOO_MANY in body: + raise _Bug(f"GET {path} returned 200 with the 429 body {TOO_MANY!r} — " + f"two responses were sent for one request") + + try: + doc = json.loads(body) + except ValueError as e: + raise _Bug(f"GET {path} returned a 200 with an unparseable body " + f"({len(body)}B, starts {body[:60]!r}) — truncated or garbled") from e + + # A 200 that isn't a JSON object is the low-heap null-body path: the buffer + # failed to allocate, the doc serialized as `null`, and it shipped as 200 + # instead of 429. That is the bug the low-heap guard fixes. + if not isinstance(doc, dict): + raise _Bug(f"GET {path} returned a 200 with non-object JSON ({body[:40]!r}) " + f"— low-heap serving should refuse (429), not 200") + + # The double-send signature: a 200 whose document doesn't match the URL. + if ("devices" in doc) != path.endswith("/devices"): + raise _Bug(f"GET {path} answered with the wrong document (keys " + f"{sorted(doc)}) — a stale response is queued on this " + f"connection, i.e. two responses were sent for one request") + finally: + if conn is not None: + conn.close() time.sleep(JSON_CHECK_DELAY_SECS) @@ -92,27 +137,45 @@ def worker(n): print(f"[hil] /json check SKIPPED — {ip} unreachable from the runner ({e})", flush=True) return - errors = [] + bugs, drops, crashes = [], [], [] threads = [ - threading.Thread(target=lambda n=n: _run(worker, n, errors)) for n in range(JSON_CHECK_WORKERS) + threading.Thread(target=lambda n=n: _run(worker, n, bugs, drops, crashes)) + for n in range(JSON_CHECK_WORKERS) ] for t in threads: t.start() for t in threads: t.join() - if errors: - failure.append(f"/json misbehaved under concurrent load: {errors[0]}") + total = JSON_CHECK_WORKERS * JSON_CHECK_REQUESTS + notes = [] + if drops: + notes.append(f"{len(drops)}/{total} shed under load") + if crashes: + notes.append(f"{len(crashes)} worker crash(es): {crashes[0]}") + suffix = f" ({'; '.join(notes)})" if notes else "" + if bugs: + bug.append(f"/json contract violation: {bugs[0]}") + elif crashes: + # A checker crash isn't a firmware failure, but it must not read as a clean pass. + bug.append(f"/json check harness error: {crashes[0]}") else: - total = JSON_CHECK_WORKERS * JSON_CHECK_REQUESTS - print(f"[hil] /json check passed ({total} concurrent requests to {ip})", flush=True) + print(f"[hil] /json check passed ({total} concurrent requests to {ip}){suffix}", flush=True) + + +class _Bug(Exception): + """A /json contract violation that must fail the build.""" -def _run(fn, n, errors): +def _run(fn, n, bugs, drops, crashes): try: - fn(n) - except Exception as e: # noqa: BLE001 - any failure here is a test failure - errors.append(str(e)) + fn(n, drops) + except _Bug as e: + bugs.append(str(e)) + except Exception as e: # noqa: BLE001 - never let a worker die silently and still "pass" + # A checker crash is not load-shedding; track it apart so it can't hide in the tally. + crashes.append(f"worker {n}: {type(e).__name__}: {e}") + print(f"[hil] /json worker {n} crashed: {type(e).__name__}: {e}", flush=True) def format_duration(seconds): diff --git a/scripts/test_json_check.py b/scripts/test_json_check.py index 0f5492a..f5e194c 100644 --- a/scripts/test_json_check.py +++ b/scripts/test_json_check.py @@ -1,10 +1,16 @@ #!/usr/bin/env python3 -"""Prove json_endpoint_check catches the serveJson double-send and passes a correct server. - -Two mock servers on raw sockets (BaseHTTPRequestHandler can't send two responses to one -request, which is the whole point): - buggy - mimics pre-fix serveJson: when busy, writes the 429 AND the 200 JSON - fixed - mimics post-fix serveJson: when busy, writes only the 429 +"""Prove json_endpoint_check catches the serveJson double-send AND tolerates load-shedding. + +Raw-socket mocks (BaseHTTPRequestHandler can't send two responses to one request, which +is the whole point): + buggy - pre-fix serveJson: when busy, writes the 429 AND the 200 JSON (the bug) + fixed - post-fix serveJson: when busy, writes only the 429 + flaky - correct load-shedding a real constrained node does: drops keep-alive + connections mid-body, and refuses with 429 when it can't afford the buffer. + MUST pass — refusing to serve is fine. + oom - the low-heap null-body bug: under pressure returns a 200 with body `null` + instead of refusing. MUST be caught — a 200 that lies about success is the + whole point of the check (this is what ESPresense#2428 fixes in firmware). """ import os import socket @@ -13,9 +19,8 @@ import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -from hil_monitor import json_endpoint_check # noqa: E402 - import hil_monitor # noqa: E402 +from hil_monitor import json_endpoint_check # noqa: E402 hil_monitor.JSON_CHECK_DELAY_SECS = 0 # don't wait 15s in a test @@ -30,32 +35,62 @@ def resp(status, reason, body): ) -def serve(sock, buggy, stop): +def serve(sock, mode, stop): serving = threading.Lock() + counter = [0] + + def respond(conn, req): + """Handle one complete request. Return False to close the connection.""" + body = DEVICES if b"/json/devices" in req else INFO + counter[0] += 1 + + if mode == "flaky": + # Correct load-shedding — refusing to serve, never a false 200: + if counter[0] % 4 == 0: + return False # drop keep-alive mid-stream -> IncompleteRead/reset + if counter[0] % 7 == 0: + # firmware's low-heap refusal (ESPresense#2428): 429, not a false 200 + conn.sendall(resp(429, "Too Many Requests", b'{"error":"low memory"}')) + return True + + if mode == "oom" and counter[0] % 3 == 0: + # The bug: a 200 that lies — buffer failed, doc serialized as null. + conn.sendall(resp(200, "OK", b"null")) + return True + + if mode == "stale503" and counter[0] % 3 == 0: + # Wrong/old firmware: low-heap refusal as 503 instead of 429. + conn.sendall(resp(503, "Service Unavailable", b'{"error":"low memory"}')) + return True + + busy = not serving.acquire(blocking=False) + if busy: + conn.sendall(resp(429, "Too Many Requests", b"Too Many Requests")) + if mode == "buggy": + conn.sendall(resp(200, "OK", body)) # the bug: no early return + return True + try: + time.sleep(0.02) # widen the window so workers actually collide + conn.sendall(resp(200, "OK", body)) + finally: + serving.release() + return True def handle(conn): conn.settimeout(5) + # TCP is a stream: a request can span recv() calls and several can arrive in one. + # Buffer and only process a request once its terminating blank line has arrived. + buf = b"" try: while True: - data = conn.recv(4096) - if not data: + chunk = conn.recv(4096) + if not chunk: return - for req in data.split(b"\r\n\r\n")[:-1]: # one response per pipelined request - # serveJson picks the document from the URL, same as the firmware - body = DEVICES if b"/json/devices" in req else INFO - busy = not serving.acquire(blocking=False) - if busy: - conn.sendall(resp(429, "Too Many Requests", b"Too Many Requests")) - if not buggy: - continue - # the bug: no early return, so a second response follows - conn.sendall(resp(200, "OK", body)) - continue - try: - time.sleep(0.02) # widen the window so workers actually collide - conn.sendall(resp(200, "OK", body)) - finally: - serving.release() + buf += chunk + while b"\r\n\r\n" in buf: # GET requests have no body; blank line ends one + req, buf = buf.split(b"\r\n\r\n", 1) + if not respond(conn, req): + return except OSError: pass finally: @@ -69,27 +104,47 @@ def handle(conn): threading.Thread(target=handle, args=(conn,), daemon=True).start() -def run(buggy): +def run(mode): sock = socket.socket() sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind(("127.0.0.1", 0)) sock.listen(8) sock.settimeout(1) stop = threading.Event() - threading.Thread(target=serve, args=(sock, buggy, stop), daemon=True).start() + threading.Thread(target=serve, args=(sock, mode, stop), daemon=True).start() - failure = [] - json_endpoint_check(f"127.0.0.1:{sock.getsockname()[1]}", failure) + bug = [] + json_endpoint_check(f"127.0.0.1:{sock.getsockname()[1]}", bug) stop.set() sock.close() - return failure + return bug -buggy = run(buggy=True) +buggy = run("buggy") assert buggy, "detector MISSED the double-send bug" print(f"buggy server -> detected: {buggy[0]}") -fixed = run(buggy=False) +fixed = run("fixed") assert not fixed, f"detector false-positived on correct server: {fixed}" print("fixed server -> clean") -print("\nOK: detector catches the bug and does not false-positive.") + +flaky = run("flaky") +assert not flaky, f"detector false-positived on load-shedding node: {flaky}" +print("flaky server -> clean (drops + 429 tolerated)") + +oom = run("oom") +assert oom, "detector MISSED the low-heap 200-null body" +print(f"oom server -> detected: {oom[0]}") + +stale503 = run("stale503") +assert stale503, "detector MISSED a 503 (wrong/old firmware — must be 429)" +print(f"stale503 srv -> detected: {stale503[0]}") + +# _run routing: an unexpected worker exception is a harness crash, tracked apart from +# load-shedding drops so it can't hide behind a "clean pass with shedding". +_bugs, _drops, _crashes = [], [], [] +hil_monitor._run(lambda n, drops: (_ for _ in ()).throw(RuntimeError("boom")), 0, _bugs, _drops, _crashes) +assert _crashes and not _bugs and not _drops, (_bugs, _drops, _crashes) +print("crash routing -> tracked as crash, not a drop") + +print("\nOK: catches double-send + 200-null, tolerates 429+drops, rejects 503, no false positives.")