fix(playback): trust completed segment writes

.env.example

@@ -2,9 +2,11 @@
 # For the default docker compose setup, MEDIA_ROOT is the main value you need to set.
 
 # Optional: layer compose files through env. Leave unset for CPU-only defaults.
-# Linux/macOS example for enabling NVIDIA override:
+# Linux example for Intel/AMD VA-API or Intel Quick Sync:
+# COMPOSE_FILE=docker-compose.yml:docker-compose.vaapi.yml
+# Linux example for NVIDIA NVENC:
 # COMPOSE_FILE=docker-compose.yml:docker-compose.nvidia.yml
-# Windows example:
+# Windows NVIDIA example:
 # COMPOSE_FILE=docker-compose.yml;docker-compose.nvidia.yml
 
 # Required for docker compose: host path to your media library.
@@ -27,6 +29,15 @@ POSTGRES_USER=silo
 POSTGRES_PASSWORD=silo
 POSTGRES_DB=silo
 POSTGRES_SHM_SIZE=8gb
+# POSTGRES_PORT=5432
+# REDIS_PORT=6379
+
+# Optional Meilisearch service. Generate a key, uncomment it, then start the
+# search profile with: docker compose --profile search up -d
+# MEILI_MASTER_KEY=replace-with-output-of-openssl-rand-hex-32
+# MEILISEARCH_IMAGE=getmeili/meilisearch:latest
+# MEILISEARCH_HOST=127.0.0.1
+# MEILISEARCH_PORT=7700
 
 # PostgreSQL auto-tuning is enabled by default in docker-compose.yml. Silo
 # applies pgtune-style OLTP recommendations through ALTER SYSTEM at startup.
@@ -40,7 +51,8 @@ POSTGRES_SHM_SIZE=8gb
 # POSTGRES_TUNE_DB_SIZE=auto
 # POSTGRES_TUNE_CONNECTIONS=100
 
-# Optional host port overrides for docker-compose.yml.
+# Optional host port overrides for docker-compose.yml. The container listeners
+# remain fixed, so these values only change the host side of each port mapping.
 # PORT=8090
 # Used only after Jellyfin-compatible app support is enabled by an admin.
 # JF_PORT=8096
@@ -77,7 +89,7 @@ POSTGRES_SHM_SIZE=8gb
 # Optional Redis override when running from source or pointing workers at an external Redis.
 # REDIS_URL=redis://localhost:6379
 
-# Optional standalone worker identity settings for docker-compose.remote-transcode.yml.
+# Optional standalone worker identity settings for custom worker deployments.
 # NODE_NAME=transcode-01
 # NODE_URL=http://transcode-01.example.com:8082
 

DEVELOPMENT.md

@@ -6,9 +6,10 @@ See [CONTRIBUTING.md](CONTRIBUTING.md) for contribution expectations, merge requ
 
 ## Prerequisites
 
-- **Go** 1.24+
-- **Bun** 1.0+
-- **PostgreSQL** 18+
+- **Go** 1.26.4+
+- **Node.js** 22+ with **pnpm** 10.32.1
+- **PostgreSQL** 18 with pgvector
+- **Redis**
 - **FFmpeg** (for transcoding support)
 
 ## Local Development
@@ -16,6 +17,13 @@ See [CONTRIBUTING.md](CONTRIBUTING.md) for contribution expectations, merge requ
 Local development remains intentionally separate from the deploy-oriented compose setup. Use [docker-compose.yml](docker-compose.yml) for local services and the source-build workflow below.
 
 ```sh
+# Create the local bootstrap configuration
+cp .env.example .env
+printf '\nSECRET_KEY=%s\nDATABASE_URL=%s\nREDIS_URL=%s\n' \
+  "$(openssl rand -base64 48)" \
+  'postgres://silo:silo@localhost:5432/silo?sslmode=disable' \
+  'redis://localhost:6379' >> .env
+
 # Start local PostgreSQL and Redis
 docker compose up -d postgres redis
 
@@ -26,7 +34,8 @@ make dev-frontend
 make dev-backend
 ```
 
-The main compose file expects `MEDIA_ROOT` to be set even if you only want the bundled PostgreSQL and Redis services, so set that in `.env` first (`cp .env.example .env`).
+The template supplies a non-empty `MEDIA_ROOT` because Compose validates the whole file even when
+you start only PostgreSQL and Redis. Change it before testing libraries against real media.
 
 If you are developing `Silo` and `silo-plugin-sdk` together, keep using the local [`go.work`](go.work) workspace. That workspace is a developer convenience only. CI and release builds run with `GOWORK=off`, so any new SDK helper used here must be pushed and tagged in `silo-plugin-sdk` before this repo can merge or release the change.
 
@@ -79,7 +88,7 @@ database URL should be read from a non-default env file.
 go test ./...
 
 # Frontend tests
-cd web && bun test
+cd web && pnpm test
 ```
 
 ## Linting
@@ -89,8 +98,8 @@ cd web && bun test
 golangci-lint run
 
 # Frontend
-cd web && bun run lint
-cd web && bun run format:check
+cd web && pnpm run lint
+cd web && pnpm run format:check
 ```
 
 ## Project Structure

README.md

