Telemetry ingestion (MQTT)¶

Vehicles publish telemetry to NanoMQ over MQTT. The realtime-engine Celery worker picks it up through an in-process bootstep and routes it into Redis.

The MQTT consumer is a Celery bootstep¶

The consumer (backend/realtime_engine/mqtt.py) is a celery.bootsteps.StartStopStep — not a standalone service or a separate Celery task. It starts when the realtime-engine worker boots and shuts down when the worker shuts down.

paho's loop_start() runs the network loop in its own background thread, so ingestion never blocks Celery task execution in the same process.

realtime-engine worker
├─ Celery Pool (task execution)
└─ MQTTConsumerStep (bootstep)
   └─ paho network thread (loop_start)
        subscribes: transit/vehicle/+/position   QoS 0
                    transit/vehicle/+/occupancy  QoS 0

The bootstep is registered globally in backend/databus/celery.py:

from realtime_engine.mqtt import MQTTConsumerStep
app.steps["worker"].add(MQTTConsumerStep)

Why a bootstep and not a separate service?¶

The earlier design had a dedicated realtime-consumer container. That added an extra Docker build target, an extra bind-mount on the backend tree, and an extra process to monitor — at a scale the project does not require. paho's network loop already runs in its own thread, so there is no contention with Celery task execution.

The MQTT_CONSUMER_ENABLED gate¶

The bootstep is registered on every worker that loads databus.celery — including the schedule-engine worker and the Celery beat process. To prevent those workers from also subscribing to the broker, the bootstep checks MQTT_CONSUMER_ENABLED at startup:

MQTT_CONSUMER_ENABLED = os.getenv("MQTT_CONSUMER_ENABLED", "false").lower() in (
    "1", "true", "yes",
)

Only the realtime-engine compose service sets this variable to true. Other workers log a single INFO line and skip activation.

Per-process client ID (commit 452d4f4)¶

Each worker instance builds its MQTT client ID from the hostname and PID:

client_id = f"databus-mqtt-consumer-{socket.gethostname()}-{os.getpid()}"

Before this fix, a fixed client ID caused an endless reconnect war: the broker treats a second connection with the same ID as a takeover, forcing the first client to reconnect, which triggers another takeover, and so on. Unique per-process IDs mean an accidental duplicate subscriber connects cleanly instead of looping. The MQTT_CONSUMER_ENABLED gate is still the real defence; the unique ID prevents the worst-case failure mode if the gate is misconfigured.

Topic subscriptions¶

On connect, the consumer subscribes to exactly two wildcard patterns:

transit/vehicle/+/position    QoS 0
transit/vehicle/+/occupancy   QoS 0

progression is not subscribed — it is decommissioned at the edge. The server computes vehicle stop status via real GPS→polyline map-matching (see Map-matching & progression). If the simulator publishes a progression leaf, the consumer drops it silently at DEBUG log level.

# 'progression' is intentionally NOT subscribed — decommissioned.

Per-leaf pipeline¶

Topic parsing¶

The consumer extracts vehicle_id and leaf from the four-part topic:

transit / vehicle / <vehicle_id> / <leaf>

Malformed topics (wrong segment count, wrong prefix) are dropped immediately.

Common guard: active run lookup¶

Before processing any payload, the consumer looks up vehicle:<id>:current_run in Redis. If the key is absent the vehicle has no active run and the message is dropped:

run_id = r.get(keys.current_run_key(vehicle_id))
if not run_id:
    logger.debug("No active run for vehicle %s — dropping %s", vehicle_id, leaf)
    return

This avoids writing telemetry for vehicles that are not on a run, which would pollute Redis with orphaned hashes.

position leaf¶

1. Parse JSON, validate against the position contract (position.validate_for_write).
2. HSET vehicle:<id>:position with the validated mapping.
3. Enqueue process_position_update.delay(run_id, vehicle_id) — the heavy work (map-matching, stop-time projection, detection) runs off the network thread.
4. Write runs:last_seen:<run_id> synchronously (see below).

occupancy leaf¶

Occupancy is handled entirely inline (no Celery task):

1. Parse JSON.
2. Strip any edge-sent occupancy_status — it is a server policy decision:

   occ_payload[occupancy.OCCUPANCY_STATUS] = occupancy.classify_status(pct)
   

3. Validate and HSET vehicle:<id>:occupancy.
4. Call detect_from_telemetry(run_id, vehicle_id, "occupancy", data) directly.

last_seen write (always synchronous)¶

After processing any leaf, the consumer writes runs:last_seen:<run_id> synchronously — before returning from the callback:

r.set(keys.last_seen_key(run_id), now().isoformat())

This key drives the stale-run scanner (scan_stale_runs, every 30 s). Writing it synchronously ensures staleness detection is never delayed by queue backlog, even when the realtime_engine Celery queue is under load.

Consumer pipeline summary¶

flowchart TD
    A[MQTT message arrives] --> B{Parse topic\nvehicle_id + leaf}
    B -- invalid --> Z[Drop silently]
    B -- valid --> C{vehicle:id:current_run\nexists in Redis?}
    C -- no --> Z
    C -- yes --> D{leaf?}
    D -- position --> E[validate_for_write\nHSET vehicle:id:position]
    E --> F[process_position_update.delay\nrun_id, vehicle_id]
    F --> G[SET runs:last_seen:run_id]
    D -- occupancy --> H[classify_status\nvalidate_for_write\nHSET vehicle:id:occupancy]
    H --> I[detect_from_telemetry\noccupancy leaf]
    I --> G
    D -- other / progression --> J[logger.debug drop]

Environment variables¶

Related pages¶

• MQTT telemetry contract — payload shapes for position and occupancy.
• Server-side processing — what process_position_update does with the enqueued data.
• Data model: Redis keys — canonical key names and ownership.
• Configuration reference — all environment variables.