Self-heal endpoint command queue wedged by a dead in-flight command

The command queue is at-most-one-in-flight, but nothing ever expired a stuck command. When an agent claimed a command (RUNNING) then died before reporting a result - killed, uninstalled, or offline mid-flight - the row stayed in-flight forever and every operator action for that endpoint returned 409 'already in flight'. Reap in-flight rows past a staleness ceiling (15m for RUNNING, 60m for PENDING) to a new STATUS_EXPIRED at enqueue time, inside the same transaction, so the queue self-heals lazily the next time someone queues work. No background job and no schema migration - status is free-text.

plugins/endpoints/api/routes.py

@@ -873,6 +873,41 @@ _CLAIM_HOLD_SECONDS = 25
 _CLAIM_POLL_INTERVAL = 1.0
 
 
+def _reap_stale_commands(asset_id):
+    """Expire in-flight commands an agent will never finish for this endpoint.
+
+    The queue is at-most-one-in-flight: a single PENDING or RUNNING row blocks
+    every new command. If the agent is killed, uninstalled, or goes offline
+    after a command is claimed (RUNNING) or before it claims one (PENDING), that
+    row would pin the queue forever and every operator action returns 409. Reap
+    rows past their staleness ceiling to STATUS_EXPIRED so the queue self-heals.
+    Caller commits (this runs inside the enqueue transaction).
+    """
+    now = _naive_utcnow()
+    running_cutoff = now - timedelta(seconds=EndpointCommand.RUNNING_STALE_SECONDS)
+    pending_cutoff = now - timedelta(seconds=EndpointCommand.PENDING_STALE_SECONDS)
+
+    stale = (EndpointCommand.query
+             .filter(EndpointCommand.assetid == asset_id,
+                     db.or_(
+                         db.and_(
+                             EndpointCommand.status == EndpointCommand.STATUS_RUNNING,
+                             EndpointCommand.claimedat.isnot(None),
+                             EndpointCommand.claimedat < running_cutoff),
+                         db.and_(
+                             EndpointCommand.status == EndpointCommand.STATUS_PENDING,
+                             EndpointCommand.createdat < pending_cutoff)))
+             .all())
+    for command in stale:
+        command.status = EndpointCommand.STATUS_EXPIRED
+        command.completedat = now
+        if not command.resultmessage:
+            command.resultmessage = (
+                'Expired: agent did not report a result before the staleness '
+                'ceiling (agent offline, killed, or reinstalled mid-command).')
+    return stale
+
+
 def _collector_auth_or_error():
     """Authorize an agent-facing command route on the endpoints collector token.
 
@@ -956,6 +991,11 @@ def enqueue_endpoint_command(asset_id: int):
                               f'No endpoint state for asset {asset_id}',
                               http_code=404)
 
+    # Self-heal a queue wedged by a command the agent will never finish (killed,
+    # uninstalled, or offline mid-flight) before enforcing at-most-one-in-flight,
+    # so a dead in-flight row cannot permanently 409 every operator action.
+    _reap_stale_commands(asset_id)
+
     pending = (EndpointCommand.query
                .filter(EndpointCommand.assetid == asset_id,
                        EndpointCommand.status.in_((

plugins/endpoints/models/endpoint.py

@@ -124,11 +124,24 @@ class EndpointCommand(db.Model):
 
     # Lifecycle: pending -> running -> succeeded|failed. A command is claimed
     # (pending -> running) by exactly one agent poll, then finalized by its
-    # result post.
+    # result post. A command that is claimed but whose agent never reports back
+    # (agent killed/uninstalled/offline mid-flight) would otherwise pin the
+    # at-most-one guard forever; the enqueue path reaps such rows to STATUS_EXPIRED
+    # so a dead in-flight command cannot wedge the endpoint's command queue.
     STATUS_PENDING = 'pending'
     STATUS_RUNNING = 'running'
     STATUS_SUCCEEDED = 'succeeded'
     STATUS_FAILED = 'failed'
+    STATUS_EXPIRED = 'expired'
+
+    # In-flight staleness ceilings. A RUNNING command is expected to finish in
+    # seconds to a couple of minutes; the generous ceiling tolerates a slow
+    # app install while still freeing the queue when the agent is gone. A
+    # PENDING command is claimed within one poll (~25s) by a live agent, so a
+    # long-unclaimed row means the agent is offline and the queue should not
+    # stay wedged behind it.
+    RUNNING_STALE_SECONDS = 15 * 60
+    PENDING_STALE_SECONDS = 60 * 60
 
     endpointcommandid = db.Column(db.Integer, primary_key=True)
 

tests/test_endpoint_command_reaper.py

@@ -0,0 +1,69 @@
+"""Regression tests for the endpoint command-queue staleness reaper.
+
+A claimed command whose agent never reports back (agent killed, uninstalled, or
+offline mid-flight) would otherwise pin the at-most-one-in-flight queue forever,
+so every operator action against that endpoint returns 409 "already in flight".
+The enqueue path reaps such rows to STATUS_EXPIRED so the queue self-heals.
+"""
+
+from datetime import timedelta
+
+
+def test_reap_stale_commands_frees_a_wedged_queue(db):
+    from plugins.endpoints.models import EndpointCommand
+    from plugins.endpoints.api.routes import _reap_stale_commands, _naive_utcnow
+
+    now = _naive_utcnow()
+    asset_id = 4242
+
+    stale_running = EndpointCommand(
+        assetid=asset_id, action='mtconnect_start',
+        status=EndpointCommand.STATUS_RUNNING,
+        createdat=now - timedelta(hours=2),
+        claimedat=now - timedelta(hours=2))
+    stale_pending = EndpointCommand(
+        assetid=asset_id, action='mtconnect_stop',
+        status=EndpointCommand.STATUS_PENDING,
+        createdat=now - timedelta(hours=2))
+    fresh_running = EndpointCommand(
+        assetid=asset_id + 1, action='mtconnect_start',
+        status=EndpointCommand.STATUS_RUNNING,
+        createdat=now, claimedat=now)
+    db.session.add_all([stale_running, stale_pending, fresh_running])
+    db.session.commit()
+
+    reaped = _reap_stale_commands(asset_id)
+    db.session.commit()
+
+    reaped_ids = {command.endpointcommandid for command in reaped}
+    assert stale_running.endpointcommandid in reaped_ids
+    assert stale_pending.endpointcommandid in reaped_ids
+    assert stale_running.status == EndpointCommand.STATUS_EXPIRED
+    assert stale_pending.status == EndpointCommand.STATUS_EXPIRED
+    assert stale_running.completedat is not None
+    assert stale_running.resultmessage
+    # A live, recently claimed command on another endpoint is untouched.
+    assert fresh_running.status == EndpointCommand.STATUS_RUNNING
+
+
+def test_reap_leaves_a_recently_claimed_running_command_in_flight(db):
+    from plugins.endpoints.models import EndpointCommand
+    from plugins.endpoints.api.routes import _reap_stale_commands, _naive_utcnow
+
+    now = _naive_utcnow()
+    asset_id = 777
+
+    # Claimed a minute ago: a slow-but-alive command must not be reaped.
+    running = EndpointCommand(
+        assetid=asset_id, action='mtconnect_start',
+        status=EndpointCommand.STATUS_RUNNING,
+        createdat=now - timedelta(minutes=1),
+        claimedat=now - timedelta(minutes=1))
+    db.session.add(running)
+    db.session.commit()
+
+    reaped = _reap_stale_commands(asset_id)
+    db.session.commit()
+
+    assert reaped == []
+    assert running.status == EndpointCommand.STATUS_RUNNING