Fix silent reconciler skips on lock contention

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
 

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")

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([])

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()

tests/unit/tasks/test_task_wrappers.py

@@ -361,23 +361,23 @@ 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),
     )
     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):