Cut poller/backend CPU on small hosts: lazy entity builds, DR dedup, Whisper cap

Fixes the CPU regression observed after PR #117 (both cores pegged,
load ~5.0 on a 2-core VPS). Benchmarked the hot paths with synthetic
BEAST traffic and addressed the measured costs:

Poller:
- Frame worker no longer builds the full entity dict (trail copy,
  comm-B snapshot, DR projection, ISO timestamp) on every decoded
  frame — only for the <=1/s-per-aircraft publishes. New decoder
  methods ingest_frame()/entity_from_state(); ingest() kept as a
  compat wrapper. ~22% cheaper per frame.
- Registry tick loop rebuilds decoder entities once per 5s snapshot
  tick (right before its only consumer) instead of every second,
  cutting that O(aircraft) rebuild by 80%.
- Dead-reckoned aircraft no longer republish every second: their
  lat/lon/altitude are wall-clock projections that always tripped the
  _entity_changed dedup, flooding Redis pub/sub, the backend WS
  fan-out, DB writes, and frontend renders — the main per-message
  regression from PR #117. The frontend already projects DR tracks
  client-side (pvb.ts), so projection-only changes are now skipped;
  real telemetry and DR on/off transitions still publish immediately.
- Dropped the redundant second sanitize_payload deep-copy per publish
  (write_entity_observation re-sanitized what publish_entity already
  had, walking all 150 trail points each time).
- BEAST transport parses escape-free frames (the common case) via
  bytearray.find slicing instead of the per-byte Python loop:
  3.7x faster (4.0us -> 1.1us per frame).

Transcription:
- New WHISPER_CPU_THREADS setting (default 1) passed to WhisperModel,
  and compose CPU limit 2.0 -> 1.0, so ctranslate2 can no longer
  saturate every core whenever a P25 call transcribes.

Tests: new test_beast_transport.py (11 tests) covering the fast path,
escaped frames, garbage resync, and incomplete buffers; 5 new DR-dedup
tests in test_bus.py. Full poller suite: 171 passed. tsc clean,
docker compose config clean.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017n1uFNfMESakg9E9qoMDrh

Author: claude

.env.example

@@ -214,6 +214,10 @@ WHISPER_MODEL=base
 WHISPER_LANGUAGE=en
 WHISPER_COMPUTE_TYPE=int8
 WHISPER_DEVICE=cpu
+# CPU threads for Whisper inference. Keep at 1 on small hosts (2-core VPS /
+# Pi) so transcription bursts can't starve the poller and API; raise if you
+# have spare cores.
+WHISPER_CPU_THREADS=1
 
 # FlashAlert & TVFR feed fallbacks (RSS/REST URL overrides)
 FLASHALERT_ENABLED=false

TASK_LOG.md

@@ -5,6 +5,19 @@ Format: `## YYYY-MM-DD — <summary>` with bullet points for details.
 
 ---
 
