diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index 78c99192..8efff8f9 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -3819,9 +3819,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht sw := httpstream.NewRollingDeadlineWriter(w) http.ServeContent(sw, r, segmentInfo.Name(), segmentInfo.ModTime(), segmentFile) if r.Method == http.MethodGet && - sw.StatusCode() == http.StatusOK && - sw.BytesWritten() == segmentInfo.Size() && - sw.Outcome(r.Context()) == httpstream.OutcomeCompleted { + sw.CompletedFullResponse(r.Context(), segmentInfo.Size()) { if segNum, parseErr := playback.ParseSegmentNumber(segmentName); parseErr == nil { transcodeSession.ReportSegmentDownloadedForGeneration(segNum, downloadGeneration) } diff --git a/internal/httpstream/rolling_deadline.go b/internal/httpstream/rolling_deadline.go index f406aa4a..2f376441 100644 --- a/internal/httpstream/rolling_deadline.go +++ b/internal/httpstream/rolling_deadline.go @@ -12,6 +12,7 @@ package httpstream import ( "context" "errors" + "fmt" "io" "net" "net/http" @@ -171,6 +172,24 @@ func (s *RollingDeadlineWriter) BytesWritten() int64 { return s.bytesWritten } +// CompletedFullResponse reports whether the complete representation was +// accepted by the transport. ServeContent returns 206 for a whole-file Range +// request such as "bytes=0-", so byte count alone is not enough to distinguish +// that successful transfer from a partial range. +func (s *RollingDeadlineWriter) CompletedFullResponse(ctx context.Context, fullSize int64) bool { + if fullSize <= 0 || s.bytesWritten != fullSize || s.Outcome(ctx) != OutcomeCompleted { + return false + } + switch s.statusCode { + case http.StatusOK: + return true + case http.StatusPartialContent: + return s.Header().Get("Content-Range") == fmt.Sprintf("bytes 0-%d/%d", fullSize-1, fullSize) + default: + return false + } +} + // Outcome classifies the first write failure, or a canceled request when no // write failure was surfaced by the transport. func (s *RollingDeadlineWriter) Outcome(ctx context.Context) StreamOutcome { diff --git a/internal/httpstream/rolling_deadline_test.go b/internal/httpstream/rolling_deadline_test.go index d64c548d..72efe289 100644 --- a/internal/httpstream/rolling_deadline_test.go +++ b/internal/httpstream/rolling_deadline_test.go @@ -271,6 +271,35 @@ func TestServeContentReadFromOutcomeCompletedAndCounted(t *testing.T) { } } +func TestCompletedFullResponse(t *testing.T) { + const body = "complete segment" + tests := []struct { + name string + rangeHeader string + method string + want bool + }{ + {name: "ordinary get", method: http.MethodGet, want: true}, + {name: "open ended full range", method: http.MethodGet, rangeHeader: "bytes=0-", want: true}, + {name: "explicit full range", method: http.MethodGet, rangeHeader: "bytes=0-15", want: true}, + {name: "partial range", method: http.MethodGet, rangeHeader: "bytes=1-", want: false}, + {name: "head", method: http.MethodHead, want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + rr := httptest.NewRecorder() + sw := newRollingDeadlineWriter(rr, time.Second, 0) + req := httptest.NewRequest(tt.method, "/segment.ts", nil) + req.Header.Set("Range", tt.rangeHeader) + http.ServeContent(sw, req, "segment.ts", time.Time{}, strings.NewReader(body)) + if got := sw.CompletedFullResponse(req.Context(), int64(len(body))); got != tt.want { + t.Fatalf("CompletedFullResponse = %v, want %v (status=%d range=%q bytes=%d)", + got, tt.want, sw.StatusCode(), rr.Header().Get("Content-Range"), sw.BytesWritten()) + } + }) + } +} + // TestReadFromPreservesCompletion exercises the io.ReaderFrom path used by // http.ServeContent (sendfile) under a server WriteTimeout shorter than the // transfer, with a source large enough to require multiple bounded slices. diff --git a/internal/jellycompat/streams.go b/internal/jellycompat/streams.go index 605a803f..93d5cba2 100644 --- a/internal/jellycompat/streams.go +++ b/internal/jellycompat/streams.go @@ -538,9 +538,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques sw := httpstream.NewRollingDeadlineWriter(w) http.ServeContent(sw, r, segmentInfo.Name(), segmentInfo.ModTime(), segment) if r.Method == http.MethodGet && - sw.StatusCode() == http.StatusOK && - sw.BytesWritten() == segmentInfo.Size() && - sw.Outcome(r.Context()) == httpstream.OutcomeCompleted { + sw.CompletedFullResponse(r.Context(), segmentInfo.Size()) { if segNum, parseErr := playback.ParseSegmentNumber(name); parseErr == nil { transcodeSession.ReportSegmentDownloadedForGeneration(segNum, downloadGeneration) } diff --git a/internal/playback/transcode.go b/internal/playback/transcode.go index 9d1b12c9..315fe5c8 100644 --- a/internal/playback/transcode.go +++ b/internal/playback/transcode.go @@ -1762,6 +1762,24 @@ 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 { + return + } + for _, entry := range entries { + segNum, parseErr := ParseSegmentNumber(entry.Name()) + if parseErr == nil && segNum < startSegment { + _ = os.Remove(filepath.Join(s.outputDir, entry.Name())) + } + } +} + // Restart kills the current ffmpeg process and starts a new one seeking to // the given position. startSegment sets -hls_segment_start_number so that // output filenames align with the expected segment numbering. Existing @@ -1803,6 +1821,7 @@ func (s *TranscodeSession) restart( s.restarting = true s.segmentGeneration++ s.segmentPruneRunning = false + previousRequestedSegment := s.lastRequestedSegment s.lastRequestedSegment = startSegment s.lastPruneFloor = startSegment - 1 cancelCurrent := s.cancel @@ -1829,6 +1848,10 @@ func (s *TranscodeSession) restart( reserveHWDevice := s.reserveHWDeviceOnRestart s.mu.Unlock() + if startSegment > previousRequestedSegment { + s.cleanSkippedSegments(startSegment) + } + // Copy-mode restarts must clean stale segments so ffmpeg writes fresh // output. Encoded transcodes keep old segments for backward seek reuse. if strings.EqualFold(opts.TargetCodecVideo, "copy") { diff --git a/internal/playback/transcode_restart_guard_test.go b/internal/playback/transcode_restart_guard_test.go index cdcd76ab..226587c1 100644 --- a/internal/playback/transcode_restart_guard_test.go +++ b/internal/playback/transcode_restart_guard_test.go @@ -2,11 +2,52 @@ package playback import ( "context" + "os" "os/exec" + "path/filepath" "testing" "time" ) +func TestForwardRestartRemovesSkippedBackBuffer(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} { + 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, + 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, 69, 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(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 // seek-freeze: while a restart is already in flight, a concurrent segment // request must WAIT for the restart's output rather than trigger another diff --git a/internal/transcodenode/server.go b/internal/transcodenode/server.go index f3d04b7c..4a220ba6 100644 --- a/internal/transcodenode/server.go +++ b/internal/transcodenode/server.go @@ -999,9 +999,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) { sw := httpstream.NewRollingDeadlineWriter(w) http.ServeContent(sw, r, segmentInfo.Name(), segmentInfo.ModTime(), segment) if r.Method == http.MethodGet && - sw.StatusCode() == http.StatusOK && - sw.BytesWritten() == segmentInfo.Size() && - sw.Outcome(r.Context()) == httpstream.OutcomeCompleted { + sw.CompletedFullResponse(r.Context(), segmentInfo.Size()) { if segmentNumber, parseErr := playback.ParseSegmentNumber(name); parseErr == nil { session.ReportSegmentDownloadedForGeneration(segmentNumber, downloadGeneration) }