Files
Kpa-clawbot 242c7c609b fix(mqtt): escalate persistent paho disconnect + recover from emit panic + expose watchdog tick (#1749) (#1810)
# Partial fix for #1749 — MQTT watchdog escalation + panic recovery +
tick exposure

Red commit: 9912bbb3e9 (CI run:
https://github.com/Kpa-clawbot/CoreScope/actions/runs?branch=fix%2Fissue-1749)

## Problem
Production CoreScope v3.9.1 (and a more recent prod recurrence on
2026-06-30 with the wcmesh source on `ssl://mqtt2.wcmesh.com:8883`)
showed two distinct watchdog failure modes:

1. **Per-source paho machinery dies silently.** `IsConnectedFn` returns
false; paho's `SetAutoReconnect(true)` never retries. The watchdog's
`processLivenessTransition` deliberately stays silent on
`LivenessDisconnected`, trusting paho to recover — so there is no
escalation path when that trust is misplaced.
2. **Watchdog goroutine death.** Three sources went silent within ~60s
of each other; no `WATCHDOG` log lines for 75 min. The most plausible
single point of failure is a panic inside `emit` (e.g. a blocked log
pipe) killing the loop with no defer/recover.

## Changes

**`cmd/ingestor/mqtt_watchdog.go`**
- Added `disconnectedReconnectMultiplier = 5` constant.
- Added `DisconnectedSinceUnix int64` (atomic) on `SourceLivenessState`.
Stamped on the first tick the source is observed disconnected; cleared
on any non-disconnected tick.
- `processLivenessTransition` now escalates: when `(now -
DisconnectedSinceUnix) > multiplier × threshold`, emits a `WATCHDOG
ESCALATION` WARN and calls `maybeForceReconnect` (subject to existing
`forceReconnectThrottle`). Distinct from the existing `LivenessStalled`
path so operators can grep escalation events independently.
- Added package-level `watchdogLastTickUnix atomic.Int64` +
`WatchdogLastTickUnix()` getter. The loop stamps it BEFORE per-source
processing — a wedged source-handler does not freeze the clock for an
external observer.
- `runLivenessWatchdogLoop` wraps each per-source
`processLivenessTransition` call in `func() { defer recover; ... }()` so
a panic in `emit` (or in any per-source code path) is logged and
skipped, not fatal. The loop continues to the next source and the next
tick.

**`cmd/ingestor/stats_file.go`**
- Added `WatchdogLastTickUnix int64` field on `IngestorStatsSnapshot`
(additive, `omitempty`); populated from `WatchdogLastTickUnix()` each
stats tick.

**`cmd/server/mqtt_status.go`**
- `MqttStatusResponse` gains `WatchdogLastTickUnix int64` (additive,
`omitempty`) sourced from the ingestor stats file; surfaced via `GET
/api/mqtt/status`.

**`config.example.json`**
- No new config field added — the multiplier is a code constant (5×) per
the issue's "N×threshold (e.g. 5×)" recommendation. The active
per-source `threshold` is the 5-minute scan threshold hard-coded at the
sole `runLivenessWatchdog` callsite (`cmd/ingestor/main.go:460`:
`runLivenessWatchdog(60*time.Second, 5*time.Minute)`), so escalation
fires at ~25 minutes (5 × 5min) of continuous disconnect — plus a
deterministic per-source jitter of 0..30s (#1810 round-1, see Taleb #4)
to avoid synchronized escalation across N sources sharing an upstream
broker outage.

## Acceptance (#1749)
- [x] Persistent `LivenessDisconnected` > N×threshold → force-reconnect
+ WARN
- [x] Watchdog goroutine liveness clock exposed (`WatchdogLastTickUnix`
in `/api/mqtt/status`)
- [x] Test: `IsConnectedFn` false for >5×threshold → assert
`ForceReconnectFn` invoked at least once
- [x] Test: panic in `emit` → assert loop recovers and continues ticking

## Test plan
- 4 new tests in `cmd/ingestor/mqtt_watchdog_1749_test.go` (all RED on
master, GREEN on this PR).
- Existing watchdog tests (`mqtt_watchdog_force_reconnect_test.go`,
`mqtt_reconnect_test.go`, r1/r2/m1 suites) continue to pass — the
escalation path is additive.

## Preflight
- TDD: red commit pushed and asserted to fail BEFORE green commit
landed.
- PII grep: clean on diff and PR body.
- Worktree: `_wt-fix-1749` on branch `fix/issue-1749`.

---------

Co-authored-by: corescope-bot <bot@corescope.dev>
Co-authored-by: openclaw-bot <bot@openclaw.local>
Co-authored-by: bot <bot@local>
2026-06-30 15:16:31 -07:00
..

MeshCore MQTT Ingestor (Go)

Standalone MQTT ingestion service for CoreScope. Connects to MQTT brokers, decodes raw MeshCore packets, and writes to the same SQLite database used by the Node.js web server.

This is the first step of a larger Go rewrite — separating MQTT ingestion from the web server.

Architecture

MQTT Broker(s)  →  Go Ingestor  →  SQLite DB  ←  Node.js Web Server
                    (this binary)     (shared)
  • Single static binary — no runtime dependencies, no CGO
  • SQLite via modernc.org/sqlite (pure Go)
  • MQTT via github.com/eclipse/paho.mqtt.golang
  • Runs alongside the Node.js server — they share the DB file
  • Does NOT serve HTTP/WebSocket — that stays in Node.js

Build

Requires Go 1.22+.

cd cmd/ingestor
go build -o corescope-ingestor .

Cross-compile for Linux (e.g., for the production VM):

GOOS=linux GOARCH=amd64 go build -o corescope-ingestor .

Run

./corescope-ingestor -config /path/to/config.json

The config file uses the same format as the Node.js config.json. The ingestor reads the mqttSources array (or legacy mqtt object) and dbPath fields.

Environment Variables

Variable Description Default
DB_PATH SQLite database path data/meshcore.db
MQTT_BROKER Single MQTT broker URL (overrides config)
MQTT_TOPIC MQTT topic (used with MQTT_BROKER) meshcore/#
CORESCOPE_INGESTOR_STATS Path to the per-second stats JSON file consumed by the server's /api/perf/io and /api/perf/write-sources endpoints (#1120) /tmp/corescope-ingestor-stats.json

Stats file (CORESCOPE_INGESTOR_STATS)

Every second the ingestor publishes a JSON snapshot of its counters (tx_inserted, obs_inserted, walCommits, backfillUpdates.*, etc.) plus a procIO block sampled from /proc/self/io (read/write/cancelled bytes per second + syscall counts). The server reads this file and surfaces the data on the Perf page so operators can self-diagnose write-volume anomalies.

The writer uses O_NOFOLLOW | O_CREAT | O_TRUNC mode 0o600, so a pre-planted symlink at the path cannot be used to clobber an arbitrary file.

Security note: the default lives in /tmp, which is world-writable on most hosts (sticky bit only protects deletion, not creation). On shared/multi-tenant hosts, override CORESCOPE_INGESTOR_STATS to point at a private directory (e.g. /var/lib/corescope/ingestor-stats.json) that only the corescope user can write to.

Minimal Config

{
  "dbPath": "data/meshcore.db",
  "mqttSources": [
    {
      "name": "local",
      "broker": "mqtt://localhost:1883",
      "topics": ["meshcore/#"]
    }
  ]
}

Full Config (same as Node.js)

The ingestor reads these fields from the existing config.json:

  • mqttSources[] — array of MQTT broker connections
    • name — display name for logging
    • broker — MQTT URL (mqtt://, mqtts://)
    • username / password — auth credentials
    • topics — array of topic patterns to subscribe
    • iataFilter — optional regional filter
  • mqtt — legacy single-broker config (auto-converted to mqttSources)
  • dbPath — SQLite DB path (default: data/meshcore.db)

Test

cd cmd/ingestor
go test -v ./...

What It Does

  1. Connects to configured MQTT brokers with auto-reconnect
  2. Subscribes to mesh packet topics (e.g., meshcore/+/+/packets)
  3. Receives raw hex packets via JSON messages ({ "raw": "...", "SNR": ..., "RSSI": ... })
  4. Decodes MeshCore packet headers, paths, and payloads (ported from decoder.js)
  5. Computes content hashes (path-independent, SHA-256-based)
  6. Writes to SQLite: transmissions + observations tables
  7. Upserts nodes from decoded ADVERT packets (with validation)
  8. Upserts observers from MQTT topic metadata

Schema Compatibility

The Go ingestor creates the same v3 schema as the Node.js server:

  • transmissions — deduplicated by content hash
  • observations — per-observer sightings with observer_idx (rowid reference)
  • nodes — mesh nodes discovered from adverts
  • observers — MQTT feed sources

Both processes can write to the same DB concurrently (SQLite WAL mode).

What's Not Ported (Yet)

  • Companion bridge format (Format 2 — meshcore/advertisement, channel messages, etc.)
  • Channel key decryption (GRP_TXT encrypted payload decryption)
  • WebSocket broadcast to browsers
  • In-memory packet store
  • Cache invalidation

These stay in the Node.js server for now.

Files

cmd/ingestor/
  main.go          — entry point, MQTT connect, message handler
  decoder.go       — MeshCore packet decoder (ported from decoder.js)
  decoder_test.go  — decoder tests (25 tests, golden fixtures)
  db.go            — SQLite writer (schema-compatible with db.js)
  db_test.go       — DB tests (schema validation, insert/upsert, E2E)
  config.go        — config struct + loader
  util.go          — shared utilities
  go.mod / go.sum  — Go module definition