Add readiness check to tests to avoid race condition flake

.github/workflows/test.yaml

@@ -72,6 +72,12 @@ jobs:
         name: Install libjpeg on Ubuntu
         run: sudo apt-get install -y libjpeg-dev
 
+      - name: Install setsid on macOS
+        if: runner.os == 'macOS'
+        run: |
+          brew install util-linux
+          echo "$(brew --prefix util-linux)/bin" >> "$GITHUB_PATH"
+
       - name: Install uv
         if: matrix.toxenv == 'oldestdeps'
         run: python -m pip install uv

.gitignore

@@ -2,6 +2,7 @@ __pycache__/
 src/__pycache__/
 *.egg-info/
 .coverage
+.coverage.*
 coverage.*
 .duct/
 .idea

CONTRIBUTING.rst

@@ -59,3 +59,9 @@ It is also possible to use pytest directly, but you will need to override the op
 `tox.ini` with `addopts`. Invocation in this way can be faster while developing::
 
     python -m pytest -o addopts="" test/test_thething.py::specific_test
+
+Some tests need the ``setsid`` command and are skipped where it is missing.
+macOS does not ship it; to run those tests there, install util-linux and put it on ``PATH``::
+
+    brew install util-linux
+    export PATH="$(brew --prefix util-linux)/bin:$PATH"

pyproject.toml

@@ -77,9 +77,6 @@ filterwarnings = [
     # matplotlib 3.5 uses deprecated pyparsing API (enablePackrat -> enable_packrat)
     "ignore:'enablePackrat' deprecated:DeprecationWarning:matplotlib._mathtext",
 ]
-markers = [
-    "flaky: mark a test as being unreliable",
-]
 norecursedirs = ["test/data"]
 
 [tool.setuptools.packages.find]
@@ -117,6 +114,7 @@ src_paths = ["src"]
 [tool.coverage.run]
 branch = true
 parallel = true
