haeaeae/silo-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(playback): keep restart buffers prunable

Browse Source
This commit is contained in:
Quick104
2026-08-11 13:42:43 -04:00
parent 7f6225bb11
commit e05c30c637
5 changed files with 73 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/handlers/playback.go
+5 -4
View File
@@ -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) {
internal/api/handlers/playback_config_test.go
+19
View File
@@ -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)
}
}
internal/playback/segment_pruner.go
+22 -7
View File
@@ -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()
internal/playback/transcode.go
+14 -3
View File
@@ -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)
}
internal/playback/transcode_restart_guard_test.go
+13
View File
@@ -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) {