diff --git a/internal/api/handlers/nodes.go b/internal/api/handlers/nodes.go index c306cc8d..016a0300 100644 --- a/internal/api/handlers/nodes.go +++ b/internal/api/handlers/nodes.go @@ -34,6 +34,10 @@ func LiveLocalSessions(sm *playback.SessionManager, nodeName string) []nodesessi live := sm.AllSessions() out := make([]nodesessions.SessionInfo, 0, len(live)) for _, s := range live { + lastServedAt := s.LastServedAt + if lastServedAt.IsZero() { + lastServedAt = s.LastActivityAt + } out = append(out, nodesessions.SessionInfo{ SessionID: s.ID, NodeName: nodeName, @@ -48,7 +52,8 @@ func LiveLocalSessions(sm *playback.SessionManager, nodeName string) []nodesessi Resolution: s.TargetResolution, HWAccel: s.TranscodeHWAccel, StartedAt: s.StartedAt.UTC().Format(time.RFC3339), - LastServedAt: s.LastActivityAt.UTC().Format(time.RFC3339), + LastServedAt: lastServedAt.UTC().Format(time.RFC3339), + BytesServed: s.BytesServed, }) } return out diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index 75dc2503..232a675a 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -3784,7 +3784,10 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht // the hidden-stream window is observable. slog.Warn("begin transport marker failed", "session", sessionID, "segment", segmentPath, "error", err) } - http.ServeFile(w, r, segmentPath) + recorder, _ := h.sessionMgr.(playback.ServedBytesRecorder) + metered := playback.NewSessionMeteredWriter(w, recorder, sessionID) + defer func() { _ = metered.Close() }() + http.ServeFile(metered, r, segmentPath) } // buildProxyManifestURL signs a stream token carrying the session's full diff --git a/internal/api/handlers/stream.go b/internal/api/handlers/stream.go index ef198fc3..d6541060 100644 --- a/internal/api/handlers/stream.go +++ b/internal/api/handlers/stream.go @@ -143,7 +143,10 @@ func (h *StreamHandler) HandleStream(w http.ResponseWriter, r *http.Request) { _ = h.sessionMgr.EndTransport(sessionID) }() } - if err := playback.ServeDirectPlay(w, r, file.FilePath); err != nil { + recorder, _ := h.sessionMgr.(playback.ServedBytesRecorder) + metered := playback.NewSessionMeteredWriter(w, recorder, sessionID) + defer func() { _ = metered.Close() }() + if err := playback.ServeDirectPlay(metered, r, file.FilePath); err != nil { h.handleTransportStartFailure(r.Context(), session, file, err) } @@ -153,13 +156,16 @@ func (h *StreamHandler) HandleStream(w http.ResponseWriter, r *http.Request) { _ = h.sessionMgr.EndTransport(sessionID) }() } + recorder, _ := h.sessionMgr.(playback.ServedBytesRecorder) + metered := playback.NewSessionMeteredWriter(w, recorder, sessionID) + defer func() { _ = metered.Close() }() seekSeconds := 0.0 if seekStr := r.URL.Query().Get("seek"); seekStr != "" { if s, err := strconv.ParseFloat(seekStr, 64); err == nil && s >= 0 { seekSeconds = s } } - if err := playback.ServeRemuxWithDVMode(w, r, file.FilePath, "mp4", seekSeconds, session.TranscodeAudio, session.AudioTrackIndex, file.PrimaryDVProfile(), session.RemuxDVMode, h.ffmpegPath()); err != nil { + if err := playback.ServeRemuxWithDVMode(metered, r, file.FilePath, "mp4", seekSeconds, session.TranscodeAudio, session.AudioTrackIndex, file.PrimaryDVProfile(), session.RemuxDVMode, h.ffmpegPath()); err != nil { h.handleTransportStartFailure(r.Context(), session, file, err) } diff --git a/internal/jellycompat/streams.go b/internal/jellycompat/streams.go index aca0d69c..55b726b3 100644 --- a/internal/jellycompat/streams.go +++ b/internal/jellycompat/streams.go @@ -157,6 +157,9 @@ func (h *PlaybackHandler) HandleVideoStream(w http.ResponseWriter, r *http.Reque stop := h.Revocation.WatchAndCut(w, playSession.UpstreamSessionID, session.StreamAppUserID, time.Now()) defer stop() } + recorder, _ := h.sessionMgr.(playback.ServedBytesRecorder) + metered := playback.NewSessionMeteredWriter(w, recorder, playSession.UpstreamSessionID) + defer func() { _ = metered.Close() }() switch method { case "remux": @@ -164,9 +167,9 @@ func (h *PlaybackHandler) HandleVideoStream(w http.ResponseWriter, r *http.Reque if resolvedAudioTrackIndex, ok := compatAudioTrackIndex(*source); ok { audioTrackIndex = resolvedAudioTrackIndex } - _ = playback.ServeRemux(w, r, file.FilePath, "mp4", seekSeconds, source.TranscodeAudio, audioTrackIndex, file.PrimaryDVProfile()) + _ = playback.ServeRemux(metered, r, file.FilePath, "mp4", seekSeconds, source.TranscodeAudio, audioTrackIndex, file.PrimaryDVProfile()) default: - _ = playback.ServeDirectPlay(w, r, file.FilePath) + _ = playback.ServeDirectPlay(metered, r, file.FilePath) } } @@ -617,7 +620,10 @@ func (h *PlaybackHandler) HandleHLSSegment(w http.ResponseWriter, r *http.Reques } } - http.ServeFile(w, r, segmentPath) + recorder, _ := h.sessionMgr.(playback.ServedBytesRecorder) + metered := playback.NewSessionMeteredWriter(w, recorder, playSession.UpstreamSessionID) + defer func() { _ = metered.Close() }() + http.ServeFile(metered, r, segmentPath) } // hlsSegmentErrorResponse maps a segment-retrieval error to a Jellyfin-faithful diff --git a/internal/nodesessions/tracker.go b/internal/nodesessions/tracker.go index a757e97c..8619ca31 100644 --- a/internal/nodesessions/tracker.go +++ b/internal/nodesessions/tracker.go @@ -19,8 +19,10 @@ const ( // namespace this tracker writes instead of duplicating the literal. KeyPrefix = "silo:sessions:" - sessionTTL = 60 * time.Second - refreshInt = 30 * time.Second + sessionTTL = 60 * time.Second + refreshInt = 30 * time.Second + transcodeIdleTTL = 180 * time.Second + sessionTypeTranscode = "transcode" ) // SessionInfo represents an active streaming session stored in Redis. @@ -130,7 +132,7 @@ func (tr *Tracker) ActiveCount() int { if _, dup := tr.sessions[id]; dup { continue } - if now.Sub(last) <= sessionTTL { + if now.Sub(last) <= idleTTLFor(tr.records[id]) { count++ } } @@ -150,7 +152,7 @@ func (tr *Tracker) Snapshot() []SessionInfo { // Session-backed entries are always live until Remove; only ephemeral // (non-session) entries age out by idle timeout. if _, isSession := tr.sessions[id]; !isSession { - if last, ok := tr.touched[id]; ok && now.Sub(last) > sessionTTL { + if last, ok := tr.touched[id]; ok && now.Sub(last) > idleTTLFor(rec) { continue } } @@ -373,7 +375,7 @@ func (tr *Tracker) refreshAll(ctx context.Context) { } for id, last := range tr.touched { _, isSession := tr.sessions[id] - if !isSession && now.Sub(last) > sessionTTL { + if !isSession && now.Sub(last) > idleTTLFor(tr.records[id]) { // Idle ephemeral session: stop refreshing and let the Redis key // expire on its own. Session-backed entries (direct/remux) are pruned // on Remove, never by idle timeout, so a quiet-but-open pour stays live. @@ -414,3 +416,10 @@ func (tr *Tracker) refreshAll(ctx context.Context) { slog.DebugContext(ctx, "session refresh pipeline failed", "component", "nodesessions", "error", err) } } + +func idleTTLFor(rec SessionInfo) time.Duration { + if rec.Type == sessionTypeTranscode { + return transcodeIdleTTL + } + return sessionTTL +} diff --git a/internal/nodesessions/tracker_idle_test.go b/internal/nodesessions/tracker_idle_test.go new file mode 100644 index 00000000..15c0695a --- /dev/null +++ b/internal/nodesessions/tracker_idle_test.go @@ -0,0 +1,37 @@ +package nodesessions + +import ( + "context" + "testing" + "time" + + "github.com/redis/go-redis/v9" +) + +func TestTranscodeIdleTTLAgreesAcrossCountSnapshotAndRefresh(t *testing.T) { + rdb := redis.NewClient(&redis.Options{Addr: "127.0.0.1:1", DialTimeout: 10 * time.Millisecond}) + t.Cleanup(func() { _ = rdb.Close() }) + tr := NewTracker(rdb, "http://node", "node", "proxy") + idle := time.Now().Add(-90 * time.Second) + tr.touched[sessionTypeTranscode] = idle + tr.records[sessionTypeTranscode] = SessionInfo{SessionID: sessionTypeTranscode, Type: sessionTypeTranscode} + tr.touched["direct"] = idle + tr.records["direct"] = SessionInfo{SessionID: "direct", Type: "direct_play"} + + if got := tr.ActiveCount(); got != 1 { + t.Fatalf("ActiveCount=%d, want only transcode", got) + } + snapshot := tr.Snapshot() + if len(snapshot) != 1 || snapshot[0].SessionID != sessionTypeTranscode { + t.Fatalf("Snapshot=%+v, want only transcode", snapshot) + } + + tr.refreshAll(context.Background()) + tr.mu.Lock() + _, transcodeRetained := tr.records[sessionTypeTranscode] + _, directRetained := tr.records["direct"] + tr.mu.Unlock() + if !transcodeRetained || directRetained { + t.Fatalf("refresh records: transcode=%v direct=%v, want true/false", transcodeRetained, directRetained) + } +} diff --git a/internal/playback/metered_writer.go b/internal/playback/metered_writer.go new file mode 100644 index 00000000..b5c65ebc --- /dev/null +++ b/internal/playback/metered_writer.go @@ -0,0 +1,80 @@ +package playback + +import ( + "io" + "net/http" +) + +const meteredWriterFlushBytes = 1 << 20 + +// ServedBytesRecorder receives server-observed byte counts. +type ServedBytesRecorder interface { + AddServedBytes(sessionID string, n int64) error +} + +// SessionMeteredWriter preserves optional ResponseWriter capabilities while +// attributing bytes to a playback session in coarse chunks. +type SessionMeteredWriter struct { + http.ResponseWriter + recorder ServedBytesRecorder + sessionID string + pending int64 +} + +func NewSessionMeteredWriter(w http.ResponseWriter, recorder ServedBytesRecorder, sessionID string) *SessionMeteredWriter { + return &SessionMeteredWriter{ResponseWriter: w, recorder: recorder, sessionID: sessionID} +} + +func (w *SessionMeteredWriter) Write(p []byte) (int, error) { + n, err := w.ResponseWriter.Write(p) + w.account(int64(n)) + return n, err +} + +func (w *SessionMeteredWriter) ReadFrom(src io.Reader) (int64, error) { + if rf, ok := w.ResponseWriter.(io.ReaderFrom); ok { + n, err := rf.ReadFrom(src) + w.account(n) + return n, err + } + return io.Copy(meteredWriteOnly{w}, src) +} + +type meteredWriteOnly struct{ io.Writer } + +func (w *SessionMeteredWriter) account(n int64) { + if n <= 0 { + return + } + w.pending += n + if w.pending >= meteredWriterFlushBytes { + w.flush() + } +} + +func (w *SessionMeteredWriter) flush() { + if w.pending <= 0 { + return + } + if w.recorder != nil { + _ = w.recorder.AddServedBytes(w.sessionID, w.pending) + } + w.pending = 0 +} + +// Close flushes the final partial chunk. Callers must defer it. +func (w *SessionMeteredWriter) Close() error { + w.flush() + return nil +} + +func (w *SessionMeteredWriter) Flush() { + if f, ok := w.ResponseWriter.(http.Flusher); ok { + f.Flush() + } +} + +// Unwrap lets http.ResponseController reach the underlying connection. +func (w *SessionMeteredWriter) Unwrap() http.ResponseWriter { + return w.ResponseWriter +} diff --git a/internal/playback/metered_writer_test.go b/internal/playback/metered_writer_test.go new file mode 100644 index 00000000..651486cb --- /dev/null +++ b/internal/playback/metered_writer_test.go @@ -0,0 +1,96 @@ +package playback + +import ( + "bytes" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/httpstream" +) + +type byteRecorder struct{ total int64 } + +func (r *byteRecorder) AddServedBytes(_ string, n int64) error { + r.total += n + return nil +} + +type readerFromResponseWriter struct { + header http.Header + body bytes.Buffer + readFromCalls int + deadlineSet bool +} + +func (w *readerFromResponseWriter) Header() http.Header { return w.header } +func (w *readerFromResponseWriter) WriteHeader(int) {} +func (w *readerFromResponseWriter) Write(p []byte) (int, error) { return w.body.Write(p) } +func (w *readerFromResponseWriter) SetWriteDeadline(time.Time) error { + w.deadlineSet = true + return nil +} +func (w *readerFromResponseWriter) ReadFrom(src io.Reader) (int64, error) { + w.readFromCalls++ + return w.body.ReadFrom(src) +} + +type readOnly struct{ io.Reader } + +func TestSessionMeteredWriterPreservesReadFromAndUnwrapChain(t *testing.T) { + base := &readerFromResponseWriter{header: make(http.Header)} + production := httpstream.NewRollingDeadlineWriter(base) + recorder := &byteRecorder{} + metered := NewSessionMeteredWriter(production, recorder, "s1") + + payload := bytes.Repeat([]byte("x"), 2<<20) + n, err := metered.ReadFrom(readOnly{bytes.NewReader(payload)}) + if err != nil { + t.Fatalf("ReadFrom: %v", err) + } + if n != int64(len(payload)) || base.readFromCalls == 0 { + t.Fatalf("ReadFrom n=%d calls=%d, want %d and fast-path calls", n, base.readFromCalls, len(payload)) + } + if err := metered.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if recorder.total != int64(len(payload)) { + t.Fatalf("recorded bytes=%d, want %d", recorder.total, len(payload)) + } + if metered.Unwrap() != production { + t.Fatal("Unwrap did not expose the next production writer") + } + if err := http.NewResponseController(metered).SetWriteDeadline(time.Now()); err != nil { + t.Fatalf("SetWriteDeadline through chain: %v", err) + } + if !base.deadlineSet { + t.Fatal("response controller did not reach base writer") + } +} + +func TestSessionMeteredWriterFallbackDoesNotRecurseAndFlushesTail(t *testing.T) { + base := httptest.NewRecorder() + recorder := &byteRecorder{} + metered := NewSessionMeteredWriter(base, recorder, "s1") + payload := []byte("final partial chunk") + + n, err := metered.ReadFrom(readOnly{bytes.NewReader(payload)}) + if err != nil { + t.Fatalf("ReadFrom fallback: %v", err) + } + if n != int64(len(payload)) { + t.Fatalf("ReadFrom n=%d, want %d", n, len(payload)) + } + if recorder.total != 0 { + t.Fatalf("tail flushed before Close: %d", recorder.total) + } + defer func() { _ = metered.Close() }() + if err := metered.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if recorder.total != int64(len(payload)) { + t.Fatalf("tail bytes=%d, want %d", recorder.total, len(payload)) + } +} diff --git a/internal/playback/session.go b/internal/playback/session.go index d52a1b1f..56002ca1 100644 --- a/internal/playback/session.go +++ b/internal/playback/session.go @@ -56,6 +56,8 @@ type Session struct { StartedAt time.Time UpdatedAt time.Time LastActivityAt time.Time + BytesServed int64 + LastServedAt time.Time activeTransportCount int replacementPlayMethod PlayMethod streamRevision uint64 @@ -179,6 +181,7 @@ type SessionManager struct { admissionDecider AdmissionDecider activeGrace time.Duration pausedGrace time.Duration + transcodeGrace time.Duration expireHook func(*Session) } @@ -241,6 +244,10 @@ const ( // pressing Play after a long pause freeze the client (issue #243). // Keep in sync with pausedSessionGrace in internal/worker/cleanup.go. DefaultPausedSessionGrace = 30 * time.Minute + + // DefaultTranscodeSessionGrace covers buffer-ahead gaps between segment + // requests without depending on local ffmpeg process state. + DefaultTranscodeSessionGrace = 10 * time.Minute ) // NewSessionManager creates a SessionManager with the given concurrency limits. @@ -248,11 +255,12 @@ const ( // maxTranscodes limits concurrent transcode streams per user. func NewSessionManager(maxStreams, maxTranscodes int) *SessionManager { return &SessionManager{ - sessions: make(map[string]*Session), - maxStreams: maxStreams, - maxTranscodes: maxTranscodes, - activeGrace: DefaultActiveSessionGrace, - pausedGrace: DefaultPausedSessionGrace, + sessions: make(map[string]*Session), + maxStreams: maxStreams, + maxTranscodes: maxTranscodes, + activeGrace: DefaultActiveSessionGrace, + pausedGrace: DefaultPausedSessionGrace, + transcodeGrace: DefaultTranscodeSessionGrace, } } @@ -286,6 +294,15 @@ func (m *SessionManager) SetLivenessGracePeriods(active, paused time.Duration) { } } +// SetTranscodeLivenessGrace overrides the unpaused transcode idle window. +func (m *SessionManager) SetTranscodeLivenessGrace(grace time.Duration) { + m.mu.Lock() + defer m.mu.Unlock() + if grace > 0 { + m.transcodeGrace = grace + } +} + // SetExpirationHook registers a callback that runs after a session is removed // by stale cleanup. The hook executes outside the manager lock. func (m *SessionManager) SetExpirationHook(fn func(*Session)) { @@ -1112,6 +1129,7 @@ func (m *SessionManager) BeginTransport(sessionID string) error { } s.activeTransportCount++ + s.LastServedAt = time.Now() m.touchSessionLocked(s) return nil } @@ -1130,6 +1148,25 @@ func (m *SessionManager) EndTransport(sessionID string) error { if s.activeTransportCount > 0 { s.activeTransportCount-- } + s.LastServedAt = time.Now() + m.touchSessionLocked(s) + return nil +} + +// AddServedBytes records bytes observed by a server-side media pour. It never +// recreates an unknown or already-reaped session. +func (m *SessionManager) AddServedBytes(sessionID string, n int64) error { + m.mu.Lock() + defer m.mu.Unlock() + s, ok := m.sessions[sessionID] + if !ok { + return ErrSessionNotFound + } + if n <= 0 { + return nil + } + s.BytesServed += n + s.LastServedAt = time.Now() m.touchSessionLocked(s) return nil } @@ -1348,6 +1385,8 @@ func (m *SessionManager) sessionIsInactiveLocked(s *Session, now time.Time, acti grace := activeIdle if s.IsPaused { grace = pausedIdle + } else if s.PlayMethod == PlayTranscode && m.transcodeGrace > grace { + grace = m.transcodeGrace } if grace <= 0 { return !lastActivity.After(now) diff --git a/internal/playback/session_served_test.go b/internal/playback/session_served_test.go new file mode 100644 index 00000000..c2606d13 --- /dev/null +++ b/internal/playback/session_served_test.go @@ -0,0 +1,80 @@ +package playback + +import ( + "errors" + "testing" + "time" +) + +func TestAddServedBytesUnknownDoesNotCreateSession(t *testing.T) { + m := NewSessionManager(0, 0) + if err := m.AddServedBytes("missing", 100); !errors.Is(err, ErrSessionNotFound) { + t.Fatalf("AddServedBytes error=%v, want ErrSessionNotFound", err) + } + if len(m.AllSessions()) != 0 { + t.Fatal("AddServedBytes recreated an unknown session") + } +} + +func TestLastServedAtOnlyAdvancesForServerObservedEvents(t *testing.T) { + m := NewSessionManager(0, 0) + s, err := m.StartSession(1, "p", 1, PlayDirect, false) + if err != nil { + t.Fatal(err) + } + if err := m.UpdateProgress(s.ID, 10, false); err != nil { + t.Fatal(err) + } + afterProgress, _ := m.GetSession(s.ID) + if !afterProgress.LastServedAt.IsZero() { + t.Fatalf("UpdateProgress advanced LastServedAt: %v", afterProgress.LastServedAt) + } + if err := m.AddServedBytes(s.ID, 7); err != nil { + t.Fatal(err) + } + afterBytes, _ := m.GetSession(s.ID) + if afterBytes.LastServedAt.IsZero() || afterBytes.BytesServed != 7 { + t.Fatalf("AddServedBytes did not update served state: %+v", afterBytes) + } + first := afterBytes.LastServedAt + time.Sleep(time.Millisecond) + if err := m.BeginTransport(s.ID); err != nil { + t.Fatal(err) + } + afterBegin, _ := m.GetSession(s.ID) + if !afterBegin.LastServedAt.After(first) { + t.Fatal("BeginTransport did not advance LastServedAt") + } +} + +func TestTranscodeLivenessGraceAndPausedPrecedence(t *testing.T) { + tests := []struct { + name string + method PlayMethod + paused bool + idle time.Duration + wantReaped bool + }{ + {name: "transcode five minutes retained", method: PlayTranscode, idle: 5 * time.Minute}, + {name: "transcode fifteen minutes reaped", method: PlayTranscode, idle: 15 * time.Minute, wantReaped: true}, + {name: "paused transcode keeps thirty minutes", method: PlayTranscode, paused: true, idle: 15 * time.Minute}, + {name: "direct keeps active grace", method: PlayDirect, idle: time.Minute, wantReaped: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + m := NewSessionManager(0, 0) + s, err := m.StartSession(1, "p", 1, tt.method, false) + if err != nil { + t.Fatal(err) + } + m.mu.Lock() + m.sessions[s.ID].IsPaused = tt.paused + m.sessions[s.ID].LastActivityAt = time.Now().Add(-tt.idle) + m.mu.Unlock() + reaped := m.CleanStale() + if (len(reaped) > 0) != tt.wantReaped { + t.Fatalf("reaped=%d, wantReaped=%v", len(reaped), tt.wantReaped) + } + }) + } +} diff --git a/internal/playback/session_test.go b/internal/playback/session_test.go index 98e3edbf..5f6e880e 100644 --- a/internal/playback/session_test.go +++ b/internal/playback/session_test.go @@ -1092,6 +1092,7 @@ func TestSetRealtimeConnection(t *testing.T) { func TestSessionManager_LimitCountsIgnoreStaleSessions(t *testing.T) { sm := playback.NewSessionManager(5, 2) sm.SetLivenessGracePeriods(20*time.Millisecond, 40*time.Millisecond) + sm.SetTranscodeLivenessGrace(20 * time.Millisecond) if _, err := sm.StartSession(1, "profile-1", 100, playback.PlayDirect, false); err != nil { t.Fatalf("StartSession direct: %v", err) @@ -1110,6 +1111,22 @@ func TestSessionManager_LimitCountsIgnoreStaleSessions(t *testing.T) { } } +func TestSessionManager_LivenessGraceDoesNotOverwriteTranscodeGrace(t *testing.T) { + sm := playback.NewSessionManager(5, 2) + sm.SetTranscodeLivenessGrace(time.Hour) + sm.SetLivenessGracePeriods(20*time.Millisecond, 40*time.Millisecond) + + if _, err := sm.StartSession(1, "profile-1", 101, playback.PlayTranscode, false); err != nil { + t.Fatalf("StartSession transcode: %v", err) + } + + time.Sleep(30 * time.Millisecond) + + if got := sm.TranscodeCount(1); got != 1 { + t.Fatalf("TranscodeCount after active grace = %d, want 1", got) + } +} + func TestSessionManager_ActiveTransportKeepsSessionLive(t *testing.T) { sm := playback.NewSessionManager(5, 2) sm.SetLivenessGracePeriods(20*time.Millisecond, 40*time.Millisecond) @@ -1366,6 +1383,7 @@ func TestSessionManager_CleanExpired_PausedGracePeriod(t *testing.T) { func TestCheckReplacementAllowedExcludesOnlyTheReplacedSession(t *testing.T) { sm := playback.NewSessionManager(10, 2) sm.SetLivenessGracePeriods(25*time.Millisecond, time.Hour) + sm.SetTranscodeLivenessGrace(25 * time.Millisecond) failed, err := sm.StartSession(1, "profile-1", 100, playback.PlayTranscode, false) if err != nil {