From b7aba37df3ba0c95fefbe5129bfded7431cfd6e9 Mon Sep 17 00:00:00 2001 From: Roger Luethi Date: Mon, 6 Jul 2026 12:22:33 +0200 Subject: [PATCH] reconciler: drain stdout to avoid deadlock run_on_change() never read the /run.sh subprocess pipe and called p.wait() with no timeout, and run() left the pipe undrained when publish=False. In these paths stdout is a subprocess.PIPE that is never consumed, so a /run.sh emitting more than the OS pipe buffer (~64 KB) blocks on write while the parent blocks forever on wait(): a deadlock that hangs the reconciler worker. Always drain the pipe. run() now reads it whether or not the output is forwarded downstream, moving the read loop back outside the publish guard. run_on_change() writes the drained lines to the log (the task publishes nowhere) and p.wait() gains the timeout=60 backstop that run() already uses. Update the affected unit tests to feed real byte streams through the production drain loop and assert that the pipe is consumed, rather than pinning the previous undrained call shape. Assisted-by: Claude:claude-opus-4-8 Signed-off-by: Roger Luethi --- osism/tasks/reconciler.py | 18 ++++++++++++---- tests/unit/tasks/test_task_wrappers.py | 30 ++++++++++++++++++++------ 2 files changed, 37 insertions(+), 11 deletions(-) diff --git a/osism/tasks/reconciler.py b/osism/tasks/reconciler.py index 1764c8f0..dab6796f 100644 --- a/osism/tasks/reconciler.py +++ b/osism/tasks/reconciler.py @@ -47,9 +47,12 @@ def run(self, publish=True): env=env, ) - if publish: - for line in io.TextIOWrapper(p.stdout, encoding="utf-8"): - utils.push_task_output(self.request.id, line) + # Always drain stdout so a chatty /run.sh cannot fill the pipe buffer + # and deadlock on wait(); only forward the lines when publishing. + with io.TextIOWrapper(p.stdout, encoding="utf-8") as stdout: + for line in stdout: + if publish: + utils.push_task_output(self.request.id, line) rc = p.wait(timeout=60) @@ -78,7 +81,14 @@ def run_on_change(self): p = subprocess.Popen( "/run.sh", shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT ) - p.wait() + + # Drain stdout into the log so a chatty /run.sh cannot fill the pipe + # buffer and deadlock on wait(); run_on_change publishes nowhere. + with io.TextIOWrapper(p.stdout, encoding="utf-8") as stdout: + for line in stdout: + logger.info(line.rstrip()) + + p.wait(timeout=60) from pottery import ReleaseUnlockedLock diff --git a/tests/unit/tasks/test_task_wrappers.py b/tests/unit/tasks/test_task_wrappers.py index 4a7e0d4d..78a47baa 100644 --- a/tests/unit/tasks/test_task_wrappers.py +++ b/tests/unit/tasks/test_task_wrappers.py @@ -428,8 +428,8 @@ def test_reconciler_run_publishes_output(mocker): lock.release.assert_called_once_with() -def test_reconciler_run_without_publish_skips_output(mocker): - """With ``publish=False`` no output is published but the process is waited on.""" +def test_reconciler_run_without_publish_drains_but_skips_output(mocker): + """With ``publish=False`` stdout is still drained but nothing is published.""" mocker.patch("osism.tasks.reconciler.utils.check_task_lock_and_exit") lock = mocker.MagicMock() lock.acquire.return_value = True @@ -437,11 +437,17 @@ def test_reconciler_run_without_publish_skips_output(mocker): push = mocker.patch("osism.tasks.reconciler.utils.push_task_output") finish = mocker.patch("osism.tasks.reconciler.utils.finish_task_output") proc = mocker.MagicMock() + # Real bytes so the production ``TextIOWrapper`` drain loop consumes the + # pipe; if the loop were skipped a filled pipe would deadlock ``wait()``. + proc.stdout = io.BytesIO(b"noise one\nnoise two\n") proc.wait.return_value = 0 mocker.patch("osism.tasks.reconciler.subprocess.Popen", return_value=proc) reconciler.run.__wrapped__(publish=False) + # The drain loop wrapped and exhausted the pipe (``TextIOWrapper`` closes + # the buffer on completion) even though nothing is forwarded downstream. + assert proc.stdout.closed push.assert_not_called() finish.assert_not_called() proc.wait.assert_called_once_with(timeout=60) @@ -493,8 +499,8 @@ def test_run_on_change_returns_none_when_lock_not_acquired(mocker): assert result is None -def test_run_on_change_runs_script_without_publishing(mocker): - """``run_on_change`` spawns ``/run.sh`` with no ``env`` kwarg and no output.""" +def test_run_on_change_drains_script_output_into_log(mocker, loguru_logs): + """``run_on_change`` spawns ``/run.sh`` and drains its output into the log.""" check = mocker.patch("osism.tasks.reconciler.utils.check_task_lock_and_exit") lock = mocker.MagicMock() lock.acquire.return_value = True @@ -502,18 +508,27 @@ def test_run_on_change_runs_script_without_publishing(mocker): push = mocker.patch("osism.tasks.reconciler.utils.push_task_output") finish = mocker.patch("osism.tasks.reconciler.utils.finish_task_output") proc = mocker.MagicMock() + # Real bytes so the production ``TextIOWrapper`` drain loop consumes the + # pipe; without draining a filled pipe would deadlock the bare ``wait()``. + proc.stdout = io.BytesIO(b"reconcile line\n") popen = mocker.patch("osism.tasks.reconciler.subprocess.Popen", return_value=proc) reconciler.run_on_change.__wrapped__() - # The exact-match assertion also proves there is no ``env`` kwarg. + # ``stdout=PIPE`` is required to drain; the exact match also proves there + # is no ``env`` kwarg (unlike ``run``). popen.assert_called_once_with( "/run.sh", shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT ) - proc.wait.assert_called_once_with() + # Pipe fully consumed (``TextIOWrapper`` closes the buffer on completion) + # and each line forwarded to the log. + assert proc.stdout.closed + assert _has_log(loguru_logs, "INFO", "reconcile line") + # A timeout backstops the final reap, matching ``run``. + proc.wait.assert_called_once_with(timeout=60) + # ``run_on_change`` publishes nowhere and has no task-lock check. push.assert_not_called() finish.assert_not_called() - # ``run_on_change`` has no task-lock check, unlike ``run``. check.assert_not_called() lock.release.assert_called_once_with() @@ -527,6 +542,7 @@ def test_run_on_change_warns_when_lock_already_released(mocker, loguru_logs): ) mocker.patch("osism.tasks.reconciler.utils.create_redlock", return_value=lock) proc = mocker.MagicMock() + proc.stdout = io.BytesIO(b"") mocker.patch("osism.tasks.reconciler.subprocess.Popen", return_value=proc) # Must not propagate.