fix(harness): exit normally at a stop, and never wait on stderr

Round 4 security review of #20 at b4da6f0:

- N-c: the stop is an ordinary interpreter exit again, so the handler
  module's atexit hooks run and its buffers are flushed. What made shutdown
  abort was harness threads holding stderr's lock; the harness's own
  diagnostics now queue for one writer that writes the descriptor directly,
  and an unanswered connection error is one of those diagnostics rather than
  a stdlib traceback. Only past the grace, beside author code still running,
  does the process run its atexit hooks within 2 s and exit directly.
- Pre-existing: no diagnostic waits on an unread stderr, so a call's
  diagnostic no longer holds its slot and the exit-1 message cannot hold the
  exit.
- N-b: the failure code is read once, checked, and that value is sent; a
  code that raises when read asserts nothing.
- N-d: the read-budget test's body is refused only at its last element, so
  removing the decode lock fails it.
- N-a: docs/protocol.md states the decode queue's latency bound.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: antoinetoussaint-byte

docs/protocol.md

@@ -94,6 +94,15 @@ budget the process holds one document being decoded and, under each running
 call's slot, a payload its contract admits — the envelope the declaration's
 `max-input-bytes` and `concurrency` state.
 
+Decoding one call at a time is first come, first served, so a small call can
+wait behind every decode queued ahead of it. The wait is bounded: admission comes
+before the queue, so at most min(concurrency, readers) calls are ahead, each one
+decode and check of at most `max-input-bytes`. A 1 MiB body whose last element
+is the one the schema refuses takes about 0.2 s, so at concurrency 32 a call
+waits at most about six seconds behind a full queue. A review measured the
+median of small calls rising from 0.58 s to 1.69 s against 24 peers looping such
+bodies, every call still answered `200`.
+
 **Concurrency is not yet one number.** Core's declaration says `0` means the
 facility default and defines none, so the harness admits 32 when the declaration
 is silent. The durable-work runtime decides how many calls it dispatches from
@@ -156,9 +165,11 @@ The code becomes a response header, so it must match
 `^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,63}$` — the grammar the durable-work runtime reads
 it with, which is narrower than core's 1 to 128 characters. A code outside it
 raises `ValueError` where it is constructed, and the handler's call then answers
-as any untyped error does: unproven. The code is read-only, and checked again
-where the header is set: a code changed after construction answers `500` with
-no failure code.
+as any untyped error does: unproven. The code is read-only, and where the
+header is set it is read exactly once, checked, and that same value is sent: a
+code changed after construction — even by a subclass whose `code` property
+returns something else on a later read — answers `500` with no failure code, and
+so does a code that raises when read.
 
 Everything else leaves the effect **unproven**, because the harness cannot know
 which side of its effect a handler stopped on. A caller that read a bare status
@@ -200,15 +211,26 @@ already open, which answer `503` with `Connection: close` — and gives the call
 in flight `SHUTDOWN_GRACE_SECONDS` (30 s) to finish — including a handler whose
 call was already answered at its deadline. Cutting them off would turn every one
 of them into an unproven effect at once. The harness exits `0` as soon as the
-last handler has returned; an idle connection holds nothing. The process ends
-with that code directly rather than through interpreter shutdown, which can
-abort while connection threads are still parked or writing a diagnostic. If the grace ends
-with handlers still running, they are cut off and the harness exits `1`, saying
-how many effects it left unproven.
+last handler has returned; an idle connection holds nothing. It is an ordinary
+interpreter exit: the handler module's `atexit` hooks run and its buffered
+writes are flushed, so a `logging` handler, a telemetry exporter or a client
+that flushes at exit still does. If the grace ends with handlers still running,
+they are cut off and the harness exits `1`, saying how many effects it left
+unproven. That exit cannot wait for interpreter shutdown beside author code
+that is still running and may hold any lock, so the `atexit` hooks run and the
+standard streams are flushed for at most `CLOSE_SECONDS` (2 s), and the process
+then ends with `1` directly; a buffered write the handler never flushed and no
+hook flushes is lost there.
 
 `stdout` and `stderr` carry diagnostics only, never completion data: a handler
 that printed its output would be indistinguishable from a library that printed a
-warning. The harness does **not** bound them. Per-call stream bounding is not
+warning. The harness's own diagnostics never wait on the stream: they queue for
+one writer that writes the descriptor directly, so a placement that does not
+read stderr cannot stall a call that holds its slot, and no harness thread holds
+the stream's lock when the process exits. Past 1,024 queued diagnostics a new
+one is dropped and counted, and the count is written at exit if the stream then accepts it. A handler's own
+`print` and `context.log` write the stream as any code does, and wait on it as
+any code does. The harness does **not** bound the streams. Per-call stream bounding is not
 something a server can do — concurrent calls share the process's descriptors —
 and `max_log_bytes` describes what the placement capturing those streams
 enforces, which is where it belongs.

pkg/contract/contract_test.go

