Files
meshcore-analyzer/cmd/ingestor
Kpa-clawbot 13bdee57d4 perf: P0 hot-path fixes (observers, neighbor-graph, observer-analytics) (#1481) (#1483)
## What

Three of the four P0s from #1481's scale-test findings. Each cuts a
distinct
hot path; together they target /api/observers,
/api/analytics/neighbor-graph,
and /api/observers/{id}/analytics — the top three live offenders.

### P0-1: 5-min atomic-pointer cache for default neighbor-graph response
- Live p95 10.8s on the most-trafficked organic endpoint.
- Background recomputer (5-min cadence per operator directive) builds
the
  default-filter (`minCount=5 minScore=0.1`, no region, no role)
  `NeighborGraphResponse` and stores it via `atomic.Pointer`.
- `handleNeighborGraph` short-circuits on the default shape; non-default
filters take the extracted `computeNeighborGraphResponse` path
(identical
  semantics to the previous inline build).

### P0-2: cache parsed `StoreObs.Timestamp` + drop RLock window
- `handleObserverAnalytics` re-parsed the RFC3339 timestamp three times
  per observation, for 60k+ observations per active observer, under
  `s.store.mu.RLock` — blocking writers for the full scan.
- `StoreObs.ParsedTime()` parses once via `sync.Once` (mirrors
  `StoreTx.ParsedDecoded`).
- Handler snapshots the `byObserver[id]` pointer slice, releases the
  RLock immediately, then iterates locally.

### P0-3: 30s cache for `/api/observers` + sargable `IN` + covering
index
- Three SQL queries on every request → ~1.7s p50 at 50-concurrent.
- Atomic-pointer 30s cache for the default (no-filter) query.
- `GetNodeLocationsByKeys` drops `LOWER(public_key) IN (...)`
(non-sargable);
  callers pre-lowercase in Go and the plain `IN` matches the existing
  `public_key` index.
- New ingestor migration `obs_observer_ts_idx_v1` adds composite index
  `idx_observations_observer_idx_timestamp(observer_idx, timestamp)` so
  `GetObserverPacketCounts` can resolve its GROUP-BY + range filter from
  the index without scanning the 1.9M-row observations table.

### P0-4: deferred
`perfMiddleware`'s global mutex was claimed to serialize every API
request.
A direct test (`50 concurrent requests through the middleware, handler
sleeps 20ms each`) shows total elapsed ≈ 25ms, not 1s — the lock is held
only for the post-handler bookkeeping (a few µs). Real impact is below
measurement noise. Skipping to avoid invasive churn on PerfStats
consumers
without a demonstrable win.

## Test plan

Red → green per P0:
- `observers_cache_test.go` — handler reads `s.observersCache` before
SQL,
  TTL boundary, atomic.Pointer (no mutex contention).
- `storeobs_parsedtime_test.go` — parses three timestamp shapes, caches
  result, no race under concurrent readers.
- `neighbor_graph_cache_test.go` — handler serves from atomic pointer
  when set, bypasses cache when `?region=` (or any non-default filter)
  is passed.

Full server + ingestor suites pass: `go test -count=1 ./...`.

## Perf proof

Before/after p50/p95/p99 (50 requests × 50 concurrent) against prod
(before)
and staging once CI deploys (after) will be posted as a PR comment per
the
operator's "no merge without proof of improvement" gate.

Closes #1481


## TDD exemption — P0-1 and P0-2 (net-new surfaces, AGENTS.md)

Per CoreScope `AGENTS.md` § "Exemptions": **net-new code surfaces with
no
prior tests to break** may land tests in the same PR without a strict
test-first → impl commit split.

- **P0-1 (neighbor-graph atomic-pointer cache)** — `neighborGraphCache`,
  `recomputeNeighborGraphCache`, `loadNeighborGraphCacheBytes`,
  `startNeighborGraphRecomputer` and the default-shape short-circuit in
  `handleNeighborGraph` were brand-new code with no pre-existing
  assertions covering them. There was no green test to first turn red.
- **P0-2 (cached `StoreObs.Timestamp` + RLock window drop)** —
  `StoreObs.ParsedTime()` and the snapshot+release pattern in
  `handleObserverAnalytics` were new surfaces; the prior code did the
  parse inline per call with no behavioural test to break.

P0-3 was authored properly red-then-green (commit `6e63ec6a` red, then
`83ae129b` green) and does NOT use this exemption.

## Default-filter detection vs frontend reality (#1483 follow-up)

The Neighbor Graph analytics tab in `public/analytics.js` fetches
`/analytics/neighbor-graph?min_count=1&min_score=0` because the
client-side sliders need the full edge set to filter from. That shape
did NOT match the `(5, 0.1)` cached default, so the UI tab still paid
the cold compute cost despite #1481 P0-1.

The #1483 follow-up commit caches BOTH shapes in the same recomputer
pass:
- `(minCount=5, minScore=0.1, no region, no role)` — `live.js`
  affinity-scoring consumer.
- `(minCount=1, minScore=0, no region, no role)` — analytics tab.

Both are served from `atomic.Pointer` with an `X-Cache-Age-Seconds`
header. The per-shape cost in the background goroutine is roughly
linear in edge count; total recompute time stays well under the
5-minute cadence on prod-scale graphs.

---------

Co-authored-by: openclaw-bot <bot@openclaw.dev>
Co-authored-by: mc-bot <mc-bot@users.noreply.github.com>
2026-05-29 02:42:21 -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