Message Processor

The message processor consumes raw ADS-B and UAT messages from its own RabbitMQ queue, maintains per-aircraft flight state in a file-backed (WAL-mode) SQLite database so it survives a process restart, enriches each flight with registration and operator data from Redis, evaluates the configured rules engine, publishes MQTT notifications when rules match, and routes completed flights to the skyfollower-archive queue (or a local SQLite fallback when RabbitMQ is unavailable). One container equals one message processor instance; scale horizontally by adding message processor containers, whether on the same host or on separate hosts -- see MESSAGE_PROCESSOR_ID below for how scripts/install.sh generates each instance's env file and service.

Configuration

Reads its configuration from environment variables via shared/config.py's load_config("rabbitmq", "redis", "mqtt", "message_processor", "map_udp"), interpolated by Compose from this host's .env (written by scripts/install.sh).

Variable	Required	Default	Description
RABBITMQ_HOST	✅	—
RABBITMQ_PORT	❌	5672
RABBITMQ_USERNAME	✅	—
RABBITMQ_PASSWORD	✅	—
REDIS_HOST	✅	—
REDIS_PORT	❌	6379
REDIS_PASSWORD	✅	—	Redis requires authentication; see shared/redis_client.py's build_redis_client()
MQTT_HOST	❌	—	Leave unset to disable MQTT entirely
MQTT_PORT	❌	1883
MQTT_USERNAME	❌	—	Optional MQTT auth; leave unset for an anonymous broker
MQTT_PASSWORD	❌	—
MAP_UDP_HOST	❌	—	Destination host for the live position/metadata/heartbeat UDP feed toward the map service (see Map UDP Publisher). Leave unset to disable entirely
MAP_UDP_PORT	❌	—
LATITUDE	✅	—	Receiver location latitude (decimal degrees). Used as the CPR reference for surface/taxi position decoding
LONGITUDE	✅	—	Receiver location longitude (decimal degrees)
CAPTURE_RAW_FRAMES	❌	false	Opt-in forensic raw-frame capture (see Raw Frame Capture below). install.sh writes CAPTURE_RAW_FRAMES=false into the node's .env (and --upgrade appends it if missing, leaving an existing value alone). Read once at startup; restart to pick up a changed value
LOG_LEVEL	❌	info	"debug" for verbose output

Timing values -- the MQTT publish cadence, the Redis heartbeat refresh and TTL, the config-poll interval, the rule-notification max lag, and reconnect backoffs -- are not environment variables. They are fixed constants in shared/timing.py; see Timing and cadences. flight_ttl_seconds remains adjustable via the config:flight_ttl_seconds Redis key (see Redis Key Dependencies).

active_flights.db (the durable active flight store) and completed_flights.db (the RabbitMQ offline fallback) are always written to /app/data, a fixed, non-configurable bind mount -- see docker-compose.message-processor.yaml.

MESSAGE_PROCESSOR_ID

MESSAGE_PROCESSOR_ID is a single flat, fleet-wide sequential number -- there is exactly one ID per processor across the whole deployment, not a per-node prefix plus a local index. It's set per instance in message-processors/{id}.env (referenced by that instance's env_file, not read from the shared .env), because scripts/install.sh decides it at generation time, not the Python process at first boot: the compose service name and container name are static values resolved at docker compose up time, so they can't depend on something a container only discovers after it's already running.

That one ID is used verbatim -- with no local/global translation -- for every artifact belonging to that processor:

What	Format
Compose service name	skyfollower-message-processor-{id}
Docker container name	skyfollower-message-processor-{id}
RabbitMQ queue name	skyfollower-message-processor-{id}
Redis heartbeat/claim key	skyfollower-message-processor-{id}
Data directory	./data/skyfollower-message-processor-{id}

The message processor declares and consumes from skyfollower-message-processor-{MESSAGE_PROCESSOR_ID}, binding that queue to the skyfollower-adsb consistent-hash exchange with a weight of 1 (see Routing in the receiver's README for the exchange's shape and its operational consequences). Because the ID is embedded in the queue name, an abandoned queue is identifiable by name alone: the RabbitMQ management UI shows skyfollower-message-processor-7 with a consumer count of zero once that message processor is gone.

On startup the message processor attempts to claim a Redis key skyfollower-message-processor-{MESSAGE_PROCESSOR_ID} using SET NX. If the key already exists (i.e., another instance with the same ID is running), the process exits immediately to prevent duplicate-ID conflicts.

