perf(server): offload blocking session load/save off the event loop

tests/test_dispatch_stage_threading.py

@@ -0,0 +1,141 @@
+# Copyright (c) 2026 Code Infinity
+# SPDX-License-Identifier: MPL-2.0
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+"""THE DISPATCH PIPELINE STAGES MUST NOT BLOCK THE EVENT LOOP EITHER.
+
+ADR-0074 (test_sync_handler_threading.py) threaded the ROUTE HANDLER so a
+blocking `def` route no longer freezes the loop. But only the handler was
+offloaded: every dispatch STAGE still runs on the loop (server.py, the
+`for stage in _*_STAGES` runners). Two of those stages do blocking network I/O
+whose cost the handler fix never touched:
+
+  * session LOAD  (_autostart_session -> Session.start(sid) -> handler.read)
+  * session SAVE  (_stage_session_save -> session.save() -> handler.write)
+
+With a network session backend (Redis/Valkey/memcached/MongoDB) each is a
+blocking `socket.recv`/`sendall` round-trip ON the loop, so even a perfectly
+async route serialises every request through its session I/O.
+
+These tests PROVE it the way ADR-0074 did — a real request through the REAL
+`server.handle` dispatch, a real clock, real threads, and a real file-backed
+session handler with a deliberate latency standing in for a slow backend (the
+same instrument as ADR-0074's `time.sleep` route; it is a REAL handler doing
+REAL file I/O plus a measured delay, not a mock of a backend's protocol).
+
+They are RED on today's code (stages on the loop) and GREEN once the blocking
+stages are offloaded with `asyncio.to_thread`, exactly as the handler was.
+"""
+import asyncio
+import threading
+import time
+
+import pytest
+
+import tina4_python.core.server as server
+from tina4_python.core.request import Request
+from tina4_python.core.middleware import Middleware
+from tina4_python.core.router import Router, get as route_get
+from tina4_python import session as session_mod
+from tina4_python.session import Session, FileSessionHandler, session_cookie_name
+
+DELAY = 0.3          # the latency each session read/write stands in for
+N = 4                # concurrent requests for the serialization proof
+
+
+class SlowFileSessionHandler(FileSessionHandler):
+    """A REAL file-backed session handler with a deliberate per-op latency.
+
+    read()/write() persist to a real file (the parent's behaviour) and then
+    sleep DELAY to represent a slow network backend's round-trip. The sleep is
+    a real blocking call - the instrument, not a mock - so a stage that runs it
+    on the loop genuinely blocks the loop for DELAY, which is the thing measured.
+    """
+
+    def read(self, session_id):
+        time.sleep(DELAY)
+        return super().read(session_id)
+
+    def write(self, session_id, data, ttl=0):
+        time.sleep(DELAY)
+        return super().write(session_id, data, ttl)
+
+
+@pytest.fixture(autouse=True)
+def _app(monkeypatch, tmp_path):
+    Router.clear()
+    Middleware.reset()
+    # Inject the REAL slow file handler into the dispatch's own Session() calls.
+    slow = SlowFileSessionHandler(str(tmp_path))
+    monkeypatch.setattr(Session, "_resolve_handler", staticmethod(lambda: slow))
+
+    @route_get("/sess")
+    async def _probe(request, response):           # async route: the handler is NOT the blocker
+        request.session.set("touched", "1")        # make the save stage persist (blocking write)
+        return response("ok")
+
+    yield
+    Router.clear()
+    Middleware.reset()
+
+
+def _request():
+    # A cookie makes autostart do a blocking READ; the route's write makes the
+    # save stage do a blocking WRITE. Both land in the dispatch, on the loop.
+    cookie = f"{session_cookie_name()}=abc123".encode()
+    scope = {
+        "type": "http", "method": "GET", "path": "/sess", "query_string": b"",
+        "scheme": "http", "headers": [(b"cookie", cookie)], "client": ("127.0.0.1", 1),
+    }
+    return server.handle(Request.from_scope(scope, b""))
+
+
+def test_session_io_does_not_stall_the_event_loop():
+    """While one request's session read+write run, the loop must keep ticking.
+
+    RED today: the session stages run on the loop, so the ticker is frozen for
+    ~2*DELAY and records almost nothing. GREEN once they are offloaded.
+    """
+    ticks = []
+
+    async def ticker():
+        for _ in range(200):
+            ticks.append(time.monotonic())
+            await asyncio.sleep(0.01)
+
+    async def drive():
+        t = asyncio.ensure_future(ticker())
+        await asyncio.sleep(0)          # let the ticker start
+        await _request()
+        t.cancel()
+
+    asyncio.run(drive())
+
+    assert len(ticks) > 20, (
+        f"the loop ticked only {len(ticks)} times while one request did its "
+        f"session read+write (~{2 * DELAY:.1f}s). The session stages ran ON the "
+        "loop, so every other in-flight request was frozen with it."
+    )
+
+
+def test_concurrent_requests_do_not_serialize_on_session_io():
+    """N concurrent requests must overlap their session I/O, not queue on it.
+
+    Serial (today): N * 2 * DELAY. Concurrent (after offloading): ~2 * DELAY.
+    The assertion sits well below the serial time so it is RED now, GREEN after,
+    and the ratio it prints is the performance improvement.
+    """
+    async def drive():
+        start = time.monotonic()
+        await asyncio.gather(*(_request() for _ in range(N)))
+        return time.monotonic() - start
+
+    elapsed = asyncio.run(drive())
+    serial = N * 2 * DELAY
+    assert elapsed < serial * 0.6, (
+        f"{N} concurrent requests took {elapsed:.2f}s; serial would be ~{serial:.2f}s "
+        f"and fully concurrent ~{2 * DELAY:.2f}s. They serialised on session I/O "
+        "running on the event loop."
+    )

