MQTT Telemetry Contract¶

Vehicles (or simulators) publish telemetry to the NanoMQ broker. The realtime-engine Celery worker subscribes to these topics via the MQTTConsumerStep bootstep and ingests them into Redis.

Broker: NanoMQ at telemetry-broker:1883 (internal) / mqtt.<domain>:8883 (TLS, production).

Topic grammar¶

transit/vehicle/<vehicle_id>/<leaf>

The MQTT consumer uses + wildcard subscriptions:

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

Supported leaves¶

Only two leaves are actively subscribed. Any other leaf is silently dropped.

position¶

Reports the vehicle's current GPS fix and motion data.

Topic: transit/vehicle/<vehicle_id>/position

QoS: 0 (at most once)

Payload (JSON):

{
    "latitude": 9.9341,
    "longitude": -84.0875,
    "bearing": 45.0,
    "speed": 8.3,
    "odometer": 12350.5,
    "timestamp": 1718800000
}

A message missing latitude or longitude is rejected and dropped. A message with a valid position but an invalid optional field (e.g., non-numeric bearing) is also rejected and dropped.

Effect of a valid position message:

1. validate_for_write(data) validates the payload.
2. r.hset(vehicle:<id>:position, mapping=...) writes the hash.
3. process_position_update.delay(run_id, vehicle_id) is enqueued for server-side processing (map-matching, detection, projections).
4. r.set(runs:last_seen:<run_id>, now().isoformat()) is written synchronously.

occupancy¶

Reports passenger load.

Topic: transit/vehicle/<vehicle_id>/occupancy

QoS: 0 (at most once)

Payload (JSON):

{
    "occupancy_percentage": 65,
    "occupancy_count": 32
}

Effect of a valid occupancy message:

1. occupancy_status in the payload is stripped.
2. classify_status(occupancy_percentage) computes the server-policy enum value.
3. validate_for_write(occ_payload) validates the combined payload.
4. r.hset(vehicle:<id>:occupancy, mapping=...) writes the hash.
5. detect_from_telemetry(run_id, vehicle_id, "occupancy", data) is called inline (not via the Celery queue) — occupancy detection is cheap and must fire lifecycle events immediately for tracking-start and restore detectors.
6. r.set(runs:last_seen:<run_id>, now().isoformat()) is written synchronously.

Routing: message dropped if no active run¶

Before processing any leaf, the consumer reads vehicle:<id>:current_run from Redis. If that key is absent or empty, the message is dropped with a DEBUG log. No telemetry is processed for vehicles not currently assigned to an active run.

This prevents processing noise from vehicles that are off-duty, parked, or whose run has already ended.

Decommissioned leaf: progression¶

Duplicate-consumer protection¶

Each consumer instance generates a unique MQTT client_id:

databus-mqtt-consumer-<hostname>-<pid>

A fixed client_id causes a second consumer process (e.g., a worker that accidentally has MQTT_CONSUMER_ENABLED=true) to collide with the first on the broker, triggering an endless reconnect war. The per-process client_id makes such accidental duplicates non-catastrophic, though the single-consumer gate (MQTT_CONSUMER_ENABLED) is the real guarantee.

Source: backend/realtime_engine/mqtt.py::build_client, commit 452d4f4.

Untrusted edge signal stance¶

All incoming MQTT data is treated as an untrusted signal. The server:

• Validates every payload before writing to Redis.
• Recomputes occupancy_status regardless of what the edge sends.
• Never trusts progression data from the edge (decommissioned).
• Drops messages for vehicles without an active run.

This prevents malformed or malicious telemetry from corrupting the operational state or causing incorrect lifecycle events.

Environment variables¶

See Operations › Configuration for the full environment variable reference.