From a53251c0715caa0ec80067898347a298c01d920e Mon Sep 17 00:00:00 2001 From: Roger Luethi Date: Tue, 9 Jun 2026 14:27:29 +0200 Subject: [PATCH 1/2] Fail fast when reconciler run cannot start An explicit "osism sync inventory" dispatches reconciler.run and waits on its output stream. The task had two paths that left that client without an outcome: - When tasks are administratively locked, check_task_lock_and_exit() calls exit(1). The resulting SystemExit propagated out of the Celery task, churning the prefork worker and publishing no terminal marker. - When another run already held the execution lock, run() silently returned None, published nothing, and the waiting client blocked until its own output-inactivity timeout before failing. Convert both into a definite result. The admin-lock SystemExit is caught and turned into a terminal marker plus a non-zero return; lock contention now fails fast with a terminal marker instead of blocking or returning None. Contention is not retried: a retry could acquire a lease the in-flight run let expire and start a second concurrent /run.sh against shared inventory state. When no client is waiting (publish=False) the failure is logged instead of streamed. The periodic run_on_change path is left unchanged; it keeps coalescing on contention and is hardened separately. Upside: on both paths a waiting client now fails within seconds with a clear message instead of stalling until its 300s output-inactivity timeout. Tradeoffs and remaining limits: - The webhook (api.py) and watchdog (service.py) callers run with publish=True, so they now push the failure marker (three stream entries) to an output stream nobody reads and that has no TTL. This is marginal waste on paths that previously skipped silently, on top of the pre-existing unbounded leak of successful runs' output. - The admin-lock path now records the Celery task as SUCCESS with result 1 where a propagating SystemExit previously recorded FAILURE. Nothing in-repo consumes the AsyncResult -- the CLI watches the stream -- so this only changes external observability such as flower. - _publish_lock_failure and the success path still do unguarded Redis writes; an exception there escapes with no terminal marker. This narrows the no-outcome surface rather than closing it -- the same windows exist on main. - Scope is reconciler-only; other task sites keep the same SystemExit churn. Assisted-by: Claude:claude-opus-4-8 Signed-off-by: Roger Luethi --- osism/tasks/reconciler.py | 93 ++++++++++----- tests/unit/tasks/test_reconciler.py | 157 +++++++++++++++++++++++++ tests/unit/tasks/test_task_wrappers.py | 16 +-- 3 files changed, 230 insertions(+), 36 deletions(-) create mode 100644 tests/unit/tasks/test_reconciler.py diff --git a/osism/tasks/reconciler.py b/osism/tasks/reconciler.py index dab6796f..3ea8bfab 100644 --- a/osism/tasks/reconciler.py +++ b/osism/tasks/reconciler.py @@ -12,6 +12,19 @@ app = Celery("reconciler") app.config_from_object(Config) +# How long an explicit run waits for the execution lock before giving up. +LOCK_ACQUIRE_TIMEOUT = 20 +# Result reported when a run cannot start (admin lock or lock contention). +LOCK_TIMEOUT_RC = 1 + + +def _publish_lock_failure(task_id, message): + # Publish a terminal marker so a client waiting on the output stream + # (osism sync inventory) fails fast with this message instead of blocking + # until its own output timeout expires. + utils.push_task_output(task_id, f"{message}\n") + utils.finish_task_output(task_id, rc=LOCK_TIMEOUT_RC) + @app.on_after_configure.connect def setup_periodic_tasks(sender, **kwargs): @@ -26,47 +39,71 @@ def setup_periodic_tasks(sender, **kwargs): @app.task(bind=True, name="osism.tasks.reconciler.run") def run(self, publish=True): - # Check if tasks are locked before execution - utils.check_task_lock_and_exit() + try: + utils.check_task_lock_and_exit() + except SystemExit: + # check_task_lock_and_exit() calls exit(1) when tasks are + # administratively locked. Convert that into a terminal result rather + # than letting SystemExit propagate out of the Celery task: a bare + # SystemExit churns the prefork worker and leaves a waiting client + # without an outcome. + message = "Tasks are locked; reconciler did not run" + logger.error(message) + if publish: + _publish_lock_failure(self.request.id, message) + return LOCK_TIMEOUT_RC lock = utils.create_redlock( key="lock_osism_tasks_reconciler_run", auto_release_time=60, ) - if lock.acquire(timeout=20): - logger.info("RUN /run.sh") + if not lock.acquire(timeout=LOCK_ACQUIRE_TIMEOUT): + # Another reconciler run holds the execution lock. Fail fast with a + # terminal marker instead of blocking or silently returning: a waiting + # client must see a definite outcome. We do not retry here -- a retry + # could acquire a lease the in-flight run let expire and start a second + # concurrent /run.sh. + message = "Reconciler busy; another run holds the execution lock" + logger.error(message) + if publish: + _publish_lock_failure(self.request.id, message) + return LOCK_TIMEOUT_RC - env = os.environ.copy() + logger.info("RUN /run.sh") - p = subprocess.Popen( - "/run.sh", - shell=True, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - env=env, - ) + env = os.environ.copy() - # 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) + p = subprocess.Popen( + "/run.sh", + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + env=env, + ) - rc = p.wait(timeout=60) + # 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) - if publish: - utils.finish_task_output(self.request.id, rc=rc) + rc = p.wait(timeout=60) - from pottery import ReleaseUnlockedLock + if publish: + utils.finish_task_output(self.request.id, rc=rc) - try: - lock.release() - except ReleaseUnlockedLock: - logger.warning( - "Lock auto-released before explicit release (auto_release_time exceeded)" - ) + from pottery import ReleaseUnlockedLock + + try: + lock.release() + except ReleaseUnlockedLock: + logger.warning( + "Lock auto-released before explicit release (auto_release_time exceeded)" + ) + + return rc @app.task(bind=True, name="osism.tasks.reconciler.run_on_change") diff --git a/tests/unit/tasks/test_reconciler.py b/tests/unit/tasks/test_reconciler.py new file mode 100644 index 00000000..fe0b5159 --- /dev/null +++ b/tests/unit/tasks/test_reconciler.py @@ -0,0 +1,157 @@ +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the reconciler Celery tasks. + +The explicit ``run`` task must always hand a waiting client a definite +outcome: when it cannot run (administrative lock or execution-lock +contention) it publishes a terminal marker and reports a non-zero result +instead of blocking or silently returning ``None``. The periodic +``run_on_change`` task keeps its coalescing silent-skip behaviour. +""" + +import io +from unittest.mock import ANY, MagicMock + +from osism.tasks import reconciler + + +def test_run_fails_fast_on_contention_with_terminal_marker(mocker): + lock = MagicMock() + lock.acquire.return_value = False + mocker.patch("osism.tasks.reconciler.utils.check_task_lock_and_exit") + mocker.patch("osism.tasks.reconciler.utils.create_redlock", return_value=lock) + popen = mocker.patch("osism.tasks.reconciler.subprocess.Popen") + push = mocker.patch("osism.tasks.reconciler.utils.push_task_output") + finish = mocker.patch("osism.tasks.reconciler.utils.finish_task_output") + + result = reconciler.run.run(publish=True) + + assert result == reconciler.LOCK_TIMEOUT_RC + push.assert_called_once() + assert push.call_args.args[1].strip() + finish.assert_called_once_with(ANY, rc=reconciler.LOCK_TIMEOUT_RC) + popen.assert_not_called() + lock.release.assert_not_called() + + +def test_run_contention_without_publish_logs_and_returns_nonzero(mocker): + lock = MagicMock() + lock.acquire.return_value = False + mocker.patch("osism.tasks.reconciler.utils.check_task_lock_and_exit") + mocker.patch("osism.tasks.reconciler.utils.create_redlock", return_value=lock) + popen = mocker.patch("osism.tasks.reconciler.subprocess.Popen") + push = mocker.patch("osism.tasks.reconciler.utils.push_task_output") + finish = mocker.patch("osism.tasks.reconciler.utils.finish_task_output") + + result = reconciler.run.run(publish=False) + + assert result == reconciler.LOCK_TIMEOUT_RC + push.assert_not_called() + finish.assert_not_called() + popen.assert_not_called() + + +def test_run_converts_admin_lock_exit_to_terminal_marker(mocker): + mocker.patch( + "osism.tasks.reconciler.utils.check_task_lock_and_exit", + side_effect=SystemExit(1), + ) + create_lock = mocker.patch("osism.tasks.reconciler.utils.create_redlock") + push = mocker.patch("osism.tasks.reconciler.utils.push_task_output") + finish = mocker.patch("osism.tasks.reconciler.utils.finish_task_output") + + result = reconciler.run.run(publish=True) + + assert result == reconciler.LOCK_TIMEOUT_RC + push.assert_called_once() + finish.assert_called_once_with(ANY, rc=reconciler.LOCK_TIMEOUT_RC) + # the admin-lock check short-circuits before the execution lock is taken + create_lock.assert_not_called() + + +def test_run_admin_lock_without_publish_returns_nonzero_quietly(mocker): + mocker.patch( + "osism.tasks.reconciler.utils.check_task_lock_and_exit", + side_effect=SystemExit(1), + ) + push = mocker.patch("osism.tasks.reconciler.utils.push_task_output") + finish = mocker.patch("osism.tasks.reconciler.utils.finish_task_output") + + result = reconciler.run.run(publish=False) + + assert result == reconciler.LOCK_TIMEOUT_RC + push.assert_not_called() + finish.assert_not_called() + + +def test_run_publishes_success_and_releases_lock(mocker): + lock = MagicMock() + lock.acquire.return_value = True + process = MagicMock() + # Real bytes so the production ``TextIOWrapper`` drain loop consumes the pipe. + process.stdout = io.BytesIO(b"line\n") + process.wait.return_value = 0 + mocker.patch("osism.tasks.reconciler.utils.check_task_lock_and_exit") + mocker.patch("osism.tasks.reconciler.utils.create_redlock", return_value=lock) + mocker.patch("osism.tasks.reconciler.subprocess.Popen", return_value=process) + push = mocker.patch("osism.tasks.reconciler.utils.push_task_output") + finish = mocker.patch("osism.tasks.reconciler.utils.finish_task_output") + + result = reconciler.run.run(publish=True) + + assert result == 0 + push.assert_called_once_with(ANY, "line\n") + finish.assert_called_once_with(ANY, rc=0) + lock.release.assert_called_once_with() + + +def test_run_without_publish_does_not_stream_but_releases(mocker): + lock = MagicMock() + lock.acquire.return_value = True + process = MagicMock() + # Real bytes so the production ``TextIOWrapper`` drain loop consumes the pipe. + process.stdout = io.BytesIO(b"noise\n") + process.wait.return_value = 0 + mocker.patch("osism.tasks.reconciler.utils.check_task_lock_and_exit") + mocker.patch("osism.tasks.reconciler.utils.create_redlock", return_value=lock) + mocker.patch("osism.tasks.reconciler.subprocess.Popen", return_value=process) + push = mocker.patch("osism.tasks.reconciler.utils.push_task_output") + finish = mocker.patch("osism.tasks.reconciler.utils.finish_task_output") + + result = reconciler.run.run(publish=False) + + assert result == 0 + push.assert_not_called() + finish.assert_not_called() + lock.release.assert_called_once_with() + + +def test_run_on_change_runs_and_releases_when_acquired(mocker): + lock = MagicMock() + lock.acquire.return_value = True + process = MagicMock() + # Real bytes so the production ``TextIOWrapper`` drain loop consumes the pipe. + process.stdout = io.BytesIO(b"noise\n") + process.wait.return_value = 0 + mocker.patch("osism.tasks.reconciler.utils.create_redlock", return_value=lock) + popen = mocker.patch( + "osism.tasks.reconciler.subprocess.Popen", return_value=process + ) + + reconciler.run_on_change.run() + + popen.assert_called_once() + process.wait.assert_called_once() + lock.release.assert_called_once_with() + + +def test_run_on_change_skips_silently_when_contended(mocker): + lock = MagicMock() + lock.acquire.return_value = False + mocker.patch("osism.tasks.reconciler.utils.create_redlock", return_value=lock) + popen = mocker.patch("osism.tasks.reconciler.subprocess.Popen") + + result = reconciler.run_on_change.run() + + assert result is None + popen.assert_not_called() diff --git a/tests/unit/tasks/test_task_wrappers.py b/tests/unit/tasks/test_task_wrappers.py index 78a47baa..6e6a6add 100644 --- a/tests/unit/tasks/test_task_wrappers.py +++ b/tests/unit/tasks/test_task_wrappers.py @@ -361,7 +361,7 @@ def test_worker_run_aborts_when_task_lock_active(mocker, module, worker): def test_reconciler_run_aborts_when_task_lock_active(mocker): - """A locked task exits before creating a Redlock or spawning ``/run.sh``.""" + """A locked task fails fast before creating a Redlock or spawning ``/run.sh``.""" mocker.patch( "osism.tasks.reconciler.utils.check_task_lock_and_exit", side_effect=SystemExit(1), @@ -369,15 +369,15 @@ def test_reconciler_run_aborts_when_task_lock_active(mocker): create = mocker.patch("osism.tasks.reconciler.utils.create_redlock") popen = mocker.patch("osism.tasks.reconciler.subprocess.Popen") - with pytest.raises(SystemExit): - reconciler.run.__wrapped__() + result = reconciler.run.__wrapped__(publish=False) + assert result == reconciler.LOCK_TIMEOUT_RC create.assert_not_called() popen.assert_not_called() -def test_reconciler_run_returns_none_when_lock_not_acquired(mocker): - """Without the lock, ``/run.sh`` is never spawned and the task returns ``None``.""" +def test_reconciler_run_fails_fast_when_lock_not_acquired(mocker): + """Without the lock, ``/run.sh`` is never spawned and the task reports failure.""" mocker.patch("osism.tasks.reconciler.utils.check_task_lock_and_exit") lock = mocker.MagicMock() lock.acquire.return_value = False @@ -386,14 +386,14 @@ def test_reconciler_run_returns_none_when_lock_not_acquired(mocker): ) popen = mocker.patch("osism.tasks.reconciler.subprocess.Popen") - result = reconciler.run.__wrapped__() + result = reconciler.run.__wrapped__(publish=False) create.assert_called_once_with( key="lock_osism_tasks_reconciler_run", auto_release_time=60 ) - lock.acquire.assert_called_once_with(timeout=20) + lock.acquire.assert_called_once_with(timeout=reconciler.LOCK_ACQUIRE_TIMEOUT) popen.assert_not_called() - assert result is None + assert result == reconciler.LOCK_TIMEOUT_RC def test_reconciler_run_publishes_output(mocker): From b3e5a5afdc48d900ab56d851170f7461e69bfc54 Mon Sep 17 00:00:00 2001 From: Roger Luethi Date: Fri, 17 Jul 2026 09:45:37 +0200 Subject: [PATCH 2/2] Clarify reconciler --task-timeout help text The --task-timeout help described the value as a timeout for a scheduled task that has not been executed yet, implying a queueing deadline. The value is passed to fetch_task_output, which uses it as an output-inactivity timeout: it resets every time a line arrives and only fires after that many seconds with no further output. Reword the help so operators set it for the behaviour it actually controls. Spell out the queued-task case explicitly rather than only denying it is a queueing deadline: a queued task produces no output while it waits, so that wait is not excluded from the timeout -- it counts against it in full, which is the most likely way an operator hits it. Assisted-by: Claude:claude-opus-4-8 Signed-off-by: Roger Luethi --- osism/commands/reconciler.py | 7 ++++++- tests/unit/commands/test_reconciler.py | 17 +++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/osism/commands/reconciler.py b/osism/commands/reconciler.py index 4f3fdbd4..aa5e38f9 100644 --- a/osism/commands/reconciler.py +++ b/osism/commands/reconciler.py @@ -40,7 +40,12 @@ def get_parser(self, prog_name): "--task-timeout", default=os.environ.get("OSISM_TASK_TIMEOUT", 300), type=int, - help="Timeout for a scheduled task that has not been executed yet", + help=( + "Seconds to wait for further task output before giving up. " + "This is an output-inactivity timeout (it resets on each line). " + "A queued task emits no output while it waits to run, so that " + "wait counts against this timeout in full." + ), ) return parser diff --git a/tests/unit/commands/test_reconciler.py b/tests/unit/commands/test_reconciler.py index 319489ae..4c5f952e 100644 --- a/tests/unit/commands/test_reconciler.py +++ b/tests/unit/commands/test_reconciler.py @@ -12,6 +12,23 @@ from osism.commands import reconciler +def test_task_timeout_help_describes_output_inactivity(): + cmd = reconciler.Sync(MagicMock(), MagicMock()) + parser = cmd.get_parser("test") + help_text = next( + action.help + for action in parser._actions + if "--task-timeout" in action.option_strings + ) + + # The value bounds how long the client waits for further task output. Time + # a task spends queued produces no output, so it counts against the + # timeout rather than being excluded from it. + assert "output" in help_text.lower() + assert "scheduled task that has not been executed" not in help_text.lower() + assert "queued" in help_text.lower() + + def test_sync_returns_nonzero_on_task_timeout(): cmd = reconciler.Sync(MagicMock(), MagicMock()) parsed_args = cmd.get_parser("test").parse_args([])