+patch = ["subprocess", "execv"]
 
 [tool.coverage.paths]
 source = [

src/con_duct/_duct_main.py

@@ -9,7 +9,7 @@
 from typing import IO, TextIO
 from con_duct._models import LogPaths, Outputs, RecordTypes, SessionMode
 from con_duct._output import TailPipe, prepare_outputs, remove_files, safe_close_files
-from con_duct._signals import SigIntHandler
+from con_duct._signals import SigIntHandler, start_thread
 from con_duct._tracker import Report, monitor_process
 
 __version__ = version("con-duct")
@@ -65,6 +65,11 @@ def execute(
             sample_interval,
         )
 
+    # Install before any command exists so no SIGINT can kill duct and orphan
+    # the command; until attach() the handler only counts them.
+    sigint_handler = SigIntHandler()
+    signal.signal(signal.SIGINT, sigint_handler)
+
     log_paths = LogPaths.create(output_prefix, pid=os.getpid())
     try:
         log_paths.prepare_paths(clobber, capture_outputs)
@@ -99,6 +104,12 @@ def execute(
     )
     files_to_close.append(report.usage_file)
 
+    if sigint_handler.sigcount:
+        safe_close_files(files_to_close)
+        remove_files(log_paths, assert_empty=True)
+        lgr.info("Received SIGINT before starting %r, not starting it", command)
+        return 128 + signal.SIGINT
+
     report.start_time = time.time()
     try:
         report.process = process = subprocess.Popen(
@@ -118,7 +129,7 @@ def execute(
         lgr.error("%s: command not found", command)
         return 127  # seems what zsh and bash return then
 
-    signal.signal(signal.SIGINT, SigIntHandler(process.pid))
+    sigint_handler.attach(process.pid)
     lgr.info("duct %s is executing %r...", __version__, full_command)
     lgr.info("Log files will be written to %s", log_paths.prefix)
     try:
@@ -145,15 +156,15 @@ def execute(
         monitoring_thread = threading.Thread(
             target=monitor_process, args=monitoring_args
         )
-        monitoring_thread.start()
+        start_thread(monitoring_thread)
     else:
         monitoring_thread = None
 
     if record_types.has_system_summary():
         env_thread = threading.Thread(target=report.collect_environment)
-        env_thread.start()
+        start_thread(env_thread)
         sys_info_thread = threading.Thread(target=report.get_system_info)
-        sys_info_thread.start()
+        start_thread(sys_info_thread)
     else:
         env_thread, sys_info_thread = None, None
 

src/con_duct/_output.py

@@ -9,6 +9,7 @@
 import time
 from typing import IO, Any, TextIO
 from con_duct._models import LogPaths, Outputs
+from con_duct._signals import start_thread
 
 
 class TailPipe:
@@ -27,7 +28,7 @@ def start(self) -> None:
         self.stop_event = threading.Event()
         self.infile = open(self.file_path, "rb")
         self.thread = threading.Thread(target=self._tail, daemon=True)
-        self.thread.start()
+        start_thread(self.thread)
 
     def fileno(self) -> int:
         assert self.infile is not None

src/con_duct/_signals.py

@@ -1,41 +1,138 @@
 """Signal handlers for con-duct."""
 
 from __future__ import annotations
+from collections.abc import Iterator
+from contextlib import contextmanager
 import logging
 import os
 import signal
+import threading
 from types import FrameType
 from typing import Optional
 
 lgr = logging.getLogger("con-duct")
 
 
+@contextmanager
+def _sigint_blocked() -> Iterator[None]:
+    """Block SIGINT in the calling thread, restoring its mask on exit.
+
+    A SIGINT that arrives meanwhile stays pending and is handled once the
+    mask is restored.
+    """
+    old_mask = signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGINT})
+    try:
+        yield
+    finally:
+        signal.pthread_sigmask(signal.SIG_SETMASK, old_mask)
+
+
+def start_thread(thread: threading.Thread) -> None:
+    """Start a thread that the kernel will never hand a SIGINT to.
+
+    A signal sent to the process is delivered to any one thread that does
+    not block it, but Python runs the handler only in the main thread. If
+    another thread takes a SIGINT, the main thread stays asleep in its wait
+    for the command and the Ctrl-C is not forwarded until the command exits.
+    A new thread inherits its creator's signal mask, so block SIGINT just
+    while starting it. Every thread duct starts must be started this way.
+
+    Parameters
+    ----------
+    thread : threading.Thread
+        The thread to start
+    """
+    with _sigint_blocked():
+        thread.start()
+
+
 class SigIntHandler:
     """
     Handler of SIGINT signals received by the process running duct.
     """
 
-    def __init__(self, pid: int) -> None:
-        """
+    def __init__(self) -> None:
+        # Installed before the command starts; SIGINTs are only counted
+        # until attach() says which process to forward them to.
+        self.pid: Optional[int] = None
+        self.sigcount: int = 0
+
+    def attach(self, pid: int) -> None:
+        """Start forwarding SIGINTs to the command.
+
         Parameters
         ----------
         pid : int
-            The PID of the process monitored by duct
+            The PID of the process monitored by duct. A SIGINT that arrived
+            while it was being started is acted on now.
         """
-        self.pid: int = pid
-        self.sigcount: int = 0
+        # A SIGINT after publishing the PID must not be forwarded both by
+        # __call__ and by the pending-signal check below.
+        with _sigint_blocked():
+            self.pid = pid
+            if self.sigcount:
+                self._act()
 
     def __call__(self, _sig: int, _frame: Optional[FrameType]) -> None:
         self.sigcount += 1
+        if self.pid is not None:
+            self._act()
+
+    def _act(self) -> None:
+        try:
+            self._forward()
+        except ProcessLookupError:
+            # The command has already exited. Python runs this handler in
+            # the main thread at its next bytecode, which can be only after
+            # the wait for the command returns.
+            _log(
+                logging.WARNING,
+                "Received SIGINT, but the command has already exited",
+            )
+
+    def _forward(self) -> None:
+        assert self.pid is not None
+        # Act before logging: the handler can interrupt duct mid-write to
+        # stderr, and a failed log write must not stop the signal.
         if self.sigcount == 1:
-            lgr.info("Received SIGINT, passing to command")
             os.kill(self.pid, signal.SIGINT)
+            _log(logging.INFO, "Received SIGINT, passing to command")
         elif self.sigcount == 2:
-            lgr.info("Received second SIGINT, again passing to command")
             os.kill(self.pid, signal.SIGINT)
+            _log(logging.INFO, "Received second SIGINT, again passing to command")
         elif self.sigcount == 3:
-            lgr.warning("Received third SIGINT, forcefully killing command process")
             os.kill(self.pid, signal.SIGKILL)
+            _log(
+                logging.WARNING,
+                "Received third SIGINT, forcefully killing command process",
+            )
         elif self.sigcount >= 4:
-            lgr.critical("Exiting duct, skipping cleanup")
+            _log(logging.CRITICAL, "Exiting duct, skipping cleanup")
             os._exit(1)
+
+
+def _log(level: int, msg: str) -> None:
+    """Log from a signal handler without ever raising.
+
+    If the signal arrived while duct was writing to stderr, logging raises
+    ``RuntimeError: reentrant call`` (even its error report does): the
+    buffered writer is held by the write the handler interrupted, which
+    cannot finish until the handler returns. Write the message straight to
+    file descriptor 2 instead, if the log level lets the message through.
+    It does not go through the log format, and it may land in the middle of
+    the log line that was being written, so it starts on a new line and says
+    why it looks different.
+    """
+    try:
+        lgr.log(level, msg)
+    except RuntimeError:
+        if not lgr.isEnabledFor(level):
+            return
+        try:
+            line = (
+                f"\ncon-duct [{logging.getLevelName(level)}] (written directly: "
+                f"the signal interrupted a log write): {msg}\n"
+            )
+            os.write(2, line.encode())
+        except OSError:
+            pass

test/data/signal_ignorer.py

@@ -1,19 +1,33 @@
 #!/usr/bin/env python3
+"""Ignore SIGINT, reporting progress through the file given as argv[1].
+
+Writes "ready" once the handler is installed, then one "SIGINT" line per
+SIGINT received, so a test can wait for each step instead of sleeping.
+"""
+
 import signal
+import sys
 import time
 from types import FrameType
 from typing import Optional
 
 
+def report(line: str) -> None:
+    with open(sys.argv[1], "a") as f:
+        f.write(line + "\n")
+
+
 def handle_signal(sig: int, _frame: Optional[FrameType]) -> None:
     print(f"Received {sig}")
+    report("SIGINT")
 
 
 if __name__ == "__main__":
     signal.signal(signal.SIGINT, handle_signal)
     signal.siginterrupt(
         signal.SIGINT, False
     )  # Restart interrupted system calls so we can test multiple SIGINTS
+    report("ready")
     t0 = time.time()
-    while time.time() - t0 < 10:
+    while time.time() - t0 < 60:
         time.sleep(0.01)