hil: only fail /json check on the actual double-send signature

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):

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.")