@@ -15,14 +15,22 @@ Join the community on [Discord](https://discord.gg/4RxuUQAEnW). If Silo is usefu
 
 ## Deploy with Docker (recommended)
 
-The easiest way to run Silo is with Docker Compose. The default stack assumes you do not already have PostgreSQL and Redis available, so it bundles PostgreSQL, Redis, FFmpeg, and the application for a one-command start.
+The easiest way to run Silo is with Docker Compose 2.24 or newer. The default stack assumes you do
+not already have PostgreSQL and Redis available, so it bundles PostgreSQL, Redis, FFmpeg, and the
+application for a one-command start.
 
 1. **Create a `.env` file**
 
    ```sh
    cp .env.example .env
+   printf '\nPOSTGRES_PASSWORD=%s\nSECRET_KEY=%s\n' \
+     "$(openssl rand -hex 24)" "$(openssl rand -base64 48)" >> .env
    ```
 
+   This replaces the development database password from `.env.example` and creates the key Silo
+   uses to encrypt stored credentials. Back up `.env` separately from PostgreSQL; losing
+   `SECRET_KEY` makes those credentials unrecoverable.
+
 2. **Set your media path**
 
    Edit `.env` and set:
@@ -43,6 +51,21 @@ The easiest way to run Silo is with Docker Compose. The default stack assumes yo
 
    If you already have PostgreSQL and Redis available, omit those bundled service examples from compose and point Silo at your existing `DATABASE_URL` and `REDIS_URL` instead.
 
+   ### Optional Intel/AMD VA-API or Intel Quick Sync
+
+   The default stack is CPU-only so it starts on hosts without `/dev/dri`. On a Linux host with
+   `/dev/dri`, enable the device overlay:
+
+   ```sh
+   docker compose -f docker-compose.yml -f docker-compose.vaapi.yml up -d
+   ```
+
+   To make that the default for this installation, set:
+
+   ```dotenv
+   COMPOSE_FILE=docker-compose.yml:docker-compose.vaapi.yml
+   ```
+
    ### Optional NVIDIA/NVENC
 
    GPU support is kept out of the default compose file so hosts without NVIDIA drivers work unchanged.
@@ -78,32 +101,39 @@ By default, data is stored under `/opt/silo`:
 
 - `/opt/silo/postgres`
 - `/opt/silo/redis`
+- `/opt/silo/plugins`
+- `/opt/silo/compat`
 - `/opt/silo/transcode`
 - `/opt/silo/catalog-seeds`
 
+The optional `search` profile also stores its index under `/opt/silo/meilisearch`.
+
 Media is mounted into the container at `/mnt/media` from the host path you set in `MEDIA_ROOT`.
 
-### Optional Profiles
+### Optional Search Profile
 
-The main compose file is integrated-first. These profiles exist for operators testing distributed mode or mirroring a split deployment shape. Most single-host installs should stay on the default integrated service, because it already includes proxying and transcoding.
+PostgreSQL full-text search works without any optional services. Meilisearch is available when you
+want its search provider:
 
 | Profile | Command | Description |
 |---|---|---|
 | default | `docker compose up -d` | Integrated server plus bundled PostgreSQL and Redis |
-| `proxy` | `docker compose --profile proxy up -d` | Start a standalone proxy service for distributed-mode testing |
-| `transcode` | `docker compose --profile transcode up -d` | Start a standalone transcode service for distributed-mode testing |
+| `search` | `docker compose --profile search up -d` | Add the optional Meilisearch service |
 
-You can enable both optional examples together:
+Before starting the `search` profile, set `MEILI_MASTER_KEY` in `.env` to the output of
+`openssl rand -hex 32`. After Silo starts, choose Meilisearch under **Admin > Settings > Search**,
+set the URL to `http://meilisearch:7700`, enter the same key as the API key, test the connection,
+and save. Restart Silo, then rebuild the catalog search index from the same page. Silo continues
+to use PostgreSQL full-text search until you select Meilisearch.
 
-```sh
-docker compose --profile proxy --profile transcode up -d
-```
+### Distributed Examples
 
-If you are splitting workers across multiple hosts, use the separate remote worker example instead of trying to stretch the main compose file across machines.
+The main Compose file includes commented proxy and transcode service examples. Most single-host
+installs should leave them commented because the integrated service already includes proxying and
+transcoding.
 
-### Advanced Remote Node Example
-
-For a dedicated remote transcode worker, use [docker-compose.remote-transcode.yml](docker-compose.remote-transcode.yml). That file is intended for a separate worker host that connects back to an existing Silo deployment using shared PostgreSQL and Redis.
+Multi-host operators can use those examples as a starting point for a dedicated worker Compose
+file connected to the deployment's shared PostgreSQL and Redis services.
 
 ### Deployment Notes
 
@@ -116,7 +146,10 @@ helper and cutover guide in [docs/continuum-to-silo-docker-migration.md](docs/co
 
 ## Configuration
 
-Silo requires only a `DATABASE_URL` when running from source or against external infrastructure. In the default Docker Compose path, the stack wires the database and Redis URLs for you. All other settings — libraries, metadata providers, transcoding, users — are managed through the admin UI after first launch.
+Silo requires `DATABASE_URL` and `SECRET_KEY` when running from source or against external
+infrastructure. In the default Docker Compose path, the stack wires the database and Redis URLs
+for you. All other settings — libraries, metadata providers, transcoding, users — are managed
+through the admin UI after first launch.
 
 ### Server Modes
 
@@ -183,24 +216,26 @@ move fully to a custom `postgresql.conf`.
 
 If you prefer running Silo without Docker:
 
-1. **Install prerequisites**: Go 1.24+, Bun 1.0+, PostgreSQL 18+, and FFmpeg.
+1. **Install prerequisites**: Go 1.26.4+, Node.js 22+, pnpm 10.32.1, PostgreSQL 18 with pgvector, Redis, and FFmpeg.
 
-2. **Start PostgreSQL and Redis** (skip if you already have them running)
+2. **Configure the source process**
+
+   ```sh
+   cp .env.example .env
+   printf '\nSECRET_KEY=%s\nDATABASE_URL=%s\nREDIS_URL=%s\n' \
+     "$(openssl rand -base64 48)" \
+     'postgres://silo:silo@localhost:5432/silo?sslmode=disable' \
+     'redis://localhost:6379' >> .env
+   ```
+
+   Change the URLs when you use existing services instead of the bundled development defaults.
+
+3. **Start PostgreSQL and Redis** (skip if you already have them running)
 
    ```sh
    docker compose up -d postgres redis
    ```
 
-   The main compose file still expects `MEDIA_ROOT` to be set even if you only want the bundled PostgreSQL and Redis services, so set that in `.env` first.
-
-3. **Configure the database connection**
-
-   ```sh
-   cp .env.example .env
-   ```
-
-   Edit `.env` and set `DATABASE_URL` to point to your PostgreSQL instance.
-
 4. **Build and run**
 
    ```sh

docker-compose.dev.yml

@@ -123,7 +123,6 @@ services:
       - silo-dev-plugins:/var/lib/silo/plugins
       - silo-dev-compat:/var/lib/silo/compat
       - silo-dev-transcode:/tmp/silo-transcode
-      - silo-dev-audiobook-covers:/var/lib/silo/audiobook-covers
       - ${MEDIA_ROOT:-/tmp/silo-dev-media}:${MEDIA_CONTAINER_ROOT:-/mnt/media}:ro
     depends_on:
       postgres:
@@ -169,7 +168,6 @@ services:
 
 volumes:
   silo-dev-air-tmp:
-  silo-dev-audiobook-covers:
   silo-dev-compat:
   silo-dev-go-build:
   silo-dev-go-mod:

docker-compose.vaapi.yml

@@ -0,0 +1,10 @@
+services:
+  silo:
+    devices:
+      - /dev/dri:/dev/dri
+
+  # Optional distributed-mode example. Most single-host installs should
+  # just use the integrated service above.
+  # silo-transcode:
+  #   devices:
+  #     - /dev/dri:/dev/dri

docker-compose.yml

@@ -8,7 +8,7 @@ services:
       POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-silo}
       POSTGRES_DB: ${POSTGRES_DB:-silo}
     ports:
-      - "5432:5432"
+      - "127.0.0.1:${POSTGRES_PORT:-5432}:5432"
     volumes:
       - ${SILO_DATA_ROOT:-/opt/silo}/postgres:/var/lib/postgresql
     command: ["postgres", "-c", "listen_addresses=*"]
@@ -22,7 +22,7 @@ services:
     image: redis:alpine
     restart: unless-stopped
     ports:
-      - "6379:6379"
+      - "127.0.0.1:${REDIS_PORT:-6379}:6379"
     volumes:
       - ${SILO_DATA_ROOT:-/opt/silo}/redis:/data
     healthcheck:
@@ -56,6 +56,10 @@ services:
     restart: unless-stopped
     environment:
       MODE: integrated
+      # Keep the container listeners fixed. PORT and JF_PORT in .env are
+      # host-side published-port overrides.
+      PORT: "8080"
+      JF_PORT: "8096"
       # Master key for at-rest credential encryption. Required — the server
       # refuses to start without it. Generate with: openssl rand -base64 48
       # Back it up SEPARATELY from your database dumps; losing it makes encrypted
@@ -71,12 +75,10 @@ services:
       - "${ABS_PORT:-13378}:13378"
     volumes:
       - ${MEDIA_ROOT:?Set MEDIA_ROOT in .env to the host media path}:${MEDIA_CONTAINER_ROOT:-/mnt/media}:ro
-      - ${MEDIA_BOOKS_ROOT:-${MEDIA_ROOT}}:${MEDIA_BOOKS_CONTAINER_ROOT:-${MEDIA_CONTAINER_ROOT:-/mnt/media}/books}:ro
       - ${SILO_DATA_ROOT:-/opt/silo}/plugins:/var/lib/silo/plugins
       - ${SILO_DATA_ROOT:-/opt/silo}/compat:/var/lib/silo/compat
       - ${SILO_DATA_ROOT:-/opt/silo}/transcode:/tmp/silo-transcode
       - ${SILO_DATA_ROOT:-/opt/silo}/catalog-seeds:/catalog-seeds:ro
-      - ${SILO_DATA_ROOT:-/opt/silo}/audiobook-covers:/var/lib/silo/audiobook-covers
       - /proc/meminfo:/host/proc/meminfo:ro
     depends_on:
       postgres:
@@ -86,9 +88,6 @@ services:
     env_file:
       - path: .env
         required: false
-    devices:
-      - /dev/dri:/dev/dri
-
   # Optional distributed-mode example. Most single-host installs should
   # just use the integrated service above.
   # silo-proxy:

internal/api/handlers/playback.go

@@ -33,6 +33,7 @@ import (
 	"github.com/Silo-Server/silo-server/internal/settingsresolve"
 	"github.com/Silo-Server/silo-server/internal/streamtoken"
 	"github.com/Silo-Server/silo-server/internal/subtitles"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
 	"github.com/Silo-Server/silo-server/internal/userstore"
 	"github.com/Silo-Server/silo-server/internal/watchstate"
 	"github.com/Silo-Server/silo-server/internal/watchsync"
@@ -229,10 +230,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) {
@@ -1374,7 +1376,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 	h.touchSessionActivity(sessionID)
 
 	segmentName := chi.URLParam(r, "name")
-	segmentPath, err := transcodeSession.GetSegment(segmentName)
+	segmentLease, err := transcodeSession.OpenSegment(segmentName)
 	if err != nil && errors.Is(err, playback.ErrSegmentNotFound) {
 		segNum, parseErr := playback.ParseSegmentNumber(segmentName)
 		if parseErr == nil {
@@ -1411,7 +1413,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 					"session", sessionID,
 					"playback_session_id", sessionID,
 				)
-				segmentPath, err = transcodeSession.WaitForSegment(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,
@@ -1473,7 +1475,7 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 					); restartErr == nil {
 						// Throttler + exit monitor re-arm via the session's
 						// restart hook.
-						segmentPath, err = transcodeSession.WaitForSegment(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
@@ -1481,7 +1483,9 @@ 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)
-							_, _ = transcodeSession.WaitForSegment(nextSegmentName, 1200*time.Millisecond)
+							if nextSegment, nextErr := transcodeSession.WaitForOpenSegment(nextSegmentName, 1200*time.Millisecond); nextErr == nil {
+								_ = nextSegment.Close()
+							}
 						}
 					}
 				}
@@ -1489,7 +1493,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.
-			segmentPath, err = transcodeSession.WaitForSegment(segmentName, 10*time.Second)
+			segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 10*time.Second)
 		}
 	}
 	if err != nil {
@@ -1501,14 +1505,17 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 		return
 	}
 
-	// Report segment download for throttle tracking.
-	if segNum, parseErr := playback.ParseSegmentNumber(segmentName); parseErr == nil {
-		transcodeSession.ReportSegmentDownloaded(segNum)
-	}
-
 	w.Header().Set("Cache-Control", "no-store, max-age=0")
 	w.Header().Set("Pragma", "no-cache")
-	http.ServeFile(w, r, segmentPath)
+	defer func() { _ = segmentLease.Close() }()
+	sw := httpstream.NewRollingDeadlineWriter(w)
+	http.ServeContent(sw, r, segmentLease.Info.Name(), segmentLease.Info.ModTime(), segmentLease.File)
+	if r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(segmentLease.Info.Size()) {
+		if segNum, parseErr := playback.ParseSegmentNumber(segmentName); parseErr == nil {
+			transcodeSession.ReportSegmentDownloadedForGeneration(segNum, segmentLease.Generation)
+		}
+	}
 }
 
 // buildProxyManifestURL signs a stream token carrying the session's full
@@ -1531,6 +1538,9 @@ func (h *PlaybackHandler) buildProxyManifestURL(card playback.RecipeCard, proxyN
 func (h *PlaybackHandler) proxyToTranscodeNode(w http.ResponseWriter, r *http.Request, transcodeNodeURL, path string) {
 	sessionID := chi.URLParam(r, "session_id")
 	targetURL := transcodeNodeURL + path
+	isSegmentRoute := strings.Contains(path, "/segment/")
+	_, segmentParseErr := playback.ParseSegmentNumber(filepath.Base(path))
+	isMediaSegment := segmentParseErr == nil
 	// Capture the signed stream token ("st") before stripping it from the URL.
 	// We forward it out-of-band as a header so the node can reconstruct after a
 	// self-restart, while keeping it out of the forwarded/logged URL.
@@ -1550,6 +1560,14 @@ func (h *PlaybackHandler) proxyToTranscodeNode(w http.ResponseWriter, r *http.Re
 		return
 	}
 	req.Header.Set("Authorization", "Bearer "+h.JWTSecret)
+	if isSegmentRoute {
+		// The node's immediate transport peer is this API process, so receiving a
+		// complete response there does not prove that the browser received it.
+		// Suppress node-local accounting and acknowledge only after the downstream
+		// writer completes. Forward range validators so both hops serve the same
+		// representation.
+		transcodeproxy.PrepareRequest(req, r)
+	}
 	// Best-effort forward of the stream token as a header so the node's
 	// reconstruct path (X-Silo-Stream-Token) can rebuild after a self-restart.
 	// Verify at the API boundary and confirm it belongs to this session; an
@@ -1573,7 +1591,7 @@ func (h *PlaybackHandler) proxyToTranscodeNode(w http.ResponseWriter, r *http.Re
 		http.Error(w, "transcode node unavailable", http.StatusBadGateway)
 		return
 	}
-	defer resp.Body.Close()
+	defer func() { _ = resp.Body.Close() }()
 
 	// The node strips "st" from the request query (kept out of node URLs/logs),
 	// so the segment/init URIs in the manifest it builds carry no token. Without
@@ -1603,16 +1621,22 @@ func (h *PlaybackHandler) proxyToTranscodeNode(w http.ResponseWriter, r *http.Re
 		return
 	}
 
-	for k, vv := range resp.Header {
-		for _, v := range vv {
-			w.Header().Add(k, v)
-		}
-	}
+	generation := resp.Header.Get(transcodeproxy.GenerationHeader)
+	transcodeproxy.CopyResponseHeaders(w.Header(), resp.Header)
 	// Proxied transcode output can stream past the server's absolute
 	// WriteTimeout; roll the write deadline with progress instead.
 	sw := httpstream.NewRollingDeadlineWriter(w)
 	sw.WriteHeader(resp.StatusCode)
-	io.Copy(sw, resp.Body)
+	if _, copyErr := io.Copy(sw, resp.Body); copyErr != nil {
+		return
+	}
+	fullSize := transcodeproxy.FullRepresentationSize(resp)
+	if isMediaSegment && generation != "" && r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(fullSize) {
+		if ackErr := transcodeproxy.Acknowledge(r.Context(), http.DefaultClient, transcodeNodeURL+path, h.JWTSecret, generation); ackErr != nil {
+			slog.WarnContext(r.Context(), "acknowledge transcode segment completion", "component", "api", "error", ackErr, "playback_session_id", sessionID)
+		}
+	}
 }
 
 // maybeStartThrottler reads throttle settings and starts the throttler if enabled.

internal/api/handlers/playback_config_test.go

@@ -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/api/handlers/playback_proxy_test.go

@@ -0,0 +1,127 @@
+package handlers
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
+)
+
+func TestProxyToTranscodeNodeAcknowledgesOnlyFullDownstreamResponse(t *testing.T) {
+	const (
+		body       = "complete segment"
+		generation = "17"
+	)
+	tests := []struct {
+		name        string
+		rangeHeader string
+		wantStatus  int
+		wantAck     int32
+	}{
+		{name: "ordinary get", wantStatus: http.StatusOK, wantAck: 1},
+		{name: "whole-file range", rangeHeader: "bytes=0-", wantStatus: http.StatusPartialContent, wantAck: 1},
+		{name: "partial range", rangeHeader: "bytes=1-", wantStatus: http.StatusPartialContent},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var acknowledgements atomic.Int32
+			seenRange := make(chan string, 1)
+			node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				switch r.Method {
+				case http.MethodGet:
+					if r.Header.Get(transcodeproxy.RequestHeader) != "1" {
+						t.Error("segment request omitted proxy marker")
+					}
+					seenRange <- r.Header.Get("Range")
+					w.Header().Set(transcodeproxy.GenerationHeader, generation)
+					http.ServeContent(w, r, "seg_00007.ts", time.Time{}, strings.NewReader(body))
+				case http.MethodPost:
+					if r.Header.Get(transcodeproxy.GenerationHeader) != generation {
+						t.Errorf("ack generation = %q, want %q", r.Header.Get(transcodeproxy.GenerationHeader), generation)
+					}
+					acknowledgements.Add(1)
+					w.WriteHeader(http.StatusNoContent)
+				default:
+					http.Error(w, "unexpected method", http.StatusMethodNotAllowed)
+				}
+			}))
+			defer node.Close()
+
+			handler := &PlaybackHandler{JWTSecret: "proxy-test-secret"}
+			req := withPlaybackRouteParam(
+				httptest.NewRequest(http.MethodGet, "/playback/transcode/public/segment/seg_00007.ts", nil),
+				"session_id",
+				"public",
+			)
+			req.Header.Set("Range", tt.rangeHeader)
+			rr := httptest.NewRecorder()
+			handler.proxyToTranscodeNode(rr, req, node.URL, "/transcode/remote/segment/seg_00007.ts")
+
+			if rr.Code != tt.wantStatus {
+				t.Fatalf("status = %d, want %d; body = %q", rr.Code, tt.wantStatus, rr.Body.String())
+			}
+			if got := <-seenRange; got != tt.rangeHeader {
+				t.Fatalf("forwarded Range = %q, want %q", got, tt.rangeHeader)
+			}
+			if got := rr.Header().Get(transcodeproxy.GenerationHeader); got != "" {
+				t.Fatalf("internal generation leaked downstream: %q", got)
+			}
+			if got := acknowledgements.Load(); got != tt.wantAck {
+				t.Fatalf("acknowledgements = %d, want %d", got, tt.wantAck)
+			}
+		})
+	}
+}
+
+func TestProxyToTranscodeNodeDoesNotAcknowledgeFailedDownstreamWrite(t *testing.T) {
+	const body = "complete segment"
+	var acknowledgements atomic.Int32
+	node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodPost {
+			acknowledgements.Add(1)
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+		w.Header().Set(transcodeproxy.GenerationHeader, "21")
+		http.ServeContent(w, r, "seg_00007.ts", time.Time{}, strings.NewReader(body))
+	}))
+	defer node.Close()
+
+	handler := &PlaybackHandler{JWTSecret: "proxy-test-secret"}
+	req := withPlaybackRouteParam(
+		httptest.NewRequest(http.MethodGet, "/playback/transcode/public/segment/seg_00007.ts", nil),
+		"session_id",
+		"public",
+	)
+	w := &failingProxyResponseWriter{header: make(http.Header), remaining: 5}
+	handler.proxyToTranscodeNode(w, req, node.URL, "/transcode/remote/segment/seg_00007.ts")
+
+	if got := acknowledgements.Load(); got != 0 {
+		t.Fatalf("failed downstream transfer produced %d acknowledgement(s)", got)
+	}
+}
+
+type failingProxyResponseWriter struct {
+	header    http.Header
+	status    int
+	remaining int
+}
+
+func (w *failingProxyResponseWriter) Header() http.Header { return w.header }
+
+func (w *failingProxyResponseWriter) WriteHeader(status int) { w.status = status }
+
+func (w *failingProxyResponseWriter) Write(p []byte) (int, error) {
+	if w.remaining <= 0 {
+		return 0, io.ErrClosedPipe
+	}
+	n := min(len(p), w.remaining)
+	w.remaining -= n
+	return n, io.ErrClosedPipe
+}

internal/api/handlers/playback_v3.go

@@ -890,7 +890,7 @@ func (h *PlaybackHandler) prepareLocalTransportV3(r *http.Request, session *play
 	sourceProfile, sourceBitDepth := sourceVideoTranscodeFactsV3(file, result)
 	seekSeconds, startSegment := configureHLSTimelineV3(result.Plan, videoCodec, 2, sourceMetadata.DurationSeconds)
 	unlock := h.tm.LockSessionLifecycle(session.ID)
-	opts := playback.TranscodeOpts{InputPath: file.FilePath, OutputDir: outputDir, OutputSubdir: outputSubdir, SessionID: session.ID, SourceVideoCodec: sourceMetadata.VideoCodec, SourceVideoProfile: sourceProfile, SourceVideoBitDepth: sourceBitDepth, SoftwareVideoDecode: sourceMetadata.SoftwareVideoDecode, VideoBitstreamFilter: videoBitstreamFilterForPlanV3(result.Plan), SeekSeconds: seekSeconds, StartSegmentNumber: startSegment, TargetResolution: result.TargetResolution, TargetCodecVideo: videoCodec, TargetCodecAudio: result.TargetAudioCodec, TargetAudioChannels: result.TargetAudioChannels, TargetAudioBitrateKbps: result.TargetAudioBitrateKbps, TargetBitrateKbps: result.TargetBitrateKbps, SegmentDuration: 2, FFmpegPath: cfg.FFmpegPath, HWAccel: cfg.HWAccel, HWDevice: cfg.HWDevice, AudioTrackIndex: plannedAudioTrackIndexV3(result, session.AudioTrackIndex), SubtitleTrackIndex: result.SubtitleTransportTrackIndex, SubtitleBurnIn: result.SubtitleBurnIn, SubtitleCodec: result.SubtitleCodec, TotalDuration: sourceMetadata.DurationSeconds, FastStart: true, NodeType: playbackNodeIntegratedV3, ExecutionMode: playbackNodeIntegratedV3, FFmpegLogSink: h.FFmpegLogSink}
+	opts := playback.TranscodeOpts{InputPath: file.FilePath, OutputDir: outputDir, OutputSubdir: outputSubdir, SessionID: session.ID, SourceVideoCodec: sourceMetadata.VideoCodec, SourceVideoProfile: sourceProfile, SourceVideoBitDepth: sourceBitDepth, SoftwareVideoDecode: sourceMetadata.SoftwareVideoDecode, VideoBitstreamFilter: videoBitstreamFilterForPlanV3(result.Plan), SeekSeconds: seekSeconds, StartSegmentNumber: startSegment, TargetResolution: result.TargetResolution, TargetCodecVideo: videoCodec, TargetCodecAudio: result.TargetAudioCodec, TargetAudioChannels: result.TargetAudioChannels, TargetAudioBitrateKbps: result.TargetAudioBitrateKbps, TargetBitrateKbps: result.TargetBitrateKbps, SegmentDuration: 2, SegmentRetentionSeconds: cfg.SegmentRetentionSeconds, FFmpegPath: cfg.FFmpegPath, HWAccel: cfg.HWAccel, HWDevice: cfg.HWDevice, AudioTrackIndex: plannedAudioTrackIndexV3(result, session.AudioTrackIndex), SubtitleTrackIndex: result.SubtitleTransportTrackIndex, SubtitleBurnIn: result.SubtitleBurnIn, SubtitleCodec: result.SubtitleCodec, TotalDuration: sourceMetadata.DurationSeconds, FastStart: true, NodeType: playbackNodeIntegratedV3, ExecutionMode: playbackNodeIntegratedV3, FFmpegLogSink: h.FFmpegLogSink}
 	ts, err := h.startLocalPlaybackTransport(r.Context(), opts)
 	if err != nil {
 		unlock()

internal/config/admin_settings.go

@@ -13,8 +13,11 @@ import (
 	"github.com/robfig/cron/v3"
 )
 
-const cloudflareURLMode = "cloudflare_token"
-const chapterThumbnailSoftwareToneMapKey = "playback.chapter_thumbnail_software_tone_map_enabled"
+const (
+	cloudflareURLMode                  = "cloudflare_token"
+	playbackSegmentRetentionSettingKey = "playback.segment_retention_seconds"
+	chapterThumbnailSoftwareToneMapKey = "playback.chapter_thumbnail_software_tone_map_enabled"
+)
 
 // adminSettingDefaults is the effective value shown by the Admin UI when no
 // row exists in server_settings. Keep these values aligned with the runtime
@@ -51,6 +54,7 @@ var adminSettingDefaults = map[string]string{
 
 	"playback.ffmpeg_path":                     "/usr/lib/jellyfin-ffmpeg/ffmpeg",
 	"playback.transcode_dir":                   DefaultTranscodeDir,
+	playbackSegmentRetentionSettingKey:         "600",
 	"playback.hw_accel":                        "auto",
 	"playback.transcode_enabled":               "true",
 	"playback.local_transcode_fallback":        "true",
@@ -301,6 +305,16 @@ func NormalizeAdminSetting(key, raw string) (string, error) {
 		return normalizeAdminInt(key, value, 1, 99)
 	case "transcode_throttle_seconds":
 		return normalizeAdminInt(key, value, 60, 86400)
+	case playbackSegmentRetentionSettingKey:
+		normalized, err := normalizeAdminInt(key, value, 0, 86400)
+		if err != nil {
+			return "", err
+		}
+		seconds, _ := strconv.Atoi(normalized)
+		if seconds != 0 && seconds < 120 {
+			return "", fmt.Errorf("%s must be 0 or between 120 and 86400", key)
+		}
+		return normalized, nil
 	case "ai.max_concurrent_jobs", "subtitle_ai.max_concurrent_jobs":
 		return normalizeAdminInt(key, value, 1, 1024)
 	case "subtitle_ai.batch_size":

internal/config/admin_settings_test.go

@@ -225,6 +225,7 @@ func TestNormalizeAdminSettingRejectsInvalidValues(t *testing.T) {
 		{key: "theme.catalog_url", value: "http://raw.githubusercontent.com/Silo-Server/silo-themes/main/catalog.json"},
 		{key: "theme.catalog_url", value: "https://example.com/catalog.json"},
 		{key: "redis.url", value: "not-a-url"},
+		{key: "playback.segment_retention_seconds", value: "119"},
 	}
 	for _, tc := range tests {
 		t.Run(tc.key, func(t *testing.T) {
@@ -235,6 +236,20 @@ func TestNormalizeAdminSettingRejectsInvalidValues(t *testing.T) {
 	}
 }
 
+func TestNormalizeAdminSettingAcceptsSegmentRetentionBounds(t *testing.T) {
+	for _, value := range []string{"0", "120", "86400"} {
+		t.Run(value, func(t *testing.T) {
+			got, err := NormalizeAdminSetting("playback.segment_retention_seconds", value)
+			if err != nil {
+				t.Fatalf("NormalizeAdminSetting: %v", err)
+			}
+			if got != value {
+				t.Fatalf("normalized retention = %q, want %q", got, value)
+			}
+		})
+	}
+}
+
 func TestNormalizeAdminSettingAcceptsApprovedThemeCatalogURL(t *testing.T) {
 	got, err := NormalizeAdminSetting(
 		"theme.catalog_url",

internal/config/config.go

@@ -157,9 +157,10 @@ func (c MatcherConfig) TVSeriesRootQueueEnabled() bool {
 
 // PlaybackConfig holds transcoding and playback settings.
 type PlaybackConfig struct {
-	FFmpegPath   string `yaml:"ffmpeg_path"`
-	TranscodeDir string `yaml:"transcode_dir"`
-	HWAccel      string `yaml:"hw_accel"`
+	FFmpegPath              string `yaml:"ffmpeg_path"`
+	TranscodeDir            string `yaml:"transcode_dir"`
+	SegmentRetentionSeconds int    `yaml:"segment_retention_seconds"`
+	HWAccel                 string `yaml:"hw_accel"`
 	// HWDevice is the GPU render device for hardware transcodes. A single
 	// path pins every GPU workload to that device; a comma-separated list
 	// (e.g. "/dev/dri/renderD128,/dev/dri/renderD129") balances workloads
@@ -479,6 +480,7 @@ func setDefaults() *configRaw {
 		Playback: PlaybackConfig{
 			FFmpegPath:                   "/usr/lib/jellyfin-ffmpeg/ffmpeg",
 			TranscodeDir:                 DefaultTranscodeDir,
+			SegmentRetentionSeconds:      600,
 			HWAccel:                      "auto",
 			ChapterThumbnailWorkers:      1,
 			ChapterThumbnailExecution:    "local",

internal/config/db_loader.go

@@ -317,6 +317,14 @@ func LoadFromDB(m map[string]string) (*Config, error) {
 	// Playback
 	cfg.Playback.FFmpegPath = stringOr(m, "playback.ffmpeg_path", "/usr/lib/jellyfin-ffmpeg/ffmpeg")
 	cfg.Playback.TranscodeDir = stringOr(m, "playback.transcode_dir", DefaultTranscodeDir)
+	segmentRetentionSeconds, err := intOr(m, playbackSegmentRetentionSettingKey, 600)
+	if err != nil {
+		return nil, err
+	}
+	if segmentRetentionSeconds != 0 && (segmentRetentionSeconds < 120 || segmentRetentionSeconds > 86400) {
+		return nil, fmt.Errorf("%s must be 0 or between 120 and 86400", playbackSegmentRetentionSettingKey)
+	}
+	cfg.Playback.SegmentRetentionSeconds = segmentRetentionSeconds
 	cfg.Playback.HWAccel = stringOr(m, "playback.hw_accel", "auto")
 	cfg.Playback.HWDevice = stringOr(m, "playback.hw_device", "")
 	chapterThumbnailWorkers, err := intOr(m, "playback.chapter_thumbnail_workers", 1)

internal/config/db_loader_test.go

@@ -35,6 +35,17 @@ func TestLoadFromDBMetadataPresignExpiryRejectsInvalidDuration(t *testing.T) {
 	}
 }
 
+func TestLoadFromDBRejectsInvalidSegmentRetention(t *testing.T) {
+	for _, value := range []string{"-1", "119", "86401"} {
+		t.Run(value, func(t *testing.T) {
+			_, err := LoadFromDB(map[string]string{playbackSegmentRetentionSettingKey: value})
+			if err == nil || !strings.Contains(err.Error(), playbackSegmentRetentionSettingKey) {
+				t.Fatalf("LoadFromDB() error = %v, want retention bounds error", err)
+			}
+		})
+	}
+}
+
 func TestLoadFromDBJellyfinWebEnabledDefaultsToTrue(t *testing.T) {
 	cfg, err := LoadFromDB(map[string]string{})
 	if err != nil {
@@ -186,6 +197,16 @@ func TestYAMLToSettingsMapJellyfinCompatEnabledDefaultsToLegacyListener(t *testi
 	}
 }
 
+func TestYAMLToSettingsMapPreservesExplicitlyDisabledSegmentRetention(t *testing.T) {
+	m := yamlSettingsMapFromString(t, `
+playback:
+  segment_retention_seconds: 0
+`)
+	if got := m[playbackSegmentRetentionSettingKey]; got != "0" {
+		t.Fatalf("segment retention = %q, want explicit disable", got)
+	}
+}
+
 func yamlSettingsMapFromString(t *testing.T, body string) map[string]string {
 	t.Helper()
 	path := t.TempDir() + "/silo.yaml"

internal/config/yaml_import.go

@@ -209,6 +209,7 @@ func YAMLToSettingsMap(path string) (map[string]string, error) {
 	// Playback
 	setIfNonEmpty(m, "playback.ffmpeg_path", raw.Playback.FFmpegPath)
 	setIfNonEmpty(m, "playback.transcode_dir", raw.Playback.TranscodeDir)
+	m[playbackSegmentRetentionSettingKey] = strconv.Itoa(raw.Playback.SegmentRetentionSeconds)
 	setIfNonEmpty(m, "playback.hw_accel", raw.Playback.HWAccel)
 	if raw.Playback.ChapterThumbnailWorkers != 0 {
 		m["playback.chapter_thumbnail_workers"] = strconv.Itoa(raw.Playback.ChapterThumbnailWorkers)

internal/httpstream/rolling_deadline.go

@@ -12,6 +12,7 @@ package httpstream
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"net/http"
@@ -171,6 +172,26 @@ func (s *RollingDeadlineWriter) BytesWritten() int64 {
 	return s.bytesWritten
 }
 
+// CompletedFullResponse reports whether the complete representation was
+// accepted by the transport without a write error. The request context is not
+// consulted because net/http may cancel it after the response reaches the
+// client but before the handler performs post-response accounting. ServeContent
+// returns 206 for a whole-file Range request such as "bytes=0-", so byte count
+// alone is not enough to distinguish that transfer from a partial range.
+func (s *RollingDeadlineWriter) CompletedFullResponse(fullSize int64) bool {
+	if fullSize <= 0 || s.bytesWritten != fullSize || s.firstWriteErr != nil {
+		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 {

internal/httpstream/rolling_deadline_test.go

@@ -271,6 +271,42 @@ func TestServeContentReadFromOutcomeCompletedAndCounted(t *testing.T) {
 	}
 }
 
+func TestCompletedFullResponse(t *testing.T) {
+	const body = "complete segment"
+	tests := []struct {
+		name             string
+		rangeHeader      string
+		method           string
+		cancelAfterWrite bool
+		want             bool
+	}{
+		{name: "ordinary get", method: http.MethodGet, want: true},
+		{name: "completed get with canceled context", method: http.MethodGet, cancelAfterWrite: true, 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)
+			ctx, cancel := context.WithCancel(context.Background())
+			t.Cleanup(cancel)
+			req := httptest.NewRequest(tt.method, "/segment.ts", nil).WithContext(ctx)
+			req.Header.Set("Range", tt.rangeHeader)
+			http.ServeContent(sw, req, "segment.ts", time.Time{}, strings.NewReader(body))
+			if tt.cancelAfterWrite {
+				cancel()
+			}
+			if got := sw.CompletedFullResponse(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.

internal/jellycompat/handlers_playback.go

@@ -184,6 +184,7 @@ type PlaybackHandler struct {
 	FFmpegPath              string
 	HWAccel                 string
 	TranscodeDir            string
+	SegmentRetentionSeconds func() int
 	// tm is the shared transcode-session lifecycle (live map, reconstruct) — the
 	// same type the native handler uses, so jellycompat gets the reconstruct cap
 	// and node-affinity rule for free. The reconstruction recipe is carried in the
@@ -268,36 +269,40 @@ func NewPlaybackHandler(
 	transcodeDir := filepath.Join(os.TempDir(), "silo-transcode")
 	ffmpegPath := ""
 	hwAccel := ""
+	segmentRetentionSeconds := 600
 	if cfg != nil {
 		if cfg.Playback.TranscodeDir != "" {
 			transcodeDir = cfg.Playback.TranscodeDir
 		}
 		ffmpegPath = cfg.Playback.FFmpegPath
 		hwAccel = cfg.Playback.HWAccel
+		segmentRetentionSeconds = cfg.Playback.SegmentRetentionSeconds
 	}
 
 	h := &PlaybackHandler{
-		cfg:            cfg,
-		content:        content,
-		codec:          codec,
-		deviceProfiles: deviceProfiles,
-		playbackStore:  playbackStore,
-		sessionMgr:     sessionMgr,
-		fileResolver:   fileResolver,
-		storeProvider:  storeProvider,
-		FFmpegPath:     ffmpegPath,
-		HWAccel:        hwAccel,
-		TranscodeDir:   transcodeDir,
-		tm:             playback.NewTranscodeManager(),
+		cfg:                     cfg,
+		content:                 content,
+		codec:                   codec,
+		deviceProfiles:          deviceProfiles,
+		playbackStore:           playbackStore,
+		sessionMgr:              sessionMgr,
+		fileResolver:            fileResolver,
+		storeProvider:           storeProvider,
+		FFmpegPath:              ffmpegPath,
+		HWAccel:                 hwAccel,
+		TranscodeDir:            transcodeDir,
+		SegmentRetentionSeconds: func() int { return segmentRetentionSeconds },
+		tm:                      playback.NewTranscodeManager(),
 	}
 	// Wire the shared transcode manager with closures so it reads the handler's
 	// (late-set) JWTSecret lazily, matching the native handler.
 	h.tm.JWTSecretFn = func() string { return h.JWTSecret }
 	h.tm.Config = func() playback.TranscodeRuntimeConfig {
 		return playback.TranscodeRuntimeConfig{
-			TranscodeDir: h.TranscodeDir,
-			FFmpegPath:   h.FFmpegPath,
-			HWAccel:      h.HWAccel,
+			TranscodeDir:            h.TranscodeDir,
+			FFmpegPath:              h.FFmpegPath,
+			HWAccel:                 h.HWAccel,
+			SegmentRetentionSeconds: h.segmentRetentionSeconds(),
 		}
 	}
 	if reg, ok := sessionMgr.(interface {
@@ -337,6 +342,13 @@ func NewPlaybackHandler(
 	return h
 }
 
+func (h *PlaybackHandler) segmentRetentionSeconds() int {
+	if h.SegmentRetentionSeconds != nil {
+		return h.SegmentRetentionSeconds()
+	}
+	return 600
+}
+
 // CleanupOrphanedTranscodes removes stale per-session transcode dirs, sparing
 // those whose recipe card still exists. Delegates to the shared manager.
 func (h *PlaybackHandler) CleanupOrphanedTranscodes() (int, error) {

internal/jellycompat/handlers_playback_config_test.go

@@ -0,0 +1,22 @@
+package jellycompat
+
+import (
+	"testing"
+
+	"github.com/Silo-Server/silo-server/internal/config"
+)
+
+func TestPlaybackHandlerReadsLiveSegmentRetention(t *testing.T) {
+	cfg := &config.Config{Playback: config.PlaybackConfig{SegmentRetentionSeconds: 300}}
+	handler := NewPlaybackHandler(cfg, nil, nil, nil, nil, nil, nil, nil)
+
+	if got := handler.tm.Config().SegmentRetentionSeconds; got != 300 {
+		t.Fatalf("configured segment retention = %d, want 300", got)
+	}
+	liveRetention := 300
+	handler.SegmentRetentionSeconds = func() int { return liveRetention }
+	liveRetention = 120
+	if got := handler.tm.Config().SegmentRetentionSeconds; got != 120 {
+		t.Fatalf("reloaded segment retention = %d, want 120", got)
+	}
+}

internal/jellycompat/router.go

@@ -103,6 +103,13 @@ func NewRouter(deps Dependencies) chi.Router {
 	}
 	userDataHandler := NewUserDataHandler(deps.ContentService, deps.UserDataService, deps.IDCodec, deps.Config)
 	playbackHandler := NewPlaybackHandler(deps.Config, deps.ContentService, deps.IDCodec, deps.DeviceProfiles, deps.PlaybackStore, deps.SessionMgr, deps.FileResolver, deps.UserStoreProvider)
+	startupSegmentRetention := playbackHandler.SegmentRetentionSeconds
+	playbackHandler.SegmentRetentionSeconds = func() int {
+		if cfg := deps.CurrentConfig(); cfg != nil {
+			return cfg.Playback.SegmentRetentionSeconds
+		}
+		return startupSegmentRetention()
+	}
 	if deps.DB != nil {
 		playbackHandler.profileStaler = recommendations.NewRepo(deps.DB)
 	}

internal/jellycompat/streams.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/go-chi/chi/v5"
 
+	"github.com/Silo-Server/silo-server/internal/httpstream"
 	"github.com/Silo-Server/silo-server/internal/models"
 	"github.com/Silo-Server/silo-server/internal/nodepool"
 	"github.com/Silo-Server/silo-server/internal/playback"
@@ -418,8 +419,8 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 		}
 	}
 
-	segmentFile := name + "." + ext
-	segmentPath, err := transcodeSession.GetSegment(segmentFile)
+	segmentName := name + "." + ext
+	segmentLease, err := transcodeSession.OpenSegment(segmentName)
 	if err != nil && errors.Is(err, playback.ErrSegmentNotFound) {
 		segNum, parseErr := playback.ParseSegmentNumber(name)
 		if parseErr == nil {
@@ -430,7 +431,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 				lastProducedAgeMS = now.Sub(decision.Progress.LastProducedAt).Milliseconds()
 			}
 			slog.InfoContext(r.Context(), "transcode segment missing", "component", "jellycompat",
-				"segment", segmentFile,
+				"segment", segmentName,
 				"requested_segment", segNum,
 				"produced_head", decision.Progress.ProducedHead,
 				"last_requested_segment", decision.Progress.LastRequestedSegment,
@@ -444,7 +445,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 			)
 			if decision.Wait {
 				slog.InfoContext(r.Context(), "transcode segment wait", "component", "jellycompat",
-					"segment", segmentFile,
+					"segment", segmentName,
 					"requested_segment", segNum,
 					"produced_head", decision.Progress.ProducedHead,
 					"last_requested_segment", decision.Progress.LastRequestedSegment,
@@ -456,10 +457,10 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 					"session", playSession.UpstreamSessionID,
 					"playback_session_id", playSession.UpstreamSessionID,
 				)
-				segmentPath, err = transcodeSession.WaitForSegment(segmentFile, 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", segmentFile,
+						"segment", segmentName,
 						"requested_segment", segNum,
 						"produced_head", decision.Progress.ProducedHead,
 						"last_requested_segment", decision.Progress.LastRequestedSegment,
@@ -479,7 +480,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 				if seekErr != nil && !errors.Is(seekErr, playback.ErrManifestNotReady) {
 					slog.ErrorContext(r.Context(), "resolve transcode seek target", "component", "jellycompat",
 						"error", seekErr,
-						"segment", segmentFile,
+						"segment", segmentName,
 						"play_session", playSessionID,
 						"session", playSession.UpstreamSessionID,
 						"playback_session_id", playSession.UpstreamSessionID,
@@ -498,7 +499,7 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 
 				if ok {
 					slog.InfoContext(r.Context(), "transcode seek restart", "component", "jellycompat",
-						"segment", segmentFile,
+						"segment", segmentName,
 						"requested_segment", segNum,
 						"produced_head", decision.Progress.ProducedHead,
 						"last_requested_segment", decision.Progress.LastRequestedSegment,
@@ -519,14 +520,14 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 						seekSeconds,
 						segNum,
 					); restartErr == nil {
-						segmentPath, err = transcodeSession.WaitForSegment(segmentFile, 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.
-			segmentPath, err = transcodeSession.WaitForSegment(segmentFile, 10*time.Second)
+			segmentLease, err = transcodeSession.WaitForOpenSegment(segmentName, 10*time.Second)
 		}
 	}
 	if err != nil {
@@ -535,11 +536,15 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques
 		return
 	}
 
-	if segNum, parseErr := playback.ParseSegmentNumber(name); parseErr == nil {
-		transcodeSession.ReportSegmentDownloaded(segNum)
+	defer func() { _ = segmentLease.Close() }()
+	sw := httpstream.NewRollingDeadlineWriter(w)
+	http.ServeContent(sw, r, segmentLease.Info.Name(), segmentLease.Info.ModTime(), segmentLease.File)
+	if r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(segmentLease.Info.Size()) {
+		if segNum, parseErr := playback.ParseSegmentNumber(name); parseErr == nil {
+			transcodeSession.ReportSegmentDownloadedForGeneration(segNum, segmentLease.Generation)
+		}
 	}
-
-	http.ServeFile(w, r, segmentPath)
 }
 
 // hlsSegmentErrorResponse maps a segment-retrieval error to a Jellyfin-faithful
@@ -1669,6 +1674,7 @@ func (h *PlaybackHandler) ensureTranscodeSession(ctx context.Context, playSessio
 		TotalDuration:       float64(source.Version.Duration),
 		FastStart:           true,
 	}
+	opts.SegmentRetentionSeconds = h.segmentRetentionSeconds()
 	if source.TranscodeAudio {
 		opts.TargetCodecVideo = "copy"
 	}

internal/playback/segment_pruner.go

@@ -0,0 +1,345 @@
+package playback
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+const (
+	segmentPruneHysteresis = 5
+	segmentPruneBatchSize  = 512
+	segmentPruneRetryDelay = 5 * time.Second
+)
+
+type segmentPruneCandidate struct {
+	number int
+	path   string
+}
+
+// scheduleSegmentPruneLocked starts one asynchronous prune pass after the
+// client has advanced far enough to make useful work. The caller must hold
+// s.mu. Segment retention is measured in media time so custom HLS segment
+// durations retain the same back-seek window.
+func (s *TranscodeSession) scheduleSegmentPruneLocked() {
+	retentionSeconds := s.opts.SegmentRetentionSeconds
+	if retentionSeconds <= 0 || s.segmentPruneRunning || s.restarting {
+		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.lastCompletedSegment-s.lastPruneHighWater < segmentPruneHysteresis {
+			return
+		}
+	} else {
+		segmentDuration := s.opts.SegmentDuration
+		if segmentDuration <= 0 {
+			segmentDuration = defaultSegmentDuration
+		}
+		retainedSegments := (retentionSeconds + segmentDuration - 1) / segmentDuration
+		floor := s.lastCompletedSegment - retainedSegments
+		beforeStartReady := s.pruneBeforeStart && floor == s.opts.StartSegmentNumber
+		if floor < s.opts.StartSegmentNumber || (!beforeStartReady && floor == s.opts.StartSegmentNumber) || floor-s.lastPruneFloor < segmentPruneHysteresis {
+			return
+		}
+	}
+
+	s.segmentPruneRunning = true
+	generation := s.segmentGeneration
+	downloadedThrough := s.lastCompletedSegment
+	s.lastPruneHighWater = downloadedThrough
+	go s.pruneDownloadedSegments(generation, downloadedThrough, false)
+}
+
+// pruneDownloadedSegments removes completed media files strictly behind floor.
+// It visits each newly expired segment number at most once in bounded batches,
+// 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, downloadedThrough int, continuation bool) {
+	started := time.Now()
+	s.mu.Lock()
+	if generation != s.segmentGeneration || s.restarting {
+		s.mu.Unlock()
+		return
+	}
+	opts := s.opts
+	outputDir := s.outputDir
+	fromFloor := s.lastPruneFloor
+	pruneBeforeStart := s.pruneBeforeStart
+	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
+	}
+	beforeStartReady := pruneBeforeStart && floor == opts.StartSegmentNumber
+	if !complete || floor < opts.StartSegmentNumber || (!beforeStartReady && floor == opts.StartSegmentNumber) || (!continuation && floor-fromFloor < segmentPruneHysteresis) {
+		s.finishSegmentPruneAttempt(generation)
+		return
+	}
+
+	segmentDuration := opts.SegmentDuration
+	if segmentDuration <= 0 {
+		segmentDuration = defaultSegmentDuration
+	}
+	freshGuard := 2*time.Duration(segmentDuration)*time.Second + 30*time.Second
+	startupEnd := opts.StartSegmentNumber + startupSegmentRequirement(opts)
+	targetFloor := floor
+	fromSegment := max(fromFloor, startupEnd)
+	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. The old
+		// range can be very sparse after a long seek, so enumerate actual files
+		// below the new start rather than walking every intervening number.
+		targetFloor = min(floor, opts.StartSegmentNumber)
+	}
+	if !pruneBeforeStart && fromSegment >= targetFloor {
+		s.finishSegmentPrune(generation, targetFloor, targetFloor, downloadedThrough, 0)
+		return
+	}
+
+	if _, err := os.Stat(outputDir); err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			s.finishSegmentPrune(generation, targetFloor, targetFloor, downloadedThrough, 0)
+			return
+		}
+		slog.Warn("stat transcode directory for pruning", "component", "playback", "error", err, "session", opts.SessionID, "playback_session_id", opts.SessionID)
+		retryFloor := fromSegment
+		if pruneBeforeStart {
+			retryFloor = fromFloor
+		}
+		s.finishSegmentPrune(generation, retryFloor, targetFloor, downloadedThrough, segmentPruneRetryDelay)
+		return
+	}
+
+	var candidates []segmentPruneCandidate
+	moreCandidates := false
+	if pruneBeforeStart {
+		entries, readErr := os.ReadDir(outputDir)
+		if readErr != nil {
+			slog.Warn("read transcode directory for preserved segment pruning", "component", "playback", "error", readErr, "session", opts.SessionID, "playback_session_id", opts.SessionID)
+			s.finishSegmentPrune(generation, fromFloor, targetFloor, downloadedThrough, segmentPruneRetryDelay)
+			return
+		}
+		for _, entry := range entries {
+			segment, parseErr := ParseSegmentNumber(entry.Name())
+			if entry.IsDir() || parseErr != nil || segment >= targetFloor || entry.Name() != segmentFilename(segment, opts) {
+				continue
+			}
+			if len(candidates) == segmentPruneBatchSize {
+				moreCandidates = true
+				break
+			}
+			candidates = append(candidates, segmentPruneCandidate{
+				number: segment,
+				path:   filepath.Join(outputDir, entry.Name()),
+			})
+		}
+	} else {
+		toSegment := min(targetFloor, fromSegment+segmentPruneBatchSize)
+		for segment := fromSegment; segment < toSegment; segment++ {
+			candidates = append(candidates, segmentPruneCandidate{
+				number: segment,
+				path:   filepath.Join(outputDir, segmentFilename(segment, opts)),
+			})
+		}
+		moreCandidates = toSegment < targetFloor
+	}
+	if len(candidates) == 0 && !moreCandidates {
+		s.finishSegmentPrune(generation, targetFloor, targetFloor, downloadedThrough, 0)
+		return
+	}
+
+	processedFloor := targetFloor
+	if !pruneBeforeStart {
+		processedFloor = candidates[len(candidates)-1].number + 1
+	}
+	var retryAfter time.Duration
+	removed := 0
+	var freedBytes int64
+	for _, candidate := range candidates {
+		info, err := os.Stat(candidate.path)
+		if err != nil {
+			if errors.Is(err, os.ErrNotExist) {
+				continue
+			}
+			if pruneBeforeStart {
+				retryAfter = max(retryAfter, segmentPruneRetryDelay)
+			} else {
+				processedFloor, retryAfter = earlierPruneRetry(processedFloor, retryAfter, candidate.number, segmentPruneRetryDelay)
+			}
+			slog.Warn("stat downloaded transcode segment", "component", "playback", "error", err, "segment", candidate.number, "session", opts.SessionID, "playback_session_id", opts.SessionID)
+			continue
+		}
+		freshUntil := info.ModTime().Add(freshGuard)
+		if delay := time.Until(freshUntil); delay > 0 {
+			if pruneBeforeStart {
+				retryAfter = max(retryAfter, delay)
+			} else {
+				processedFloor, retryAfter = earlierPruneRetry(processedFloor, retryAfter, candidate.number, delay)
+			}
+			continue
+		}
+
+		// Serialize the unlink with restart and shutdown generation changes.
+		// Once either advances the generation, this pass cannot delete files
+		// generated for a replacement timeline or session object.
+		s.mu.Lock()
+		if generation != s.segmentGeneration || s.restarting {
+			s.mu.Unlock()
+			return
+		}
+		removeErr := os.Remove(candidate.path)
+		s.mu.Unlock()
+		if removeErr != nil {
+			if !errors.Is(removeErr, os.ErrNotExist) {
+				if pruneBeforeStart {
+					retryAfter = max(retryAfter, segmentPruneRetryDelay)
+				} else {
+					processedFloor, retryAfter = earlierPruneRetry(processedFloor, retryAfter, candidate.number, segmentPruneRetryDelay)
+				}
+				slog.Warn("remove downloaded transcode segment", "component", "playback", "error", removeErr, "segment", candidate.number, "session", opts.SessionID, "playback_session_id", opts.SessionID)
+			}
+			continue
+		}
+		removed++
+		freedBytes += info.Size()
+	}
+	if pruneBeforeStart && (moreCandidates || retryAfter > 0) {
+		processedFloor = fromFloor
+	}
+
+	s.finishSegmentPrune(generation, processedFloor, targetFloor, downloadedThrough, retryAfter)
+	if removed > 0 {
+		slog.Info("pruned downloaded transcode segments",
+			"component", "playback",
+			"count", removed,
+			"freed_bytes", freedBytes,
+			"floor_segment", processedFloor,
+			"duration_ms", time.Since(started).Milliseconds(),
+			"session", opts.SessionID,
+			"playback_session_id", opts.SessionID,
+		)
+	}
+}
+
+// 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
+	}
+
+	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)
+	}
+	return currentFloor, currentDelay
+}
+
+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 {
+		return
+	}
+	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()
+			if generation != s.segmentGeneration || s.restarting {
+				s.mu.Unlock()
+				return
+			}
+			s.mu.Unlock()
+			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

@@ -0,0 +1,388 @@
+package playback
+
+import (
+	"errors"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestReportSegmentDownloadedPrunesOnlyExpiredBackBuffer(t *testing.T) {
+	dir := t.TempDir()
+	old := time.Now().Add(-time.Minute)
+	for segment := 0; segment <= 15; segment++ {
+		writePrunerTestFile(t, filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})), []byte("segment"), old)
+	}
+	writePrunerTestFile(t, filepath.Join(dir, "init.mp4"), []byte("init"), old)
+	writePrunerTestFile(t, filepath.Join(dir, "stream.m3u8"), []byte("manifest"), old)
+	writePrunerTestFile(t, filepath.Join(dir, "seg_00004.ts.tmp"), []byte("temporary"), old)
+	writePrunerTestFile(t, filepath.Join(dir, "notes.txt"), []byte("other"), old)
+	// A newly-created file below the floor may belong to a replacement ffmpeg
+	// process and must survive this pass.
+	writePrunerTestFile(t, filepath.Join(dir, segmentFilename(4, TranscodeOpts{})), []byte("fresh"), time.Now())
+
+	session := &TranscodeSession{
+		opts: TranscodeOpts{
+			SessionID:               "prune-test",
+			SegmentDuration:         2,
+			SegmentRetentionSeconds: 10,
+		},
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: 0,
+	}
+	session.ReportSegmentDownloaded(15)
+	waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(9, TranscodeOpts{})))
+	t.Cleanup(func() {
+		session.mu.Lock()
+		session.segmentGeneration++
+		session.segmentPruneRunning = false
+		session.mu.Unlock()
+	})
+
+	for _, segment := range []int{0, 1, 2, 4, 10, 11, 12, 13, 14, 15} {
+		assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})))
+	}
+	for _, segment := range []int{3, 5, 6, 7, 8, 9} {
+		assertPrunerFileMissing(t, filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})))
+	}
+	for _, name := range []string{"init.mp4", "stream.m3u8", "seg_00004.ts.tmp", "notes.txt"} {
+		assertPrunerFileExists(t, filepath.Join(dir, name))
+	}
+}
+
+func TestFreshSegmentIsRetriedAfterGuardExpires(t *testing.T) {
+	dir := t.TempDir()
+	opts := TranscodeOpts{
+		SessionID:               "retry-test",
+		SegmentDuration:         2,
+		SegmentRetentionSeconds: 10,
+	}
+	guard := 2*time.Duration(opts.SegmentDuration)*time.Second + 30*time.Second
+	path := filepath.Join(dir, segmentFilename(3, opts))
+	writePrunerTestFile(t, path, []byte("segment"), time.Now().Add(-guard+time.Second))
+	session := &TranscodeSession{
+		opts:                 opts,
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: 0,
+	}
+
+	session.ReportSegmentDownloaded(9)
+	waitForPrunerFileMissing(t, path)
+	waitForPrunePass(t, session, 4)
+}
+
+func TestPruningContinuesAcrossBoundedBatches(t *testing.T) {
+	dir := t.TempDir()
+	old := time.Now().Add(-time.Minute)
+	opts := TranscodeOpts{
+		SessionID:               "batch-test",
+		SegmentDuration:         2,
+		SegmentRetentionSeconds: 10,
+	}
+	for segment := 0; segment <= 600; segment++ {
+		writePrunerTestFile(t, filepath.Join(dir, segmentFilename(segment, opts)), []byte("segment"), old)
+	}
+	session := &TranscodeSession{
+		opts:                 opts,
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: 0,
+	}
+
+	session.ReportSegmentDownloaded(600)
+	waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(594, opts)))
+	waitForPrunePass(t, session, 595)
+
+	for _, segment := range []int{0, 1, 2, 595, 600} {
+		assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(segment, opts)))
+	}
+}
+
+func TestForwardRestartPrunesOnlyFilesPresentAcrossSparseGap(t *testing.T) {
+	dir := t.TempDir()
+	old := time.Now().Add(-time.Minute)
+	opts := TranscodeOpts{
+		SessionID:               "sparse-restart-test",
+		SegmentDuration:         2,
+		SegmentRetentionSeconds: 10,
+		StartSegmentNumber:      100_000,
+	}
+	for _, segment := range []int{3, 70, 99_999, 100_000} {
+		writePrunerTestFile(t, filepath.Join(dir, segmentFilename(segment, opts)), []byte("segment"), old)
+	}
+	writePrunerTestFile(t, filepath.Join(dir, "seg_00004.ts.tmp"), []byte("temporary"), old)
+	session := &TranscodeSession{
+		opts:                 opts,
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: opts.StartSegmentNumber,
+		lastCompletedSegment: opts.StartSegmentNumber - 1,
+		pruneBeforeStart:     true,
+	}
+
+	session.ReportSegmentDownloaded(opts.StartSegmentNumber + 5)
+	waitForPrunerFileMissing(t, filepath.Join(dir, segmentFilename(99_999, opts)))
+	waitForPrunePass(t, session, opts.StartSegmentNumber)
+
+	for _, segment := range []int{3, 70, 99_999} {
+		assertPrunerFileMissing(t, filepath.Join(dir, segmentFilename(segment, opts)))
+	}
+	assertPrunerFileExists(t, filepath.Join(dir, segmentFilename(100_000, opts)))
+	assertPrunerFileExists(t, filepath.Join(dir, "seg_00004.ts.tmp"))
+}
+
+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{}))
+	writePrunerTestFile(t, path, []byte("segment"), time.Now().Add(-time.Minute))
+	session := &TranscodeSession{
+		opts:                 TranscodeOpts{SegmentDuration: 2},
+		outputDir:            dir,
+		lastPruneFloor:       -1,
+		lastRequestedSegment: 0,
+	}
+
+	session.ReportSegmentDownloaded(500)
+	time.Sleep(20 * time.Millisecond)
+	assertPrunerFileExists(t, path)
+}
+
+func TestOldGenerationDownloadDoesNotAdvanceRestartedSession(t *testing.T) {
+	session := &TranscodeSession{
+		opts:                 TranscodeOpts{SegmentRetentionSeconds: 600},
+		lastRequestedSegment: 100,
+	}
+	oldGeneration := session.SegmentGeneration()
+
+	session.mu.Lock()
+	session.segmentGeneration++
+	session.lastRequestedSegment = 20
+	session.mu.Unlock()
+
+	session.ReportSegmentDownloadedForGeneration(101, oldGeneration)
+	if got := session.LastRequestedSegment(); got != 20 {
+		t.Fatalf("last requested segment = %d, want restarted position 20", got)
+	}
+}
+
+func TestCloseInvalidatesInFlightSegmentGeneration(t *testing.T) {
+	session := &TranscodeSession{
+		opts:                 TranscodeOpts{SegmentRetentionSeconds: 600},
+		outputDir:            t.TempDir(),
+		lastRequestedSegment: 20,
+	}
+	downloadGeneration := session.SegmentGeneration()
+
+	if err := session.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	session.ReportSegmentDownloadedForGeneration(100, downloadGeneration)
+	if got := session.LastRequestedSegment(); got != 20 {
+		t.Fatalf("last requested segment = %d, want closed-session position 20", got)
+	}
+}
+
+func TestOpenSegmentDescriptorSurvivesUnlink(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "seg_00001.ts")
+	want := []byte("complete segment")
+	writePrunerTestFile(t, path, want, time.Now())
+	session := &TranscodeSession{outputDir: dir}
+
+	segment, err := session.OpenSegment("seg_00001.ts")
+	if err != nil {
+		t.Fatalf("OpenSegment: %v", err)
+	}
+	defer func() { _ = segment.Close() }()
+	if err := os.Remove(path); err != nil {
+		t.Fatalf("remove opened segment: %v", err)
+	}
+	got, err := io.ReadAll(segment.File)
+	if err != nil {
+		t.Fatalf("read opened segment: %v", err)
+	}
+	if string(got) != string(want) {
+		t.Fatalf("opened segment = %q, want %q", got, want)
+	}
+}
+
+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 TestGenerationTokenRejectsPriorSessionIncarnation(t *testing.T) {
+	oldDir := t.TempDir()
+	newDir := t.TempDir()
+	const name = "seg_00007.ts"
+	writePrunerTestFile(t, filepath.Join(oldDir, name), []byte("old segment"), time.Now())
+	writePrunerTestFile(t, filepath.Join(newDir, name), []byte("new segment"), time.Now())
+
+	oldSession := &TranscodeSession{outputDir: oldDir}
+	oldLease, err := oldSession.OpenSegment(name)
+	if err != nil {
+		t.Fatalf("open old segment: %v", err)
+	}
+	defer func() { _ = oldLease.Close() }()
+
+	newSession := &TranscodeSession{outputDir: newDir}
+	newLease, err := newSession.OpenSegment(name)
+	if err != nil {
+		t.Fatalf("open new segment: %v", err)
+	}
+	defer func() { _ = newLease.Close() }()
+	if oldLease.Generation != newLease.Generation {
+		t.Fatalf("numeric generations differ: old=%d new=%d; test requires restart collision", oldLease.Generation, newLease.Generation)
+	}
+	if oldLease.GenerationToken == newLease.GenerationToken {
+		t.Fatal("separate session objects reused an opaque generation token")
+	}
+
+	newSession.ReportSegmentDownloadedForGenerationToken(7, oldLease.GenerationToken)
+	if got := newSession.LastRequestedSegment(); got != 0 {
+		t.Fatalf("prior-session token advanced replacement to %d", got)
+	}
+	newSession.ReportSegmentDownloadedForGenerationToken(7, newLease.GenerationToken)
+	if got := newSession.LastRequestedSegment(); got != 7 {
+		t.Fatalf("current token advanced replacement to %d, want 7", got)
+	}
+}
+
+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 {
+		t.Fatalf("write %s: %v", filepath.Base(path), err)
+	}
+	if err := os.Chtimes(path, modTime, modTime); err != nil {
+		t.Fatalf("set times for %s: %v", filepath.Base(path), err)
+	}
+}
+
+func waitForPrunePass(t *testing.T, session *TranscodeSession, wantFloor int) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		session.mu.Lock()
+		running := session.segmentPruneRunning
+		floor := session.lastPruneFloor
+		session.mu.Unlock()
+		if !running && floor >= wantFloor {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("segment prune pass did not finish")
+}
+
+func waitForPrunerFileMissing(t *testing.T, path string) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, err := os.Stat(path); os.IsNotExist(err) {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("expected %s to be removed", filepath.Base(path))
+}
+
+func assertPrunerFileExists(t *testing.T, path string) {
+	t.Helper()
+	if _, err := os.Stat(path); err != nil {
+		t.Fatalf("expected %s to exist: %v", filepath.Base(path), err)
+	}
+}
+
+func assertPrunerFileMissing(t *testing.T, path string) {
+	t.Helper()
+	if _, err := os.Stat(path); !os.IsNotExist(err) {
+		t.Fatalf("expected %s to be removed, stat error = %v", filepath.Base(path), err)
+	}
+}

internal/playback/transcode.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/Silo-Server/silo-server/internal/models"
+	"github.com/google/uuid"
 )
 
 func init() {
@@ -52,15 +53,16 @@ type TranscodeOpts struct {
 	StreamOriginSeconds float64
 	// CopySeekAnchorResolved distinguishes a valid zero-second origin from
 	// older/shared recipes that never resolved a copy seek anchor.
-	CopySeekAnchorResolved bool
-	TargetResolution       string // e.g., 1080p, 720p
-	TargetCodecVideo       string // e.g., h264 (or hevc if allowed)
-	TargetCodecAudio       string // e.g., aac
-	SegmentDuration        int    // seconds, default 6
-	StartSegmentNumber     int    // -hls_segment_start_number, default 0
-	FFmpegPath             string // optional explicit ffmpeg binary path
-	HWAccel                string // auto, qsv, vaapi, nvenc, none
-	HWDevice               string // e.g., /dev/dri/renderD128 (default if empty)
+	CopySeekAnchorResolved  bool
+	TargetResolution        string // e.g., 1080p, 720p
+	TargetCodecVideo        string // e.g., h264 (or hevc if allowed)
+	TargetCodecAudio        string // e.g., aac
+	SegmentDuration         int    // seconds, default 6
+	SegmentRetentionSeconds int    // downloaded media retained behind the client; 0 disables pruning
+	StartSegmentNumber      int    // -hls_segment_start_number, default 0
+	FFmpegPath              string // optional explicit ffmpeg binary path
+	HWAccel                 string // auto, qsv, vaapi, nvenc, none
+	HWDevice                string // e.g., /dev/dri/renderD128 (default if empty)
 	// AvoidHWDevice asks the initial multi-device allocator to prefer any other
 	// present render device. It is a process-local startup hint used after an
 	// early GPU failure; the selected concrete device remains fully reserved and
@@ -119,6 +121,13 @@ type TranscodeSession struct {
 	done                 chan struct{} // closed when the monitor goroutine finishes
 	stdinPipe            io.WriteCloser
 	lastRequestedSegment int
+	lastCompletedSegment int
+	lastPruneFloor       int
+	lastPruneHighWater   int
+	segmentPruneRunning  bool
+	pruneBeforeStart     bool
+	segmentGeneration    uint64
+	segmentIncarnation   string
 	throttler            *TranscodeThrottler
 	stderrLinesLogged    int
 	stderrBytesLogged    int
@@ -246,6 +255,10 @@ func StartTranscode(ctx context.Context, opts TranscodeOpts) (*TranscodeSession,
 		done:                     make(chan struct{}),
 		stderr:                   newBoundedTailBuffer(stderrTailMaxBytes),
 		lastRequestedSegment:     opts.StartSegmentNumber,
+		lastCompletedSegment:     opts.StartSegmentNumber - 1,
+		lastPruneFloor:           opts.StartSegmentNumber - 1,
+		lastPruneHighWater:       opts.StartSegmentNumber - 1,
+		segmentIncarnation:       uuid.NewString(),
 		reserveHWDeviceOnRestart: reserveHWDeviceOnRestart,
 	}
 
@@ -1878,6 +1891,59 @@ func (s *TranscodeSession) GetSegment(name string) (string, error) {
 	return segPath, nil
 }
 
+// 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.
+type SegmentLease struct {
+	File            *os.File
+	Info            os.FileInfo
+	Generation      uint64
+	GenerationToken string
+}
+
+// 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, ErrSegmentNotFound
+		}
+		return nil, fmt.Errorf("open segment: %w", err)
+	}
+	info, err := segment.Stat()
+	if err != nil {
+		_ = segment.Close()
+		return nil, fmt.Errorf("stat segment: %w", err)
+	}
+	if info.Size() <= 0 {
+		_ = segment.Close()
+		return nil, ErrSegmentNotFound
+	}
+	if s.segmentIncarnation == "" {
+		s.segmentIncarnation = uuid.NewString()
+	}
+	return &SegmentLease{
+		File:            segment,
+		Info:            info,
+		Generation:      s.segmentGeneration,
+		GenerationToken: s.segmentGenerationTokenLocked(),
+	}, nil
+}
+
 // Close terminates the ffmpeg process and removes the temporary output directory.
 func (s *TranscodeSession) Close() error {
 	return s.shutdown(true)
@@ -1912,6 +1978,8 @@ func (s *TranscodeSession) shutdown(removeOutput bool) error {
 	defer s.mu.Unlock()
 
 	s.running = false
+	s.segmentGeneration++
+	s.segmentPruneRunning = false
 
 	// Clean up temporary directory.
 	if removeOutput && s.outputDir != "" {
@@ -2027,6 +2095,22 @@ func (s *TranscodeSession) restart(
 		return nil
 	}
 	s.restarting = true
+	s.segmentGeneration++
+	s.segmentPruneRunning = false
+	previousRequestedSegment := s.lastRequestedSegment
+	forwardRestart := startSegment > previousRequestedSegment
+	s.lastRequestedSegment = startSegment
+	s.lastCompletedSegment = 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 - 1
 	cancelCurrent := s.cancel
 	done := s.done
 	s.mu.Unlock()
@@ -2123,6 +2207,7 @@ func (s *TranscodeSession) restart(
 	s.restarting = false
 	s.stdinPipe = stdinPipe
 	s.lastRequestedSegment = startSegment
+	s.lastCompletedSegment = startSegment - 1
 	s.done = make(chan struct{})
 	hook := s.restartHook
 	s.mu.Unlock()
@@ -2186,6 +2271,49 @@ 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) (*SegmentLease, error) {
+	deadline := time.After(timeout)
+	for {
+		segment, err := s.OpenSegment(name)
+		if err == nil {
+			return segment, nil
+		}
+		if !errors.Is(err, ErrSegmentNotFound) {
+			return nil, err
+		}
+
+		s.mu.Lock()
+		running := s.running
+		restarting := s.restarting
+		waitErr := s.waitErr
+		s.mu.Unlock()
+
+		if restarting {
+			select {
+			case <-deadline:
+				return nil, ErrSegmentNotFound
+			case <-time.After(100 * time.Millisecond):
+				continue
+			}
+		}
+		if !running && waitErr != nil {
+			return nil, fmt.Errorf("%w: %w", ErrTranscodeFailed, waitErr)
+		}
+		if !running {
+			return nil, ErrSegmentNotFound
+		}
+
+		select {
+		case <-deadline:
+			return nil, ErrSegmentNotFound
+		case <-time.After(100 * time.Millisecond):
+		}
+	}
+}
+
 // RewriteManifestPaths prefixes relative segment references in an HLS manifest
 // with segPrefix (e.g. "segment/") and optionally appends rawQuery as a query
 // string. This ensures the HLS player's segment requests match server routes
@@ -2433,9 +2561,54 @@ func (s *TranscodeSession) RestartSeekTarget(segNum int) (float64, bool, error)
 func (s *TranscodeSession) ReportSegmentDownloaded(segNum int) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	s.reportSegmentDownloadedLocked(segNum)
+}
+
+// SegmentGeneration identifies the current ffmpeg timeline. Callers that hold
+// an open segment across a restart use it to prevent the old response from
+// advancing the replacement process's download position.
+func (s *TranscodeSession) SegmentGeneration() uint64 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.segmentGeneration
+}
+
+// ReportSegmentDownloadedForGeneration records completion only if the served
+// file belongs to the current ffmpeg timeline.
+func (s *TranscodeSession) ReportSegmentDownloadedForGeneration(segNum int, generation uint64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if generation != s.segmentGeneration {
+		return
+	}
+	s.reportSegmentDownloadedLocked(segNum)
+}
+
+// ReportSegmentDownloadedForGenerationToken records a completion relayed over
+// HTTP only when it belongs to this exact session object and FFmpeg timeline.
+// The opaque incarnation prevents a delayed proxy acknowledgement from matching
+// a reconstructed session whose numeric generation restarted at zero.
+func (s *TranscodeSession) ReportSegmentDownloadedForGenerationToken(segNum int, generationToken string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if generationToken == "" || generationToken != s.segmentGenerationTokenLocked() {
+		return
+	}
+	s.reportSegmentDownloadedLocked(segNum)
+}
+
+func (s *TranscodeSession) segmentGenerationTokenLocked() string {
+	return s.segmentIncarnation + ":" + strconv.FormatUint(s.segmentGeneration, 10)
+}
+
+func (s *TranscodeSession) reportSegmentDownloadedLocked(segNum int) {
 	if segNum > s.lastRequestedSegment {
 		s.lastRequestedSegment = segNum
 	}
+	if segNum > s.lastCompletedSegment {
+		s.lastCompletedSegment = segNum
+	}
+	s.scheduleSegmentPruneLocked()
 }
 
 // LastRequestedSegment returns the highest segment number downloaded by the client.

internal/playback/transcode_manager.go

@@ -20,10 +20,11 @@ import (
 // internal/playback does not import internal/config (avoiding an import cycle);
 // each embedding handler adapts its own config snapshot into this shape.
 type TranscodeRuntimeConfig struct {
-	TranscodeDir string
-	FFmpegPath   string
-	HWAccel      string
-	HWDevice     string
+	TranscodeDir            string
+	FFmpegPath              string
+	HWAccel                 string
+	HWDevice                string
+	SegmentRetentionSeconds int
 }
 
 // sessionReconstructor is the SessionManager capability used to re-register a
@@ -579,6 +580,7 @@ func (m *TranscodeManager) doReconstructTranscode(ctx context.Context, sessionID
 	// operator config change applies to reconstructed sessions too.
 	opts.HWAccel = cfg.HWAccel
 	opts.HWDevice = cfg.HWDevice
+	opts.SegmentRetentionSeconds = cfg.SegmentRetentionSeconds
 
 	// Resume near the segment the client is actually requesting. The card records
 	// the original start; if the client has played past it, spawning ffmpeg at the

internal/playback/transcode_restart_guard_test.go

@@ -2,11 +2,148 @@ package playback
 
 import (
 	"context"
+	"os"
 	"os/exec"
+	"path/filepath"
 	"testing"
 	"time"
 )
 
+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()
+	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,
+		lastRequestedSegment: 70,
+		lastCompletedSegment: 70,
+		lastPruneFloor:       -1,
+		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)
+	}
+	// Restart itself has no downstream-completion proof for the replacement
+	// generation, so it must not unlink even old files synchronously.
+	for _, segment := range []int{3, 40, 64, 65, 69, 70, 100} {
+		path := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))
+		if _, err := os.Stat(path); err != nil {
+			t.Errorf("segment %d was removed during restart: %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{})))
+	for _, segment := range []int{3, 40, 64, 65, 69, 70} {
+		path := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))
+		if _, err := os.Stat(path); !os.IsNotExist(err) {
+			t.Errorf("expired pre-restart segment %d survived completed-download pruning: %v", segment, err)
+		}
+	}
+	if _, err := os.Stat(filepath.Join(dir, segmentFilename(100, TranscodeOpts{}))); err != nil {
+		t.Fatalf("replacement startup segment was removed: %v", err)
+	}
+}
+
+func TestForwardRestartWithoutCompletedDownloadPreservesFreshFiles(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} {
+		if err := os.WriteFile(filepath.Join(dir, segmentFilename(segment, TranscodeOpts{})), []byte("segment"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	session := &TranscodeSession{
+		outputDir:            dir,
+		lastRequestedSegment: 70,
+		lastCompletedSegment: -1,
+		lastPruneFloor:       -1,
+		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)
+	}
+	time.Sleep(20 * time.Millisecond)
+	for _, segment := range []int{3, 40, 70} {
+		if _, err := os.Stat(filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))); err != nil {
+			t.Errorf("fresh segment %d was removed without a completed download: %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)
+		}
+	}
+	session := &TranscodeSession{
+		outputDir:            dir,
+		lastRequestedSegment: 70,
+		lastCompletedSegment: 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, 70} {
+		path := filepath.Join(dir, segmentFilename(segment, TranscodeOpts{}))
+		if _, err := os.Stat(path); err != nil {
+			t.Errorf("disabled retention removed segment %d: %v", segment, 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

internal/proxy/segment_completion_test.go

@@ -0,0 +1,133 @@
+package proxy
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Silo-Server/silo-server/internal/config"
+	"github.com/Silo-Server/silo-server/internal/nodeconfig"
+	"github.com/Silo-Server/silo-server/internal/streamtoken"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
+)
+
+func TestTranscodeProxyAcknowledgesOnlyFullDownstreamResponse(t *testing.T) {
+	const (
+		body       = "complete segment"
+		generation = "incarnation:17"
+	)
+	tests := []struct {
+		name        string
+		rangeHeader string
+		wantStatus  int
+		wantAck     int32
+	}{
+		{name: "ordinary get", wantStatus: http.StatusOK, wantAck: 1},
+		{name: "whole-file range", rangeHeader: "bytes=0-", wantStatus: http.StatusPartialContent, wantAck: 1},
+		{name: "partial range", rangeHeader: "bytes=1-", wantStatus: http.StatusPartialContent},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var acknowledgements atomic.Int32
+			seenRange := make(chan string, 1)
+			node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				switch r.Method {
+				case http.MethodGet:
+					if r.Header.Get(transcodeproxy.RequestHeader) != "1" {
+						t.Error("segment request omitted proxy marker")
+					}
+					seenRange <- r.Header.Get("Range")
+					w.Header().Set(transcodeproxy.GenerationHeader, generation)
+					http.ServeContent(w, r, "seg_00007.ts", time.Time{}, strings.NewReader(body))
+				case http.MethodPost:
+					if r.Header.Get(transcodeproxy.GenerationHeader) != generation {
+						t.Errorf("ack generation = %q, want %q", r.Header.Get(transcodeproxy.GenerationHeader), generation)
+					}
+					acknowledgements.Add(1)
+					w.WriteHeader(http.StatusNoContent)
+				default:
+					http.Error(w, "unexpected method", http.StatusMethodNotAllowed)
+				}
+			}))
+			defer node.Close()
+
+			server := newCompletionTestProxy(node.Client())
+			claims := &streamtoken.Claims{SessionID: "public", TranscodeNode: node.URL}
+			req := httptest.NewRequest(http.MethodGet, "/stream/transcode/token/segment/seg_00007.ts", nil)
+			req.Header.Set("Range", tt.rangeHeader)
+			rr := httptest.NewRecorder()
+			server.proxyToTranscodeNode(rr, req, claims, "/transcode/remote/segment/seg_00007.ts")
+
+			if rr.Code != tt.wantStatus {
+				t.Fatalf("status = %d, want %d; body = %q", rr.Code, tt.wantStatus, rr.Body.String())
+			}
+			if got := <-seenRange; got != tt.rangeHeader {
+				t.Fatalf("forwarded Range = %q, want %q", got, tt.rangeHeader)
+			}
+			if got := rr.Header().Get(transcodeproxy.GenerationHeader); got != "" {
+				t.Fatalf("internal generation leaked downstream: %q", got)
+			}
+			if got := acknowledgements.Load(); got != tt.wantAck {
+				t.Fatalf("acknowledgements = %d, want %d", got, tt.wantAck)
+			}
+		})
+	}
+}
+
+func TestTranscodeProxyDoesNotAcknowledgeFailedDownstreamWrite(t *testing.T) {
+	var acknowledgements atomic.Int32
+	node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodPost {
+			acknowledgements.Add(1)
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+		w.Header().Set(transcodeproxy.GenerationHeader, "incarnation:21")
+		http.ServeContent(w, r, "seg_00007.ts", time.Time{}, strings.NewReader("complete segment"))
+	}))
+	defer node.Close()
+
+	server := newCompletionTestProxy(node.Client())
+	claims := &streamtoken.Claims{SessionID: "public", TranscodeNode: node.URL}
+	req := httptest.NewRequest(http.MethodGet, "/stream/transcode/token/segment/seg_00007.ts", nil)
+	w := &failingCompletionResponseWriter{header: make(http.Header), remaining: 5}
+	server.proxyToTranscodeNode(w, req, claims, "/transcode/remote/segment/seg_00007.ts")
+
+	if got := acknowledgements.Load(); got != 0 {
+		t.Fatalf("failed downstream transfer produced %d acknowledgement(s)", got)
+	}
+}
+
+func newCompletionTestProxy(client *http.Client) *Server {
+	watcher := nodeconfig.NewWatcher(nil, nil, nil, nodeconfig.BootstrapOverrides{})
+	cfg := &config.Config{}
+	cfg.Auth.JWTSecret = "proxy-test-secret"
+	watcher.SetConfigForTest(cfg)
+	server := NewServer(watcher, nil)
+	server.httpClient = client
+	return server
+}
+
+type failingCompletionResponseWriter struct {
+	header    http.Header
+	status    int
+	remaining int
+}
+
+func (w *failingCompletionResponseWriter) Header() http.Header { return w.header }
+
+func (w *failingCompletionResponseWriter) WriteHeader(status int) { w.status = status }
+
+func (w *failingCompletionResponseWriter) Write(p []byte) (int, error) {
+	if w.remaining <= 0 {
+		return 0, io.ErrClosedPipe
+	}
+	n := min(len(p), w.remaining)
+	w.remaining -= n
+	return n, io.ErrClosedPipe
+}

internal/proxy/server.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"log/slog"
 	"net/http"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"time"
@@ -13,10 +14,12 @@ import (
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/cors"
 
+	"github.com/Silo-Server/silo-server/internal/httpstream"
 	"github.com/Silo-Server/silo-server/internal/nodeconfig"
 	"github.com/Silo-Server/silo-server/internal/nodesessions"
 	"github.com/Silo-Server/silo-server/internal/playback"
 	"github.com/Silo-Server/silo-server/internal/streamtoken"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
 )
 
 // Server is the HTTP handler for proxy mode.
@@ -343,6 +346,9 @@ func (s *Server) proxyToTranscodeNode(w http.ResponseWriter, r *http.Request, cl
 	}
 
 	targetURL := claims.TranscodeNode + path
+	isSegmentRoute := strings.Contains(path, "/segment/")
+	_, segmentParseErr := playback.ParseSegmentNumber(filepath.Base(path))
+	isMediaSegment := segmentParseErr == nil
 	if rawQuery := r.URL.RawQuery; rawQuery != "" {
 		targetURL = fmt.Sprintf("%s?%s", targetURL, rawQuery)
 	}
@@ -353,6 +359,9 @@ func (s *Server) proxyToTranscodeNode(w http.ResponseWriter, r *http.Request, cl
 		return
 	}
 	req.Header.Set("Authorization", "Bearer "+cfg.Auth.JWTSecret)
+	if isSegmentRoute {
+		transcodeproxy.PrepareRequest(req, r)
+	}
 	// Forward the verified stream token so the transcode node can self-reconstruct
 	// a lost session after its OWN restart: the token carries the full byte-affecting
 	// recipe, so the node can re-spawn ffmpeg seeked to the requested segment instead
@@ -368,16 +377,21 @@ func (s *Server) proxyToTranscodeNode(w http.ResponseWriter, r *http.Request, cl
 		http.Error(w, "transcode node unavailable", http.StatusBadGateway)
 		return
 	}
-	defer resp.Body.Close()
+	defer func() { _ = resp.Body.Close() }()
 
-	// Copy response headers
-	for k, vv := range resp.Header {
-		for _, v := range vv {
-			w.Header().Add(k, v)
+	generation := resp.Header.Get(transcodeproxy.GenerationHeader)
+	transcodeproxy.CopyResponseHeaders(w.Header(), resp.Header)
+	sw := httpstream.NewRollingDeadlineWriter(w)
+	sw.WriteHeader(resp.StatusCode)
+	if _, copyErr := io.Copy(sw, resp.Body); copyErr != nil {
+		return
+	}
+	if isMediaSegment && generation != "" && r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(transcodeproxy.FullRepresentationSize(resp)) {
+		if ackErr := transcodeproxy.Acknowledge(r.Context(), s.httpClient, claims.TranscodeNode+path, cfg.Auth.JWTSecret, generation); ackErr != nil {
+			slog.WarnContext(r.Context(), "acknowledge transcode segment completion", "component", "proxy", "error", ackErr, "playback_session_id", claims.SessionID)
 		}
 	}
-	w.WriteHeader(resp.StatusCode)
-	io.Copy(w, resp.Body)
 }
 
 func (s *Server) handleForceReload(w http.ResponseWriter, r *http.Request) {

internal/transcodenode/server.go

@@ -19,10 +19,12 @@ import (
 	"golang.org/x/sync/singleflight"
 
 	"github.com/Silo-Server/silo-server/internal/chapterthumbs"
+	"github.com/Silo-Server/silo-server/internal/httpstream"
 	"github.com/Silo-Server/silo-server/internal/nodeconfig"
 	"github.com/Silo-Server/silo-server/internal/nodesessions"
 	"github.com/Silo-Server/silo-server/internal/playback"
 	"github.com/Silo-Server/silo-server/internal/streamtoken"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
 )
 
 // TranscodeStartRequest is the JSON body for POST /transcode/start.
@@ -387,6 +389,7 @@ func (s *Server) Handler() http.Handler {
 		r.Delete("/transcode/{session_id}", s.handleStop)
 		r.Get("/transcode/{session_id}/master.m3u8", s.handleManifest)
 		r.Get("/transcode/{session_id}/segment/{name}", s.handleSegment)
+		r.Post("/transcode/{session_id}/segment/{name}/downloaded", s.handleSegmentDownloaded)
 		r.Post("/admin/force-reload", s.handleForceReload)
 		r.Get("/status", s.handleStatus)
 	})
@@ -481,27 +484,28 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
 	outputDir := filepath.Join(cfg.Playback.TranscodeDir, req.SessionID)
 
 	opts := playback.TranscodeOpts{
-		InputPath:              req.InputPath,
-		OutputDir:              outputDir,
-		SessionID:              req.SessionID,
-		SourceVideoCodec:       req.SourceVideoCodec,
-		SourceVideoProfile:     req.SourceVideoProfile,
-		SourceVideoBitDepth:    req.SourceVideoBitDepth,
-		SoftwareVideoDecode:    req.SoftwareVideoDecode,
-		VideoBitstreamFilter:   req.VideoBitstreamFilter,
-		SeekSeconds:            req.SeekSeconds,
-		StreamOriginSeconds:    req.StreamOriginSeconds,
-		CopySeekAnchorResolved: req.CopySeekAnchorResolved,
-		StartSegmentNumber:     req.StartSegmentNumber,
-		TargetResolution:       req.TargetResolution,
-		TargetCodecVideo:       req.TargetCodecVideo,
-		TargetCodecAudio:       req.TargetCodecAudio,
-		TargetAudioChannels:    req.TargetAudioChannels,
-		TargetAudioBitrateKbps: req.TargetAudioBitrateKbps,
-		TargetBitrateKbps:      req.TargetBitrateKbps,
-		SegmentDuration:        req.SegmentDuration,
-		FFmpegPath:             cfg.Playback.FFmpegPath,
-		HWAccel:                req.HWAccel,
+		InputPath:               req.InputPath,
+		OutputDir:               outputDir,
+		SessionID:               req.SessionID,
+		SourceVideoCodec:        req.SourceVideoCodec,
+		SourceVideoProfile:      req.SourceVideoProfile,
+		SourceVideoBitDepth:     req.SourceVideoBitDepth,
+		SoftwareVideoDecode:     req.SoftwareVideoDecode,
+		VideoBitstreamFilter:    req.VideoBitstreamFilter,
+		SeekSeconds:             req.SeekSeconds,
+		StreamOriginSeconds:     req.StreamOriginSeconds,
+		CopySeekAnchorResolved:  req.CopySeekAnchorResolved,
+		StartSegmentNumber:      req.StartSegmentNumber,
+		TargetResolution:        req.TargetResolution,
+		TargetCodecVideo:        req.TargetCodecVideo,
+		TargetCodecAudio:        req.TargetCodecAudio,
+		TargetAudioChannels:     req.TargetAudioChannels,
+		TargetAudioBitrateKbps:  req.TargetAudioBitrateKbps,
+		TargetBitrateKbps:       req.TargetBitrateKbps,
+		SegmentDuration:         req.SegmentDuration,
+		SegmentRetentionSeconds: cfg.Playback.SegmentRetentionSeconds,
+		FFmpegPath:              cfg.Playback.FFmpegPath,
+		HWAccel:                 req.HWAccel,
 		// This node's configured device (or device list — StartTranscode
 		// resolves it to one GPU), matching what reconstruction uses so fresh
 		// and reconstructed sessions balance identically.
@@ -706,6 +710,7 @@ func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSe
 	// rebuild. Run as a transcode node, not integrated (card.TranscodeOpts defaults).
 	opts.HWAccel = cfg.Playback.HWAccel
 	opts.HWDevice = cfg.Playback.HWDevice
+	opts.SegmentRetentionSeconds = cfg.Playback.SegmentRetentionSeconds
 	opts.NodeType = "transcode"
 	opts.ExecutionMode = "transcode_node"
 
@@ -900,7 +905,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {
 		s.touchSession(sessionID)
 	}
 
-	segPath, err := session.GetSegment(name)
+	segmentLease, err := session.OpenSegment(name)
 	if err != nil && err == playback.ErrSegmentNotFound {
 		segNum, parseErr := playback.ParseSegmentNumber(name)
 		if parseErr == nil {
@@ -935,7 +940,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {
 					"session", sessionID,
 					"playback_session_id", sessionID,
 				)
-				segPath, err = session.WaitForSegment(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,
@@ -980,7 +985,7 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {
 						seekSeconds,
 						segNum,
 					); restartErr == nil {
-						segPath, err = session.WaitForSegment(name, 30*time.Second)
+						segmentLease, err = session.WaitForOpenSegment(name, 30*time.Second)
 					}
 				}
 				if !ok && session.IsCopyVideo() {
@@ -990,7 +995,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.
-			segPath, err = session.WaitForSegment(name, 10*time.Second)
+			segmentLease, err = session.WaitForOpenSegment(name, 10*time.Second)
 		}
 	}
 	if err != nil {
@@ -1000,7 +1005,46 @@ 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")
-	http.ServeFile(w, r, segPath)
+	proxied := r.Header.Get(transcodeproxy.RequestHeader) == "1"
+	if proxied {
+		w.Header().Set(transcodeproxy.GenerationHeader, segmentLease.GenerationToken)
+	}
+	defer func() { _ = segmentLease.Close() }()
+	sw := httpstream.NewRollingDeadlineWriter(w)
+	http.ServeContent(sw, r, segmentLease.Info.Name(), segmentLease.Info.ModTime(), segmentLease.File)
+	if !proxied && r.Method == http.MethodGet &&
+		sw.CompletedFullResponse(segmentLease.Info.Size()) {
+		if segmentNumber, parseErr := playback.ParseSegmentNumber(name); parseErr == nil {
+			session.ReportSegmentDownloadedForGeneration(segmentNumber, segmentLease.Generation)
+		}
+	}
+}
+
+// handleSegmentDownloaded records completion only after the central API has
+// delivered the full segment to its downstream playback client. The generation
+// returned with the original segment response prevents a delayed acknowledgement
+// from advancing a replacement FFmpeg timeline.
+func (s *Server) handleSegmentDownloaded(w http.ResponseWriter, r *http.Request) {
+	sessionID := chi.URLParam(r, "session_id")
+	name := chi.URLParam(r, "name")
+	segmentNumber, err := playback.ParseSegmentNumber(name)
+	if err != nil {
+		http.Error(w, "invalid segment", http.StatusBadRequest)
+		return
+	}
+	generationToken := r.Header.Get(transcodeproxy.GenerationHeader)
+	if generationToken == "" {
+		http.Error(w, "invalid segment generation", http.StatusBadRequest)
+		return
+	}
+
+	session, ok := s.acquireSessionTouched(sessionID)
+	if !ok {
+		http.Error(w, "session not found", http.StatusNotFound)
+		return
+	}
+	session.ReportSegmentDownloadedForGenerationToken(segmentNumber, generationToken)
+	w.WriteHeader(http.StatusNoContent)
 }
 
 func (s *Server) handleForceReload(w http.ResponseWriter, r *http.Request) {

internal/transcodenode/server_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"strings"
 	"testing"
@@ -19,6 +20,7 @@ import (
 	"github.com/Silo-Server/silo-server/internal/nodesessions"
 	"github.com/Silo-Server/silo-server/internal/playback"
 	"github.com/Silo-Server/silo-server/internal/streamtoken"
+	"github.com/Silo-Server/silo-server/internal/transcodeproxy"
 )
 
 const testSecret = "node-reconstruct-test-secret"
@@ -34,11 +36,122 @@ func newTestServer(t *testing.T) *Server {
 	cfg.Playback.TranscodeDir = t.TempDir()
 	w.SetConfigForTest(cfg)
 	return &Server{
-		watcher:  w,
-		sessions: make(map[string]*playback.TranscodeSession),
+		watcher:    w,
+		sessions:   make(map[string]*playback.TranscodeSession),
+		lastAccess: make(map[string]time.Time),
 	}
 }
 
+func TestProxiedSegmentCompletionRequiresDownstreamAcknowledgement(t *testing.T) {
+	truePath, err := exec.LookPath("true")
+	if err != nil {
+		t.Skipf("`true` not found in PATH: %v", err)
+	}
+
+	const sessionID = "proxy-completion-session"
+	server := newTestServer(t)
+	outputDir := t.TempDir()
+	session, err := playback.StartTranscode(context.Background(), playback.TranscodeOpts{
+		SessionID:               sessionID,
+		OutputDir:               outputDir,
+		FFmpegPath:              truePath,
+		TargetCodecVideo:        "h264",
+		TargetCodecAudio:        "aac",
+		SegmentDuration:         2,
+		SegmentRetentionSeconds: 600,
+	})
+	if err != nil {
+		t.Fatalf("StartTranscode: %v", err)
+	}
+	t.Cleanup(func() { _ = session.Close() })
+	server.sessions[sessionID] = session
+	server.lastAccess[sessionID] = time.Now()
+
+	const segmentName = "seg_00007.ts"
+	if err := os.WriteFile(filepath.Join(outputDir, segmentName), []byte("complete segment"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	segmentReq := withNodeRouteParams(
+		httptest.NewRequest(http.MethodGet, "/transcode/"+sessionID+"/segment/"+segmentName, nil),
+		map[string]string{"session_id": sessionID, "name": segmentName},
+	)
+	segmentReq.Header.Set(transcodeproxy.RequestHeader, "1")
+	segmentRR := httptest.NewRecorder()
+	server.handleSegment(segmentRR, segmentReq)
+	if segmentRR.Code != http.StatusOK {
+		t.Fatalf("segment status = %d, body = %q", segmentRR.Code, segmentRR.Body.String())
+	}
+	generation := segmentRR.Header().Get(transcodeproxy.GenerationHeader)
+	if generation == "" {
+		t.Fatal("proxied segment omitted generation")
+	}
+	if got := session.LastRequestedSegment(); got != 0 {
+		t.Fatalf("node counted proxy-hop completion as client completion: got %d, want 0", got)
+	}
+
+	ackReq := withNodeRouteParams(
+		httptest.NewRequest(http.MethodPost, "/transcode/"+sessionID+"/segment/"+segmentName+"/downloaded", nil),
+		map[string]string{"session_id": sessionID, "name": segmentName},
+	)
+	ackReq.Header.Set(transcodeproxy.GenerationHeader, generation)
+	ackRR := httptest.NewRecorder()
+	server.handleSegmentDownloaded(ackRR, ackReq)
+	if ackRR.Code != http.StatusNoContent {
+		t.Fatalf("ack status = %d, body = %q", ackRR.Code, ackRR.Body.String())
+	}
+	if got := session.LastRequestedSegment(); got != 7 {
+		t.Fatalf("acknowledged segment = %d, want 7", got)
+	}
+}
+
+func TestProxiedSegmentCompletionRejectsStaleGeneration(t *testing.T) {
+	truePath, err := exec.LookPath("true")
+	if err != nil {
+		t.Skipf("`true` not found in PATH: %v", err)
+	}
+
+	const sessionID = "stale-proxy-completion-session"
+	server := newTestServer(t)
+	session, err := playback.StartTranscode(context.Background(), playback.TranscodeOpts{
+		SessionID:        sessionID,
+		OutputDir:        t.TempDir(),
+		FFmpegPath:       truePath,
+		TargetCodecVideo: "h264",
+		TargetCodecAudio: "aac",
+		SegmentDuration:  2,
+	})
+	if err != nil {
+		t.Fatalf("StartTranscode: %v", err)
+	}
+	t.Cleanup(func() { _ = session.Close() })
+	server.sessions[sessionID] = session
+	server.lastAccess[sessionID] = time.Now()
+
+	const segmentName = "seg_00007.ts"
+	ackReq := withNodeRouteParams(
+		httptest.NewRequest(http.MethodPost, "/transcode/"+sessionID+"/segment/"+segmentName+"/downloaded", nil),
+		map[string]string{"session_id": sessionID, "name": segmentName},
+	)
+	ackReq.Header.Set(transcodeproxy.GenerationHeader, "999")
+	ackRR := httptest.NewRecorder()
+	server.handleSegmentDownloaded(ackRR, ackReq)
+	if ackRR.Code != http.StatusNoContent {
+		t.Fatalf("ack status = %d, body = %q", ackRR.Code, ackRR.Body.String())
+	}
+	if got := session.LastRequestedSegment(); got != 0 {
+		t.Fatalf("stale acknowledgement advanced segment to %d", got)
+	}
+}
+
+func withNodeRouteParams(req *http.Request, values map[string]string) *http.Request {
+	routeCtx := chi.NewRouteContext()
+	for key, value := range values {
+		routeCtx.URLParams.Add(key, value)
+	}
+	return req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, routeCtx))
+}
+
 func TestHandleStartRequireReadyRejectsExitedFFmpeg(t *testing.T) {
 	server := newTestServer(t)
 	ffmpegPath := filepath.Join(t.TempDir(), "failing-ffmpeg.sh")

internal/transcodeproxy/completion.go

@@ -0,0 +1,110 @@
+// Package transcodeproxy owns the private HTTP completion contract shared by
+// Silo's integrated API proxy and its dedicated proxy node.
+package transcodeproxy
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const (
+	// RequestHeader marks a segment hop whose immediate consumer is another
+	// Silo server, not the playback client. The transcode node defers download
+	// accounting until that server confirms downstream completion.
+	RequestHeader = "X-Silo-Transcode-Proxy"
+	// GenerationHeader carries an opaque session-incarnation and FFmpeg-timeline
+	// token. It is private to Silo hops and must not be forwarded to the client.
+	GenerationHeader = "X-Silo-Transcode-Segment-Generation"
+)
+
+var representationRequestHeaders = []string{
+	"Range",
+	"If-Range",
+	"If-Match",
+	"If-None-Match",
+	"If-Modified-Since",
+	"If-Unmodified-Since",
+}
+
+// PrepareRequest makes the upstream node serve the same byte representation
+// requested by the downstream client and suppresses completion at this hop.
+func PrepareRequest(upstream, downstream *http.Request) {
+	upstream.Header.Set(RequestHeader, "1")
+	for _, name := range representationRequestHeaders {
+		for _, value := range downstream.Header.Values(name) {
+			upstream.Header.Add(name, value)
+		}
+	}
+}
+
+// CopyResponseHeaders forwards public response metadata while retaining the
+// generation token inside the trusted Silo hop.
+func CopyResponseHeaders(dst, src http.Header) {
+	for name, values := range src {
+		if http.CanonicalHeaderKey(name) == GenerationHeader {
+			continue
+		}
+		for _, value := range values {
+			dst.Add(name, value)
+		}
+	}
+}
+
+// FullRepresentationSize resolves the total entity size needed to distinguish
+// a whole-file 206 response from an ordinary partial range.
+func FullRepresentationSize(resp *http.Response) int64 {
+	if resp == nil {
+		return -1
+	}
+	if resp.StatusCode == http.StatusOK {
+		return resp.ContentLength
+	}
+	if resp.StatusCode != http.StatusPartialContent {
+		return -1
+	}
+	_, total, ok := strings.Cut(resp.Header.Get("Content-Range"), "/")
+	if !ok || total == "*" {
+		return -1
+	}
+	size, err := strconv.ParseInt(total, 10, 64)
+	if err != nil || size <= 0 {
+		return -1
+	}
+	return size
+}
+
+// Acknowledge reports a completed downstream response to the transcode node.
+// The opaque generation token makes delayed acknowledgements harmless after a
+// restart or reconstruction.
+func Acknowledge(ctx context.Context, client *http.Client, targetURL, jwtSecret, generation string) error {
+	if client == nil {
+		client = http.DefaultClient
+	}
+	// The downstream request is commonly canceled as soon as its response body
+	// reaches the client, while the handler is still issuing this acknowledgement.
+	// Preserve request-scoped values, but give the post-response hop its own
+	// bounded lifetime so a successful delivery is not mistaken for a disconnect.
+	ackCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ackCtx, http.MethodPost, targetURL+"/downloaded", nil)
+	if err != nil {
+		return fmt.Errorf("build acknowledgement: %w", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+jwtSecret)
+	req.Header.Set(GenerationHeader, generation)
+	resp, err := client.Do(req)
+	if err != nil {
+		return fmt.Errorf("send acknowledgement: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10))
+	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
+		return fmt.Errorf("acknowledgement status %d", resp.StatusCode)
+	}
+	return nil
+}

internal/transcodeproxy/completion_test.go

@@ -0,0 +1,42 @@
+package transcodeproxy
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestAcknowledgeSurvivesCompletedDownstreamRequest(t *testing.T) {
+	const (
+		jwtSecret  = "test-secret"
+		generation = "incarnation:generation"
+	)
+
+	acknowledged := make(chan struct{}, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			t.Errorf("method = %q, want POST", r.Method)
+		}
+		if got := r.Header.Get("Authorization"); got != "Bearer "+jwtSecret {
+			t.Errorf("Authorization = %q", got)
+		}
+		if got := r.Header.Get(GenerationHeader); got != generation {
+			t.Errorf("generation = %q, want %q", got, generation)
+		}
+		acknowledged <- struct{}{}
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	t.Cleanup(server.Close)
+
+	requestCtx, cancelRequest := context.WithCancel(context.Background())
+	cancelRequest()
+	if err := Acknowledge(requestCtx, server.Client(), server.URL+"/segment", jwtSecret, generation); err != nil {
+		t.Fatalf("Acknowledge() error = %v", err)
+	}
+	select {
+	case <-acknowledged:
+	default:
+		t.Fatal("acknowledgement request was not delivered")
+	}
+}

web/src/pages/admin-settings/PlaybackSettings.tsx

@@ -16,6 +16,7 @@ import {
 const KEYS = [
   "playback.ffmpeg_path",
   "playback.transcode_dir",
+  "playback.segment_retention_seconds",
   "playback.hw_accel",
   "playback.hw_device",
   "playback.transcode_enabled",
@@ -197,6 +198,13 @@ export default function PlaybackSettings() {
         </FieldGroup>
 
         <FieldGroup label="Segments">
+          <SettingField
+            label="Transcode Back Buffer (seconds)"
+            type="number"
+            hint="Keeps this much already-downloaded media for instant backward seeking, then reclaims older transcode segments. Use 0 to disable cleanup; enabled values must be at least 120 seconds. Pair with transcode throttling to bound both behind- and ahead-of-client disk usage."
+            value={form.getValue("playback.segment_retention_seconds")}
+            onChange={(v) => form.setValue("playback.segment_retention_seconds", v)}
+          />
           <SettingField
             label="Chapter Thumbnail Workers"
             type="number"