feat(playback): formalize resumable direct streams and stall observability (#464)

* feat(playback): formalize resumable direct streams and stall observability

Implements #443: strong stat-based ETag + If-Range on original-file direct
play (via http.ServeContent), stream-end outcome classification in
RollingDeadlineWriter (stalled_reap vs client_gone vs completed) with a
structured log event and Prometheus counters, the direct_stream_resume_v1
protocol-v3 capability, and a contract doc. Progressive remux is explicitly
excluded from the resume contract.

Code written by OpenAI Codex CLI (gpt-5.6-sol) from a Claude-authored spec;
reviewed and verified by Claude.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(playback): harden direct stream resume contract

* test(playback): cover resume platform contracts

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Quick
2026-07-24 14:22:12 -04:00
committed by GitHub
co-authored by Claude Fable 5
parent 22fec4ed2d
commit ee31a1f0e2
17 changed files with 1065 additions and 13 deletions
+43
View File
@@ -0,0 +1,43 @@
# Direct Stream Resumption
Protocol v3 advertises `direct_stream_resume_v1` for playback plans whose
delivery is `original_http`. The capability formalizes resumption of the
original file by issuing sequential authorized HTTP requests. It does not
change authentication, authorization, or playback-session ownership.
Original-file responses follow the HTTP byte-range contract:
- `Accept-Ranges: bytes` advertises byte addressing.
- A satisfiable `Range` request returns `206 Partial Content` with the selected
interval in `Content-Range`.
- Open-ended and suffix byte ranges are supported.
- A range starting at or past end of file, or another invalid range, returns
`416 Requested Range Not Satisfiable` with `Content-Range: bytes */<size>`.
- `HEAD` returns the same representation headers as `GET` without a body.
On Linux, macOS, and Windows, each response carries a strong, opaque `ETag`
derived from the open file's filesystem identity, change time, modification
time, and size. The validator is stable while the playback plan's original-file
entity is unchanged, but changes for same-size replacements even when their
modification time is preserved. Platforms that cannot expose a durable
filesystem revision omit the validator instead of hashing an entire media file
before each request. On those platforms an ETag-based `If-Range` request cannot
match and safely falls back to a full `200 OK` response.
A client resuming a transfer sends both `Range` and `If-Range` with the
validator. When it still matches, the server returns the requested `206`
response. When the entity changed, the server ignores the range and returns the
entire current entity as `200 OK`, preventing bytes from different revisions
from being combined. `If-None-Match` uses the same validator for ordinary
conditional requests.
Playback sessions already treat each transport request independently.
Sequential ranged requests refresh transport activity, and cleanup never
expires a session while one of those transport requests is active. A late
request within the paused-session grace remains valid under the same rules as
the initial request.
The capability does not apply to progressive remux delivery
`server_remux_progressive`, which is not byte-resumable. It applies only to
`original_http`; remux and transcode transports retain their existing
contracts.
+4 -4
View File
@@ -353,7 +353,7 @@ func (h *PlaybackHandler) HandlePlaybackCapabilityV3(w http.ResponseWriter, r *h
writeJSON(w, http.StatusOK, response)
return
}
response.Features = []string{playback.FeaturePlaybackPlanV3, playback.FeatureMedia3Only, playback.FeatureDetailedDecodeV3, playback.FeatureLayoutPassthrough, playback.FeatureRouteDiagnostics, playback.FeatureDeviceQuirksV3, playback.FeatureSeekReanchorV3}
response.Features = playback.ServerFeaturesV3()
response.Deliveries = []playback.DeliveryV3{playback.DeliveryOriginalHTTPV3, playback.DeliveryRemuxProgressiveV3, playback.DeliveryRemuxHLSV3, playback.DeliveryTranscodeHLSV3}
response.Transformations = h.transformationRegistryV3(r.Context()).Advertised()
writeJSON(w, http.StatusOK, response)
@@ -536,7 +536,7 @@ func (h *PlaybackHandler) startPlannedPlaybackV3(r *http.Request, userID int, pr
abort()
return playback.DecisionResponseV3{}, &transportErrorV3{reason: "subtitle_artifact_unavailable", message: "Failed to prepare the selected subtitle artifact.", cause: err}
}
response := playback.DecisionResponseV3{ProtocolVersion: playback.ProtocolV3, ServerFeatures: []string{playback.FeaturePlaybackPlanV3, playback.FeatureMedia3Only, playback.FeatureRouteDiagnostics, playback.FeatureDeviceQuirksV3, playback.FeatureSeekReanchorV3}, Outcome: playback.OutcomePlayableV3, SessionID: session.ID, PlaybackPlan: result.Plan}
response := playback.DecisionResponseV3{ProtocolVersion: playback.ProtocolV3, ServerFeatures: playback.ServerFeaturesV3(), Outcome: playback.OutcomePlayableV3, SessionID: session.ID, PlaybackPlan: result.Plan}
record := playback.AttemptRecordV3{PlaybackAttemptID: req.PlaybackAttemptID, SessionID: session.ID, UserID: userID, ProfileID: profileID, RequestedMediaFileID: requestedFile.ID, EffectiveMediaFileID: effectiveFile.ID, CurrentPlanID: result.Plan.PlanID, CurrentPlan: *result.Plan, NormalizedRequest: req, RequestDigest: requestDigest, ExpiresAt: time.Now().Add(playback.MaxTokenTTL)}
if err := h.updateV3SessionState(r.Context(), session, effectiveFile, result, transport); err != nil {
transport.rollback()
@@ -1270,7 +1270,7 @@ func (h *PlaybackHandler) executeReplanV3(r *http.Request, record *playback.Atte
}
}
}
response := playback.DecisionResponseV3{ProtocolVersion: playback.ProtocolV3, ServerFeatures: []string{playback.FeaturePlaybackPlanV3, playback.FeatureMedia3Only, playback.FeatureRouteDiagnostics, playback.FeatureDeviceQuirksV3, playback.FeatureSeekReanchorV3}, Outcome: playback.OutcomePlayableV3, SessionID: session.ID, PlaybackPlan: result.Plan}
response := playback.DecisionResponseV3{ProtocolVersion: playback.ProtocolV3, ServerFeatures: playback.ServerFeaturesV3(), Outcome: playback.OutcomePlayableV3, SessionID: session.ID, PlaybackPlan: result.Plan}
updated := *record
updated.CurrentPlanID = result.Plan.PlanID
updated.CurrentPlan = *result.Plan
@@ -1846,7 +1846,7 @@ func decisionResponseFromAttemptV3(record *playback.AttemptRecordV3) playback.De
if plan.RuntimeCorrections == nil {
plan.RuntimeCorrections = []string{}
}
return playback.DecisionResponseV3{ProtocolVersion: playback.ProtocolV3, ServerFeatures: []string{playback.FeaturePlaybackPlanV3, playback.FeatureMedia3Only, playback.FeatureRouteDiagnostics, playback.FeatureDeviceQuirksV3, playback.FeatureSeekReanchorV3}, Outcome: playback.OutcomePlayableV3, SessionID: record.SessionID, PlaybackPlan: &plan}
return playback.DecisionResponseV3{ProtocolVersion: playback.ProtocolV3, ServerFeatures: playback.ServerFeaturesV3(), Outcome: playback.OutcomePlayableV3, SessionID: record.SessionID, PlaybackPlan: &plan}
}
func completedReplanResponseMatchesAttemptV3(raw json.RawMessage, record *playback.AttemptRecordV3) bool {
+4 -1
View File
@@ -137,7 +137,10 @@ func TestHandlePlaybackCapabilityV3ReadsFlagPerRequest(t *testing.T) {
handler.v3FlagMu.Lock()
handler.v3Flags = nil
handler.v3FlagMu.Unlock()
if response := request(); !response.Enabled || len(response.Deliveries) != 4 || !playback.HasFeatureV3(response.Features, playback.FeatureSeekReanchorV3) {
if response := request(); !response.Enabled ||
len(response.Deliveries) != 4 ||
!playback.HasFeatureV3(response.Features, playback.FeatureSeekReanchorV3) ||
!playback.HasFeatureV3(response.Features, playback.FeatureDirectStreamResumeV3) {
t.Fatalf("enabled response = %#v", response)
}
}
+73
View File
@@ -8,6 +8,7 @@ import (
"net/http/httptest"
"os"
"testing"
"time"
"github.com/go-chi/chi/v5"
@@ -35,6 +36,78 @@ func (m *hookedSessionManager) BeginTransport(sessionID string) error {
return m.SessionManager.BeginTransport(sessionID)
}
func TestHandleStream_PausedSessionResumesWithDelayedRangeRequest(t *testing.T) {
const (
contentID = "movie-1"
sessionRouteKey = "session_id"
)
filePath := writePlaybackTestMediaFile(t, "movie.mp4")
file := &models.MediaFile{
ID: 42,
ContentID: contentID,
FilePath: filePath,
Duration: 3600,
}
sessionMgr := playback.NewSessionManager(0, 0)
session, err := sessionMgr.StartSession(1, "profile-1", 42, playback.PlayDirect, false)
if err != nil {
t.Fatalf("StartSession: %v", err)
}
if err := sessionMgr.UpdateProgress(session.ID, 1, true); err != nil {
t.Fatalf("UpdateProgress(paused): %v", err)
}
handler := NewStreamHandler(sessionMgr, testPlaybackFileResolver{file: file})
request := func(rangeHeader, ifRange string) *httptest.ResponseRecorder {
t.Helper()
req := playbackTestRequest(
http.MethodGet,
"/api/v1/stream/"+session.ID,
nil,
map[string]string{sessionRouteKey: session.ID},
)
if rangeHeader != "" {
req.Header.Set("Range", rangeHeader)
}
if ifRange != "" {
req.Header.Set("If-Range", ifRange)
}
rr := httptest.NewRecorder()
handler.HandleStream(rr, req)
return rr
}
initial := request("", "")
if initial.Code != http.StatusOK {
t.Fatalf("initial status = %d, body = %s", initial.Code, initial.Body.String())
}
etag := initial.Header().Get("ETag")
if etag == "" {
t.Fatal("initial response omitted ETag")
}
const (
activeGrace = 5 * time.Millisecond
pausedGrace = 5 * time.Second
)
time.Sleep(20 * time.Millisecond)
sessionMgr.CleanInactive(activeGrace, pausedGrace)
if _, err := sessionMgr.GetSession(session.ID); err != nil {
t.Fatalf("paused session expired before ranged resume: %v", err)
}
resumed := request("bytes=2-", etag)
if resumed.Code != http.StatusPartialContent {
t.Fatalf("resume status = %d, body = %s", resumed.Code, resumed.Body.String())
}
if got := resumed.Body.String(); got != "deo" {
t.Fatalf("resume body = %q, want %q", got, "deo")
}
if live, err := sessionMgr.GetSession(session.ID); err != nil || live.ID != session.ID {
t.Fatalf("ranged request did not preserve session %q: session=%#v err=%v", session.ID, live, err)
}
}
func TestHandleStream_AbortsSessionWhenDirectPlayFileDisappearsAfterPreflight(t *testing.T) {
filePath := writePlaybackTestMediaFile(t, "movie.mkv")
file := &models.MediaFile{
+70 -1
View File
@@ -10,7 +10,10 @@
package httpstream
import (
"context"
"errors"
"io"
"net"
"net/http"
"os"
"strconv"
@@ -34,6 +37,15 @@ const (
readFromChunk int64 = 64 << 20
)
// StreamOutcome classifies how a streaming response ended.
type StreamOutcome string
const (
OutcomeCompleted StreamOutcome = "completed"
OutcomeStalledReap StreamOutcome = "stalled_reap"
OutcomeClientGone StreamOutcome = "client_gone"
)
// StallWindow returns the configured stall window for streaming responses.
func StallWindow() time.Duration {
if v := os.Getenv(stallWindowEnv); v != "" {
@@ -58,6 +70,10 @@ type RollingDeadlineWriter struct {
step time.Duration
lastBump time.Time
disabled bool
statusCode int
bytesWritten int64
firstWriteErr error
}
// NewRollingDeadlineWriter wraps w with the configured stall window.
@@ -95,12 +111,20 @@ func (s *RollingDeadlineWriter) Header() http.Header { return s.w.Header() }
func (s *RollingDeadlineWriter) WriteHeader(code int) {
s.bump()
if s.statusCode == 0 {
s.statusCode = code
}
s.w.WriteHeader(code)
}
func (s *RollingDeadlineWriter) Write(p []byte) (int, error) {
s.bump()
return s.w.Write(p)
if s.statusCode == 0 {
s.statusCode = http.StatusOK
}
n, err := s.w.Write(p)
s.recordWrite(int64(n), err)
return n, err
}
// ReadFrom preserves the underlying ResponseWriter's io.ReaderFrom fast path
@@ -116,8 +140,12 @@ func (s *RollingDeadlineWriter) ReadFrom(r io.Reader) (int64, error) {
var total int64
for {
s.bump()
if s.statusCode == 0 {
s.statusCode = http.StatusOK
}
n, err := rf.ReadFrom(io.LimitReader(r, readFromChunk))
total += n
s.recordWrite(n, err)
if err != nil {
return total, err
}
@@ -132,7 +160,48 @@ func (s *RollingDeadlineWriter) Flush() {
_ = s.rc.Flush()
}
// StatusCode returns the response status observed by the wrapper.
func (s *RollingDeadlineWriter) StatusCode() int {
return s.statusCode
}
// BytesWritten returns the number of response body bytes accepted by the
// underlying writer.
func (s *RollingDeadlineWriter) BytesWritten() int64 {
return s.bytesWritten
}
// 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 {
if isTimeoutError(s.firstWriteErr) {
return OutcomeStalledReap
}
if s.firstWriteErr != nil || (ctx != nil && ctx.Err() != nil) {
return OutcomeClientGone
}
return OutcomeCompleted
}
// Unwrap lets http.ResponseController traverse to the underlying writer.
func (s *RollingDeadlineWriter) Unwrap() http.ResponseWriter { return s.w }
func (s *RollingDeadlineWriter) recordWrite(n int64, err error) {
s.bytesWritten += n
if err != nil && s.firstWriteErr == nil {
s.firstWriteErr = err
}
}
func isTimeoutError(err error) bool {
if err == nil {
return false
}
if errors.Is(err, os.ErrDeadlineExceeded) {
return true
}
var netErr net.Error
return errors.As(err, &netErr) && netErr.Timeout()
}
type writerOnly struct{ io.Writer }
+159 -5
View File
@@ -2,11 +2,14 @@ package httpstream
import (
"bufio"
"context"
"fmt"
"io"
"net"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"time"
@@ -86,9 +89,13 @@ func TestUnwrappedStreamStillKilledAtWriteTimeout(t *testing.T) {
// connection: a client that stops reading must cause a write error within
// roughly the stall window, not never.
func TestStalledClientReaped(t *testing.T) {
handlerDone := make(chan error, 1)
type result struct {
err error
outcome StreamOutcome
}
handlerDone := make(chan result, 1)
srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
sw := newRollingDeadlineWriter(w, 1*time.Second, 0)
sw := newRollingDeadlineWriter(w, 500*time.Millisecond, 0)
sw.WriteHeader(http.StatusOK)
buf := make([]byte, 1<<20)
var err error
@@ -98,7 +105,7 @@ func TestStalledClientReaped(t *testing.T) {
}
sw.Flush()
}
handlerDone <- err
handlerDone <- result{err: err, outcome: sw.Outcome(r.Context())}
}))
srv.Config.WriteTimeout = 0 // isolate: only the rolling deadline may reap
srv.Start()
@@ -116,15 +123,154 @@ func TestStalledClientReaped(t *testing.T) {
}
select {
case err := <-handlerDone:
if err == nil {
case got := <-handlerDone:
if got.err == nil {
t.Fatal("handler finished 256MB into a non-reading client without error")
}
if got.outcome != OutcomeStalledReap {
t.Fatalf("outcome = %q, want %q (error: %v)", got.outcome, OutcomeStalledReap, got.err)
}
case <-time.After(30 * time.Second):
t.Fatal("stalled client was never reaped by the rolling deadline")
}
}
func TestDisconnectedClientClassifiedClientGone(t *testing.T) {
type result struct {
err error
outcome StreamOutcome
}
startWriting := make(chan struct{})
handlerDone := make(chan result, 1)
srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
sw := newRollingDeadlineWriter(w, 5*time.Second, 0)
sw.WriteHeader(http.StatusOK)
_, _ = sw.Write([]byte("ready"))
sw.Flush()
<-startWriting
buf := make([]byte, 1<<20)
var err error
for i := 0; i < 256; i++ {
if _, err = sw.Write(buf); err != nil {
break
}
sw.Flush()
}
handlerDone <- result{err: err, outcome: sw.Outcome(r.Context())}
}))
srv.Config.WriteTimeout = 0
srv.Start()
defer srv.Close()
conn, err := net.Dial("tcp", strings.TrimPrefix(srv.URL, "http://"))
if err != nil {
t.Fatalf("dial: %v", err)
}
if _, err := fmt.Fprintf(conn, "GET / HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
t.Fatalf("request: %v", err)
}
if _, err := bufio.NewReader(conn).ReadString('\n'); err != nil {
t.Fatalf("status line: %v", err)
}
if err := conn.Close(); err != nil {
t.Fatalf("close: %v", err)
}
close(startWriting)
select {
case got := <-handlerDone:
if got.err == nil {
t.Fatal("handler completed after the client disconnected")
}
if got.outcome != OutcomeClientGone {
t.Fatalf("outcome = %q, want %q (error: %v)", got.outcome, OutcomeClientGone, got.err)
}
case <-time.After(10 * time.Second):
t.Fatal("handler did not observe the client disconnect")
}
}
func TestWriteOutcomeCompletedAndCounted(t *testing.T) {
rr := httptest.NewRecorder()
sw := newRollingDeadlineWriter(rr, time.Second, 0)
sw.WriteHeader(http.StatusCreated)
const body = "completed body"
n, err := sw.Write([]byte(body))
if err != nil {
t.Fatal(err)
}
if n != len(body) || sw.BytesWritten() != int64(len(body)) {
t.Fatalf("bytes = (%d, %d), want %d", n, sw.BytesWritten(), len(body))
}
if sw.StatusCode() != http.StatusCreated {
t.Fatalf("status = %d, want %d", sw.StatusCode(), http.StatusCreated)
}
if outcome := sw.Outcome(context.Background()); outcome != OutcomeCompleted {
t.Fatalf("outcome = %q, want %q", outcome, OutcomeCompleted)
}
}
func TestServeContentReadFromOutcomeCompletedAndCounted(t *testing.T) {
const totalSize = 2 << 20
filePath := filepath.Join(t.TempDir(), "source.bin")
if err := os.WriteFile(filePath, bytesOf('x', totalSize), 0o600); err != nil {
t.Fatal(err)
}
type result struct {
status int
bytes int64
outcome StreamOutcome
}
handlerDone := make(chan result, 1)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
f, err := os.Open(filePath)
if err != nil {
t.Error(err)
return
}
defer f.Close()
stat, err := f.Stat()
if err != nil {
t.Error(err)
return
}
sw := newRollingDeadlineWriter(w, 5*time.Second, 0)
http.ServeContent(sw, r, stat.Name(), stat.ModTime(), f)
handlerDone <- result{status: sw.StatusCode(), bytes: sw.BytesWritten(), outcome: sw.Outcome(r.Context())}
}))
defer srv.Close()
resp, err := http.Get(srv.URL)
if err != nil {
t.Fatal(err)
}
n, readErr := io.Copy(io.Discard, resp.Body)
closeErr := resp.Body.Close()
if readErr != nil || closeErr != nil {
t.Fatalf("read = %v, close = %v", readErr, closeErr)
}
if n != totalSize {
t.Fatalf("response bytes = %d, want %d", n, totalSize)
}
select {
case got := <-handlerDone:
if got.status != http.StatusOK {
t.Fatalf("status = %d, want 200", got.status)
}
if got.bytes != totalSize {
t.Fatalf("counted bytes = %d, want %d", got.bytes, totalSize)
}
if got.outcome != OutcomeCompleted {
t.Fatalf("outcome = %q, want %q", got.outcome, OutcomeCompleted)
}
case <-time.After(5 * time.Second):
t.Fatal("handler did not complete")
}
}
// 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.
@@ -173,6 +319,14 @@ type slowReader struct {
delay time.Duration
}
func bytesOf(value byte, size int) []byte {
buf := make([]byte, size)
for i := range buf {
buf[i] = value
}
return buf
}
func (s *slowReader) Read(p []byte) (int, error) {
time.Sleep(s.delay)
if len(p) > 32<<10 {
+67 -1
View File
@@ -1,9 +1,12 @@
package playback
import (
"fmt"
"log/slog"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"github.com/Silo-Server/silo-server/internal/httpstream"
@@ -54,7 +57,8 @@ func MimeFromExtension(name string) string {
func ServeDirectPlay(w http.ResponseWriter, r *http.Request, filePath string) error {
// Media bodies routinely take longer than the server's absolute
// WriteTimeout; roll the write deadline with progress instead.
w = httpstream.NewRollingDeadlineWriter(w)
streamWriter := httpstream.NewRollingDeadlineWriter(w)
w = streamWriter
f, err := os.Open(filePath)
if err != nil {
if os.IsNotExist(err) {
@@ -71,10 +75,72 @@ func ServeDirectPlay(w http.ResponseWriter, r *http.Request, filePath string) er
http.Error(w, "internal server error", http.StatusInternalServerError)
return err
}
w = &directPlayResponseWriter{
RollingDeadlineWriter: streamWriter,
size: stat.Size(),
}
w.Header().Del("ETag")
if etag := directPlayEntityTag(f, stat); etag != "" {
w.Header().Set("ETag", etag)
}
// Set Content-Type explicitly so ServeContent does not sniff.
w.Header().Set("Content-Type", MimeFromExtension(filePath))
hadRange := len(r.Header.Values("Range")) > 0
hadIfRange := len(r.Header.Values("If-Range")) > 0
directStreamActive.Inc()
http.ServeContent(w, r, stat.Name(), stat.ModTime(), f)
outcome := streamWriter.Outcome(r.Context())
status := streamWriter.StatusCode()
bytesSent := streamWriter.BytesWritten()
rangeStart := directStreamRangeStart(status, w.Header().Get("Content-Range"))
recordDirectStreamEnd(outcome, status, bytesSent, rangeStart)
slog.InfoContext(r.Context(), "direct stream ended",
"component", "playback",
"outcome", outcome,
"status", status,
"bytes_sent", bytesSent,
"range_requested", hadRange,
"range_start", rangeStart,
"had_if_range", hadIfRange,
)
return nil
}
type directPlayResponseWriter struct {
*httpstream.RollingDeadlineWriter
size int64
}
func (w *directPlayResponseWriter) WriteHeader(status int) {
if status == http.StatusRequestedRangeNotSatisfiable {
w.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", w.size))
}
w.RollingDeadlineWriter.WriteHeader(status)
}
func directStreamRangeStart(status int, contentRange string) int64 {
if status != http.StatusPartialContent {
return -1
}
value, ok := strings.CutPrefix(strings.TrimSpace(contentRange), "bytes ")
if !ok {
return -1
}
bounds, _, ok := strings.Cut(value, "/")
if !ok {
return -1
}
start, _, ok := strings.Cut(strings.TrimSpace(bounds), "-")
if !ok || start == "" {
return -1
}
parsedStart, err := strconv.ParseInt(strings.TrimSpace(start), 10, 64)
if err != nil || parsedStart < 0 {
return -1
}
return parsedStart
}
+19
View File
@@ -0,0 +1,19 @@
package playback
import (
"crypto/sha256"
"fmt"
"os"
)
const directPlayETagVersion = "dsr1"
func directPlayEntityTag(file *os.File, info os.FileInfo) string {
revision, ok := directPlayFilesystemRevision(file, info)
if !ok {
return ""
}
digest := sha256.Sum256([]byte(directPlayETagVersion + "\x00" + revision))
return fmt.Sprintf("\"%s-%x\"", directPlayETagVersion, digest)
}
+44
View File
@@ -0,0 +1,44 @@
package playback
import (
"net/http"
"github.com/Silo-Server/silo-server/internal/httpstream"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
var (
directStreamActive = promauto.NewGauge(prometheus.GaugeOpts{
Name: "silo_direct_stream_active",
Help: "Number of original-file direct streams currently being served.",
})
directStreamEnds = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "silo_direct_stream_ends_total",
Help: "Number of original-file direct streams by terminal outcome.",
}, []string{"outcome"})
directStreamRangeResumes = promauto.NewCounter(prometheus.CounterOpts{
Name: "silo_direct_stream_range_resumes_total",
Help: "Number of successful original-file byte-range resumes after byte zero.",
})
directStreamInvalidRanges = promauto.NewCounter(prometheus.CounterOpts{
Name: "silo_direct_stream_invalid_range_total",
Help: "Number of original-file direct stream requests rejected with HTTP 416.",
})
directStreamBytesSent = promauto.NewCounter(prometheus.CounterOpts{
Name: "silo_direct_stream_bytes_sent_total",
Help: "Number of original-file direct stream response body bytes sent.",
})
)
func recordDirectStreamEnd(outcome httpstream.StreamOutcome, status int, bytesSent, rangeStart int64) {
directStreamActive.Dec()
directStreamEnds.WithLabelValues(string(outcome)).Inc()
directStreamBytesSent.Add(float64(bytesSent))
if status == http.StatusPartialContent && rangeStart > 0 {
directStreamRangeResumes.Inc()
}
if status == http.StatusRequestedRangeNotSatisfiable {
directStreamInvalidRanges.Inc()
}
}
@@ -0,0 +1,25 @@
//go:build darwin
package playback
import (
"fmt"
"os"
"syscall"
)
func directPlayFilesystemRevision(_ *os.File, info os.FileInfo) (string, bool) {
stat, ok := info.Sys().(*syscall.Stat_t)
if !ok {
return "", false
}
return fmt.Sprintf(
"darwin:%x:%x:%x:%x:%x:%x",
uint64(stat.Dev),
stat.Ino,
stat.Ctimespec.Sec,
stat.Ctimespec.Nsec,
info.ModTime().UnixNano(),
info.Size(),
), true
}
@@ -0,0 +1,9 @@
//go:build !darwin && !linux && !windows
package playback
import "os"
func directPlayFilesystemRevision(*os.File, os.FileInfo) (string, bool) {
return "", false
}
@@ -0,0 +1,25 @@
//go:build linux
package playback
import (
"fmt"
"os"
"syscall"
)
func directPlayFilesystemRevision(_ *os.File, info os.FileInfo) (string, bool) {
stat, ok := info.Sys().(*syscall.Stat_t)
if !ok {
return "", false
}
return fmt.Sprintf(
"linux:%x:%x:%x:%x:%x:%x",
uint64(stat.Dev),
stat.Ino,
stat.Ctim.Sec,
stat.Ctim.Nsec,
info.ModTime().UnixNano(),
info.Size(),
), true
}
@@ -0,0 +1,49 @@
//go:build windows
package playback
import (
"fmt"
"os"
"unsafe"
"golang.org/x/sys/windows"
)
type windowsFileBasicInfo struct {
CreationTime int64
LastAccessTime int64
LastWriteTime int64
ChangeTime int64
FileAttributes uint32
}
func directPlayFilesystemRevision(file *os.File, _ os.FileInfo) (string, bool) {
handle := windows.Handle(file.Fd())
var identity windows.ByHandleFileInformation
if err := windows.GetFileInformationByHandle(handle, &identity); err != nil {
return "", false
}
var basic windowsFileBasicInfo
if err := windows.GetFileInformationByHandleEx(
handle,
windows.FileBasicInfo,
(*byte)(unsafe.Pointer(&basic)),
uint32(unsafe.Sizeof(basic)),
); err != nil {
return "", false
}
size := uint64(identity.FileSizeHigh)<<32 | uint64(identity.FileSizeLow)
fileID := uint64(identity.FileIndexHigh)<<32 | uint64(identity.FileIndexLow)
return fmt.Sprintf(
"windows:%x:%x:%x:%x:%x",
identity.VolumeSerialNumber,
fileID,
basic.ChangeTime,
basic.LastWriteTime,
size,
), true
}
+359
View File
@@ -0,0 +1,359 @@
package playback
import (
"fmt"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"runtime"
"strings"
"testing"
"time"
"github.com/Silo-Server/silo-server/internal/httpstream"
"github.com/prometheus/client_golang/prometheus"
dto "github.com/prometheus/client_model/go"
)
const (
directPlayDarwinGOOS = "darwin"
directPlayLinuxGOOS = "linux"
directPlayWindowsGOOS = "windows"
)
func TestServeDirectPlayHTTPContract(t *testing.T) {
const content = "0123456789abcdefghijklmnopqrstuvwxyz"
filePath := filepath.Join(t.TempDir(), "fixture.mp4")
if err := os.WriteFile(filePath, []byte(content), 0o600); err != nil {
t.Fatal(err)
}
serve := func(method, rangeHeader, ifRange, ifNoneMatch string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(method, "/stream", nil)
if rangeHeader != "" {
req.Header.Set("Range", rangeHeader)
}
if ifRange != "" {
req.Header.Set("If-Range", ifRange)
}
if ifNoneMatch != "" {
req.Header.Set("If-None-Match", ifNoneMatch)
}
rr := httptest.NewRecorder()
if err := ServeDirectPlay(rr, req, filePath); err != nil {
t.Fatalf("ServeDirectPlay: %v", err)
}
return rr
}
full := serve(http.MethodGet, "", "", "")
if full.Code != http.StatusOK {
t.Fatalf("full status = %d, want 200", full.Code)
}
if body := full.Body.String(); body != content {
t.Fatalf("full body = %q, want %q", body, content)
}
if got := full.Header().Get("Accept-Ranges"); got != "bytes" {
t.Fatalf("Accept-Ranges = %q, want bytes", got)
}
etag := full.Header().Get("ETag")
validatorRequired := platformRequiresDirectPlayValidator()
if validatorRequired && etag == "" {
t.Fatalf("ETag omitted on supported platform %s", runtime.GOOS)
}
if !validatorRequired && etag != "" {
t.Fatalf("ETag = %q on unsupported platform %s, want omitted validator", etag, runtime.GOOS)
}
if etag != "" && (strings.HasPrefix(etag, "W/") || !strings.HasPrefix(etag, "\"") || !strings.HasSuffix(etag, "\"")) {
t.Fatalf("ETag = %q, want a strong quoted validator", etag)
}
t.Run("HEAD", func(t *testing.T) {
rr := serve(http.MethodHead, "", "", "")
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rr.Code)
}
if rr.Body.Len() != 0 {
t.Fatalf("body length = %d, want 0", rr.Body.Len())
}
if rr.Header().Get("ETag") != etag {
t.Fatalf("ETag = %q, want %q", rr.Header().Get("ETag"), etag)
}
if rr.Header().Get("Accept-Ranges") != "bytes" {
t.Fatalf("Accept-Ranges = %q, want bytes", rr.Header().Get("Accept-Ranges"))
}
if rr.Header().Get("Content-Length") != fmt.Sprint(len(content)) {
t.Fatalf("Content-Length = %q, want %d", rr.Header().Get("Content-Length"), len(content))
}
})
tests := []struct {
name string
rangeHeader string
wantStatus int
wantRange string
wantBody string
wantStart int64
}{
{
name: "bounded range",
rangeHeader: "bytes=5-9",
wantStatus: http.StatusPartialContent,
wantRange: fmt.Sprintf("bytes 5-9/%d", len(content)),
wantBody: content[5:10],
wantStart: 5,
},
{
name: "suffix range",
rangeHeader: "bytes=-4",
wantStatus: http.StatusPartialContent,
wantRange: fmt.Sprintf("bytes %d-%d/%d", len(content)-4, len(content)-1, len(content)),
wantBody: content[len(content)-4:],
wantStart: int64(len(content) - 4),
},
{
name: "open ended range",
rangeHeader: "bytes=10-",
wantStatus: http.StatusPartialContent,
wantRange: fmt.Sprintf("bytes 10-%d/%d", len(content)-1, len(content)),
wantBody: content[10:],
wantStart: 10,
},
{
name: "syntactically invalid range",
rangeHeader: "bytes=invalid",
wantStatus: http.StatusRequestedRangeNotSatisfiable,
wantRange: fmt.Sprintf("bytes */%d", len(content)),
},
{
name: "unsatisfiable range",
rangeHeader: "bytes=999-1000",
wantStatus: http.StatusRequestedRangeNotSatisfiable,
wantRange: fmt.Sprintf("bytes */%d", len(content)),
},
{
name: "range starts at EOF",
rangeHeader: fmt.Sprintf("bytes=%d-", len(content)),
wantStatus: http.StatusRequestedRangeNotSatisfiable,
wantRange: fmt.Sprintf("bytes */%d", len(content)),
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
resumesBefore := counterValue(t, directStreamRangeResumes)
rr := serve(http.MethodGet, tt.rangeHeader, "", "")
if rr.Code != tt.wantStatus {
t.Fatalf("status = %d, want %d; body = %q", rr.Code, tt.wantStatus, rr.Body.String())
}
if got := rr.Header().Get("Content-Range"); got != tt.wantRange {
t.Fatalf("Content-Range = %q, want %q", got, tt.wantRange)
}
if tt.wantStatus == http.StatusPartialContent && rr.Body.String() != tt.wantBody {
t.Fatalf("body = %q, want %q", rr.Body.String(), tt.wantBody)
}
if tt.wantStatus == http.StatusPartialContent {
if got := directStreamRangeStart(rr.Code, rr.Header().Get("Content-Range")); got != tt.wantStart {
t.Fatalf("range start = %d, want %d", got, tt.wantStart)
}
if got := counterValue(t, directStreamRangeResumes); got != resumesBefore+1 {
t.Fatalf("resume counter = %v, want %v", got, resumesBefore+1)
}
}
})
}
t.Run("matching If-Range", func(t *testing.T) {
if !validatorRequired {
t.Skip("platform does not expose a durable file revision")
}
rr := serve(http.MethodGet, "bytes=7-", etag, "")
if rr.Code != http.StatusPartialContent {
t.Fatalf("status = %d, want 206", rr.Code)
}
if body := rr.Body.String(); body != content[7:] {
t.Fatalf("body = %q, want %q", body, content[7:])
}
})
t.Run("stale If-Range", func(t *testing.T) {
rr := serve(http.MethodGet, "bytes=7-", "\"stale\"", "")
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rr.Code)
}
if body := rr.Body.String(); body != content {
t.Fatalf("body = %q, want full entity %q", body, content)
}
})
t.Run("If-None-Match", func(t *testing.T) {
if !validatorRequired {
t.Skip("platform does not expose a durable file revision")
}
rr := serve(http.MethodGet, "", "", etag)
if rr.Code != http.StatusNotModified {
t.Fatalf("status = %d, want 304", rr.Code)
}
if rr.Body.Len() != 0 {
t.Fatalf("body length = %d, want 0", rr.Body.Len())
}
})
}
func TestServeDirectPlayChangedEntityRejectsOldIfRange(t *testing.T) {
if !platformRequiresDirectPlayValidator() {
t.Skip("platform does not expose a durable file revision")
}
dir := t.TempDir()
filePath := filepath.Join(dir, "fixture.mp4")
const original = "original bytes"
if err := os.WriteFile(filePath, []byte(original), 0o600); err != nil {
t.Fatal(err)
}
originalTime := time.Now().Add(-10 * time.Second).Truncate(time.Second)
if err := os.Chtimes(filePath, originalTime, originalTime); err != nil {
t.Fatal(err)
}
first := httptest.NewRecorder()
if err := ServeDirectPlay(first, httptest.NewRequest(http.MethodGet, "/stream", nil), filePath); err != nil {
t.Fatal(err)
}
oldETag := first.Header().Get("ETag")
if oldETag == "" {
t.Fatalf("ETag omitted on supported platform %s", runtime.GOOS)
}
const replacement = "replaced bytes"
if len(replacement) != len(original) {
t.Fatal("test fixture must preserve file size")
}
if err := os.WriteFile(filePath, []byte(replacement), 0o600); err != nil {
t.Fatal(err)
}
if err := os.Chtimes(filePath, originalTime, originalTime); err != nil {
t.Fatal(err)
}
req := httptest.NewRequest(http.MethodGet, "/stream", nil)
req.Header.Set("Range", "bytes=5-")
req.Header.Set("If-Range", oldETag)
rr := httptest.NewRecorder()
if err := ServeDirectPlay(rr, req, filePath); err != nil {
t.Fatal(err)
}
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rr.Code)
}
if body, err := io.ReadAll(rr.Result().Body); err != nil || string(body) != replacement {
t.Fatalf("body = %q, err = %v; want full replacement entity", body, err)
}
if newETag := rr.Header().Get("ETag"); newETag == oldETag {
t.Fatalf("ETag did not change after replacement: %q", newETag)
}
}
func TestDirectPlayEntityTagOmitsUnsupportedRevision(t *testing.T) {
filePath := filepath.Join(t.TempDir(), "fixture.mp4")
if err := os.WriteFile(filePath, []byte("abcdef"), 0o600); err != nil {
t.Fatal(err)
}
file, err := os.Open(filePath)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
if err := file.Close(); err != nil {
t.Errorf("close fixture: %v", err)
}
})
info, err := file.Stat()
if err != nil {
t.Fatal(err)
}
if got := directPlayEntityTag(file, fileInfoWithoutSystem{FileInfo: info}); got != "" {
t.Fatalf("ETag without durable revision = %q, want omitted validator", got)
}
}
type fileInfoWithoutSystem struct {
os.FileInfo
}
func (fileInfoWithoutSystem) Sys() any {
return nil
}
func platformRequiresDirectPlayValidator() bool {
switch runtime.GOOS {
case directPlayDarwinGOOS, directPlayLinuxGOOS, directPlayWindowsGOOS:
return true
default:
return false
}
}
func TestServeDirectPlayStalledWriteIncrementsOutcomeMetric(t *testing.T) {
filePath := filepath.Join(t.TempDir(), "fixture.mp4")
if err := os.WriteFile(filePath, []byte("media bytes"), 0o600); err != nil {
t.Fatal(err)
}
stalledEnds := directStreamEnds.WithLabelValues(string(httpstream.OutcomeStalledReap))
endsBefore := counterValue(t, stalledEnds)
activeBefore := gaugeValue(t, directStreamActive)
writer := &deadlineResponseWriter{header: make(http.Header)}
if err := ServeDirectPlay(writer, httptest.NewRequest(http.MethodGet, "/stream", nil), filePath); err != nil {
t.Fatal(err)
}
if writer.status != http.StatusOK {
t.Fatalf("status = %d, want %d", writer.status, http.StatusOK)
}
if got := counterValue(t, stalledEnds); got != endsBefore+1 {
t.Fatalf("stalled end counter = %v, want %v", got, endsBefore+1)
}
if got := gaugeValue(t, directStreamActive); got != activeBefore {
t.Fatalf("active stream gauge = %v, want restored value %v", got, activeBefore)
}
}
func counterValue(t testing.TB, counter prometheus.Counter) float64 {
t.Helper()
metric := &dto.Metric{}
if err := counter.Write(metric); err != nil {
t.Fatalf("read counter: %v", err)
}
return metric.GetCounter().GetValue()
}
func gaugeValue(t testing.TB, gauge prometheus.Gauge) float64 {
t.Helper()
metric := &dto.Metric{}
if err := gauge.Write(metric); err != nil {
t.Fatalf("read gauge: %v", err)
}
return metric.GetGauge().GetValue()
}
type deadlineResponseWriter struct {
header http.Header
status int
}
func (w *deadlineResponseWriter) Header() http.Header {
return w.header
}
func (w *deadlineResponseWriter) WriteHeader(status int) {
w.status = status
}
func (w *deadlineResponseWriter) Write([]byte) (int, error) {
return 0, os.ErrDeadlineExceeded
}
+18 -1
View File
@@ -20,6 +20,7 @@ const (
FeatureRouteDiagnostics = "playback_route_diagnostics"
FeatureDeviceQuirksV3 = "device_quirks_v1"
FeatureSeekReanchorV3 = "seek_reanchor_v1"
FeatureDirectStreamResumeV3 = "direct_stream_resume_v1"
PlanRecipeVersionV3 = "v3.2"
ClientDV7ToDV81V3 = "client_dv7_to_dv81"
ClientDV7ToHDR10V3 = "client_dv7_to_hdr10"
@@ -30,6 +31,22 @@ const (
DeviceQuirkRegistryRevisionV3 = "2026-07-13.1"
)
// ServerFeaturesV3 returns the complete feature set advertised by protocol-v3
// capability and decision responses. A fresh slice prevents callers from
// mutating the shared contract.
func ServerFeaturesV3() []string {
return []string{
FeaturePlaybackPlanV3,
FeatureMedia3Only,
FeatureDetailedDecodeV3,
FeatureLayoutPassthrough,
FeatureRouteDiagnostics,
FeatureDeviceQuirksV3,
FeatureSeekReanchorV3,
FeatureDirectStreamResumeV3,
}
}
type DecisionOutcomeV3 string
const (
@@ -812,7 +829,7 @@ func HasFeatureV3(features []string, wanted string) bool {
func NewTerminalResponseV3(reason, message string, retryable bool) DecisionResponseV3 {
return DecisionResponseV3{
ProtocolVersion: ProtocolV3,
ServerFeatures: []string{FeaturePlaybackPlanV3, FeatureMedia3Only, FeatureDeviceQuirksV3, FeatureSeekReanchorV3},
ServerFeatures: ServerFeaturesV3(),
Outcome: OutcomeAdaptationUnavailableV3,
Terminal: &TerminalV3{Reason: reason, Message: message, Retryable: retryable},
}
+38
View File
@@ -18,6 +18,44 @@ func hasDegradationWarningV3(warnings []DegradationWarningV3, code string) bool
return false
}
func TestServerFeaturesV3ReturnsCompleteIndependentSlices(t *testing.T) {
first := ServerFeaturesV3()
second := ServerFeaturesV3()
expected := map[string]struct{}{
FeaturePlaybackPlanV3: {},
FeatureMedia3Only: {},
FeatureDetailedDecodeV3: {},
FeatureLayoutPassthrough: {},
FeatureRouteDiagnostics: {},
FeatureDeviceQuirksV3: {},
FeatureSeekReanchorV3: {},
FeatureDirectStreamResumeV3: {},
}
if len(first) != len(expected) {
t.Fatalf("server features = %v, want %d entries", first, len(expected))
}
seen := make(map[string]struct{}, len(first))
for _, feature := range first {
if _, ok := expected[feature]; !ok {
t.Fatalf("server features contain unexpected %q: %v", feature, first)
}
if _, duplicate := seen[feature]; duplicate {
t.Fatalf("server features contain duplicate %q: %v", feature, first)
}
seen[feature] = struct{}{}
}
for feature := range expected {
if _, ok := seen[feature]; !ok {
t.Fatalf("server features omitted %q: %v", feature, first)
}
}
first[0] = "mutated"
if second[0] != FeaturePlaybackPlanV3 {
t.Fatalf("feature slices share backing storage: %v", second)
}
}
func TestStartRequestV3Validation(t *testing.T) {
index := 1
req := validStartRequestV3()
@@ -1,6 +1,7 @@
package playback
import (
"fmt"
"testing"
"time"
)
@@ -45,3 +46,61 @@ func TestPausedSessionSurvivesIntentionalPause(t *testing.T) {
t.Fatal("session survived past the paused grace; abandoned sessions must still be reaped")
}
}
func TestSequentialRangedTransportsSurviveIdleAndPausedGrace(t *testing.T) {
m := NewSessionManager(0, 0)
session, err := m.StartSession(1, "profile-1", 100, PlayDirect, false)
if err != nil {
t.Fatalf("StartSession: %v", err)
}
setLastActivity := func(age time.Duration) {
t.Helper()
m.mu.Lock()
s := m.sessions[session.ID]
s.LastActivityAt = time.Now().Add(-age)
s.UpdatedAt = s.LastActivityAt
m.mu.Unlock()
}
assertPresent := func(stage string) {
t.Helper()
if _, err := m.GetSession(session.ID); err != nil {
t.Fatalf("%s: session was cleaned: %v", stage, err)
}
}
const activeGrace = 2 * time.Minute
for cycle := 1; cycle <= 3; cycle++ {
setLastActivity(activeGrace / 2)
m.CleanInactive(activeGrace, DefaultPausedSessionGrace)
assertPresent(fmt.Sprintf("idle gap before transport %d", cycle))
if err := m.BeginTransport(session.ID); err != nil {
t.Fatalf("BeginTransport(%d): %v", cycle, err)
}
setLastActivity(activeGrace + time.Minute)
m.CleanInactive(activeGrace, DefaultPausedSessionGrace)
assertPresent(fmt.Sprintf("active transport %d", cycle))
if err := m.EndTransport(session.ID); err != nil {
t.Fatalf("EndTransport(%d): %v", cycle, err)
}
}
if err := m.UpdateProgress(session.ID, 42, true); err != nil {
t.Fatalf("UpdateProgress(paused): %v", err)
}
setLastActivity(DefaultPausedSessionGrace - time.Minute)
m.CleanInactive(activeGrace, DefaultPausedSessionGrace)
assertPresent("late paused idle gap")
if err := m.BeginTransport(session.ID); err != nil {
t.Fatalf("BeginTransport(late range): %v", err)
}
setLastActivity(DefaultPausedSessionGrace + time.Minute)
m.CleanInactive(activeGrace, DefaultPausedSessionGrace)
assertPresent("late ranged transport active")
if err := m.EndTransport(session.ID); err != nil {
t.Fatalf("EndTransport(late range): %v", err)
}
assertPresent("completed ranged transport sequence")
}