upd({ meshcore_auto_add_contacts: v })}
helper="Enables firmware CMD 58 (set_autoadd_config) at connect so AIDA automatically adds every node it hears an advert from as a contact, enabling DM send/decrypt without manual contact exchange"
/>
@@ -418,13 +466,13 @@ export default function MeshCoreConnection() {
upd({ meshcore_auto_reconnect: v })}
helper="Automatically reconnect to the MeshCore companion if the link drops"
/>
upd({ meshcore_max_reconnect_attempts: v })}
min={0}
helper="Maximum reconnect attempts before giving up (0 = unlimited)"
diff --git a/work/docker-compose.yml b/work/docker-compose.yml
index ffcead4..417604b 100644
--- a/work/docker-compose.yml
+++ b/work/docker-compose.yml
@@ -68,11 +68,11 @@ services:
memory: 64M
healthcheck:
- test: ["CMD-SHELL", "test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ \"$(cat /tmp/meshai.link 2>/dev/null)\" = up ] || exit 1"]
+ test: ["CMD-SHELL", "curl -f -s -o /dev/null http://localhost:8080/ || exit 1"]
interval: 30s
timeout: 10s
retries: 3
- start_period: 15s
+ start_period: 240s
logging:
driver: "json-file"
diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py
index 06109a1..15f295b 100644
--- a/work/meshai/env/store.py
+++ b/work/meshai/env/store.py
@@ -1,5 +1,6 @@
"""Environmental data store with tick-based adapter polling."""
+import asyncio
import hashlib
import json
import logging
@@ -438,20 +439,38 @@ class EnvironmentalStore:
from meshai import coverage as _cov
return _cov.resolve_adapter_coverage(adapter, self._coverage_bbox, "native")
- def refresh(self) -> bool:
+ async def refresh(self) -> bool:
"""Called every second from main loop. Ticks each adapter.
+ Adapter tick() calls (blocking network I/O) run concurrently in
+ worker threads and are AWAITED to completion before ingest, so the
+ event loop stays responsive while fetches are in flight. Ingest
+ (DB/EventBus work) then runs on the loop thread once all ticks are
+ done, exactly as before, so no thread ever overlaps ingest.
+
Returns:
True if any data changed
"""
changed = False
- for name, adapter in self._adapters.items():
- try:
- if adapter.tick():
- changed = True
+ adapters = list(self._adapters.items())
+ if not adapters:
+ self._purge_expired()
+ return changed
+
+ results = await asyncio.gather(
+ *(asyncio.to_thread(adapter.tick) for _, adapter in adapters),
+ return_exceptions=True,
+ )
+ for (name, adapter), result in zip(adapters, results):
+ if isinstance(result, Exception):
+ logger.warning("Env adapter %s error: %s", name, result)
+ continue
+ if result:
+ changed = True
+ try:
self._ingest(name, adapter)
- except Exception as e:
- logger.warning("Env adapter %s error: %s", name, e)
+ except Exception as e:
+ logger.warning("Env adapter %s error: %s", name, e)
self._purge_expired()
return changed
diff --git a/work/meshai/main.py b/work/meshai/main.py
index 82c6a9e..ac24b45 100644
--- a/work/meshai/main.py
+++ b/work/meshai/main.py
@@ -2,6 +2,7 @@
import argparse
import asyncio
+import concurrent.futures
import logging
import os
import signal
@@ -153,13 +154,46 @@ class MeshAI:
while self._running:
await asyncio.sleep(1)
- # Periodic MeshMonitor refresh
- if self.meshmonitor_sync:
- self.meshmonitor_sync.maybe_refresh()
-
- # Periodic data store refresh and health computation
+ # Run the mesh/env/meshmonitor pollers concurrently so blocking
+ # network I/O (tick() fetches) never starves this loop — and
+ # therefore never starves the dashboard, which shares this same
+ # asyncio loop. Each refresh() internally awaits its own due
+ # ticks in worker threads; meshmonitor_sync.maybe_refresh is
+ # synchronous, so it is offloaded to a thread here directly.
+ # We await the WHOLE cycle before the next iteration, so no
+ # tick() thread ever overlaps the next cycle's bookkeeping.
+ _refresh_tasks = {}
+ if self.data_store:
+ _refresh_tasks['data'] = self.data_store.refresh()
+ if self.env_store:
+ _refresh_tasks['env'] = self.env_store.refresh()
+ if self.meshmonitor_sync:
+ _refresh_tasks['mm'] = asyncio.to_thread(self.meshmonitor_sync.maybe_refresh)
+
+ if _refresh_tasks:
+ _refresh_results = dict(zip(
+ _refresh_tasks.keys(),
+ await asyncio.gather(*_refresh_tasks.values(), return_exceptions=True),
+ ))
+ else:
+ _refresh_results = {}
+
+ refreshed = _refresh_results.get('data')
+ if isinstance(refreshed, Exception):
+ logger.warning("Data store refresh error: %s", refreshed)
+ refreshed = False
+
+ env_changed = _refresh_results.get('env')
+ if isinstance(env_changed, Exception):
+ logger.debug("Env refresh error: %s", env_changed)
+ env_changed = False
+
+ _mm_result = _refresh_results.get('mm')
+ if isinstance(_mm_result, Exception):
+ logger.warning("MeshMonitor sync refresh error: %s", _mm_result)
+
+ # Periodic data store health computation
if self.data_store:
- refreshed = self.data_store.refresh()
# Recompute health after refresh
if refreshed and self.health_engine:
self.health_engine.compute(self.data_store)
@@ -210,10 +244,9 @@ class MeshAI:
except Exception:
pass
- # Environmental feed refresh
+ # Environmental feed alerting/broadcast (refresh already ran above)
if self.env_store:
try:
- env_changed = self.env_store.refresh()
if env_changed and self.alert_engine:
env_alerts = self.alert_engine.check_environmental(self.env_store)
if env_alerts:
@@ -1006,6 +1039,14 @@ def main() -> None:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
+ # Size the default executor generously: mesh sources + env adapters
+ # (~7 sources, ~15 env adapters) now fetch concurrently via
+ # asyncio.to_thread() every tick, so they need thread headroom to avoid
+ # queuing behind each other on the default executor's small pool.
+ loop.set_default_executor(
+ concurrent.futures.ThreadPoolExecutor(max_workers=24, thread_name_prefix="meshai-io")
+ )
+
def signal_handler(sig, frame):
logger.info(f"Received signal {sig}")
loop.create_task(bot.stop())
diff --git a/work/meshai/mesh_data_store.py b/work/meshai/mesh_data_store.py
index 44047db..97c1592 100644
--- a/work/meshai/mesh_data_store.py
+++ b/work/meshai/mesh_data_store.py
@@ -6,6 +6,7 @@ This module replaces mesh_sources.py with a clean three-layer architecture:
- Layer 3: Consumers read unified model (no field guessing)
"""
+import asyncio
import json
import logging
import sqlite3
@@ -441,12 +442,16 @@ class MeshDataStore:
if stale_nums:
logger.info(f"Purged {len(stale_nums)} stale nodes (not heard in {STALE_NODE_THRESHOLD_DAYS} days)")
- def refresh(self) -> bool:
+ async def refresh(self) -> bool:
"""Tick-based refresh. Called every second from the main loop.
- Delegates to source tick() for sources that support it.
- Only does a full rebuild when nodes/edges/topology change.
- Only does a lightweight update when only packets change.
+ Delegates to source tick() for sources that support it. Due sources'
+ tick() calls (blocking network I/O) run concurrently in worker
+ threads and are AWAITED to completion before any bookkeeping, so the
+ event loop (and therefore the dashboard) stays responsive while
+ fetches are in flight. Only does a full rebuild when nodes/edges/
+ topology change. Only does a lightweight update when only packets
+ change.
Returns:
True if any data changed
@@ -456,26 +461,38 @@ class MeshDataStore:
needs_rebuild = False
needs_packet_update = False
+ due: list[tuple[str, object]] = []
for name, source in self._sources.items():
# Check if this source supports tick-based polling
if hasattr(source, 'tick') and hasattr(source, '_tick_interval'):
if now - source._last_tick >= source._tick_interval:
- endpoint = source.tick()
- if endpoint:
- any_changed = True
- # Major changes require full rebuild
- if endpoint in ("nodes", "edges", "traceroutes", "topology", "telemetry"):
- needs_rebuild = True
- # Packet-only changes are lightweight
- elif endpoint in ("packets",):
- needs_packet_update = True
- # stats, counts, channels, solar, network just update cached data
+ due.append((name, source))
else:
# Legacy fallback for sources without tick support
if source.maybe_refresh():
any_changed = True
needs_rebuild = True
+ if due:
+ results = await asyncio.gather(
+ *(asyncio.to_thread(source.tick) for _, source in due),
+ return_exceptions=True,
+ )
+ for (name, source), result in zip(due, results):
+ if isinstance(result, Exception):
+ logger.warning(f"Source {name} tick failed: {result}")
+ continue
+ endpoint = result
+ if endpoint:
+ any_changed = True
+ # Major changes require full rebuild
+ if endpoint in ("nodes", "edges", "traceroutes", "topology", "telemetry"):
+ needs_rebuild = True
+ # Packet-only changes are lightweight
+ elif endpoint in ("packets",):
+ needs_packet_update = True
+ # stats, counts, channels, solar, network just update cached data
+
if needs_rebuild:
self._rebuild()
self._purge_stale_nodes()
diff --git a/work/meshai/transport/meshcore_transport.py b/work/meshai/transport/meshcore_transport.py
index 9be229d..6d15bba 100644
--- a/work/meshai/transport/meshcore_transport.py
+++ b/work/meshai/transport/meshcore_transport.py
@@ -36,6 +36,13 @@ _TELEMETRY_MIN_INTERVAL_SECONDS = 300
# dropped from the auto-poll rotation (a manual "Poll now" un-sticks it).
_TELEMETRY_MAX_FAILURES = 3
+# --- Companion-link keepalive tuning --------------------------------------
+# MeshMonitor's MeshCore vnode (the shared companion-link server meshai
+# attaches to) reaps any client idle >5 min, where "idle" means no bytes seen
+# FROM the client — a periodic LOCAL query resets that clock. 120s is well
+# inside the 300s reaper window with margin to spare.
+_KEEPALIVE_INTERVAL_SECONDS = 120
+
# Numeric Cayenne-LPP type id → decoded field name. Ids not in this map are
# passed through as ``lpp_`` so nothing is silently dropped.
_LPP_ID_TO_FIELD = {
@@ -132,6 +139,8 @@ class MeshCoreTransport(MeshTransport):
self._advert_task = None
# asyncio.Task handle for the telemetry auto-poll loop; None when inactive.
self._telemetry_task = None
+ # asyncio.Task handle for the companion-link keepalive loop; None when inactive.
+ self._keepalive_task = None
# Telemetry availability/bookkeeping (shared by poller + on-demand):
# _telemetry_cache: contact-id -> {contact, data, polled_at, available}
# _telemetry_failures: contact-id -> consecutive-timeout count
@@ -577,15 +586,37 @@ class MeshCoreTransport(MeshTransport):
)
return acked or (not result.is_error())
+ def _resolve_mc_channel_idx(self, meshcore_channel: str) -> Optional[int]:
+ """Resolve a config channel name to the companion's channel slot.
+
+ Tries an exact match first (fast path, preserves existing behavior
+ for e.g. ``#bot``). Falls back to a match that ignores a single
+ leading ``#`` and case, since after the radio moved to MeshMonitor's
+ vnode the companion enumerates region channels WITHOUT the leading
+ ``#`` that meshai's region_routes config still carries (e.g. config
+ ``#sc-id-aida`` vs. companion ``sc-id-aida``) — same channel/key,
+ just a display-name difference upstream.
+ """
+ idx = self._chan_name_to_idx.get(meshcore_channel)
+ if idx is not None:
+ return idx
+ canon = meshcore_channel[1:] if meshcore_channel.startswith("#") else meshcore_channel
+ canon = canon.casefold()
+ for name, slot in self._chan_name_to_idx.items():
+ name_canon = name[1:] if name.startswith("#") else name
+ if name_canon.casefold() == canon:
+ return slot
+ return None
+
async def _do_mc_broadcast_async(self, text: str, meshcore_channel: str) -> bool:
"""Channel broadcast on the MC loop (replaces send_message() broadcast branch)."""
if self._mc is None:
return False
- idx = self._chan_name_to_idx.get(meshcore_channel)
+ idx = self._resolve_mc_channel_idx(meshcore_channel)
if idx is None:
# Lazy async re-enumeration (no _run_coro deadlock risk).
await self._enumerate_channels_async()
- idx = self._chan_name_to_idx.get(meshcore_channel)
+ idx = self._resolve_mc_channel_idx(meshcore_channel)
if idx is None:
logger.warning("MC channel '%s' not on companion; skipping", meshcore_channel)
return False
@@ -1776,6 +1807,66 @@ class MeshCoreTransport(MeshTransport):
if task is not None and self._loop is not None and self._loop.is_running():
self._loop.call_soon_threadsafe(task.cancel)
+ # ------------------------------------------------------------------
+ # Companion-link keepalive (Task on the dedicated loop)
+ # ------------------------------------------------------------------
+
+ async def _keepalive_loop(self) -> None:
+ """Quiet LOCAL companion-link keepalive (Task on the dedicated loop).
+
+ MeshMonitor's MeshCore vnode disconnects any client idle >5 min,
+ where "idle" means no bytes seen FROM the client — its
+ ``lastActivity`` only updates on data we send it, never on data it
+ sends us. The self-advert (every 24h by default) and telemetry poll
+ (30 min default, and only when contacts are configured) are both far
+ too infrequent to keep that clock fresh, so the link was silently
+ reaped and never recovered (``meshcore_auto_reconnect`` is the
+ recovery safety net; this loop is the prevention).
+
+ Every ``_KEEPALIVE_INTERVAL_SECONDS`` (while connected), issues
+ ``commands.get_time()`` — a single-byte companion opcode (CMD 0x05)
+ that reads the node's own onboard clock and returns CURRENT_TIME.
+ It carries no destination/contact and has no mesh-routing semantics
+ (unlike send_advert/send_msg/send_chan_msg), so the firmware answers
+ it purely locally over the companion link — it does not key the
+ radio or emit an RF packet. Runs directly on the MC loop (NOT
+ through the send queue/pacing — it is a device-info query, not a
+ mesh send, so it should never wait behind or delay a real send).
+
+ Stops on CancelledError (disconnect). A transient query failure is
+ logged and ignored — the loop keeps ticking every interval either
+ way, since the point is resetting the vnode's clock on our next
+ successful frame, not the query result itself.
+ """
+ try:
+ while True:
+ await asyncio.sleep(_KEEPALIVE_INTERVAL_SECONDS)
+ if not self._connected or self._mc is None:
+ return
+ try:
+ await self._mc.commands.get_time()
+ logger.debug("MC: companion-link keepalive query sent")
+ except Exception as exc:
+ logger.debug("MC: keepalive get_time failed (non-fatal): %s", exc)
+ except asyncio.CancelledError:
+ logger.debug("MC: keepalive task cancelled")
+ raise
+
+ def _schedule_keepalive(self) -> None:
+ """Create the keepalive asyncio.Task on the dedicated loop (thread-safe)."""
+ def _arm() -> None:
+ self._keepalive_task = asyncio.get_event_loop().create_task(
+ self._keepalive_loop()
+ )
+ self._loop.call_soon_threadsafe(_arm)
+
+ def _cancel_keepalive(self) -> None:
+ """Cancel the keepalive task (thread-safe). Called at disconnect."""
+ task = self._keepalive_task
+ self._keepalive_task = None
+ if task is not None and self._loop is not None and self._loop.is_running():
+ self._loop.call_soon_threadsafe(task.cancel)
+
# ------------------------------------------------------------------
# Internal coroutines (run on the dedicated loop)
# ------------------------------------------------------------------
@@ -1963,6 +2054,12 @@ class MeshCoreTransport(MeshTransport):
if telem_interval > 0:
self._schedule_telemetry_poll()
+ # Arm the quiet local companion-link keepalive — unconditional (not
+ # a mesh operation, no config gate): protects against the
+ # MeshMonitor vnode's 5-min idle reaper regardless of advert/
+ # telemetry cadence.
+ self._schedule_keepalive()
+
logger.info(
"MeshCoreTransport: connected as %s (pubkey %s)",
self._self_info.get("name", "unknown"),
@@ -1971,9 +2068,10 @@ class MeshCoreTransport(MeshTransport):
def disconnect(self) -> None:
"""Disconnect and stop the event loop thread."""
- # Cancel periodic advert + telemetry poll before tearing down the loop.
+ # Cancel periodic advert + telemetry poll + keepalive before tearing down the loop.
self._cancel_periodic_advert()
self._cancel_telemetry_poll()
+ self._cancel_keepalive()
if self._mc is not None:
try:
self._run_coro(self._do_disconnect(), timeout=10.0)
@@ -2078,12 +2176,12 @@ class MeshCoreTransport(MeshTransport):
)
return True
# Resolve NAME → slot against the live companion channel table.
- idx = self._chan_name_to_idx.get(meshcore_channel)
+ idx = self._resolve_mc_channel_idx(meshcore_channel)
if idx is None:
# One lazy re-enumeration in case the table changed since
# connect (e.g. a channel was provisioned after startup).
self._enumerate_channels()
- idx = self._chan_name_to_idx.get(meshcore_channel)
+ idx = self._resolve_mc_channel_idx(meshcore_channel)
if idx is None:
# Never blind-send to a guessed slot.
logger.warning(
@@ -2263,10 +2361,65 @@ class MeshCoreTransport(MeshTransport):
self._connected = False
logger.warning("MeshCoreTransport: DISCONNECTED event received")
+ async def _post_reconnect_setup_async(self) -> None:
+ """Redo connect()'s LOCAL post-connect setup after an auto-reconnect.
+
+ connect() does this setup once, on the initial connect: rebuild
+ ``_chan_name_to_idx`` (so channel-name broadcasts can resolve a
+ slot) and arm the companion-link keepalive (so MeshMonitor's vnode
+ doesn't reap the link again at 5 min idle). The meshcore lib's
+ auto-reconnect only re-establishes the socket and fires CONNECTED
+ (-> ``_on_connect_event``) — it does not repeat that setup, so a
+ reconnected link was left with an empty channel table and no
+ keepalive until the reaper cut it again.
+
+ Both steps here are local companion queries/timers only —
+ ``_enumerate_channels_async`` calls ``get_channel()`` and the
+ keepalive calls ``get_time()`` (see their docstrings); neither
+ keys the radio or emits an RF packet. This deliberately excludes
+ connect()'s ``send_advert()`` — that IS a transmission, and must
+ stay confined to the initial connect() path, never replayed on
+ reconnect.
+
+ Keepalive re-arm is cancel-then-schedule (idempotent) so it never
+ double-schedules the task.
+
+ Runs as a fire-and-forget task on the dedicated MC loop (see
+ ``_on_connect_event``) rather than being awaited inline via
+ ``_run_coro``: the meshcore lib invokes ``_on_connect_event`` from
+ within that same loop (like ``_on_new_contact``), so a blocking
+ ``_run_coro().result()`` call here would deadlock it.
+ """
+ try:
+ await self._enumerate_channels_async()
+ except Exception:
+ logger.warning(
+ "MeshCore: post-reconnect channel re-enumeration failed", exc_info=True
+ )
+ try:
+ self._cancel_keepalive()
+ self._schedule_keepalive()
+ except Exception:
+ logger.warning(
+ "MeshCore: post-reconnect keepalive re-arm failed", exc_info=True
+ )
+
def _on_connect_event(self, event=None) -> None:
- """Track link state: CONNECTED (auto-reconnect succeeded)."""
+ """Track link state: CONNECTED (auto-reconnect succeeded).
+
+ Schedules ``_post_reconnect_setup_async`` fire-and-forget on the
+ dedicated MC loop — see that method's docstring for why this must
+ not block (``_run_coro`` would deadlock from inside this callback,
+ exactly as noted in ``_on_new_contact``).
+ """
self._connected = True
logger.info("MeshCoreTransport: CONNECTED event received")
+ try:
+ loop = getattr(self, "_loop", None)
+ if loop is not None and loop.is_running():
+ asyncio.run_coroutine_threadsafe(self._post_reconnect_setup_async(), loop)
+ except Exception:
+ logger.debug("MeshCore: scheduling post-reconnect setup failed", exc_info=True)
# ------------------------------------------------------------------
# Node identity / topology (MeshTransport abstract methods)
diff --git a/work/tests/test_generic_http.py b/work/tests/test_generic_http.py
index 2b40fab..854aa76 100644
--- a/work/tests/test_generic_http.py
+++ b/work/tests/test_generic_http.py
@@ -8,6 +8,7 @@ Ported-behavior coverage:
* geometry-path Point -> centroid extraction
"""
from __future__ import annotations
+import asyncio
import json
@@ -203,7 +204,7 @@ def test_cold_start_silent_first_poll_seeds_persists_no_emit():
# Stub the network fetch with one active outage.
adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM])
- store.refresh() # poll 1 == pre-existing backlog
+ asyncio.run(store.refresh()) # poll 1 == pre-existing backlog
# Nothing broadcast on the cold-start poll...
assert captured == [], "first poll must broadcast NOTHING (cold-start seed)"
@@ -220,14 +221,14 @@ def test_cold_start_silent_first_poll_seeds_persists_no_emit():
def test_later_poll_broadcasts_newly_received_item():
store, adapter, captured = _make_store_with_generic()
adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM])
- store.refresh() # poll 1 — seed silently
+ asyncio.run(store.refresh()) # poll 1 — seed silently
assert captured == []
# A genuinely NEW outage appears on a later poll -> it must broadcast.
new_item = dict(IDAHO_POWER_ITEM, omsOutageId="456", omsCustomerCount=99)
adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM, new_item])
adapter._last_poll.clear() # force cadence to elapse
- store.refresh() # poll 2
+ asyncio.run(store.refresh()) # poll 2
assert len(captured) == 1, "only the newly-received outage broadcasts"
assert captured[0].category == "power_outage"
@@ -367,7 +368,7 @@ def test_build_generic_detail_reader():
from meshai.notifications.env_reporter import EnvReporter
store, adapter, captured = _make_store_with_generic()
adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM])
- store.refresh()
+ asyncio.run(store.refresh())
text = EnvReporter().build_generic_detail()
assert "idaho_power" in text
diff --git a/work/tests/test_store_received_delta.py b/work/tests/test_store_received_delta.py
index 77555fc..1b253ef 100644
--- a/work/tests/test_store_received_delta.py
+++ b/work/tests/test_store_received_delta.py
@@ -14,6 +14,7 @@ These tests drive the real EnvironmentalStore + EventBus with a fake adapter
whose per-poll batch we control, and assert exactly which events reach the bus.
"""
from __future__ import annotations
+import asyncio
from meshai.env.store import EnvironmentalStore, _key_ext
from meshai.config import EnvironmentalConfig
@@ -79,7 +80,7 @@ def test_first_poll_seeds_and_broadcasts_nothing():
store, adapter, captured = _make_store()
adapter.set_batch(["A", "B", "C"])
- store.refresh() # poll 1 — the backlog
+ asyncio.run(store.refresh()) # poll 1 — the backlog
assert captured == [], "first poll must broadcast NOTHING (backlog seed)"
@@ -88,11 +89,11 @@ def test_second_poll_emits_only_newly_received():
store, adapter, captured = _make_store()
adapter.set_batch(["A", "B", "C"])
- store.refresh() # poll 1: seed
+ asyncio.run(store.refresh()) # poll 1: seed
assert _emitted_ids(captured) == []
adapter.set_batch(["A", "B", "C", "D"])
- store.refresh() # poll 2: only D is new
+ asyncio.run(store.refresh()) # poll 2: only D is new
assert _emitted_ids(captured) == ["D"]
@@ -100,11 +101,11 @@ def test_unchanged_poll_emits_nothing():
store, adapter, captured = _make_store()
adapter.set_batch(["A", "B", "C"])
- store.refresh() # poll 1: seed
+ asyncio.run(store.refresh()) # poll 1: seed
adapter.set_batch(["A", "B", "C", "D"])
- store.refresh() # poll 2: D
+ asyncio.run(store.refresh()) # poll 2: D
adapter.set_batch(["A", "B", "C", "D"])
- store.refresh() # poll 3: nothing new
+ asyncio.run(store.refresh()) # poll 3: nothing new
assert _emitted_ids(captured) == ["D"], "poll 3 has no new items"
@@ -113,21 +114,21 @@ def test_restart_reseeds_and_never_rebroadcasts_backlog():
# Process 1 sees A,B,C,D and broadcasts D.
store1, adapter1, cap1 = _make_store()
adapter1.set_batch(["A", "B", "C"])
- store1.refresh()
+ asyncio.run(store1.refresh())
adapter1.set_batch(["A", "B", "C", "D"])
- store1.refresh()
+ asyncio.run(store1.refresh())
assert _emitted_ids(cap1) == ["D"]
# RESTART: a fresh store has an empty seen-set. The SAME backlog [A,B,C,D]
# arriving on its first poll must be re-seeded silently, not re-broadcast.
store2, adapter2, cap2 = _make_store()
adapter2.set_batch(["A", "B", "C", "D"])
- store2.refresh()
+ asyncio.run(store2.refresh())
assert cap2 == [], "restart must NEVER re-broadcast the existing backlog"
# And a genuinely new item after the restart still broadcasts once.
adapter2.set_batch(["A", "B", "C", "D", "E"])
- store2.refresh()
+ asyncio.run(store2.refresh())
assert _emitted_ids(cap2) == ["E"]
@@ -135,9 +136,9 @@ def test_stable_key_prevents_reemit_when_batch_reorders():
# The same real-world items in a different order are NOT "newly received".
store, adapter, captured = _make_store()
adapter.set_batch(["A", "B", "C"])
- store.refresh() # seed
+ asyncio.run(store.refresh()) # seed
adapter.set_batch(["C", "A", "B"]) # reordered, same items
- store.refresh()
+ asyncio.run(store.refresh())
assert captured == [], "reordering the same items emits nothing"
@@ -147,12 +148,12 @@ def test_disabled_for_days_then_backlog_is_not_broadcast():
store, adapter, captured = _make_store()
backlog = [f"evt{i}" for i in range(200)]
adapter.set_batch(backlog)
- store.refresh() # first poll after re-enable
+ asyncio.run(store.refresh()) # first poll after re-enable
assert captured == [], "a days-old backlog is seeded silently, never sent"
# Only a truly new arrival afterward is announced.
adapter.set_batch(backlog + ["fresh"])
- store.refresh()
+ asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["fresh"]
@@ -290,11 +291,11 @@ def test_persistent_preseed_known_suppressed_new_emitted():
assert len(store._seen["wzdx"]) == 5
adapter.set_batch(known)
- store.refresh()
+ asyncio.run(store.refresh())
assert captured == [], "all 5 are durably-known → zero broadcast"
adapter.set_batch(known + ["z_new"])
- store.refresh()
+ asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["z_new"], "only the not-in-table id broadcasts"
@@ -307,9 +308,9 @@ def test_persistent_preseed_cross_tick_staging_no_leak():
store, captured = _build_store(_GENERIC_NAME, adapter)
adapter.set_batch(["A"])
- store.refresh() # tick 1: only A present
+ asyncio.run(store.refresh()) # tick 1: only A present
adapter.set_batch(["A", "B"])
- store.refresh() # tick 2: B appears (backlog)
+ asyncio.run(store.refresh()) # tick 2: B appears (backlog)
assert captured == [], "B is durably-known — must NOT leak on a later tick"
# CONTROL: identical staging but NO durable rows → B leaks (proves the
@@ -325,11 +326,11 @@ def test_persistent_preseed_cross_tick_staging_no_leak():
# Re-point ctrl events to a fresh source with no durable rows.
for e in ctrl._batch:
e["source"] = "wzdx_ctrl"
- store2.refresh()
+ asyncio.run(store2.refresh())
ctrl.set_batch(["A", "B"])
for e in ctrl._batch:
e["source"] = "wzdx_ctrl"
- store2.refresh()
+ asyncio.run(store2.refresh())
assert [e.title for e in cap2] == ["B"], "without a durable record, B leaks"
@@ -343,11 +344,11 @@ def test_incremental_empty_first_tick_then_only_new_broadcasts():
store, captured = _build_store(_GENERIC_NAME, adapter)
adapter.set_batch([]) # empty first tick
- store.refresh()
+ asyncio.run(store.refresh())
assert captured == [], "empty tick emits nothing"
adapter.set_batch(["A", "B", "C"]) # backlog A,B + new C
- store.refresh()
+ asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["C"], "only the never-received C broadcasts"
@@ -360,18 +361,18 @@ def test_restart_against_same_persistent_db_never_rebroadcasts():
a1 = _FakeWZDx()
store1, cap1 = _build_store(_GENERIC_NAME, a1)
a1.set_batch(backlog)
- store1.refresh()
+ asyncio.run(store1.refresh())
assert cap1 == [], "process 1: durable backlog is silent"
# RESTART: brand-new store, same persistent DB → pre-seed reloads.
a2 = _FakeWZDx()
store2, cap2 = _build_store(_GENERIC_NAME, a2)
a2.set_batch(backlog)
- store2.refresh()
+ asyncio.run(store2.refresh())
assert cap2 == [], "restart must NEVER re-broadcast the durable backlog"
a2.set_batch(backlog + ["E"])
- store2.refresh()
+ asyncio.run(store2.refresh())
assert _emitted_ids(cap2) == ["E"], "a genuinely-new item still broadcasts once"
@@ -385,11 +386,11 @@ def test_persistent_preseed_quake_by_event_id():
assert len(store._seen["usgs_quake"]) == 2
adapter.set_batch(["us1000aaaa", "us1000bbbb"])
- store.refresh()
+ asyncio.run(store.refresh())
assert captured == [], "both quakes already received → zero broadcast"
adapter.set_batch(["us1000aaaa", "us1000bbbb", "us1000cccc"])
- store.refresh()
+ asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["us1000cccc"], "only the new quake broadcasts"
@@ -402,11 +403,11 @@ def test_no_durable_rows_falls_back_to_silent_first_poll():
assert "wzdx" not in store._seeded, "0 durable rows → not pre-marked seeded"
adapter.set_batch(["A", "B"])
- store.refresh()
+ asyncio.run(store.refresh())
assert captured == [], "first non-empty poll on a fresh DB is silent"
adapter.set_batch(["A", "B", "C"])
- store.refresh()
+ asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["C"]
@@ -486,9 +487,9 @@ def test_persistent_preseed_roads511_by_external_id():
assert len(store._seen["511"]) == 4
adapter.set_batch(known)
- store.refresh()
+ asyncio.run(store.refresh())
assert captured == [], "all 4 durably-known 511 rows → zero broadcast"
adapter.set_batch(known + ["511_99"])
- store.refresh()
+ asyncio.run(store.refresh())
assert _emitted_ids(captured) == ["511_99"], "only the not-in-table id broadcasts"
diff --git a/work/tests/test_store_wzdx_persist.py b/work/tests/test_store_wzdx_persist.py
index 13422ee..4e60a97 100644
--- a/work/tests/test_store_wzdx_persist.py
+++ b/work/tests/test_store_wzdx_persist.py
@@ -20,6 +20,7 @@ adapter whose per-poll coalesced set we control, then assert directly against
traffic_events AND against the bus (nothing must ever be dispatched).
"""
from __future__ import annotations
+import asyncio
from meshai.env.store import EnvironmentalStore
from meshai.config import EnvironmentalConfig
@@ -147,7 +148,7 @@ def test_first_poll_persists_current_set_and_broadcasts_nothing():
store, captured = _build_store(adapter)
adapter.set_zones(ZONES3)
- store.refresh() # first (cold-start) poll
+ asyncio.run(store.refresh()) # first (cold-start) poll
rows = _wzdx_rows()
exts = {r["external_id"] for r in rows}
@@ -171,7 +172,7 @@ def test_columns_match_summary_and_dm_queries():
adapter = _FakeWZDx()
store, _ = _build_store(adapter)
adapter.set_zones([ZONES3[1]]) # the full_closure I-84 zone
- store.refresh()
+ asyncio.run(store.refresh())
r = _wzdx_rows()[0]
assert r["road"] == "I-84"
@@ -188,12 +189,12 @@ def test_subsequent_poll_reconciles_removed_zone():
adapter = _FakeWZDx()
store, captured = _build_store(adapter)
adapter.set_zones(ZONES3)
- store.refresh()
+ asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 3
# Next poll: US-20 dropped out; I-84 + ID-55 remain.
adapter.set_zones([ZONES3[1], ZONES3[2]])
- store.refresh()
+ asyncio.run(store.refresh())
exts = {r["external_id"] for r in _wzdx_rows()}
assert exts == {ZONES3[1]["ext"], ZONES3[2]["ext"]}, (
@@ -207,11 +208,11 @@ def test_empty_or_failed_fetch_does_not_wipe_existing_rows():
adapter = _FakeWZDx()
store, _ = _build_store(adapter)
adapter.set_zones(ZONES3)
- store.refresh()
+ asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 3
adapter.set_raw([]) # empty/failed poll
- store.refresh()
+ asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 3, (
"an empty fetch must NEVER wipe the existing active set")
@@ -225,7 +226,7 @@ def test_upsert_preserves_first_seen_at_and_refreshes_end_at():
z = dict(ZONES3[0]); z["end_at"] = 1000
adapter.set_zones([z])
- store.refresh()
+ asyncio.run(store.refresh())
r1 = _wzdx_rows()[0]
first_seen = r1["first_seen_at"]
assert r1["end_at"] == 1000
@@ -233,7 +234,7 @@ def test_upsert_preserves_first_seen_at_and_refreshes_end_at():
# Same zone reappears with a LATER end_at.
z2 = dict(ZONES3[0]); z2["end_at"] = 5000
adapter.set_zones([z2])
- store.refresh()
+ asyncio.run(store.refresh())
r2 = _wzdx_rows()[0]
assert r2["first_seen_at"] == first_seen, "first_seen_at must be preserved"
assert r2["end_at"] == 5000, "end_at must refresh from the feed"
@@ -254,7 +255,7 @@ def test_expiry_end_at_preserved_for_not_expired_filter():
"sub_type": "x", "impact": "partial", "end_at": now - 10_000}, # expired
]
adapter.set_zones(zones)
- store.refresh()
+ asyncio.run(store.refresh())
# All 3 persisted (ingest does not itself drop expired rows) ...
assert len(_wzdx_rows()) == 3
@@ -269,7 +270,7 @@ def test_id_less_zone_is_skipped_not_fatal():
store, _ = _build_store(adapter)
good = ZONES3[0]
adapter.set_zones([good])
- store.refresh()
+ asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 1
# Poll with the good zone plus an id-less junk event.
@@ -277,7 +278,7 @@ def test_id_less_zone_is_skipped_not_fatal():
junk = {"source": "wzdx", "event_id": None, "external_id": None,
"lat": 5.0, "lon": 5.0, "normalized": {}, "fetched_at": 0}
adapter._batch.append(junk)
- store.refresh()
+ asyncio.run(store.refresh())
rows = _wzdx_rows()
assert {r["external_id"] for r in rows} == {good["ext"]}, (
@@ -296,7 +297,7 @@ def test_bulk_current_set_persists_all_like_the_real_127():
for i in range(127)
]
adapter.set_zones(zones)
- store.refresh()
+ asyncio.run(store.refresh())
assert len(_wzdx_rows()) == 127
assert _summary_visible_count(now=0) == 127, (