From e05c30c6375f1133fa954fa3b5e2716122cba3d2 Mon Sep 17 00:00:00 2001 From: Quick104 <31828688+Quick104@users.noreply.github.com> Date: Tue, 11 Aug 2026 13:42:43 -0400 Subject: [PATCH] fix(playback): keep restart buffers prunable --- internal/api/handlers/playback.go | 9 +++--- internal/api/handlers/playback_config_test.go | 19 ++++++++++++ internal/playback/segment_pruner.go | 29 ++++++++++++++----- internal/playback/transcode.go | 17 +++++++++-- .../playback/transcode_restart_guard_test.go | 13 +++++++++ 5 files changed, 73 insertions(+), 14 deletions(-) create mode 100644 internal/api/handlers/playback_config_test.go diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index b64774a7..a8d0ea68 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -229,10 +229,11 @@ func NewPlaybackHandler(sessionMgr SessionManagerInterface, opts ...FilePathReso h.tm.Config = func() playback.TranscodeRuntimeConfig { c := h.playbackConfig() return playback.TranscodeRuntimeConfig{ - TranscodeDir: c.TranscodeDir, - FFmpegPath: c.FFmpegPath, - HWAccel: c.HWAccel, - HWDevice: c.HWDevice, + TranscodeDir: c.TranscodeDir, + FFmpegPath: c.FFmpegPath, + HWAccel: c.HWAccel, + HWDevice: c.HWDevice, + SegmentRetentionSeconds: c.SegmentRetentionSeconds, } } h.tm.StartThrottler = func(ctx context.Context, ts *playback.TranscodeSession) { diff --git a/internal/api/handlers/playback_config_test.go b/internal/api/handlers/playback_config_test.go new file mode 100644 index 00000000..fc8b2b90 --- /dev/null +++ b/internal/api/handlers/playback_config_test.go @@ -0,0 +1,19 @@ +package handlers + +import ( + "testing" + + "github.com/Silo-Server/silo-server/internal/config" + "github.com/Silo-Server/silo-server/internal/playback" +) + +func TestPlaybackHandlerTranscodeRuntimeConfigIncludesSegmentRetention(t *testing.T) { + handler := NewPlaybackHandler(playback.NewSessionManager(0, 0)) + handler.PlaybackConfig = func() config.PlaybackConfig { + return config.PlaybackConfig{SegmentRetentionSeconds: 300} + } + + if got := handler.tm.Config().SegmentRetentionSeconds; got != 300 { + t.Fatalf("configured segment retention = %d, want 300", got) + } +} diff --git a/internal/playback/segment_pruner.go b/internal/playback/segment_pruner.go index 6833c46d..0efd8050 100644 --- a/internal/playback/segment_pruner.go +++ b/internal/playback/segment_pruner.go @@ -41,7 +41,8 @@ func (s *TranscodeSession) scheduleSegmentPruneLocked() { } retainedSegments := (retentionSeconds + segmentDuration - 1) / segmentDuration floor := s.lastRequestedSegment - retainedSegments - if floor <= s.opts.StartSegmentNumber || floor-s.lastPruneFloor < segmentPruneHysteresis { + beforeStartReady := s.pruneBeforeStart && floor == s.opts.StartSegmentNumber + if floor < s.opts.StartSegmentNumber || (!beforeStartReady && floor == s.opts.StartSegmentNumber) || floor-s.lastPruneFloor < segmentPruneHysteresis { return } } @@ -68,6 +69,7 @@ func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, downloaded opts := s.opts outputDir := s.outputDir fromFloor := s.lastPruneFloor + pruneBeforeStart := s.pruneBeforeStart s.mu.Unlock() floor, complete, err := segmentRetentionFloor(outputDir, opts, downloadedThrough) @@ -76,7 +78,8 @@ func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, downloaded s.finishSegmentPrune(generation, fromFloor, fromFloor, downloadedThrough, segmentPruneRetryDelay) return } - if !complete || floor <= opts.StartSegmentNumber || (!continuation && floor-fromFloor < segmentPruneHysteresis) { + beforeStartReady := pruneBeforeStart && floor == opts.StartSegmentNumber + if !complete || floor < opts.StartSegmentNumber || (!beforeStartReady && floor == opts.StartSegmentNumber) || (!continuation && floor-fromFloor < segmentPruneHysteresis) { s.finishSegmentPruneAttempt(generation) return } @@ -87,20 +90,29 @@ func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, downloaded } freshGuard := 2*time.Duration(segmentDuration)*time.Second + 30*time.Second startupEnd := opts.StartSegmentNumber + startupSegmentRequirement(opts) + targetFloor := floor fromSegment := max(fromFloor, startupEnd) - toSegment := min(floor, fromSegment+segmentPruneBatchSize) + if pruneBeforeStart { + // A forward restart intentionally keeps the previous generation's back + // buffer until the replacement has produced its own. Retire that older + // range first, including its startup files, then resume normal pruning + // without touching the replacement process's startup window. + fromSegment = max(fromFloor, 0) + targetFloor = min(floor, opts.StartSegmentNumber) + } + toSegment := min(targetFloor, fromSegment+segmentPruneBatchSize) if fromSegment >= toSegment { - s.finishSegmentPrune(generation, floor, floor, downloadedThrough, 0) + s.finishSegmentPrune(generation, targetFloor, targetFloor, downloadedThrough, 0) return } if _, err := os.Stat(outputDir); err != nil { if errors.Is(err, os.ErrNotExist) { - s.finishSegmentPrune(generation, toSegment, floor, downloadedThrough, 0) + s.finishSegmentPrune(generation, toSegment, targetFloor, downloadedThrough, 0) return } slog.Warn("stat transcode directory for pruning", "component", "playback", "error", err, "session", opts.SessionID, "playback_session_id", opts.SessionID) - s.finishSegmentPrune(generation, fromSegment, floor, downloadedThrough, segmentPruneRetryDelay) + s.finishSegmentPrune(generation, fromSegment, targetFloor, downloadedThrough, segmentPruneRetryDelay) return } @@ -146,7 +158,7 @@ func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, downloaded freedBytes += info.Size() } - s.finishSegmentPrune(generation, processedFloor, floor, downloadedThrough, retryAfter) + s.finishSegmentPrune(generation, processedFloor, targetFloor, downloadedThrough, retryAfter) if removed > 0 { slog.Info("pruned downloaded transcode segments", "component", "playback", @@ -245,6 +257,9 @@ func (s *TranscodeSession) finishSegmentPrune( if processedFloor > s.lastPruneFloor { s.lastPruneFloor = processedFloor } + if s.pruneBeforeStart && s.lastPruneFloor >= s.opts.StartSegmentNumber { + s.pruneBeforeStart = false + } if retryAfter > 0 { time.AfterFunc(retryAfter, func() { s.mu.Lock() diff --git a/internal/playback/transcode.go b/internal/playback/transcode.go index bdf5f4eb..8d1b1c6e 100644 --- a/internal/playback/transcode.go +++ b/internal/playback/transcode.go @@ -123,6 +123,7 @@ type TranscodeSession struct { lastPruneFloor int lastPruneHighWater int segmentPruneRunning bool + pruneBeforeStart bool segmentGeneration uint64 throttler *TranscodeThrottler stderrLinesLogged int @@ -2065,7 +2066,8 @@ func (s *TranscodeSession) cleanSkippedSegments(downloadedThrough int, opts Tran startupEnd := opts.StartSegmentNumber + startupSegmentRequirement(opts) for _, entry := range entries { segNum, parseErr := ParseSegmentNumber(entry.Name()) - if parseErr != nil || segNum < startupEnd || segNum >= floor { + inStartupWindow := segNum >= opts.StartSegmentNumber && segNum < startupEnd + if parseErr != nil || inStartupWindow || segNum >= floor { continue } if removeErr := os.Remove(filepath.Join(s.outputDir, entry.Name())); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) { @@ -2116,8 +2118,17 @@ func (s *TranscodeSession) restart( s.segmentGeneration++ s.segmentPruneRunning = false previousRequestedSegment := s.lastRequestedSegment + forwardRestart := startSegment > previousRequestedSegment s.lastRequestedSegment = startSegment - s.lastPruneFloor = startSegment - 1 + if forwardRestart && s.opts.SegmentRetentionSeconds > 0 { + // Keep the old prune cursor reachable until the replacement process has + // produced a complete back buffer. Otherwise every forward restart leaves + // one retained window permanently behind startSegment. + s.pruneBeforeStart = true + } else { + s.lastPruneFloor = startSegment - 1 + s.pruneBeforeStart = false + } s.lastPruneHighWater = startSegment cancelCurrent := s.cancel done := s.done @@ -2143,7 +2154,7 @@ func (s *TranscodeSession) restart( reserveHWDevice := s.reserveHWDeviceOnRestart s.mu.Unlock() - if startSegment > previousRequestedSegment { + if forwardRestart { s.cleanSkippedSegments(previousRequestedSegment, opts) } diff --git a/internal/playback/transcode_restart_guard_test.go b/internal/playback/transcode_restart_guard_test.go index abae0d15..cac81ed3 100644 --- a/internal/playback/transcode_restart_guard_test.go +++ b/internal/playback/transcode_restart_guard_test.go @@ -16,11 +16,15 @@ func TestForwardRestartPreservesConfiguredBackBuffer(t *testing.T) { } dir := t.TempDir() + old := time.Now().Add(-time.Minute) for _, segment := range []int{3, 40, 64, 65, 69, 70, 100} { name := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})) if err := os.WriteFile(name, []byte("segment"), 0o600); err != nil { t.Fatal(err) } + if err := os.Chtimes(name, old, old); err != nil { + t.Fatal(err) + } } session := &TranscodeSession{ outputDir: dir, @@ -50,6 +54,15 @@ func TestForwardRestartPreservesConfiguredBackBuffer(t *testing.T) { t.Errorf("retained segment %d was removed: %v", segment, err) } } + + // Once the replacement generation has produced a complete back buffer, + // the preserved range must become eligible again rather than leaking one + // full window after every forward seek. + session.ReportSegmentDownloaded(105) + waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(70, TranscodeOpts{}))) + if _, err := os.Stat(filepath.Join(dir, segmentFilename(100, TranscodeOpts{}))); err != nil { + t.Fatalf("replacement startup segment was removed: %v", err) + } } func TestForwardRestartKeepsSegmentsWhenRetentionDisabled(t *testing.T) {