mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
Fix event-loop starvation, MeshCore stability, config-page hardening
- mesh_data_store.py / env/store.py: make refresh() async, offload blocking polls via asyncio.to_thread/gather so 7 lockstep sources no longer starve the shared event loop. - main.py: gather pollers concurrently + set_default_executor thread pool. - Dockerfile / docker-compose.yml: healthcheck now curls the dashboard for a real liveness signal instead of a process-exists check. - transport/meshcore_transport.py: MeshCore keepalive loop (get_time() every 120s), reconnect re-arm (_post_reconnect_setup_async from _on_connect_event), and MC channel-name normalization (_resolve_mc_channel_idx strips a leading #). - dashboard-frontend: MeshCoreConnection.tsx config-page hardening, new ErrorBoundary component, wired into App.tsx. - tests: fix ~40 call sites broken by refresh() becoming async ( test_generic_http.py, test_store_received_delta.py, test_store_wzdx_persist.py) by wrapping with asyncio.run(), matching this suite's existing convention for calling async code from sync test functions. Verified: all 40 tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Matt Johnson
parent
8460ab50e0
commit
2b448e064b
12 changed files with 462 additions and 129 deletions
|
|
@ -88,8 +88,8 @@ VOLUME ["/data"]
|
|||
EXPOSE 8080
|
||||
|
||||
# Health check - verify bot process is alive via PID file
|
||||
HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
|
||||
CMD test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ "$(cat /tmp/meshai.link 2>/dev/null)" = up ] || exit 1
|
||||
HEALTHCHECK --interval=30s --timeout=10s --start-period=240s --retries=3 \
|
||||
CMD curl -f -s -o /dev/null http://localhost:8080/ || exit 1
|
||||
|
||||
# Entrypoint writes default config on first run, then starts the bot
|
||||
ENTRYPOINT ["/app/docker-entrypoint.sh"]
|
||||
|
|
|
|||
|
|
@ -24,12 +24,14 @@ import MeshCoreDangerZones from './pages/MeshCoreDangerZones'
|
|||
import Coverage from './pages/Coverage'
|
||||
import { ToastProvider } from './components/ToastProvider'
|
||||
import { DirtyProvider } from './context/DirtyContext'
|
||||
import ErrorBoundary from './components/ErrorBoundary'
|
||||
|
||||
function App() {
|
||||
return (
|
||||
<DirtyProvider>
|
||||
<ToastProvider>
|
||||
<Layout>
|
||||
<ErrorBoundary>
|
||||
<Routes>
|
||||
{/* Core routes */}
|
||||
<Route path="/" element={<Dashboard />} />
|
||||
|
|
@ -69,6 +71,7 @@ function App() {
|
|||
<Route path="/meshcore/companion" element={<MeshCoreCompanion />} />
|
||||
<Route path="/meshcore/danger-zones" element={<MeshCoreDangerZones />} />
|
||||
</Routes>
|
||||
</ErrorBoundary>
|
||||
</Layout>
|
||||
</ToastProvider>
|
||||
</DirtyProvider>
|
||||
|
|
|
|||
49
work/dashboard-frontend/src/components/ErrorBoundary.tsx
Normal file
49
work/dashboard-frontend/src/components/ErrorBoundary.tsx
Normal file
|
|
@ -0,0 +1,49 @@
|
|||
// App-wide render-error safety net. Wraps <Routes> in App.tsx so an
|
||||
// unhandled error thrown while rendering any page degrades to a recoverable
|
||||
// "Something went wrong" card instead of a blank white screen.
|
||||
import { Component, type ErrorInfo, type ReactNode } from 'react'
|
||||
import { AlertTriangle } from 'lucide-react'
|
||||
|
||||
interface Props {
|
||||
children: ReactNode
|
||||
}
|
||||
|
||||
interface State {
|
||||
hasError: boolean
|
||||
error: Error | null
|
||||
}
|
||||
|
||||
export default class ErrorBoundary extends Component<Props, State> {
|
||||
state: State = { hasError: false, error: null }
|
||||
|
||||
static getDerivedStateFromError(error: Error): State {
|
||||
return { hasError: true, error }
|
||||
}
|
||||
|
||||
componentDidCatch(error: Error, errorInfo: ErrorInfo) {
|
||||
console.error('MeshAI dashboard render error:', error, errorInfo)
|
||||
}
|
||||
|
||||
render() {
|
||||
if (this.state.hasError) {
|
||||
return (
|
||||
<div className="flex items-center justify-center h-64">
|
||||
<div className="bg-bg-card border border-red-500/20 rounded p-6 max-w-md w-full text-center space-y-3">
|
||||
<AlertTriangle className="mx-auto text-red-400" size={28} />
|
||||
<div className="text-slate-200 font-medium">Something went wrong</div>
|
||||
<div className="text-xs text-slate-500">
|
||||
{this.state.error?.message ?? 'An unexpected error occurred while rendering this page.'}
|
||||
</div>
|
||||
<button
|
||||
onClick={() => window.location.reload()}
|
||||
className="px-4 py-2 bg-accent hover:bg-accent/80 rounded text-white text-sm transition-colors"
|
||||
>
|
||||
Reload
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
return this.props.children
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
import { useState, useEffect, useCallback } from 'react'
|
||||
import { Link } from 'react-router-dom'
|
||||
import { Save, RotateCcw, RefreshCw, Check, ChevronRight, Trash2, Eye, EyeOff, Copy } from 'lucide-react'
|
||||
import { Save, RotateCcw, RefreshCw, Check, ChevronRight, Trash2, Eye, EyeOff, Copy, X } from 'lucide-react'
|
||||
import { TextInput, NumberInput, Toggle, ListInput, SelectInput } from './Config'
|
||||
import SerialPortPicker from '@/components/SerialPortPicker'
|
||||
import { notifyRestartRequired } from '@/components/RestartBanner'
|
||||
|
|
@ -58,6 +58,9 @@ export default function MeshCoreConnection() {
|
|||
const [error, setError] = useState<string | null>(null)
|
||||
const [success, setSuccess] = useState<string | null>(null)
|
||||
const [hasChanges, setHasChanges] = useState(false)
|
||||
// Set when one or both config fetches fail; the form still renders (using
|
||||
// safe defaults for whatever didn't load) instead of blanking the page.
|
||||
const [loadError, setLoadError] = useState<string | null>(null)
|
||||
|
||||
// Test send state
|
||||
const [channelsActive, setChannelsActive] = useState(false)
|
||||
|
|
@ -81,22 +84,41 @@ export default function MeshCoreConnection() {
|
|||
|
||||
const fetchConfig = useCallback(async () => {
|
||||
setLoading(true)
|
||||
try {
|
||||
const [data, mcCtx] = await Promise.all([
|
||||
apiFetchConfig('connection') as Promise<ConnectionConfig>,
|
||||
apiFetchConfig('meshcore_context') as Promise<MeshcoreContextCfg>,
|
||||
])
|
||||
setConfig(data)
|
||||
setOriginalConfig(JSON.parse(JSON.stringify(data)))
|
||||
setMcContext(mcCtx)
|
||||
setOriginalMcContext(JSON.parse(JSON.stringify(mcCtx)))
|
||||
setHasChanges(false)
|
||||
setError(null)
|
||||
} catch (err) {
|
||||
setError(err instanceof Error ? err.message : 'Unknown error')
|
||||
} finally {
|
||||
setLoading(false)
|
||||
// Promise.allSettled so one section failing to load can't blank the
|
||||
// whole page — each section is set independently from its own result,
|
||||
// and any failure(s) are surfaced as a dismissible banner alongside the
|
||||
// still-usable form (see the `!config` dead-end this replaces).
|
||||
const [connResult, mcResult] = await Promise.allSettled([
|
||||
apiFetchConfig('connection') as Promise<ConnectionConfig>,
|
||||
apiFetchConfig('meshcore_context') as Promise<MeshcoreContextCfg>,
|
||||
])
|
||||
|
||||
const errors: string[] = []
|
||||
|
||||
if (connResult.status === 'fulfilled') {
|
||||
setConfig(connResult.value)
|
||||
setOriginalConfig(JSON.parse(JSON.stringify(connResult.value)))
|
||||
} else {
|
||||
errors.push(`connection (${connResult.reason instanceof Error ? connResult.reason.message : String(connResult.reason)})`)
|
||||
// Fall back to an empty object (never null) so the form below always
|
||||
// has something to render safe defaults from, and so the dirty-check
|
||||
// diff below still has a baseline to compare edits against.
|
||||
setConfig((c) => c ?? {})
|
||||
setOriginalConfig((c) => c ?? {})
|
||||
}
|
||||
|
||||
if (mcResult.status === 'fulfilled') {
|
||||
setMcContext(mcResult.value)
|
||||
setOriginalMcContext(JSON.parse(JSON.stringify(mcResult.value)))
|
||||
} else {
|
||||
errors.push(`bot behavior (${mcResult.reason instanceof Error ? mcResult.reason.message : String(mcResult.reason)})`)
|
||||
// Leave mcContext null — the "Bot behavior" card only renders when it
|
||||
// is present, so a null value just hides that card cleanly.
|
||||
}
|
||||
|
||||
setLoadError(errors.length ? `Couldn't load ${errors.join(' and ')}. Showing defaults — edits below are still safe to make and save.` : null)
|
||||
setHasChanges(false)
|
||||
setLoading(false)
|
||||
}, [])
|
||||
|
||||
useEffect(() => {
|
||||
|
|
@ -207,10 +229,13 @@ export default function MeshCoreConnection() {
|
|||
}
|
||||
|
||||
useEffect(() => {
|
||||
if (config && originalConfig && mcContext && originalMcContext) {
|
||||
// mcContext may be null (its fetch failed) — don't let that block
|
||||
// detecting changes to the connection fields, which always load or
|
||||
// fall back to an empty-object baseline in fetchConfig.
|
||||
if (config && originalConfig) {
|
||||
const changed =
|
||||
JSON.stringify(config) !== JSON.stringify(originalConfig) ||
|
||||
JSON.stringify(mcContext) !== JSON.stringify(originalMcContext)
|
||||
(!!mcContext && !!originalMcContext && JSON.stringify(mcContext) !== JSON.stringify(originalMcContext))
|
||||
setHasChanges(changed)
|
||||
}
|
||||
}, [config, originalConfig, mcContext, originalMcContext])
|
||||
|
|
@ -221,22 +246,24 @@ export default function MeshCoreConnection() {
|
|||
}, [hasChanges, setDirty])
|
||||
|
||||
const upd = (patch: Partial<ConnectionConfig>) =>
|
||||
setConfig((c) => (c ? { ...c, ...patch } : c))
|
||||
// Build off an empty object rather than bailing when config is still
|
||||
// null (e.g. mid-retry) — edits should never be silently dropped.
|
||||
setConfig((c) => ({ ...(c ?? {}), ...patch }))
|
||||
|
||||
const saveConfig = async () => {
|
||||
if (!config || !mcContext) return
|
||||
if (!config) return
|
||||
setSaving(true)
|
||||
setError(null)
|
||||
setSuccess(null)
|
||||
try {
|
||||
// PUT the whole objects so sibling fields (Meshtastic connection fields,
|
||||
// any other meshcore_context keys) are preserved.
|
||||
const results = await Promise.all([
|
||||
apiUpdateConfig('connection', config),
|
||||
apiUpdateConfig('meshcore_context', mcContext),
|
||||
])
|
||||
// any other meshcore_context keys) are preserved. Only PUT
|
||||
// meshcore_context if it actually loaded — its fetch may have failed.
|
||||
const puts: Promise<{ restart_required?: boolean }>[] = [apiUpdateConfig('connection', config)]
|
||||
if (mcContext) puts.push(apiUpdateConfig('meshcore_context', mcContext))
|
||||
const results = await Promise.all(puts)
|
||||
setOriginalConfig(JSON.parse(JSON.stringify(config)))
|
||||
setOriginalMcContext(JSON.parse(JSON.stringify(mcContext)))
|
||||
if (mcContext) setOriginalMcContext(JSON.parse(JSON.stringify(mcContext)))
|
||||
setHasChanges(false)
|
||||
setDirty(false)
|
||||
setSuccess('MeshCore connection saved successfully')
|
||||
|
|
@ -276,13 +303,10 @@ export default function MeshCoreConnection() {
|
|||
)
|
||||
}
|
||||
|
||||
if (!config) {
|
||||
return (
|
||||
<div className="flex items-center justify-center h-64">
|
||||
<div className="text-red-400">Failed to load connection config</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
// Always render the form below, even if the fetch failed — `cfg` supplies
|
||||
// safe defaults so the connection-type/host/port fields and Save/Discard
|
||||
// stay usable. The failure itself is surfaced by the loadError banner.
|
||||
const cfg: ConnectionConfig = config ?? {}
|
||||
|
||||
return (
|
||||
<div className="max-w-2xl mx-auto space-y-6">
|
||||
|
|
@ -320,6 +344,30 @@ export default function MeshCoreConnection() {
|
|||
</div>
|
||||
</div>
|
||||
|
||||
{/* Load-error banner — dismissible, with its own Retry so the page
|
||||
never dead-ends when a config section fails to fetch. The form
|
||||
below stays fully editable regardless. */}
|
||||
{loadError && (
|
||||
<div className="p-3 text-sm bg-red-500/10 text-red-400 border border-red-500/20 flex items-start gap-3">
|
||||
<div className="flex-1">{loadError}</div>
|
||||
<button
|
||||
onClick={fetchConfig}
|
||||
className="flex items-center gap-1 px-2 py-1 bg-red-500/20 hover:bg-red-500/30 rounded text-red-300 text-xs shrink-0"
|
||||
>
|
||||
<RefreshCw size={12} />
|
||||
Retry
|
||||
</button>
|
||||
<button
|
||||
onClick={() => setLoadError(null)}
|
||||
title="Dismiss"
|
||||
aria-label="Dismiss load error"
|
||||
className="text-red-400 hover:text-red-200 shrink-0"
|
||||
>
|
||||
<X size={14} />
|
||||
</button>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* Status messages */}
|
||||
{error && (
|
||||
<div className="p-3 text-sm bg-red-500/10 text-red-400 border border-red-500/20">{error}</div>
|
||||
|
|
@ -341,7 +389,7 @@ export default function MeshCoreConnection() {
|
|||
</p>
|
||||
<SelectInput
|
||||
label="Connection Type"
|
||||
value={config.meshcore_conn_type ?? 'tcp'}
|
||||
value={cfg.meshcore_conn_type ?? 'tcp'}
|
||||
onChange={(v) => upd({ meshcore_conn_type: v })}
|
||||
options={[
|
||||
{ value: 'tcp', label: 'TCP (companion)' },
|
||||
|
|
@ -350,11 +398,11 @@ export default function MeshCoreConnection() {
|
|||
]}
|
||||
helper="TCP for a companion frame server, Serial for a USB node, BLE for Bluetooth"
|
||||
/>
|
||||
{(config.meshcore_conn_type ?? 'tcp') === 'tcp' && (
|
||||
{(cfg.meshcore_conn_type ?? 'tcp') === 'tcp' && (
|
||||
<div className="grid grid-cols-2 gap-4">
|
||||
<TextInput
|
||||
label="MeshCore Host"
|
||||
value={config.meshcore_host ?? ''}
|
||||
value={cfg.meshcore_host ?? ''}
|
||||
onChange={(v) => upd({ meshcore_host: v })}
|
||||
placeholder="192.168.1.100"
|
||||
helper="IP or hostname of the companion frame server"
|
||||
|
|
@ -362,7 +410,7 @@ export default function MeshCoreConnection() {
|
|||
/>
|
||||
<NumberInput
|
||||
label="MeshCore Port"
|
||||
value={config.meshcore_port ?? 5525}
|
||||
value={cfg.meshcore_port ?? 5525}
|
||||
onChange={(v) => upd({ meshcore_port: v })}
|
||||
min={1}
|
||||
max={65535}
|
||||
|
|
@ -370,27 +418,27 @@ export default function MeshCoreConnection() {
|
|||
/>
|
||||
</div>
|
||||
)}
|
||||
{(config.meshcore_conn_type ?? 'tcp') === 'serial' && (
|
||||
{(cfg.meshcore_conn_type ?? 'tcp') === 'serial' && (
|
||||
<>
|
||||
<SerialPortPicker
|
||||
label="MeshCore Serial Port"
|
||||
value={config.meshcore_serial_port ?? ''}
|
||||
value={cfg.meshcore_serial_port ?? ''}
|
||||
onChange={(v) => upd({ meshcore_serial_port: v })}
|
||||
helper="USB-attached MeshCore node — Detect fills a stable by-id path"
|
||||
/>
|
||||
<NumberInput
|
||||
label="Baud Rate"
|
||||
value={config.meshcore_baud ?? 115200}
|
||||
value={cfg.meshcore_baud ?? 115200}
|
||||
onChange={(v) => upd({ meshcore_baud: v })}
|
||||
min={1200}
|
||||
helper="Serial baud rate (default 115200)"
|
||||
/>
|
||||
</>
|
||||
)}
|
||||
{(config.meshcore_conn_type ?? 'tcp') === 'ble' && (
|
||||
{(cfg.meshcore_conn_type ?? 'tcp') === 'ble' && (
|
||||
<TextInput
|
||||
label="BLE Address"
|
||||
value={config.meshcore_ble_address ?? ''}
|
||||
value={cfg.meshcore_ble_address ?? ''}
|
||||
onChange={(v) => upd({ meshcore_ble_address: v })}
|
||||
placeholder="AA:BB:CC:DD:EE:FF"
|
||||
helper="Leave blank to scan/pair the first available device"
|
||||
|
|
@ -406,7 +454,7 @@ export default function MeshCoreConnection() {
|
|||
</div>
|
||||
<Toggle
|
||||
label="Auto-add contacts (AIDA adds any node it hears — required to DM anyone)"
|
||||
checked={config.meshcore_auto_add_contacts ?? true}
|
||||
checked={cfg.meshcore_auto_add_contacts ?? true}
|
||||
onChange={(v) => 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() {
|
|||
<div className="mt-4 space-y-4 pl-6 border-l border-[#1e2a3a]">
|
||||
<Toggle
|
||||
label="Auto-reconnect (MeshCore)"
|
||||
checked={config.meshcore_auto_reconnect ?? true}
|
||||
checked={cfg.meshcore_auto_reconnect ?? true}
|
||||
onChange={(v) => upd({ meshcore_auto_reconnect: v })}
|
||||
helper="Automatically reconnect to the MeshCore companion if the link drops"
|
||||
/>
|
||||
<NumberInput
|
||||
label="Max Reconnect Attempts"
|
||||
value={config.meshcore_max_reconnect_attempts ?? 5}
|
||||
value={cfg.meshcore_max_reconnect_attempts ?? 5}
|
||||
onChange={(v) => upd({ meshcore_max_reconnect_attempts: v })}
|
||||
min={0}
|
||||
helper="Maximum reconnect attempts before giving up (0 = unlimited)"
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
33
work/meshai/env/store.py
vendored
33
work/meshai/env/store.py
vendored
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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_<id>`` 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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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, (
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue