diff --git a/docs/architecture/direct-stream-resume.md b/docs/architecture/direct-stream-resume.md new file mode 100644 index 00000000..5bc38cf2 --- /dev/null +++ b/docs/architecture/direct-stream-resume.md @@ -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 */`. +- `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. diff --git a/internal/api/handlers/playback_v3.go b/internal/api/handlers/playback_v3.go index 851c0c45..a5617eba 100644 --- a/internal/api/handlers/playback_v3.go +++ b/internal/api/handlers/playback_v3.go @@ -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 { diff --git a/internal/api/handlers/playback_v3_test.go b/internal/api/handlers/playback_v3_test.go index 25ef7c2a..ed7a54b0 100644 --- a/internal/api/handlers/playback_v3_test.go +++ b/internal/api/handlers/playback_v3_test.go @@ -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) } } diff --git a/internal/api/handlers/stream_test.go b/internal/api/handlers/stream_test.go index 5aefc50c..348d38fc 100644 --- a/internal/api/handlers/stream_test.go +++ b/internal/api/handlers/stream_test.go @@ -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{ diff --git a/internal/httpstream/rolling_deadline.go b/internal/httpstream/rolling_deadline.go index de681385..f406aa4a 100644 --- a/internal/httpstream/rolling_deadline.go +++ b/internal/httpstream/rolling_deadline.go @@ -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 } diff --git a/internal/httpstream/rolling_deadline_test.go b/internal/httpstream/rolling_deadline_test.go index f762d2d7..d64c548d 100644 --- a/internal/httpstream/rolling_deadline_test.go +++ b/internal/httpstream/rolling_deadline_test.go @@ -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 { diff --git a/internal/playback/directplay.go b/internal/playback/directplay.go index 8dd19d3d..8a211f2a 100644 --- a/internal/playback/directplay.go +++ b/internal/playback/directplay.go @@ -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 +} diff --git a/internal/playback/directplay_etag.go b/internal/playback/directplay_etag.go new file mode 100644 index 00000000..6880c5bf --- /dev/null +++ b/internal/playback/directplay_etag.go @@ -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) +} diff --git a/internal/playback/directplay_metrics.go b/internal/playback/directplay_metrics.go new file mode 100644 index 00000000..b94b4505 --- /dev/null +++ b/internal/playback/directplay_metrics.go @@ -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() + } +} diff --git a/internal/playback/directplay_revision_darwin.go b/internal/playback/directplay_revision_darwin.go new file mode 100644 index 00000000..881ded52 --- /dev/null +++ b/internal/playback/directplay_revision_darwin.go @@ -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 +} diff --git a/internal/playback/directplay_revision_fallback.go b/internal/playback/directplay_revision_fallback.go new file mode 100644 index 00000000..6f48c0f8 --- /dev/null +++ b/internal/playback/directplay_revision_fallback.go @@ -0,0 +1,9 @@ +//go:build !darwin && !linux && !windows + +package playback + +import "os" + +func directPlayFilesystemRevision(*os.File, os.FileInfo) (string, bool) { + return "", false +} diff --git a/internal/playback/directplay_revision_linux.go b/internal/playback/directplay_revision_linux.go new file mode 100644 index 00000000..baf2106e --- /dev/null +++ b/internal/playback/directplay_revision_linux.go @@ -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 +} diff --git a/internal/playback/directplay_revision_windows.go b/internal/playback/directplay_revision_windows.go new file mode 100644 index 00000000..e2fd43a6 --- /dev/null +++ b/internal/playback/directplay_revision_windows.go @@ -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 +} diff --git a/internal/playback/directplay_test.go b/internal/playback/directplay_test.go new file mode 100644 index 00000000..8357433b --- /dev/null +++ b/internal/playback/directplay_test.go @@ -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 +} diff --git a/internal/playback/protocol_v3.go b/internal/playback/protocol_v3.go index 20efbce3..b8157451 100644 --- a/internal/playback/protocol_v3.go +++ b/internal/playback/protocol_v3.go @@ -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}, } diff --git a/internal/playback/protocol_v3_test.go b/internal/playback/protocol_v3_test.go index c1dd2946..898a94de 100644 --- a/internal/playback/protocol_v3_test.go +++ b/internal/playback/protocol_v3_test.go @@ -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() diff --git a/internal/playback/session_paused_grace_test.go b/internal/playback/session_paused_grace_test.go index 5e67f222..b8f5231e 100644 --- a/internal/playback/session_paused_grace_test.go +++ b/internal/playback/session_paused_grace_test.go @@ -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") +}