Merge pull request #23 from jeromekelleher/on-demand-sidecars

Build encoder-server static sidecars on demand

Author: jeromekelleher

biofuse/encoder_client.py

@@ -201,14 +201,17 @@ async def start(
         try:
             await self._handshake()
         except BaseException as exc:
+            # If the subprocess has exited on its own during startup
+            # the handshake failure is a downstream symptom (refused
+            # connect, RST mid frame, premature EOF). Surface a single
+            # clean OSError so the CLI prints one error line and
+            # points the user at the server's own log. Check
+            # ``is_alive`` before ``aclose`` — the latter signals stop
+            # and joins the child, after which ``exitcode`` is set
+            # regardless of why the child exited.
+            child_died_on_own = not proc.is_alive()
             await self.aclose()
-            # If the subprocess has exited during startup the handshake
-            # failure is a downstream symptom (refused connect, RST mid
-            # frame, premature EOF). Surface a single clean OSError so
-            # the CLI prints one error line and points the user at the
-            # server's own log, instead of leaking the trio / socket
-            # exception type to the caller.
-            if proc.exitcode is not None and not isinstance(exc, KeyboardInterrupt):
+            if child_died_on_own and not isinstance(exc, KeyboardInterrupt):
                 raise OSError(
                     errno.EIO,
                     f"{spec.name}-server exited during startup; "

biofuse/encoder_server.py

@@ -5,9 +5,9 @@
 ``AF_UNIX`` listening socket. The owning :class:`~biofuse.formats.FormatSpec`
 selects:
 
-- which static sidecar files to precompute and cache (``.bim``/``.fam``
-  for PLINK; ``.sample``/``.bgen.bgi`` for BGEN), via
-  :meth:`FormatSpec.build_static_files`;
+- which static sidecar files to build on demand for the metadata
+  handshake (``.bim``/``.fam`` for PLINK; ``.sample``/``.bgen.bgi``
+  for BGEN), via :meth:`FormatSpec.build_static_files`;
 - which encoder class to construct per accepted connection, via
   :meth:`FormatSpec.encoder_factory`.
 
@@ -37,10 +37,13 @@
 class _ServerSession:
     """Server-side state shared across all connection threads.
 
-    Holds the ``VczReader``, the precomputed static-file bodies (in
-    the format spec's declared order), and the streaming-file size.
-    Immutable after construction; safe to read concurrently from any
-    thread without locking.
+    Holds the ``VczReader`` and the active format spec. The static
+    sidecar bytes and the streaming-file size are both built on demand
+    inside the metadata handshake (see ``_make_metadata_reply``)
+    rather than cached here — the parent fuse handler is the canonical
+    owner of that metadata after the handshake completes. Immutable
+    after construction; safe to read concurrently from any thread
+    without locking.
     """
 
     def __init__(
@@ -50,19 +53,6 @@ def __init__(
     ) -> None:
         self.reader = reader
         self.spec = spec
-        self.static_files: dict[str, bytes] = spec.build_static_files(reader)
-        missing = set(spec.static_suffixes) - set(self.static_files)
-        extra = set(self.static_files) - set(spec.static_suffixes)
-        if missing or extra:
-            raise ValueError(
-                f"{spec.name}: build_static_files returned keys "
-                f"{sorted(self.static_files)}; expected {list(spec.static_suffixes)}"
-            )
-        # Open one throwaway encoder to read ``total_size`` — encoder
-        # construction is I/O-free, so this is cheap. Per-connection
-        # encoders are constructed fresh in ``_handle_connection``.
-        with spec.encoder_factory(reader) as encoder:
-            self.stream_size: int = int(encoder.total_size)
 
 
 def _recv_exact_sync(sock: socket.socket, n: int) -> bytes:
@@ -85,6 +75,113 @@ def _recv_exact_sync(sock: socket.socket, n: int) -> bytes:
     return bytes(buf)
 
 
+def _make_error_reply(exc: BaseException, context: str) -> bytes:
+    """Pack an errno reply for ``exc`` and log the cause.
+
+    Non-``OSError`` causes log a one-line ERROR with the errno and
+    the exception text, plus a DEBUG-only traceback. Used by both
+    the encoder-construction failure path and the per-dispatch
+    exception handler.
+    """
+    err = encoder_protocol.errno_for_exception(exc)
+    if not isinstance(exc, OSError):
+        logger.error("encoder-server %s; replying with errno %d: %s", context, err, exc)
+        logger.debug("encoder-server %s traceback", context, exc_info=True)
+    return encoder_protocol.pack_error_reply(err)
+
+
+def _make_metadata_reply(session: "_ServerSession") -> bytes:
+    """Build the reply for a ``TAG_GET_METADATA`` request.
+
+    The static sidecars and the streaming-file size are built on
+    demand from the format spec, serialised into the reply, then
+    dropped: the parent fuse handler owns the canonical copy after
+    the handshake. Encoder construction is I/O-free so reading
+    ``total_size`` here is cheap.
+    """
+    static_files = session.spec.build_static_files(session.reader)
+    missing = set(session.spec.static_suffixes) - set(static_files)
+    extra = set(static_files) - set(session.spec.static_suffixes)
+    if missing or extra:
+        raise ValueError(
+            f"{session.spec.name}: build_static_files returned keys "
+            f"{sorted(static_files)}; expected {list(session.spec.static_suffixes)}"
+        )
+    with session.spec.encoder_factory(session.reader) as encoder:
+        stream_size = int(encoder.total_size)
+    ordered_bodies = [static_files[suffix] for suffix in session.spec.static_suffixes]
+    return encoder_protocol.pack_metadata_reply(ordered_bodies, stream_size)
+
+
+def _make_read_reply(conn_sock: socket.socket, encoder, tname: str) -> bytes | None:
+    """Build the reply for a ``TAG_READ`` request, or ``None`` on
+    truncated payload (caller terminates the connection)."""
+    payload = _recv_exact_sync(conn_sock, encoder_protocol.REQ_READ_PAYLOAD_SIZE)
+    if len(payload) < encoder_protocol.REQ_READ_PAYLOAD_SIZE:
+        return None
+    off, size = encoder_protocol.parse_read_payload(payload)
+    t_read = time.monotonic()
+    data = encoder.read(off, size)
+    logger.debug(
+        "%s: encoder.read off=%d size=%d in %.3fs",
+        tname,
+        off,
+        size,
+        time.monotonic() - t_read,
+    )
+    return encoder_protocol.pack_read_reply(data)
+
+
+def _send_reply(conn_sock: socket.socket, reply: bytes) -> bool:
+    """Send ``reply`` on the socket; return ``True`` on success and
+    ``False`` if the send raised ``OSError`` (already logged)."""
+    try:
+        conn_sock.sendall(reply)
+        return True
+    except OSError as exc:
+        logger.warning("encoder-server send failed: %s", exc)
+        return False
+
+
+def _serve_connection(
+    conn_sock: socket.socket,
+    session: "_ServerSession",
+    encoder,
+    tname: str,
+) -> None:
+    """Run the tag-dispatch loop for one connected client.
+
+    Exits cleanly on EOF, unknown tag, truncated payload, or a send
+    failure. Exceptions raised inside a dispatch case are converted
+    to errno replies and the loop continues to the next request.
+    """
+    while True:
+        try:
+            tag_buf = _recv_exact_sync(conn_sock, 1)
+        except EOFError as exc:
+            logger.warning("encoder-server frame error: %s", exc)
+            return
+        if len(tag_buf) == 0:
+            return
+        tag = bytes(tag_buf)
+        try:
+            if tag == encoder_protocol.TAG_GET_METADATA:
+                reply = _make_metadata_reply(session)
+            elif tag == encoder_protocol.TAG_READ:
+                reply = _make_read_reply(conn_sock, encoder, tname)
+            else:
+                logger.warning(
+                    "encoder-server: unknown tag %r; closing connection", tag
+                )
+                return
+        except Exception as exc:  # noqa: BLE001 - any error becomes errno reply
+            reply = _make_error_reply(exc, "dispatch raised")
+        if reply is None:
+            return
+        if not _send_reply(conn_sock, reply):
+            return
+
+
 def _handle_connection(conn_sock: socket.socket, session: _ServerSession) -> None:
     """Run one client connection synchronously. Returns on EOF.
 
@@ -102,73 +199,15 @@ def _handle_connection(conn_sock: socket.socket, session: _ServerSession) -> Non
             t_enc = time.monotonic()
             encoder_cm = session.spec.encoder_factory(session.reader)
         except Exception as exc:  # noqa: BLE001 - any error becomes errno reply
-            err = encoder_protocol.errno_for_exception(exc)
-            if not isinstance(exc, OSError):
-                logger.exception(
-                    "encoder-server encoder construction failed; replying with errno"
-                )
-            try:
-                conn_sock.sendall(encoder_protocol.pack_error_reply(err))
-            except OSError as send_exc:
-                logger.warning("encoder-server send failed: %s", send_exc)
+            _send_reply(
+                conn_sock, _make_error_reply(exc, "encoder construction failed")
+            )
             return
         with encoder_cm as encoder:
             logger.debug(
                 "%s: encoder created in %.3fs", tname, time.monotonic() - t_enc
             )
-            while True:
-                try:
-                    tag_buf = _recv_exact_sync(conn_sock, 1)
-                except EOFError as exc:
-                    logger.warning("encoder-server frame error: %s", exc)
-                    return
-                if len(tag_buf) == 0:
-                    return
-                tag = bytes(tag_buf)
-                try:
-                    if tag == encoder_protocol.TAG_GET_METADATA:
-                        ordered_bodies = [
-                            session.static_files[suffix]
-                            for suffix in session.spec.static_suffixes
-                        ]
-                        reply = encoder_protocol.pack_metadata_reply(
-                            ordered_bodies, session.stream_size
-                        )
-                    elif tag == encoder_protocol.TAG_READ:
-                        payload = _recv_exact_sync(
-                            conn_sock, encoder_protocol.REQ_READ_PAYLOAD_SIZE
-                        )
-                        if len(payload) < encoder_protocol.REQ_READ_PAYLOAD_SIZE:
-                            return
-                        off, size = encoder_protocol.parse_read_payload(payload)
-                        t_read = time.monotonic()
-                        data = encoder.read(off, size)
-                        logger.debug(
-                            "%s: encoder.read off=%d size=%d in %.3fs",
-                            tname,
-                            off,
-                            size,
-                            time.monotonic() - t_read,
-                        )
-                        reply = encoder_protocol.pack_read_reply(data)
-                    else:
-                        logger.warning(
-                            "encoder-server: unknown tag %r; closing connection",
-                            tag,
-                        )
-                        return
-                except Exception as exc:  # noqa: BLE001 - any error becomes errno reply
-                    err = encoder_protocol.errno_for_exception(exc)
-                    if not isinstance(exc, OSError):
-                        logger.exception(
-                            "encoder-server dispatch raised; replying with EIO"
-                        )
-                    reply = encoder_protocol.pack_error_reply(err)
-                try:
-                    conn_sock.sendall(reply)
-                except OSError as exc:
-                    logger.warning("encoder-server send failed: %s", exc)
-                    return
+            _serve_connection(conn_sock, session, encoder, tname)
     logger.debug("%s: conn thread exit", tname)
 
 
@@ -253,12 +292,15 @@ def _server_main(
     ``make_reader`` consumes.
 
     Any exception raised before ``serve_forever`` starts (reader
-    construction, ``_ServerSession`` construction — the latter eagerly
-    walks the variants to build the static sidecars and may reject e.g.
-    multi-allelic input) is caught here so multiprocessing's default
-    handler does not print a traceback. The cause is logged at ERROR
-    (visible at default verbosity); the traceback only surfaces at
-    DEBUG.
+    construction, ``_ServerSession`` construction) is caught here so
+    multiprocessing's default handler does not print a traceback. The
+    cause is logged at ERROR (visible at default verbosity); the
+    traceback only surfaces at DEBUG. Static-sidecar build errors
+    (e.g. multi-allelic input rejection) now surface inside
+    ``_handle_connection`` on the first metadata request rather than
+    at session construction; the handler converts them to errno
+    replies so the parent sees a clean ``OSError`` from
+    ``_handshake``.
     """
     log_config.apply()
     try:

tests/test_encoder_client.py

@@ -344,22 +344,25 @@ class TestRealSubprocess:
     @pytest.mark.parametrize(
         "spec", [formats.PLINK_SPEC, formats.BGEN_SPEC], ids=["plink", "bgen"]
     )
-    async def test_start_fast_fails_on_subprocess_startup_error(
+    async def test_start_fast_fails_on_handshake_build_error(
         self, fx_multiallelic_vcz, tmp_path, spec
     ):
         """``EncoderClient.start()`` must surface a clean ``OSError`` and
         return well under the 10 s connect deadline when the subprocess
-        catches a startup error (here: multi-allelic VCZ) and exits."""
+        rejects the input during the metadata handshake. Static-file
+        build (here: multi-allelic VCZ) runs on demand inside
+        ``_handle_connection``; failures become errno replies on the
+        handshake socket."""
         socket_path = tmp_path / "encoder.sock"
         t0 = trio.current_time()
-        with pytest.raises(OSError, match="exited during startup") as excinfo:
+        with pytest.raises(OSError, match="encoder-server reported errno") as excinfo:
             await encoder_client.EncoderClient.start(
                 str(fx_multiallelic_vcz.path), socket_path, spec
             )
         elapsed = trio.current_time() - t0
         assert excinfo.value.errno == errno.EIO
         assert elapsed < encoder_client._CONNECT_DEADLINE_S, (
-            f"start() should fast-fail when child dies, took {elapsed:.2f}s"
+            f"start() should fast-fail on handshake error, took {elapsed:.2f}s"
         )
 
     async def test_max_alleles_filter_through_start(