fix(gateway): contain output pump shutdown races
This commit is contained in:
@@ -136,10 +136,18 @@ def _pump(name: str, stream: IO[bytes]) -> None:
|
||||
"""Read lines from `stream`, prefix with `[name]`, write to
|
||||
stdout. Runs in its own thread per child; daemon=True so a
|
||||
blocked read doesn't keep the process alive after main exits."""
|
||||
for raw in iter(stream.readline, b""):
|
||||
line = raw.decode("utf-8", errors="replace").rstrip("\n")
|
||||
sys.stdout.write(f"[{name}] {line}\n")
|
||||
sys.stdout.flush()
|
||||
try:
|
||||
for raw in iter(stream.readline, b""):
|
||||
line = raw.decode("utf-8", errors="replace").rstrip("\n")
|
||||
sys.stdout.write(f"[{name}] {line}\n")
|
||||
sys.stdout.flush()
|
||||
except (OSError, ValueError) as exc:
|
||||
# The manager closes a dead child's pipe after wait() and before a
|
||||
# restart. A pump can be between readline calls at that exact moment;
|
||||
# closed-stream errors are normal completion, not uncaught thread
|
||||
# failures. Preserve genuinely unexpected I/O diagnostics.
|
||||
if not stream.closed:
|
||||
_log(f"{name} output pump stopped: {type(exc).__name__}: {exc}")
|
||||
|
||||
|
||||
def _spawn(spec: _DaemonSpec) -> subprocess.Popen[bytes]:
|
||||
|
||||
@@ -23,6 +23,7 @@ from bot_bottle.gateway.bootstrap import (
|
||||
_DaemonManager,
|
||||
_argv_for_daemon,
|
||||
_env_for_daemon,
|
||||
_pump,
|
||||
_selected_daemons,
|
||||
)
|
||||
from tests._bin import SLEEP
|
||||
@@ -565,5 +566,26 @@ class TestMainEndToEnd(unittest.TestCase):
|
||||
self.assertIn("no daemons selected", out)
|
||||
|
||||
|
||||
class _FailingStream:
|
||||
def __init__(self, *, closed: bool) -> None:
|
||||
self.closed = closed
|
||||
|
||||
def readline(self) -> bytes:
|
||||
raise ValueError("I/O operation on closed file")
|
||||
|
||||
|
||||
class TestOutputPump(unittest.TestCase):
|
||||
def test_closed_stream_race_is_normal_completion(self) -> None:
|
||||
with patch("bot_bottle.gateway.bootstrap._log") as log:
|
||||
_pump("egress", _FailingStream(closed=True)) # type: ignore[arg-type]
|
||||
log.assert_not_called()
|
||||
|
||||
def test_unexpected_io_failure_is_logged(self) -> None:
|
||||
with patch("bot_bottle.gateway.bootstrap._log") as log:
|
||||
_pump("egress", _FailingStream(closed=False)) # type: ignore[arg-type]
|
||||
log.assert_called_once()
|
||||
self.assertIn("output pump stopped", log.call_args.args[0])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user