feat(run-lifecycle-mqtt): MQTT consumer + full run lifecycle pipeline#59
Open
dotjae wants to merge 6 commits into
Open
feat(run-lifecycle-mqtt): MQTT consumer + full run lifecycle pipeline#59dotjae wants to merge 6 commits into
dotjae wants to merge 6 commits into
Conversation
UpdateRunViewSet
- Flatten serializer.validated_data["details"] into payload before
process_event. Guards like is_*_authorized read actor_role at the
top level of payload (consistent with the Celery path in
realtime_engine/tasks.py), so without this every UI-driven
cancel/interrupt/short-turn/run_rejected returned 422 with
"actor_role '' is not authorized…".
Transitions
- INTERRUPT_RUN, SHORT_TURN_RUN, COMPLETE_RUN: add sync_lifecycle_state
so Redis's run:{id} hash reflects the terminal state. Without this
the simulator's RunBinder polled an "In Progress" value forever and
the UI never recognized the terminal state.
- RUN_TRACKING_LOST: stop removing the run from runs:tracking — the
set is scan_stale_runs's work queue, and removing on lost prevented
the eventual RUN_TRACKING_EXPIRED transition.
- RUN_TRACKING_EXPIRED: add sync_lifecycle_state + remove_from_tracking_set
so the run finally exits the queue when fully terminal.
Also includes RunHistoryView (read-only GET /api/runs/{id}/history/)
in views.py — drives the history panel in the simulator's Runs tab.
Note: committed against feat/run-lifecycle-mqtt which still has
unrelated WIP from another contributor (uncommitted) — only the two
files touched here are part of this fix.
Replaces the management-command mqtt_consumer with an in-worker
Celery bootstep at realtime_engine/mqtt.py. It subscribes to
transit/vehicle/+/{position,progression,occupancy}, updates Redis
on every ping, and emits lifecycle events (run_tracking_started,
run_started, complete_run, run_tracking_restored) based on the
current run state.
- backend/realtime_engine/mqtt.py: new MQTT bootstep
- backend/realtime_engine/management/commands/*: removed
(mqtt_consumer + bootstrap_simulator_runs no longer needed)
- backend/databus/celery.py: register the bootstep on workers that
set MQTT_CONSUMER_ENABLED=true
- backend/api/urls.py: add /api/runs/<uuid>/history/
- backend/Dockerfile, compose.dev.yml, compose.prod.yml: wire
telemetry-broker + realtime-engine environment so the consumer
attaches at startup
- backend/uv.lock: pick up paho-mqtt
- docs/content/processes/run-lifecycle.md, README, realtime_engine/README:
document the new flow
…update related views and URLs feat(run): add RunStateViewSet for retrieving current run lifecycle state docs: add README for run lifecycle states refactor(redis): change vehicle data key from 'data' to 'metadata' across scripts and tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
realtime_engine/mqtt.py): subscribes to vehicle telemetry broker, drives run lifecycle FSM transitions in real timeruns/domain/): complete guards, actions, and transitions for all lifecycle states — including flatten/sync fixes for terminal transitionsschedule_engine/tasks.py): ported publisher builders into the unified backend, removed legacypublisher/,scheduler/, andscripts/top-level service directoriesdatabus/celery.py): addedbuild_vehicle_positionsandbuild_trip_updatesbeat schedules (15 s cadence)schedule_engine/fake_stop_times.py): handle dict-shaped run/progression entries to unblock dev simulation