fix(playback): preserve retained segments across restarts

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)
 		}
 	}
 }

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)
 		}
 	}
 }

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()
 }

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 {

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):
 		}
 	}

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

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)
 		}
 	}
 }