File indexing completed on 2026-08-12 09:36:14
0001 """Terminalize workflow-execution rows abandoned as 'running'.
0002
0003 Before the runner finalized its execution record on every exit path
0004 (swf-testbed workflow_runner.py, 2026-07-24), a crashed or killed run
0005 left its row claiming 'running' forever, and every surface reading the
0006 record repeated the lie ("26 executions running" beside all-idle
0007 lanes). This one-shot repair marks such rows terminated, with the end
0008 time taken from the run record's last activity for the execution's run
0009 when one exists, else the execution's own start time. The runner fix
0010 prevents new ones; there is no scheduled janitor — a system needing
0011 one is writing garbage.
0012
0013 Dry-run by default; --apply writes. A row is stuck when status is
0014 'running' with no end_time and a start older than --hours (default 12).
0015 """
0016
0017 from datetime import timedelta
0018
0019 from django.core.management.base import BaseCommand
0020 from django.db.models.fields.json import KeyTextTransform
0021 from django.utils import timezone
0022
0023 from monitor_app.models import RunState, SystemStateEvent
0024 from monitor_app.workflow_models import WorkflowExecution
0025
0026
0027 class Command(BaseCommand):
0028 help = __doc__
0029
0030 def add_arguments(self, parser):
0031 parser.add_argument('--apply', action='store_true',
0032 help='write the repair (default: report only)')
0033 parser.add_argument('--hours', type=float, default=12,
0034 help='running-with-no-end age threshold')
0035
0036 def handle(self, *args, **options):
0037 cutoff = timezone.now() - timedelta(hours=options['hours'])
0038 stuck = list(
0039 WorkflowExecution.objects
0040 .filter(status='running', end_time__isnull=True,
0041 start_time__lt=cutoff)
0042 .order_by('start_time'))
0043 execution_runs = {}
0044 for run_number, execution_key in (
0045 RunState.objects
0046 .annotate(execution_key=KeyTextTransform(
0047 'execution_id', 'metadata'))
0048 .exclude(execution_key__isnull=True)
0049 .values_list('run_number', 'execution_key')):
0050 execution_runs.setdefault(execution_key, run_number)
0051
0052 repaired = 0
0053 for execution in stuck:
0054 run_number = execution_runs.get(execution.execution_id)
0055 last_activity = None
0056 if run_number is not None:
0057 last_activity = (
0058 SystemStateEvent.objects
0059 .filter(run_number=run_number)
0060 .order_by('-timestamp')
0061 .values_list('timestamp', flat=True)
0062 .first())
0063 end_time = last_activity or execution.start_time
0064 self.stdout.write(
0065 f'{execution.execution_id}: running since '
0066 f'{execution.start_time:%Y-%m-%d %H:%M} -> terminated '
0067 f'at {end_time:%Y-%m-%d %H:%M}')
0068 if options['apply']:
0069 execution.status = 'terminated'
0070 execution.end_time = end_time
0071 execution.save(update_fields=['status', 'end_time'])
0072 repaired += 1
0073
0074 mode = 'APPLIED' if options['apply'] else 'DRY RUN'
0075 self.stdout.write(f'{mode}: {repaired} stuck executions')