tina4_python/core/server.py

@@ -2923,6 +2923,29 @@ def _stage_head_strip(ctx: DispatchContext) -> None:
     _stage_head_strip,
 )
 
+#: Response stages that do BLOCKING I/O and must run off the event loop (via
+#: asyncio.to_thread in the dispatch runner), so one request's session write or
+#: log write never freezes the loop for every other in-flight request. The rest
+#: of _RESPONSE_STAGES is CPU-only and stays on the loop. See the LOAD offload in
+#: handle() and the handler threading in _invoke_handler (ADR-0074).
+_OFFLOAD_RESPONSE_STAGES = frozenset({_stage_request_log, _stage_session_save})
+
+
+async def _run_response_stage(stage, ctx: "DispatchContext") -> None:
+    """Run one response stage, OFF the event loop if it does blocking I/O.
+
+    Session SAVE (backend write) and the request-log file write block; offload
+    those to a worker thread so one request's I/O never freezes the loop for the
+    rest. The CPU-only stages (headers, CORS, toolbar, HEAD strip) stay on the
+    loop. ctx is request-local, so running a stage in a thread is safe. The
+    branch lives HERE, not in handle(), so the dispatch runner stays flat
+    (test_dispatch_pipeline::test_the_god_function_does_not_come_back).
+    """
+    if stage in _OFFLOAD_RESPONSE_STAGES:
+        await asyncio.to_thread(stage, ctx)
+    else:
+        stage(ctx)
+
 
 def _resolve_request_id(request: Request) -> str:
     """Honour a well-formed inbound X-Request-ID (sanitized), else generate one.
@@ -2962,7 +2985,13 @@ async def handle(request: Request) -> Response:
     from tina4_python.frond.engine import set_request_session, reset_request_session
     session_token = set_request_session(None)
     try:
-        _init_session(request)
+        # Session LOAD reads the backend (Redis/Valkey/memcached/MongoDB do a
+        # blocking socket round-trip; the file backend a blocking file read).
+        # Offload it so it does not freeze the loop — the handler was threaded
+        # for the same reason (ADR-0074), but only the handler was. The Session
+        # and its backend connection are request-local, so a worker thread adds
+        # no cross-request sharing.
+        await asyncio.to_thread(_init_session, request)
         # form_token() binds each token to this request's session (CSRF).
         set_request_session(request.session)
 
@@ -2988,7 +3017,7 @@ async def handle(request: Request) -> Response:
                     break
 
         for stage in _RESPONSE_STAGES:
-            stage(ctx)
+            await _run_response_stage(stage, ctx)
 
         return ctx.response
     finally:
@@ -3825,6 +3854,7 @@ def _find_available_port(start: int, max_tries: int = 10) -> int:
     import socket
     try:
         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         s.bind(("127.0.0.1", start))
         s.close()
         return start