commit 1f678fe87fe2197f9abdaaebbe553423a9cc8784 Author: Mikkeli Date: Fri Oct 9 20:15:13 2026 +0900 Add recorder: raw CAN and GNSS sessions with failsafes and a storage bound Reads SocketCAN and gpsd directly. Sessions start on the first frame or on 'ctl start', end after 15 s of bus idle or on 'ctl stop'; blips under 3 s are discarded. candump -L chunks fdatasync'd every second and zstd'd once closed; oldest sessions deleted to stay under 50 GB / above 20 GB free. Kernel and queue drops, link changes and clock steps are recorded as events. Bench-tested on vcan with the first car capture (byte-identical) and installed on the car Orin as yukart-recorder.service. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz diff --git a/.cargo/config.toml b/.cargo/config.toml new file mode 100644 index 0000000..9673d0b --- /dev/null +++ b/.cargo/config.toml @@ -0,0 +1,4 @@ +# The services run on the Jetson (aarch64). Build for it with: +# cargo build --release --target aarch64-unknown-linux-gnu +[target.aarch64-unknown-linux-gnu] +linker = "aarch64-linux-gnu-gcc" diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..27c7aab --- /dev/null +++ b/.editorconfig @@ -0,0 +1,17 @@ +root = true + +[*] +charset = utf-8 +end_of_line = lf +insert_final_newline = true +trim_trailing_whitespace = true +indent_style = space + +[*.{rs,toml}] +indent_size = 4 + +[*.{py,dart,json,yaml,yml}] +indent_size = 2 + +[*.md] +trim_trailing_whitespace = false diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..734cc5a --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +/target +*.pdb + +# Cargo.lock IS committed: these crates build binaries, not libraries. diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..264cb7c --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,250 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "cc" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f74872d07caf508b30a21f6836e7d7016a2eaf7d9ff4f48deaa58cd8a0407630" +dependencies = [ + "find-msvc-tools", + "jobserver", + "libc", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" + +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "find-msvc-tools" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aedcfb3409746eddb02b9e19ebda1c3394f759a152e48ee875a0844d1b955484" + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom", + "libc", +] + +[[package]] +name = "libc" +version = "0.2.190" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "pkg-config" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "svc-core" +version = "0.1.0" +dependencies = [ + "libc", + "serde_json", + "signal-hook", + "zstd", +] + +[[package]] +name = "syn" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "unicode-ident" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" + +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64d80649ab6db9d9f6f9c80a40becd948eda4714a0a5ac8c4d157a32231c7882" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.1.0+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ef0a8027ec3ee71300ab3bcbcd0393f434aa72b91ca6d635a39941deae8eea0" +dependencies = [ + "cc", + "pkg-config", +] diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..5da3470 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "svc-core" +version = "0.1.0" +edition = "2021" +description = "Acquisition and processing cluster: recorder now; can-bridge, obd-poller, gnss, decoder later" +publish = false + +[[bin]] +name = "recorder" +path = "src/recorder/main.rs" + +[dependencies] +libc = "0.2" +serde_json = "1" +signal-hook = "0.3" +zstd = "0.13" + +[profile.release] +strip = true diff --git a/README.md b/README.md new file mode 100644 index 0000000..968c601 --- /dev/null +++ b/README.md @@ -0,0 +1,32 @@ +--- +type: reference +--- +# Core services + +Part of **Project Yukart**, making an old 2008 Mitsubishi Z27AG a bit smarter with modern technologies. +This file serves as the minimalist portal to other docs + + +## What this repo is about + +The acquisition and processing cluster on the Orin, developed and deployed as one +unit: `recorder` now; `can-bridge`, `obd-poller`, `gnss` and `decoder` as they are +needed. + +| Binary | Installed as | Does | +|---|---|---| +| `recorder` | `yukart-recorder.service` | Raw CAN + GNSS to disk, in sessions, under a storage bound | + +```bash +scripts/deploy-recorder.sh # build, glibc check, install on the car Orin +yukart-recorder ctl status # on the Orin +yukart-recorder ctl start "cold start" # manual session, ends only on stop +yukart-recorder ctl mark "indicator left on" +yukart-recorder ctl stop +``` + +Recordings live in `/var/lib/yukart/recordings/`, one directory per session — +layout in [the initial plan](docs/impl/plans/INITIAL.md#session-layout). + +- For implementation practice guidelines, refer to the [guidelines](docs/GUIDELINES.md); +- For implementation plans and history, refer to [them](docs/impl/); diff --git a/deploy/yukart-recorder.service b/deploy/yukart-recorder.service new file mode 100644 index 0000000..6ec7389 --- /dev/null +++ b/deploy/yukart-recorder.service @@ -0,0 +1,24 @@ +[Unit] +Description=Yukart recorder: raw CAN and GNSS sessions to disk +Documentation=gitea:yukart/svc-core +# Tier 1, ordered after nothing: CAN adapter and gpsd come and go, and the +# recorder waits for them itself rather than through unit dependencies. +StartLimitIntervalSec=0 + +[Service] +User=mikkeli +ExecStart=/usr/local/bin/yukart-recorder run +Restart=always +RestartSec=2 +# /var/lib/yukart/recordings and /run/yukart-recorder, created and chowned by systemd. +StateDirectory=yukart/recordings +RuntimeDirectory=yukart-recorder +# Shutdown closes the session and fsyncs; seconds at most, but never cut it short. +TimeoutStopSec=20 +KillSignal=SIGTERM +Nice=-5 +IOSchedulingClass=best-effort +IOSchedulingPriority=0 + +[Install] +WantedBy=multi-user.target diff --git a/docs/GUIDELINES.md b/docs/GUIDELINES.md new file mode 100644 index 0000000..2dc802b --- /dev/null +++ b/docs/GUIDELINES.md @@ -0,0 +1,33 @@ +--- +type: guidelines +--- +# Core services guidelines + +## 1. Listen-only until `obd-poller` exists +- Nothing here opens a CAN socket for writing. The recorder only reads, and the + host brings `can_pcan` up listen-only. +- `obd-poller` will be the single writer. Until it lands, any transmit is a bug. + +## 2. Capture never waits +- A frame missed is gone. Producers hand frames to the writer without blocking; + what doesn't fit is **counted**, recorded as an event, and never silently lost. +- Capture reads SocketCAN and gpsd directly. It does not depend on the broker, a + display, or any other service being up. + +## 3. Every failure leaves complete files +- Append, flush and `fdatasync` about once a second. Power loss costs at most the + last second. +- Replace files by write-temp, fsync, rename. Never rewrite in place. +- Anything found half-done at startup is repaired or marked, not deleted. + +## 4. Raw stays raw +- Recordings are `candump -L` text plus the gpsd JSON as received, so `can-utils`, + `canplayer` and anything already written against the first captures keep working. +- Decoding happens downstream of the recording, never instead of it. + +## 5. Time is evidence, not truth +- The car unit's clock can boot stale and be stepped mid-drive. Record what is + needed to correct it later (sync state, monotonic time, observed steps, GNSS + time beside local receive time) rather than trusting it. + +## 6. To be added diff --git a/docs/impl/plans/INITIAL.md b/docs/impl/plans/INITIAL.md new file mode 100644 index 0000000..379280b --- /dev/null +++ b/docs/impl/plans/INITIAL.md @@ -0,0 +1,108 @@ +--- +type: plan +status: active +updated: 2026-10-09 +--- +# Initial plan + +## Order + +1. **`recorder`** — raw CAN and GNSS to disk. Built first because every later + piece is developed against what it records. +2. `obd-poller` — standard OBD-II PIDs at a low rate, the cross-check for + reverse-engineered signals. The first transmitter: `can_pcan` leaves + listen-only for it, a deliberate step. +3. `can-bridge` and `gnss` — publish onto the broker once it exists (project-yukart + Phase 3; MQTT at ~890 frames/s is still unproven). +4. `decoder` — frames to named signals via the DBC in `contracts`. + +## Recorder + +Reads SocketCAN and gpsd directly. It is tier 1, and the broker is a single point +of failure that raw capture should not share. When `can-bridge` exists, the +recorder still owns raw capture; it subscribes to the bus only for decoded +signals. + +### Session triggers + +| Trigger | Starts on | Ends on | +|---|---|---| +| auto | the first CAN frame | `--idle` (15 s) with no frames | +| manual | `ctl start [label]` | `ctl stop`, or shutdown | + +- **No traffic, no session.** The PEAK attached at a desk, attached to a parked + car with the bus asleep, or unplugged, produces nothing — the socket is retried + every second in silence. +- **Cranking stays inside one session.** A bus dropout or adapter re-enumeration + shorter than the idle timeout is recorded as `can_link` events, not as a split. +- **Blips are discarded.** An auto session with under `--min-secs` (3 s) of + traffic and no marks is deleted at close — door unlocks and brief plug-ins. +- `ctl start` during an auto session promotes it to manual. After `ctl stop`, auto + start waits for the bus to go idle, so a stop sticks. +- `ctl auto off` (or `--no-auto`) leaves only manual sessions. + +The power service planned in orin-nano-env-setup (`docs/impl/plans/POWER.md`) +is the intended second client: `ctl mark crank` for a recovered input loss, and +`systemctl stop yukart-recorder` before poweroff. + +### Session layout + +``` +/var/lib/yukart/recordings/ + .last-seq high-water mark; numbers are never reused + 000042-20261009T163736/ sequence number orders; the local time is for humans + session.json status open → closed | truncated; trigger, label, host, + boot_id, start/end realtime + monotonic + clock sync, + frame and drop counts, end reason + can-0001.log.zst candump -L, closed chunks (5 min or 64 MB), zstd + can-0002.log the open chunk, plain + gnss.ndjson {"rx": , "msg": } + events.ndjson marks, can_link, gnss_link, clock_step, drops, storage_bound +``` + +`zstdcat can-*.log.zst can-*.log | canplayer` replays a session. + +### Failsafes + +| Failure | Behaviour | +|---|---| +| Power loss, crash, SIGKILL | open chunk is fdatasync'd every second; at the next start the session is marked `truncated` and its chunks compressed | +| Crash mid-compression | compression writes `.zst.tmp` then renames; the plain chunk is removed only after | +| Adapter unplugged, link down | reopen every second; recorded as `can_link` events | +| gpsd down | reconnect every 3 s; CAN capture unaffected | +| Writer slower than the bus | 100 000-frame queue (~110 s); overflow counted as `queue_drops` | +| Kernel socket overflow | `SO_RXQ_OVFL` counted as `kernel_drops` | +| Disk filling | oldest closed sessions deleted to stay under `--cap-gb` (50) and above `--min-free-gb` (20); the open session is never deleted | +| Control socket fails | logged; auto capture continues | + +Measured on the first capture: ~33 KB/s of candump text, ~5.5 KB/s after zstd — +**about 20 MB per hour of driving**. The 50 GB default holds years of it; the cap +is there for the failure where something else fills the disk. + +### Time + +The car unit has no RTC battery yet and can start with a stale clock that chrony +steps later. CAN frames carry the kernel's receive timestamp (`SO_TIMESTAMP`, +CLOCK_REALTIME), and GNSS reports carry their local receive time — **the same +clock**, so CAN and GNSS line up within a session however wrong that clock is. + +To get absolute time back afterwards: + +- `session.json` records the kernel's sync state at start and end. +- `clock_step` events record every step over 50 ms with its size. +- Each TPV holds GNSS UTC beside its local receive time; their difference is the + clock error, plus the SIM7600's NMEA latency (no PPS: expect tens to hundreds + of milliseconds, constant enough to calibrate out). + +At 1 Hz GNSS against ~890 frames/s, alignment to a few hundred milliseconds is +what this gives. It is enough to match CAN speed against GNSS speed; it is not +enough to time individual frames against position. + +## Open + +- **Session marking from the app.** `app-flutter`'s mark button reaches the + recorder through `svc-gw`, which will forward to the control socket. +- **Native builds.** `scripts/deploy-recorder.sh` cross-compiles from the + workstation while the development Orin is offline; `check-glibc.sh` gates it. +- **Copy-off.** Recordings stay on the car until the parked-at-home step in the + power plan, or a manual `rsync`. diff --git a/docs/impl/reports/RECORDER-BENCH.md b/docs/impl/reports/RECORDER-BENCH.md new file mode 100644 index 0000000..f21afcc --- /dev/null +++ b/docs/impl/reports/RECORDER-BENCH.md @@ -0,0 +1,37 @@ +--- +type: report +date: 2026-10-09 +--- +# Recorder bench test + +Replaying the first in-car capture (`z27ag-20261009-163736-ign.log`, 10 s, +8925 frames) onto a `vcan0` with its original timing, against a fake gpsd on the +workstation and the real gpsd on the car Orin. + +## Scenarios + +| # | Scenario | Result | +|---|---|---| +| 1 | Full replay, `mark` at t≈3 s | auto session; 8925 frames; ID#DATA sequence **byte-identical** to the source; 3 chunks at `--chunk-secs 4`, all compressed | +| 2 | 1 s replay | discarded at close (under `--min-secs`, no marks) — after fixing the span to first→last frame; the first version counted the idle wait too and kept it | +| 3 | `ctl start bench-test`, no CAN, `mark`, `stop` | manual session with label, mark and GNSS; no CAN chunk left behind | +| 4 | `vcan0` deleted mid-session, recreated 1 s later | one session; `Network is down` → `No such device` → `socket open` recorded as `can_link` events | +| 5 | SIGTERM mid-session | session closed, reason `shutdown`, exit 0 | +| 6 | SIGKILL mid-session (3 s replay), restart | all 2676 frames on disk; marked `truncated` and compressed at restart | +| 7 | `--cap-gb 0` | every older closed session deleted, the active one kept; sequence numbers continue from `.last-seq` | + +On the car Orin (aarch64 cross-build, highest GLIBC_2.34 against its 2.35): the +same full replay recorded 8925 frames and 55 KB of zstd; real gpsd reports (no +fix indoors) recorded with the last TPV before start stamped `before_start`. +Resident memory at idle: 5 MB. + +## Not tested + +- Real PCAN-USB hot-plug in the car (the `vcan` deletion stands in for it). +- Sustained throughput beyond 890 frames/s, and the queue-overflow path. +- A clock step during a session (the detection is in place; nothing stepped). + +## Installed + +`yukart-recorder.service` on `mikkeli-orin-nano`, enabled, running as `mikkeli`, +waiting for `can_pcan`. Recordings: `/var/lib/yukart/recordings/`. diff --git a/rust-toolchain.toml b/rust-toolchain.toml new file mode 100644 index 0000000..292fe49 --- /dev/null +++ b/rust-toolchain.toml @@ -0,0 +1,2 @@ +[toolchain] +channel = "stable" diff --git a/scripts/check-glibc.sh b/scripts/check-glibc.sh new file mode 100755 index 0000000..19b14cd --- /dev/null +++ b/scripts/check-glibc.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash +# Fail if a binary references a glibc symbol version newer than the car unit has. +# +# Detects drift between the development Orin and the car Orin. It does not fix +# it: the fix is building on a unit whose release matches the target's. +# +# usage: scripts/check-glibc.sh [ceiling] +set -euo pipefail + +CEILING="${GLIBC_CEILING:-2.35}" +BIN="${1:?usage: check-glibc.sh [ceiling]}" +[ $# -ge 2 ] && CEILING="$2" + +# A check that cannot check must not report success. +command -v objdump >/dev/null 2>&1 || { echo "BROKEN: objdump unavailable, nothing was checked"; exit 2; } +[ -f "$BIN" ] || { echo "BROKEN: no such binary: $BIN"; exit 2; } + +mapfile -t versions < <(objdump -T "$BIN" 2>/dev/null | grep -oE 'GLIBC_[0-9]+\.[0-9]+' | sort -Vu) +if [ "${#versions[@]}" -eq 0 ]; then + echo "BROKEN: no glibc symbol versions found in $BIN — statically linked, or objdump read nothing" + exit 2 +fi + +max="${versions[-1]#GLIBC_}" +echo "$(basename "$BIN"): ${#versions[@]} glibc versions referenced, highest GLIBC_$max, ceiling GLIBC_$CEILING" + +if [ "$(printf '%s\n%s\n' "$max" "$CEILING" | sort -V | tail -1)" != "$CEILING" ]; then + echo "FAIL: GLIBC_$max exceeds the ceiling. Symbols responsible:" + objdump -T "$BIN" 2>/dev/null | grep -F "GLIBC_$max" | awk '{print " " $(NF-1), $NF}' | sort -u + echo + echo "Build on a unit matching the car Orin's release, or raise the ceiling if" + echo "the car unit was upgraded too." + exit 1 +fi +echo "OK" diff --git a/scripts/deploy-recorder.sh b/scripts/deploy-recorder.sh new file mode 100755 index 0000000..7441ec6 --- /dev/null +++ b/scripts/deploy-recorder.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Build the recorder for aarch64, check it against the car's glibc, and install it +# on an Orin as the yukart-recorder service. +# +# usage: scripts/deploy-recorder.sh [host] (default: mikkeli-orin-nano) +# +# Cross-compiles from the workstation (the documented fallback): the development +# Orin is the preferred native build host. Either way check-glibc.sh gates it. +set -euo pipefail + +HOST="${1:-mikkeli-orin-nano}" +cd "$(dirname "$0")/.." +TARGET=aarch64-unknown-linux-gnu +BIN="target/$TARGET/release/recorder" + +cargo build --release --target "$TARGET" +scripts/check-glibc.sh "$BIN" + +# Binary lives under ~/infra per the host repo's guidelines; /usr/local/bin gets a symlink. +ssh "$HOST" 'mkdir -p ~/infra/svc-core' +scp -q "$BIN" "$HOST:infra/svc-core/recorder.new" +scp -q deploy/yukart-recorder.service "$HOST:infra/svc-core/yukart-recorder.service" +ssh "$HOST" 'set -e + mv ~/infra/svc-core/recorder.new ~/infra/svc-core/recorder + sudo ln -sfn "$HOME/infra/svc-core/recorder" /usr/local/bin/yukart-recorder + sudo install -m 0644 ~/infra/svc-core/yukart-recorder.service /etc/systemd/system/yukart-recorder.service + sudo systemctl daemon-reload + sudo systemctl enable yukart-recorder.service >/dev/null + sudo systemctl restart yukart-recorder.service + sleep 1 + systemctl --no-pager --lines=5 status yukart-recorder.service' diff --git a/src/recorder/can.rs b/src/recorder/can.rs new file mode 100644 index 0000000..3b7cea4 --- /dev/null +++ b/src/recorder/can.rs @@ -0,0 +1,177 @@ +//! Raw SocketCAN reception: kernel timestamps, the kernel's own drop counter, +//! and error frames. Read-only — the socket is never written to. + +use std::ffi::CString; +use std::fmt::Write as _; +use std::io; +use std::mem::{size_of, zeroed}; +use std::time::Duration; + +pub struct Frame { + /// CLOCK_REALTIME at reception in the driver, from SO_TIMESTAMP. + pub sec: i64, + pub usec: i64, + /// As delivered: carries the EFF / RTR / ERR flag bits. + pub id: u32, + pub len: u8, + pub data: [u8; 8], +} + +impl Frame { + /// One `candump -L` line, without the newline, so captures stay replayable + /// with `canplayer` and readable by anything already written against them. + pub fn write_candump(&self, iface: &str, out: &mut String) { + let _ = write!(out, "({}.{:06}) {} ", self.sec, self.usec, iface); + if self.id & libc::CAN_ERR_FLAG != 0 { + let _ = write!(out, "{:08X}", self.id & (libc::CAN_ERR_MASK | libc::CAN_ERR_FLAG)); + } else if self.id & libc::CAN_EFF_FLAG != 0 { + let _ = write!(out, "{:08X}", self.id & libc::CAN_EFF_MASK); + } else { + let _ = write!(out, "{:03X}", self.id & libc::CAN_SFF_MASK); + } + out.push('#'); + if self.id & libc::CAN_RTR_FLAG != 0 { + out.push('R'); + } else { + for b in &self.data[..self.len.min(8) as usize] { + let _ = write!(out, "{:02X}", b); + } + } + } +} + +pub enum Recv { + /// A frame, and how many frames the kernel dropped since the previous one. + Frame(Frame, u32), + Timeout, +} + +pub struct CanSocket { + fd: i32, + kernel_drops: u32, +} + +impl Drop for CanSocket { + fn drop(&mut self) { + unsafe { libc::close(self.fd) }; + } +} + +fn setopt(fd: i32, level: i32, name: i32, val: &T) -> io::Result<()> { + let r = unsafe { + libc::setsockopt(fd, level, name, val as *const T as *const libc::c_void, size_of::() as u32) + }; + if r < 0 { Err(io::Error::last_os_error()) } else { Ok(()) } +} + +impl CanSocket { + /// Fails with NotFound-like errors while the adapter is unplugged; callers retry. + pub fn open(iface: &str, timeout: Duration) -> io::Result { + let name = CString::new(iface).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + let idx = unsafe { libc::if_nametoindex(name.as_ptr()) }; + if idx == 0 { + return Err(io::Error::last_os_error()); + } + let fd = unsafe { libc::socket(libc::PF_CAN, libc::SOCK_RAW | libc::SOCK_CLOEXEC, libc::CAN_RAW) }; + if fd < 0 { + return Err(io::Error::last_os_error()); + } + let sock = CanSocket { fd, kernel_drops: 0 }; + + // Error frames are evidence (bus-off, wrong bitrate, a dying adapter), so record them all. + setopt(fd, libc::SOL_CAN_RAW, libc::CAN_RAW_ERR_FILTER, &libc::CAN_ERR_MASK)?; + setopt(fd, libc::SOL_SOCKET, libc::SO_TIMESTAMP, &1i32)?; + setopt(fd, libc::SOL_SOCKET, libc::SO_RXQ_OVFL, &1i32)?; + // Clamped to net.core.rmem_max; best effort. + let _ = setopt(fd, libc::SOL_SOCKET, libc::SO_RCVBUF, &(4i32 << 20)); + let tv = libc::timeval { tv_sec: timeout.as_secs() as _, tv_usec: timeout.subsec_micros() as _ }; + setopt(fd, libc::SOL_SOCKET, libc::SO_RCVTIMEO, &tv)?; + + let mut addr: libc::sockaddr_can = unsafe { zeroed() }; + addr.can_family = libc::AF_CAN as u16; + addr.can_ifindex = idx as i32; + let r = unsafe { + libc::bind(fd, &addr as *const _ as *const libc::sockaddr, size_of::() as u32) + }; + if r < 0 { + return Err(io::Error::last_os_error()); + } + Ok(sock) + } + + pub fn recv(&mut self) -> io::Result { + let mut frame: libc::can_frame = unsafe { zeroed() }; + let mut iov = libc::iovec { + iov_base: &mut frame as *mut _ as *mut libc::c_void, + iov_len: size_of::(), + }; + let mut cbuf = [0u64; 16]; + let mut msg: libc::msghdr = unsafe { zeroed() }; + msg.msg_iov = &mut iov; + msg.msg_iovlen = 1; + msg.msg_control = cbuf.as_mut_ptr() as *mut libc::c_void; + msg.msg_controllen = size_of::<[u64; 16]>() as _; + + let n = unsafe { libc::recvmsg(self.fd, &mut msg, 0) }; + if n < 0 { + let e = io::Error::last_os_error(); + return match e.kind() { + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => Ok(Recv::Timeout), + io::ErrorKind::Interrupted => Ok(Recv::Timeout), + _ => Err(e), + }; + } + + let (mut sec, mut usec) = (None, 0); + let mut drops = self.kernel_drops; + unsafe { + let mut c = libc::CMSG_FIRSTHDR(&msg); + while !c.is_null() { + let (level, ty) = ((*c).cmsg_level, (*c).cmsg_type); + if level == libc::SOL_SOCKET && ty == libc::SCM_TIMESTAMP { + let tv = std::ptr::read_unaligned(libc::CMSG_DATA(c) as *const libc::timeval); + sec = Some(tv.tv_sec as i64); + usec = tv.tv_usec as i64; + } else if level == libc::SOL_SOCKET && ty == libc::SO_RXQ_OVFL { + drops = std::ptr::read_unaligned(libc::CMSG_DATA(c) as *const u32); + } + c = libc::CMSG_NXTHDR(&msg, c); + } + } + let sec = sec.unwrap_or_else(|| { + let t = crate::clock::realtime(); + usec = ((t.fract()) * 1e6) as i64; + t.trunc() as i64 + }); + let new_drops = drops.wrapping_sub(self.kernel_drops); + self.kernel_drops = drops; + + Ok(Recv::Frame( + Frame { sec, usec, id: frame.can_id, len: frame.can_dlc, data: frame.data }, + new_drops, + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn line(id: u32, data: &[u8]) -> String { + let mut d = [0u8; 8]; + d[..data.len()].copy_from_slice(data); + let mut s = String::new(); + Frame { sec: 1791531456, usec: 435801, id, len: data.len() as u8, data: d }.write_candump("can_pcan", &mut s); + s + } + + #[test] + fn candump_format_matches_existing_captures() { + assert_eq!(line(0x449, &[0x6F, 0x5F, 0x6E, 0x71, 0xFF, 0xFD, 0x33, 0xFF]), + "(1791531456.435801) can_pcan 449#6F5F6E71FFFD33FF"); + assert_eq!(line(0x0C0, &[0x6D, 0xFF, 0, 7, 0x0C]), "(1791531456.435801) can_pcan 0C0#6DFF00070C"); + assert_eq!(line(0x18DAF110 | libc::CAN_EFF_FLAG, &[1]), "(1791531456.435801) can_pcan 18DAF110#01"); + assert_eq!(line(0x7DF | libc::CAN_RTR_FLAG, &[]), "(1791531456.435801) can_pcan 7DF#R"); + assert_eq!(line(0x04 | libc::CAN_ERR_FLAG, &[0; 8]), "(1791531456.435801) can_pcan 20000004#0000000000000000"); + } +} diff --git a/src/recorder/clock.rs b/src/recorder/clock.rs new file mode 100644 index 0000000..64b729e --- /dev/null +++ b/src/recorder/clock.rs @@ -0,0 +1,40 @@ +//! The car unit has no RTC battery: the clock boots stale and is stepped later by +//! chrony (NTP or GNSS). Recordings therefore carry enough to correct timestamps +//! afterwards — sync state at start, monotonic time, and every step observed. + +use serde_json::{json, Value}; + +fn clock(id: libc::clockid_t) -> f64 { + let mut ts: libc::timespec = unsafe { std::mem::zeroed() }; + unsafe { libc::clock_gettime(id, &mut ts) }; + ts.tv_sec as f64 + ts.tv_nsec as f64 * 1e-9 +} + +pub fn realtime() -> f64 { + clock(libc::CLOCK_REALTIME) +} + +pub fn monotonic() -> f64 { + clock(libc::CLOCK_MONOTONIC) +} + +/// realtime − monotonic. Constant while the clock only slews; a jump means a step. +pub fn offset() -> f64 { + realtime() - monotonic() +} + +pub fn boot_id() -> String { + std::fs::read_to_string("/proc/sys/kernel/random/boot_id").map(|s| s.trim().to_string()).unwrap_or_default() +} + +/// Kernel view of synchronisation, read without privilege (adjtimex with modes = 0). +pub fn sync_state() -> Value { + let mut tx: libc::timex = unsafe { std::mem::zeroed() }; + let state = unsafe { libc::adjtimex(&mut tx) }; + let unsync = state == libc::TIME_ERROR || tx.status & libc::STA_UNSYNC != 0; + json!({ "synced": !unsync, "maxerror_us": tx.maxerror, "esterror_us": tx.esterror }) +} + +pub fn hostname() -> String { + std::fs::read_to_string("/proc/sys/kernel/hostname").map(|s| s.trim().to_string()).unwrap_or_default() +} diff --git a/src/recorder/main.rs b/src/recorder/main.rs new file mode 100644 index 0000000..3f4a794 --- /dev/null +++ b/src/recorder/main.rs @@ -0,0 +1,428 @@ +//! recorder — persists raw CAN and GNSS to disk, session by session. +//! +//! Reads SocketCAN and gpsd directly, not through the broker: capture is tier 1 and +//! must not depend on anything that can be late or down. +//! +//! Sessions start when CAN frames arrive (auto) or on `ctl start` (manual), and an +//! auto session ends after `--idle` seconds without frames: ignition off, adapter +//! unplugged, or plugged in somewhere with no bus. An adapter with no traffic never +//! creates a session, so leaving the PEAK attached at a desk produces nothing. + +mod can; +mod clock; +mod session; +mod sources; +mod storage; + +use std::io::{Read, Write}; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::process::ExitCode; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::mpsc::{sync_channel, RecvTimeoutError}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use serde_json::json; + +use session::{Limits, Session}; +use sources::Event; +use storage::Bound; + +const USAGE: &str = "\ +usage: + recorder run [options] + --iface NAME CAN interface (can_pcan) + --root DIR recordings directory (/var/lib/yukart/recordings) + --gpsd HOST:PORT gpsd, '' to disable (127.0.0.1:2947) + --ctl PATH control socket (/run/yukart-recorder/ctl.sock) + --idle SECS end an auto session after this long without frames (15) + --min-secs SECS discard auto sessions shorter than this with no marks (3) + --chunk-secs SECS rotate the CAN log after this long (300) + --chunk-mb MB ... or this size (64) + --cap-gb GB total size of all recordings (50) + --min-free-gb GB keep at least this much free on the filesystem (20) + --no-auto only record on 'ctl start' + recorder ctl [--ctl PATH] COMMAND + status | start [LABEL] | stop | mark TEXT | auto on|off"; + +struct Config { + iface: String, + root: PathBuf, + gpsd: String, + ctl: String, + idle: Duration, + min_secs: u64, + limits: Limits, + bound: Bound, + auto: bool, +} + +fn parse_run(args: &[String]) -> Result { + let mut c = Config { + iface: "can_pcan".into(), + root: "/var/lib/yukart/recordings".into(), + gpsd: "127.0.0.1:2947".into(), + ctl: "/run/yukart-recorder/ctl.sock".into(), + idle: Duration::from_secs(15), + min_secs: 3, + limits: Limits { chunk_secs: 300, chunk_bytes: 64 << 20 }, + bound: Bound { cap_bytes: 50 << 30, min_free_bytes: 20 << 30 }, + auto: true, + }; + let mut it = args.iter(); + while let Some(a) = it.next() { + if a == "--no-auto" { + c.auto = false; + continue; + } + let v = it.next().ok_or(format!("{a} needs a value"))?; + let num = || v.parse::().map_err(|_| format!("{a}: not a number: {v}")); + match a.as_str() { + "--iface" => c.iface = v.clone(), + "--root" => c.root = v.into(), + "--gpsd" => c.gpsd = v.clone(), + "--ctl" => c.ctl = v.clone(), + "--idle" => c.idle = Duration::from_secs(num()?), + "--min-secs" => c.min_secs = num()?, + "--chunk-secs" => c.limits.chunk_secs = num()?, + "--chunk-mb" => c.limits.chunk_bytes = num()? << 20, + "--cap-gb" => c.bound.cap_bytes = num()? << 30, + "--min-free-gb" => c.bound.min_free_bytes = num()? << 30, + _ => return Err(format!("unknown option {a}")), + } + } + Ok(c) +} + +fn ctl_client(args: &[String]) -> ExitCode { + let mut path = "/run/yukart-recorder/ctl.sock".to_string(); + let mut rest = args; + if rest.first().map(String::as_str) == Some("--ctl") && rest.len() >= 2 { + path = rest[1].clone(); + rest = &rest[2..]; + } + if rest.is_empty() { + eprintln!("{USAGE}"); + return ExitCode::from(2); + } + let run = || -> std::io::Result { + let mut s = UnixStream::connect(&path)?; + s.set_read_timeout(Some(Duration::from_secs(6)))?; + s.write_all(rest.join(" ").as_bytes())?; + s.shutdown(std::net::Shutdown::Write)?; + let mut out = String::new(); + s.read_to_string(&mut out)?; + Ok(out) + }; + match run() { + Ok(out) => { + print!("{out}"); + if out.starts_with("error") { ExitCode::from(1) } else { ExitCode::SUCCESS } + } + Err(e) => { + eprintln!("recorder ctl: {path}: {e}"); + ExitCode::from(1) + } + } +} + +struct Recorder { + cfg: Config, + session: Option, + compress: std::sync::mpsc::Sender, + auto: bool, + /// After a manual stop, auto-start waits for the bus to go idle first, + /// otherwise it would reopen a session immediately. + suppressed: bool, + last_frame: Option, + last_tpv: Option<(f64, String)>, + can_link: (bool, String), + gnss_link: (bool, String), + kernel_drops: u64, + queue_drops: Arc, + queue_drops_seen: u64, +} + +impl Recorder { + fn start(&mut self, trigger: &'static str, label: Option) -> Result<(), String> { + let deleted = storage::enforce(&self.cfg.root, &self.cfg.bound, None); + let seq = storage::next_seq(&self.cfg.root); + let mut s = Session::open(&self.cfg.root, seq, trigger, label, &self.cfg.iface, self.compress.clone()) + .map_err(|e| format!("cannot open session: {e}"))?; + eprintln!("session {} started ({trigger})", s.name); + if !deleted.is_empty() { + let _ = s.event("storage_bound", json!({ "deleted": deleted })); + } + let _ = s.event("can_link", json!({ "up": self.can_link.0, "detail": self.can_link.1 })); + let _ = s.event("gnss_link", json!({ "up": self.gnss_link.0, "detail": self.gnss_link.1 })); + if let Some((rx, line)) = &self.last_tpv { + let _ = s.gnss(*rx, line, true); + } + self.session = Some(s); + Ok(()) + } + + fn stop(&mut self, reason: &str) { + let Some(s) = self.session.take() else { return }; + let short = s.trigger == "auto" && s.marks == 0 && s.bus_span().as_secs() < self.cfg.min_secs; + let name = s.name.clone(); + if short { + // A wake-up blip (door unlock, a brief plug-in) is not worth a session. + match s.discard() { + Ok(()) => eprintln!("session {name} discarded: under {}s of traffic, no marks", self.cfg.min_secs), + Err(e) => eprintln!("session {name} discard failed: {e}"), + } + return; + } + match s.close(reason) { + Ok(_) => eprintln!("session {name} closed: {reason}"), + Err(e) => eprintln!("session {name} close failed: {e}"), + } + } + + fn ctl(&mut self, cmd: &str) -> String { + let (verb, arg) = cmd.split_once(' ').map(|(v, a)| (v, a.trim())).unwrap_or((cmd, "")); + match verb { + "status" => { + json!({ + "auto": self.auto, + "suppressed": self.suppressed, + "session": self.session.as_ref().map(Session::status), + "can": { "up": self.can_link.0, "detail": self.can_link.1, + "last_frame_age_s": self.last_frame.map(|t| t.elapsed().as_secs_f64()) }, + "gnss": { "up": self.gnss_link.0, "detail": self.gnss_link.1 }, + "storage": { "root": self.cfg.root, "used_bytes": storage::usage(&self.cfg.root), + "free_bytes": storage::free_bytes(&self.cfg.root), + "cap_bytes": self.cfg.bound.cap_bytes }, + "clock": clock::sync_state(), + }) + .to_string() + } + "start" => { + let label = (!arg.is_empty()).then(|| arg.to_string()); + if let Some(s) = &mut self.session { + // Promote an auto session: it now ends only on 'stop'. + s.trigger = "manual"; + let _ = s.event("promoted", json!({ "label": label })); + return format!("ok: {} now manual", s.name); + } + match self.start("manual", label) { + Ok(()) => format!("ok: {}", self.session.as_ref().unwrap().name), + Err(e) => format!("error: {e}"), + } + } + "stop" => match &self.session { + Some(s) => { + let name = s.name.clone(); + self.stop("manual stop"); + self.suppressed = true; + format!("ok: {name} closed") + } + None => "ok: no session".into(), + }, + "mark" => match &mut self.session { + Some(s) if !arg.is_empty() => { + let _ = s.event("mark", json!({ "text": arg })); + format!("ok: mark {} in {}", s.marks, s.name) + } + Some(_) => "error: mark needs text".into(), + None => "error: no session; 'start' first".into(), + }, + "auto" => match arg { + "on" => { self.auto = true; "ok: auto on".into() } + "off" => { self.auto = false; "ok: auto off".into() } + _ => "error: auto on|off".into(), + }, + _ => format!("error: unknown command '{verb}'"), + } + } + + fn handle(&mut self, ev: Event) { + match ev { + Event::Can(f) => { + self.last_frame = Some(Instant::now()); + if self.session.is_none() && self.auto && !self.suppressed { + if let Err(e) = self.start("auto", None) { + eprintln!("{e}"); + } + } + if let Some(s) = &mut self.session { + if let Err(e) = s.frame(&f) { + eprintln!("write: {e}"); + } + } + } + Event::KernelDrops(n) => { + self.kernel_drops += n as u64; + if let Some(s) = &mut self.session { + s.kernel_drops += n as u64; + let _ = s.event("kernel_drops", json!({ "frames": n })); + } + } + Event::CanLink(up, detail) => { + eprintln!("can: {detail}"); + self.can_link = (up, detail.clone()); + if let Some(s) = &mut self.session { + let _ = s.event("can_link", json!({ "up": up, "detail": detail })); + } + } + Event::Gnss(rx, line) => { + if line.contains("\"class\":\"TPV\"") { + self.last_tpv = Some((rx, line.clone())); + } + if let Some(s) = &mut self.session { + let _ = s.gnss(rx, &line, false); + } + } + Event::GnssLink(up, detail) => { + eprintln!("gnss: {detail}"); + self.gnss_link = (up, detail.clone()); + if let Some(s) = &mut self.session { + let _ = s.event("gnss_link", json!({ "up": up, "detail": detail })); + } + } + Event::Ctl(cmd, reply) => { + let out = self.ctl(&cmd); + let _ = reply.send(out); + } + } + } + + fn bus_idle(&self) -> bool { + self.last_frame.map(|t| t.elapsed() >= self.cfg.idle).unwrap_or(true) + } +} + +fn run(cfg: Config) -> ExitCode { + if let Err(e) = std::fs::create_dir_all(&cfg.root) { + eprintln!("recorder: {}: {e}", cfg.root.display()); + return ExitCode::from(1); + } + let compress = storage::spawn_compressor(); + for p in storage::recover(&cfg.root) { + let _ = compress.send(p); + } + + let stop = Arc::new(AtomicBool::new(false)); + for sig in [signal_hook::consts::SIGTERM, signal_hook::consts::SIGINT] { + let _ = signal_hook::flag::register(sig, stop.clone()); + } + + // ~100 s of frames at the measured 890 frames/s. + let (tx, rx) = sync_channel::(100_000); + let queue_drops = Arc::new(AtomicU64::new(0)); + sources::spawn_can(cfg.iface.clone(), tx.clone(), queue_drops.clone(), stop.clone()); + if !cfg.gpsd.is_empty() { + sources::spawn_gnss(cfg.gpsd.clone(), tx.clone(), stop.clone()); + } + if let Some(dir) = Path::new(&cfg.ctl).parent() { + let _ = std::fs::create_dir_all(dir); + } + if let Err(e) = sources::spawn_ctl(cfg.ctl.clone(), tx.clone()) { + // Manual control is lost, auto capture is not: keep going. + eprintln!("recorder: control socket {}: {e}", cfg.ctl); + } + drop(tx); + + eprintln!("recorder {} on {} → {}", env!("CARGO_PKG_VERSION"), cfg.iface, cfg.root.display()); + let mut r = Recorder { + auto: cfg.auto, + cfg, + session: None, + compress, + suppressed: false, + last_frame: None, + last_tpv: None, + can_link: (false, "not yet opened".into()), + gnss_link: (false, "not yet connected".into()), + kernel_drops: 0, + queue_drops, + queue_drops_seen: 0, + }; + + let mut last_sync = Instant::now(); + let mut last_bound = Instant::now(); + let mut clock_ref = clock::offset(); + loop { + match rx.recv_timeout(Duration::from_millis(200)) { + Ok(ev) => r.handle(ev), + Err(RecvTimeoutError::Timeout) => {} + Err(RecvTimeoutError::Disconnected) => break, + } + if stop.load(Ordering::Relaxed) { + // Drain what is already queued so the last frames before shutdown are kept. + while let Ok(ev) = rx.try_recv() { + if !matches!(ev, Event::Ctl(..)) { + r.handle(ev); + } + } + r.stop("shutdown"); + break; + } + if r.bus_idle() { + r.suppressed = false; + if r.session.as_ref().map(|s| s.trigger == "auto").unwrap_or(false) { + r.stop("bus idle"); + } + } + if last_sync.elapsed() >= Duration::from_secs(1) { + last_sync = Instant::now(); + let offset = clock::offset(); + let step = offset - clock_ref; + clock_ref = offset; + let qd = r.queue_drops.load(Ordering::Relaxed); + let new_qd = qd - r.queue_drops_seen; + r.queue_drops_seen = qd; + if let Some(s) = &mut r.session { + if step.abs() > 0.05 { + // chrony stepped the clock: everything before this is off by `step`. + let _ = s.event("clock_step", json!({ "step_s": step, "clock": clock::sync_state() })); + } + if new_qd > 0 { + s.queue_drops += new_qd; + let _ = s.event("queue_drops", json!({ "frames": new_qd })); + } + if let Err(e) = s.sync() { + eprintln!("sync: {e}"); + } + match s.maybe_rotate(&r.cfg.limits) { + Ok(true) => last_bound = Instant::now() - Duration::from_secs(3600), + Ok(false) => {} + Err(e) => eprintln!("rotate: {e}"), + } + } + } + if last_bound.elapsed() >= Duration::from_secs(60) { + last_bound = Instant::now(); + let active = r.session.as_ref().map(|s| s.dir.clone()); + let deleted = storage::enforce(&r.cfg.root, &r.cfg.bound, active.as_deref()); + if !deleted.is_empty() { + eprintln!("storage bound: deleted {deleted:?}"); + if let Some(s) = &mut r.session { + let _ = s.event("storage_bound", json!({ "deleted": deleted })); + } + } + } + } + eprintln!("recorder: stopped"); + ExitCode::SUCCESS +} + +fn main() -> ExitCode { + let args: Vec = std::env::args().skip(1).collect(); + match args.first().map(String::as_str) { + Some("run") => match parse_run(&args[1..]) { + Ok(cfg) => run(cfg), + Err(e) => { + eprintln!("recorder: {e}\n{USAGE}"); + ExitCode::from(2) + } + }, + Some("ctl") => ctl_client(&args[1..]), + _ => { + eprintln!("{USAGE}"); + ExitCode::from(2) + } + } +} diff --git a/src/recorder/session.rs b/src/recorder/session.rs new file mode 100644 index 0000000..ad52c02 --- /dev/null +++ b/src/recorder/session.rs @@ -0,0 +1,260 @@ +//! One recording session: a directory of append-only files. +//! +//! ```text +//! 000042-20261009T163736/ +//! session.json metadata; "status": open → closed (or truncated, if found open at startup) +//! can-0001.log.zst candump -L text, closed chunks compressed +//! can-0002.log the open chunk, plain text, fdatasync'd every second +//! gnss.ndjson every gpsd report, with the local receive time +//! events.ndjson marks, clock steps, link changes, drops +//! ``` +//! +//! Durability: power loss costs at most the last second of the open chunk. Closed +//! chunks are compressed to a temp file and renamed, so a crash mid-compression +//! leaves the plain chunk intact. + +use std::fs::{self, File, OpenOptions}; +use std::io::{self, BufWriter, Write}; +use std::os::unix::io::AsRawFd; +use std::path::{Path, PathBuf}; +use std::sync::mpsc::Sender; +use std::time::{Duration, Instant}; + +use serde_json::{json, Value}; + +use crate::can::Frame; +use crate::clock; + +pub struct Limits { + pub chunk_secs: u64, + pub chunk_bytes: u64, +} + +pub struct Session { + pub dir: PathBuf, + pub name: String, + pub trigger: &'static str, + iface: String, + meta: Value, + started: Instant, + chunk_idx: u32, + chunk_started: Instant, + chunk_bytes: u64, + can: BufWriter, + gnss: BufWriter, + events: BufWriter, + line: String, + pub frames: u64, + first_frame: Option, + last_frame: Option, + pub marks: u32, + pub kernel_drops: u64, + pub queue_drops: u64, + compress: Sender, +} + +fn append(path: &Path) -> io::Result> { + Ok(BufWriter::with_capacity(256 << 10, OpenOptions::new().create(true).append(true).open(path)?)) +} + +fn sync(w: &mut BufWriter) -> io::Result<()> { + w.flush()?; + w.get_ref().sync_data() +} + +pub fn sync_dir(dir: &Path) { + if let Ok(d) = File::open(dir) { + unsafe { libc::fsync(d.as_raw_fd()) }; + } +} + +/// Atomic replace: readers see the old or the new file, never half of one. +pub fn write_json(path: &Path, v: &Value) -> io::Result<()> { + let tmp = path.with_extension("json.tmp"); + let mut f = File::create(&tmp)?; + f.write_all(serde_json::to_string_pretty(v)?.as_bytes())?; + f.write_all(b"\n")?; + f.sync_all()?; + fs::rename(&tmp, path)?; + if let Some(d) = path.parent() { + sync_dir(d); + } + Ok(()) +} + +fn chunk_name(i: u32) -> String { + format!("can-{i:04}.log") +} + +impl Session { + pub fn open(root: &Path, seq: u64, trigger: &'static str, label: Option, iface: &str, + compress: Sender) -> io::Result { + let rt = clock::realtime(); + let stamp = { + let t = rt as libc::time_t; + let mut tm: libc::tm = unsafe { std::mem::zeroed() }; + unsafe { libc::localtime_r(&t, &mut tm) }; + format!("{:04}{:02}{:02}T{:02}{:02}{:02}", tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday, + tm.tm_hour, tm.tm_min, tm.tm_sec) + }; + // The sequence number orders sessions; the timestamp is for humans and may be stale. + let name = format!("{seq:06}-{stamp}"); + let dir = root.join(&name); + fs::create_dir_all(&dir)?; + let meta = json!({ + "format": 1, + "name": name, + "seq": seq, + "status": "open", + "trigger": trigger, + "label": label, + "host": clock::hostname(), + "boot_id": clock::boot_id(), + "iface": iface, + "recorder": env!("CARGO_PKG_VERSION"), + "start": { "realtime": rt, "monotonic": clock::monotonic(), "clock": clock::sync_state() }, + }); + write_json(&dir.join("session.json"), &meta)?; + let s = Session { + can: append(&dir.join(chunk_name(1)))?, + gnss: append(&dir.join("gnss.ndjson"))?, + events: append(&dir.join("events.ndjson"))?, + dir, name, trigger, meta, + iface: iface.to_string(), + started: Instant::now(), + chunk_idx: 1, + chunk_started: Instant::now(), + chunk_bytes: 0, + line: String::with_capacity(64), + frames: 0, first_frame: None, last_frame: None, marks: 0, kernel_drops: 0, queue_drops: 0, + compress, + }; + sync_dir(root); + Ok(s) + } + + pub fn frame(&mut self, f: &Frame) -> io::Result<()> { + self.line.clear(); + f.write_candump(&self.iface, &mut self.line); + self.line.push('\n'); + self.can.write_all(self.line.as_bytes())?; + self.chunk_bytes += self.line.len() as u64; + self.frames += 1; + let now = Instant::now(); + self.first_frame.get_or_insert(now); + self.last_frame = Some(now); + Ok(()) + } + + pub fn gnss(&mut self, rx: f64, raw: &str, stale: bool) -> io::Result<()> { + let msg = serde_json::from_str::(raw).unwrap_or_else(|_| Value::String(raw.to_string())); + let mut v = json!({ "rx": rx, "msg": msg }); + if stale { + // The last report seen before this session began, so a session starts with a position. + v["before_start"] = json!(true); + } + writeln!(self.gnss, "{v}") + } + + pub fn event(&mut self, kind: &str, mut extra: Value) -> io::Result<()> { + if !extra.is_object() { + extra = json!({}); + } + extra["rt"] = json!(clock::realtime()); + extra["mono"] = json!(clock::monotonic()); + extra["kind"] = json!(kind); + if kind == "mark" { + self.marks += 1; + } + writeln!(self.events, "{extra}") + } + + /// First to last frame: the trailing idle wait that ended the session does not count. + pub fn bus_span(&self) -> Duration { + match (self.first_frame, self.last_frame) { + (Some(a), Some(b)) => b - a, + _ => Duration::ZERO, + } + } + + /// Called about once a second: makes everything so far survive power loss. + pub fn sync(&mut self) -> io::Result<()> { + sync(&mut self.can)?; + sync(&mut self.gnss)?; + sync(&mut self.events) + } + + /// Returns true when a chunk was closed (a good moment to enforce the storage bound). + pub fn maybe_rotate(&mut self, lim: &Limits) -> io::Result { + if self.chunk_bytes == 0 + || (self.chunk_started.elapsed().as_secs() < lim.chunk_secs && self.chunk_bytes < lim.chunk_bytes) + { + return Ok(false); + } + sync(&mut self.can)?; + let closed = self.dir.join(chunk_name(self.chunk_idx)); + self.chunk_idx += 1; + self.can = append(&self.dir.join(chunk_name(self.chunk_idx)))?; + sync_dir(&self.dir); + self.chunk_started = Instant::now(); + self.chunk_bytes = 0; + let _ = self.compress.send(closed); + Ok(true) + } + + pub fn close(mut self, reason: &str) -> io::Result { + self.sync()?; + let last = self.dir.join(chunk_name(self.chunk_idx)); + drop(self.can); + if self.chunk_bytes == 0 { + let _ = fs::remove_file(&last); + } else { + let _ = self.compress.send(last); + } + self.meta["status"] = json!("closed"); + self.meta["end"] = json!({ + "realtime": clock::realtime(), "monotonic": clock::monotonic(), + "reason": reason, "clock": clock::sync_state(), + }); + self.meta["frames"] = json!(self.frames); + self.meta["chunks"] = json!(if self.chunk_bytes == 0 { self.chunk_idx - 1 } else { self.chunk_idx }); + self.meta["marks"] = json!(self.marks); + self.meta["drops"] = json!({ "kernel": self.kernel_drops, "recorder_queue": self.queue_drops }); + write_json(&self.dir.join("session.json"), &self.meta)?; + Ok(self.dir) + } + + /// Drop the session entirely, without queueing its chunk for compression. + pub fn discard(self) -> io::Result<()> { + let dir = self.dir.clone(); + drop(self); + fs::remove_dir_all(dir) + } + + pub fn status(&self) -> Value { + json!({ + "name": self.name, "trigger": self.trigger, "frames": self.frames, + "seconds": self.started.elapsed().as_secs(), "chunk": self.chunk_idx, "marks": self.marks, + "drops": { "kernel": self.kernel_drops, "recorder_queue": self.queue_drops }, + }) + } +} + +/// Compresses closed chunks off the writer thread: `x.log` → `x.log.zst.tmp` → `x.log.zst`, then +/// removes `x.log`. Every step leaves at least one complete copy on disk. +pub fn compress_chunk(plain: &Path) -> io::Result<()> { + let zst = plain.with_extension("log.zst"); + let tmp = plain.with_extension("log.zst.tmp"); + { + let mut input = File::open(plain)?; + let out = File::create(&tmp)?; + let mut enc = zstd::stream::Encoder::new(out, 9)?; + io::copy(&mut input, &mut enc)?; + enc.finish()?.sync_all()?; + } + fs::rename(&tmp, &zst)?; + if let Some(d) = plain.parent() { + sync_dir(d); + } + fs::remove_file(plain) +} diff --git a/src/recorder/sources.rs b/src/recorder/sources.rs new file mode 100644 index 0000000..17c6b80 --- /dev/null +++ b/src/recorder/sources.rs @@ -0,0 +1,149 @@ +//! Producer threads. Each owns one input, reconnects on its own, and never blocks +//! on the writer: CAN frames that do not fit the queue are counted, not waited for. + +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::{TcpStream, ToSocketAddrs}; +use std::os::unix::fs::PermissionsExt; +use std::os::unix::net::UnixListener; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::mpsc::{self, SyncSender, TrySendError}; +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +use crate::can::{CanSocket, Frame, Recv}; +use crate::clock; + +pub enum Event { + Can(Frame), + /// Frames the kernel dropped before we read them (SO_RXQ_OVFL). + KernelDrops(u32), + CanLink(bool, String), + Gnss(f64, String), + GnssLink(bool, String), + Ctl(String, mpsc::Sender), +} + +pub fn spawn_can(iface: String, tx: SyncSender, queue_drops: Arc, stop: Arc) { + thread::Builder::new().name("can".into()).spawn(move || { + let mut reported: Option = None; + while !stop.load(Ordering::Relaxed) { + let mut sock = match CanSocket::open(&iface, Duration::from_millis(200)) { + Ok(s) => s, + Err(e) => { + // An unplugged adapter is a normal state; say so once, then poll quietly. + let msg = format!("{iface}: {e}"); + if reported.as_deref() != Some(&msg) { + let _ = tx.send(Event::CanLink(false, msg.clone())); + reported = Some(msg); + } + thread::sleep(Duration::from_secs(1)); + continue; + } + }; + let _ = tx.send(Event::CanLink(true, format!("{iface}: socket open"))); + reported = None; + while !stop.load(Ordering::Relaxed) { + match sock.recv() { + Ok(Recv::Timeout) => {} + Ok(Recv::Frame(f, drops)) => { + if drops > 0 { + let _ = tx.try_send(Event::KernelDrops(drops)); + } + match tx.try_send(Event::Can(f)) { + Ok(()) => {} + Err(TrySendError::Full(_)) => { + queue_drops.fetch_add(1, Ordering::Relaxed); + } + Err(TrySendError::Disconnected(_)) => return, + } + } + Err(e) => { + // ENETDOWN / ENODEV: the link went down or the adapter left. + let _ = tx.send(Event::CanLink(false, format!("{iface}: {e}"))); + reported = Some(String::new()); + thread::sleep(Duration::from_millis(500)); + break; + } + } + } + } + }).expect("spawn can thread"); +} + +pub fn spawn_gnss(addr: String, tx: SyncSender, stop: Arc) { + thread::Builder::new().name("gnss".into()).spawn(move || { + // Report a failing gpsd once, not every retry. + let mut reported_down = false; + while !stop.load(Ordering::Relaxed) { + let conn = addr + .to_socket_addrs() + .ok() + .and_then(|mut a| a.next()) + .ok_or_else(|| std::io::Error::other("unresolvable")) + .and_then(|a| TcpStream::connect_timeout(&a, Duration::from_secs(2))); + let mut stream = match conn { + Ok(s) => s, + Err(e) => { + if !reported_down { + let _ = tx.send(Event::GnssLink(false, format!("gpsd {addr}: {e}"))); + reported_down = true; + } + thread::sleep(Duration::from_secs(3)); + continue; + } + }; + let _ = stream.set_read_timeout(Some(Duration::from_secs(1))); + if stream.write_all(b"?WATCH={\"enable\":true,\"json\":true};\n").is_err() { + continue; + } + let _ = tx.send(Event::GnssLink(true, format!("gpsd {addr}: watching"))); + let mut reader = BufReader::new(stream); + let mut line = String::new(); + loop { + if stop.load(Ordering::Relaxed) { + return; + } + match reader.read_line(&mut line) { + Ok(0) => break, + Ok(_) => { + if line.ends_with('\n') { + let rx = clock::realtime(); + let _ = tx.send(Event::Gnss(rx, line.trim_end().to_string())); + line.clear(); + } + } + Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {} + Err(_) => break, + } + } + let _ = tx.send(Event::GnssLink(false, format!("gpsd {addr}: connection lost"))); + reported_down = true; + thread::sleep(Duration::from_secs(1)); + } + }).expect("spawn gnss thread"); +} + +/// One command per connection, one reply line back. +pub fn spawn_ctl(path: String, tx: SyncSender) -> std::io::Result<()> { + let _ = std::fs::remove_file(&path); + let listener = UnixListener::bind(&path)?; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o660))?; + thread::Builder::new().name("ctl".into()).spawn(move || { + for conn in listener.incoming() { + let Ok(mut conn) = conn else { continue }; + let _ = conn.set_read_timeout(Some(Duration::from_secs(2))); + let mut buf = Vec::new(); + let _ = (&mut conn).take(4096).read_to_end(&mut buf); + let cmd = String::from_utf8_lossy(&buf).trim().to_string(); + let (rtx, rrx) = mpsc::channel(); + if tx.send(Event::Ctl(cmd, rtx)).is_err() { + return; + } + let reply = rrx.recv_timeout(Duration::from_secs(5)).unwrap_or_else(|_| "error: no reply".into()); + let _ = conn.write_all(reply.as_bytes()); + let _ = conn.write_all(b"\n"); + } + })?; + Ok(()) +} diff --git a/src/recorder/storage.rs b/src/recorder/storage.rs new file mode 100644 index 0000000..07cb242 --- /dev/null +++ b/src/recorder/storage.rs @@ -0,0 +1,134 @@ +//! The storage bound, enforced by the writer: oldest closed sessions go first, the +//! session in progress never does. A full disk would stop capture silently; losing +//! the oldest drive is recoverable. + +use std::fs; +use std::path::{Path, PathBuf}; + +use serde_json::Value; + +use crate::session::{compress_chunk, write_json}; + +pub struct Bound { + pub cap_bytes: u64, + pub min_free_bytes: u64, +} + +/// Session directories, oldest first. Ordered by sequence number, not by clock. +pub fn sessions(root: &Path) -> Vec<(u64, PathBuf)> { + let mut v: Vec<(u64, PathBuf)> = fs::read_dir(root) + .into_iter() + .flatten() + .flatten() + .filter(|e| e.file_type().map(|t| t.is_dir()).unwrap_or(false)) + .filter_map(|e| { + let seq = e.file_name().to_str()?.split('-').next()?.parse().ok()?; + Some((seq, e.path())) + }) + .collect(); + v.sort(); + v +} + +/// Never reuses a number, even after the bound has deleted every session: copies +/// taken off the car must not collide with later sessions of the same name. +pub fn next_seq(root: &Path) -> u64 { + let mark = root.join(".last-seq"); + let stored = fs::read_to_string(&mark).ok().and_then(|s| s.trim().parse().ok()).unwrap_or(0); + let seen = sessions(root).last().map(|(s, _)| *s).unwrap_or(0); + let next = stored.max(seen) + 1; + let tmp = root.join(".last-seq.tmp"); + if fs::write(&tmp, format!("{next}\n")).is_ok() { + let _ = fs::rename(&tmp, &mark); + } + next +} + +fn du(p: &Path) -> u64 { + fs::read_dir(p) + .into_iter() + .flatten() + .flatten() + .map(|e| match e.file_type() { + Ok(t) if t.is_dir() => du(&e.path()), + _ => e.metadata().map(|m| m.len()).unwrap_or(0), + }) + .sum() +} + +pub fn free_bytes(p: &Path) -> u64 { + let c = std::ffi::CString::new(p.as_os_str().as_encoded_bytes()).unwrap_or_default(); + let mut s: libc::statvfs = unsafe { std::mem::zeroed() }; + if unsafe { libc::statvfs(c.as_ptr(), &mut s) } != 0 { + return u64::MAX; + } + s.f_bavail as u64 * s.f_frsize as u64 +} + +pub fn usage(root: &Path) -> u64 { + du(root) +} + +/// Returns the names of sessions deleted. +pub fn enforce(root: &Path, b: &Bound, active: Option<&Path>) -> Vec { + let mut deleted = Vec::new(); + let mut total = du(root); + for (_, dir) in sessions(root) { + if total <= b.cap_bytes && free_bytes(root) >= b.min_free_bytes { + break; + } + if Some(dir.as_path()) == active { + continue; + } + let size = du(&dir); + if fs::remove_dir_all(&dir).is_ok() { + total = total.saturating_sub(size); + deleted.push(dir.file_name().unwrap_or_default().to_string_lossy().into_owned()); + } + } + deleted +} + +/// At startup no session is active, so anything still marked open was cut short +/// (power loss, crash). Mark it, compress its chunks, and drop half-written temp files. +pub fn recover(root: &Path) -> Vec { + let mut pending = Vec::new(); + for (_, dir) in sessions(root) { + let meta_path = dir.join("session.json"); + if let Some(mut meta) = fs::read_to_string(&meta_path).ok().and_then(|s| serde_json::from_str::(&s).ok()) { + if meta["status"] == "open" { + meta["status"] = Value::from("truncated"); + let _ = write_json(&meta_path, &meta); + } + } + for e in fs::read_dir(&dir).into_iter().flatten().flatten() { + let p = e.path(); + let name = p.file_name().unwrap_or_default().to_string_lossy().into_owned(); + if name.ends_with(".tmp") { + let _ = fs::remove_file(&p); + } else if name.starts_with("can-") && name.ends_with(".log") { + if e.metadata().map(|m| m.len() == 0).unwrap_or(false) { + let _ = fs::remove_file(&p); + } else { + pending.push(p); + } + } + } + } + pending +} + +pub fn spawn_compressor() -> std::sync::mpsc::Sender { + let (tx, rx) = std::sync::mpsc::channel::(); + std::thread::Builder::new() + .name("compress".into()) + .spawn(move || { + for p in rx { + if let Err(e) = compress_chunk(&p) { + eprintln!("compress {}: {e} (plain chunk kept)", p.display()); + } + } + }) + .expect("spawn compressor"); + tx +}