feat: add unified relay service - #1
Conversation
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 76a78fece7
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if result.Error != nil { | ||
| utils.SugarLogger.Errorf("[DB] Failed to batch insert %d messages: %v", len(batch), result.Error) |
There was a problem hiding this comment.
When SQLite returns a transient error—such as database is locked while the documented shelter process accesses the shared file—this branch only logs it; the worker immediately resets the batch after writeBatch returns, permanently discarding every frame in it. Requeue or retry failed batches so a temporary database problem does not corrupt telemetry capture.
Useful? React with 👍 / 👎.
| if config.DatabasePath == "" { | ||
| config.DatabasePath = "relay.db" | ||
| } |
There was a problem hiding this comment.
When the production image is run without DATABASE_PATH, this relative fallback resolves to /relay.db because the final Docker stage has no WORKDIR; the image's declared /data volume is therefore unused. Recreating or upgrading the container then deletes all buffered telemetry despite mounting /data, so the container default should resolve inside that volume.
Useful? React with 👍 / 👎.
| utils.SugarLogger.Warnf("[MQ][cloud] Cannot subscribe to %s: CLOUD_MQTT_HOST not configured", topic) | ||
| return | ||
| } | ||
| subscribedTopics[topic] = handler |
There was a problem hiding this comment.
When the cloud connection completes while the SubscribePong startup goroutine is registering its handler, this map write can overlap onConnectFn iterating the same map. Native Go maps do not permit concurrent iteration and writes, so an ordinary fast broker connection can race or terminate the relay with concurrent map iteration and map write; protect the registry with a mutex or finish registration before connecting.
Useful? React with 👍 / 👎.
| warnAfter := config.PingInterval * 2 | ||
| for { | ||
| lastPing := FindLastSuccessfulPing() | ||
| ageMs := time.Now().UnixMilli() - int64(lastPing.Ping) |
There was a problem hiding this comment.
After the first successful pong, lastPing.Ping contains the UnixMicro value written by PublishPing, but this subtracts it from UnixMilli. The resulting age is a large negative number, so the warning condition never becomes true during subsequent Mapache outages; use matching timestamp units before comparing with warnAfter.
Useful? React with 👍 / 👎.
PublishPing stores the ping as UnixMicro, but the staleness watchdog subtracted it from UnixMilli. The resulting age was a large negative number, so the warning never fired during a Mapache outage — the exact condition it exists to report. Rate-limit the warning to once a minute now that it can actually fire. The check itself stays on its old cadence so TCM Status reacts quickly, but a road car sits unreachable for days at a time and logging every poll would put tens of thousands of lines a day onto the SD card.
writeBatch only logged the error and the worker reset the batch immediately after, so a single transient failure discarded every frame in it — up to DB_BATCH_SIZE of them. "database is locked" is expected in normal operation once shelter holds the same file, so this would put holes in the telemetry capture rather than being a rare edge case. Retry with a short backoff before giving up. The insert is a single transaction, so a failed attempt wrote nothing and a retry cannot duplicate rows. Also close the queue under the write lock. QueueDBWrite checked the stopped flag and then sent on the channel, so a sender that passed the check before StopDBQueue closed it would panic on a closed channel.
The final image stage has no WORKDIR, so the built-in relative DATABASE_PATH default resolved to /relay.db and the declared /data volume went unused. Recreating or upgrading the container then discarded all buffered telemetry despite /data being mounted.
Every CAN frame spawned a goroutine to publish itself, which puts the whole bus through the scheduler on a single-board target. Publishing inline instead is only safe if publishing cannot block: paho's Publish waits on an internal channel bounded by MessageChannelDepth and gives up only after WriteTimeout (30s by default), so a broker that is connected but slow — a degrading cell uplink rather than a dropped one — would stall the socketcan reader. That stalls unix.Read, overruns the kernel CAN buffer, and drops frames before the relay ever sees them. Give each broker its own queue drained by its own worker, so a slow cloud uplink cannot block a local consumer, and shed with a periodic count rather than a line per frame. Skip the throttle bookkeeping and the payload allocation entirely when no broker wants the frame, which is the common case under the publish intervals. Guard the subscription registry with a mutex. Subscribe registers from the SubscribePong goroutine while onConnect ranges over the same map, so a fast broker connection could terminate the relay with "concurrent map iteration and map write". Handle SIGINT/SIGTERM so the write queue flushes and the database closes on the way out. Previously docker stop discarded whatever was buffered, up to a full batch of frames.
The 0x201 layout was inherited from TCM-26's Jetson and carried fields the Pi cannot produce: two extra CPU cores, GPU utilization, frequency and temperature, and three power-rail readings. Sharing the Mapache decoder was the stated reason to keep them, but no p987 decoder exists yet, so the compatibility is with nothing. Cut the layout to what the board actually reports — 4 cores, no GPU, no power rails — and spend one of the freed bytes on the failure mode this hardware really has. Under-voltage on a Pi corrupts SD cards and is invisible in every other metric. The firmware get_throttled word is the only place the live throttle bits are exposed, since the Pi throttles in firmware rather than through the kernel thermal governor. That attribute is deprecated upstream, so fall back to the rpi_volt hwmon alarm, which carries the sticky under-voltage bit only (BIT(16)) and nothing thermal. Saturate the encoded fields rather than truncating: a bogus reading should pin a field, not alias to a plausible small number on a dashboard. BREAKING CHANGE: the 0x201 payload is now 29 bytes with a new field order. Any decoder written against the 44-byte TCM-26 layout must be updated.
paho's loggers default to NOOPLogger, and the cloud client sets ConnectRetry, so a wrong host, port, or credential retried forever in complete silence — the only symptom was pongs that never arrived. Route paho's errors into zap, dropping the stacktrace (always the same paho internals) and sampling to one line per message per minute, since an unreachable broker retries every 5s indefinitely. Fail fast when a broker host is set without a port, which previously produced a "tcp://host:" URL, and log both endpoints at startup so the configured target is visible rather than inferred. Log the database path resolved rather than raw. The default is relative and lands wherever the workdir points, which is what made the volume bug above hard to see. Silence GORM's "record not found", which is a normal result here: the ping watchdog polls for a successful pong every couple of seconds and finds none until the car first reaches Mapache. At the default level that wrote a colorized SQL dump to the SD card on every poll, forever, while the car was offline. Drop the ANSI color outside DEV too.
Vehicle identity travels in the topic, not the payload, and Mapache's ingest splits on "/" and requires exactly four segments, reading the vehicle from segment 1. A VEHICLE_ID or bus label containing a slash shifts every field and the message is dropped on arrival — silently, because we publish at QoS 0 and never learn it was rejected. The car would buffer to SQLite and upload nothing. MQTT also forbids wildcards in a topic being published to. Refuse to start rather than sanitizing: a silently renamed vehicle scatters data under the wrong id, which is worse than not booting.
The encoders and parsers are pure functions over bytes and are decoded downstream by fixed offset, so they are the parts worth pinning. Covers the shared message header, the 0x200 and 0x201 payloads, the virtual CAN and socketcan frame parsers, the publish throttle, interface and port parsing, and the topic-shape guard. Includes cases for the two bugs that motivated the parsers being extracted: that parsed frame data is copied out of a reused read buffer, and that a CAN FD DLC on a classic socket is clamped to 8 bytes.
Gate the image build on the tests so a tree that fails its own suite cannot be pushed to GHCR.
Matches how the rest of the services are set up.
1 participant
relay/Go service: reads socketcan interfaces directly (raw AF_CAN, one reader per interface with reopen backoff) and publishes to local + cloud MQTT brokersp987/{vehicle_id}/{bus}/0x{can_id}with bus labels fromCAN_INTERFACES(can0:pcan); virtual/housekeeping frames publish undertcmp987_messagewith partial unsynced index)LOCAL_MQTT_HOSTdisables, matching cloud behavior)RETENTION_HOURSpurge for synced messages + old pings (Pi Zero 2 W RAM/SD constraints)example.env, multi-arch GHCR workflow (tcm-987/relay),scripts/release.sh,scripts/setup-can.sh(listen-only bring-up, vcan support)BREAKING — resource metrics 0x201 is now a 29-byte Pi-native layout, not TCM-26's 44 bytes:
get_throttledword with therpi_volthwmon alarm as fallbackp987today, so nothing breaks yetReview fixes:
WORKDIR /dataso the defaultDATABASE_PATHresolves inside the declared volumeAlso in this branch:
docker stoppreviously dropped up to a full batchNOOPLogger, so a misconfigured cloud broker retried forever in silence); validate and log both broker endpointsrecord not found, which wrote a colorized SQL dump to the SD card every 2.3s while the car was offlineVEHICLE_ID/ bus labels containing/,+,#— they break the 4-segment topic shape the ingest requires and the message is dropped server-side with no signal backVerified by running the arm64 image: 5.8 MiB RSS steady state, 41.4 MB image, database lands at
/data/relay.db, and a SIGTERM 3ms after injecting frames persists all of them.