+## 2026-07-06 — CPU usage optimization for small hosts (2-core VPS)
+
+- **Motivation**: User reported both cores pegged (load ~5.0 on 2 cores) after the ADS-B pipeline overhaul (PR #117). Benchmarked the hot paths with synthetic BEAST traffic and fixed the measured costs.
+- **Lazy entity building in the BEAST frame worker** ([beast_decoder.py](poller/normalizers/beast_decoder.py), [adsb.py](poller/pollers/adsb.py)):
+  - Split `ingest()` into `ingest_frame()` (state-only decode, runs per frame) and `entity_from_state()` (full entity dict build). The frame worker now builds the entity dict — trail copy, comm-B snapshot, DR projection, ISO timestamp — only when the ≤1/s-per-aircraft publish gate passes instead of on every decoded frame (~22% cheaper per frame; `ingest()` kept as a compat wrapper).
+  - Best Mode arbitration semantics preserved: "beast seen" is still only recorded for positioned aircraft.
+- **Registry tick loop rebuild moved to snapshot cadence** ([adsb.py](poller/pollers/adsb.py)): `snapshot_entities()` (an O(aircraft) full entity rebuild) ran every second while its only consumer — the enriched snapshot publish — runs every 5s. It now runs once per snapshot tick, immediately before the publish, cutting that work by 80%. DR still advances through a total feed outage since the rebuild happens on the same tick that publishes.
+- **Dead-reckoning republish suppression** ([bus.py](poller/bus.py)): while an aircraft is dead-reckoned, its lat/lon/altitude are synthetic wall-clock projections that made `_entity_changed` report a change every second — so previously-quiet stale aircraft flooded Redis pub/sub, the backend WS fan-out, DB writes, and frontend renders (the main per-message CPU regression from PR #117). Since the frontend already projects DR tracks client-side (pvb.ts), `_entity_changed` now skips lat/lon/altitude when both states are DR. Real telemetry changes and DR on/off transitions (`position_dr` added to compare keys) still publish immediately.
+- **Halved per-publish sanitize cost** ([bus.py](poller/bus.py), [db.py](poller/db.py)): `write_entity_observation` re-ran `sanitize_payload` (a recursive deep-copy that walks all 150 trail points, ~216µs per call) on entities its only caller `publish_entity` had already sanitized. New `sanitized=True` flag skips the redundant pass.
+- **BEAST transport fast path** ([beast_transport.py](poller/pollers/beast_transport.py)): frames whose wire body contains no 0x1A escape byte (the overwhelmingly common case) are now sliced out at C speed with `bytearray.find` instead of the per-byte Python loop — 3.7× faster frame parsing (4.0µs → 1.1µs/frame). Escaped/incomplete/garbage frames still go through `parse_frame`.
+- **Whisper transcription CPU cap** ([docker-compose.yml](docker-compose.yml), [transcription/config.py](transcription/config.py), [transcription/main.py](transcription/main.py), [.env.example](.env.example)): ctranslate2 grabbed every core whenever a P25 call transcribed, starving the poller/backend on a 2-core host. New `WHISPER_CPU_THREADS` setting (default 1) passed to `WhisperModel`, and the container's compose CPU limit reduced from 2.0 to 1.0. Base/int8 on one thread still transcribes short P25 clips faster than realtime.
+- **Tests**: new [test_beast_transport.py](poller/tests/test_beast_transport.py) (11 tests: fast path, escaped MLAT/signal/message bytes, garbage resync, incomplete buffers, 0x31 skipping) and 5 DR-dedup tests in [test_bus.py](poller/tests/test_bus.py). Full poller suite: 171 passed.
+
 ## 2026-07-06 — ADS-B pipeline overhaul: dead reckoning, signal-gap handling, stale-track rendering
 
 - **Server-side dead reckoning** ([beast_decoder.py](poller/normalizers/beast_decoder.py), [config.py](poller/config.py)):

docker-compose.yml

@@ -192,7 +192,9 @@ services:
       resources:
         limits:
           memory: 2g
-          cpus: '2.0'
+          # Transcription is a background task — never let it saturate the
+          # host (map/API responsiveness comes first on 2-core deployments).
+          cpus: '1.0'
         reservations:
           memory: 512m
           cpus: '0.5'

poller/bus.py

@@ -92,7 +92,10 @@ async def publish_entity(
         return
 
     try:
-        await write_entity_observation(entity, record_observation=record_observation)
+        # sanitized=True: this entity already went through sanitize_payload()
+        # above — skipping the second recursive deep-copy (which walks every
+        # trail point) roughly halves the per-publish CPU cost.
+        await write_entity_observation(entity, record_observation=record_observation, sanitized=True)
     except Exception as exc:
         import traceback
         logger.warning("DB write failed for %s: %s\n%s", entity.get("entity_id"), exc, traceback.format_exc())
@@ -126,25 +129,40 @@ async def close():
         _redis = None
 
 
+_COMPARE_KEYS = (
+    "entity_type",
+    "source",
+    "display_name",
+    "lat",
+    "lon",
+    "altitude",
+    "heading",
+    "speed",
+    "vertical_rate",
+    "status",
+    "identity",
+    "tags",
+    "position_stale",
+    "position_dr",
+    "trail_pts",
+    "comm_b",
+)
+
+# While an aircraft is being dead-reckoned, its lat/lon/altitude are synthetic
+# projections that advance with wall-clock time — every rebuild "changes" them
+# even though no new data arrived. The frontend projects DR tracks client-side
+# (pvb.ts), so republishing each projection is pure overhead: skip those keys
+# when both the previous and current state are dead-reckoned. Real telemetry
+# (heading, speed, vertical_rate, squawk, ...) and the DR on/off transitions
+# themselves still publish immediately.
+_DR_SKIP_KEYS = frozenset(("lat", "lon", "altitude"))
+
+
 def _entity_changed(previous: dict, current: dict) -> bool:
-    compare_keys = (
-        "entity_type",
-        "source",
-        "display_name",
-        "lat",
-        "lon",
-        "altitude",
-        "heading",
-        "speed",
-        "vertical_rate",
-        "status",
-        "identity",
-        "tags",
-        "position_stale",
-        "trail_pts",
-        "comm_b",
-    )
-    for key in compare_keys:
+    both_dr = bool(previous.get("position_dr")) and bool(current.get("position_dr"))
+    for key in _COMPARE_KEYS:
+        if both_dr and key in _DR_SKIP_KEYS:
+            continue
         if previous.get(key) != current.get(key):
             return True
     return False

poller/db.py

@@ -122,11 +122,17 @@ async def close_db():
         _pool = None
 
 
-async def write_entity_observation(entity: dict, record_observation: bool = True):
-    """Upsert entity row and append an observation. Runs geofence check if positioned."""
+async def write_entity_observation(entity: dict, record_observation: bool = True, sanitized: bool = False):
+    """Upsert entity row and append an observation. Runs geofence check if positioned.
+
+    Pass sanitized=True when the payload already went through sanitize_payload()
+    (publish_entity does) to skip a redundant recursive deep-copy of the entity —
+    trail-carrying aircraft entities make that copy expensive.
+    """
     if _pool is None:
         return
-    entity = sanitize_payload(entity)
+    if not sanitized:
+        entity = sanitize_payload(entity)
 
     from geofence import check_geofences  # lazy — breaks bus→db→geofence→bus cycle
 

poller/normalizers/beast_decoder.py

@@ -95,6 +95,28 @@ def __init__(self):
         self._last_prune_ts: float = 0.0
 
     def ingest(self, message_bytes: bytes, *, mlat_ticks: int | None = None, signal: int | None = None) -> dict | None:
+        """Decode one frame and return the full entity dict (or None).
+
+        Convenience wrapper around ingest_frame() + entity_from_state(). The
+        hot path (adsb frame worker) uses those two directly so the entity
+        dict is only built for the ~1/s-per-aircraft publishes instead of on
+        every decoded frame.
+        """
+        ac = self.ingest_frame(message_bytes, mlat_ticks=mlat_ticks, signal=signal)
+        if ac is None:
+            return None
+        return self._to_entity(ac)
+
+    def entity_from_state(self, ac: _AircraftState, now: float | None = None) -> dict | None:
+        """Build the publishable entity dict for a decoded aircraft state."""
+        return self._to_entity(ac, now=now)
+
+    def ingest_frame(self, message_bytes: bytes, *, mlat_ticks: int | None = None, signal: int | None = None) -> _AircraftState | None:
+        """Decode one Mode S frame into aircraft state, without building an entity.
+
+        Returns the updated _AircraftState (which may not have a position yet),
+        or None if the frame was rejected/undecodable.
+        """
         if pms is None:
             if not self._warned_missing_dep:
                 logger.warning("[adsb] pyModeS not available; BEAST decode disabled")
@@ -189,7 +211,7 @@ def ingest(self, message_bytes: bytes, *, mlat_ticks: int | None = None, signal:
             self._prune_stale()
             self._last_prune_ts = now
 
-        return self._to_entity(ac)
+        return ac
 
     def seed_reference(self, icao: str, lat: float, lon: float, ts: float | None = None) -> bool:
         """Seed a position reference from another source (OpenSky / ultrafeeder).

poller/pollers/adsb.py

@@ -224,20 +224,29 @@ async def _process_beast_frames(self):
         while True:
             msg, mlat_ticks, signal = await self._beast_queue.get()
             try:
-                entity = self._beast_decoder.ingest(msg, mlat_ticks=mlat_ticks, signal=signal)
-                if entity:
-                    icao = (entity.get("identity") or {}).get("icao24", "").lower()
+                # State-only decode: the full entity dict (trail copy, comm-B
+                # snapshot, DR projection, ISO timestamp) is built lazily below,
+                # only for the ≤1/s-per-aircraft publishes — not per frame.
+                state = self._beast_decoder.ingest_frame(msg, mlat_ticks=mlat_ticks, signal=signal)
+                # Positioned aircraft only, matching the previous behavior where
+                # a position-less decode produced no entity: recording "beast"
+                # for unpositioned aircraft would wrongly suppress the
+                # ultrafeeder/OpenSky sources in Best Mode arbitration.
+                if state is not None and state.lat is not None and state.lon is not None:
+                    icao = state.icao
                     self._record_source_seen(icao, "beast")
-                    self._unified_entities[icao] = entity
                     now = time.time()
                     if now - _last_published.get(icao, 0.0) >= _BEAST_PUBLISH_MIN_INTERVAL:
-                        _last_published[icao] = now
-                        # Dead-reckoned positions are estimates — keep them out
-                        # of the observation history (trails stay real fixes only).
-                        await publish_entity(
-                            entity,
-                            record_observation=not entity.get("position_dr"),
-                        )
+                        entity = self._beast_decoder.entity_from_state(state, now=now)
+                        if entity:
+                            _last_published[icao] = now
+                            self._unified_entities[icao] = entity
+                            # Dead-reckoned positions are estimates — keep them out
+                            # of the observation history (trails stay real fixes only).
+                            await publish_entity(
+                                entity,
+                                record_observation=not entity.get("position_dr"),
+                            )
             except Exception as exc:
                 logger.warning("[adsb] frame processing error: %s", exc)
 
@@ -249,7 +258,6 @@ async def _process_beast_frames(self):
 
     async def _registry_tick_loop(self):
         _SNAPSHOT_INTERVAL = 5  # publish full snapshot every N ticks (seconds)
-        _last_frames_seen = self._transport.frames_seen
 
         while True:
             await asyncio.sleep(1.0)
@@ -269,19 +277,6 @@ async def _registry_tick_loop(self):
                         del self._last_seen_by_source[icao]
 
             try:
-                # Pull fresh BEAST positions into the unified registry when new
-                # frames have arrived since the last tick — skips the O(aircraft)
-                # entity reconstruction when the decoder state is unchanged.
-                # Also refresh on every snapshot tick regardless: dead-reckoned
-                # display positions advance with wall-clock time, so a total
-                # feed gap must not freeze the published snapshot.
-                current_frames = self._transport.frames_seen
-                if current_frames != _last_frames_seen or self._tick_count % _SNAPSHOT_INTERVAL == 0:
-                    _last_frames_seen = current_frames
-                    for ac in self._beast_decoder.snapshot_entities():
-                        icao = (ac.get("identity") or {}).get("icao24", "").lower()
-                        self._unified_entities[icao] = ac
-
                 # Evict entries silent for more than 2 minutes (runs every tick, cheap)
                 stale_cutoff = 120.0
                 to_remove = [
@@ -294,6 +289,16 @@ async def _registry_tick_loop(self):
                 # Publish full enriched snapshot at reduced cadence — individual
                 # entity updates still arrive in real time via publish_entity().
                 if self._tick_count % _SNAPSHOT_INTERVAL == 0:
+                    # Rebuild decoder entities only here, immediately before the
+                    # snapshot that consumes them — nothing reads the registry
+                    # between snapshots (real-time flow goes via publish_entity),
+                    # so the previous every-tick O(aircraft) rebuild was 80%
+                    # wasted work. Rebuilding on the snapshot tick also keeps
+                    # dead reckoning advancing through a total feed outage.
+                    for ac in self._beast_decoder.snapshot_entities():
+                        ac_icao = (ac.get("identity") or {}).get("icao24", "").lower()
+                        self._unified_entities[ac_icao] = ac
+
                     snapshot_ents = [
                         entity for icao, entity in self._unified_entities.items()
                         if self._should_publish_from_source(icao, entity.get("source", "unknown"))

poller/pollers/beast_transport.py

@@ -21,6 +21,10 @@
 
 logger = logging.getLogger(__name__)
 
+# BEAST frame type -> Mode S payload length in bytes.
+# 0x31 = 2-byte short squitter (skipped), 0x32 = 7-byte short, 0x33 = 14-byte long.
+_PAYLOAD_LEN = {0x31: 2, 0x32: 7, 0x33: 14}
+
 
 class BeastTransport:
     """Manages a BEAST TCP connection and delivers raw Mode S frames.
@@ -117,9 +121,30 @@ def consume_buffer(
         buffer after this call.
         """
         pos = 0
+        blen = len(buffer)
         messages: list[tuple[bytes, int, int]] = []
 
-        while pos < len(buffer):
+        while pos < blen:
+            # Fast path: a frame whose wire body contains no 0x1A byte needs no
+            # unescaping, so it can be sliced out directly at C speed. This is
+            # the overwhelmingly common case (0x1A appears in ~0.4% of body
+            # bytes), and it avoids the per-byte Python loop in parse_frame().
+            if buffer[pos] == 0x1A and pos + 1 < blen:
+                payload_len = _PAYLOAD_LEN.get(buffer[pos + 1])
+                if payload_len is not None:
+                    body_start = pos + 2
+                    body_end = body_start + 7 + payload_len
+                    if body_end <= blen and buffer.find(0x1A, body_start, body_end) == -1:
+                        if buffer[pos + 1] != 0x31:  # 0x31 short squitter: skip
+                            messages.append((
+                                bytes(buffer[body_start + 7:body_end]),
+                                int.from_bytes(buffer[body_start:body_start + 6], "big"),
+                                buffer[body_start + 6],
+                            ))
+                        pos = body_end
+                        continue
+
+            # Slow path: escapes present, buffer incomplete, or garbage bytes.
             consumed, message = self.parse_frame(memoryview(buffer)[pos:])
             if consumed == 0:
                 break          # incomplete frame; wait for more data
@@ -161,7 +186,7 @@ def parse_frame(
             return -(next_sync + 1), None
 
         frame_type = view[1]
-        payload_len = {0x31: 2, 0x32: 7, 0x33: 14}.get(frame_type)
+        payload_len = _PAYLOAD_LEN.get(frame_type)
         if payload_len is None:
             return -1, None