diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..8624ff3 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,13 @@ +.DS_Store +.claude/ +.cyclonedds/ +.draco/ +.zenoh-c/ +__pycache__/ +bin/ +internal/ddsgen/ +myenv/ +recordings/ +.env +docker-compose.override.yml +.git diff --git a/README.md b/README.md index 6c545ac..65912a8 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ A Go application that synchronously records multi-modal sensor data: - **Depth** frames from a CycloneDDS topic (RVL-compressed, lossless) - **Odometry** messages from a CycloneDDS topic - **Lowstate** (IMU/joints/battery/etc.) from a CycloneDDS topic +- **LLM traces** (prompt/response records) polled from a co-located OM1 process's Prometheus trace-export endpoint All streams are timestamped and organized into session directories for easy alignment and analysis. @@ -87,6 +88,33 @@ The profile has it enabled because this fleet's G1 has a D435i fitted (measured 14.9 Hz, 480x270 `16UC1`, RVL-compressed to ~34% of raw, lossless). On a G1 without one, set `ENABLE_DEPTH=false`. +#### DDS readers reconnect on their own after a sensor restart + +`lidar`, `pointcloud`, `depth`, `odom`, and `lowstate` each hold one +long-lived DDS reader for the life of the process -- a session rotation only +swaps their output files (see `persistentStreams` in `cmd/main/main.go`), it +never resubscribes. + +Plain DDS discovery does not always recover cleanly if the underlying sensor +or its driver process restarts (a physical power-cycle, say): a writer +announces itself with a burst of discovery packets right at startup and then +falls back to a much slower steady-state interval, so a reader that's been +sitting idle for a while can simply miss that burst and never learn the new +writer exists -- nothing retries that specific handshake afterward. A reader +created fresh (a full process restart, or a brand-new subscriber from a tool +like `ros2 topic hz`) sends its own discovery burst immediately and tends to +match right away, which is what makes this failure mode easy to miss until a +stream has been silently empty for hours. + +Each of the five streams above is registered with +`heartbeat.Monitor.RegisterRecoverable` instead of plain `Register`: once a +stream is found broken (the same "recorder NOT WORKING" condition that gets +logged), the monitor calls that stream's `Reconnect()` once per check +interval until it recovers. `Reconnect()` tears down and recreates just that +stream's DDS participant/reader -- forcing a fresh discovery burst -- without +touching output files, so no manual restart should be needed after a sensor +bounces. + #### The G1's down camera is the RealSense RGB view There is no third USB camera on a G1 -- only `OM1FRONTCAM` and `OM1REARCAM`. @@ -139,6 +167,9 @@ Any of the following override the selected profile / defaults: - `ODOM_DDS_TOPIC` - DDS topic for odometry data (default: `rt/odom`) - `LOWSTATE_DDS_DOMAIN` - CycloneDDS domain ID for lowstate (default: `0`) - `LOWSTATE_DDS_TOPIC` - DDS topic for lowstate data (default: `rt/lowstate`) +- `TRACES_ENABLED` - Poll a co-located OM1 process's Prometheus trace-export endpoint and record its LLM traces (default: `false` — most deployments have no OM1 process to poll; see "LLM trace export" below) +- `TRACES_URL` - OM1's metrics endpoint (default: `http://localhost:9090/metrics`) +- `TRACES_POLL_INTERVAL` - How often to poll `TRACES_URL` (default: `30s`) - `RECORDINGS_DIR` - Base directory for recordings (default: `recordings`) - `SESSION_ROTATE_INTERVAL` - Close the current session and open a fresh one on this cadence, so each segment can be uploaded without waiting for the whole run to end (default: `5m`; `0` disables rotation — one session for the life of the process, as before this feature existed) - `UPLOAD_ENABLED` - Upload each rotated-away session to the openmind-api (default: `true`; actual uploading additionally requires `OPENMIND_API_URL` and `OPENMIND_API_KEY` — see "Cloud upload" below) @@ -248,7 +279,8 @@ recordings/ ├── depth_frames.bin # RVL-compressed depth frames (lossless) ├── depth_timestamps.csv # unix_ns,seq,byte_offset,byte_length,method,width,height,encoding,mono_ns ├── odom_frames.bin # Raw odometry messages - └── odom_timestamps.csv # Timestamps: unix_ns,seq,byte_offset,mono_ns + ├── odom_timestamps.csv # Timestamps: unix_ns,seq,byte_offset,mono_ns + └── traces.jsonl # LLM trace records polled from OM1 (if TRACES_ENABLED set) ``` Every timestamps CSV carries a **`mono_ns`** column alongside its wall-clock @@ -474,6 +506,33 @@ its own copy of it (a snapshot taken when that segment closes), so each segment's directory stays self-contained for `align_recording.py` / `fix_session_time.py` without needing the boot session alongside it. +## LLM trace export + +Set `TRACES_ENABLED=true` to poll a co-located [OM1](https://github.com/OpenMind/OM1) +process's Prometheus trace-export endpoint (`use_tracer.prometheus_export.enabled` +in OM1's config) and record each LLM interaction trace into the current +session's `traces.jsonl` -- picked up by the normal upload pipeline like +every other stream, with no extra bucket-path configuration needed. + +This is a persistent stream: it keeps one poll loop and one dedup cursor +(the newest trace timestamp already written) running for the life of the +process, only swapping which session's `traces.jsonl` it writes to on +rotation -- so OM1 re-serving its whole buffer on every poll never produces +duplicate lines, even across a session rotation. The cursor is also +persisted to `RECORDINGS_DIR/.traces_cursor`, so a full process restart +resumes from it too, instead of re-ingesting OM1's whole buffered backlog +into whatever session happens to be open at that moment. + +OM1's exporter only buffers its most recent 200 trace records; poll more +often than that window fills at your deployment's trace rate +(`TRACES_POLL_INTERVAL`, default `30s`) or older records will be evicted on +the OM1 side before this recorder ever sees them. + +`TRACES_URL` points at OM1's main `/metrics` endpoint -- `om1_trace_info` +is broadcast there alongside every other metric, so this poller (like OM1's +own fleet-wide Prometheus, if one scrapes the same endpoint) will also see +full prompt/response text as unbounded label values on that metric. + ## Scheduling `SCHEDULE_FILE` points at an optional YAML file describing a daily diff --git a/cmd/main/main.go b/cmd/main/main.go index 6aef31d..85c5abe 100644 --- a/cmd/main/main.go +++ b/cmd/main/main.go @@ -25,6 +25,7 @@ import ( "om1-telemetry/internal/retention" "om1-telemetry/internal/schedule" "om1-telemetry/internal/session" + "om1-telemetry/internal/traces" "om1-telemetry/internal/upload" "om1-telemetry/internal/video" ) @@ -32,6 +33,13 @@ import ( // scheduleTickInterval is how often the schedule reconciler checks for a state change. const scheduleTickInterval = 30 * time.Second +// stuckStreamThreshold is consecutive 30s heartbeat failures (~5 minutes) +// before a recoverable stream escalates from reconnecting to a full restart. +const stuckStreamThreshold = 10 + +// shutdownUploadTimeout bounds the synchronous upload of the live session at shutdown. +const shutdownUploadTimeout = 10 * time.Minute + func main() { // Sampled before anything else, so the session's monotonic anchor covers // the whole run. @@ -63,7 +71,24 @@ func main() { } mon := heartbeat.NewMonitor(30 * time.Second) - registerHeartbeats(mon, cfg, videoHeartbeatNames) + + restartSignal := make(chan string, 1) + mon.SetStuckThreshold(stuckStreamThreshold, func(stream string) { + select { + case restartSignal <- stream: + default: + } + }) + + var rs *recorderSet + streamsFn := func() *persistentStreams { + if rs == nil { + return nil + } + return &rs.persistent + } + + registerHeartbeats(mon, cfg, videoHeartbeatNames, streamsFn) hbCtx, hbCancel := context.WithCancel(context.Background()) go mon.Run(hbCtx) @@ -118,9 +143,9 @@ func main() { // Uploading (never cap enforcement) is gated on ctl.Uploading -- see retention.RunSweeps. sweepCtx, sweepCancel := context.WithCancel(context.Background()) - go retention.RunSweeps(sweepCtx, uploader, recordingsDir, bootTimebasePath, currentDirFn, cfg.Retention, ctl) + go retention.RunSweeps(sweepCtx, uploader, recordingsDir, bootTimebasePath, currentDirFn, cfg.Retention, ctl, cfg.Upload.DeleteAfterUpload) - rs := startRecorders(cfg, mon, recordingsDir, sess, videoHeartbeatNames) + rs = startRecorders(cfg, mon, recordingsDir, sess, videoHeartbeatNames) videoURLs := make([]string, 0, len(cfg.Video)) for _, vc := range cfg.Video { @@ -191,7 +216,10 @@ func main() { retention.SnapshotTimebase(bootTimebasePath, finished.RealDir()) ctl.SetRecording(false) slog.Info("recording paused by schedule", "session", finished.RealDir()) - retention.UploadFinishedSessionAsync(&uploadWG, uploader, ctl, recordingsDir, bootTimebasePath, finished, uploadDelete) + persistent := rs.persistent + finishedStart := time.Unix(0, finished.StartUnixNs()) + retention.UploadFinishedSessionAsync(&uploadWG, uploader, ctl, recordingsDir, bootTimebasePath, finished, uploadDelete, + func(ctx context.Context) { persistent.awaitSegments(ctx, finishedStart) }) } // resumeRecording is the schedule's open-half: same as a rotation's open-half. @@ -208,7 +236,7 @@ func main() { currentMu.Unlock() sess, cfg = next, nextCfg - registerHeartbeats(mon, cfg, videoHeartbeatNames) + registerHeartbeats(mon, cfg, videoHeartbeatNames, streamsFn) rs = startRecorders(cfg, mon, recordingsDir, sess, videoHeartbeatNames) ctl.SetRecording(true) slog.Info("recording resumed by schedule", "session", sess.RealDir()) @@ -242,12 +270,19 @@ func main() { reconcileSchedule() } + var restartRequested bool + loop: for { select { case <-shutdownSignal: break loop + case stream := <-restartSignal: + slog.Error("restarting process to recover a stuck stream", "stream", stream) + restartRequested = true + break loop + case <-scheduleC: reconcileSchedule() @@ -276,7 +311,7 @@ loop: currentMu.Unlock() sess, cfg = next, nextCfg - registerHeartbeats(mon, cfg, videoHeartbeatNames) + registerHeartbeats(mon, cfg, videoHeartbeatNames, streamsFn) persistent := rs.persistent rs = startEphemeral(cfg, mon) @@ -285,7 +320,9 @@ loop: slog.Info("session rotated", "session", sess.RealDir()) - retention.UploadFinishedSessionAsync(&uploadWG, uploader, ctl, recordingsDir, bootTimebasePath, finished, uploadDelete) + finishedStart := time.Unix(0, finished.StartUnixNs()) + retention.UploadFinishedSessionAsync(&uploadWG, uploader, ctl, recordingsDir, bootTimebasePath, finished, uploadDelete, + func(ctx context.Context) { persistent.awaitSegments(ctx, finishedStart) }) } } @@ -302,7 +339,11 @@ loop: if uploader != nil && ctl.Uploading() { opts := retention.UploadOptions(bootTimebasePath, sess.RealDir()) - retention.UploadSession(ctl, uploader, sess.RealDir(), retention.APISessionDir(recordingsDir, sess.RealDir()), time.Unix(0, sess.StartUnixNs()), uploadDelete, opts) + // No awaitReady needed: rs.Stop() above already drains ffmpeg's + // final segment (and its relocation) before returning. + ctx, cancel := context.WithTimeout(context.Background(), shutdownUploadTimeout) + retention.UploadSession(ctx, ctl, uploader, sess.RealDir(), retention.APISessionDir(recordingsDir, sess.RealDir()), time.Unix(0, sess.StartUnixNs()), uploadDelete, opts, nil) + cancel() } } uploadWG.Wait() @@ -314,29 +355,58 @@ loop: "dir", sess.RealDir(), "note", "it will be dated on a later start if it ever syncs") } + + if restartRequested { + os.Exit(1) + } } -func registerHeartbeats(mon *heartbeat.Monitor, cfg config.Config, videoHeartbeatNames []string) { +// registerHeartbeats registers every stream's heartbeat check, wiring the +// five DDS-backed streams up with a reconnect callback. +func registerHeartbeats(mon *heartbeat.Monitor, cfg config.Config, videoHeartbeatNames []string, streams func() *persistentStreams) { if cfg.EnableLidar { - mon.Register(lidar.HeartbeatName, 10) // ~10 Hz nominal + mon.RegisterRecoverable(lidar.HeartbeatName, 10, func() { + if p := streams(); p != nil && p.lidarStream != nil { + p.lidarStream.Reconnect() + } + }) } if cfg.EnablePointCloud { - mon.Register(pointcloud.HeartbeatName, 10) // ~10 Hz nominal + mon.RegisterRecoverable(pointcloud.HeartbeatName, 10, func() { + if p := streams(); p != nil && p.pointCloudStream != nil { + p.pointCloudStream.Reconnect() + } + }) } if cfg.EnableDepth { - mon.Register(depth.HeartbeatName, 10) // measured 14.9 Hz on a G1; floor at 10 + mon.RegisterRecoverable(depth.HeartbeatName, 10, func() { + if p := streams(); p != nil && p.depthStream != nil { + p.depthStream.Reconnect() + } + }) } if cfg.EnableOdom { - mon.Register(odom.HeartbeatName, 30) // Go2 ~150, G1 ~495; 30 = safe floor + mon.RegisterRecoverable(odom.HeartbeatName, 30, func() { + if p := streams(); p != nil && p.odomStream != nil { + p.odomStream.Reconnect() + } + }) } if cfg.EnableLowstate { - mon.Register(lowstate.HeartbeatName, 400) // Go2 ~500, G1 ~1053; 400 = safe floor + mon.RegisterRecoverable(lowstate.HeartbeatName, 400, func() { + if p := streams(); p != nil && p.lowstateStream != nil { + p.lowstateStream.Reconnect() + } + }) } mon.Register(network.HeartbeatName, 0) // 0 = liveness check only mon.Register(audio.HeartbeatName, 0) // 0 = liveness check only if cfg.Features.Enabled() { mon.Register(features.HeartbeatName, 0) // 0 = liveness check only } + if cfg.Traces.Enabled { + mon.Register(traces.HeartbeatName, 0) + } for _, name := range videoHeartbeatNames { mon.Register(name, 0) } @@ -364,6 +434,9 @@ func unregisterHeartbeats(mon *heartbeat.Monitor, cfg config.Config, videoHeartb if cfg.Features.Enabled() { mon.Unregister(features.HeartbeatName) } + if cfg.Traces.Enabled { + mon.Unregister(traces.HeartbeatName) + } for _, name := range videoHeartbeatNames { mon.Unregister(name) } @@ -383,6 +456,7 @@ type persistentStreams struct { lowstateStream *lowstate.LowstateStream videoStreams []*video.VideoRTSPStream audioStream *audio.AudioRTSPStream + traceStream *traces.TraceStream } func newPersistentStreams(cfg config.Config, mon *heartbeat.Monitor, recordingsDir string, sess *session.Session, videoHeartbeatNames []string) persistentStreams { @@ -415,6 +489,11 @@ func newPersistentStreams(cfg config.Config, mon *heartbeat.Monitor, recordingsD lowstateCfg.Monitor = mon p.lowstateStream = lowstate.New(lowstateCfg) } + if cfg.Traces.Enabled { + tracesCfg := cfg.Traces.TraceStreamConfig() + tracesCfg.Monitor = mon + p.traceStream = traces.New(tracesCfg) + } p.videoStreams = make([]*video.VideoRTSPStream, 0, len(cfg.Video)) for i, vc := range cfg.Video { @@ -467,6 +546,9 @@ func (p *persistentStreams) start() { if p.lowstateStream != nil { p.lowstateStream.Start() } + if p.traceStream != nil { + p.traceStream.Start() + } for _, vs := range p.videoStreams { vs.Start() } @@ -489,6 +571,9 @@ func (p *persistentStreams) stop() { if p.lowstateStream != nil { p.lowstateStream.Stop() } + if p.traceStream != nil { + p.traceStream.Stop() + } for _, vs := range p.videoStreams { vs.Stop() } @@ -525,6 +610,11 @@ func (p *persistentStreams) rotate(cfg config.Config, sess *session.Session) { slog.Error("lowstate: rotate failed", "err", err) } } + if p.traceStream != nil { + if err := p.traceStream.Rotate(cfg.Traces.OutputFile); err != nil { + slog.Error("traces: rotate failed", "err", err) + } + } for i, vs := range p.videoStreams { vc := cfg.Video[i] out := realPath(vc.OutputFile, sessionRealDir) @@ -536,6 +626,25 @@ func (p *persistentStreams) rotate(cfg config.Config, sess *session.Session) { p.audioStream.Rotate(sessionStart, audioTs, framesFileFor(audioOut, "_packets.csv")) } +// awaitSegments waits for every video/audio stream to relocate +// sessionStart's segment before upload -- ffmpeg closes segments a full rotation interval late, so this must run first. +func (p *persistentStreams) awaitSegments(ctx context.Context, sessionStart time.Time) { + var wg sync.WaitGroup + for _, vs := range p.videoStreams { + wg.Add(1) + go func(vs *video.VideoRTSPStream) { + defer wg.Done() + vs.WaitSegment(ctx, sessionStart) + }(vs) + } + wg.Add(1) + go func() { + defer wg.Done() + p.audioStream.WaitSegment(ctx, sessionStart) + }() + wg.Wait() +} + // realPath rewrites path to sit in realDir instead of whatever directory it // was built from. func realPath(path, realDir string) string { diff --git a/config/config.go b/config/config.go index cca2e35..0998ce7 100644 --- a/config/config.go +++ b/config/config.go @@ -15,6 +15,7 @@ import ( "om1-telemetry/internal/network" "om1-telemetry/internal/odom" "om1-telemetry/internal/pointcloud" + "om1-telemetry/internal/traces" "om1-telemetry/internal/upload" "om1-telemetry/internal/video" ) @@ -37,6 +38,7 @@ type Config struct { Odom OdomConfig Lowstate LowstateConfig Network NetworkConfig + Traces TracesConfig SessionRotateInterval time.Duration Upload UploadConfig Retention RetentionConfig @@ -142,6 +144,16 @@ type NetworkConfig struct { DataFile string } +// TracesConfig configures polling a co-located OM1 process's Prometheus +// trace-export endpoint (see internal/traces). +type TracesConfig struct { + Enabled bool + URL string + PollInterval time.Duration + OutputFile string + CursorFile string +} + // RecordingsDir is the root under which sessions are created. The session // package needs it before Load runs, because it decides the session directory // name -- which depends on whether the clock can be trusted yet. @@ -227,6 +239,13 @@ func Load(sessionDir string) Config { PollInterval: envDuration("NET_POLL_INTERVAL", 5*time.Second), DataFile: filepath.Join(sessionDir, "network_status.csv"), }, + Traces: TracesConfig{ + Enabled: envBool("TRACES_ENABLED", false), + URL: envStr("TRACES_URL", "http://localhost:9090/metrics"), + PollInterval: envDuration("TRACES_POLL_INTERVAL", 30*time.Second), + OutputFile: filepath.Join(sessionDir, "traces.jsonl"), + CursorFile: filepath.Join(RecordingsDir(), ".traces_cursor"), + }, SessionRotateInterval: envDuration("SESSION_ROTATE_INTERVAL", 5*time.Minute), Upload: UploadConfig{ Enabled: envBool("UPLOAD_ENABLED", true), @@ -346,6 +365,15 @@ func (c NetworkConfig) NetworkStreamConfig() network.Config { } } +func (c TracesConfig) TraceStreamConfig() traces.Config { + return traces.Config{ + URL: c.URL, + PollInterval: c.PollInterval, + OutputFile: c.OutputFile, + CursorFile: c.CursorFile, + } +} + func envStr(key, defaultValue string) string { if value := os.Getenv(key); value != "" { return value diff --git a/go.mod b/go.mod index 94224d9..03aa10f 100644 --- a/go.mod +++ b/go.mod @@ -1,15 +1,18 @@ module om1-telemetry -go 1.25 +go 1.25.0 require ( github.com/klauspost/compress v1.18.6 - github.com/stretchr/testify v1.11.1 - golang.org/x/sys v0.35.0 + github.com/prometheus/common v0.71.0 + github.com/stretchr/testify v1.12.1 + golang.org/x/sys v0.47.0 gopkg.in/yaml.v3 v3.0.1 ) require ( - github.com/davecgh/go-spew v1.1.1 // indirect - github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + go.yaml.in/yaml/v3 v3.0.5 // indirect + google.golang.org/protobuf v1.36.12 // indirect ) diff --git a/go.sum b/go.sum index 2f5977f..32e644c 100644 --- a/go.sum +++ b/go.sum @@ -1,13 +1,23 @@ -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= -github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= -github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= -github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= -golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.71.0 h1:9KDAKb7Mj3HEVKyFCK6Dc/HIwlBzZIN2l7/lrHl3KK8= +github.com/prometheus/common v0.71.0/go.mod h1:CLJ5H8TEsGX8bl31BdMkfhIZ+QmZ9tBPPotUxUbfcmk= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/internal/audio/stream.go b/internal/audio/stream.go index 69d13ce..038fa71 100644 --- a/internal/audio/stream.go +++ b/internal/audio/stream.go @@ -68,6 +68,17 @@ type segmentTarget struct { sessionStart time.Time timestampsFile string framesWrite *recordutil.FrameCSVWriter + + // ready closes once this target's first segment lands (or definitively + // won't, on relocation failure) -- see WaitSegment. + ready chan struct{} + readyOnce sync.Once +} + +// markReady unblocks any WaitSegment callers for this target. Safe to call +// more than once (a target may receive several segments, or none). +func (t *segmentTarget) markReady() { + t.readyOnce.Do(func() { close(t.ready) }) } func New(cfg Config) *AudioRTSPStream { @@ -127,7 +138,7 @@ func newSegmentTarget(sessionStart time.Time, timestampsFile, framesFile string) if framesFile != "" { fw = recordutil.NewFrameCSVWriter(framesFile) } - return &segmentTarget{sessionStart: sessionStart, timestampsFile: timestampsFile, framesWrite: fw} + return &segmentTarget{sessionStart: sessionStart, timestampsFile: timestampsFile, framesWrite: fw, ready: make(chan struct{})} } // targetFor returns the target whose session started most recently at or @@ -147,6 +158,29 @@ func (a *AudioRTSPStream) targetFor(startWallClock time.Time) *segmentTarget { return best } +// WaitSegment blocks until the target for sessionStart has relocated its +// first segment, or ctx ends; reports whether a segment actually arrived. +func (a *AudioRTSPStream) WaitSegment(ctx context.Context, sessionStart time.Time) bool { + a.targetMu.Lock() + var target *segmentTarget + for _, t := range a.targets { + if t.sessionStart.Equal(sessionStart) { + target = t + break + } + } + a.targetMu.Unlock() + if target == nil { + return false + } + select { + case <-target.ready: + return true + case <-ctx.Done(): + return false + } +} + func (a *AudioRTSPStream) loop(ctx context.Context) { defer close(a.done) a.ensureTarget() @@ -175,7 +209,6 @@ func (a *AudioRTSPStream) record(ctx context.Context) error { } start := time.Now() - startMono := clock.MonoNs() pattern := filepath.Join(a.cfg.ScratchDir, "%Y%m%dT%H%M%S.ogg") cmd := exec.CommandContext(ctx, "ffmpeg", @@ -238,7 +271,7 @@ func (a *AudioRTSPStream) record(ctx context.Context) error { // Must fully drain stdout before calling Wait: Wait closes the pipe once // it reaps the process, and reading after that races the close. - a.watchSegments(stdout, start, startMono) + a.watchSegments(stdout) waitErr := cmd.Wait() close(hbStop) @@ -251,14 +284,14 @@ func (a *AudioRTSPStream) record(ctx context.Context) error { // (filename,start_seconds,end_seconds), and relocates, indexes, and // extracts frame timestamps for each segment as it completes. Returns once // ffmpeg closes stdout, i.e. once the process exits. -func (a *AudioRTSPStream) watchSegments(stdout io.Reader, processStart time.Time, processStartMonoNs int64) { +func (a *AudioRTSPStream) watchSegments(stdout io.Reader) { scanner := bufio.NewScanner(stdout) for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) if line == "" { continue } - name, startSeconds, ok := parseSegmentListLine(line) + name, startSeconds, endSeconds, ok := parseSegmentListLine(line) if !ok { slog.Warn("audio: unrecognized segment_list line", "line", line) continue @@ -266,7 +299,8 @@ func (a *AudioRTSPStream) watchSegments(stdout io.Reader, processStart time.Time // ffmpeg's segment_list reports the filename as written to its // pattern, without the directory -- join it back to ScratchDir. scratchFile := filepath.Join(a.cfg.ScratchDir, name) - a.finishSegment(scratchFile, startSeconds, processStart, processStartMonoNs) + duration := time.Duration((endSeconds - startSeconds) * float64(time.Second)) + a.finishSegment(scratchFile, duration, time.Now(), clock.MonoNs()) } if err := scanner.Err(); err != nil { slog.Warn("audio: segment_list read error", "err", err) @@ -275,24 +309,27 @@ func (a *AudioRTSPStream) watchSegments(stdout io.Reader, processStart time.Time // parseSegmentListLine parses one "-segment_list_type csv" line: // filename,start_seconds,end_seconds. -func parseSegmentListLine(line string) (file string, startSeconds float64, ok bool) { +func parseSegmentListLine(line string) (file string, startSeconds, endSeconds float64, ok bool) { parts := strings.Split(line, ",") - if len(parts) < 2 { - return "", 0, false + if len(parts) < 3 { + return "", 0, 0, false } start, err := strconv.ParseFloat(strings.TrimSpace(parts[1]), 64) if err != nil { - return "", 0, false + return "", 0, 0, false + } + end, err := strconv.ParseFloat(strings.TrimSpace(parts[2]), 64) + if err != nil { + return "", 0, 0, false } - return strings.TrimSpace(parts[0]), start, true + return strings.TrimSpace(parts[0]), start, end, true } -// finishSegment relocates a just-closed segment from the scratch directory -// into the current target's session directory, indexes it, and kicks off -// its (async) frame extraction. -func (a *AudioRTSPStream) finishSegment(scratchFile string, startSeconds float64, processStart time.Time, processStartMonoNs int64) { - startWallClock := processStart.Add(time.Duration(startSeconds * float64(time.Second))) - startMonoNs := processStartMonoNs + int64(startSeconds*float64(time.Second)) +// finishSegment relocates a just-closed segment into the current target's +// session directory, indexes it, and kicks off its (async) frame extraction. +func (a *AudioRTSPStream) finishSegment(scratchFile string, duration time.Duration, observedNow time.Time, observedMonoNs int64) { + startWallClock := observedNow.Add(-duration) + startMonoNs := observedMonoNs - duration.Nanoseconds() target := a.targetFor(startWallClock) finalFile := filepath.Join(filepath.Dir(target.timestampsFile), @@ -300,6 +337,7 @@ func (a *AudioRTSPStream) finishSegment(scratchFile string, startSeconds float64 if err := os.Rename(scratchFile, finalFile); err != nil { slog.Error("audio: could not relocate segment", "err", err) + target.markReady() return } @@ -308,6 +346,7 @@ func (a *AudioRTSPStream) finishSegment(scratchFile string, startSeconds float64 "file", target.timestampsFile, "err", err) } slog.Info("audio segment closed", "file", finalFile) + target.markReady() if target.framesWrite == nil { return diff --git a/internal/audio/stream_test.go b/internal/audio/stream_test.go index 94d457f..01127dd 100644 --- a/internal/audio/stream_test.go +++ b/internal/audio/stream_test.go @@ -1,6 +1,7 @@ package audio import ( + "context" "fmt" "os" "path/filepath" @@ -74,18 +75,19 @@ func TestStop_idempotent(t *testing.T) { } func TestParseSegmentListLine_parsesFilenameAndStart(t *testing.T) { - file, start, ok := parseSegmentListLine("20260826T185832.ogg,12.500000,15.000000") + file, start, end, ok := parseSegmentListLine("20260826T185832.ogg,12.500000,15.000000") require.True(t, ok) require.Equal(t, "20260826T185832.ogg", file) require.InDelta(t, 12.5, start, 1e-9) + require.InDelta(t, 15.0, end, 1e-9) } func TestParseSegmentListLine_rejectsMalformedLine(t *testing.T) { - _, _, ok := parseSegmentListLine("not,a,valid,,line") + _, _, _, ok := parseSegmentListLine("not,a,valid,,line") require.False(t, ok, "a non-numeric start field must be rejected") - _, _, ok = parseSegmentListLine("onlyonefield") - require.False(t, ok, "a line missing the start field must be rejected") + _, _, _, ok = parseSegmentListLine("onlyonefield") + require.False(t, ok, "a line missing the start/end fields must be rejected") } func TestFinishSegment_relocatesFileIndexesItAndPrefixesWithStem(t *testing.T) { @@ -103,7 +105,8 @@ func TestFinishSegment_relocatesFileIndexesItAndPrefixesWithStem(t *testing.T) { processStart := time.Unix(1_800_000_000, 0) stream.Rotate(processStart, filepath.Join(sessionDir, "audio_timestamps.csv"), "") // no frames file: skip async ffprobe - stream.finishSegment(scratchFile, 12.5, processStart, 5_000_000_000) + segmentStart := processStart.Add(12500 * time.Millisecond) + stream.finishSegment(scratchFile, 0, segmentStart, 5_000_000_000) wantFinal := filepath.Join(sessionDir, "audio_20260826T185832.ogg") require.FileExists(t, wantFinal, "the segment must be relocated with its stem prefixed") @@ -113,8 +116,7 @@ func TestFinishSegment_relocatesFileIndexesItAndPrefixesWithStem(t *testing.T) { require.NoError(t, err) content := string(data) require.Contains(t, content, "recording_start_unix_ns,segment_file,mono_ns") - wantStartUnixNs := processStart.Add(12500 * time.Millisecond).UnixNano() - require.Contains(t, content, fmt.Sprintf("%d", wantStartUnixNs), "the indexed start time must include the segment's offset into the stream") + require.Contains(t, content, fmt.Sprintf("%d", segmentStart.UnixNano()), "the indexed start time must match the segment's own start") require.Contains(t, content, "audio_20260826T185832.ogg") } @@ -132,10 +134,9 @@ func TestWatchSegments_joinsBareFilenameFromSegmentListWithScratchDir(t *testing OutputFile: filepath.Join(t.TempDir(), "audio.ogg"), ScratchDir: scratchDir, }) - processStart := time.Unix(1_800_000_000, 0) - stream.Rotate(processStart, filepath.Join(sessionDir, "audio_timestamps.csv"), "") + stream.Rotate(time.Now(), filepath.Join(sessionDir, "audio_timestamps.csv"), "") - stream.watchSegments(strings.NewReader("20260826T185832.ogg,0.000000,3.000000\n"), processStart, 0) + stream.watchSegments(strings.NewReader("20260826T185832.ogg,0.000000,3.000000\n")) require.FileExists(t, filepath.Join(sessionDir, "audio_20260826T185832.ogg"), "watchSegments must join the segment_list's bare filename with ScratchDir before relocating") @@ -198,6 +199,130 @@ func TestTargetFor_attributesByOwnTimestampEvenWhenRotateRacesAhead(t *testing.T require.NoFileExists(t, filepath.Join(sessionN1Dir, "audio_seg.ogg")) } +// Regression: a segment starting just after a rotation boundary must land +// in the new session even after a long-running process's clock has drifted. +func TestFinishSegment_notMisledByAccumulatedProcessDrift(t *testing.T) { + scratchDir := t.TempDir() + sessionNDir := t.TempDir() + sessionN1Dir := t.TempDir() + + sessionNStart := time.Unix(1_800_000_000, 0) + sessionN1Start := sessionNStart.Add(5 * time.Minute) + + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "audio.ogg"), + TimestampsFile: filepath.Join(sessionNDir, "audio_timestamps.csv"), + SessionStart: sessionNStart, + ScratchDir: scratchDir, + }) + stream.ensureTarget() + stream.Rotate(sessionN1Start, filepath.Join(sessionN1Dir, "audio_timestamps.csv"), "") + + segmentDuration := 5 * time.Minute + trueSegmentStart := sessionN1Start.Add(1 * time.Second) + observedClose := trueSegmentStart.Add(segmentDuration) + + seg := filepath.Join(scratchDir, "seg.ogg") + require.NoError(t, os.WriteFile(seg, []byte("a"), 0o644)) + + stream.finishSegment(seg, segmentDuration, observedClose, 0) + + require.FileExists(t, filepath.Join(sessionN1Dir, "audio_seg.ogg"), + "a segment observed to close one full span after starting just past the rotation boundary must land in the new session") + require.NoFileExists(t, filepath.Join(sessionNDir, "audio_seg.ogg")) +} + +// Regression: an uploader triggered right on rotation must be able to wait +// for the session's own segment to land instead of racing finishSegment. +func TestWaitSegment_blocksUntilFinishSegmentRelocatesThatSessionsSegment(t *testing.T) { + scratchDir := t.TempDir() + sessionDir := t.TempDir() + sessionStart := time.Unix(1_800_000_000, 0) + + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "audio.ogg"), + TimestampsFile: filepath.Join(sessionDir, "audio_timestamps.csv"), + SessionStart: sessionStart, + ScratchDir: scratchDir, + }) + stream.ensureTarget() + + waited := make(chan bool, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + waited <- stream.WaitSegment(ctx, sessionStart) + }() + + select { + case <-waited: + t.Fatal("WaitSegment returned before finishSegment relocated anything") + case <-time.After(20 * time.Millisecond): + } + + seg := filepath.Join(scratchDir, "seg.ogg") + require.NoError(t, os.WriteFile(seg, []byte("a"), 0o644)) + stream.finishSegment(seg, 0, sessionStart, 0) + + require.True(t, <-waited, "WaitSegment must return true once finishSegment relocates the session's segment") +} + +// A target that never receives a segment (RTSP never delivered data for +// that session) must not hang WaitSegment past its context. +func TestWaitSegment_returnsFalseWhenContextEndsFirst(t *testing.T) { + sessionStart := time.Unix(1_800_000_000, 0) + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "audio.ogg"), + SessionStart: sessionStart, + ScratchDir: t.TempDir(), + }) + stream.ensureTarget() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + require.False(t, stream.WaitSegment(ctx, sessionStart)) +} + +// A relocation failure must still unblock waiters rather than leaving them +// to time out every time. +func TestWaitSegment_returnsTrueImmediatelyAfterRelocationFailure(t *testing.T) { + sessionDir := t.TempDir() + sessionStart := time.Unix(1_800_000_000, 0) + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "audio.ogg"), + TimestampsFile: filepath.Join(sessionDir, "audio_timestamps.csv"), + SessionStart: sessionStart, + ScratchDir: t.TempDir(), + }) + stream.ensureTarget() + + // scratchFile does not exist, so os.Rename inside finishSegment fails. + stream.finishSegment(filepath.Join(t.TempDir(), "missing.ogg"), 0, sessionStart, 0) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + require.True(t, stream.WaitSegment(ctx, sessionStart), + "a relocation failure must still mark the target ready so callers don't wait out the full timeout") +} + +// No target was ever installed for this sessionStart. +func TestWaitSegment_returnsFalseForUnknownSessionStart(t *testing.T) { + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "audio.ogg"), + ScratchDir: t.TempDir(), + }) + stream.ensureTarget() + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + require.False(t, stream.WaitSegment(ctx, time.Unix(1_900_000_000, 0))) +} + func TestAppendSegmentEntry_emptyPath_isNoOp(t *testing.T) { err := appendSegmentEntry("", time.Now(), 0, "/data/audio.ogg") require.NoError(t, err) diff --git a/internal/compress/zstd.go b/internal/compress/zstd.go index 67a9bdc..b8a941c 100644 --- a/internal/compress/zstd.go +++ b/internal/compress/zstd.go @@ -15,6 +15,10 @@ var wholeFileTargets = []string{ "lowstate_frames.bin", "odom_frames.bin", "lidar_scans.bin", + // Per-frame timestamp CSVs: repetitive numeric text, and large enough at + // lowstate's 400Hz to be worth compressing like their sibling data files. + "lowstate_timestamps.csv", + "odom_timestamps.csv", } // WholeFiles replaces each of wholeFileTargets with a zstd-compressed copy. diff --git a/internal/compress/zstd_test.go b/internal/compress/zstd_test.go index e1f2a3d..d0f485d 100644 --- a/internal/compress/zstd_test.go +++ b/internal/compress/zstd_test.go @@ -32,15 +32,20 @@ func TestWholeFiles_roundTripsAndDeletesOriginal(t *testing.T) { require.Equal(t, original, decompressed) } -func TestWholeFiles_compressesAllThreeTargets(t *testing.T) { +func TestWholeFiles_compressesAllTargets(t *testing.T) { dir := t.TempDir() writeFile(t, dir, "lowstate_frames.bin", []byte("aaaaaaaaaaaaaaaa")) writeFile(t, dir, "odom_frames.bin", []byte("bbbbbbbbbbbbbbbb")) writeFile(t, dir, "lidar_scans.bin", []byte("cccccccccccccccc")) + writeFile(t, dir, "lowstate_timestamps.csv", []byte("dddddddddddddddd")) + writeFile(t, dir, "odom_timestamps.csv", []byte("eeeeeeeeeeeeeeee")) require.NoError(t, WholeFiles(dir)) - for _, name := range []string{"lowstate_frames.zstd", "odom_frames.zstd", "lidar_scans.zstd"} { + for _, name := range []string{ + "lowstate_frames.zstd", "odom_frames.zstd", "lidar_scans.zstd", + "lowstate_timestamps.zstd", "odom_timestamps.zstd", + } { require.FileExists(t, filepath.Join(dir, name)) } } @@ -73,4 +78,5 @@ func TestCompressWholeFile_isIdempotentOnRetry(t *testing.T) { func TestZstdName(t *testing.T) { require.Equal(t, "lowstate_frames.zstd", zstdName("lowstate_frames.bin")) + require.Equal(t, "lowstate_timestamps.zstd", zstdName("lowstate_timestamps.csv")) } diff --git a/internal/ddscore/participant.go b/internal/ddscore/participant.go index af0a0c3..e143e99 100644 --- a/internal/ddscore/participant.go +++ b/internal/ddscore/participant.go @@ -32,30 +32,75 @@ type Participant struct { entity Entity } +// createTimeout bounds how long a dds_create_* cgo call is given to return: +// the same degraded-domain conditions that make dds_delete hang can also make entity creation hang. +const createTimeout = 5 * time.Second + +// boundedCreate runs create with a bound of createTimeout, abandoning (and +// leaking) it on timeout instead of blocking its caller forever. +func boundedCreate(create func() C.dds_entity_t) (Entity, error) { + done := make(chan C.dds_entity_t, 1) + go func() { done <- create() }() + select { + case ret := <-done: + return check(ret) + case <-time.After(createTimeout): + return 0, errors.New("timed out") + } +} + func NewParticipant(domainID uint32) (*Participant, error) { - e, err := check(C.dds_create_participant(C.dds_domainid_t(domainID), nil, nil)) + e, err := boundedCreate(func() C.dds_entity_t { + return C.dds_create_participant(C.dds_domainid_t(domainID), nil, nil) + }) if err != nil { return nil, fmt.Errorf("dds_create_participant(domain=%d): %w", domainID, err) } return &Participant{entity: e}, nil } -func (p *Participant) Close() error { - if ret := C.dds_delete(C.dds_entity_t(p.entity)); ret < 0 { - return fmt.Errorf("dds_delete(participant): %w", retcodeError(ret)) +// closeTimeout bounds how long dds_delete is given to return: a never-connected entity +// can hang the call forever in native code, uninterruptible from Go. +const closeTimeout = 5 * time.Second + +// closeEntity deletes entity, abandoning (and leaking) the underlying +// cgo call if it does not return within closeTimeout. +func closeEntity(entity C.dds_entity_t, what string) error { + done := make(chan error, 1) + go func() { + if ret := C.dds_delete(entity); ret < 0 { + done <- retcodeError(ret) + return + } + done <- nil + }() + select { + case err := <-done: + if err != nil { + return fmt.Errorf("dds_delete(%s): %w", what, err) + } + return nil + case <-time.After(closeTimeout): + return fmt.Errorf("dds_delete(%s): timed out after %s, abandoning", what, closeTimeout) } - return nil } +func (p *Participant) Close() error { + return closeEntity(C.dds_entity_t(p.entity), "participant") +} + +// CreateTopic creates topic. cname is freed inside the goroutine after the cgo call +// returns, not via defer, to avoid a use-after-free if a timed-out call eventually runs. func (p *Participant) CreateTopic(name string, descriptor unsafe.Pointer) (Entity, error) { cname := C.CString(name) - defer C.free(unsafe.Pointer(cname)) - - e, err := check(C.dds_create_topic( - C.dds_entity_t(p.entity), - (*C.dds_topic_descriptor_t)(descriptor), - cname, nil, nil, - )) + e, err := boundedCreate(func() C.dds_entity_t { + defer C.free(unsafe.Pointer(cname)) + return C.dds_create_topic( + C.dds_entity_t(p.entity), + (*C.dds_topic_descriptor_t)(descriptor), + cname, nil, nil, + ) + }) if err != nil { return 0, fmt.Errorf("dds_create_topic(%q): %w", name, err) } @@ -63,7 +108,9 @@ func (p *Participant) CreateTopic(name string, descriptor unsafe.Pointer) (Entit } func (p *Participant) CreateReader(topic Entity) (Entity, error) { - e, err := check(C.dds_create_reader(C.dds_entity_t(p.entity), C.dds_entity_t(topic), nil, nil)) + e, err := boundedCreate(func() C.dds_entity_t { + return C.dds_create_reader(C.dds_entity_t(p.entity), C.dds_entity_t(topic), nil, nil) + }) if err != nil { return 0, fmt.Errorf("dds_create_reader: %w", err) } @@ -71,7 +118,9 @@ func (p *Participant) CreateReader(topic Entity) (Entity, error) { } func (p *Participant) CreateWriter(topic Entity) (Entity, error) { - e, err := check(C.dds_create_writer(C.dds_entity_t(p.entity), C.dds_entity_t(topic), nil, nil)) + e, err := boundedCreate(func() C.dds_entity_t { + return C.dds_create_writer(C.dds_entity_t(p.entity), C.dds_entity_t(topic), nil, nil) + }) if err != nil { return 0, fmt.Errorf("dds_create_writer: %w", err) } @@ -107,38 +156,53 @@ type WaitSet struct { cond Entity } +// NewWaitSet builds a waitset in its own goroutine, bounded by createTimeout: +// any of its three cgo calls can hang on a degraded domain, same as boundedCreate. func NewWaitSet(participant *Participant, reader Entity) (*WaitSet, error) { - ws, err := check(C.dds_create_waitset(C.dds_entity_t(participant.entity))) - if err != nil { - return nil, fmt.Errorf("dds_create_waitset: %w", err) + type result struct { + ws *WaitSet + err error } - - cond, err := check(C.dds_create_readcondition(C.dds_entity_t(reader), C.DDS_ANY_STATE)) - if err != nil { - _ = C.dds_delete(C.dds_entity_t(ws)) - return nil, fmt.Errorf("dds_create_readcondition: %w", err) + done := make(chan result, 1) + go func() { + ws, err := check(C.dds_create_waitset(C.dds_entity_t(participant.entity))) + if err != nil { + done <- result{nil, fmt.Errorf("dds_create_waitset: %w", err)} + return + } + cond, err := check(C.dds_create_readcondition(C.dds_entity_t(reader), C.DDS_ANY_STATE)) + if err != nil { + _ = C.dds_delete(C.dds_entity_t(ws)) + done <- result{nil, fmt.Errorf("dds_create_readcondition: %w", err)} + return + } + if ret := C.dds_waitset_attach(C.dds_entity_t(ws), C.dds_entity_t(cond), C.dds_attach_t(cond)); ret < 0 { + _ = C.dds_delete(C.dds_entity_t(ws)) + done <- result{nil, fmt.Errorf("dds_waitset_attach: %w", retcodeError(C.dds_return_t(ret)))} + return + } + done <- result{&WaitSet{entity: ws, cond: cond}, nil} + }() + select { + case r := <-done: + return r.ws, r.err + case <-time.After(createTimeout): + return nil, fmt.Errorf("dds_create_waitset(construction): timed out after %s, abandoning", createTimeout) } - - if ret := C.dds_waitset_attach(C.dds_entity_t(ws), C.dds_entity_t(cond), C.dds_attach_t(cond)); ret < 0 { - _ = C.dds_delete(C.dds_entity_t(ws)) - return nil, fmt.Errorf("dds_waitset_attach: %w", retcodeError(ret)) - } - - return &WaitSet{entity: ws, cond: cond}, nil } +// Wait blocks for up to timeout; on error it sleeps for timeout too, since +// callers retry on error with no backoff of their own. func (w *WaitSet) Wait(timeout time.Duration) (bool, error) { var xs [1]C.dds_attach_t ret := C.dds_waitset_wait(C.dds_entity_t(w.entity), &xs[0], 1, C.dds_duration_t(timeout.Nanoseconds())) if ret < 0 { + time.Sleep(timeout) return false, fmt.Errorf("dds_waitset_wait: %w", retcodeError(ret)) } return ret > 0, nil } func (w *WaitSet) Close() error { - if ret := C.dds_delete(C.dds_entity_t(w.entity)); ret < 0 { - return fmt.Errorf("dds_delete(waitset): %w", retcodeError(ret)) - } - return nil + return closeEntity(C.dds_entity_t(w.entity), "waitset") } diff --git a/internal/depth/stream.go b/internal/depth/stream.go index 5a91d24..e9544a8 100644 --- a/internal/depth/stream.go +++ b/internal/depth/stream.go @@ -35,6 +35,10 @@ type DepthStream struct { cancel context.CancelFunc wg sync.WaitGroup + // reconnect asks a running record() to tear down and recreate its DDS + // subscription. Buffered so a request never blocks; see Reconnect. + reconnect chan struct{} + filesMu sync.Mutex files *outputFiles } @@ -49,7 +53,7 @@ type outputFiles struct { } func New(cfg Config) *DepthStream { - return &DepthStream{cfg: cfg} + return &DepthStream{cfg: cfg, reconnect: make(chan struct{}, 1)} } func (d *DepthStream) Start() { @@ -92,6 +96,15 @@ func (d *DepthStream) Rotate(dataFile, timestampsFile string) error { return nil } +// Reconnect tears down and recreates the DDS subscription without touching +// output files. Non-blocking; see heartbeat.Monitor.RegisterRecoverable. +func (d *DepthStream) Reconnect() { + select { + case d.reconnect <- struct{}{}: + default: + } +} + func (d *DepthStream) loop(ctx context.Context) { defer d.wg.Done() for ctx.Err() == nil { @@ -146,6 +159,9 @@ func (d *DepthStream) record(ctx context.Context) error { case <-ctx.Done(): d.flush() return nil + case <-d.reconnect: + d.flush() + return fmt.Errorf("reconnect requested") case <-syncTicker.C: d.flush() case sample, ok := <-receiver: diff --git a/internal/heartbeat/heartbeat.go b/internal/heartbeat/heartbeat.go index 3d577ab..b3a704f 100644 --- a/internal/heartbeat/heartbeat.go +++ b/internal/heartbeat/heartbeat.go @@ -12,6 +12,9 @@ import ( type Monitor struct { streams sync.Map interval time.Duration + + stuckThreshold int + stuckHandler func(stream string) } type state struct { @@ -22,19 +25,60 @@ type state struct { lastTime time.Time registered time.Time warned bool + + // reconnect, if set, is called (in its own goroutine) every time a check + // finds the stream still broken -- see RegisterRecoverable. + reconnect func() + reconnecting atomic.Bool + + // consecutiveBad counts checks in a row that found the stream broken; + // reset the moment it recovers. Drives the stuck-handler escalation. + consecutiveBad int + escalated bool + + // everWorked latches true on the first tick ever; only a stream that + // worked and got stuck is eligible to escalate. + everWorked bool } func NewMonitor(checkInterval time.Duration) *Monitor { return &Monitor{interval: checkInterval} } +// SetStuckThreshold calls handler once, from check's goroutine, if a +// recoverable stream is still broken after threshold consecutive checks. +func (m *Monitor) SetStuckThreshold(threshold int, handler func(stream string)) { + m.stuckThreshold = threshold + m.stuckHandler = handler +} + func (m *Monitor) Register(name string, expectedHz float64) { + m.register(name, expectedHz, nil) +} + +// RegisterRecoverable is like Register, but reconnect is called once per +// check interval for as long as the stream stays broken. +func (m *Monitor) RegisterRecoverable(name string, expectedHz float64, reconnect func()) { + m.register(name, expectedHz, reconnect) +} + +// register (re)creates name's tracking state, carrying consecutiveBad and +// escalated over from any existing entry so periodic re-registration can't reset them. +func (m *Monitor) register(name string, expectedHz float64, reconnect func()) { now := time.Now() - m.streams.Store(name, &state{ + s := &state{ expectedHz: expectedHz, lastTime: now, registered: now, - }) + reconnect: reconnect, + } + if prev, ok := m.streams.Load(name); ok { + p := prev.(*state) + s.consecutiveBad = p.consecutiveBad + s.escalated = p.escalated + s.everWorked = p.everWorked + } + m.streams.Store(name, s) } // Unregister stops tracking name, so a deliberately stopped stream isn't flagged NOT WORKING. @@ -72,6 +116,9 @@ func (m *Monitor) check() { s := v.(*state) current := s.ticks.Load() + if current > 0 { + s.everWorked = true + } delta := current - s.lastTicks elapsed := now.Sub(s.lastTime).Seconds() var rate float64 @@ -117,6 +164,27 @@ func (m *Monitor) check() { s.warned = false } + if bad && s.reconnect != nil && s.reconnecting.CompareAndSwap(false, true) { + reconnect := s.reconnect + go func() { + defer s.reconnecting.Store(false) + reconnect() + }() + } + + if bad && s.reconnect != nil && s.everWorked { + s.consecutiveBad++ + if m.stuckThreshold > 0 && s.consecutiveBad >= m.stuckThreshold && !s.escalated { + s.escalated = true + slog.Error("recorder still broken after repeated reconnects; escalating", + "stream", name, "consecutive_failures", s.consecutiveBad) + m.stuckHandler(name) + } + } else { + s.consecutiveBad = 0 + s.escalated = false + } + s.lastTicks = current s.lastTime = now return true diff --git a/internal/heartbeat/heartbeat_test.go b/internal/heartbeat/heartbeat_test.go index 77d086a..652060a 100644 --- a/internal/heartbeat/heartbeat_test.go +++ b/internal/heartbeat/heartbeat_test.go @@ -2,6 +2,7 @@ package heartbeat import ( "context" + "sync/atomic" "testing" "time" @@ -142,3 +143,168 @@ func TestCheck_noWarnWhenTicksArrive(t *testing.T) { v, _ := mon.streams.Load("lidar") require.False(t, v.(*state).warned, "should not warn when stream is ticking at expected rate") } + +// Reproduces the real gap this closes: a DDS reader that never received the +// writer's post-restart discovery burst just sits there silently forever. +func TestRegisterRecoverable_callsReconnectWhenBroken(t *testing.T) { + mon := NewMonitor(5 * time.Millisecond) + called := make(chan struct{}, 1) + mon.RegisterRecoverable("pointcloud", 10, func() { + select { + case called <- struct{}{}: + default: + } + }) + + time.Sleep(20 * time.Millisecond) + mon.check() + + select { + case <-called: + case <-time.After(2 * time.Second): + t.Fatal("reconnect was not called for a stream that never received a message") + } +} + +func TestRegisterRecoverable_doesNotCallReconnectWhileHealthy(t *testing.T) { + mon := NewMonitor(5 * time.Millisecond) + called := make(chan struct{}, 1) + mon.RegisterRecoverable("lidar", 10, func() { + select { + case called <- struct{}{}: + default: + } + }) + + for range 100 { + mon.Tick("lidar") + } + time.Sleep(20 * time.Millisecond) + mon.check() + + select { + case <-called: + t.Fatal("reconnect must not be called for a healthy stream") + case <-time.After(50 * time.Millisecond): + } +} + +// check() must not pile up a new reconnect goroutine on top of one still running. +func TestRegisterRecoverable_doesNotOverlapReconnects(t *testing.T) { + mon := NewMonitor(5 * time.Millisecond) + var calls atomic.Int32 + release := make(chan struct{}) + mon.RegisterRecoverable("odom", 30, func() { + calls.Add(1) + <-release + }) + + time.Sleep(20 * time.Millisecond) + mon.check() + mon.check() + + time.Sleep(20 * time.Millisecond) + require.Equal(t, int32(1), calls.Load(), "a reconnect already in flight must not be duplicated") + close(release) +} + +// Reader-level reconnects can loop forever without ever restoring data if +// the real fault is upstream of the reader, so a stream that worked and +// then got stuck must escalate exactly once, not on every remaining bad check. +func TestSetStuckThreshold_escalatesOnceAtThreshold(t *testing.T) { + mon := NewMonitor(5 * time.Millisecond) + mon.RegisterRecoverable("pointcloud", 10, func() {}) + mon.Tick("pointcloud") // it worked once, then got stuck + + var stuckCalls int + var lastStream string + mon.SetStuckThreshold(2, func(stream string) { + stuckCalls++ + lastStream = stream + }) + + time.Sleep(20 * time.Millisecond) + mon.check() // consumes the initial tick; establishes the baseline + require.Zero(t, stuckCalls) + + mon.check() + require.Zero(t, stuckCalls, "must not escalate before reaching the threshold") + + mon.check() + require.Equal(t, 1, stuckCalls, "must escalate once the threshold is reached") + require.Equal(t, "pointcloud", lastStream) + + mon.check() + require.Equal(t, 1, stuckCalls, "must not escalate again while still broken") +} + +// A stream with no hardware ever attached (e.g. no camera connected) never +// ticks at all; restarting it can never help, so it must never escalate no +// matter how long it stays "broken" -- otherwise the process would restart +// itself forever because of a permanently, expectedly absent device. +func TestSetStuckThreshold_neverEscalatesAStreamThatNeverWorked(t *testing.T) { + mon := NewMonitor(5 * time.Millisecond) + mon.RegisterRecoverable("depth", 10, func() {}) + + var stuckCalls int + mon.SetStuckThreshold(2, func(string) { stuckCalls++ }) + + time.Sleep(20 * time.Millisecond) + for range 20 { + mon.check() + } + require.Zero(t, stuckCalls, "a stream that has never ticked must never escalate") +} + +func TestSetStuckThreshold_recoveryResetsTheCount(t *testing.T) { + mon := NewMonitor(5 * time.Millisecond) + mon.RegisterRecoverable("odom", 10, func() {}) + + var stuckCalls int + mon.SetStuckThreshold(2, func(string) { stuckCalls++ }) + + for range 50 { + mon.Tick("odom") + } + time.Sleep(20 * time.Millisecond) + mon.check() // consumes the ticks; establishes the baseline + + mon.check() + require.Zero(t, stuckCalls, "one failure right after recovering must not immediately re-escalate") +} + +// Session rotation re-registers every stream every few minutes; that must +// not reset a stream's progress toward the stuck threshold, or a stream +// broken for hours would never escalate. +func TestSetStuckThreshold_survivesReRegistration(t *testing.T) { + mon := NewMonitor(5 * time.Millisecond) + mon.RegisterRecoverable("pointcloud", 10, func() {}) + mon.Tick("pointcloud") + + var stuckCalls int + mon.SetStuckThreshold(3, func(string) { stuckCalls++ }) + + time.Sleep(20 * time.Millisecond) + mon.check() // consumes the initial tick + mon.check() // consecutiveBad=1 + require.Zero(t, stuckCalls, "must not escalate before reaching the threshold") + + mon.RegisterRecoverable("pointcloud", 10, func() {}) + time.Sleep(20 * time.Millisecond) + mon.check() // consecutiveBad=2 despite the re-registration in between + require.Zero(t, stuckCalls) + mon.check() // consecutiveBad=3 + require.Equal(t, 1, stuckCalls, "re-registration must not reset progress toward the stuck threshold") +} + +func TestSetStuckThreshold_disabledByDefault(t *testing.T) { + mon := NewMonitor(5 * time.Millisecond) + mon.RegisterRecoverable("lidar", 10, func() {}) + + time.Sleep(20 * time.Millisecond) + require.NotPanics(t, func() { + for range 10 { + mon.check() + } + }) +} diff --git a/internal/lidar/reconnect_test.go b/internal/lidar/reconnect_test.go new file mode 100644 index 0000000..d657567 --- /dev/null +++ b/internal/lidar/reconnect_test.go @@ -0,0 +1,98 @@ +package lidar + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "om1-telemetry/internal/ddscore" + "om1-telemetry/internal/lidar/ddstest" +) + +func countDataLines(t *testing.T, path string) int { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return 0 + } + t.Fatalf("read %s: %v", path, err) + } + n := 0 + for _, l := range strings.Split(strings.TrimSpace(string(data)), "\n") { + if l != "" { + n++ + } + } + return n +} + +func waitForLines(t *testing.T, path string, want int, timeout time.Duration, msgAndArgs ...any) { + t.Helper() + deadline := time.After(timeout) + for { + if countDataLines(t, path) >= want { + return + } + select { + case <-deadline: + t.Fatalf("timed out waiting for %d line(s) in %s (have %d): %v", + want, path, countDataLines(t, path), msgAndArgs) + case <-time.After(10 * time.Millisecond): + } + } +} + +// Proves Reconnect() tears the subscription down and rebuilds it without +// losing the stream's ability to receive. +func TestReconnect_stillReceivesAfterRecreatingSubscription(t *testing.T) { + if testing.Short() { + t.Skip("skipping DDS loopback test in short mode") + } + + t.Setenv("CYCLONEDDS_URI", ``) + + const ( + domainID = 79 + topicName = "reconnect_test_scan" + ) + + wp, err := ddscore.NewParticipant(domainID) + require.NoError(t, err) + defer func() { _ = wp.Close() }() + + writer, err := ddstest.SetupWriter(wp, topicName) + require.NoError(t, err) + + dir := t.TempDir() + tsFile := filepath.Join(dir, "timestamps.csv") + stream := New(Config{ + DDSDomainID: domainID, + DDSTopic: topicName, + TimestampsFile: tsFile, + DataFile: filepath.Join(dir, "data.bin"), + }) + stream.Start() + defer stream.Stop() + + require.NoError(t, ddstest.PublishScan(writer, 8, 0)) + waitForLines(t, tsFile, 1, 2*time.Second) + + stream.Reconnect() + time.Sleep(300 * time.Millisecond) + + deadline := time.After(5 * time.Second) + for countDataLines(t, tsFile) < 2 { + require.NoError(t, ddstest.PublishScan(writer, 8, 1)) + select { + case <-deadline: + t.Fatalf("a sample published after Reconnect was never received by the recreated subscription (have %d line(s))", + countDataLines(t, tsFile)) + case <-time.After(100 * time.Millisecond): + } + } +} diff --git a/internal/lidar/stream.go b/internal/lidar/stream.go index 5a6a06c..09a96bf 100644 --- a/internal/lidar/stream.go +++ b/internal/lidar/stream.go @@ -35,6 +35,10 @@ type LidarStream struct { cancel context.CancelFunc wg sync.WaitGroup + // reconnect asks a running record() to tear down and recreate its DDS + // subscription. Buffered so a request never blocks; see Reconnect. + reconnect chan struct{} + filesMu sync.Mutex files *outputFiles } @@ -49,7 +53,7 @@ type outputFiles struct { } func New(cfg Config) *LidarStream { - return &LidarStream{cfg: cfg} + return &LidarStream{cfg: cfg, reconnect: make(chan struct{}, 1)} } func (l *LidarStream) Start() { @@ -92,6 +96,15 @@ func (l *LidarStream) Rotate(dataFile, timestampsFile string) error { return nil } +// Reconnect tears down and recreates the DDS subscription without touching +// output files. Non-blocking; see heartbeat.Monitor.RegisterRecoverable. +func (l *LidarStream) Reconnect() { + select { + case l.reconnect <- struct{}{}: + default: + } +} + func (l *LidarStream) loop(ctx context.Context) { defer l.wg.Done() for ctx.Err() == nil { @@ -146,6 +159,9 @@ func (l *LidarStream) record(ctx context.Context) error { case <-ctx.Done(): l.flush() return nil + case <-l.reconnect: + l.flush() + return fmt.Errorf("reconnect requested") case <-syncTicker.C: l.flush() case sample, ok := <-receiver: diff --git a/internal/lowstate/stream.go b/internal/lowstate/stream.go index 7ac104b..15995a3 100644 --- a/internal/lowstate/stream.go +++ b/internal/lowstate/stream.go @@ -34,6 +34,10 @@ type LowstateStream struct { cancel context.CancelFunc wg sync.WaitGroup + // reconnect asks a running record() to tear down and recreate its DDS + // subscription. Buffered so a request never blocks; see Reconnect. + reconnect chan struct{} + filesMu sync.Mutex files *outputFiles } @@ -48,7 +52,7 @@ type outputFiles struct { } func New(cfg Config) *LowstateStream { - return &LowstateStream{cfg: cfg} + return &LowstateStream{cfg: cfg, reconnect: make(chan struct{}, 1)} } func (l *LowstateStream) Start() { @@ -91,6 +95,15 @@ func (l *LowstateStream) Rotate(dataFile, timestampsFile string) error { return nil } +// Reconnect tears down and recreates the DDS subscription without touching +// output files. Non-blocking; see heartbeat.Monitor.RegisterRecoverable. +func (l *LowstateStream) Reconnect() { + select { + case l.reconnect <- struct{}{}: + default: + } +} + func (l *LowstateStream) loop(ctx context.Context) { defer l.wg.Done() for ctx.Err() == nil { @@ -145,6 +158,9 @@ func (l *LowstateStream) record(ctx context.Context) error { case <-ctx.Done(): l.flush() return nil + case <-l.reconnect: + l.flush() + return fmt.Errorf("reconnect requested") case <-syncTicker.C: l.flush() case sample, ok := <-receiver: diff --git a/internal/odom/stream.go b/internal/odom/stream.go index 45b4f00..27e93a9 100644 --- a/internal/odom/stream.go +++ b/internal/odom/stream.go @@ -33,6 +33,10 @@ type OdomStream struct { cancel context.CancelFunc wg sync.WaitGroup + // reconnect asks a running record() to tear down and recreate its DDS + // subscription. Buffered so a request never blocks; see Reconnect. + reconnect chan struct{} + filesMu sync.Mutex files *outputFiles } @@ -47,7 +51,7 @@ type outputFiles struct { } func New(cfg Config) *OdomStream { - return &OdomStream{cfg: cfg} + return &OdomStream{cfg: cfg, reconnect: make(chan struct{}, 1)} } func (o *OdomStream) Start() { @@ -90,6 +94,15 @@ func (o *OdomStream) Rotate(dataFile, timestampsFile string) error { return nil } +// Reconnect tears down and recreates the DDS subscription without touching +// output files. Non-blocking; see heartbeat.Monitor.RegisterRecoverable. +func (o *OdomStream) Reconnect() { + select { + case o.reconnect <- struct{}{}: + default: + } +} + func (o *OdomStream) loop(ctx context.Context) { defer o.wg.Done() for ctx.Err() == nil { @@ -144,6 +157,9 @@ func (o *OdomStream) record(ctx context.Context) error { case <-ctx.Done(): o.flush() return nil + case <-o.reconnect: + o.flush() + return fmt.Errorf("reconnect requested") case <-syncTicker.C: o.flush() case sample, ok := <-receiver: diff --git a/internal/pointcloud/stream.go b/internal/pointcloud/stream.go index 09af1f6..ae8f710 100644 --- a/internal/pointcloud/stream.go +++ b/internal/pointcloud/stream.go @@ -41,6 +41,10 @@ type PointCloudStream struct { // needs no lock of its own. encoder *zstd.Encoder + // reconnect asks a running record() to tear down and recreate its DDS + // subscription. Buffered so a request never blocks; see Reconnect. + reconnect chan struct{} + filesMu sync.Mutex files *outputFiles } @@ -55,7 +59,7 @@ type outputFiles struct { } func New(cfg Config) *PointCloudStream { - return &PointCloudStream{cfg: cfg} + return &PointCloudStream{cfg: cfg, reconnect: make(chan struct{}, 1)} } func (c *PointCloudStream) Start() { @@ -104,6 +108,15 @@ func (c *PointCloudStream) Rotate(dataFile, timestampsFile string) error { return nil } +// Reconnect tears down and recreates the DDS subscription without touching +// output files. Non-blocking; see heartbeat.Monitor.RegisterRecoverable. +func (c *PointCloudStream) Reconnect() { + select { + case c.reconnect <- struct{}{}: + default: + } +} + func (c *PointCloudStream) loop(ctx context.Context) { defer c.wg.Done() for ctx.Err() == nil { @@ -172,6 +185,9 @@ func (c *PointCloudStream) record(ctx context.Context) error { case <-ctx.Done(): c.flush() return nil + case <-c.reconnect: + c.flush() + return fmt.Errorf("reconnect requested") case <-syncTicker.C: c.flush() case sample, ok := <-receiver: diff --git a/internal/retention/retention.go b/internal/retention/retention.go index 25192e6..2a461d5 100644 --- a/internal/retention/retention.go +++ b/internal/retention/retention.go @@ -22,8 +22,19 @@ import ( // uploadMarkerName marks a session directory as already uploaded. const uploadMarkerName = ".uploaded" -// uploadTimeout bounds one session's upload, whether kicked off by rotation, schedule, or shutdown. -const uploadTimeout = 10 * time.Minute +// minSessionAge is a second, time-based line of defense protecting the +// currently-recording session, alongside the dir == currentDir check. +const minSessionAge = 30 * time.Second + +// tooYoungToSweep reports whether dir's own recorded start time is within +// minSessionAge of now. +func tooYoungToSweep(dir string) bool { + start := ReadStartedAt(dir) + if start.IsZero() { + return false + } + return time.Since(start) < minSessionAge +} // IsUploaded reports whether dir was already fully uploaded. func IsUploaded(dir string) bool { @@ -125,19 +136,23 @@ func SnapshotTimebase(bootPath, sessionDir string) { } // UploadSession uploads one finished session directory; on failure the files are kept locally for a later retry. +// awaitReady, if non-nil, blocks first so still-relocating streams can settle before dir's contents are listed. // // dir is claimed for the duration of the upload, so a concurrent call for // the same dir (another upload path, or retention's cap enforcement) skips -// it instead of racing. -func UploadSession(ctl *control.State, client *upload.Client, dir, sessionDir string, startedAt time.Time, deleteAfter bool, opts upload.Options) { +// it instead of racing. No deadline is added here beyond ctx: each request is +// bounded by the caller and the client, and an interrupted upload resumes +// from its last completed file. +func UploadSession(ctx context.Context, ctl *control.State, client *upload.Client, dir, sessionDir string, startedAt time.Time, deleteAfter bool, opts upload.Options, awaitReady func(context.Context)) { if !ctl.TryClaimDir(dir) { slog.Info("retention: skipping upload, already in flight", "dir", dir) return } defer ctl.ReleaseDir(dir) - ctx, cancel := context.WithTimeout(context.Background(), uploadTimeout) - defer cancel() + if awaitReady != nil { + awaitReady(ctx) + } if err := client.UploadSession(ctx, dir, sessionDir, startedAt, opts); err != nil { slog.Error("session upload failed; files kept locally for retry", "dir", dir, "err", err) @@ -153,8 +168,13 @@ func UploadSession(ctl *control.State, client *upload.Client, dir, sessionDir st } } -// UploadFinishedSessionAsync kicks off finished's upload in the background if uploading is enabled. -func UploadFinishedSessionAsync(wg *sync.WaitGroup, uploader *upload.Client, ctl *control.State, recordingsDir, bootTimebasePath string, finished *session.Session, uploadDelete bool) { +// asyncUploadTimeout bounds a finished session's background upload (awaitReady included), so a +// stream stuck forever there can't block shutdown's wg.Wait() and defeat the restart recovering it. +var asyncUploadTimeout = 10 * time.Minute // var so tests can shrink it + +// UploadFinishedSessionAsync kicks off finished's upload in the background +// if uploading is enabled; awaitReady is passed straight through to UploadSession. +func UploadFinishedSessionAsync(wg *sync.WaitGroup, uploader *upload.Client, ctl *control.State, recordingsDir, bootTimebasePath string, finished *session.Session, uploadDelete bool, awaitReady func(context.Context)) { if uploader == nil || !ctl.Uploading() { return } @@ -162,37 +182,43 @@ func UploadFinishedSessionAsync(wg *sync.WaitGroup, uploader *upload.Client, ctl wg.Add(1) go func(dir, apiDir string, startedAt time.Time, opts upload.Options) { defer wg.Done() - UploadSession(ctl, uploader, dir, apiDir, startedAt, uploadDelete, opts) + ctx, cancel := context.WithTimeout(context.Background(), asyncUploadTimeout) + defer cancel() + UploadSession(ctx, ctl, uploader, dir, apiDir, startedAt, uploadDelete, opts, awaitReady) }(finished.RealDir(), APISessionDir(recordingsDir, finished.RealDir()), time.Unix(0, finished.StartUnixNs()), opts) } // Sweep runs catch-up uploads and cap enforcement once; kept as a building block for tests. -func Sweep(ctl *control.State, uploader *upload.Client, recordingsDir, bootTimebasePath, currentDir string, maxBytes int64) { +func Sweep(ctx context.Context, ctl *control.State, uploader *upload.Client, recordingsDir, bootTimebasePath, currentDir string, maxBytes int64, deleteAfter bool) { dirs, err := session.ListClosed(recordingsDir) if err != nil { slog.Warn("retention: cannot list session directories", "dir", recordingsDir, "err", err) return } - protected := func(dir string) bool { - return dir == currentDir || BootSessionDir(bootTimebasePath, dir) + uploadProtected := func(dir string) bool { return dir == currentDir || tooYoungToSweep(dir) } + deleteProtected := func(dir string) bool { + return dir == currentDir || BootSessionDir(bootTimebasePath, dir) || tooYoungToSweep(dir) } - CatchUpUploads(ctl, uploader, dirs, protected, recordingsDir, bootTimebasePath) - EnforceRetentionCap(ctl, recordingsDir, dirs, protected, maxBytes) + CatchUpUploads(ctx, ctl, uploader, dirs, uploadProtected, recordingsDir, bootTimebasePath, deleteAfter) + EnforceRetentionCap(ctl, recordingsDir, dirs, deleteProtected, maxBytes) } // CatchUpUploads retries every closed, not-yet-uploaded, non-protected directory in dirs, oldest first. -func CatchUpUploads(ctl *control.State, uploader *upload.Client, dirs []string, protected func(string) bool, recordingsDir, bootTimebasePath string) { +func CatchUpUploads(ctx context.Context, ctl *control.State, uploader *upload.Client, dirs []string, protected func(string) bool, recordingsDir, bootTimebasePath string, deleteAfter bool) { if uploader == nil { return } for _, dir := range dirs { + if ctx.Err() != nil { + return + } if protected(dir) || IsUploaded(dir) { continue } opts := UploadOptions(bootTimebasePath, dir) - UploadSession(ctl, uploader, dir, APISessionDir(recordingsDir, dir), ReadStartedAt(dir), false, opts) + UploadSession(ctx, ctl, uploader, dir, APISessionDir(recordingsDir, dir), ReadStartedAt(dir), deleteAfter, opts, nil) } } @@ -265,22 +291,24 @@ func EnforceRetentionCap(ctl *control.State, recordingsDir string, dirs []string } // RunSweeps drives catch-up uploads and cap enforcement on separate tickers until ctx is canceled. -func RunSweeps(ctx context.Context, uploader *upload.Client, recordingsDir, bootTimebasePath string, currentDir func() string, cfg config.RetentionConfig, ctl *control.State) { +func RunSweeps(ctx context.Context, uploader *upload.Client, recordingsDir, bootTimebasePath string, currentDir func() string, cfg config.RetentionConfig, ctl *control.State, deleteAfter bool) { interval := cfg.SweepInterval if interval <= 0 { interval = 5 * time.Minute } - listClosed := func() (dirs []string, protected func(string) bool, ok bool) { + listClosed := func() (dirs []string, uploadProtected, deleteProtected func(string) bool, ok bool) { dirs, err := session.ListClosed(recordingsDir) if err != nil { slog.Warn("retention: cannot list session directories", "dir", recordingsDir, "err", err) - return nil, nil, false + return nil, nil, nil, false } cd := currentDir() - return dirs, func(dir string) bool { - return dir == cd || BootSessionDir(bootTimebasePath, dir) - }, true + uploadProtected = func(dir string) bool { return dir == cd || tooYoungToSweep(dir) } + deleteProtected = func(dir string) bool { + return dir == cd || BootSessionDir(bootTimebasePath, dir) || tooYoungToSweep(dir) + } + return dirs, uploadProtected, deleteProtected, true } var wg sync.WaitGroup @@ -296,8 +324,8 @@ func RunSweeps(ctx context.Context, uploader *upload.Client, recordingsDir, boot case <-ctx.Done(): return case <-ticker.C: - if dirs, protected, ok := listClosed(); ok { - EnforceRetentionCap(ctl, recordingsDir, dirs, protected, cfg.MaxBytes) + if dirs, _, deleteProtected, ok := listClosed(); ok { + EnforceRetentionCap(ctl, recordingsDir, dirs, deleteProtected, cfg.MaxBytes) } } } @@ -314,8 +342,8 @@ func RunSweeps(ctx context.Context, uploader *upload.Client, recordingsDir, boot if !ctl.Uploading() { return } - if dirs, protected, ok := listClosed(); ok { - CatchUpUploads(ctl, uploader, dirs, protected, recordingsDir, bootTimebasePath) + if dirs, uploadProtected, _, ok := listClosed(); ok { + CatchUpUploads(ctx, ctl, uploader, dirs, uploadProtected, recordingsDir, bootTimebasePath, deleteAfter) } } for { diff --git a/internal/retention/retention_test.go b/internal/retention/retention_test.go index 2e548ba..ad89d1a 100644 --- a/internal/retention/retention_test.go +++ b/internal/retention/retention_test.go @@ -3,6 +3,7 @@ package retention import ( "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "os" @@ -15,7 +16,9 @@ import ( "github.com/stretchr/testify/require" "om1-telemetry/config" + "om1-telemetry/internal/clock" "om1-telemetry/internal/control" + "om1-telemetry/internal/session" "om1-telemetry/internal/upload" ) @@ -156,7 +159,7 @@ func TestSweep_nilUploaderSkipsCatchUpButStillEnforcesCap(t *testing.T) { writeFile(t, underCap, "a.bin", []byte("data")) ctl := control.New() - Sweep(ctl, nil, root, filepath.Join(root, "boot_timebase.jsonl"), "", 100) + Sweep(context.Background(), ctl, nil, root, filepath.Join(root, "boot_timebase.jsonl"), "", 100, false) require.False(t, IsUploaded(underCap), "without an uploader nothing can be marked uploaded") require.DirExists(t, underCap, "comfortably under the cap, so nothing needs to be deleted") @@ -164,7 +167,7 @@ func TestSweep_nilUploaderSkipsCatchUpButStillEnforcesCap(t *testing.T) { overCap := filepath.Join(root, "2026-08-15", "2026-08-15_00-00-00") writeFile(t, overCap, "b.bin", make([]byte, 200)) - Sweep(ctl, nil, root, filepath.Join(root, "boot_timebase.jsonl"), "", 100) + Sweep(context.Background(), ctl, nil, root, filepath.Join(root, "boot_timebase.jsonl"), "", 100, false) require.NoDirExists(t, underCap, "with no uploader configured, cap enforcement must still delete the oldest directory rather than let the disk fill up") @@ -236,7 +239,7 @@ func TestSweep_uploadsOldestUnmarkedFirstAndSkipsProtectedAndAlreadyUploaded(t * bootTimebasePath := filepath.Join(root, "does-not-exist.jsonl") // no boot session protection in play - Sweep(control.New(), client, root, bootTimebasePath, live, 0) + Sweep(context.Background(), control.New(), client, root, bootTimebasePath, live, 0, false) require.True(t, IsUploaded(oldest), "the oldest not-yet-uploaded closed dir must get uploaded") require.False(t, IsUploaded(live), "the currently-open session must never be swept") @@ -249,6 +252,80 @@ func TestSweep_uploadsOldestUnmarkedFirstAndSkipsProtectedAndAlreadyUploaded(t * require.Zero(t, api.createCalls[APISessionDir(root, live)]) } +// Reproduces the race this protection closes: a session directory that +// exists on disk but hasn't yet been reported as "current" must not upload. +func TestSweep_neverUploadsAVeryYoungDirEvenIfNotReportedCurrent(t *testing.T) { + api, apiURL := newMinimalFakeAPI(t) + client := upload.New(upload.Config{BaseURL: apiURL, APIKey: "k"}) + + root := t.TempDir() + justOpened := filepath.Join(root, "2026-08-31", "2026-08-31_21-22-33") + startedNs := time.Now().Add(-1 * time.Second).UnixNano() + writeFile(t, justOpened, "meta.json", []byte(fmt.Sprintf(`{"session_start_unix_ns":%d}`, startedNs))) + + bootTimebasePath := filepath.Join(root, "does-not-exist.jsonl") + + Sweep(context.Background(), control.New(), client, root, bootTimebasePath, "", 0, false) + + require.False(t, IsUploaded(justOpened), + "a directory younger than minSessionAge must never be uploaded, even when it isn't reported as the current session") + + api.mu.Lock() + defer api.mu.Unlock() + require.Zero(t, api.createCalls[APISessionDir(root, justOpened)]) +} + +// The age check must not block legitimate catch-up uploads -- only ones +// still inside the race window. +func TestSweep_uploadsADirOlderThanMinSessionAge(t *testing.T) { + api, apiURL := newMinimalFakeAPI(t) + client := upload.New(upload.Config{BaseURL: apiURL, APIKey: "k"}) + + root := t.TempDir() + old := filepath.Join(root, "2026-08-31", "2026-08-31_21-22-33") + startedNs := time.Now().Add(-2 * minSessionAge).UnixNano() + writeFile(t, old, "meta.json", []byte(fmt.Sprintf(`{"session_start_unix_ns":%d}`, startedNs))) + + bootTimebasePath := filepath.Join(root, "does-not-exist.jsonl") + + Sweep(context.Background(), control.New(), client, root, bootTimebasePath, "", 0, false) + + require.True(t, IsUploaded(old), "a directory well past minSessionAge must upload normally") + + api.mu.Lock() + defer api.mu.Unlock() + require.Equal(t, 1, api.createCalls[APISessionDir(root, old)]) +} + +// A failed upload for the boot session's own directory must still be +// retried by later sweeps, without ever touching its live clock journal. +func TestSweep_retriesBootSessionDirOnUploadButNeverDeletesIt(t *testing.T) { + api, apiURL := newMinimalFakeAPI(t) + client := upload.New(upload.Config{BaseURL: apiURL, APIKey: "k"}) + + root := t.TempDir() + bootDir := filepath.Join(root, "2026-08-14", "2026-08-14_00-00-00") + startedNs := time.Now().Add(-2 * minSessionAge).UnixNano() + writeFile(t, bootDir, "meta.json", []byte(fmt.Sprintf(`{"session_start_unix_ns":%d}`, startedNs))) + writeFile(t, bootDir, "lidar_scans.bin", make([]byte, 200)) + bootTimebasePath := filepath.Join(bootDir, clock.TimebaseName) + require.NoError(t, os.WriteFile(bootTimebasePath, []byte(`{"kind":"start"}`+"\n"), 0o644)) + + Sweep(context.Background(), control.New(), client, root, bootTimebasePath, "", 0, false) + + require.True(t, IsUploaded(bootDir), + "a catch-up sweep must retry the boot session's own directory, not strand it behind one failed attempt") + api.mu.Lock() + require.Equal(t, 1, api.createCalls[APISessionDir(root, bootDir)]) + api.mu.Unlock() + require.FileExists(t, bootTimebasePath, "the live clock journal must never be swept away, uploaded or not") + + Sweep(context.Background(), control.New(), client, root, bootTimebasePath, "", 10, false) + + require.DirExists(t, bootDir, "cap enforcement must never delete the boot session directory while its journal is live") + require.FileExists(t, bootTimebasePath) +} + // Rotation's async upload and a concurrent catch-up sweep must never both // upload the same directory at once. func TestUploadSession_concurrentCallsForSameDirOnlyUploadOnce(t *testing.T) { @@ -273,7 +350,7 @@ func TestUploadSession_concurrentCallsForSameDirOnlyUploadOnce(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - UploadSession(ctl, client, dir, "api/dir", time.Time{}, false, upload.Options{}) + UploadSession(context.Background(), ctl, client, dir, "api/dir", time.Time{}, false, upload.Options{}, nil) }() } @@ -286,6 +363,84 @@ func TestUploadSession_concurrentCallsForSameDirOnlyUploadOnce(t *testing.T) { "the second concurrent UploadSession call for the same dir must be skipped entirely, not just deduplicated after the fact") } +// Proves awaitReady runs to completion, and any files it adds are visible, +// before UploadSession's network calls start. +func TestUploadSession_awaitReadyCompletesBeforeUploadStarts(t *testing.T) { + api, apiURL := newMinimalFakeAPI(t) + client := upload.New(upload.Config{BaseURL: apiURL, APIKey: "k"}) + + dir := t.TempDir() + writeFile(t, dir, "meta.json", []byte(`{}`)) + + release := make(chan struct{}) + var awaitReadyDone atomic.Bool + awaitReady := func(ctx context.Context) { + <-release + writeFile(t, dir, "front_camera_raw_seg.mp4", []byte("video")) + awaitReadyDone.Store(true) + } + + done := make(chan struct{}) + go func() { + defer close(done) + UploadSession(context.Background(), control.New(), client, dir, "api/dir", time.Time{}, false, upload.Options{}, awaitReady) + }() + + sessionCreated := func() bool { + api.mu.Lock() + defer api.mu.Unlock() + return api.createCalls["api/dir"] > 0 + } + require.Never(t, sessionCreated, 50*time.Millisecond, 5*time.Millisecond, + "the session must not be created -- which is when regularFiles snapshots dir -- until awaitReady returns") + + close(release) + <-done + + require.True(t, awaitReadyDone.Load()) + require.True(t, sessionCreated()) + require.True(t, IsUploaded(dir), "the session, including the file awaitReady added, must be reported as fully uploaded") +} + +// Regression test for a real production deadlock: a recorder stream stuck +// badly enough to never satisfy awaitReady -- exactly what trips the +// heartbeat monitor's stuck-stream restart -- must not be able to block +// UploadFinishedSessionAsync's goroutine forever. It used to run awaitReady +// under context.Background(), so main's shutdown-time wg.Wait() (guarding the +// os.Exit that lets the container actually restart and recover the stream) +// never returned, silently defeating the restart it was blocking on. +func TestUploadFinishedSessionAsync_boundsAHungAwaitReady(t *testing.T) { + orig := asyncUploadTimeout + asyncUploadTimeout = 20 * time.Millisecond + t.Cleanup(func() { asyncUploadTimeout = orig }) + + _, apiURL := newMinimalFakeAPI(t) + client := upload.New(upload.Config{BaseURL: apiURL, APIKey: "k"}) + + sess, err := session.Open(t.TempDir(), clock.NewWithSync(clock.SyncYes)) + require.NoError(t, err) + + // Shaped like the real awaitSegments/WaitSegment: it only ever returns via + // ctx, never on its own -- so this proves the fix is the bounded context, + // not some change to how awaitReady itself behaves. + awaitReady := func(ctx context.Context) { <-ctx.Done() } + + var wg sync.WaitGroup + UploadFinishedSessionAsync(&wg, client, control.New(), t.TempDir(), "", sess, false, awaitReady) + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("UploadFinishedSessionAsync's goroutine never returned -- a hung awaitReady must not be able to block it forever") + } +} + // Cap enforcement must not block on a stuck catch-up upload. func TestRunSweeps_capEnforcementNotBlockedByStuckCatchUpUpload(t *testing.T) { const sweepInterval = 20 * time.Millisecond @@ -311,7 +466,7 @@ func TestRunSweeps_capEnforcementNotBlockedByStuckCatchUpUpload(t *testing.T) { done := make(chan struct{}) go func() { - RunSweeps(ctx, client, root, filepath.Join(root, "missing-boot.jsonl"), func() string { return "" }, cfg, control.New()) + RunSweeps(ctx, client, root, filepath.Join(root, "missing-boot.jsonl"), func() string { return "" }, cfg, control.New(), false) close(done) }() @@ -357,7 +512,7 @@ func TestRunSweeps_uploadDisabled_skipsCatchUpButStillEnforcesCap(t *testing.T) done := make(chan struct{}) go func() { - RunSweeps(ctx, client, root, filepath.Join(root, "missing-boot.jsonl"), func() string { return "" }, cfg, ctl) + RunSweeps(ctx, client, root, filepath.Join(root, "missing-boot.jsonl"), func() string { return "" }, cfg, ctl, false) close(done) }() @@ -403,7 +558,7 @@ func TestRunSweeps_uploadTrigger_runsImmediateSweep(t *testing.T) { done := make(chan struct{}) go func() { - RunSweeps(ctx, client, root, filepath.Join(root, "missing-boot.jsonl"), func() string { return "" }, cfg, ctl) + RunSweeps(ctx, client, root, filepath.Join(root, "missing-boot.jsonl"), func() string { return "" }, cfg, ctl, false) close(done) }() @@ -445,7 +600,7 @@ func TestRunSweeps_capDisabled_stillRunsCatchUpUploads(t *testing.T) { done := make(chan struct{}) go func() { - RunSweeps(ctx, client, root, filepath.Join(root, "missing-boot.jsonl"), func() string { return "" }, cfg, ctl) + RunSweeps(ctx, client, root, filepath.Join(root, "missing-boot.jsonl"), func() string { return "" }, cfg, ctl, false) close(done) }() diff --git a/internal/session/session.go b/internal/session/session.go index d54f18d..bfe7a27 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -440,7 +440,8 @@ func SortedNames(dirs []string) []string { return out } -// ListClosed returns every dated session directory under root, oldest first. +// ListClosed returns every dated session directory under root, oldest +// first. It does not verify a directory is actually closed. func ListClosed(root string) ([]string, error) { dateEntries, err := os.ReadDir(root) if err != nil { diff --git a/internal/traces/stream.go b/internal/traces/stream.go new file mode 100644 index 0000000..a9e56b6 --- /dev/null +++ b/internal/traces/stream.go @@ -0,0 +1,340 @@ +// Package traces polls a co-located OM1 process's Prometheus trace-export +// endpoint and appends any new records to the session directory as traces.jsonl. +package traces + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/prometheus/common/expfmt" + "github.com/prometheus/common/model" + + "om1-telemetry/internal/heartbeat" + "om1-telemetry/internal/recordutil" +) + +// HeartbeatName is the stream identifier for the heartbeat monitor. +const HeartbeatName = "traces" + +// traceMetricFamily is the Prometheus metric name OM1's exporter uses. +const traceMetricFamily = "om1_trace_info" + +// httpTimeout bounds a single poll request, so an unreachable/hung OM1 +// process can never stall the loop past one poll interval. +const httpTimeout = 10 * time.Second + +type Config struct { + // URL is OM1's metrics endpoint, e.g. http://localhost:9090/metrics. + URL string + + PollInterval time.Duration + + OutputFile string + + // CursorFile persists the dedup cursor across process restarts. Optional. + CursorFile string + + Monitor *heartbeat.Monitor +} + +// TraceStream polls Config.URL and appends new records to the current +// output file, deduplicating by timestamp across polls, rotations, and restarts. +type TraceStream struct { + cfg Config + running atomic.Bool + cancel context.CancelFunc + wg sync.WaitGroup + hc *http.Client + + fileMu sync.Mutex + file *os.File + + lastSeenMu sync.Mutex + lastSeen time.Time +} + +func New(cfg Config) *TraceStream { + return &TraceStream{cfg: cfg, hc: &http.Client{Timeout: httpTimeout}} +} + +func (s *TraceStream) Start() { + if s.running.Swap(true) { + return + } + ctx, cancel := context.WithCancel(context.Background()) + s.cancel = cancel + s.wg.Add(1) + go s.loop(ctx) +} + +func (s *TraceStream) Stop() { + if !s.running.Swap(false) { + return + } + s.cancel() + s.wg.Wait() + s.closeFile() + slog.Info("traces stream stopped") +} + +// Rotate switches the stream's output to a new file without touching the +// poll loop or the dedup cursor. +func (s *TraceStream) Rotate(outputFile string) error { + result, err := recordutil.OpenForAppend(outputFile) + if err != nil { + return fmt.Errorf("rotate: %w", err) + } + + s.fileMu.Lock() + old := s.file + s.file = result.File + s.fileMu.Unlock() + + if old != nil { + if err := old.Sync(); err != nil { + slog.Warn("traces: output sync failed", "err", err) + } + if err := old.Close(); err != nil { + slog.Warn("traces: output close failed", "err", err) + } + } + return nil +} + +func (s *TraceStream) loop(ctx context.Context) { + defer s.wg.Done() + + if err := s.ensureFileOpen(); err != nil { + slog.Error("traces: cannot open output file; stream disabled", "err", err) + return + } + s.loadCursor() + + slog.Info("traces recorder started", "url", s.cfg.URL, "interval", s.cfg.PollInterval) + + ticker := time.NewTicker(s.cfg.PollInterval) + defer ticker.Stop() + + for { + s.poll(ctx) + s.cfg.Monitor.Tick(HeartbeatName) + + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +func (s *TraceStream) ensureFileOpen() error { + s.fileMu.Lock() + defer s.fileMu.Unlock() + if s.file != nil { + return nil + } + result, err := recordutil.OpenForAppend(s.cfg.OutputFile) + if err != nil { + return err + } + s.file = result.File + return nil +} + +// poll fetches one batch and appends any records newer than the dedup +// cursor. Errors are logged and retried on the next tick. +func (s *TraceStream) poll(ctx context.Context) { + records, err := s.fetch(ctx) + if err != nil { + slog.Warn("traces: poll failed", "err", err) + return + } + if len(records) == 0 { + return + } + + sort.Slice(records, func(i, j int) bool { return records[i].parsedTS.Before(records[j].parsedTS) }) + + s.lastSeenMu.Lock() + cursor := s.lastSeen + s.lastSeenMu.Unlock() + + var wrote int + newest := cursor + for _, rec := range records { + if !rec.parsedTS.After(cursor) { + continue + } + if err := s.write(rec); err != nil { + slog.Warn("traces: write failed", "err", err) + return + } + wrote++ + if rec.parsedTS.After(newest) { + newest = rec.parsedTS + } + } + if wrote == 0 { + return + } + + s.lastSeenMu.Lock() + s.lastSeen = newest + s.lastSeenMu.Unlock() + s.saveCursor(newest) +} + +// loadCursor restores the dedup cursor from CursorFile, if set and present, +// so a full process restart doesn't re-ingest OM1's whole buffered backlog. +func (s *TraceStream) loadCursor() { + if s.cfg.CursorFile == "" { + return + } + raw, err := os.ReadFile(s.cfg.CursorFile) + if err != nil { + return + } + ts, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(string(raw))) + if err != nil { + slog.Warn("traces: cannot parse persisted cursor; starting from empty", "err", err) + return + } + s.lastSeenMu.Lock() + s.lastSeen = ts + s.lastSeenMu.Unlock() +} + +// saveCursor persists the dedup cursor so loadCursor can restore it after a +// restart. Best-effort: failures are only logged. +func (s *TraceStream) saveCursor(ts time.Time) { + if s.cfg.CursorFile == "" { + return + } + if dir := filepath.Dir(s.cfg.CursorFile); dir != "" && dir != "." { + if err := os.MkdirAll(dir, 0o755); err != nil { + slog.Warn("traces: cannot create cursor file directory", "err", err) + return + } + } + if err := os.WriteFile(s.cfg.CursorFile, []byte(ts.Format(time.RFC3339Nano)), 0o644); err != nil { + slog.Warn("traces: cannot persist cursor", "err", err) + } +} + +// line is the JSON shape written to OutputFile, matching OM1's own +// traces/tracer_.jsonl records field-for-field. +type line struct { + Timestamp string `json:"ts"` + Generation int `json:"generation"` + LLMInput string `json:"llm_input"` + LLMOutput json.RawMessage `json:"llm_output"` +} + +type record struct { + line + parsedTS time.Time +} + +func (s *TraceStream) write(rec record) error { + buf, err := json.Marshal(rec.line) + if err != nil { + return fmt.Errorf("marshal: %w", err) + } + + s.fileMu.Lock() + defer s.fileMu.Unlock() + if _, err := s.file.Write(append(buf, '\n')); err != nil { + return fmt.Errorf("write: %w", err) + } + return s.file.Sync() +} + +// fetch requests and parses one scrape of the trace-export endpoint. +func (s *TraceStream) fetch(ctx context.Context) ([]record, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.cfg.URL, nil) + if err != nil { + return nil, fmt.Errorf("build request: %w", err) + } + resp, err := s.hc.Do(req) + if err != nil { + return nil, fmt.Errorf("request: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("unexpected status %s", resp.Status) + } + + parser := expfmt.NewTextParser(model.LegacyValidation) + families, err := parser.TextToMetricFamilies(resp.Body) + if err != nil { + return nil, fmt.Errorf("parse: %w", err) + } + + family, ok := families[traceMetricFamily] + if !ok { + return nil, nil + } + + records := make([]record, 0, len(family.GetMetric())) + for _, m := range family.GetMetric() { + labels := make(map[string]string, len(m.GetLabel())) + for _, l := range m.GetLabel() { + labels[l.GetName()] = l.GetValue() + } + + ts := labels["ts"] + parsedTS, err := time.Parse(time.RFC3339Nano, ts) + if err != nil { + slog.Warn("traces: record with unparseable ts, skipping", "ts", ts, "err", err) + continue + } + + generation, err := strconv.Atoi(labels["generation"]) + if err != nil { + generation = 0 + } + + llmOutput := labels["llm_output"] + if !json.Valid([]byte(llmOutput)) { + llmOutput = "[]" + } + + records = append(records, record{ + line: line{ + Timestamp: ts, + Generation: generation, + LLMInput: labels["llm_input"], + LLMOutput: json.RawMessage(llmOutput), + }, + parsedTS: parsedTS, + }) + } + return records, nil +} + +func (s *TraceStream) closeFile() { + s.fileMu.Lock() + f := s.file + s.file = nil + s.fileMu.Unlock() + if f == nil { + return + } + if err := f.Sync(); err != nil { + slog.Warn("traces: output sync failed", "err", err) + } + if err := f.Close(); err != nil { + slog.Warn("traces: output close failed", "err", err) + } +} diff --git a/internal/traces/stream_test.go b/internal/traces/stream_test.go new file mode 100644 index 0000000..7c46c7a --- /dev/null +++ b/internal/traces/stream_test.go @@ -0,0 +1,250 @@ +package traces + +import ( + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func testConfig(t *testing.T, url string) Config { + return Config{ + URL: url, + PollInterval: 20 * time.Millisecond, + OutputFile: filepath.Join(t.TempDir(), "traces.jsonl"), + } +} + +// exposition builds a minimal om1_trace_info exposition-format body, as +// OM1's real exporter would serialize it. +func exposition(records ...[5]string) string { + var b strings.Builder + b.WriteString("# HELP om1_trace_info test\n# TYPE om1_trace_info gauge\n") + for _, r := range records { + fmt.Fprintf(&b, `om1_trace_info{seq="%s",ts="%s",generation="%s",llm_input="%s",llm_output="%s"} 1`+"\n", + r[0], r[1], r[2], r[3], r[4]) + } + return b.String() +} + +func serveText(t *testing.T, body func() string) *httptest.Server { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + _, _ = w.Write([]byte(body())) + })) + t.Cleanup(server.Close) + return server +} + +func waitFor(t *testing.T, timeout time.Duration, fn func() bool) { + t.Helper() + deadline := time.After(timeout) + for { + if fn() { + return + } + select { + case <-deadline: + t.Fatal("timed out waiting for condition") + case <-time.After(5 * time.Millisecond): + } + } +} + +func fileLines(t *testing.T, path string) []string { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return nil + } + t.Fatalf("read %s: %v", path, err) + } + var lines []string + for _, l := range strings.Split(strings.TrimSpace(string(data)), "\n") { + if l != "" { + lines = append(lines, l) + } + } + return lines +} + +func TestNew_returnsNonNilStream(t *testing.T) { + require.NotNil(t, New(testConfig(t, "http://example.invalid"))) +} + +func TestStartStop_cleanLifecycle(t *testing.T) { + server := serveText(t, func() string { return exposition() }) + stream := New(testConfig(t, server.URL)) + stream.Start() + time.Sleep(50 * time.Millisecond) + + done := make(chan struct{}) + go func() { + stream.Stop() + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Stop() did not return within 5s") + } +} + +func TestStart_idempotent(t *testing.T) { + server := serveText(t, func() string { return exposition() }) + stream := New(testConfig(t, server.URL)) + stream.Start() + stream.Start() + stream.Stop() +} + +func TestStop_beforeStart_isNoOp(t *testing.T) { + require.NotPanics(t, func() { New(testConfig(t, "http://example.invalid")).Stop() }) +} + +func TestStop_idempotent(t *testing.T) { + server := serveText(t, func() string { return exposition() }) + stream := New(testConfig(t, server.URL)) + stream.Start() + stream.Stop() + stream.Stop() +} + +func TestPoll_writesNewRecordsOnce(t *testing.T) { + rec := [5]string{"0", "2026-08-31T17:41:36.027193146Z", "1", "hello there", "[]"} + server := serveText(t, func() string { return exposition(rec) }) + + cfg := testConfig(t, server.URL) + stream := New(cfg) + stream.Start() + defer stream.Stop() + + waitFor(t, 2*time.Second, func() bool { return len(fileLines(t, cfg.OutputFile)) == 1 }) + + time.Sleep(150 * time.Millisecond) + require.Len(t, fileLines(t, cfg.OutputFile), 1, "an already-seen record must not be duplicated across polls") + require.Contains(t, fileLines(t, cfg.OutputFile)[0], `"llm_input":"hello there"`) +} + +func TestPoll_appendsNewRecordsAcrossPolls(t *testing.T) { + rec1 := [5]string{"0", "2026-08-31T17:41:36.000000000Z", "1", "first", "[]"} + rec2 := [5]string{"1", "2026-08-31T17:41:37.000000000Z", "1", "second", "[]"} + + var showSecond atomic.Bool + server := serveText(t, func() string { + if showSecond.Load() { + return exposition(rec1, rec2) + } + return exposition(rec1) + }) + + cfg := testConfig(t, server.URL) + stream := New(cfg) + stream.Start() + defer stream.Stop() + + waitFor(t, 2*time.Second, func() bool { return len(fileLines(t, cfg.OutputFile)) == 1 }) + + showSecond.Store(true) + waitFor(t, 2*time.Second, func() bool { return len(fileLines(t, cfg.OutputFile)) == 2 }) + + lines := fileLines(t, cfg.OutputFile) + require.Contains(t, lines[0], `"llm_input":"first"`) + require.Contains(t, lines[1], `"llm_input":"second"`) +} + +func TestPoll_handlesEscapedQuotesAndNewlines(t *testing.T) { + rec := [5]string{"0", "2026-08-31T17:41:36.000000000Z", "1", `line one\nline two \"quoted\"`, "[]"} + server := serveText(t, func() string { return exposition(rec) }) + + cfg := testConfig(t, server.URL) + stream := New(cfg) + stream.Start() + defer stream.Stop() + + waitFor(t, 2*time.Second, func() bool { return len(fileLines(t, cfg.OutputFile)) == 1 }) + + lines := fileLines(t, cfg.OutputFile) + require.Contains(t, lines[0], `line one\nline two \"quoted\"`) +} + +func TestRotate_preservesDedupCursor(t *testing.T) { + rec := [5]string{"0", "2026-08-31T17:41:36.000000000Z", "1", "hello", "[]"} + server := serveText(t, func() string { return exposition(rec) }) + + cfg := testConfig(t, server.URL) + stream := New(cfg) + stream.Start() + defer stream.Stop() + + waitFor(t, 2*time.Second, func() bool { return len(fileLines(t, cfg.OutputFile)) == 1 }) + + secondFile := filepath.Join(t.TempDir(), "traces2.jsonl") + require.NoError(t, stream.Rotate(secondFile)) + + time.Sleep(150 * time.Millisecond) + require.Empty(t, fileLines(t, secondFile), "a record already written before Rotate must not reappear in the new file") +} + +// Reproduces the real bug: a fresh process's cursor started at zero and +// re-ingested OM1's whole buffered backlog into whatever session was open. +func TestCursorFile_survivesProcessRestart(t *testing.T) { + rec := [5]string{"0", "2026-08-31T17:41:36.000000000Z", "1", "hello", "[]"} + server := serveText(t, func() string { return exposition(rec) }) + + cursorFile := filepath.Join(t.TempDir(), "cursor") + firstOutput := filepath.Join(t.TempDir(), "traces.jsonl") + + first := New(Config{ + URL: server.URL, PollInterval: 20 * time.Millisecond, + OutputFile: firstOutput, CursorFile: cursorFile, + }) + first.Start() + waitFor(t, 2*time.Second, func() bool { return len(fileLines(t, firstOutput)) == 1 }) + first.Stop() + + secondOutput := filepath.Join(t.TempDir(), "traces.jsonl") + second := New(Config{ + URL: server.URL, PollInterval: 20 * time.Millisecond, + OutputFile: secondOutput, CursorFile: cursorFile, + }) + second.Start() + defer second.Stop() + + time.Sleep(150 * time.Millisecond) + require.Empty(t, fileLines(t, secondOutput), + "a record already written by a prior process must not be re-written after a restart") +} + +func TestPoll_skipsUnparseableTimestamp(t *testing.T) { + body := "# HELP om1_trace_info test\n# TYPE om1_trace_info gauge\n" + + `om1_trace_info{seq="0",ts="not-a-timestamp",generation="1",llm_input="x",llm_output="[]"} 1` + "\n" + server := serveText(t, func() string { return body }) + + cfg := testConfig(t, server.URL) + stream := New(cfg) + stream.Start() + defer stream.Stop() + + time.Sleep(100 * time.Millisecond) + require.Empty(t, fileLines(t, cfg.OutputFile), "a record with an unparseable timestamp must be skipped, not crash the poller") +} + +func TestPoll_serverUnreachable_doesNotPanic(t *testing.T) { + cfg := testConfig(t, "http://127.0.0.1:1") + stream := New(cfg) + require.NotPanics(t, func() { + stream.Start() + time.Sleep(50 * time.Millisecond) + stream.Stop() + }) +} diff --git a/internal/upload/client.go b/internal/upload/client.go index f93dbb3..c3c502e 100644 --- a/internal/upload/client.go +++ b/internal/upload/client.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "io" + "log/slog" "mime/multipart" "net/http" "os" @@ -77,12 +78,50 @@ type presignedPOST struct { type sessionResp struct { SessionID string `json:"session_id"` S3Prefix string `json:"s3_prefix"` - Status string `json:"status"` Upload *presignedPOST `json:"upload"` } +// uploadStateName journals, inside the session directory, which files already reached S3 for a session ID. +const uploadStateName = ".upload-state.json" + +type uploadState struct { + SessionID string `json:"session_id"` + Done []string `json:"done"` +} + +func (s uploadState) pending(files []string) []string { + done := make(map[string]bool, len(s.Done)) + for _, name := range s.Done { + done[name] = true + } + var out []string + for _, name := range files { + if !done[name] { + out = append(out, name) + } + } + return out +} + +func loadUploadState(localDir string) uploadState { + var st uploadState + if raw, err := os.ReadFile(filepath.Join(localDir, uploadStateName)); err == nil { + _ = json.Unmarshal(raw, &st) + } + return st +} + +func saveUploadState(localDir string, st uploadState) error { + raw, err := json.Marshal(st) + if err != nil { + return err + } + return os.WriteFile(filepath.Join(localDir, uploadStateName), raw, 0o644) +} + // UploadSession preprocesses localDir, uploads every regular file under it under sessionDir, then marks -// the session complete. Best-effort: on error the session is marked "failed" server-side for a later retry. +// the session complete. Progress is journaled in localDir, so a retry continues the same server-side +// session and skips files that already reached S3. func (c *Client) UploadSession(ctx context.Context, localDir, sessionDir string, startedAt time.Time, opts Options) error { if !c.cfg.Ready() { return errors.New("upload: not configured (base URL / API key unset)") @@ -107,20 +146,40 @@ func (c *Client) UploadSession(ctx context.Context, localDir, sessionDir string, if err != nil { return fmt.Errorf("upload: create session: %w", err) } - if sess.Status == "complete" { - return nil + statePath := filepath.Join(localDir, uploadStateName) + + st := loadUploadState(localDir) + if st.SessionID != sess.SessionID { + st = uploadState{SessionID: sess.SessionID} + } + // The preserved journal's conversion is re-sent on every attempt: its source is still growing. + live := strings.TrimSuffix(opts.PreserveJSONL, "l") + var stMu sync.Mutex + done := func(name string) { + if name == live { + return + } + stMu.Lock() + defer stMu.Unlock() + st.Done = append(st.Done, name) + if err := saveUploadState(localDir, st); err != nil { + slog.Warn("upload: could not journal progress", "dir", localDir, "err", err) + } } - if err := c.uploadFiles(ctx, sess, localDir, files); err != nil { - c.fail(ctx, sess.SessionID, err) + if err := c.uploadFiles(ctx, sess, localDir, st.pending(files), done); err != nil { return fmt.Errorf("upload: %w", err) } - - return c.complete(ctx, sess.SessionID) + if err := c.complete(ctx, sess.SessionID); err != nil { + return err + } + _ = os.Remove(statePath) + return nil } -// uploadFiles uploads every file in files, up to Concurrency at a time; the first failure cancels the rest. -func (c *Client) uploadFiles(ctx context.Context, sess *sessionResp, localDir string, files []string) error { +// uploadFiles uploads every file in files, up to Concurrency at a time, calling done after each success; +// the first failure cancels the rest. +func (c *Client) uploadFiles(ctx context.Context, sess *sessionResp, localDir string, files []string, done func(name string)) error { uploadCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -131,20 +190,20 @@ func (c *Client) uploadFiles(ctx context.Context, sess *sessionResp, localDir st var once sync.Once var firstErr error - fail := func(name string, err error) { - once.Do(func() { - firstErr = fmt.Errorf("%s: %w", name, err) - cancel() - }) - } - for _, name := range files { sem <- struct{}{} wg.Add(1) go func(name string) { defer wg.Done() defer func() { <-sem }() - c.uploadOne(uploadCtx, sess, localDir, name, post, fail) + if err := c.uploadOne(uploadCtx, sess, localDir, name, post); err != nil { + once.Do(func() { + firstErr = fmt.Errorf("%s: %w", name, err) + cancel() + }) + return + } + done(name) }(name) } wg.Wait() @@ -152,38 +211,32 @@ func (c *Client) uploadFiles(ctx context.Context, sess *sessionResp, localDir st return firstErr } -// uploadOne uploads a single file, choosing multipart vs. direct-POST by size, reporting failure via fail. -func (c *Client) uploadOne(ctx context.Context, sess *sessionResp, localDir, name string, post *postBox, fail func(name string, err error)) { +// uploadOne uploads a single file, choosing multipart vs. direct-POST by size. +func (c *Client) uploadOne(ctx context.Context, sess *sessionResp, localDir, name string, post *postBox) error { path := filepath.Join(localDir, name) info, err := os.Stat(path) if err != nil { - fail(name, fmt.Errorf("stat %s: %w", path, err)) - return + return fmt.Errorf("stat %s: %w", path, err) } if info.Size() >= c.cfg.MultipartThreshold { - if err := c.uploadMultipart(ctx, sess.SessionID, path, name); err != nil { - fail(name, err) - } - return + return c.uploadMultipart(ctx, sess.SessionID, path, name) } p, err := post.get(ctx, c) if err != nil { - fail(name, fmt.Errorf("renew policy: %w", err)) - return - } - if err := c.uploadDirect(ctx, p, sess.S3Prefix+name, path); err != nil { - // Retry once against a freshly-presigned policy in case the old one expired. - p, rerr := post.forceRenew(ctx, c) - if rerr != nil { - fail(name, err) - return - } - if err := c.uploadDirect(ctx, p, sess.S3Prefix+name, path); err != nil { - fail(name, err) - } + return fmt.Errorf("renew policy: %w", err) + } + err = c.uploadDirect(ctx, p, sess.S3Prefix+name, path) + if err == nil { + return nil + } + // Retry once against a freshly-presigned policy in case the old one expired. + p, rerr := post.forceRenew(ctx, c) + if rerr != nil { + return err } + return c.uploadDirect(ctx, p, sess.S3Prefix+name, path) } // postBox holds the presigned-POST policy files share for direct uploads within one session. @@ -248,19 +301,6 @@ func (c *Client) complete(ctx context.Context, sessionID string) error { map[string]string{"status": "complete"}, nil) } -// fail best-effort marks a session failed server-side. -func (c *Client) fail(ctx context.Context, sessionID string, cause error) { - if sessionID == "" { - return - } - msg := cause.Error() - if len(msg) > 500 { - msg = msg[:500] - } - _ = c.doJSON(ctx, http.MethodPost, "/data/collection/sessions/"+sessionID+"/complete", - map[string]string{"status": "failed", "error": msg}, nil) -} - // uploadDirect POSTs one file straight to S3 using a presigned POST policy. func (c *Client) uploadDirect(ctx context.Context, post *presignedPOST, key, path string) error { f, err := os.Open(path) @@ -324,7 +364,8 @@ type completedPart struct { ETag string `json:"etag"` } -// uploadMultipart sends one file through the API's S3-multipart endpoints, reading it in PartSize chunks. +// uploadMultipart sends one file through the API's S3-multipart endpoints, uploading up to +// Concurrency parts at once so no single connection's loss/RTT caps the whole file's throughput. func (c *Client) uploadMultipart(ctx context.Context, sessionID, path, filename string) error { f, err := os.Open(path) if err != nil { @@ -332,6 +373,17 @@ func (c *Client) uploadMultipart(ctx context.Context, sessionID, path, filename } defer func() { _ = f.Close() }() + info, err := f.Stat() + if err != nil { + return err + } + size := info.Size() + partSize := c.cfg.PartSize + numParts := (size + partSize - 1) / partSize + if numParts == 0 { + numParts = 1 + } + var start struct { UploadID string `json:"upload_id"` } @@ -345,26 +397,52 @@ func (c *Client) uploadMultipart(ctx context.Context, sessionID, path, filename map[string]string{"filename": filename, "upload_id": start.UploadID}, nil) } + uploadCtx, cancel := context.WithCancel(ctx) + defer cancel() + + sem := make(chan struct{}, c.cfg.Concurrency) + var wg sync.WaitGroup + var mu sync.Mutex var parts []completedPart - buf := make([]byte, c.cfg.PartSize) - for partNumber := int64(1); ; partNumber++ { - n, readErr := io.ReadFull(f, buf) - if n > 0 { - etag, err := c.uploadPart(ctx, sessionID, filename, start.UploadID, partNumber, buf[:n]) + var once sync.Once + var firstErr error + + for partNumber := int64(1); partNumber <= numParts; partNumber++ { + offset := (partNumber - 1) * partSize + n := partSize + if offset+n > size { + n = size - offset + } + + sem <- struct{}{} + wg.Add(1) + go func(partNumber, offset, n int64) { + defer wg.Done() + defer func() { <-sem }() + + buf := make([]byte, n) + if _, err := f.ReadAt(buf, offset); err != nil { + once.Do(func() { firstErr = err; cancel() }) + return + } + etag, err := c.uploadPart(uploadCtx, sessionID, filename, start.UploadID, partNumber, buf) if err != nil { - abort() - return err + once.Do(func() { firstErr = err; cancel() }) + return } + mu.Lock() parts = append(parts, completedPart{PartNumber: partNumber, ETag: etag}) - } - if readErr == io.EOF || readErr == io.ErrUnexpectedEOF { - break - } - if readErr != nil { - abort() - return readErr - } + mu.Unlock() + }(partNumber, offset, n) } + wg.Wait() + + if firstErr != nil { + abort() + return firstErr + } + + sort.Slice(parts, func(i, j int) bool { return parts[i].PartNumber < parts[j].PartNumber }) return c.doJSON(ctx, http.MethodPost, "/data/collection/sessions/"+sessionID+"/files/complete", map[string]any{"filename": filename, "upload_id": start.UploadID, "parts": parts}, nil) diff --git a/internal/upload/client_test.go b/internal/upload/client_test.go index da14cf5..78b0819 100644 --- a/internal/upload/client_test.go +++ b/internal/upload/client_test.go @@ -9,6 +9,7 @@ import ( "net/http/httptest" "os" "path/filepath" + "strconv" "sync" "testing" "time" @@ -22,11 +23,14 @@ type fakeAPI struct { s3URL string - sessions map[string]*fakeSession - nextID int - preComplete string // session_dir that should short-circuit as already complete - failRequests map[string]int - multipartData map[string][]byte // uploadID -> reassembled bytes + sessions map[string]*fakeSession + nextID int + preComplete string // session_dir the server reports as already "complete" on creation + failRequests map[string]int + multipartParts map[string]map[int64][]byte // uploadID -> part number -> that part's bytes + multipartData map[string][]byte // uploadID -> parts reassembled in part-number order, set on complete + failPosts map[string]int // file name -> direct-POST attempts still to reject + postCount map[string]int // file name -> direct-POST attempts received // postDelay makes every direct-POST S3 upload take that long, so concurrency can be measured by wall-clock time. postDelay time.Duration @@ -46,9 +50,12 @@ type fakeSession struct { func newFakeAPI(t *testing.T) (*fakeAPI, *httptest.Server, *httptest.Server) { api := &fakeAPI{ - sessions: map[string]*fakeSession{}, - failRequests: map[string]int{}, - multipartData: map[string][]byte{}, + sessions: map[string]*fakeSession{}, + failRequests: map[string]int{}, + multipartParts: map[string]map[int64][]byte{}, + multipartData: map[string][]byte{}, + failPosts: map[string]int{}, + postCount: map[string]int{}, } s3 := httptest.NewServer(http.HandlerFunc(api.handleS3)) @@ -105,27 +112,34 @@ func (a *fakeAPI) createSession(w http.ResponseWriter, r *http.Request) { a.mu.Lock() defer a.mu.Unlock() + status := "uploading" if body.SessionDir == a.preComplete { - w.WriteHeader(http.StatusOK) - _ = json.NewEncoder(w).Encode(map[string]any{ - "session_id": "already-done", - "status": "complete", - }) - return + status = "complete" + } + + // Like the real API, an in-progress session for the same dir is handed back instead of a new one. + for _, sess := range a.sessions { + if sess.sessionDir == body.SessionDir && sess.status == "uploading" { + a.writeSession(w, http.StatusOK, sess) + return + } } a.nextID++ id := fmt.Sprintf("sess-%d", a.nextID) sess := &fakeSession{ id: id, sessionDir: body.SessionDir, startedAt: body.StartedAt, - status: "uploading", prefix: "cust/key/" + id + "/", + status: status, prefix: "cust/key/" + id + "/", uploaded: map[string][]byte{}, } a.sessions[id] = sess + a.writeSession(w, http.StatusCreated, sess) +} - w.WriteHeader(http.StatusCreated) +func (a *fakeAPI) writeSession(w http.ResponseWriter, code int, sess *fakeSession) { + w.WriteHeader(code) _ = json.NewEncoder(w).Encode(map[string]any{ - "session_id": id, + "session_id": sess.id, "s3_prefix": sess.prefix, "status": sess.status, "upload": map[string]any{ @@ -191,7 +205,7 @@ func (a *fakeAPI) startMultipart(w http.ResponseWriter, r *http.Request) { _ = json.NewDecoder(r.Body).Decode(&body) uploadID := "up-" + body.Filename a.mu.Lock() - a.multipartData[uploadID] = nil + a.multipartParts[uploadID] = map[int64][]byte{} a.mu.Unlock() w.WriteHeader(http.StatusOK) _ = json.NewEncoder(w).Encode(map[string]string{"upload_id": uploadID, "key": "k/" + body.Filename}) @@ -220,6 +234,16 @@ func (a *fakeAPI) completeMultipart(w http.ResponseWriter, r *http.Request) { } `json:"parts"` } _ = json.NewDecoder(r.Body).Decode(&body) + + a.mu.Lock() + parts := a.multipartParts[body.UploadID] + var reassembled []byte + for _, p := range body.Parts { + reassembled = append(reassembled, parts[p.PartNumber]...) + } + a.multipartData[body.UploadID] = reassembled + a.mu.Unlock() + w.WriteHeader(http.StatusOK) _ = json.NewEncoder(w).Encode(map[string]string{"message": "Upload completed", "key": body.Filename}) } @@ -250,6 +274,18 @@ func (a *fakeAPI) handleS3(w http.ResponseWriter, r *http.Request) { return } key := r.FormValue("key") + name := filepath.Base(key) + a.mu.Lock() + a.postCount[name]++ + reject := a.failPosts[name] > 0 + if reject { + a.failPosts[name]-- + } + a.mu.Unlock() + if reject { + w.WriteHeader(http.StatusInternalServerError) + return + } f, _, err := r.FormFile("file") if err != nil { w.WriteHeader(http.StatusBadRequest) @@ -270,10 +306,11 @@ func (a *fakeAPI) handleS3(w http.ResponseWriter, r *http.Request) { case r.Method == http.MethodPut && r.URL.Path == "/part": uploadID := r.URL.Query().Get("upload_id") part := r.URL.Query().Get("part") + partNumber, _ := strconv.ParseInt(part, 10, 64) data, _ := io.ReadAll(r.Body) a.mu.Lock() - a.multipartData[uploadID] = append(a.multipartData[uploadID], data...) + a.multipartParts[uploadID][partNumber] = data a.mu.Unlock() w.Header().Set("ETag", `"etag-`+uploadID+`-`+part+`"`) @@ -337,12 +374,14 @@ func TestUploadSession_uploadsFilesConcurrently(t *testing.T) { require.LessOrEqual(t, api.maxRunning, 4, "concurrency should be bounded by Config.Concurrency") } -func TestUploadSession_alreadyCompleteShortCircuits(t *testing.T) { +// Regression: the client used to skip uploading whenever createSession +// reported the session already "complete", which could silently drop files. +func TestUploadSession_uploadsEvenWhenServerReportsAlreadyComplete(t *testing.T) { api, apiSrv, _ := newFakeAPI(t) api.preComplete = "recordings/done" dir := t.TempDir() - writeFile(t, dir, "meta.json", []byte(`{}`)) + writeFile(t, dir, "meta.json", []byte(`{"ok":true}`)) c := New(Config{BaseURL: apiSrv.URL, APIKey: "test-key"}) err := c.UploadSession(context.Background(), dir, "recordings/done", time.Now(), Options{}) @@ -350,7 +389,11 @@ func TestUploadSession_alreadyCompleteShortCircuits(t *testing.T) { api.mu.Lock() defer api.mu.Unlock() - require.Empty(t, api.sessions, "should not have created a new session for one already complete") + require.Len(t, api.sessions, 1) + for _, sess := range api.sessions { + require.Equal(t, []byte(`{"ok":true}`), sess.uploaded["meta.json"], + "local files must still be uploaded even when the server claims the session is already complete") + } } func TestUploadSession_skipsDotfiles(t *testing.T) { @@ -412,23 +455,60 @@ func TestUploadSession_largeFileUsesMultipart(t *testing.T) { require.Equal(t, big, got, "reassembled multipart bytes must match the source file") } -func TestUploadSession_failureMarksSessionFailed(t *testing.T) { - api, apiSrv, s3 := newFakeAPI(t) - s3.Close() // every S3 request will now fail to connect +func TestUploadSession_retryResumesSameSessionAndSkipsUploadedFiles(t *testing.T) { + api, apiSrv, _ := newFakeAPI(t) + api.failPosts["b.bin"] = 2 // the first attempt's initial try and its policy-renew retry both fail + + dir := t.TempDir() + writeFile(t, dir, "a.bin", []byte("aaa")) + writeFile(t, dir, "b.bin", []byte("bbb")) + + c := New(Config{BaseURL: apiSrv.URL, APIKey: "test-key", Concurrency: 1}) + require.Error(t, c.UploadSession(context.Background(), dir, "recordings/resume", time.Now(), Options{})) + require.FileExists(t, filepath.Join(dir, uploadStateName)) + + require.NoError(t, c.UploadSession(context.Background(), dir, "recordings/resume", time.Now(), Options{})) + require.NoFileExists(t, filepath.Join(dir, uploadStateName)) + + api.mu.Lock() + defer api.mu.Unlock() + require.Len(t, api.sessions, 1, "the retry must continue the server-side session, not open a new one") + for _, sess := range api.sessions { + require.Equal(t, "complete", sess.status) + require.Equal(t, []byte("aaa"), sess.uploaded["a.bin"]) + require.Equal(t, []byte("bbb"), sess.uploaded["b.bin"]) + } + require.Equal(t, 1, api.postCount["a.bin"], "a file that already reached S3 must not be re-sent") + require.Equal(t, 3, api.postCount["b.bin"]) +} + +func TestUploadSession_preservedJournalIsResentOnRetry(t *testing.T) { + api, apiSrv, _ := newFakeAPI(t) + api.failPosts["meta.json"] = 2 dir := t.TempDir() + writeFile(t, dir, "clock_timebase.jsonl", []byte("{\"kind\":\"start\"}\n")) writeFile(t, dir, "meta.json", []byte(`{}`)) - c := New(Config{BaseURL: apiSrv.URL, APIKey: "test-key"}) - err := c.UploadSession(context.Background(), dir, "recordings/fails", time.Now(), Options{}) - require.Error(t, err) + c := New(Config{BaseURL: apiSrv.URL, APIKey: "test-key", Concurrency: 1}) + opts := Options{PreserveJSONL: "clock_timebase.jsonl"} + require.Error(t, c.UploadSession(context.Background(), dir, "recordings/live", time.Now(), opts)) + + f, err := os.OpenFile(filepath.Join(dir, "clock_timebase.jsonl"), os.O_APPEND|os.O_WRONLY, 0o644) + require.NoError(t, err) + _, err = f.WriteString("{\"kind\":\"sync\"}\n") + require.NoError(t, err) + require.NoError(t, f.Close()) + + require.NoError(t, c.UploadSession(context.Background(), dir, "recordings/live", time.Now(), opts)) api.mu.Lock() defer api.mu.Unlock() - require.Len(t, api.sessions, 1) + require.Equal(t, 2, api.postCount["clock_timebase.json"], "the still-growing journal must be re-sent") for _, sess := range api.sessions { - require.Equal(t, "failed", sess.status) - require.NotEmpty(t, sess.failReason) + var records []map[string]any + require.NoError(t, json.Unmarshal(sess.uploaded["clock_timebase.json"], &records)) + require.Len(t, records, 2) } } diff --git a/internal/upload/preprocess.go b/internal/upload/preprocess.go index 145183f..914d172 100644 --- a/internal/upload/preprocess.go +++ b/internal/upload/preprocess.go @@ -83,7 +83,7 @@ func jsonlToJSONArray(src, dst string) (err error) { }() w := bufio.NewWriter(out) - if _, err := w.WriteString("["); err != nil { + if _, err := w.WriteString("[\n"); err != nil { return err } @@ -99,7 +99,7 @@ func jsonlToJSONArray(src, dst string) (err error) { return fmt.Errorf("%s: invalid json line: %s", src, line) } if !first { - if _, err := w.WriteString(","); err != nil { + if _, err := w.WriteString(",\n"); err != nil { return err } } @@ -111,7 +111,11 @@ func jsonlToJSONArray(src, dst string) (err error) { if err := sc.Err(); err != nil { return fmt.Errorf("%s: %w", src, err) } - if _, err := w.WriteString("]"); err != nil { + if first { + if _, err := w.WriteString("]"); err != nil { + return err + } + } else if _, err := w.WriteString("\n]"); err != nil { return err } return w.Flush() diff --git a/internal/upload/preprocess_test.go b/internal/upload/preprocess_test.go index 7dbfaf8..2641b9a 100644 --- a/internal/upload/preprocess_test.go +++ b/internal/upload/preprocess_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "os" "path/filepath" + "strings" "testing" "time" @@ -30,6 +31,30 @@ func TestConvertJSONLToJSON_producesValidArrayAndRemovesSource(t *testing.T) { require.Equal(t, "sample", records[1]["kind"]) } +func TestConvertJSONLToJSON_putsEachRecordOnItsOwnLine(t *testing.T) { + dir := t.TempDir() + writeFile(t, dir, "traces.jsonl", + []byte("{\"ts\":\"a\"}\n{\"ts\":\"b\"}\n{\"ts\":\"c\"}\n")) + + require.NoError(t, convertJSONLToJSON(dir, Options{})) + + raw, err := os.ReadFile(filepath.Join(dir, "traces.json")) + require.NoError(t, err) + + lines := strings.Split(strings.TrimRight(string(raw), "\n"), "\n") + require.Equal(t, []string{ + "[", + `{"ts":"a"},`, + `{"ts":"b"},`, + `{"ts":"c"}`, + "]", + }, lines, "each record must be readable and diffable on its own line, not packed into one giant line") + + var records []map[string]any + require.NoError(t, json.Unmarshal(raw, &records)) + require.Len(t, records, 3) +} + func TestConvertJSONLToJSON_emptyFileProducesEmptyArray(t *testing.T) { dir := t.TempDir() writeFile(t, dir, "video_features.jsonl", []byte("")) diff --git a/internal/video/stream.go b/internal/video/stream.go index 7053e69..475f4f5 100644 --- a/internal/video/stream.go +++ b/internal/video/stream.go @@ -70,6 +70,17 @@ type segmentTarget struct { sessionStart time.Time timestampsFile string framesWrite *recordutil.FrameCSVWriter + + // ready closes once this target's first segment lands (or definitively + // won't, on relocation failure) -- see WaitSegment. + ready chan struct{} + readyOnce sync.Once +} + +// markReady unblocks any WaitSegment callers for this target. Safe to call +// more than once (a target may receive several segments, or none). +func (t *segmentTarget) markReady() { + t.readyOnce.Do(func() { close(t.ready) }) } func New(cfg Config) *VideoRTSPStream { @@ -133,7 +144,7 @@ func newSegmentTarget(sessionStart time.Time, timestampsFile, framesFile string) if framesFile != "" { fw = recordutil.NewFrameCSVWriter(framesFile) } - return &segmentTarget{sessionStart: sessionStart, timestampsFile: timestampsFile, framesWrite: fw} + return &segmentTarget{sessionStart: sessionStart, timestampsFile: timestampsFile, framesWrite: fw, ready: make(chan struct{})} } // targetFor returns the target whose session started most recently at or @@ -153,6 +164,29 @@ func (v *VideoRTSPStream) targetFor(startWallClock time.Time) *segmentTarget { return best } +// WaitSegment blocks until the target for sessionStart has relocated its +// first segment, or ctx ends; reports whether a segment actually arrived. +func (v *VideoRTSPStream) WaitSegment(ctx context.Context, sessionStart time.Time) bool { + v.targetMu.Lock() + var target *segmentTarget + for _, t := range v.targets { + if t.sessionStart.Equal(sessionStart) { + target = t + break + } + } + v.targetMu.Unlock() + if target == nil { + return false + } + select { + case <-target.ready: + return true + case <-ctx.Done(): + return false + } +} + func (v *VideoRTSPStream) loop(ctx context.Context) { defer close(v.done) v.ensureTarget() @@ -182,7 +216,6 @@ func (v *VideoRTSPStream) record(ctx context.Context) error { } start := time.Now() - startMono := clock.MonoNs() pattern := filepath.Join(v.cfg.ScratchDir, "%Y%m%dT%H%M%S.mp4") cmd := exec.CommandContext(ctx, "ffmpeg", @@ -239,7 +272,7 @@ func (v *VideoRTSPStream) record(ctx context.Context) error { // Must fully drain stdout before calling Wait: Wait closes the pipe once // it reaps the process, and reading after that races the close. - v.watchSegments(stdout, start, startMono) + v.watchSegments(stdout) waitErr := cmd.Wait() close(hbStop) @@ -252,14 +285,14 @@ func (v *VideoRTSPStream) record(ctx context.Context) error { // (filename,start_seconds,end_seconds), and relocates, indexes, and // extracts frame timestamps for each segment as it completes. Returns once // ffmpeg closes stdout, i.e. once the process exits. -func (v *VideoRTSPStream) watchSegments(stdout io.Reader, processStart time.Time, processStartMonoNs int64) { +func (v *VideoRTSPStream) watchSegments(stdout io.Reader) { scanner := bufio.NewScanner(stdout) for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) if line == "" { continue } - name, startSeconds, ok := parseSegmentListLine(line) + name, startSeconds, endSeconds, ok := parseSegmentListLine(line) if !ok { slog.Warn("video: unrecognized segment_list line", "camera", v.cfg.HeartbeatName, "line", line) continue @@ -267,7 +300,8 @@ func (v *VideoRTSPStream) watchSegments(stdout io.Reader, processStart time.Time // ffmpeg's segment_list reports the filename as written to its // pattern, without the directory -- join it back to ScratchDir. scratchFile := filepath.Join(v.cfg.ScratchDir, name) - v.finishSegment(scratchFile, startSeconds, processStart, processStartMonoNs) + duration := time.Duration((endSeconds - startSeconds) * float64(time.Second)) + v.finishSegment(scratchFile, duration, time.Now(), clock.MonoNs()) } if err := scanner.Err(); err != nil { slog.Warn("video: segment_list read error", "camera", v.cfg.HeartbeatName, "err", err) @@ -276,24 +310,27 @@ func (v *VideoRTSPStream) watchSegments(stdout io.Reader, processStart time.Time // parseSegmentListLine parses one "-segment_list_type csv" line: // filename,start_seconds,end_seconds. -func parseSegmentListLine(line string) (file string, startSeconds float64, ok bool) { +func parseSegmentListLine(line string) (file string, startSeconds, endSeconds float64, ok bool) { parts := strings.Split(line, ",") - if len(parts) < 2 { - return "", 0, false + if len(parts) < 3 { + return "", 0, 0, false } start, err := strconv.ParseFloat(strings.TrimSpace(parts[1]), 64) if err != nil { - return "", 0, false + return "", 0, 0, false + } + end, err := strconv.ParseFloat(strings.TrimSpace(parts[2]), 64) + if err != nil { + return "", 0, 0, false } - return strings.TrimSpace(parts[0]), start, true + return strings.TrimSpace(parts[0]), start, end, true } -// finishSegment relocates a just-closed segment from the scratch directory -// into the current target's session directory, indexes it, and kicks off -// its (async) frame extraction. -func (v *VideoRTSPStream) finishSegment(scratchFile string, startSeconds float64, processStart time.Time, processStartMonoNs int64) { - startWallClock := processStart.Add(time.Duration(startSeconds * float64(time.Second))) - startMonoNs := processStartMonoNs + int64(startSeconds*float64(time.Second)) +// finishSegment relocates a just-closed segment into the current target's +// session directory, indexes it, and kicks off its (async) frame extraction. +func (v *VideoRTSPStream) finishSegment(scratchFile string, duration time.Duration, observedNow time.Time, observedMonoNs int64) { + startWallClock := observedNow.Add(-duration) + startMonoNs := observedMonoNs - duration.Nanoseconds() target := v.targetFor(startWallClock) finalFile := filepath.Join(filepath.Dir(target.timestampsFile), @@ -301,6 +338,7 @@ func (v *VideoRTSPStream) finishSegment(scratchFile string, startSeconds float64 if err := os.Rename(scratchFile, finalFile); err != nil { slog.Error("video: could not relocate segment", "camera", v.cfg.HeartbeatName, "err", err) + target.markReady() return } @@ -309,6 +347,7 @@ func (v *VideoRTSPStream) finishSegment(scratchFile string, startSeconds float64 "file", target.timestampsFile, "err", err) } slog.Info("video segment closed", "camera", v.cfg.HeartbeatName, "file", finalFile) + target.markReady() if target.framesWrite == nil { return diff --git a/internal/video/stream_test.go b/internal/video/stream_test.go index d64901c..88b9d62 100644 --- a/internal/video/stream_test.go +++ b/internal/video/stream_test.go @@ -1,6 +1,7 @@ package video import ( + "context" "fmt" "os" "path/filepath" @@ -116,18 +117,19 @@ func TestRecord_createsOutputFileDirectory(t *testing.T) { } func TestParseSegmentListLine_parsesFilenameAndStart(t *testing.T) { - file, start, ok := parseSegmentListLine("20260826T185832.mp4,12.500000,15.000000") + file, start, end, ok := parseSegmentListLine("20260826T185832.mp4,12.500000,15.000000") require.True(t, ok) require.Equal(t, "20260826T185832.mp4", file) require.InDelta(t, 12.5, start, 1e-9) + require.InDelta(t, 15.0, end, 1e-9) } func TestParseSegmentListLine_rejectsMalformedLine(t *testing.T) { - _, _, ok := parseSegmentListLine("not,a,valid,,line") + _, _, _, ok := parseSegmentListLine("not,a,valid,,line") require.False(t, ok, "a non-numeric start field must be rejected") - _, _, ok = parseSegmentListLine("onlyonefield") - require.False(t, ok, "a line missing the start field must be rejected") + _, _, _, ok = parseSegmentListLine("onlyonefield") + require.False(t, ok, "a line missing the start/end fields must be rejected") } func TestFinishSegment_relocatesFileIndexesItAndPrefixesWithStem(t *testing.T) { @@ -145,7 +147,8 @@ func TestFinishSegment_relocatesFileIndexesItAndPrefixesWithStem(t *testing.T) { processStart := time.Unix(1_800_000_000, 0) stream.Rotate(processStart, filepath.Join(sessionDir, "front_camera_timestamps.csv"), "") // no frames file: skip async ffprobe - stream.finishSegment(scratchFile, 12.5, processStart, 5_000_000_000) + segmentStart := processStart.Add(12500 * time.Millisecond) + stream.finishSegment(scratchFile, 0, segmentStart, 5_000_000_000) wantFinal := filepath.Join(sessionDir, "front_camera_20260826T185832.mp4") require.FileExists(t, wantFinal, "the segment must be relocated with its camera stem prefixed") @@ -155,8 +158,7 @@ func TestFinishSegment_relocatesFileIndexesItAndPrefixesWithStem(t *testing.T) { require.NoError(t, err) content := string(data) require.Contains(t, content, "recording_start_unix_ns,segment_file,mono_ns") - wantStartUnixNs := processStart.Add(12500 * time.Millisecond).UnixNano() - require.Contains(t, content, fmt.Sprintf("%d", wantStartUnixNs), "the indexed start time must include the segment's offset into the stream") + require.Contains(t, content, fmt.Sprintf("%d", segmentStart.UnixNano()), "the indexed start time must match the segment's own start") require.Contains(t, content, "front_camera_20260826T185832.mp4") } @@ -174,10 +176,9 @@ func TestWatchSegments_joinsBareFilenameFromSegmentListWithScratchDir(t *testing OutputFile: filepath.Join(t.TempDir(), "front_camera.mp4"), ScratchDir: scratchDir, }) - processStart := time.Unix(1_800_000_000, 0) - stream.Rotate(processStart, filepath.Join(sessionDir, "front_camera_timestamps.csv"), "") + stream.Rotate(time.Now(), filepath.Join(sessionDir, "front_camera_timestamps.csv"), "") - stream.watchSegments(strings.NewReader("20260826T185832.mp4,0.000000,3.000000\n"), processStart, 0) + stream.watchSegments(strings.NewReader("20260826T185832.mp4,0.000000,3.000000\n")) require.FileExists(t, filepath.Join(sessionDir, "front_camera_20260826T185832.mp4"), "watchSegments must join the segment_list's bare filename with ScratchDir before relocating") @@ -240,6 +241,130 @@ func TestTargetFor_attributesByOwnTimestampEvenWhenRotateRacesAhead(t *testing.T require.NoFileExists(t, filepath.Join(sessionN1Dir, "front_camera_seg.mp4")) } +// Regression: a segment starting just after a rotation boundary must land +// in the new session even after a long-running process's clock has drifted. +func TestFinishSegment_notMisledByAccumulatedProcessDrift(t *testing.T) { + scratchDir := t.TempDir() + sessionNDir := t.TempDir() + sessionN1Dir := t.TempDir() + + sessionNStart := time.Unix(1_800_000_000, 0) + sessionN1Start := sessionNStart.Add(5 * time.Minute) + + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "front_camera.mp4"), + TimestampsFile: filepath.Join(sessionNDir, "front_camera_timestamps.csv"), + SessionStart: sessionNStart, + ScratchDir: scratchDir, + }) + stream.ensureTarget() + stream.Rotate(sessionN1Start, filepath.Join(sessionN1Dir, "front_camera_timestamps.csv"), "") + + segmentDuration := 5 * time.Minute + trueSegmentStart := sessionN1Start.Add(1 * time.Second) + observedClose := trueSegmentStart.Add(segmentDuration) + + seg := filepath.Join(scratchDir, "seg.mp4") + require.NoError(t, os.WriteFile(seg, []byte("a"), 0o644)) + + stream.finishSegment(seg, segmentDuration, observedClose, 0) + + require.FileExists(t, filepath.Join(sessionN1Dir, "front_camera_seg.mp4"), + "a segment observed to close one full span after starting just past the rotation boundary must land in the new session") + require.NoFileExists(t, filepath.Join(sessionNDir, "front_camera_seg.mp4")) +} + +// Regression: an uploader triggered right on rotation must be able to wait +// for the session's own segment to land instead of racing finishSegment. +func TestWaitSegment_blocksUntilFinishSegmentRelocatesThatSessionsSegment(t *testing.T) { + scratchDir := t.TempDir() + sessionDir := t.TempDir() + sessionStart := time.Unix(1_800_000_000, 0) + + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "front_camera.mp4"), + TimestampsFile: filepath.Join(sessionDir, "front_camera_timestamps.csv"), + SessionStart: sessionStart, + ScratchDir: scratchDir, + }) + stream.ensureTarget() + + waited := make(chan bool, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + waited <- stream.WaitSegment(ctx, sessionStart) + }() + + select { + case <-waited: + t.Fatal("WaitSegment returned before finishSegment relocated anything") + case <-time.After(20 * time.Millisecond): + } + + seg := filepath.Join(scratchDir, "seg.mp4") + require.NoError(t, os.WriteFile(seg, []byte("a"), 0o644)) + stream.finishSegment(seg, 0, sessionStart, 0) + + require.True(t, <-waited, "WaitSegment must return true once finishSegment relocates the session's segment") +} + +// A target that never receives a segment (RTSP never delivered data for +// that session) must not hang WaitSegment past its context. +func TestWaitSegment_returnsFalseWhenContextEndsFirst(t *testing.T) { + sessionStart := time.Unix(1_800_000_000, 0) + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "front_camera.mp4"), + SessionStart: sessionStart, + ScratchDir: t.TempDir(), + }) + stream.ensureTarget() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + require.False(t, stream.WaitSegment(ctx, sessionStart)) +} + +// A relocation failure must still unblock waiters rather than leaving them +// to time out every time. +func TestWaitSegment_returnsTrueImmediatelyAfterRelocationFailure(t *testing.T) { + sessionDir := t.TempDir() + sessionStart := time.Unix(1_800_000_000, 0) + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "front_camera.mp4"), + TimestampsFile: filepath.Join(sessionDir, "front_camera_timestamps.csv"), + SessionStart: sessionStart, + ScratchDir: t.TempDir(), + }) + stream.ensureTarget() + + // scratchFile does not exist, so os.Rename inside finishSegment fails. + stream.finishSegment(filepath.Join(t.TempDir(), "missing.mp4"), 0, sessionStart, 0) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + require.True(t, stream.WaitSegment(ctx, sessionStart), + "a relocation failure must still mark the target ready so callers don't wait out the full timeout") +} + +// No target was ever installed for this sessionStart. +func TestWaitSegment_returnsFalseForUnknownSessionStart(t *testing.T) { + stream := New(Config{ + RTSPURL: "rtsp://192.0.2.1:8554/unreachable", + OutputFile: filepath.Join(t.TempDir(), "front_camera.mp4"), + ScratchDir: t.TempDir(), + }) + stream.ensureTarget() + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + require.False(t, stream.WaitSegment(ctx, time.Unix(1_900_000_000, 0))) +} + func TestAppendSegmentEntry_emptyPath_isNoOp(t *testing.T) { err := appendSegmentEntry("", time.Now(), 0, "/data/video.mp4") require.NoError(t, err)