From dc2f4266bca0473c5dc539cddc3707a85a21fb67 Mon Sep 17 00:00:00 2001 From: "benjamin.baker" Date: Sun, 30 Aug 2026 14:32:08 -0400 Subject: [PATCH] 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 | 40 ++++++++++++++++ plugins/endpoints/models/endpoint.py | 15 +++++- tests/test_endpoint_command_reaper.py | 69 +++++++++++++++++++++++++++ 3 files changed, 123 insertions(+), 1 deletion(-) create mode 100644 tests/test_endpoint_command_reaper.py diff --git a/plugins/endpoints/api/routes.py b/plugins/endpoints/api/routes.py index 043aeb4c..b83cf4d8 100644 --- a/plugins/endpoints/api/routes.py +++ b/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_(( diff --git a/plugins/endpoints/models/endpoint.py b/plugins/endpoints/models/endpoint.py index c3e543c2..e0da120d 100644 --- a/plugins/endpoints/models/endpoint.py +++ b/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) diff --git a/tests/test_endpoint_command_reaper.py b/tests/test_endpoint_command_reaper.py new file mode 100644 index 00000000..1c912fdc --- /dev/null +++ b/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