Bound live-stream latency drift with periodic live-edge resync

Over hours the wall drifted ~15s behind real-time: the Pi decodes a hair
slower than real-time, and since RTSP-over-TCP delivers every byte, that
small deficit buffers up instead of being dropped — and a live stream
can't be seeked back to the live edge.

Add player.resync_seconds (0 = off): the daemon reconnects each stream to
the live edge on that interval via mpv loadfile-replace over IPC, cycling
one tile at a time so only a single tile ever blips. Reusing the running
mpv process means no window teardown. At 600s this caps drift to a couple
seconds. config.example.yaml ships it at 600; README documents the knob
and the "also lighten decode load" caveat.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Levi Woodard
2026-07-02 14:09:07 -05:00
parent ed9814ab4e
commit 840324825b
5 changed files with 69 additions and 0 deletions

View File

@@ -269,6 +269,13 @@ pegged, work through:
of accumulating a backlog. Capping `player.max_fps` trims render load too. of accumulating a backlog. Capping `player.max_fps` trims render load too.
- **Audio is off by default** (`--no-audio`), which skips one audio decoder per - **Audio is off by default** (`--no-audio`), which skips one audio decoder per
stream. Set `player.audio: true` if you actually want camera sound. stream. Set `player.audio: true` if you actually want camera sound.
- **Latency slowly creeping up over hours** (e.g. seconds of drift after a
couple hours) means the Pi is decoding a hair behind real-time, so RTSP-over-
TCP quietly buffers the deficit — and a live stream can't be seeked back to
"now." Set `player.resync_seconds` (e.g. `600`) so the daemon reconnects each
stream to the live edge on that interval, staggered one tile at a time, which
caps the drift to a few seconds. If drift is large, also lighten decode load
(low substreams, cap `max_fps`) so the Pi keeps up between resyncs.
## Troubleshooting ## Troubleshooting

View File

@@ -22,6 +22,9 @@ player:
max_fps: 0 # cap rendered fps (0 = uncapped); trims render load max_fps: 0 # cap rendered fps (0 = uncapped); trims render load
audio: false # streams are muted by default (saves CPU too); audio: false # streams are muted by default (saves CPU too);
# set true to decode and play camera audio # set true to decode and play camera audio
resync_seconds: 600 # reconnect each stream to the live edge on this
# interval (staggered) so latency can't slowly
# drift; 0 = off. 600 keeps drift to a few sec.
restart_backoff_seconds: 3 restart_backoff_seconds: 3
extra_args: [] extra_args: []

View File

@@ -92,6 +92,13 @@ type Player struct {
// feeds is unwatchable with sound, and skipping the audio decoder saves // feeds is unwatchable with sound, and skipping the audio decoder saves
// CPU per stream. // CPU per stream.
Audio bool `yaml:"audio,omitempty"` Audio bool `yaml:"audio,omitempty"`
// ResyncSeconds, when > 0, makes the daemon reconnect each stream to the
// live edge on this interval (staggered across tiles). Live RTSP can't be
// seeked, so latency that slowly accumulates when the Pi decodes a hair
// behind real-time is only cleared by reopening the stream. Each tile is
// resynced about once per interval; e.g. 600 keeps drift well under a few
// seconds. 0 = off.
ResyncSeconds int `yaml:"resync_seconds,omitempty"`
// RestartBackoffSeconds is how long to wait before relaunching a stream // RestartBackoffSeconds is how long to wait before relaunching a stream
// that exited or stalled. // that exited or stalled.
RestartBackoffSeconds int `yaml:"restart_backoff_seconds,omitempty"` RestartBackoffSeconds int `yaml:"restart_backoff_seconds,omitempty"`

View File

@@ -200,6 +200,10 @@ func (d *Daemon) applyLayout(ctx context.Context, name string) error {
d.wg.Add(1) d.wg.Add(1)
go func() { defer d.wg.Done(); d.placeLoop(loCtx, places) }() go func() { defer d.wg.Done(); d.placeLoop(loCtx, places) }()
} }
if secs := cfg.Player.ResyncSeconds; secs > 0 && len(players) > 0 {
d.wg.Add(1)
go func() { defer d.wg.Done(); d.resyncLoop(loCtx, players, time.Duration(secs)*time.Second) }()
}
d.mu.Lock() d.mu.Lock()
d.players = players d.players = players
@@ -290,6 +294,38 @@ func (d *Daemon) placeLoop(ctx context.Context, places []placement) {
} }
} }
// resyncLoop reconnects one stream at a time to the live edge, cycling through
// all of them so each is resynced roughly once per interval. Spreading the
// reconnects (rather than doing them all at once) means only a single tile
// ever blips, and it bounds the latency drift that live RTSP accumulates when
// the Pi decodes slightly behind real-time.
func (d *Daemon) resyncLoop(ctx context.Context, players []*player.Player, interval time.Duration) {
step := interval / time.Duration(len(players))
if step < time.Second {
step = time.Second
}
ticker := time.NewTicker(step)
defer ticker.Stop()
i := 0
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
}
p := players[i%len(players)]
i++
if !p.Running() {
continue
}
if err := p.Reload(); err != nil {
d.log.Debug("resync failed", "slot", p.Slot, "camera", p.Name, "err", err)
continue
}
d.log.Debug("resynced stream to live edge", "slot", p.Slot, "camera", p.Name)
}
}
// sleepCtx sleeps for d or until ctx is cancelled; false means cancelled. // sleepCtx sleeps for d or until ctx is cancelled; false means cancelled.
func sleepCtx(ctx context.Context, dur time.Duration) bool { func sleepCtx(ctx context.Context, dur time.Duration) bool {
t := time.NewTimer(dur) t := time.NewTimer(dur)

View File

@@ -331,6 +331,22 @@ func isStatusLine(s string) bool {
return false return false
} }
// Reload reconnects mpv to its stream, snapping playback back to the live
// edge. Live RTSP can't be seeked, so latency that accumulates when the Pi
// runs a hair behind real-time is only cleared by reopening the stream. The
// daemon calls this on a stagger so drift stays bounded without a visible
// wall-wide blip. It reuses the running mpv process (no window teardown).
func (p *Player) Reload() error {
reply, err := p.Command("loadfile", p.URL, "replace")
if err != nil {
return err
}
if reply["error"] != "success" {
return fmt.Errorf("mpv loadfile: %v", reply["error"])
}
return nil
}
func sleep(ctx context.Context, d time.Duration) bool { func sleep(ctx context.Context, d time.Duration) bool {
t := time.NewTimer(d) t := time.NewTimer(d)
defer t.Stop() defer t.Stop()