The ID is sequential (operator-supplied count), not random, deliberately: scale-down safety depends on operators being able to reason about creation order -- remove the highest-numbered/most-recently-bound instance to avoid a large RabbitMQ consistent-hash reshuffle (see Routing's "Remove from the end, never the middle" guidance). A random ID would solve cross-node uniqueness just as well but would destroy that property.

docker-compose.message-processor.yaml, as fetched from the repo, is static and re-fetched on every install and upgrade: the shared x-message-processor-environment anchor and one profile-gated message-processor template service that never runs on its own. scripts/install.sh's collect_message_processor_env() asks whether this run is replacing an existing processor (adopts and confirms one specific ID) or adding new ones (asks how many are currently implemented fleet-wide and how many this host will add, then computes the new range as existing_count+1 through existing_count+num_new), and for each ID writes message-processors/{id}.env (mode 0600, containing MESSAGE_PROCESSOR_ID) and a skyfollower-message-processor-{id} service in the generated docker-compose.instances.yaml, which extends the template with container_name, env_file and the data volume. The role's .env sets COMPOSE_FILE to both compose files. Re-running it later to add more processors to the same node only adds IDs with no env file yet and leaves already-running ones untouched.

An install that still holds generated service blocks inside docker-compose.message-processor.yaml is converted by install.sh --upgrade before that file is replaced: each block becomes an env file and an entry in docker-compose.instances.yaml, the original is saved as docker-compose.message-processor.yaml.bak, and service names, container names and data directories are unchanged. See docker-compose.message-processor.yaml's own comments for why deploy.replicas can't substitute for any of this (each replica would need its own volume and its own derivable ID, and Compose doesn't provide either).

Decoding

Raw Mode-S/ADS-B frames are decoded via pyModeS 3.x's single unified decode() call, which returns every decodable field for a message in one dict. The message processor extracts fields purely by presence — if a field is in the result, it's used; there's no downlink-format or typecode dispatch, and no downlink-format allowlist. Message types that don't populate any field the message processor cares about (e.g. ACAS RA broadcasts) simply produce nothing and are dropped, with nothing to explicitly filter.

Any message pyModeS flags as CRC-invalid is rejected outright — but this only provides real protection for DF17/18 (extended squitter), where crc_valid reflects an actual CRC-remainder-equals-zero check. For DF0/4/5/11/16/20/21 (including squawk, DF5/21), pyModeS hardcodes crc_valid=True unconditionally, since their CRC field encodes the ICAO address itself rather than providing an independent integrity check — there's no single-message corruption signal available for those message types at all. A squawk value is trusted once decoded; there's nothing further to verify it against outside of multi-message/pipe-mode decoding, which this message processor doesn't use.

wake_turbulence_category is receiver-decode-only — registry/Mictronics enrichment is never allowed to seed it (shared/enrichment.py's split_merged_aircraft() strips the key from the Redis-merged aircraft document before it reaches flight state), and each new live reading simply replaces the last one (no first-wins protection; there's only one writer). A single shared mapping table (_WAKE_TURBULENCE_MAP) collapses both decode paths' possible source values down to three canonical strings — light, medium, heavy:

• 1090 (_decode_1090): pyModeS's wake_vortex field (TC=4 aircraft category only — pyModeS is aware of which identification sub-type a category code came from, so TC=3 gliders/UAVs and TC=2 surface vehicles never produce a wake_vortex value in the first place).
• 978 (_decode_978): pyModeS978's category field (an EmitterCategory enum member), via the same table.

Both libraries expose the same 7 DO-260B wake/emitter categories (Light, Medium 1, Medium 2, High Vortex Aircraft, Heavy, High Performance, Rotorcraft). Super has no code point in this data at all — an A380 and a 767 broadcast the identical Heavy value — so it's not a reachable output. Rotorcraft/High Performance describe emitter type, not wake-turbulence weight class, so they're intentionally left unset rather than mapped to anything; the remaining 5 collapse to light/medium/heavy by weight band.

The same source field is also forwarded verbatim, as the raw <set><subcategory> code the aircraft broadcasts (A1–A7, B1/B2/ B4/B6/B7, C1…, _emitter_category_code). It rides in the flight's aircraft dict as emitter_category — first-sighting-wins, like adsb_version, and likewise not part of the AircraftRecord enrichment model. Its only consumer is the map service, which uses it as a last-resort icon-shape hint for aircraft that carry no type enrichment at all (no type_designator, no Mictronics hit); wake_turbulence_category derivation is unaffected by it.

Redis Key Dependencies

Keys read

Key pattern	Purpose
EVALSHA → shared/lua/merge_aircraft.lua	Aircraft registration and type enrichment (read once per new flight). Not a direct key read: the message processor calls this script (SCRIPT LOADed at startup, and reloaded once on NoScriptError after a Redis restart flushes the script cache) with icao_hex as its sole argument and has no visibility into what it reads. The script itself performs the underlying JSON.GETs against aircraft:mictronics:{icao_hex}, aircraft:registry:{icao_hex}, and aircraft:livery:{icao_hex} server-side and returns the deep-merged result (each later source winning on any field overlap — livery over registry over mictronics) in a single round-trip.
operator:{DESIGNATOR}	Airline operator enrichment (read once per flight when ident is first seen)
EVALSHA → shared/lua/route_airports.lua	Resolves route:{ident} into its full ordered array of airport:{code} records. Read at most once per flight, the moment ident/position/altitude/heading are all known (not on every message, and not at archive time). See "Route Leg Resolution" below.
config:rules:version	SHA-256 hash polled every CONFIG_POLL_INTERVAL_SECONDS (30 s); a change triggers a rule reload. Used only as a fast-path "unchanged" signal — a missing or body-skewed version key falls through to hashing config:rules itself (see below), so it can't leave a processor silently ruleless
config:rules	JSON rules array; loaded when the version changes, when its own SHA-256 doesn't match the recorded version, or whenever no rules are currently loaded
config:areas:version	SHA-256 hash polled every 30 s; same fast-path / fall-through behaviour as config:rules:version
config:areas	GeoJSON FeatureCollection; loaded on the same conditions as config:rules
config:flight_ttl_seconds	Plain scalar, read once at startup and cached (not hot-reloaded — restart to pick up a changed value); defaults to 300 if unset. Shared with the archive processor, which uses the same value to detect flights split by a processor-count resize.

Keys written

Key pattern	Purpose
skyfollower-message-processor-{ID}	Liveness key; claimed with NX on startup, refreshed every HEARTBEAT_INTERVAL_SECONDS with a HEARTBEAT_TTL_SECONDS TTL (see shared/timing.py)
registration:{REGISTRATION}	Reverse-lookup index (registration → ICAO hex); written NX when aircraft enrichment is found and a registration exists
metrics:message_processor:{ID}:registration_misses:{hour|today|lifetime}	Incremented each time an icao_hex aircraft enrichment lookup (_enrich_aircraft()) returns no result. Operator-lookup misses are not counted here — they have their own dedicated operator_misses key below.
metrics:message_processor:{ID}:operator_misses:{today|lifetime}	Incremented each time an operator:{designator} lookup (_enrich_operator()) returns no result. No hour period — operator misses are lower-volume and only tracked today/lifetime.
metrics:message_processor:{ID}:total_messages_processed:{hour|today|lifetime}	Incremented for every message an attempt was made to decode (including CRC-corrupt/no-content messages) — the same point messages_per_second's own rate tracker records at, in _on_message. Bucketed by wall-clock processing time, not received_at — a backlog replay counts entirely toward the current hour/day, not the messages' original timestamps.
rule_triggers:{identifier}:lifetime and rule_triggers:{identifier}:{YYYY-MM-DD}	How many times each rule has fired — counted once per flight the first time a rule matches it (the per-matched-rule loop in _update_flight runs once per (flight, rule), since RulesEngine.evaluate() skips a rule already in flight.matched_rules), not once per message. Keyed by rule identifier, so a deleted-then-recreated identifier starts clean. Same non-blocking pattern as the three counters above — _KeyedCounterAccumulator on the hot path, flushed by the telemetry thread via _flush_rule_trigger_counts() in one pipelined round trip: INCRBY the never-expiring lifetime key, INCRBY + EXPIRE (RULE_TRIGGER_DAY_TTL_SECONDS, 31 days) today's UTC day key. The management-ui backend reads and sums these on demand for the Rules editor's "Triggered since created" / "Triggered in last 30 days" display; management-ui's DELETE /api/rules/{identifier} removes a deleted rule's keys.

All three period counters above share the same reset mechanism and are write-only from this component's own perspective:

• Real clock-boundary resets: hour/today periods are written via EVALSHA against shared/lua/incr_period_counter.lua (SCRIPT LOADed once at startup into self._incr_period_counter_sha, matching the merge_aircraft.lua/route_airports.lua pattern above), which sets EXPIREAT to the real next UTC hour/midnight boundary (shared/metrics.py's next_period_boundary()) only the instant a call creates the key — never on a later increment within the same period, so the window can't slide forward. Redis's own TTL expiry deletes the key at the boundary; the next increment after that recreates it fresh, a genuine reset with no external scheduler involved.
• lifetime periods have no TTL and instead are explicitly DELETEd once, in start(), before any message is processed — "lifetime" means "since this process instance started," not "forever." Redis is a separate, persistent service, so a container restart alone does not clear these keys.
• No per-message Redis round trip: _on_message/_enrich_aircraft/ _enrich_operator only ever touch a small in-memory accumulator (_CounterAccumulator, a lock-protected pending count — record()/ flush_and_reset()). The existing telemetry thread (already running on its own cadence, already off the message-consuming path) is the only place that ever flushes an accumulated delta into Redis, via _flush_period_counters().
• Not self-published over MQTT/Home Assistant. core-health reads these three Redis key families and publishes them on this component's behalf, using the exact topic paths/unique_id/device block this component's own _publish_telemetry()/_publish_ha_autodiscovery() would otherwise have used — nothing on the wire distinguishes the two. See core-health/README.md's "Counter mimicry" section.

Route Leg Resolution

origin/destination are resolved at most once per flight — not at archive time, and not on every message. _maybe_resolve_route() is called from _update_flight() after each message is applied to the flight's state, and runs the actual Redis lookup and resolution logic the moment all of the following have been seen for the flight (in any order across messages):

• a route-bearing ident — present, not "00000000" (both already enforced before flight.ident is ever set at all — see _update_flight), and not just the aircraft's own tail number re-broadcast as the callsign (shared/enrichment.py's ident_matches_registration(), dash-insensitive — the same check _enrich_operator() uses to decide whether to look up an airline operator, matching SkyFollower-legacy's setIdent()/_getOperator() precedent)
• a position (latitude/longitude)
• an altitude
• a heading

Once resolution runs, flight.route_resolution_attempted is set the moment the outcome is final — resolved, or confidently ruled out — so a valid ident with no known route (or a settled ambiguous/rejected leg) is never re-queried against Redis on every subsequent message for the rest of the flight. There's one deliberate exception: a multi-leg route whose recent headings haven't stabilized yet (see "Heading stability" below) is re-evaluated on later messages instead of being settled prematurely — but the airport records fetched from Redis are cached on flight.route_candidate_airports the first time, so every retry re-runs only the local (no-I/O) resolution logic, never a second EVALSHA. The resolution and sanity-check logic itself lives in message_processor/route_resolver.py as a set of pure functions, independent of Redis.

route:{ident} (written by the vrs-standing-data runner) is a raw, dash-delimited string of ICAO airport codes — a simple point-to-point route has 2 codes, but a same-day out-and-back reusing one callsign (e.g. KMIA-KJFK-KMIA) or a multi-stop "milk run" can have 3 or more. EVALSHA route_airports.lua resolves the whole string into an ordered array of full airport:{code} records in one round trip (see shared/lua/route_airports.lua); it returns an empty array if the ident has no known route, or if even one code in the route has no matching airport record.

Because resolution runs as soon as the four conditions above are met — not at the end of the flight — the position/velocity history it reasons over is whatever the flight has accumulated so far, not necessarily its eventual full track. This is deliberate: it's what makes origin/destination available early enough for the same flight's own rules-engine evaluation or MQTT notifications to see them, at the cost of the sanity check below only being as strong as the track recorded up to that point.

• 2 airports: no ambiguity — they're the origin and destination directly.
• 3+ airports: the flight is only actually flying one adjacent pair, so it's resolved low-altitude-first, falling back to a cruise heuristic:
  • Proximity + climb/descent — if the flight's earliest position is below 10,000ft and near exactly one waypoint (or near a waypoint that appears more than once, e.g. the round-trip case, disambiguated by which occurrence's climb/descent direction is structurally possible — climbing away only makes sense for an occurrence with a next hop, descending toward only for one with a previous hop), that's a near-dispositive signal for whether this is the leg's origin (climbing away) or destination (descending toward).
  • Heading vs. bearing — otherwise (cruise altitude, or the proximity check was inconclusive), the flight's most recent observed heading is compared against each candidate leg's great-circle initial bearing. The closest candidate must be within 30° of the observed heading and at least 15° clearer than the second-closest candidate to count as resolved — this is what makes the common out-and-back case tractable (the two legs' bearings are roughly 180° apart), while still refusing to guess among several similarly-plausible legs on a genuine multi-city route. This heuristic only ever runs once heading data has stabilized (see below) — never against a single instantaneous reading.
  • Either heuristic returning nothing resolved (ambiguous proximity with no valid vertical-trend match, or heading inconclusive) leaves both fields None rather than guessing.

Heading stability: a single instantaneous heading reading is not trustworthy on its own — an aircraft circling in a holding pattern (e.g. diverted around weather) sweeps its heading through a full circle, and can momentarily point in almost any direction, including one that happens to align with a completely different leg's bearing. heading_is_stable() requires at least 3 recent heading samples to agree within 20° of each other before the heading-vs-bearing heuristic is trusted at all; ordinary cruise flight naturally produces consistent consecutive headings, while circling does not. When headings aren't yet stable, resolution defers rather than settling — see "one deliberate exception" above.

Sanity check: whatever pair is selected — a direct 2-airport pass-through or a resolved multi-leg pair — is checked against the flight's actual observed track before being trusted. route:{ident} is community-maintained VRS standing data; a callsign can carry a stale or mismatched route with no way to detect that from the string alone. Two geometric checks run against every recorded position, plus a heading check against the latest one:

• Cross-track distance — the perpendicular distance from the position to the great-circle line between the candidate origin and destination; if any position exceeds min(max(150nm, 30% × route_distance), 300nm), the pair is rejected. The 30% term keeps short hops tight; the 300nm ceiling keeps long routes from allowing more than real routing variance (jet stream, ATC, weather).

• Along-track bound — cross-track distance alone only measures perpendicular distance to the infinite line through both airports; it says nothing about whether the position actually falls between the two endpoints. A position must project onto that line within [0, route_distance] (plus a fixed 50nm slack for ordinary terminal-area maneuvering) or the pair is rejected. This specifically catches a holding pattern whose (now-stabilized) heading matches a different, wrong leg's bearing closely enough, and happens to sit within that wrong leg's cross-track tolerance too, while actually being well beyond its destination — real example: an aircraft actually flying KJFK→KMIA, holding at 22,000ft east of Jacksonville on a KJFK-KMIA-KMCO-KJFK routing, whose stabilized northbound holding heading (~350°) is a closer match to the wrong KMIA→KMCO leg's bearing (~341°) than to the correct KJFK→KMIA leg (~202°) and passes that wrong leg's cross-track check (~52nm, under its ~150nm threshold) — but projects ~100nm past KMCO along that leg's line, which the along-track bound catches and rejects.

• Heading vs. destination — when the heading is stable (see below), the most recent heading is compared to the bearing from the latest position to the candidate destination; more than 100° off rejects the pair. Skipped below 10,000ft and within 100nm of either endpoint, where holds, vectoring and terminal maneuvering legitimately point away.

Any check failing rejects the pair — treated the same as an unresolvable case, never a partial guess. All checks run once, at resolution time; later positions don't re-validate an accepted pair.

All-or-nothing: origin/destination are set only when exactly one unambiguous, sanity-checked pair was resolved. Any failure along the way — no route data, an unresolvable leg, or a rejected sanity check — leaves both fields None, never a partial or best-guess value.

Whenever a finalized attempt leaves origin/destination unset — no known route, an unresolvable/ambiguous leg, or a rejected sanity check — a DEBUG log line records the ident, icao_hex, the exact route_airports.lua response that was rejected, and (where applicable) the specific reason (e.g. which sanity check failed, at which position, and by how much). Nothing is logged for the "not final yet" case (heading still stabilizing) — there's nothing to report until an attempt actually settles.

MQTT Topics Published

All topics use the root SkyFollower.

Topic	Payload	Retained
SkyFollower/message-processor/{ID}/status	ONLINE or OFFLINE	Yes
SkyFollower/message-processor/{ID}/statistic/{name}	One retained topic per stat (see fields below)	Yes
SkyFollower/rule/{IDENTIFIER}	JSON flight snapshot (no positions/velocities) with rule key	No

Statistic topic suffixes ({name}):

Field	Format	Description
started_at	UTC ISO-8601 timestamp	Process start time
messages_per_second	Float as string	Rolling 30-second average message rate
processing_time_hwm_ms	Float as string	End-to-end processing time high-water mark since last publish; resets on publish
message_latency_hwm_ms	Float as string	Receipt-through-processed latency high-water mark since last publish, including RabbitMQ queue wait time (a superset of processing_time_hwm_ms); resets on publish. Wall-clock based (time.time() - msg.received_at), unlike processing_time_hwm_ms's monotonic clock, since received_at is stamped on the receiver host and crosses the RabbitMQ hop -- sensitive to NTP drift between hosts
rules_engine_hwm_ns	Float as string	Rules engine duration high-water mark since last publish, in nanoseconds; resets on publish
local_archive_queue_depth	Integer as string	Completed flights queued in completed_flights.db fallback
dead_letter_queue_depth	Integer as string	Completed flights dead-lettered after repeatedly failing to publish (see Dead-Lettering Poison Flights)
active_flights	Integer as string	Flights currently tracked in the active store
rabbitmq_connected	True or False	Whether an active RabbitMQ connection is held
rules_version	String	Last 8 characters of the SHA-256 hash of the rules config this instance has actually loaded ("unknown" until the first successful load). Lags the canonical value published by core-health if a bad payload was pushed and rejected, or if this instance hasn't polled since the last save. Truncated only at publish; the engine keeps the full hash for its own comparison
areas_version	String	Last 8 characters of the SHA-256 hash of the areas config this instance has actually loaded ("unknown" until the first successful load). Same semantics as rules_version

registration_misses_{hour,today,lifetime}, operator_misses_{today,lifetime}, and total_messages_processed_{hour,today,lifetime} are not in the table above — this component is write-only for those three counters (accumulates in memory, flushes to Redis on the telemetry cadence) and does not publish their MQTT statistic topics or Home Assistant discovery entries itself. core-health publishes them on this component's behalf instead — see the "Keys written" Redis table above for the mechanism. aircraft_type_misses_{hour,today} no longer exists at all (it was a dead metric, never incremented anywhere).

Each stat is published as its own retained topic (not a combined JSON payload) every MQTT_PUBLISH_INTERVAL_SECONDS. Home Assistant autodiscovery payloads are published to homeassistant/sensor/SkyFollower_message_processor_{ID}_{field}/config on MQTT connect; each sensor's state_topic points directly at its own SkyFollower/message-processor/{ID}/statistic/{field} topic — no value_template needed.

Map UDP Publisher

An optional, fire-and-forget UDP feed of live aircraft data toward the map service. Fully disabled -- no socket ever created, no send ever attempted -- unless MAP_UDP_HOST is set; same optional-endpoint convention as MQTT_HOST above.

When enabled, unicasts one UTF-8 JSON object per UDP datagram to MAP_UDP_HOST:MAP_UDP_PORT, of three types distinguished by a top-level "type" field. All three carry processor_id (MESSAGE_PROCESSOR_ID -- the same value used for the message_processor:{id}:heartbeat Redis key and MQTT topics), which the map service uses to update a per-processor liveness roster from any of the three, not just heartbeat -- see map/README.md's "Processor Roster" section:

• position -- sent at most once per MAP_UDP_POSITION_MIN_INTERVAL_SECONDS (shared/timing.py) (default 1s) per aircraft -- sub-second position updates aren't perceptible on a map, and this is the single biggest lever on UDP volume / map-Redis write rate. Throttled per icao_hex on the source message's own received_at, not wall-clock send time (see _MapUdpPublisher.should_send_position). A flat object merging whatever Position/Velocity fields that particular message carried (icao_hex, processor_id, ts, lat, lon, alt, velocity, hdg, vs -- field names are deliberately abbreviated on this wire protocol, unlike the full Position/Velocity model field names they're translated from, since this is the single highest-frequency, uncompressed-on-the-UDP-hop payload in the whole system); a field absent from that message is omitted, not sent as null, matching Position.to_dict()/Velocity.to_dict()'s existing convention.
• metadata -- sent the first time a flight's ident/aircraft enrichment/operator/registrant/squawk/origin/destination are known, and again whenever one of those changes (_maybe_publish_map_metadata). Never throttled by MAP_UDP_POSITION_MIN_INTERVAL_SECONDS -- this change-gating is already its own throttle. Reuses the exact same CompletedFlight-shape payload the SkyFollower/rule/{IDENTIFIER} MQTT notification publishes (positions/velocities/_id popped, empty operator/registrant/origin/destination/force_archive omitted) minus the rule key, via a shared _build_flight_notification_payload() helper — see Flight.map_metadata_hash, persisted across messages, for how a change is detected -- plus processor_id and flight_ttl_seconds (the map service adopts its evict window from the first one it sees). Independently of that change-gated path, a dedicated _map_metadata_resend_loop also resends every currently-active flight's metadata datagram unconditionally (skipping any flight whose last_message is older than flight_ttl_seconds by wall clock, so the map never resurrects an aircraft it already evicted) every MAP_METADATA_RESEND_INTERVAL_SECONDS (60s, shared/timing.py), via _resend_all_map_metadata (enumeration pattern mirrors _force_evict_all) -- this does not update map_metadata_hash, so it never masks a genuine later change. Without this, a flight whose metadata never changes again after its first send would get exactly one metadata datagram for its entire duration; since UDP delivery isn't confirmed and the map service's own Redis has no persistence, this periodic resend is what actually lets a map-service restart recover a still-active flight's full metadata within one flight_ttl_seconds window, not just its position.
• heartbeat -- {"type": "heartbeat", "processor_id": "mp-1", "ts": 1725720000.0}. A fixed-interval liveness beacon from a dedicated _map_heartbeat_loop, independent of aircraft traffic entirely -- it has no icao_hex and never touches any flight's state. Runs on a separate MAP_HEARTBEAT_INTERVAL_SECONDS (5s, shared/timing.py) timer from _heartbeat_loop above (the unrelated Redis NX duplicate-ID-guard refresh on HEARTBEAT_INTERVAL_SECONDS) -- don't confuse the two. Traffic-reduction skip: on each 5s tick, _map_heartbeat_loop checks _MapUdpPublisher.last_sent_at (the timestamp of the last datagram this publisher sent, of any type) and skips sending a standalone heartbeat if that was within the last MAP_HEARTBEAT_INTERVAL_SECONDS -- so a processor with steady position/metadata traffic almost never emits one; only an idle/ low-traffic processor sees it fire on (nearly) every tick. Not throttled by MAX_MESSAGE_LAG_SECONDS -- unlike position/metadata, a heartbeat is about the processor being up now, not about a source message's recency.

position/metadata both apply the same staleness suppression SkyFollower/rule/{IDENTIFIER} notifications already do: a message older than MAX_MESSAGE_LAG_SECONDS at emit time is suppressed rather than sent (logged at debug level). heartbeat is exempt -- see above.

position additionally skips a message that's out of order for its aircraft (#1956), independent of MAX_MESSAGE_LAG_SECONDS: that gate only catches a message that's stale in absolute wall-clock terms, but a message can be well within it and still be older than a position already sent for the same icao_hex (cross-receiver skew -- see "Out-of-order messages don't regress last_message" above). Sending it anyway would visibly snap a currently-displayed aircraft backward on the map.

A slow, unreachable, or misconfigured MAP_UDP_HOST/MAP_UDP_PORT can never affect the rest of the pipeline -- _MapUdpPublisher.send() wraps the underlying socket.sendto() call and swallows any exception rather than propagating it. The first send() failure of an outage is logged at WARNING so a misconfigured or unreachable destination is visible at the default log level; repeated failures from the same outage drop to DEBUG to avoid flooding the log, and the next successful send resets that so a later, separate outage warns again.

Startup logs one line naming whether the feed is enabled (with its destination) or disabled. Setting only one of MAP_UDP_HOST/MAP_UDP_PORT is also logged at WARNING at startup, since that combination is always a misconfiguration -- either the feed silently stays disabled (MAP_UDP_PORT set without MAP_UDP_HOST) or it enables with a nonsensical port (MAP_UDP_HOST set without MAP_UDP_PORT, which defaults to 0).

Raw Frame Capture (Forensic)

CAPTURE_RAW_FRAMES (default off, written explicitly to .env by install.sh) is general-purpose forensic infrastructure for investigating a decode anomaly (a bad CPR position, a corrupted velocity/altitude/squawk/ident, a burst of CRC failures, a future pyModeS regression) after the fact, without needing a bespoke diagnostic built for each one. It was motivated by, but is not specific to, the #1835/#1836/#1841 CPR-teleport investigation.

When enabled, every message for a tracked (or newly created) flight -- whether it decodes into usable data or not -- is recorded verbatim as a raw_frames row alongside that flight's positions/velocities, tagged decoded: true/false. A decode failure (too short, a raised exception, CRC invalid) and a message that decodes cleanly but yields nothing this system currently extracts (e.g. an ACAS RA broadcast) are deliberately not distinguished -- both are decoded: false; the raw hex itself is fully replayable offline regardless of which applies. Unlike positions/ velocities, raw_frames has no uniqueness constraint on (icao_hex, timestamp): real, legitimate frames from the same aircraft have been observed 75-345 microseconds apart (dual 978+1090 reporting), and a capture path that could silently drop one to a timestamp collision would defeat its own purpose.

Edge case worth knowing about: with the flag on, a completely garbled first-ever message from a never-before-seen aircraft now creates a flight row too, purely to have somewhere to attach the captured frame -- something that never happens with the flag off, where a message that fails to decode still short-circuits before any Flight is touched.

When a flight completes with the flag on and at least one captured frame, the same completed-flight object -- raw_frames included -- is additionally published (best-effort; not routed through the fallback/SQLite durability _archive() uses) to a second, short-lived RabbitMQ queue, skyfollower-archive-raw-frames, declared only while CAPTURE_RAW_FRAMES is on. That queue carries an 8-hour x-message-ttl and a 100MB x-max-length-bytes cap (whichever limit is hit first evicts the oldest message) and has no dedicated consumer service of its own -- it's meant to be manually inspected or drained (the RabbitMQ management UI, or an ad hoc script) while actively investigating something, not polled continuously. core-health deliberately does not publish any Home Assistant sensors for it, despite it matching the same SKYFOLLOWER_RABBITMQ_RESOURCE_PATTERN every other SkyFollower-owned queue does (required there for RabbitMQ ACL purposes only).

The permanent archive path never carries raw frames, unconditionally._archive() -- the only code path that can ever reach skyfollower-archive / S3 -- always serializes with raw_frames excluded, regardless of CAPTURE_RAW_FRAMES's value. This isn't gated on the setting at all: it doesn't matter whether the flag is on, off, or a future bug in the capture-side gating left raw_frames populated when it shouldn't be, this one exclusion always drops the field before anything reaches permanent storage.

Fault Tolerance

When RabbitMQ is unavailable at startup or during operation, completed flights are written to completed_flights.db (SQLite WAL mode) in data_dir. On the next successful RabbitMQ reconnect, the fallback queue is drained oldest-first before new messages are consumed. Redis and MQTT failures are handled gracefully and logged; enrichment lookups that fail leave the flight partially enriched rather than dropping it.

Draining is also attempted independently every MQTT_PUBLISH_INTERVAL_SECONDS, not just on a detected reconnect. _archive() queues a completed flight to the fallback on any publish exception without necessarily affecting _rmq_connected or the consume side at all — so a run of publish-only failures (e.g. a broker-side rejection on the skyfollower-archive routing key, with the consumer connection itself unaffected) would otherwise never trigger a drain again, since that's only ever spawned from the consumer's own reconnect path. The periodic check is a cheap no-op when the queue is already empty.

Both triggers go through the same drain_in_background(): it spawns the actual drain on a background thread and returns immediately, so a slow drain (e.g. a large backlog) never delays that telemetry cycle's publish. A single-flight guard (a non-blocking lock) ensures only one drain is ever in progress at a time regardless of which trigger started it — if the periodic tick fires while the reconnect-triggered drain is still running, it's a no-op rather than a second overlapping drain, which could otherwise select the same queued row twice and publish it twice.

The fallback queue is the shared FallbackQueue (see shared/README.md) rather than a component-local class.

Dead-Lettering Poison Flights

A completed flight that fails to publish on every drain attempt — not because RabbitMQ is down, but because something about that specific record causes a deterministic failure — would otherwise retry forever, and since drain() always re-selects the oldest row first, it would also block every other queued flight behind it indefinitely. FallbackQueue tracks a per-row retry count: below the threshold (5, hardcoded), a failure behaves exactly as before — stop the drain pass, retry from the top next time. At the threshold, the row is judged permanently poison: it's written out as a standalone JSON file under dead_letters/queue/ in data_dir (capped at 100MB total, oldest file evicted first) for manual inspection, and the drain pass continues to whatever's queued behind it instead of stopping. There's no automated replay path — a dead-lettered file is purely something an operator inspects or discards out-of-band (data_dir is already a host-mounted volume, same as completed_flights.db itself).

A raw attempt count alone isn't safe: _drain_fallback() is called on every successful RabbitMQ reconnect (not just on the MQTT_PUBLISH_INTERVAL_SECONDS tick), so a flapping connection reconnecting every few seconds could otherwise burn through the retry threshold within seconds — dead-lettering a flight that was never actually poison, just unlucky enough to be at the head of the queue during a brief instability. FallbackQueue also enforces a minimum time between attempts on the same row (30 seconds, hardcoded, independent of how often _drain_fallback() itself gets called), so reaching the threshold always takes a real, bounded amount of elapsed time — not just a burst of rapid reconnect attempts.

A missing archive queue is not poison. _archive() and _drain_fallback() publish completed flights with mandatory=True against the well-known skyfollower-archive queue, which only exists once an archive-processor has connected and declared it. A deployment that deliberately runs no archive processor never has that queue, so every publish returns pika.exceptions.UnroutableError — a healthy-connection condition, not a per-flight fault and not a recoverable outage. The fallback queue is constructed with that exception type classified as non-poison: those rows retry forever and are never dead-lettered, so turning on an archive-processor later drains the entire accumulated backlog automatically. Disk growth for a deployment running no archive processor is instead bounded by a ring-buffer cap on the retryable table itself (100MB, the same ceiling the dead-letter directory uses); once such a deployment accumulates past it, the oldest completed flights are evicted first, logged as a capacity eviction. Any other publish failure (a real connection loss, or an archive-processor rejecting a specific payload for content reasons) still dead-letters at the threshold exactly as above.

Active flight store durability & crash recovery

active_flights.db (SQLite, WAL mode, data_dir) holds every currently tracked flight and is committed to disk after every message. A process end — whether a deliberate stop (SIGTERM/SIGINT) or an ungraceful one (OOM-kill, docker kill, host crash) — is recovered identically on the next startup; there is no special "flush everything" shutdown path for either of those, since nothing needs force-archiving when the store already survives on its own. That assumes the store itself survives, though — see "Decommissioning a Message Processor" below for the one case where it deliberately won't (permanently destroying the container and its volume together).

On startup, the message processor reopens active_flights.db and recovers whatever flights were still tracked. Recovery is driven by message timestamps, not by how long the container was down: message_clock (_message_clock in main.py) is floored at the most recent last_message among recovered flights (not wall-clock "now"), and only advances as messages are actually consumed — _update_flight() bumps it to max(message_clock, msg.received_at) on every message. _evict_stale() (which drives both eviction from the active store and, as a consequence, archiving) compares each flight's last_message against message_clock - flight_ttl_seconds, never against wall-clock time.

This gating matters in two situations, both of which produce the same symptom — active flights sitting unevicted, local_archive_queue_depth flat — for a reason that is correct, not a stuck or broken pipeline:

• Recovering after a restart. A recovered flight is not archived just because real time passed while the container was stopped — if a continuation message for that aircraft is sitting in the RabbitMQ backlog, it resumes the same flight once the message processor reconnects and drains the backlog. A genuine gap longer than flight_ttl_seconds — whether it happens live or is discovered while replaying a backlog — still correctly splits the flight into two records.
• Offline replay with nothing live behind it (see tools/traffic-replayer). If the only traffic a message processor ever sees is a finite replayed capture and no live receiver is attached, message_clock advances only as far as the capture's last message and then simply stops — there is nothing left to bump it further. Flights still active when the replay ends stay in the active store indefinitely, neither refreshing nor expiring, until either more messages arrive (a further replay, or live traffic) or the process is stopped. This can look identical to a stuck archive path from the outside; it isn't one.

MQTT rule notifications for messages older than max_message_lag_seconds are suppressed during backlog replay (logged at debug) to avoid flooding MQTT the instant a message processor reconnects after downtime; the rule still fires and is still recorded in matched_rules/the eventual archived flight.

Out-of-order messages don't regress last_message. A message can still arrive out of order for a given aircraft even with the receiver's own publish-ordering guarantee in place (#1956) — cross-receiver skew (a 1090 and a 978 receiver, or redundant antennas, each publishing independently) is expected, if rare. _update_flight() only ever advances flight.last_message forward; a message older than what's already stored still counts toward total_messages and is still persisted (positions/velocities sort correctly on load regardless of processing order), but it cannot roll last_message backward. Without that guard, a single out-of-order message could make the next, perfectly ordinary message compute a gap that appears to exceed flight_ttl_seconds against the wrongly-rolled-back value, force-archiving and splitting a flight that never actually ended. The same message is also skipped for the map UDP position send (see below) — even though it may be well under max_message_lag_seconds, publishing it would visibly snap a currently-displayed aircraft backward on the map.

Because active_flights.db only depends on MESSAGE_PROCESSOR_ID (which determines the RabbitMQ queue consumed, skyfollower-message-processor-{MESSAGE_PROCESSOR_ID}) and not on container identity, moving the file to a replacement container with the same MESSAGE_PROCESSOR_ID resumes tracking the same way a restart does. This is exactly scripts/install.sh's "replacing an existing message processor?" flow -- see MESSAGE_PROCESSOR_ID above. One caveat: the Redis heartbeat key (skyfollower-message-processor-{MESSAGE_PROCESSOR_ID}, SET NX with a HEARTBEAT_TTL_SECONDS TTL) must expire — or be deleted manually — before a replacement container can claim the same ID.

Decommissioning a Message Processor

Everything above assumes active_flights.db survives the process ending — true for a restart, a crash, or moving the file to a replacement container. It is not true if the intent is to permanently remove a message processor: destroy its container and its volume together (e.g. shrinking the fleet — see the "treat processors as a stack" operating rule elsewhere in this repo). Anything still active in active_flights.db at that moment is gone for good unless it's forced through eviction/archival first.

SIGUSR1 triggers exactly that: a decommission sequence that force-evicts every active flight regardless of message_clock/TTL, waits for the local archive fallback queue to drain, and only then runs the same graceful shutdown SIGTERM/SIGINT already use. SIGTERM/SIGINT themselves are untouched by this — they keep meaning "restart, resume from disk," exactly as described above.

Procedure:

1. docker kill -s SIGUSR1 <container>
2. Wait for the container to exit on its own. No further action is needed — the process forces every active flight through the normal archive path, waits for the local fallback queue to drain, and shuts itself down.
3. Check the container's logs for a dead-letter warning (see below) before continuing.
4. Only once the container has exited: docker compose down -v, or otherwise delete the data directory.

The wait in step 2 is indefinite, by design. The process polls the local archive fallback queue (local_archive_queue_depth) until it reaches zero and does not give up on its own — if RabbitMQ is down, the correct assumption is that it eventually comes back, and a decommission racing an unrelated broker outage should not exit while flights are sitting only in local fallback storage, not yet durable anywhere else. If an operator needs to abandon a stuck wait (e.g. RabbitMQ is gone for good and the data loss is accepted), the deliberate escape hatch is a manual docker kill -9 (SIGKILL) on the container — a conscious operator decision, not a timeout this tool applies automatically.

Dead-lettered flights are never waited on. The dead-letter queue (dead_letter_queue_depth) exists specifically for items expected to never successfully drain via retry — waiting on it too would let a single poison message hang a decommission forever. If it's nonzero once the retryable queue has drained, the process logs the count at high severity and continues shutting down regardless. Those flights' contents need manual attention (see "Dead-Lettering Poison Flights" above) before the volume is destroyed, if they matter.

Rules Engine

Rules and areas are loaded from Redis (config:rules / config:areas) and hot-reloaded every 30 seconds. The config:*:version SHA-256 keys are a fast-path "nothing changed" signal, not the sole gate: if a version key is missing (a deployment from before it existed, a partially-restored volume, a manual seed), was written out of step with its body, or the engine is holding no rules while config:rules has content, the reload falls through to hashing the body itself. So a processor started against a Redis that has config:rules but no config:rules:version still loads those rules on its next poll — no UI save required. The rules engine is implemented in message-processor/rules_engine.py.

Each rule must have a unique identifier, an enabled boolean, and a non-empty conditions array. All conditions within a rule are ANDed together. A rule fires at most once per flight per identifier (the identifier is added to flight.matched_rules on first match, and subsequent evaluations skip it).

Conditions are sorted by evaluation cost before each evaluation pass — cheap field comparisons run before expensive geographic checks. See Rules & Areas for the full condition type reference.

A rule may also carry an optional force_archive boolean (default false). It isn't a condition — it's an ordinary property alongside identifier, so any number of independent rules can each set it. When a matched rule has force_archive: true, flight.force_archive becomes (and stays) true for that flight's lifetime; the archive processor reads this to override its external-only archive skip (see archive-processor/README.md) for external-only flights the user does care about — a specific aircraft, an area, etc. — without archiving every external contact indiscriminately. A matched force_archive rule still publishes its normal MQTT rule notification like any other rule; nothing about the flag suppresses that.