From 73a78eef205142ffb4f8a3ea7aeeefb4dd1856d2 Mon Sep 17 00:00:00 2001 From: legop3 Date: Sun, 27 Sep 2026 18:41:24 -0400 Subject: [PATCH] slop --- plans/rover learning service.md | 468 ++++++++++++++++++ server/index.js | 1 + .../src/configuration/configuration.test.js | 1 + server/src/configuration/definition.js | 2 + server/src/services/commandService/index.js | 9 +- .../src/services/commandService/recording.js | 55 ++ .../roverLearningService/configuration.js | 19 + .../services/roverLearningService/index.js | 39 ++ .../roverLearningService/recording.js | 119 +++++ .../services/roverLearningService/runtime.js | 254 ++++++++++ .../services/roverLearningService/storage.js | 59 +++ 11 files changed, 1024 insertions(+), 2 deletions(-) create mode 100644 plans/rover learning service.md create mode 100644 server/src/services/commandService/recording.js create mode 100644 server/src/services/roverLearningService/configuration.js create mode 100644 server/src/services/roverLearningService/index.js create mode 100644 server/src/services/roverLearningService/recording.js create mode 100644 server/src/services/roverLearningService/runtime.js create mode 100644 server/src/services/roverLearningService/storage.js diff --git a/plans/rover learning service.md b/plans/rover learning service.md new file mode 100644 index 00000000..b9103b3b --- /dev/null +++ b/plans/rover learning service.md @@ -0,0 +1,468 @@ +# Rover learning service implementation tracker + +## Resume here + +Status: implementation authorized; first recording/lifecycle increment is coded. +Training, model catalog, autonomous control, and the VIP card are not implemented. +The standalone +CPU benchmark exists at `perf/rover-model-benchmark.py` and was run successfully +on the user's actual server. No autonomous-driving capability is implemented. + +This document preserves the conversation's decisions across compactions and +usage cutoffs. Keep the checklist and handoff section current during authorized +implementation. Do not treat proposals or open questions as user decisions. + +The user explicitly authorized feature implementation with: "okay cool. begin +implementation, use the md to track. IPI". Continue the agreed implementation +without requesting IPI again. Follow the current +AGENTS.md instructions: targeted investigation, minimal edits, no new tests unless +requested, proportional direct verification, and no development processes left +running. This development machine is not the actual rover server. + +Next work: verify real recording output on the server, then define the actual +feature/action schema and implement training/checkpoint publication. The VIP entry +point is `webui/src/components/VipPanel/index.jsx`; verified cards are in its +`isVerified` branch, and client settings use `useSettingsNamespace`. The common +browser rover command sender is `webui/src/controls/commandPipeline.js`. Finish +the control-path inventory before declaring every capability learnable. + +## Intended product + +Learn from ordinary human rover use, then let a user select a saved model and +press Start to have it roam and operate the rover's controls. No destination or +structured route is required. Users should not need dedicated demonstration +sessions before collection becomes useful. Build the complete collection and +training pipeline; do not reduce the intended product to a single-route demo. + +Chat is out of scope. The user explicitly dropped it to keep the model compact. +This is a learned control policy, not a language model generating instructions. + +Learning useful behavior is an experimental objective, not a guaranteed result +of accumulating recordings. Autonomous performance must be observed; low training +loss or fast inference alone does not establish competent driving. + +## Agreed requirements + +- One self-contained server service, provisionally `roverLearningService`. +- Minimal integration elsewhere: configuration/startup, command recording hook, + checked command submission hook, and the necessary UI connection. +- The service owns its video recorder processes. Do not move this feature into + replay, rover management, turn management, or unrelated services. +- Configuration defaults to disabled. Disabled means no active feature runtime. +- Record every rover control input, video, and all available rover sensors. +- Model inputs use recent control history/current state, useful sensors, and + video. Model outputs can use the rover's supported operating controls, not just + wheel speeds. Available hardware/control capabilities must be accounted for. +- Record broadly; choosing training features must not discard raw source data + prematurely within the configured retention window. +- Retain a reasonable bounded rolling training dataset, with ongoing background + training. The model architecture/parameter count remains bounded as it learns. +- Periodically publish immutable named model versions. Two random words plus a + unique identifier/sequence is the proposed naming scheme. +- Keep every published model permanently. No automatic model pruning. +- Users can choose which compatible model controls their rover. +- Add the UI card at the top of the VIP tab, using the existing verified-feature + mechanism. The card is absent when the feature is disabled. +- Model control must use the same permissions and eligibility rules as manual + rover control. Verified feature access does not grant rover control. +- Favorites are client-side saved model IDs only. No server favorite/pin state, + endpoints, or retention behavior tied to favorites. +- All server information needed to populate the card arrives in one coherent + state object. Action commands remain separate from that state object. + +Earlier suggestions to prune old model versions or retain only server-pinned +favorites are superseded. Rolling retention applies to training data, not the +published model catalog. + +## Architecture and containment + +Proposed internal ownership (not a requirement to scaffold empty directories): + +```text +server/src/services/roverLearningService/ + configuration.js Schema/defaults; no runtime side effects + index.js Small enable/disable lifecycle entry point + runtime.js Enabled runtime coordination + recording/ Commands, sensors, video processes, synchronization + dataset/ Rolling storage, training sampling, retention + training/ Training scheduling and checkpoint production + models/ Permanent catalog, names, metadata, publication + driving/ Inference sessions and action scheduling + workers/ Python model/training implementation + socketGateway.js Full card state and authorized user actions +``` + +Persistent video, datasets, checkpoints, and catalog metadata belong under the +application's existing data-path convention, not the service source directory. +The agreed root is `/rover-learning/`, normally `/data/rover-learning/` in +the deployed container (`SERVER_DATA_DIR`), or `server/data/rover-learning/` with +the default development path. Intended layout: + +```text +rover-learning/ + recordings/ Rolling video/events/session metadata (implemented) + models/ Permanent published models (pending) + training/ Resumable training/working state (pending) + catalog.sqlite Session/model/progress index (proposed; not created yet) +``` + +Only `recordings/` currently exists when enabled. Its session manifests provide +the initial retention index; no placeholder model/training database is created. +Python compute and video processing must not block the Node event loop. Worker +dependencies, deployment packaging, and process shutdown must be included in the +implementation plan; installing Python packages at server startup is not assumed. + +The outside command system should know only about optional observation and +checked submission. It should know nothing about model architecture, datasets, +training, model names, or video recorders. + +### Existing integration findings (recheck before edits) + +- `server/index.js` explicitly loads services. +- `server/src/configuration/definition.js` assembles service-owned configuration + fragments. `configuration/index.js` provides live reload handlers. +- `server/src/services/commandService/index.js` exports `issueCommand()` and + `commandEvents`. Existing `observation` events intentionally omit payloads; + they are analytics events and insufficient as training examples. +- `issueCommand()` is a low-level send function. Many authorization, turn, + cooldown, private-drive safety, and overcurrent checks occur in the browser + command handler before calling it. Direct autonomous calls must not bypass + those checks. +- `roverManager.managerEvents` already emits decoded `sensor` events and rover + lifecycle events. The sensor event is not necessarily every raw telemetry field; + inventory what is available before claiming complete sensor capture. +- `fleetReportService` demonstrates subscriptions and configuration lifecycle, + but its persistent storage/lifecycle is not automatically the desired disabled + behavior for this new service. +- `replayEngineV2` demonstrates FFmpeg capture and RTSP/TCP source handling. Reuse + applicable media conventions without making this feature depend on replay + being enabled or sharing replay's retention ownership. + +### Command recording + +Proposed optional observation contract includes timestamp, rover, operator/source, +requested command payload, dispatched payload, outcome, and command ID where +available. Requested and dispatched values can differ due to safety processing. +Distinguish dispatched/acknowledged commands from measured physical execution. + +Record human, autonomous, and service-generated commands with distinct provenance. +Autonomous predictions are not automatically correct human training labels. +Include rejected/ignored outcomes where relevant; do not present them as executed +actions. Inventory control paths outside the common browser command handler so +camera/accessory controls are not silently omitted. + +The hook only enqueues bounded work. Recording failures or disk pressure must not +block/fail manual command delivery. Preserve the existing payload-free analytics +event contract instead of putting detailed recordings into general logging. +Requested payload snapshots must precede any in-place command transformation. + +### Checked command sending and ownership + +Proposed implementation: extract a reusable checked submission path from the +current browser command handler, preserving existing manual behavior. Autonomy +submits on behalf of its initiating user's actual control context. Do not invent +an admin identity or send directly to the rover WebSocket. + +Recheck permissions server-side for actions and continued control. Losing the +turn/assignment/connection or otherwise losing eligibility stops the autonomous +session. A safe stop must still be possible after ordinary permission is revoked. +The exact policy for an incoming manual control command during autonomy is open; +immediate takeover is the proposed behavior. Prevent two controllers fighting. + +Model actions describe operating controls, not arbitrary server maintenance. +Inventory reboot/update/raw commands and capability restrictions explicitly rather +than allowing unrestricted command dictionaries from model outputs. + +### Disabled lifecycle + +Only configuration metadata and a small lifecycle handler remain registered while +disabled, to support live enablement. Lazy-load runtime modules on enable. + +When disabled there are no feature recording subscriptions, command observers, +video/Python workers, runtime timers, dataset scans, open feature databases, +created directories, inference/training, or feature socket handlers. The existing +config/feature mechanism hides the card without starting this runtime. + +On disable: stop autonomous motion, detach observers, stop and await owned child +processes, flush/close storage appropriately, remove socket handlers/timers, and +release runtime state. Preserve all saved data/models. Serialize enable/disable +transitions so overlapping reloads cannot leave duplicate workers or listeners. + +## Data and model design + +The benchmark architecture is a representative workload, not a selected final +policy. It uses four RGB frames at 160x120, 32 sensor and 32 control/state values +per frame, a spatial CNN and GRU, eight continuous outputs, and sixteen button +logits. These counts are placeholders, not the actual rover control schema. + +- Past controls/current state are inputs; the human's next action is a training + target. Avoid future-action leakage into model input. +- Maintain temporal coherence. Independent guesses can hesitate or oscillate; + multiple human intentions must not simply average into meaningless controls. +- Continuous controls and discrete button/toggle events require different output + semantics. Do not retrigger a one-shot action every inference tick. +- Include camera pose/state when available, and supported control capabilities. +- Preserve synchronized timestamps and sequence boundaries. Server receive time + does not prove which frame the remote driver saw. Exact displayed-frame timing + may need a small frontend timing hook; that is not yet designed or authorized as + a settled integration requirement. +- Human takeovers/corrections are valuable demonstrations. Do not indiscriminately + discard playful behavior: the goal includes human-like unstructured roaming. +- An action timeout/stale-video stop and existing hardware safety remain outside + the model's learned decision-making. + +Training proposal: shuffle windows from retained sessions, including recent and +older examples. Publish after a configured amount of training work, potentially +with a minimum elapsed interval. Steps/examples are less ambiguous than epochs +over a continuously changing dataset. Continue training independently of active +inference; published models are immutable and do not change beneath a session. + +Background work should be resource-limited and lower priority. Pausing after +sufficient reuse of unchanged data is proposed; continuous repeated passes are +not automatically useful. A small reserve of older valuable sessions was suggested +to reduce forgetting but is not a settled retention requirement. + +Training readers must not lose files mid-batch to retention. Publish checkpoints +atomically so clients cannot select partially written models. Store sufficient +architecture, feature/action schema, normalization, and training metadata to load +each permanent model correctly or report it as incompatible. + +## Model catalog and UI mechanics + +Proposed model names: `curious-otter-0042`, `amber-finch-0043`. Names are display +labels; stable unique IDs are used for selection and local favorites. Record +creation time, architecture/schema identity, training progress, and dataset +provenance/summary. Never assume a newer model is better. + +Card scope: model selection with local favorites, Start/Stop, current session +status, and compact training status. Reuse current VIP appearance and existing +settings persistence rather than inventing a new storage layer. UI implementation +locations have not been inspected yet. + +Illustrative single state object (not a finalized protocol): + +```js +{ + available: true, + roverId, + control: { canStart, canStop, reason }, + session: { status, modelId, startedBy, startedAt }, + models: [{ id, name, createdAt, trainingSteps, compatible }], + training: { status, examplesProcessed, latestModelId } +} +``` + +Derive per-user permissions on the server and recheck on execution. State updates +should be coherent and event-driven/coalesced, not broadcast at camera frequency. +The permanent catalog can grow; avoid blocking command handling while assembling +or delivering card state. Any future pagination must preserve the user's single +card-state contract rather than silently introducing independent UI data feeds. + +Selection semantics proposed earlier: stop before switching an active model, +reset temporal state, and resume only explicitly. Confirm exact behavior as part +of UI design. Client favorites do not affect server retention or permissions. + +## Hardware and measured feasibility + +Actual server: Fedora 43, two Xeon E5-2620 v4 processors, 16 physical cores / 32 +threads, AVX2, about 125 GiB RAM with 118 GiB available in the initial sample. +The server only runs the rover server. Approximately 80% CPU idle was measured +under normal working load, not on an otherwise idle test machine. + +The relevant benchmark is the custom vision/control workload, not the earlier +Qwen text benchmark. Server run: Python 3.14.7, PyTorch 2.14.0+cpu, FP32, +488,008 parameters, four RGB frames at 160x120. Training batch size 16. + +| Mode | Threads | Rover/training batch | p50 ms | p95 ms | Max ms | Examples/s | CPU cores used | Peak MiB | +|---|---:|---:|---:|---:|---:|---:|---:|---:| +| Inference | 4 | 1 | 5.8 | 8.0 | 16.5 | 165.43 | 3.97 | 321 | +| Inference | 4 | 2 | 10.4 | 13.6 | 24.1 | 187.39 | 3.96 | 334 | +| Inference | 4 | 4 | 18.7 | 25.1 | 34.1 | 207.08 | 3.96 | 357 | +| Inference | 8 | 1 | 4.4 | 5.9 | 12.9 | 219.98 | 7.95 | 318 | +| Inference | 8 | 2 | 7.5 | 9.2 | 13.2 | 272.28 | 7.95 | 330 | +| Inference | 8 | 4 | 15.3 | 17.5 | 26.0 | 263.88 | 7.81 | 359 | +| Inference | 16 | 1 | 4.9 | 7.3 | 23.1 | 191.94 | 15.77 | 318 | +| Inference | 16 | 2 | 7.5 | 10.9 | 21.0 | 257.54 | 15.76 | 333 | +| Inference | 16 | 4 | 11.6 | 17.9 | 33.6 | 319.84 | 14.98 | 359 | +| Training | 4 | 16 | 210.7 | 227.4 | 245.5 | 75.36 | 3.79 | 655 | +| Training | 8 | 16 | 141.2 | 172.8 | 176.5 | 108.57 | 7.09 | 673 | +| Training | 16 | 16 | 121.6 | 228.9 | 263.4 | 112.37 | 13.29 | 669 | + +Inference latency is per entire batch, not per rover within it. CPU measurements +are saturation runs, not paced 10 Hz operation. Eight threads is a promising +starting point, not a finalized resource allocation. At 10 Hz there are 36,000 +training windows per recorded rover-hour; one pass takes about 5.5 compute-only +minutes at eight threads. Dataset hours and repeat passes multiply that cost. + +Limits: synthetic data/random weights, no decoding/resizing/network/dataset I/O, +no concurrent inference and training, and no measured autonomous driving quality. +The target of 10 decisions/second is provisional. A larger model or higher image +resolution requires renewed measurement. These results do not mandate a GPU. + +Approximate storage for this architecture: 1.9 MiB FP32 inference weights; 5.6 MiB +for weights plus Adam optimizer state, before metadata/serialization overhead. +One hundred inference versions are about 186 MiB. Dataset video dominates storage. +Published inference artifacts and resumable training state need not be identical. + +## Open implementation decisions + +- Exact operating-control catalog, continuous/event action encoding, capability + masking, selected sensors, model architecture/history, and update rate. +- Dataset disk budget, segment format, sampling strategy, session quality markers, + and whether to retain a reserved historical subset. +- Time alignment between video observed by the driver and command generation. +- Python packaging, IPC, process supervision, CPU limits, and worker failure policy. +- Checkpoint cadence/names, resume-state retention, evaluation metadata, and how + new versions become selectable. Do not add an unrequested admin approval flow. +- Exact ownership/takeover behavior, disconnect/turn-loss handling, model switch + behavior, and which user may stop an active session under existing permissions. +- Card fields and existing client persistence mechanism; current UI paths still + need targeted inspection. + +## Implementation checklist + +- [x] Discuss scope: autonomous roaming, full rover controls, no chat. +- [x] Add standalone CPU benchmark and verify its execution locally. +- [x] Obtain and interpret results from the actual server. +- [x] Record containment, default-disabled behavior, VIP placement, permissions, + permanent model retention, client favorites, and single card-state requirement. +- [x] Obtain IPI for feature implementation. +- [ ] Inspect remaining UI/control paths and settle concrete contracts. +- [x] Add service-owned config and dormant lifecycle integration. +- [ ] Add optional detailed command observer and shared checked submission path. +- [ ] Implement timestamped recording, video worker ownership, and bounded dataset. + Code is implemented for common commands, raw/decoded rover sensor frames, + host stats, and existing rover state events. Live video/timing/retention + verification and full control-path inventory remain; do not mark complete yet. +- [ ] Implement feature/action encoding and training worker with resource limits. +- [ ] Implement atomic checkpoint publication and permanent model catalog. +- [ ] Implement model inference, ownership, takeover, stops, and worker cleanup. +- [ ] Implement service socket gateway with complete per-user card state. +- [ ] Implement verified VIP card at top, model selection, local favorites, controls. +- [ ] Verify disabled startup and enable/disable cleanup directly. +- [ ] Verify manual-control behavior remains intact and autonomous permission loss + stops motion using an appropriate authorized runtime. +- [ ] Measure real video pipeline and concurrent training/inference on the server. +- [ ] Evaluate autonomous behavior with supervised rover runs before claiming + useful driving performance. + +Verification must follow user policy: no new/expanded tests unless explicitly +requested; use existing small relevant checks or actual affected runtime. Do not +automatically run full suites/builds. Hardware/server checks must be handed to +the user when the development environment cannot perform them. + +## Handoff log + +### First implementation increment + +Implemented files: + +- `server/src/services/roverLearningService/configuration.js`: default disabled; + recording budget 20 GiB, minimum free space 2 GiB, session duration 300 seconds. + These are initial implementation defaults, adjustable in the existing admin + configuration editor, not measured final production sizing. +- `server/src/services/roverLearningService/index.js`: serialized lifecycle bridge; + lazy runtime require; no recording side effects while disabled. +- `server/src/services/roverLearningService/runtime.js`: enabled-only subscriptions, + recorder ownership, bounded startup/rotation event buffering, retention scheduling, + disconnect/rotation handling, and signal/disable cleanup. +- `server/src/services/roverLearningService/recording.js`: per-session metadata, + bounded NDJSON output, and FFmpeg RTSP/TCP video packet copy into Matroska. + No camera re-encoding, Python process, training, or autonomous commands yet. +- `server/src/services/roverLearningService/storage.js`: oldest-first recording + retention, free-space enforcement, and interrupted-session recovery. +- `server/src/services/commandService/recording.js`: optional failure-isolated + request/dispatch/result/ack observer with payload snapshots and IDs. Source is + `client` or `server`; browser automation is not falsely classified as proven + intentional human action. Async safety timers do not retain client attribution. + +External changes are limited to startup, configuration definition, command-hook +wiring, and adding the new feature to an existing configuration test's expected +feature list (required by the feature addition). No new tests were created. +Raw sensor capture reads `lastSensor.raw` during the existing synchronous sensor +event; rover manager/sensor pipeline did not need modification. + +Recording format: + +```text +recordings/-/ + session.json Schema version, rover/session identity, timing description, + start/end/reason/error, loss count, video exit information + events.ndjson Unix ms + process monotonic ns, kind, data + video.mkv Original video codec; receiver wall-clock PTS retained +``` + +Request/dispatch/ack IDs let later training join requested and applied commands. +The recorded `source` is provenance, not proof of a correct demonstration. +Queued events preserve their observation timestamp; file order around a new +session's snapshot is not necessarily chronological. Sort by timestamps when +constructing training windows. Rotations can have video gaps; do not train across +session boundaries as if footage were continuous. + +Limits and follow-up: + +- Retention is checked every 15 seconds and is a soft disk budget, not a filesystem + quota. Active files are closed before becoming eligible for pruning. Low free + space pauses capture. Models/training paths are never traversed by retention. +- Record only rovers with a connected human driver who currently passes the + existing control checks, including deterrence. Membership is required even for + admins so an admin login alone does not record unattended rovers. Preserve pauses + within an eligible human-control session. Driver/turn/mode/permission changes + reconcile recording immediately; telemetry buffering uses the same eligibility. + This supersedes the initial all-connected-rovers policy at the user's request. + Autonomous sessions will need explicit exclusion/provenance when implemented; + there are currently no autonomous controllers. +- Each event stream has a 2 MiB pending-write bound; startup/rotation buffering has + a shared 4 MiB bound. Session loss counters/capture-gap events identify overload. +- Video timestamps represent packet receipt at this server, not the frame a human + browser displayed. Actual FFmpeg output PTS still needs inspection on the server. +- Enabled recorders are stopped on disable and awaited; process-exit hooks kill + surviving children. Existing MediaMTX shutdown can exit the whole Node process + before async metadata flush finishes; next enable marks those sessions interrupted. +- Storage leases for concurrent training reads are pending with dataset sampling. +- Safe command submission extraction, autonomous provenance, training, permanent + catalog, model selection, client favorites, and UI remain pending. + +Verification completed: + +- `node --check` on new recording/lifecycle modules and modified command service: + passed (repeat only for subsequently changed modules as needed). +- `node --test server/src/configuration/configuration.test.js`: 14 passed, 1 failed. + The failing pre-existing default-values assertion omits + `$.discord.channels.liveStatus`; the current HEAD already defines it as empty. + New configuration validation, descriptions, normalization, and feature flags + passed. Leave the unrelated Discord assertion untouched. +- Direct disabled service startup with a disposable `SERVER_DATA_DIR` returned + `{"runtimeLoaded":false,"storageCreated":false}` and exited normally. Temporary + configuration storage was removed. No development service was left running. +- Local FFmpeg help confirms the RTSP transport and timeout options used. +- No full build, full lint, new tests, live server startup, or rover hardware run. + +Server verification needed for this increment: + +1. Deploy normally, initially disabled; confirm no `rover-learning` directory or + learning FFmpeg workers are created by a fresh disabled startup. +2. Enable `roverLearning.enabled` through the existing admin configuration. An + online unattended rover must not produce recordings. Take human control and + confirm session files appear beneath the actual data mount; pauses should keep + recording. Release control/disconnect the last eligible driver and confirm + capture stops. Repeat across turn changes and mode/permission changes. +3. Drive, move the camera, and operate accessories. Check that event records contain + requested and dispatched payloads with matching IDs, sensor frames, and host stats. +4. For a completed session, inspect video with: + `ffprobe -v error -select_streams v:0 -read_intervals '%+#3' -show_entries packet=pts_time,dts_time -of json /actual/session/video.mkv`. + Compare packet timestamps (seconds) with event timestamps (milliseconds). + Report FFmpeg/session errors rather than assuming timestamp alignment is correct. +5. Disable: confirm owned FFmpeg processes exit, session metadata closes, no further + writes occur, and saved recordings remain. Re-enable to check clean restart. +6. Verify rotation and retention with an intentionally chosen small recording + budget, and confirm manual driving remains responsive during recording. + +Return the session metadata, a small representative event sample (excluding any +unwanted personal text), FFprobe timestamps, and any `roverLearning` errors. Actual +server capture/timing and live cleanup remain unverified until these checks run. + +When resuming after implementation begins, append concrete changed files, commands +run/results, unresolved failures, decisions approved by the user, and the next +bounded task here. Do not mark a checkbox complete merely because code exists; +record verification limits and outstanding server/hardware work explicitly. diff --git a/server/index.js b/server/index.js index 105a92ad..46b0692e 100644 --- a/server/index.js +++ b/server/index.js @@ -68,6 +68,7 @@ require('./src/services/batteryManager'); // subscriptions see fully decoded state without becoming an initialization // dependency of either control path. require('./src/services/fleetReportService'); +require('./src/services/roverLearningService'); require('./src/services/replayEngineV2'); // Replay delivery is a core service. It must subscribe before the optional // Discord feature so web requests always have a local delivery path. diff --git a/server/src/configuration/configuration.test.js b/server/src/configuration/configuration.test.js index 0664c4c2..57139c71 100644 --- a/server/src/configuration/configuration.test.js +++ b/server/src/configuration/configuration.test.js @@ -184,6 +184,7 @@ test('service definitions generate public feature paths without a separate regis { key: 'discord', path: ['discord', 'enabled'] }, { key: 'socials', path: ['socials', 'enabled'] }, { key: 'fleetReports', path: ['fleetReports', 'enabled'] }, + { key: 'roverLearning', path: ['roverLearning', 'enabled'] }, ]); }); diff --git a/server/src/configuration/definition.js b/server/src/configuration/definition.js index 9b48f156..fb0de1c6 100644 --- a/server/src/configuration/definition.js +++ b/server/src/configuration/definition.js @@ -22,6 +22,7 @@ const barcodeScanner = require('../services/barcodeScannerService/configuration' const commands = require('../services/operatorCommandService/configuration'); const discord = require('../services/discordBotService/configuration'); const fleetReports = require('../services/fleetReportService/configuration'); +const roverLearning = require('../services/roverLearningService/configuration'); /* Object property order is preserved by JSON serialization and JSON Schema @@ -55,6 +56,7 @@ const definitions = [ sessionConfiguration.socials, sessionConfiguration.driverAd, fleetReports, + roverLearning, ]; const defaultConfig = Object.fromEntries( diff --git a/server/src/services/commandService/index.js b/server/src/services/commandService/index.js index a0d0dbec..aec12caa 100644 --- a/server/src/services/commandService/index.js +++ b/server/src/services/commandService/index.js @@ -11,6 +11,7 @@ const logger = require('../../globals/logger').child('commandService'); const { isHeadlightBlocked } = require('../../rewards/definitions/darkness'); const homeAssistantService = require('../homeAssistantService'); const overcurrentProtectionService = require('../overcurrentProtectionService'); +const { subscribeCommandRecording, observeCommand, recordCommandRequest } = require('./recording'); // Command observations are intentionally separate from the global event bus. // Drive and motor commands can run at control-loop frequency, and publishing @@ -96,6 +97,7 @@ function issueCommand(roverId, payload) { const normalizedPayload = normalizeOutboundCommandPayload(payload); const message = { ...normalizedPayload, id }; record.ws.send(JSON.stringify(message)); + observeCommand({ phase: 'dispatched', roverId, commandId: id, command: normalizedPayload }); pendingCommands.set(id, { roverId, ts: Date.now(), type: normalizedPayload.type }); logger.info('Issued command', roverId, normalizedPayload.type, id); return id; @@ -130,6 +132,7 @@ function handleAck(msg) { const pending = pendingCommands.get(msg.id); if (!pending) return; pendingCommands.delete(msg.id); + observeCommand({ phase: 'acknowledged', roverId: pending.roverId, commandId: msg.id, acknowledgement: msg }); logger.info('Command acknowledged', pending.roverId, pending.type, msg.status); io.emit('commandAck', { roverId: pending.roverId, @@ -255,6 +258,7 @@ module.exports = { getRecentDriveActivity, setDriveCooldown, commandEvents, + subscribeCommandRecording, }; io.on('connection', (socket) => { @@ -392,8 +396,9 @@ io.on('connection', (socket) => { } } - socket.on('command', handleCommand); - socket.on('command:issue', handleCommand); + const recordedCommand = (request, reply) => recordCommandRequest(socket, request, reply, handleCommand); + socket.on('command', recordedCommand); + socket.on('command:issue', recordedCommand); socket.on('command:updateAllRovers', (_payload = {}, cb) => { const reply = typeof cb === 'function' ? cb : () => {}; diff --git a/server/src/services/commandService/recording.js b/server/src/services/commandService/recording.js new file mode 100644 index 00000000..a97117a4 --- /dev/null +++ b/server/src/services/commandService/recording.js @@ -0,0 +1,55 @@ +// Optional detailed command observation, separate from payload-free analytics. +// Subscribers must only enqueue work; recorder failures cannot fail a command. +const { AsyncLocalStorage } = require('node:async_hooks'); +const { randomUUID } = require('node:crypto'); + +const observers = new Set(); +const requests = new AsyncLocalStorage(); + +function subscribeCommandRecording(observer) { + observers.add(observer); + return () => observers.delete(observer); +} + +function observeCommand(event) { + if (!observers.size) return; + const scope = requests.getStore(); + const context = scope?.active ? scope.context : null; + for (const observer of observers) { + try { + // Snapshot before safety code or a caller mutates a nested payload. Each + // subscriber receives its own copy and cannot change command delivery. + observer(structuredClone({ + ts: Date.now(), + monotonicNs: process.hrtime.bigint().toString(), + source: context ? 'client' : 'server', + ...context, + ...event, + })); + } catch { + // Recording is deliberately best-effort at this boundary. The owning + // recorder reports its storage health independently of driver replies. + } + } +} + +function recordCommandRequest(socket, request, reply, execute) { + if (!observers.size) return execute(request, reply); + const context = { requestId: randomUUID(), socketId: socket.id, userId: socket.data?.userId || null }; + const scope = { active: true, context }; + return requests.run(scope, () => { + try { + observeCommand({ phase: 'requested', roverId: request?.roverId, command: request }); + return execute(request, (result) => { + observeCommand({ phase: 'result', roverId: request?.roverId, result }); + if (typeof reply === 'function') reply(result); + }); + } finally { + // The command handler is synchronous. A safety timer created inside it + // inherits async context but its later commands are server actions. + scope.active = false; + } + }); +} + +module.exports = { subscribeCommandRecording, observeCommand, recordCommandRequest }; diff --git a/server/src/services/roverLearningService/configuration.js b/server/src/services/roverLearningService/configuration.js new file mode 100644 index 00000000..af0542b1 --- /dev/null +++ b/server/src/services/roverLearningService/configuration.js @@ -0,0 +1,19 @@ +// Metadata only: loading the configuration must never start the learning runtime. +const { strictObject, boolean, integer, number } = require('../../configuration/schemaHelpers'); + +module.exports = { + key: 'roverLearning', + feature: true, + defaultValue: { + enabled: false, + recording: { maxGiB: 20, minimumFreeGiB: 2, sessionSeconds: 300 }, + }, + schema: strictObject({ + enabled: boolean({ description: 'Enable rover learning recording. Disabled starts no recording workers or storage. Training and autonomous control are not implemented yet.' }), + recording: strictObject({ + maxGiB: number({ minimum: 0.1, maximum: 100000, description: 'Rolling recording budget in GiB. Checked every 15 seconds; active files can temporarily exceed it. Published models are never pruned.' }), + minimumFreeGiB: number({ minimum: 0.1, maximum: 100000, description: 'Pause recording below this amount of available filesystem space, in GiB.' }), + sessionSeconds: integer({ minimum: 30, maximum: 3600, description: 'Rotate video and telemetry sessions after this many seconds so completed sessions can be pruned.' }), + }, { description: 'Bounded recording storage. Limits never delete published models.', required: ['maxGiB', 'minimumFreeGiB', 'sessionSeconds'] }), + }, { description: 'Optional rover learning service. Currently provides passive recording only.', required: ['enabled', 'recording'] }), +}; diff --git a/server/src/services/roverLearningService/index.js b/server/src/services/roverLearningService/index.js new file mode 100644 index 00000000..a69ccbeb --- /dev/null +++ b/server/src/services/roverLearningService/index.js @@ -0,0 +1,39 @@ +// The only resident part while disabled is this configuration lifecycle bridge. +const { loadConfig, registerConfigurationHandler } = require('../../configuration'); +const logger = require('../../globals/logger').child('roverLearning'); + +let runtime = null; +let transitions = Promise.resolve(); +let shuttingDown = false; + +function apply(config) { + const transition = transitions.then(async () => { + if (runtime) { + const previous = runtime; + runtime = null; + await previous.stop(); + } + if (!config?.enabled || shuttingDown) return; + const { createRuntime } = require('./runtime'); + const next = createRuntime({ config, logger }); + try { + await next.start(); + runtime = next; + } catch (error) { + await next.stop(); + throw error; + } + }); + transitions = transition.catch(() => undefined); + return transition; +} + +registerConfigurationHandler('roverLearning', apply); +apply(loadConfig().roverLearning).catch((error) => { + logger.error('Unable to start rover learning recording', { error: error.message }); +}); + +// Signal handlers are installed by the enabled runtime, not by this bridge. +module.exports = { + stop: () => { shuttingDown = true; return apply({ enabled: false }); }, +}; diff --git a/server/src/services/roverLearningService/recording.js b/server/src/services/roverLearningService/recording.js new file mode 100644 index 00000000..74d5798e --- /dev/null +++ b/server/src/services/roverLearningService/recording.js @@ -0,0 +1,119 @@ +// One bounded recording session. Video packets retain receiver wall-clock PTS; +// telemetry uses the same server clock, not an assumed browser playback clock. +const fs = require('node:fs'); +const fsp = require('node:fs/promises'); +const path = require('node:path'); +const { randomUUID } = require('node:crypto'); +const { spawn } = require('node:child_process'); + +const MAX_BUFFER_BYTES = 2 * 1024 * 1024; + +async function saveMetadata(directory, metadata) { + const filename = path.join(directory, 'session.json'); + await fsp.writeFile(`${filename}.tmp`, JSON.stringify(metadata, null, 2)); + await fsp.rename(`${filename}.tmp`, filename); +} + +async function createRecording({ root, roverId, snapshot, logger }) { + const id = `${Date.now()}-${randomUUID()}`; + const directory = path.join(root, id); + await fsp.mkdir(directory); + const metadata = { + version: 1, id, roverId, startedAt: Date.now(), endedAt: null, + timing: 'Video PTS uses server receiver Unix seconds. Events use Unix milliseconds and process monotonic nanoseconds. Browser display delay is not measured.', + video: { file: 'video.mkv', codec: 'source-copy', timestampUnit: 'seconds', origin: 'unix' }, + events: 'events.ndjson', droppedEvents: 0, + }; + await saveMetadata(directory, metadata); + const output = fs.createWriteStream(path.join(directory, 'events.ndjson'), { flags: 'wx' }); + let failure = null; + let stopping = false; + let stopPromise; + let child; + let childClosed = false; + let stderr = ''; + let videoResult = null; + let resolveVideo; + const videoClosed = new Promise((resolve) => { resolveVideo = resolve; }); + output.on('error', (error) => { + failure = error.message; + // Stop file growth immediately when telemetry is no longer writable. + child?.kill('SIGTERM'); + }); + + function append(kind, data, timestamp = {}) { + if (stopping || failure) return false; + try { + const line = JSON.stringify({ + ts: Date.now(), monotonicNs: process.hrtime.bigint().toString(), ...timestamp, kind, data, + }) + '\n'; + // Never wait for disk from a control/sensor callback. Explicit loss counts + // prevent an overloaded recorder from pretending its history is complete. + if (output.writableLength + Buffer.byteLength(line) > MAX_BUFFER_BYTES) { + metadata.droppedEvents += 1; + return true; + } + output.write(line); + } catch (error) { + metadata.droppedEvents += 1; + failure = error.message; + child?.kill('SIGTERM'); + } + return true; + } + + append('snapshot', snapshot); + // Read the local published stream, as replay does, without depending on the + // replay service or re-encoding camera video. No audio/chat is captured. + child = spawn(process.env.FFMPEG_BIN || 'ffmpeg', [ + '-hide_banner', '-loglevel', 'warning', '-nostdin', '-n', + '-rtsp_transport', 'tcp', '-timeout', '10000000', + '-use_wallclock_as_timestamps', '1', '-copyts', + '-i', `rtsp://127.0.0.1:8554/${encodeURIComponent(roverId)}`, + '-map', '0:v:0', '-an', '-c:v', 'copy', '-avoid_negative_ts', 'disabled', + '-f', 'matroska', path.join(directory, 'video.mkv'), + ], { stdio: ['ignore', 'ignore', 'pipe'] }); + child.stderr.on('data', (chunk) => { stderr = (stderr + chunk.toString()).slice(-4096); }); + child.on('error', (error) => { failure = error.message; }); + child.once('close', (code, signal) => { + childClosed = true; + videoResult = { code, signal, stderr }; + if (!stopping) failure ||= `Video recorder exited (${code ?? signal})`; + resolveVideo(); + }); + + async function stop(reason) { + if (stopPromise) return stopPromise; + stopPromise = (async () => { + append('sessionEnd', { reason, droppedEvents: metadata.droppedEvents }); + stopping = true; + if (!childClosed) child.kill('SIGTERM'); + // A stuck RTSP read must not leave a recorder behind after disable. + const killTimer = setTimeout(() => { if (!childClosed) child.kill('SIGKILL'); }, 2000); + await videoClosed; + clearTimeout(killTimer); + await new Promise((resolve) => { + if (output.closed) return resolve(); + output.once('close', resolve); + output.end(); + }); + metadata.endedAt = Date.now(); + metadata.reason = reason; + metadata.error = failure; + metadata.video.result = videoResult; + await saveMetadata(directory, metadata); + if (failure) logger.warn('Recording session incomplete', { roverId, id, error: failure, stderr }); + })(); + return stopPromise; + } + + return { + id, directory, startedAt: metadata.startedAt, append, stop, + get failed() { return Boolean(failure); }, + get stopping() { return stopping; }, + // Process exit is synchronous; normal disable uses stop() and awaits close. + kill: () => { if (!childClosed) child.kill('SIGKILL'); }, + }; +} + +module.exports = { createRecording, saveMetadata }; diff --git a/server/src/services/roverLearningService/runtime.js b/server/src/services/roverLearningService/runtime.js new file mode 100644 index 00000000..28eab2a7 --- /dev/null +++ b/server/src/services/roverLearningService/runtime.js @@ -0,0 +1,254 @@ +// Enabled-only recording runtime. No training or driving worker is started here. +const fs = require('node:fs/promises'); +const path = require('node:path'); +const { resolveDataPath } = require('../../helpers/dataPaths'); +const roverManager = require('../roverManager'); +const io = require('../../globals/io'); +const { turnEvents } = require('../turnService'); +const { modeEvents } = require('../modeManager'); +const { isAdmin, roleEvents } = require('../roleService'); +const { isDeterred, verificationEvents } = require('../verificationService'); +const { subscribeCommandRecording, commandEvents } = require('../commandService'); +const { createRecording } = require('./recording'); +const { recoverRecordings, enforceRetention } = require('./storage'); + +function createRuntime({ config, logger }) { + const root = resolveDataPath('rover-learning'); + const recordingsRoot = path.join(root, 'recordings'); + const sessions = new Map(); + const unsubscribers = []; + let running = false; + let timer; + let maintenance = Promise.resolve(); + let stopPromise; + let diskPaused = false; + let busy = false; + let wakeRequested = false; + let pending = []; + let pendingBytes = 0; + let droppedPending = 0; + let eligibleRovers = new Set(); + + function hasHumanController(roverId) { + const record = roverManager.rovers.get(roverId); + if (!record?.ws) return false; + // Admin permission alone must not record the whole fleet. Require actual + // driver membership, a live browser socket, and current control eligibility. + for (const socketId of record.drivers || []) { + const socket = io.sockets.sockets.get(socketId); + if (socket?.connected && (isAdmin(socket) || !isDeterred(socket)) + && roverManager.canDrive(roverId, socket)) return true; + } + return false; + } + + function currentEligibleRovers() { + return new Set([...roverManager.rovers.keys()].filter(hasHumanController)); + } + + function refreshEligibility() { + const next = currentEligibleRovers(); + const changed = next.size !== eligibleRovers.size || [...next].some((id) => !eligibleRovers.has(id)); + eligibleRovers = next; + if (!changed) return; + // Stop the stream immediately even if disk maintenance is in progress. + // The serialized maintenance pass still owns closing/removing the session. + for (const [id, session] of sessions) { + if (!next.has(id)) session.stop('no-human-controller').catch((error) => { + logger.warn('Unable to finalize unattended recording', { roverId: id, error: error.message }); + }); + } + schedule(); + } + + function append(kind, event) { + if (!running || event?.roverId == null) return; + const roverId = String(event.roverId); + if (!hasHumanController(roverId)) return; + const timestamp = { ts: Date.now(), monotonicNs: process.hrtime.bigint().toString() }; + if (sessions.get(roverId)?.append(kind, event, timestamp)) return; + if (diskPaused || !roverManager.rovers.has(roverId)) return; + // Session creation/rotation is asynchronous. Keep a bounded snapshot of + // events arriving during that gap, with their original observation times. + try { + const serialized = JSON.stringify(event); + const bytes = Buffer.byteLength(serialized); + if (pendingBytes + bytes > 4 * 1024 * 1024) { + droppedPending += 1; + return; + } + pending.push({ roverId, kind, serialized, timestamp, bytes }); + pendingBytes += bytes; + } catch { droppedPending += 1; } + } + + function listen(emitter, name, handler) { + emitter.on(name, handler); + unsubscribers.push(() => emitter.off(name, handler)); + } + + function snapshot(record) { + // Do not serialize the live record: it contains sockets and circular state. + return { + meta: record.meta, sensors: record.lastSensor, hostStats: record.lastHostStats, + batteryState: record.batteryState, headlightState: record.headlightState, + laserState: record.laserState, drivers: [...(record.drivers || [])], + locked: record.locked, privateOpen: record.privateOpen, + }; + } + + async function closeSession(roverId, reason) { + const session = sessions.get(roverId); + if (!session) return; + try { + await session.stop(reason); + } finally { + // Keep closing children reachable by the synchronous process-exit hook. + sessions.delete(roverId); + } + } + + async function maintain() { + if (!running) return; + pending = pending.filter((event) => hasHumanController(event.roverId)); + pendingBytes = pending.reduce((sum, event) => sum + event.bytes, 0); + for (const [roverId, session] of sessions) { + const record = roverManager.rovers.get(roverId); + if (!hasHumanController(roverId) || session.stopping || session.failed || Date.now() - session.startedAt >= config.recording.sessionSeconds * 1000) { + await closeSession(roverId, !record?.ws ? 'offline' : !hasHumanController(roverId) ? 'no-human-controller' : session.failed ? 'recorder-failed' : 'rotation'); + } + } + if (!running) return; + const retention = await enforceRetention( + recordingsRoot, new Set([...sessions.values()].map((session) => session.id)), + config.recording.maxGiB * 1024 ** 3, config.recording.minimumFreeGiB * 1024 ** 3, + ); + if (retention.paused) { + if (!diskPaused) logger.warn('Recording paused for disk budget', retention); + diskPaused = true; + pending = []; + pendingBytes = 0; + // Close active files to make them eligible for pruning on the next tick. + await Promise.all([...sessions.keys()].map((id) => closeSession(id, 'disk-budget'))); + return; + } + if (diskPaused) logger.info('Recording resumed after disk budget recovery'); + diskPaused = false; + for (const [key, record] of roverManager.rovers) { + if (!running) break; + const roverId = String(key); + if (!hasHumanController(roverId) || sessions.has(roverId)) continue; + const session = await createRecording({ root: recordingsRoot, roverId, snapshot: snapshot(record), logger }); + sessions.set(roverId, session); + // Filesystem awaits can span a disconnect or turn change. + if (!running || !hasHumanController(roverId)) { + await closeSession(roverId, running ? 'no-human-controller' : 'disabled'); + continue; + } + if (droppedPending) { + session.append('captureGap', { droppedEventsAcrossRovers: droppedPending }); + droppedPending = 0; + } + pending = pending.filter((event) => { + if (event.roverId !== roverId) return true; + session.append(event.kind, JSON.parse(event.serialized), event.timestamp); + pendingBytes -= event.bytes; + return false; + }); + } + } + + function schedule() { + if (!running) return; + if (busy) { wakeRequested = true; return; } + clearTimeout(timer); + busy = true; + // The next tick is scheduled after completion: disk slowness cannot build + // an unbounded backlog of retention tasks or overlapping recorder starts. + maintenance = maintain().catch(async (error) => { + logger.error('Recording maintenance failed; closing recorders', { error: error.message }); + await Promise.allSettled([...sessions.keys()].map((id) => closeSession(id, 'storage-error'))); + }).finally(() => { + busy = false; + if (running) timer = setTimeout(schedule, wakeRequested ? 0 : 15000); + wakeRequested = false; + }); + } + + function killChildren() { for (const session of sessions.values()) session.kill(); } + function onSignal() { + // Other server services can exit the process before asynchronous cleanup + // finishes. Signal children immediately, and keep an exit fallback too. + for (const session of sessions.values()) session.stop('server-shutdown').catch(() => undefined); + stop().catch((error) => logger.warn('Recording shutdown failed', { error: error.message })); + } + + async function start() { + await fs.mkdir(recordingsRoot, { recursive: true }); + await recoverRecordings(recordingsRoot); + running = true; + eligibleRovers = currentEligibleRovers(); + unsubscribers.push(subscribeCommandRecording((event) => append('command', event))); + // ACK analytics also provide latency/type context; detailed acknowledgements + // arrive through the recording hook and share the command ID. + listen(commandEvents, 'observation', (event) => append('commandOutcome', event)); + listen(roverManager.managerEvents, 'sensor', (event) => append('sensor', { + ...event, + // The manager stores the exact frame before emitting this synchronous + // event, so raw capture needs no extra hook in rover sensor processing. + frame: roverManager.rovers.get(event.roverId)?.lastSensor?.raw, + })); + for (const kind of ['hostStats', 'driver', 'switch', 'lock', 'private', 'privateSafety', 'help', 'dockGuard']) { + listen(roverManager.managerEvents, kind, (event) => { + append(kind, event); + if (['driver', 'switch', 'lock', 'private', 'privateSafety'].includes(kind)) refreshEligibility(); + }); + } + for (const kind of ['activeDriver', 'queue']) { + listen(turnEvents, kind, (event) => { append(`turn:${kind}`, event); refreshEligibility(); }); + } + listen(modeEvents, 'change', refreshEligibility); + listen(roleEvents, 'change', refreshEligibility); + listen(verificationEvents, 'change', refreshEligibility); + listen(roverManager.managerEvents, 'rover', (event) => { + append('rover', { roverId: event.roverId, action: event.action, snapshot: event.record ? snapshot(event.record) : null }); + if (event.action === 'upsert' || event.action === 'removed') refreshEligibility(); + }); + process.on('SIGTERM', onSignal); + process.on('SIGINT', onSignal); + process.on('exit', killChildren); + busy = true; + maintenance = maintain(); + try { + await maintenance; + } finally { + busy = false; + } + if (running) timer = setTimeout(schedule, wakeRequested ? 0 : 15000); + wakeRequested = false; + logger.info('Rover learning recording enabled', { root }); + } + + async function stop() { + if (stopPromise) return stopPromise; + running = false; + clearTimeout(timer); + for (const unsubscribe of unsubscribers.splice(0)) unsubscribe(); + stopPromise = (async () => { + await maintenance.catch(() => undefined); + const results = await Promise.allSettled([...sessions.keys()].map((id) => closeSession(id, 'disabled'))); + process.off('SIGTERM', onSignal); + process.off('SIGINT', onSignal); + process.off('exit', killChildren); + pending = []; + pendingBytes = 0; + const failed = results.find((result) => result.status === 'rejected'); + if (failed) throw failed.reason; + })(); + return stopPromise; + } + + return { start, stop }; +} + +module.exports = { createRuntime }; diff --git a/server/src/services/roverLearningService/storage.js b/server/src/services/roverLearningService/storage.js new file mode 100644 index 00000000..e174b628 --- /dev/null +++ b/server/src/services/roverLearningService/storage.js @@ -0,0 +1,59 @@ +// Recording retention only. Published models and training state are never visited. +const fs = require('node:fs/promises'); +const path = require('node:path'); +const { saveMetadata } = require('./recording'); + +async function listRecordings(root) { + const recordings = []; + for (const entry of await fs.readdir(root, { withFileTypes: true })) { + // Directory names are generated, never rover IDs or user-supplied paths. + if (!entry.isDirectory() || !/^\d+-[0-9a-f-]{36}$/.test(entry.name)) continue; + const directory = path.join(root, entry.name); + let bytes = 0; + for (const file of await fs.readdir(directory, { withFileTypes: true })) { + if (!file.isFile()) continue; + bytes += (await fs.stat(path.join(directory, file.name))).size; + } + recordings.push({ id: entry.name, directory, bytes, startedAt: Number(entry.name.split('-')[0]) }); + } + return recordings.sort((a, b) => a.startedAt - b.startedAt); +} + +async function recoverRecordings(root) { + for (const recording of await listRecordings(root)) { + let metadata; + try { + metadata = JSON.parse(await fs.readFile(path.join(recording.directory, 'session.json'), 'utf8')); + } catch (error) { + if (error.code !== 'ENOENT' && !(error instanceof SyntaxError)) throw error; + // An interrupted initial metadata write is retained as an unusable session + // until ordinary retention removes it, rather than blocking all startup. + metadata = { version: 1, id: recording.id, startedAt: recording.startedAt }; + } + if (metadata.endedAt != null) continue; + await saveMetadata(recording.directory, { + ...metadata, endedAt: Date.now(), reason: 'interrupted', + error: 'Previous process did not close this session; final timestamps and video may be incomplete.', + }); + } +} + +async function enforceRetention(root, activeIds, maxBytes, minimumFreeBytes) { + const recordings = await listRecordings(root); + let bytes = recordings.reduce((total, recording) => total + recording.bytes, 0); + const stat = await fs.statfs(root); + let free = stat.bavail * stat.bsize; + for (const recording of recordings) { + if (bytes < maxBytes && free >= minimumFreeBytes) break; + if (activeIds.has(recording.id)) continue; + await fs.rm(recording.directory, { recursive: true }); + bytes -= recording.bytes; + // Re-read actual free space; sparse/compressed files and other services can + // make a deleted file's logical size a misleading estimate of freed space. + const current = await fs.statfs(root); + free = current.bavail * current.bsize; + } + return { bytes, free, paused: bytes >= maxBytes || free < minimumFreeBytes }; +} + +module.exports = { recoverRecordings, enforceRetention };