src/monitor_app/run_state_transitions.py

0001 """Apply run lifecycle messages to RunState — the E0-E1 state write side.
0002
0003 The E0-E1 state machine (swf-testbed docs/e0-e1-state-machine.md) stamps
0004 every run lifecycle message with the state and substate in effect, and
0005 prescribes concurrent recording in the testbed database. The workflow
0006 runner creates the initial RunState row at launch; this module advances
0007 it from the messages as the monitor's ActiveMQ processor records them —
0008 one consumer, one truth. Without it every namespace lane on the Snapper
0009 Time history shows its launch-time state forever.
0010
0011 Vocabulary: state/substate copy the message stamps verbatim (beam/
0012 not_ready, run/physics, run/standby). The terminal marker follows the
0013 established fast-processing convention the read side already renders:
0014 state 'ended', phase 'completed'. The lane active check accepts both
0015 'run' (stamped) and 'running' (fast-processing legacy).
0016 """
0017
0018 import logging
0019
0020 from django.utils import timezone
0021
0022 logger = logging.getLogger(__name__)
0023
0024 RUN_LIFECYCLE_TYPES = (
0025     'run_imminent', 'start_run', 'pause_run', 'resume_run', 'end_run',
0026 )
0027
0028
0029 def apply_run_lifecycle_message(data) -> bool:
0030     """Advance RunState from one run lifecycle message; True if applied."""
0031     from .models import RunState
0032
0033     msg_type = data.get('msg_type')
0034     if msg_type not in RUN_LIFECYCLE_TYPES:
0035         return False
0036     try:
0037         run_number = int(str(data.get('run_id')))
0038     except (TypeError, ValueError):
0039         logger.error(
0040             "run lifecycle message %s carries no usable run_id: %r",
0041             msg_type, data.get('run_id'))
0042         return False
0043
0044     now = timezone.now()
0045     # Agents may start mid-run or belong to workflows whose launcher did
0046     # not create the row; transitions must never fall on the floor.
0047     row, created = RunState.objects.get_or_create(
0048         run_number=run_number,
0049         defaults={
0050             'phase': 'initializing',
0051             'state': 'imminent',
0052             'substate': 'preparing',
0053             'target_worker_count': 0,
0054             'state_changed_at': now,
0055             'metadata': {},
0056         },
0057     )
0058
0059     previous = (row.phase, row.state, row.substate)
0060     if msg_type == 'end_run':
0061         row.state = 'ended'
0062         row.substate = None
0063         row.phase = 'completed'
0064     else:
0065         stamped_state = data.get('state')
0066         if stamped_state:
0067             row.state = str(stamped_state)
0068         stamped_substate = data.get('substate')
0069         if stamped_substate:
0070             row.substate = str(stamped_substate)
0071         if msg_type in ('start_run', 'pause_run', 'resume_run'):
0072             row.phase = 'physics'
0073     row.state_changed_at = now
0074
0075     # The datataking projection joins namespace through
0076     # metadata.execution_id; rows created here must carry it too.
0077     metadata = row.metadata if isinstance(row.metadata, dict) else {}
0078     execution_id = data.get('execution_id')
0079     if execution_id and not metadata.get('execution_id'):
0080         metadata['execution_id'] = execution_id
0081         row.metadata = metadata
0082
0083     row.save(update_fields=[
0084         'phase', 'state', 'substate', 'state_changed_at', 'metadata',
0085         'updated_at',
0086     ])
0087     logger.info(
0088         "RunState %s: %s -> %s/%s (%s)%s",
0089         run_number, msg_type, row.state, row.substate or '-', row.phase,
0090         ' [row created]' if created else '')
0091
0092     # The datataking snapper component republishes on REST-path state
0093     # changes (RunStateViewSet); this ORM path must do the same or the
0094     # lanes go stale despite a correct RunState.
0095     if created or (row.phase, row.state, row.substate) != previous:
0096         try:
0097             from .snapper_datataking import publish_datataking_state
0098             publish_datataking_state()
0099         except Exception:
0100             logger.exception(
0101                 "datataking republish failed after RunState %s %s",
0102                 run_number, msg_type)
0103     return True