## Summary Reverts the part of PR #1233 (commit `498fbc03`) that routed the MQTT envelope's `timestamp` field into `PacketData.Timestamp` for `transmissions.first_seen` and `observations.timestamp`. Packet ordering is restored to server ingest time — the client clock is untrusted. `UpsertObserverAt` + `MAX(MIN(existing, ingestNow), rxTime)` for observer/node `last_seen` (PR #1233's other half) is preserved unchanged. `parseEnvelopeTime` / `resolveRxTime` helpers are preserved — they still feed the observer.last_seen path. ## Diagnosis — Voodoo3 tx 304114 on staging Staging `tx_id = 304114` in channel `#test` has 5 observations: | # | observer | reported timestamp | comment | |---|-----------|--------------------|---------| | 1 | Voodoo3 | 18:42 | broken client RTC — ingested first, locks `first_seen` | | 2 | Voodoo3 | 18:42 | broken client RTC | | 3 | Voodoo3 | 18:42 | broken client RTC | | 4 | Voodoo3 | 18:42 | broken client RTC | | 5 | other obs | 01:42 | genuine receive time | 4 of 5 observations carry stale 18:42 timestamps from Voodoo3's own broken clock. Because Voodoo3 ingested first, PR #1233's code wrote `transmissions.first_seen = 18:42` (envelope value). Downstream aggregators that compute `MAX(first_seen)` per channel saw 18:42 as the latest activity, and `/api/channels` for `#test` displayed `lastActivity` ~7h+ in the past plus a stale heartbeat in the row preview — hiding the genuinely-newest message (Voodoo3's `tst hmdpt` at 01:42). ## Why PR #1233's premise fails PR #1233 assumed: > Uploaders stamp `timestamp` when the radio receives the frame and > freeze it; the MQTT message is published late, but the timestamp > field is not re-stamped at publish. A buffered packet uploaded > hours late still carries its true receive time. That holds ONLY when the uploader's wall clock is correct. Observers in the field (Voodoo3 here, surely others) have broken local clocks. Their envelope timestamps are not a true receive time — they're a broken-clock receive time, which is just garbage with extra steps. The server clock is the only one we control, so packet ordering must use it. ## Fix ### `cmd/ingestor/db.go` - `BuildPacketData`: `PacketData.Timestamp = time.Now().UTC().Format(time.RFC3339)`, NOT `msg.Timestamp`. Docstring updated to cite #1370 and explain why `msg.Timestamp` is no longer read here. ### `cmd/ingestor/main.go` - Channel-companion path: `Timestamp: ingestNow` (was `rxTime`). - DM-companion path: `Timestamp: ingestNow` (was `rxTime`). - Local `rxTime := resolveRxTime(msg, tag)` removed from both paths (no remaining consumers in those scopes). ### Preserved (NOT touched) - `resolveRxTime`, `parseEnvelopeTime` — still used by `handleMessage` to populate `mqttMsg.Timestamp` and to call `UpsertObserverAt`, which feeds `observer.last_seen` and `observer.last_packet_at`. - All three `MAX(MIN(existing, ingestNow), rxTime)` guards (#1233 observer.last_seen, observer.last_packet_at, node.last_seen). - `MQTTPacketMessage.Timestamp` struct field. ## Tests | File | Asserts | |------|---------| | `cmd/ingestor/ingest_time_regression_1370_test.go` (3 cases) | Raw-packet, channel-companion, and DM-companion `handleMessage` paths. Feed envelope `timestamp = T_now - 7h`; assert stored `transmissions.first_seen` (RFC3339) and `observations.timestamp` (epoch) are server wall clock (±5s). Each case fails on master under PR #1233's premise. | ### Adjusted test - `cmd/ingestor/db_test.go::TestBuildPacketData` — PR #1233 had asserted `pkt.Timestamp == "2026-05-16T10:00:00Z"` (the envelope value propagating). Now asserts the opposite: `pkt.Timestamp` is non-empty AND is NOT the envelope value. Comment cites #1370 and why the expectation flipped. ### Verified still-green - `cmd/ingestor/rxtime_test.go` (`TestParseEnvelopeTime`, `TestResolveRxTime`) — helpers untouched, still cover envelope parsing for the observer.last_seen path. - `cmd/server/channels_message_order_1366_test.go` (#1366). - `cmd/server/db_channel_messages_perf_test.go` (#1368 perf budget). ## Commits - `a9b7efc3` — RED: 3 `handleMessage` assertion-fail tests + test name collision check. - `5a0891f0` — GREEN: revert envelope→PacketData.Timestamp plumbing in `cmd/ingestor/{db,main}.go` + flip `TestBuildPacketData`. Fixes #1370 --------- Co-authored-by: corescope-bot <bot@corescope.dev>
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 connectionsname— display name for loggingbroker— MQTT URL (mqtt://,mqtts://)username/password— auth credentialstopics— array of topic patterns to subscribeiataFilter— optional regional filter
mqtt— legacy single-broker config (auto-converted tomqttSources)dbPath— SQLite DB path (default:data/meshcore.db)
Test
cd cmd/ingestor
go test -v ./...
What It Does
- Connects to configured MQTT brokers with auto-reconnect
- Subscribes to mesh packet topics (e.g.,
meshcore/+/+/packets) - Receives raw hex packets via JSON messages (
{ "raw": "...", "SNR": ..., "RSSI": ... }) - Decodes MeshCore packet headers, paths, and payloads (ported from
decoder.js) - Computes content hashes (path-independent, SHA-256-based)
- Writes to SQLite:
transmissions+observationstables - Upserts
nodesfrom decoded ADVERT packets (with validation) - Upserts
observersfrom MQTT topic metadata
Schema Compatibility
The Go ingestor creates the same v3 schema as the Node.js server:
transmissions— deduplicated by content hashobservations— per-observer sightings withobserver_idx(rowid reference)nodes— mesh nodes discovered from advertsobservers— 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