@@ -100,8 +100,9 @@ func TestTheFailureCodeHeaderIsSetInOnePlace(t *testing.T) {
 	if occurrences := strings.Count(runner, "FAILURE_CODE_HEADER"); occurrences != 2 {
 		t.Fatalf("the failure-code header is named %d times: once in the import and once where it is set", occurrences)
 	}
-	if !strings.Contains(runner, "self.send_header(FAILURE_CODE_HEADER, answer.failure.code)") {
-		t.Error("the failure-code header no longer carries the handler's own declared code")
+	if !strings.Contains(runner, "self.send_header(FAILURE_CODE_HEADER, code)") ||
+		strings.Count(runner, "code = _code_of(answer.failure)") != 1 {
+		t.Error("the failure-code header no longer carries the handler's own declared code, read once and checked")
 	}
 }
 

pkg/harness/codefly_runnable/protocol.py

@@ -80,6 +80,10 @@
 # number, not one any placement resolved; see docs/protocol.md.
 HARNESS_CONCURRENCY = 32
 
+# How long the harness gives its own connection threads to end, and its
+# diagnostics to be written, once the calls in flight have finished.
+CLOSE_SECONDS = 2.0
+
 # How long a call in flight has to finish after a signal. A handler interrupted
 # mid-effect leaves it unproven, so the process waits rather than making every
 # call in flight unproven at once.

pkg/harness/codefly_runnable/runner.py

@@ -14,10 +14,12 @@
 
 from __future__ import annotations
 
+import atexit
 import importlib
 import io
 import json
 import os
+import queue
 import re
 import signal
 import socket
@@ -34,6 +36,7 @@
 from .contract import CONTRACT_FILE, RECEIPT_ATTRIBUTE, Contract
 from .protocol import (
     ADDRESS_VARIABLE,
+    CLOSE_SECONDS,
     FAILURE_CODE_HEADER,
     FAILURE_CODE_PATTERN,
     HARNESS_CONCURRENCY,
@@ -142,6 +145,58 @@ def drain(self, timeout: float) -> int:
             return self._active
 
 
+class _Diagnostics:
+    """The harness's own diagnostics, written so that nothing waits on stderr.
+
+    A placement that never reads the stream would otherwise stall whichever
+    thread printed — and a request thread printing holds its call's slot, so
+    every later call would be refused. Messages queue for one writer, which
+    writes the descriptor directly and so holds no stream lock that a thread
+    or interpreter shutdown could wait on. Past the queue's bound a message is
+    dropped and counted rather than waited for.
+    """
+
+    def __init__(self, capacity: int = 1024) -> None:
+        self._queue: queue.Queue[str] = queue.Queue(capacity)
+        self._dropped = 0
+        self._lock = threading.Lock()
+        self._closing = threading.Event()
+        self._writer = threading.Thread(target=self._write, name="codefly-diagnostics", daemon=True)
+        self._writer.start()
+
+    def say(self, message: str) -> None:
+        try:
+            self._queue.put_nowait(message)
+        except queue.Full:
+            with self._lock:
+                self._dropped += 1
+
+    def _write(self) -> None:
+        while not (self._closing.is_set() and self._queue.empty()):
+            try:
+                message = self._queue.get(timeout=0.05)
+            except queue.Empty:
+                continue
+            data = f"[codefly] {message}\n".encode("utf-8", "backslashreplace")
+            try:
+                while data:
+                    data = data[os.write(2, data):]
+            except OSError:
+                pass
+
+    def close(self, timeout: float) -> None:
+        """Write what is queued, for at most this long."""
+        with self._lock:
+            dropped, self._dropped = self._dropped, 0
+        if dropped:
+            self.say(f"{dropped} diagnostics were dropped while stderr was not being read")
+        self._closing.set()
+        self._writer.join(timeout)
+
+
+_DIAGNOSTICS = _Diagnostics()
+
+
 def main() -> int:
     """Serve this runnable until it is signalled. The return value is the exit code."""
     root = Path(__file__).resolve().parent.parent
@@ -161,7 +216,8 @@ def main() -> int:
         calls = _Calls(contract.concurrency or HARNESS_CONCURRENCY)
         server = _server(address, contract, calls, handle, receipt_of)
     except (OSError, KeyError, ValueError, ProtocolError, AttributeError, ImportError) as err:
-        print(f"[codefly] {err}", file=sys.stderr)
+        _DIAGNOSTICS.say(str(err))
+        _DIAGNOSTICS.close(CLOSE_SECONDS)
         return 1
 
     stopping = threading.Event()
@@ -190,26 +246,38 @@ def stop(*_: object) -> None:
     # the last one has returned; an idle connection holds nothing.
     remaining = calls.drain(SHUTDOWN_GRACE_SECONDS)
     if remaining:
-        print(
-            f"[codefly] {remaining} handlers were still running when the {SHUTDOWN_GRACE_SECONDS:g}s grace ended; "
-            "their effects are unproven",
-            file=sys.stderr,
+        _DIAGNOSTICS.say(
+            f"{remaining} handlers were still running when the {SHUTDOWN_GRACE_SECONDS:g}s grace ended; "
+            "their effects are unproven"
         )
-    _exit(1 if remaining else 0)
-    return 1 if remaining else 0
+    # No thread the harness started holds a stream lock — its diagnostics are
+    # written to the descriptor directly — so interpreter shutdown cannot
+    # abort on one, and the process exits normally: the handler module's
+    # atexit hooks run and its buffers are flushed.
+    _DIAGNOSTICS.close(CLOSE_SECONDS)
+    code = 1 if remaining else 0
+    if remaining:
+        _exit_now(code)
+    return code
 
 
-def _exit(code: int) -> None:
-    """End the process with this code, whatever its connection threads are doing.
+def _exit_now(code: int) -> None:
+    """End the process with this code while threads the harness cannot stop are still running.
 
-    Interpreter finalization with daemon threads still parked on connections —
-    or holding the lock of a stream they are writing a diagnostic to — can
-    abort instead of exiting, and a placement would read a crash where the code
-    says what happened to the calls. Streams get a bounded moment to flush.
+    A handler past the grace is author code, and it may hold any lock —
+    including a stream's — so interpreter shutdown beside it could abort, and a
+    placement would read a crash where the code says what happened to the
+    calls. The handler module's atexit hooks still run and the standard
+    streams are flushed, for at most CLOSE_SECONDS, before the process ends.
     """
-    flushing = threading.Thread(target=lambda: (sys.stdout.flush(), sys.stderr.flush()), daemon=True)
-    flushing.start()
-    flushing.join(2.0)
+    def finish() -> None:
+        atexit._run_exitfuncs()
+        sys.stdout.flush()
+        sys.stderr.flush()
+
+    finishing = threading.Thread(target=finish, daemon=True)
+    finishing.start()
+    finishing.join(CLOSE_SECONDS)
     os._exit(code)
 
 
@@ -287,6 +355,10 @@ def process_request(self, request, client_address) -> None:
         self.register(request)
         super().process_request(request, client_address)
 
+    def handle_error(self, request, client_address) -> None:
+        """An exception a connection thread did not answer: a diagnostic, not a traceback on stderr."""
+        _DIAGNOSTICS.say(f"a connection ended on {_describe(sys.exc_info()[1])}")
+
     def register(self, request) -> None:
         """Count a new connection in, idle, closing the one idle the longest if it is at the bound."""
         with self._lock:
@@ -520,7 +592,8 @@ def _decode(self, body: bytes) -> Call | _Answer:
         def _answer(self, answer: _Answer) -> None:
             # Anything but a success may have left request bytes unread, and a
             # stopping harness keeps no connection for a call it will refuse.
-            if answer.failure is not None and not _asserts(answer.failure):
+            code = _code_of(answer.failure) if answer.failure is not None else None
+            if answer.failure is not None and code is None:
                 # Checked again where it is sent: a code changed after it was
                 # constructed asserts nothing, and never reaches a header.
                 answer = _Answer(500, code="internal", message="the handler's failure code changed after it was checked")
@@ -530,7 +603,7 @@ def _answer(self, answer: _Answer) -> None:
                 # The one place this header is set. A caller reads it before the
                 # status, and without it no status proves anything.
                 self.send_response(answer.status)
-                self.send_header(FAILURE_CODE_HEADER, answer.failure.code)
+                self.send_header(FAILURE_CODE_HEADER, code)
                 body = _error("failed_precondition", str(answer.failure))
             elif answer.status == 200:
                 self.send_response(200)
@@ -548,7 +621,7 @@ def _answer(self, answer: _Answer) -> None:
             self.end_headers()
             self.wfile.write(body)
             if answer.message:
-                print(f"[codefly] {answer.message}", file=sys.stderr)
+                _DIAGNOSTICS.say(answer.message)
 
         def log_request(self, *_: object) -> None:
             """No request log: an answer's diagnostics are printed by _answer."""
@@ -559,9 +632,17 @@ def log_message(self, *_: object) -> None:
     return _Server(host, port, Handler, readers=readers, connections=readers + calls.limit + IDLE_CONNECTIONS)
 
 
-def _asserts(failure: HandlerFailure) -> bool:
-    """Whether a failure's code is still one the grammar admits."""
-    return isinstance(failure.code, str) and FAILURE_CODE_PATTERN.fullmatch(failure.code) is not None
+def _code_of(failure: HandlerFailure) -> str | None:
+    """The failure's code, read once, or None when it is no longer one the grammar admits.
+
+    Read once because the value checked must be the value sent: a code that
+    could change between the two reads would reach the header unchecked.
+    """
+    try:
+        code = failure.code
+    except Exception:  # noqa: BLE001 — a code that cannot be read asserts nothing
+        return None
+    return code if isinstance(code, str) and FAILURE_CODE_PATTERN.fullmatch(code) else None
 
 
 def _listen(address: str) -> tuple[str, int]: