From e2899ff6190ca1c41077aa71c234f2de7952fa71 Mon Sep 17 00:00:00 2001 From: Quick104 <31828688+Quick104@users.noreply.github.com> Date: Tue, 11 Aug 2026 12:59:25 -0400 Subject: [PATCH] fix(playback): preserve retained segments across restarts --- internal/api/handlers/playback.go | 19 ++- internal/jellycompat/streams.go | 17 ++- internal/playback/segment_pruner.go | 132 +++++++++++++++--- internal/playback/segment_pruner_test.go | 93 +++++++++++- internal/playback/transcode.go | 91 ++++++++---- .../playback/transcode_restart_guard_test.go | 56 ++++++-- internal/transcodenode/server.go | 17 ++- 7 files changed, 344 insertions(+), 81 deletions(-) diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index 8efff8f9..29e4d8a9 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -3683,8 +3683,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht h.touchSessionActivity(sessionID) segmentName := chi.URLParam(r, "name") - downloadGeneration := transcodeSession.SegmentGeneration() - segmentFile, segmentInfo, err := transcodeSession.OpenSegment(segmentName) + segmentLease, err := transcodeSession.OpenSegment(segmentName) if err != nil && errors.Is(err, playback.ErrSegmentNotFound) { segNum, parseErr := playback.ParseSegmentNumber(segmentName) if parseErr == nil { @@ -3721,7 +3720,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht "session", sessionID, "playback_session_id", sessionID, ) - segmentFile, segmentInfo, err = transcodeSession.WaitForOpenSegment(segmentName, decision.WaitTimeout) + segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, decision.WaitTimeout) if err != nil && errors.Is(err, playback.ErrSegmentNotFound) { slog.InfoContext(r.Context(), "transcode segment wait timeout", "component", "api", "segment", segmentName, @@ -3783,7 +3782,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht ); restartErr == nil { // Throttler + exit monitor re-arm via the session's // restart hook. - segmentFile, segmentInfo, err = transcodeSession.WaitForOpenSegment(segmentName, 30*time.Second) + segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 30*time.Second) if err == nil && strings.EqualFold(transcodeSession.Opts().TargetCodecVideo, "copy") { // Copy-mode seeks can resume as soon as the target segment // exists, but that sometimes leaves the player one segment @@ -3791,7 +3790,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht // for a single lookahead fragment when available so the // first resumed playback window is less brittle. nextSegmentName := fmt.Sprintf("seg_%05d.m4s", segNum+1) - if nextSegment, _, nextErr := transcodeSession.WaitForOpenSegment(nextSegmentName, 1200*time.Millisecond); nextErr == nil { + if nextSegment, nextErr := transcodeSession.WaitForOpenSegment(nextSegmentName, 1200*time.Millisecond); nextErr == nil { _ = nextSegment.Close() } } @@ -3801,7 +3800,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht } else if transcodeSession.IsRunning() { // Non-numbered segment (e.g., init.mp4 for fMP4 HLS). // Wait briefly — the init segment is written almost immediately. - segmentFile, segmentInfo, err = transcodeSession.WaitForOpenSegment(segmentName, 10*time.Second) + segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 10*time.Second) } } if err != nil { @@ -3815,13 +3814,13 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht w.Header().Set("Cache-Control", "no-store, max-age=0") w.Header().Set("Pragma", "no-cache") - defer func() { _ = segmentFile.Close() }() + defer func() { _ = segmentLease.Close() }() sw := httpstream.NewRollingDeadlineWriter(w) - http.ServeContent(sw, r, segmentInfo.Name(), segmentInfo.ModTime(), segmentFile) + http.ServeContent(sw, r, segmentLease.Info.Name(), segmentLease.Info.ModTime(), segmentLease.File) if r.Method == http.MethodGet && - sw.CompletedFullResponse(r.Context(), segmentInfo.Size()) { + sw.CompletedFullResponse(r.Context(), segmentLease.Info.Size()) { if segNum, parseErr := playback.ParseSegmentNumber(segmentName); parseErr == nil { - transcodeSession.ReportSegmentDownloadedForGeneration(segNum, downloadGeneration) + transcodeSession.ReportSegmentDownloadedForGeneration(segNum, segmentLease.Generation) } } } diff --git a/internal/jellycompat/streams.go b/internal/jellycompat/streams.go index 93d5cba2..249b8da3 100644 --- a/internal/jellycompat/streams.go +++ b/internal/jellycompat/streams.go @@ -417,8 +417,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques } segmentName := name + "." + ext - downloadGeneration := transcodeSession.SegmentGeneration() - segment, segmentInfo, err := transcodeSession.OpenSegment(segmentName) + segmentLease, err := transcodeSession.OpenSegment(segmentName) if err != nil && errors.Is(err, playback.ErrSegmentNotFound) { segNum, parseErr := playback.ParseSegmentNumber(name) if parseErr == nil { @@ -455,7 +454,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques "session", playSession.UpstreamSessionID, "playback_session_id", playSession.UpstreamSessionID, ) - segment, segmentInfo, err = transcodeSession.WaitForOpenSegment(segmentName, decision.WaitTimeout) + segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, decision.WaitTimeout) if err != nil && errors.Is(err, playback.ErrSegmentNotFound) { slog.InfoContext(r.Context(), "transcode segment wait timeout", "component", "jellycompat", "segment", segmentName, @@ -518,14 +517,14 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques seekSeconds, segNum, ); restartErr == nil { - segment, segmentInfo, err = transcodeSession.WaitForOpenSegment(segmentName, 30*time.Second) + segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 30*time.Second) } } } } else if transcodeSession.IsRunning() { // Non-numbered segment (e.g. init.mp4 for fMP4 HLS). // Wait briefly — the init segment is written almost immediately. - segment, segmentInfo, err = transcodeSession.WaitForOpenSegment(segmentName, 10*time.Second) + segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 10*time.Second) } } if err != nil { @@ -534,13 +533,13 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques return } - defer func() { _ = segment.Close() }() + defer func() { _ = segmentLease.Close() }() sw := httpstream.NewRollingDeadlineWriter(w) - http.ServeContent(sw, r, segmentInfo.Name(), segmentInfo.ModTime(), segment) + http.ServeContent(sw, r, segmentLease.Info.Name(), segmentLease.Info.ModTime(), segmentLease.File) if r.Method == http.MethodGet && - sw.CompletedFullResponse(r.Context(), segmentInfo.Size()) { + sw.CompletedFullResponse(r.Context(), segmentLease.Info.Size()) { if segNum, parseErr := playback.ParseSegmentNumber(name); parseErr == nil { - transcodeSession.ReportSegmentDownloadedForGeneration(segNum, downloadGeneration) + transcodeSession.ReportSegmentDownloadedForGeneration(segNum, segmentLease.Generation) } } } diff --git a/internal/playback/segment_pruner.go b/internal/playback/segment_pruner.go index 9a0dbd58..ebc65eaa 100644 --- a/internal/playback/segment_pruner.go +++ b/internal/playback/segment_pruner.go @@ -2,9 +2,11 @@ package playback import ( "errors" + "fmt" "log/slog" "os" "path/filepath" + "strings" "time" ) @@ -23,19 +25,32 @@ func (s *TranscodeSession) scheduleSegmentPruneLocked() { if retentionSeconds <= 0 || s.segmentPruneRunning || s.restarting { return } - segmentDuration := s.opts.SegmentDuration - if segmentDuration <= 0 { - segmentDuration = defaultSegmentDuration - } - retainedSegments := (retentionSeconds + segmentDuration - 1) / segmentDuration - floor := s.lastRequestedSegment - retainedSegments - if floor <= s.opts.StartSegmentNumber || floor-s.lastPruneFloor < segmentPruneHysteresis { - return + + if strings.EqualFold(s.opts.TargetCodecVideo, "copy") { + // Copy-mode fragments follow source keyframes, so their real media-time + // floor is resolved asynchronously from EXTINF durations below. Use the + // download high-water mark only to avoid reparsing the manifest for every + // segment response. + if s.lastRequestedSegment-s.lastPruneHighWater < segmentPruneHysteresis { + return + } + } else { + segmentDuration := s.opts.SegmentDuration + if segmentDuration <= 0 { + segmentDuration = defaultSegmentDuration + } + retainedSegments := (retentionSeconds + segmentDuration - 1) / segmentDuration + floor := s.lastRequestedSegment - retainedSegments + if floor <= s.opts.StartSegmentNumber || floor-s.lastPruneFloor < segmentPruneHysteresis { + return + } } s.segmentPruneRunning = true generation := s.segmentGeneration - go s.pruneDownloadedSegments(generation, floor) + downloadedThrough := s.lastRequestedSegment + s.lastPruneHighWater = downloadedThrough + go s.pruneDownloadedSegments(generation, downloadedThrough, false) } // pruneDownloadedSegments removes completed media files strictly behind floor. @@ -43,7 +58,7 @@ func (s *TranscodeSession) scheduleSegmentPruneLocked() { // avoiding repeated full-directory scans when FFmpeg has generated far ahead // of the client. The current process's startup window remains present so real // manifest reloads cannot get stuck behind startupFilesReady after cleanup. -func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, floor int) { +func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, downloadedThrough int, continuation bool) { started := time.Now() s.mu.Lock() if generation != s.segmentGeneration || s.restarting { @@ -55,6 +70,17 @@ func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, floor int) fromFloor := s.lastPruneFloor s.mu.Unlock() + floor, complete, err := segmentRetentionFloor(outputDir, opts, downloadedThrough) + if err != nil { + slog.Warn("resolve transcode segment retention floor", "component", "playback", "error", err, "session", opts.SessionID, "playback_session_id", opts.SessionID) + s.finishSegmentPrune(generation, fromFloor, fromFloor, downloadedThrough, segmentPruneRetryDelay) + return + } + if !complete || floor <= opts.StartSegmentNumber || (!continuation && floor-fromFloor < segmentPruneHysteresis) { + s.finishSegmentPruneAttempt(generation) + return + } + segmentDuration := opts.SegmentDuration if segmentDuration <= 0 { segmentDuration = defaultSegmentDuration @@ -64,17 +90,17 @@ func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, floor int) fromSegment := max(fromFloor, startupEnd) toSegment := min(floor, fromSegment+segmentPruneBatchSize) if fromSegment >= toSegment { - s.finishSegmentPrune(generation, floor, floor, 0) + s.finishSegmentPrune(generation, floor, floor, downloadedThrough, 0) return } if _, err := os.Stat(outputDir); err != nil { if errors.Is(err, os.ErrNotExist) { - s.finishSegmentPrune(generation, toSegment, floor, 0) + s.finishSegmentPrune(generation, toSegment, floor, 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, segmentPruneRetryDelay) + s.finishSegmentPrune(generation, fromSegment, floor, downloadedThrough, segmentPruneRetryDelay) return } @@ -120,7 +146,7 @@ func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, floor int) freedBytes += info.Size() } - s.finishSegmentPrune(generation, processedFloor, floor, retryAfter) + s.finishSegmentPrune(generation, processedFloor, floor, downloadedThrough, retryAfter) if removed > 0 { slog.Info("pruned downloaded transcode segments", "component", "playback", @@ -134,6 +160,60 @@ func (s *TranscodeSession) pruneDownloadedSegments(generation uint64, floor int) } } +// segmentRetentionFloor returns the first segment that must remain to preserve +// the configured media-time window behind downloadedThrough. Encoded HLS uses +// fixed-duration fragments. Copy HLS follows source keyframes, so its floor is +// derived from the current manifest's actual EXTINF durations. complete is +// false when the manifest does not yet cover the full requested back buffer. +func segmentRetentionFloor(outputDir string, opts TranscodeOpts, downloadedThrough int) (floor int, complete bool, err error) { + retentionSeconds := opts.SegmentRetentionSeconds + if retentionSeconds <= 0 { + return 0, false, nil + } + + if !strings.EqualFold(opts.TargetCodecVideo, "copy") { + segmentDuration := opts.SegmentDuration + if segmentDuration <= 0 { + segmentDuration = defaultSegmentDuration + } + retainedSegments := (retentionSeconds + segmentDuration - 1) / segmentDuration + return downloadedThrough - retainedSegments, true, nil + } + + manifest, err := os.ReadFile(filepath.Join(outputDir, "stream.m3u8")) + if err != nil { + return 0, false, fmt.Errorf("read copy manifest: %w", err) + } + timeline, err := parseManifestTimeline(manifest) + if err != nil { + return 0, false, fmt.Errorf("parse copy manifest: %w", err) + } + + downloadedIndex := -1 + for i, entry := range timeline.entries { + if entry.duration <= 0 { + return 0, false, fmt.Errorf("copy segment %d has non-positive duration %.6f", entry.number, entry.duration) + } + if entry.number == downloadedThrough { + downloadedIndex = i + } + } + if downloadedIndex < 0 { + return 0, false, nil + } + + floor = downloadedThrough + retainedSeconds := 0.0 + for i := downloadedIndex - 1; i >= 0; i-- { + floor = timeline.entries[i].number + retainedSeconds += timeline.entries[i].duration + if retainedSeconds >= float64(retentionSeconds) { + return floor, true, nil + } + } + return 0, false, nil +} + func earlierPruneRetry(currentFloor int, currentDelay time.Duration, segment int, delay time.Duration) (int, time.Duration) { if segment < currentFloor { return segment, max(delay, time.Millisecond) @@ -141,7 +221,23 @@ func earlierPruneRetry(currentFloor int, currentDelay time.Duration, segment int return currentFloor, currentDelay } -func (s *TranscodeSession) finishSegmentPrune(generation uint64, processedFloor, targetFloor int, retryAfter time.Duration) { +func (s *TranscodeSession) finishSegmentPruneAttempt(generation uint64) { + s.mu.Lock() + defer s.mu.Unlock() + if generation != s.segmentGeneration { + return + } + s.segmentPruneRunning = false + s.scheduleSegmentPruneLocked() +} + +func (s *TranscodeSession) finishSegmentPrune( + generation uint64, + processedFloor int, + targetFloor int, + downloadedThrough int, + retryAfter time.Duration, +) { s.mu.Lock() defer s.mu.Unlock() if generation != s.segmentGeneration { @@ -158,10 +254,14 @@ func (s *TranscodeSession) finishSegmentPrune(generation uint64, processedFloor, return } s.mu.Unlock() - go s.pruneDownloadedSegments(generation, targetFloor) + go s.pruneDownloadedSegments(generation, downloadedThrough, true) }) return } + if processedFloor < targetFloor { + go s.pruneDownloadedSegments(generation, downloadedThrough, true) + return + } s.segmentPruneRunning = false s.scheduleSegmentPruneLocked() } diff --git a/internal/playback/segment_pruner_test.go b/internal/playback/segment_pruner_test.go index 691a85f0..9e0fb234 100644 --- a/internal/playback/segment_pruner_test.go +++ b/internal/playback/segment_pruner_test.go @@ -1,9 +1,11 @@ package playback import ( + "errors" "io" "os" "path/filepath" + "strings" "testing" "time" ) @@ -101,6 +103,63 @@ func TestPruningContinuesAcrossBoundedBatches(t *testing.T) { } } +func TestCopyRetentionUsesManifestDurations(t *testing.T) { + dir := t.TempDir() + old := time.Now().Add(-time.Minute) + opts := TranscodeOpts{ + SessionID: "copy-duration-test", + SourceVideoCodec: "h264", + TargetCodecVideo: "copy", + SegmentDuration: 2, + SegmentRetentionSeconds: 10, + } + manifest := strings.Join([]string{ + "#EXTM3U", + "#EXT-X-VERSION:7", + "#EXT-X-MEDIA-SEQUENCE:8", + "#EXTINF:1.0,", + "seg_00008.m4s", + "#EXTINF:2.0,", + "seg_00009.m4s", + "#EXTINF:3.0,", + "seg_00010.m4s", + "#EXTINF:4.0,", + "seg_00011.m4s", + "#EXTINF:5.0,", + "seg_00012.m4s", + "#EXTINF:6.0,", + "seg_00013.m4s", + "#EXTINF:7.0,", + "seg_00014.m4s", + "#EXTINF:8.0,", + "seg_00015.m4s", + "", + }, "\n") + writePrunerTestFile(t, filepath.Join(dir, "stream.m3u8"), []byte(manifest), old) + for segment := 8; segment <= 15; segment++ { + writePrunerTestFile(t, filepath.Join(dir, segmentFilename(segment, opts)), []byte("segment"), old) + } + + session := &TranscodeSession{ + opts: opts, + outputDir: dir, + lastPruneFloor: -1, + lastRequestedSegment: 0, + } + session.ReportSegmentDownloaded(15) + waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(12, opts))) + t.Cleanup(func() { + session.mu.Lock() + session.segmentGeneration++ + session.segmentPruneRunning = false + session.mu.Unlock() + }) + + assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(13, opts))) + assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(14, opts))) + assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(15, opts))) +} + func TestReportSegmentDownloadedDisabledKeepsSegments(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, segmentFilename(3, TranscodeOpts{})) @@ -159,7 +218,7 @@ func TestOpenSegmentDescriptorSurvivesUnlink(t *testing.T) { writePrunerTestFile(t, path, want, time.Now()) session := &TranscodeSession{outputDir: dir} - segment, _, err := session.OpenSegment("seg_00001.ts") + segment, err := session.OpenSegment("seg_00001.ts") if err != nil { t.Fatalf("OpenSegment: %v", err) } @@ -167,7 +226,7 @@ func TestOpenSegmentDescriptorSurvivesUnlink(t *testing.T) { if err := os.Remove(path); err != nil { t.Fatalf("remove opened segment: %v", err) } - got, err := io.ReadAll(segment) + got, err := io.ReadAll(segment.File) if err != nil { t.Fatalf("read opened segment: %v", err) } @@ -176,6 +235,36 @@ func TestOpenSegmentDescriptorSurvivesUnlink(t *testing.T) { } } +func TestOpenSegmentLeaseCapturesGeneration(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "seg_00001.ts") + writePrunerTestFile(t, path, []byte("complete segment"), time.Now()) + session := &TranscodeSession{outputDir: dir, segmentGeneration: 7} + + segment, err := session.OpenSegment("seg_00001.ts") + if err != nil { + t.Fatalf("OpenSegment: %v", err) + } + defer func() { _ = segment.Close() }() + if segment.Generation != 7 { + t.Fatalf("segment generation = %d, want 7", segment.Generation) + } +} + +func TestOpenSegmentWaitsThroughRestartInsteadOfLeasingStaleFile(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "seg_00001.ts") + writePrunerTestFile(t, path, []byte("stale segment"), time.Now()) + session := &TranscodeSession{outputDir: dir, restarting: true} + + if segment, err := session.OpenSegment("seg_00001.ts"); !errors.Is(err, ErrSegmentNotFound) { + if segment != nil { + _ = segment.Close() + } + t.Fatalf("OpenSegment error = %v, want ErrSegmentNotFound", err) + } +} + func writePrunerTestFile(t *testing.T, path string, contents []byte, modTime time.Time) { t.Helper() if err := os.WriteFile(path, contents, 0o600); err != nil { diff --git a/internal/playback/transcode.go b/internal/playback/transcode.go index 315fe5c8..05abff87 100644 --- a/internal/playback/transcode.go +++ b/internal/playback/transcode.go @@ -94,6 +94,7 @@ type TranscodeSession struct { stdinPipe io.WriteCloser lastRequestedSegment int lastPruneFloor int + lastPruneHighWater int segmentPruneRunning bool segmentGeneration uint64 throttler *TranscodeThrottler @@ -207,6 +208,7 @@ func StartTranscode(ctx context.Context, opts TranscodeOpts) (*TranscodeSession, stderr: newBoundedTailBuffer(stderrTailMaxBytes), lastRequestedSegment: opts.StartSegmentNumber, lastPruneFloor: opts.StartSegmentNumber - 1, + lastPruneHighWater: opts.StartSegmentNumber, reserveHWDeviceOnRestart: reserveHWDeviceOnRestart, } @@ -1625,29 +1627,48 @@ func (s *TranscodeSession) GetSegment(name string) (string, error) { return segPath, nil } -// OpenSegment opens a completed segment for serving. Keeping the descriptor -// open while the response is written makes it a filesystem-backed lease: a +// SegmentLease binds an opened segment descriptor to the FFmpeg generation it +// came from. Keeping the descriptor open makes it a filesystem-backed lease: a // concurrent prune may unlink the directory entry, but the response can still // read the complete file on POSIX filesystems. -func (s *TranscodeSession) OpenSegment(name string) (*os.File, os.FileInfo, error) { +type SegmentLease struct { + File *os.File + Info os.FileInfo + Generation uint64 +} + +// Close releases the opened segment descriptor. +func (l *SegmentLease) Close() error { + return l.File.Close() +} + +// OpenSegment opens a completed segment for serving and captures its FFmpeg +// generation atomically with the open. +func (s *TranscodeSession) OpenSegment(name string) (*SegmentLease, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.restarting { + return nil, ErrSegmentNotFound + } + clean := filepath.Base(name) segment, err := os.Open(filepath.Join(s.outputDir, clean)) if err != nil { if os.IsNotExist(err) { - return nil, nil, ErrSegmentNotFound + return nil, ErrSegmentNotFound } - return nil, nil, fmt.Errorf("open segment: %w", err) + return nil, fmt.Errorf("open segment: %w", err) } info, err := segment.Stat() if err != nil { _ = segment.Close() - return nil, nil, fmt.Errorf("stat segment: %w", err) + return nil, fmt.Errorf("stat segment: %w", err) } if info.Size() <= 0 { _ = segment.Close() - return nil, nil, ErrSegmentNotFound + return nil, ErrSegmentNotFound } - return segment, info, nil + return &SegmentLease{File: segment, Info: info, Generation: s.segmentGeneration}, nil } // Close terminates the ffmpeg process and removes the temporary output directory. @@ -1762,20 +1783,35 @@ func (s *TranscodeSession) cleanStaleSegments(startSegment int) { } } -// cleanSkippedSegments removes media that a forward seek jumped past. The -// prior back buffer cannot remain eligible for the normal pruner after the -// process start number advances, so leaving it in place would accumulate one -// abandoned retention window per forward seek. Open response descriptors keep -// any in-flight transfer valid after the directory entries are unlinked. -func (s *TranscodeSession) cleanSkippedSegments(startSegment int) { - entries, err := os.ReadDir(s.outputDir) - if err != nil { +// cleanSkippedSegments removes media older than the retained window before a +// forward restart. The configured back buffer and the current process's startup +// files remain available for immediate backward recovery. +func (s *TranscodeSession) cleanSkippedSegments(downloadedThrough int, opts TranscodeOpts) { + if opts.SegmentRetentionSeconds <= 0 { return } + floor, complete, err := segmentRetentionFloor(s.outputDir, opts, downloadedThrough) + if err != nil { + slog.Warn("resolve skipped transcode retention floor", "component", "playback", "error", err, "session", opts.SessionID, "playback_session_id", opts.SessionID) + return + } + if !complete { + return + } + + entries, err := os.ReadDir(s.outputDir) + if err != nil { + slog.Warn("read transcode directory for skipped cleanup", "component", "playback", "error", err, "session", opts.SessionID, "playback_session_id", opts.SessionID) + return + } + startupEnd := opts.StartSegmentNumber + startupSegmentRequirement(opts) for _, entry := range entries { segNum, parseErr := ParseSegmentNumber(entry.Name()) - if parseErr == nil && segNum < startSegment { - _ = os.Remove(filepath.Join(s.outputDir, entry.Name())) + if parseErr != nil || segNum < startupEnd || segNum >= floor { + continue + } + if removeErr := os.Remove(filepath.Join(s.outputDir, entry.Name())); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) { + slog.Warn("remove skipped transcode segment", "component", "playback", "error", removeErr, "segment", segNum, "session", opts.SessionID, "playback_session_id", opts.SessionID) } } } @@ -1824,6 +1860,7 @@ func (s *TranscodeSession) restart( previousRequestedSegment := s.lastRequestedSegment s.lastRequestedSegment = startSegment s.lastPruneFloor = startSegment - 1 + s.lastPruneHighWater = startSegment cancelCurrent := s.cancel done := s.done s.mu.Unlock() @@ -1849,7 +1886,7 @@ func (s *TranscodeSession) restart( s.mu.Unlock() if startSegment > previousRequestedSegment { - s.cleanSkippedSegments(startSegment) + s.cleanSkippedSegments(previousRequestedSegment, opts) } // Copy-mode restarts must clean stale segments so ffmpeg writes fresh @@ -1990,15 +2027,15 @@ func (s *TranscodeSession) WaitForSegment(name string, timeout time.Duration) (s // WaitForOpenSegment is the opened-file counterpart to WaitForSegment. It // closes the stat-to-open race for HTTP serving while preserving the same // restart and ffmpeg-exit error contract. -func (s *TranscodeSession) WaitForOpenSegment(name string, timeout time.Duration) (*os.File, os.FileInfo, error) { +func (s *TranscodeSession) WaitForOpenSegment(name string, timeout time.Duration) (*SegmentLease, error) { deadline := time.After(timeout) for { - segment, info, err := s.OpenSegment(name) + segment, err := s.OpenSegment(name) if err == nil { - return segment, info, nil + return segment, nil } if !errors.Is(err, ErrSegmentNotFound) { - return nil, nil, err + return nil, err } s.mu.Lock() @@ -2010,21 +2047,21 @@ func (s *TranscodeSession) WaitForOpenSegment(name string, timeout time.Duration if restarting { select { case <-deadline: - return nil, nil, ErrSegmentNotFound + return nil, ErrSegmentNotFound case <-time.After(100 * time.Millisecond): continue } } if !running && waitErr != nil { - return nil, nil, fmt.Errorf("%w: %w", ErrTranscodeFailed, waitErr) + return nil, fmt.Errorf("%w: %w", ErrTranscodeFailed, waitErr) } if !running { - return nil, nil, ErrSegmentNotFound + return nil, ErrSegmentNotFound } select { case <-deadline: - return nil, nil, ErrSegmentNotFound + return nil, ErrSegmentNotFound case <-time.After(100 * time.Millisecond): } } diff --git a/internal/playback/transcode_restart_guard_test.go b/internal/playback/transcode_restart_guard_test.go index 226587c1..abae0d15 100644 --- a/internal/playback/transcode_restart_guard_test.go +++ b/internal/playback/transcode_restart_guard_test.go @@ -9,14 +9,57 @@ import ( "time" ) -func TestForwardRestartRemovesSkippedBackBuffer(t *testing.T) { +func TestForwardRestartPreservesConfiguredBackBuffer(t *testing.T) { truePath, err := exec.LookPath("true") if err != nil { t.Skipf("`true` not found in PATH: %v", err) } dir := t.TempDir() - for _, segment := range []int{3, 40, 69, 70, 100} { + 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) + } + } + session := &TranscodeSession{ + outputDir: dir, + lastRequestedSegment: 70, + opts: TranscodeOpts{ + OutputDir: dir, + TargetCodecVideo: "h264", + SegmentDuration: 4, + SegmentRetentionSeconds: 20, + StartSegmentNumber: 0, + FFmpegPath: truePath, + }, + } + + if err := session.Restart(context.Background(), 400, 100); err != nil { + t.Fatalf("Restart: %v", err) + } + for _, segment := range []int{3, 40, 64} { + path := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})) + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Errorf("skipped segment %d survived forward restart: %v", segment, err) + } + } + for _, segment := range []int{65, 69, 70, 100} { + path := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})) + if _, err := os.Stat(path); err != nil { + t.Errorf("retained segment %d was removed: %v", segment, err) + } + } +} + +func TestForwardRestartKeepsSegmentsWhenRetentionDisabled(t *testing.T) { + truePath, err := exec.LookPath("true") + if err != nil { + t.Skipf("`true` not found in PATH: %v", err) + } + + dir := t.TempDir() + for _, segment := range []int{3, 40, 70} { name := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})) if err := os.WriteFile(name, []byte("segment"), 0o600); err != nil { t.Fatal(err) @@ -37,15 +80,12 @@ func TestForwardRestartRemovesSkippedBackBuffer(t *testing.T) { if err := session.Restart(context.Background(), 400, 100); err != nil { t.Fatalf("Restart: %v", err) } - for _, segment := range []int{3, 40, 69, 70} { + for _, segment := range []int{3, 40, 70} { path := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})) - if _, err := os.Stat(path); !os.IsNotExist(err) { - t.Errorf("skipped segment %d survived forward restart: %v", segment, err) + if _, err := os.Stat(path); err != nil { + t.Errorf("disabled retention removed segment %d: %v", segment, err) } } - if _, err := os.Stat(filepath.Join(dir, segmentFilename(100, TranscodeOpts{}))); err != nil { - t.Fatalf("segment at restart point was removed: %v", err) - } } // TestSegmentRecoveryDecisionWaitsWhileRestarting covers half of issue #243's diff --git a/internal/transcodenode/server.go b/internal/transcodenode/server.go index 4a220ba6..4534520f 100644 --- a/internal/transcodenode/server.go +++ b/internal/transcodenode/server.go @@ -894,8 +894,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) { s.touchSession(sessionID) } - downloadGeneration := session.SegmentGeneration() - segment, segmentInfo, err := session.OpenSegment(name) + segmentLease, err := session.OpenSegment(name) if err != nil && err == playback.ErrSegmentNotFound { segNum, parseErr := playback.ParseSegmentNumber(name) if parseErr == nil { @@ -930,7 +929,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) { "session", sessionID, "playback_session_id", sessionID, ) - segment, segmentInfo, err = session.WaitForOpenSegment(name, decision.WaitTimeout) + segmentLease, err = session.WaitForOpenSegment(name, decision.WaitTimeout) if err != nil && err == playback.ErrSegmentNotFound { slog.InfoContext(r.Context(), "transcode segment wait timeout", "component", "transcodenode", "segment", name, @@ -975,7 +974,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) { seekSeconds, segNum, ); restartErr == nil { - segment, segmentInfo, err = session.WaitForOpenSegment(name, 30*time.Second) + segmentLease, err = session.WaitForOpenSegment(name, 30*time.Second) } } if !ok && session.IsCopyVideo() { @@ -985,7 +984,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) { } else if session.IsRunning() { // Non-numbered segment (e.g., init.mp4 for fMP4 HLS). // Wait briefly — the init segment is written almost immediately. - segment, segmentInfo, err = session.WaitForOpenSegment(name, 10*time.Second) + segmentLease, err = session.WaitForOpenSegment(name, 10*time.Second) } } if err != nil { @@ -995,13 +994,13 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) { w.Header().Set("Cache-Control", "no-store, max-age=0") w.Header().Set("Pragma", "no-cache") - defer func() { _ = segment.Close() }() + defer func() { _ = segmentLease.Close() }() sw := httpstream.NewRollingDeadlineWriter(w) - http.ServeContent(sw, r, segmentInfo.Name(), segmentInfo.ModTime(), segment) + http.ServeContent(sw, r, segmentLease.Info.Name(), segmentLease.Info.ModTime(), segmentLease.File) if r.Method == http.MethodGet && - sw.CompletedFullResponse(r.Context(), segmentInfo.Size()) { + sw.CompletedFullResponse(r.Context(), segmentLease.Info.Size()) { if segmentNumber, parseErr := playback.ParseSegmentNumber(name); parseErr == nil { - session.ReportSegmentDownloadedForGeneration(segmentNumber, downloadGeneration) + session.ReportSegmentDownloadedForGeneration(segmentNumber, segmentLease.Generation) } } }