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.
This commit is contained in:
committed by
cproudlock
parent
50987c96ad
commit
5f8061c55d
@@ -938,6 +938,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.
|
||||
|
||||
@@ -1021,6 +1056,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_((
|
||||
|
||||
Reference in New Issue
Block a user