Add rolling transcode segment cleanup

Author: tces1

README.md

@@ -98,6 +98,7 @@ Copy `config.example.json` and change the upstream URL:
     "max_sessions": 2,
     "buffer_pause_seconds": 300,
     "buffer_resume_seconds": 120,
+    "segment_retention_seconds": 300,
     "idle_timeout_seconds": 60
   }
 }
@@ -117,5 +118,6 @@ EmbyTranscoder keeps local FFmpeg sessions tied to Emby playback check-ins:
 - HLS playlist and segment requests refresh media activity.
 - When transcoded media gets more than `buffer_pause_seconds` ahead of playback, FFmpeg is paused.
 - When buffered media falls back under `buffer_resume_seconds`, FFmpeg resumes.
+- Segments older than `segment_retention_seconds` behind the current playback position are deleted from the local cache.
 - If neither playback activity nor HLS access arrives before `idle_timeout_seconds`, the idle reaper stops the session.
 - A new `master.m3u8` request with a different upstream stream URL, such as a seek with a different `StartTimeTicks`, restarts the local session.

config.example.json

@@ -16,6 +16,7 @@
     "max_sessions": 2,
     "buffer_pause_seconds": 300,
     "buffer_resume_seconds": 120,
+    "segment_retention_seconds": 300,
     "idle_timeout_seconds": 60
   },
   "clients": [

docker/README.md

@@ -30,10 +30,11 @@ Edit `config/config.json` before startup, or copy it to `config/config.local.jso
 - video output is capped at 1920x1080 while preserving aspect ratio
 - playbackinfo rewrites prewarm the transcode session before the first playlist request
 - ffmpeg runs with low-latency startup and GOP settings to reduce first-segment delay
+- old `segment_*.ts` files are deleted once they are more than `transcode.segment_retention_seconds` behind playback
 
 For GitHub Actions publishing to Docker Hub, set `DOCKERHUB_USERNAME` and `DOCKERHUB_TOKEN` in the repository secrets.
 
-The transcode cache is stored in `docker/data/transcode`.
+The transcode cache is stored in `docker/data/transcode`; active sessions keep only the retained back-buffer plus the forward buffer.
 
 ## Operations
 

docker/config/config.json

@@ -16,6 +16,7 @@
     "max_sessions": 2,
     "buffer_pause_seconds": 300,
     "buffer_resume_seconds": 120,
+    "segment_retention_seconds": 300,
     "idle_timeout_seconds": 60
   },
   "clients": [

internal/config/config.go

@@ -26,18 +26,20 @@ type Upstream struct {
 }
 
 type Transcode struct {
-	Enabled             bool          `json:"enabled"`
-	FFmpegPath          string        `json:"ffmpeg_path"`
-	TempDir             string        `json:"temp_dir"`
-	HardwareDecode      string        `json:"hardware_decode"`
-	HardwareDevice      string        `json:"hardware_device"`
-	MaxSessions         int           `json:"max_sessions"`
-	BufferPauseSeconds  int           `json:"buffer_pause_seconds"`
-	BufferResumeSeconds int           `json:"buffer_resume_seconds"`
-	IdleTimeoutSeconds  int           `json:"idle_timeout_seconds"`
-	BufferPause         time.Duration `json:"-"`
-	BufferResume        time.Duration `json:"-"`
-	IdleTimeout         time.Duration `json:"-"`
+	Enabled                 bool          `json:"enabled"`
+	FFmpegPath              string        `json:"ffmpeg_path"`
+	TempDir                 string        `json:"temp_dir"`
+	HardwareDecode          string        `json:"hardware_decode"`
+	HardwareDevice          string        `json:"hardware_device"`
+	MaxSessions             int           `json:"max_sessions"`
+	BufferPauseSeconds      int           `json:"buffer_pause_seconds"`
+	BufferResumeSeconds     int           `json:"buffer_resume_seconds"`
+	SegmentRetentionSeconds int           `json:"segment_retention_seconds"`
+	IdleTimeoutSeconds      int           `json:"idle_timeout_seconds"`
+	BufferPause             time.Duration `json:"-"`
+	BufferResume            time.Duration `json:"-"`
+	SegmentRetention        time.Duration `json:"-"`
+	IdleTimeout             time.Duration `json:"-"`
 }
 
 func (t *Transcode) UnmarshalJSON(data []byte) error {
@@ -85,13 +87,14 @@ func Default() Config {
 			URL: "http://127.0.0.1:8096",
 		},
 		Transcode: Transcode{
-			Enabled:             true,
-			FFmpegPath:          "/usr/bin/ffmpeg",
-			TempDir:             "/var/lib/emby-transcoder/transcode",
-			MaxSessions:         2,
-			BufferPauseSeconds:  300,
-			BufferResumeSeconds: 120,
-			IdleTimeoutSeconds:  60,
+			Enabled:                 true,
+			FFmpegPath:              "/usr/bin/ffmpeg",
+			TempDir:                 "/var/lib/emby-transcoder/transcode",
+			MaxSessions:             2,
+			BufferPauseSeconds:      300,
+			BufferResumeSeconds:     120,
+			SegmentRetentionSeconds: 300,
+			IdleTimeoutSeconds:      60,
 		},
 		Clients: []ClientProfile{
 			{Name: "emby_android_tv", Match: []string{"Emby for Android TV", "Android TV"}, Transcode: true},
@@ -149,10 +152,14 @@ func normalize(cfg *Config) {
 	if cfg.Transcode.BufferResumeSeconds <= 0 {
 		cfg.Transcode.BufferResumeSeconds = 120
 	}
+	if cfg.Transcode.SegmentRetentionSeconds <= 0 {
+		cfg.Transcode.SegmentRetentionSeconds = 300
+	}
 	if cfg.Transcode.IdleTimeoutSeconds <= 0 {
 		cfg.Transcode.IdleTimeoutSeconds = 60
 	}
 	cfg.Transcode.BufferPause = time.Duration(cfg.Transcode.BufferPauseSeconds) * time.Second
 	cfg.Transcode.BufferResume = time.Duration(cfg.Transcode.BufferResumeSeconds) * time.Second
+	cfg.Transcode.SegmentRetention = time.Duration(cfg.Transcode.SegmentRetentionSeconds) * time.Second
 	cfg.Transcode.IdleTimeout = time.Duration(cfg.Transcode.IdleTimeoutSeconds) * time.Second
 }

internal/config/config_test.go

@@ -32,6 +32,9 @@ func TestDefaultConfigIsUsable(t *testing.T) {
 	if cfg.Transcode.BufferResumeSeconds != 120 {
 		t.Fatalf("buffer resume seconds = %d", cfg.Transcode.BufferResumeSeconds)
 	}
+	if cfg.Transcode.SegmentRetentionSeconds != 300 {
+		t.Fatalf("segment retention seconds = %d", cfg.Transcode.SegmentRetentionSeconds)
+	}
 	if cfg.Server.Debug {
 		t.Fatal("server debug should default false")
 	}
@@ -42,7 +45,7 @@ func TestLoadMergesJSONOverDefaults(t *testing.T) {
 	err := os.WriteFile(path, []byte(`{
 		"server": {"listen": ":9000", "debug": true},
 		"upstream": {"url": "http://emby.local:8096"},
-		"transcode": {"max_sessions": 4, "buffer_pause_seconds": 600, "buffer_resume_seconds": 90},
+		"transcode": {"max_sessions": 4, "buffer_pause_seconds": 600, "buffer_resume_seconds": 90, "segment_retention_seconds": 180},
 		"clients": [{"name": "yamby", "match": ["Yamby"], "transcode": true}]
 	}`), 0o600)
 	if err != nil {
@@ -69,6 +72,9 @@ func TestLoadMergesJSONOverDefaults(t *testing.T) {
 	if cfg.Transcode.BufferPauseSeconds != 600 || cfg.Transcode.BufferResumeSeconds != 90 {
 		t.Fatalf("buffer thresholds = %d/%d", cfg.Transcode.BufferPauseSeconds, cfg.Transcode.BufferResumeSeconds)
 	}
+	if cfg.Transcode.SegmentRetentionSeconds != 180 {
+		t.Fatalf("segment retention seconds = %d", cfg.Transcode.SegmentRetentionSeconds)
+	}
 	if !cfg.Server.Debug {
 		t.Fatal("server debug should be loaded from config")
 	}

internal/proxy/server.go

@@ -52,6 +52,7 @@ func NewWithTransport(cfg config.Config, transport http.RoundTripper) (*Server,
 		HardwareDevice:        cfg.Transcode.HardwareDevice,
 		BufferPauseThreshold:  cfg.Transcode.BufferPause,
 		BufferResumeThreshold: cfg.Transcode.BufferResume,
+		SegmentRetention:      cfg.Transcode.SegmentRetention,
 		IdleTimeout:           cfg.Transcode.IdleTimeout,
 	})
 	if err != nil {

internal/transcode/handler.go

@@ -145,6 +145,7 @@ func (h Handler) sessionForSegment(id string, segmentIndex int, name string, r *
 	}
 	request := requestFromHTTP(id, inputURL, segmentReq)
 	request.SegmentStartIndex = segmentIndex
+	request.SegmentRequest = true
 	request.RequestedStartTimeTicks = int64Query(r.URL.Query().Get("StartTimeTicks"))
 	if request.RequestedStartTimeTicks == 0 {
 		request.RequestedStartTimeTicks = request.StartTimeTicks
@@ -188,12 +189,15 @@ func (h Handler) startupWait() time.Duration {
 }
 
 func segmentReusable(session *Session, segmentIndex int, name string) bool {
-	if fileExists(filepath.Join(session.Dir, name)) {
-		return true
+	if segmentIndex < session.OldestSegmentKept {
+		return false
 	}
 	if segmentIndex < session.SegmentStartIndex {
 		return false
 	}
+	if fileExists(filepath.Join(session.Dir, name)) {
+		return true
+	}
 	highest := session.HighestSegmentSeen
 	if highest < session.SegmentStartIndex {
 		highest = session.SegmentStartIndex

internal/transcode/manager.go

@@ -35,6 +35,7 @@ type Options struct {
 	HardwareDevice        string
 	BufferPauseThreshold  time.Duration
 	BufferResumeThreshold time.Duration
+	SegmentRetention      time.Duration
 	BufferCheckInterval   time.Duration
 	IdleTimeout           time.Duration
 	ReapInterval          time.Duration
@@ -53,6 +54,7 @@ type Request struct {
 	StartTimeTicks          int64
 	RequestedStartTimeTicks int64
 	SegmentStartIndex       int
+	SegmentRequest          bool
 	Media                   MediaInfo
 }
 
@@ -73,6 +75,7 @@ type Session struct {
 	StartTimeTicks          int64
 	RequestedStartTimeTicks int64
 	SegmentStartIndex       int
+	OldestSegmentKept       int
 	HighestSegmentSeen      int
 	Media                   MediaInfo
 	Dir                     string
@@ -155,6 +158,9 @@ func normalizeManagerOptions(options Options) Options {
 	if options.BufferResumeThreshold <= 0 {
 		options.BufferResumeThreshold = 2 * time.Minute
 	}
+	if options.SegmentRetention <= 0 {
+		options.SegmentRetention = 5 * time.Minute
+	}
 	if options.BufferCheckInterval <= 0 {
 		options.BufferCheckInterval = time.Second
 	}
@@ -410,6 +416,7 @@ func (m *Manager) RecordProgress(event PlaybackEvent) int {
 	now := time.Now()
 	count := 0
 	var actions []bufferAction
+	var cleanups []segmentCleanupTask
 	for _, session := range m.sessions {
 		if !matchesPlayback(session, event) {
 			continue
@@ -430,9 +437,15 @@ func (m *Manager) RecordProgress(event PlaybackEvent) int {
 		if action, ok := m.bufferActionLocked(session); ok {
 			actions = append(actions, action)
 		}
+		if cleanup, ok := m.segmentCleanupTaskLocked(session); ok {
+			cleanups = append(cleanups, cleanup)
+		}
 		count++
 	}
 	m.mu.Unlock()
+	for _, cleanup := range cleanups {
+		cleanupOldSegments(cleanup)
+	}
 	for _, action := range actions {
 		m.applyBufferAction(action)
 	}
@@ -535,6 +548,9 @@ func (m *Manager) ReconcileBuffers() {
 }
 
 func shouldRestart(session *Session, request Request) bool {
+	if request.SegmentRequest && session.OldestSegmentKept > session.SegmentStartIndex && request.SegmentStartIndex < session.OldestSegmentKept {
+		return true
+	}
 	if request.StartTimeTicks != session.StartTimeTicks {
 		return true
 	}
@@ -571,6 +587,9 @@ func touchSession(session *Session, request Request, now time.Time, mediaAccess
 	session.StartTimeTicks = request.StartTimeTicks
 	session.RequestedStartTimeTicks = request.RequestedStartTimeTicks
 	session.SegmentStartIndex = request.SegmentStartIndex
+	if session.OldestSegmentKept < request.SegmentStartIndex {
+		session.OldestSegmentKept = request.SegmentStartIndex
+	}
 	if session.HighestSegmentSeen < request.SegmentStartIndex {
 		session.HighestSegmentSeen = request.SegmentStartIndex
 	}
@@ -591,6 +610,61 @@ type bufferAction struct {
 	bufferTicks int64
 }
 
+type segmentCleanupTask struct {
+	sessionID     string
+	dir           string
+	beforeSegment int
+}
+
+func (m *Manager) segmentCleanupTaskLocked(session *Session) (segmentCleanupTask, bool) {
+	if session == nil || session.Dir == "" || m.options.SegmentRetention <= 0 {
+		return segmentCleanupTask{}, false
+	}
+	retentionTicks := durationToTicks(m.options.SegmentRetention)
+	if retentionTicks <= 0 || session.PositionTicks <= retentionTicks {
+		return segmentCleanupTask{}, false
+	}
+	cutoffTicks := session.PositionTicks - retentionTicks
+	beforeSegment := int(cutoffTicks / defaultSegmentTicks)
+	if beforeSegment <= session.OldestSegmentKept {
+		return segmentCleanupTask{}, false
+	}
+	session.OldestSegmentKept = beforeSegment
+	return segmentCleanupTask{sessionID: session.ID, dir: session.Dir, beforeSegment: beforeSegment}, true
+}
+
+func cleanupOldSegments(task segmentCleanupTask) {
+	entries, err := os.ReadDir(task.dir)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			logging.Errorf("transcode cleanup error id=%s err=%v", task.sessionID, err)
+		}
+		return
+	}
+
+	deleted := 0
+	for _, entry := range entries {
+		if entry.IsDir() {
+			continue
+		}
+		segmentIndex, ok := segmentIndexFromName(entry.Name())
+		if !ok || segmentIndex >= task.beforeSegment {
+			continue
+		}
+		path := filepath.Join(task.dir, entry.Name())
+		if err := os.Remove(path); err != nil {
+			if !os.IsNotExist(err) {
+				logging.Errorf("transcode cleanup error id=%s file=%s err=%v", task.sessionID, entry.Name(), err)
+			}
+			continue
+		}
+		deleted++
+	}
+	if deleted > 0 {
+		logging.Infof("transcode cleanup id=%s deleted_segments=%d before_segment=%d", task.sessionID, deleted, task.beforeSegment)
+	}
+}
+
 func (m *Manager) bufferActionLocked(session *Session) (bufferAction, bool) {
 	if session == nil || session.process == nil {
 		return bufferAction{}, false