diff --git a/docs/superpowers/plans/2026-10-21-playback-correctness-isolation.md b/docs/superpowers/plans/2026-10-21-playback-correctness-isolation.md new file mode 100644 index 0000000..50650a9 --- /dev/null +++ b/docs/superpowers/plans/2026-10-21-playback-correctness-isolation.md @@ -0,0 +1,46 @@ +# Playback Correctness and Isolation Implementation Plan + +**Goal:** Fix the five playback defects from the 2026-10-21 review and make playback capability queries side-effect-free, without replacing the player or adding dependencies. + +**Scope:** The user approved implementing the reviewed fixes. Work in the current clean checkout; do not commit, push, deploy, or alter unrelated files. + +## Design and invariants + +- Keep existing media IDs, stream routes, direct playback and external-player URLs compatible. +- Add optional `session_id` to local HLS requests. Authenticated handlers supply the actual user identity; a client parameter never supplies the owner. New web playback sessions use random IDs. Legacy clients are isolated by user and, where available, Emby play-session/device identity. +- Keep transcoder media lookup/logging separate from its internal task key. A task key incorporates user, media and playback session. Each actual ffmpeg run writes to a fresh generation directory, so delayed segment requests never read a replacement run's output. +- Stop only the authenticated caller's playback session. Check media visibility and scoped playback tokens on stop, just as on playlist/segment requests. Segment and playlist queries must match the running generation. +- Cloud child requests resolve their real session media first, compare query media, enforce token scope and visibility against the real media, and reject mismatches before upstream fetch. +- Treat player business positions as absolute source seconds. Only local HLS converts to/from a relative ffmpeg timeline. +- Invalidate outstanding cloud-transcode requests on media/unmount, explicit source/quality changes, and newer requests. Guard both fulfilled and rejected responses. +- Select local fallbacks only from available local qualities. Keep the existing 115-first preference when a cloud quality is available. +- `PlaybackInfo` only reads capabilities/status. Explicit `StartTranscode` requests enqueue cloud work. Retain pending/ready response contracts, polling, and retries. + +## Tasks + +1. Add regression tests for local-scope isolation, stale generations, stop authorization, cloud media binding, pure capability reads, timeline conversion, available-local selection and stale async results. Run tests before implementation and verify the intended failures. +2. Add tested playback helpers; route skip/undo/progress/quality/subtitle calculations through the common absolute timeline; guard cloud transcode results and fix local fallback choice. +3. Extend existing transcoder/stream methods with an optional scope, preserve legacy internal calls, isolate task keys/output directories, and pass caller scope from web and Emby handlers. Bind web local HLS URLs and stop requests to captured session IDs. +4. Enforce real-media authorization on cloud child requests and authenticated ownership on local stop. +5. Split read-only cloud capability lookup from explicit cloud enqueue. Update existing cloud tests to use the explicit start operation and test no push during reads. +6. Run focused and full Go tests, frontend tests, TypeScript/Vite build, lint diagnostics and review the diff for security, lifecycle and API compatibility. Fix regressions before reporting completion. + +## Verification commands + +- `node --experimental-strip-types --test src/pages/hlsPlayback.test.ts src/pages/directPlaySeek.test.ts src/pages/playerPageModel.test.ts src/pages/playbackQuality.test.ts src/pages/playbackRequestGuard.test.ts src/utils/skipSegments.test.ts` +- `go test ./internal/service ./internal/handler ./internal/middleware -count=1` +- `go test ./...` +- `npm run build` and targeted ESLint for changed frontend files. + +## Out of scope + +New remux/audio-only ffmpeg pipelines, speculative browser codec policy changes, player replacement, broad UI refactors, performance improvement claims without measurements, and production deployment. + +## Verification outcome + +- Six frontend playback regression scripts pass; targeted ESLint and the TypeScript/Vite production build pass. The existing bundle-size warning remains. +- Playback/stream/transcode/progress tests in service, handler and middleware pass, including scoped output reads, stale generation rejection, scoped stops, device-to-segment propagation, read-only capability queries and concurrent cloud start deduplication. +- Whole-repository tests pass with only `TestQueueLibraryRootScanImportsNewLibraryMedia` and `TestSQLiteRestoreStagesWithoutChangingLiveDatabase` excluded. Both were run independently and reproduce Windows file-lock / SQLite errors in unmodified scan/database code. They are not silently disabled in source. +- FFmpeg is not installed in this environment, so the existing real-transcode lifecycle test skips. Actual video, upstream 115 and browser end-to-end playback remain unverified. +- Clients with no playback-session or device identifier retain per-user legacy isolation; new web requests always carry an independent session ID. +- No commit, push or deployment was performed. diff --git a/internal/handler/cloud115_playback.go b/internal/handler/cloud115_playback.go index 2f61591..e99f7ff 100644 --- a/internal/handler/cloud115_playback.go +++ b/internal/handler/cloud115_playback.go @@ -124,10 +124,32 @@ func cloud115HLSSessionHandler(svc *service.Container) gin.HandlerFunc { c.JSON(http.StatusBadRequest, gin.H{"error": "missing media_id"}) return } - if !enforceScopedPlaybackToken(c, mediaID) { + if c.Request.Context().Err() != nil { + c.AbortWithStatus(499) return } - err := svc.Cloud115.HLSProxy().ServeChild( + actualMediaID, err := svc.Cloud115.HLSProxy().SessionMediaID(c.Param("session"), c.Param("key")) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"code": "hls_session_not_found", "error": "hls session not found"}) + return + } + if mediaID != actualMediaID { + c.JSON(http.StatusForbidden, gin.H{"error": "playback session media mismatch"}) + return + } + if !enforceScopedPlaybackToken(c, actualMediaID) { + return + } + if svc.Repo == nil || svc.Repo.Media == nil { + c.JSON(http.StatusNotFound, gin.H{"error": "media not found"}) + return + } + m, findErr := svc.Repo.Media.FindByID(c.Request.Context(), actualMediaID) + if findErr != nil || m == nil || !mediaVisibleForRequest(c, svc, m) { + c.JSON(http.StatusNotFound, gin.H{"error": "media not found"}) + return + } + err = svc.Cloud115.HLSProxy().ServeChild( c.Request.Context(), c.Writer, c.Request, diff --git a/internal/handler/emby_playback.go b/internal/handler/emby_playback.go index d9782ba..151d6ea 100644 --- a/internal/handler/emby_playback.go +++ b/internal/handler/emby_playback.go @@ -418,7 +418,11 @@ func embyVideoHLSPlaylistHandler(svc *service.Container) gin.HandlerFunc { c.Status(http.StatusNotFound) return } - err = svc.Stream.ServeHLSPlaylist(c.Writer, c.Request, c.Param("id")) + scope, ok := localHLSPlaybackScope(c) + if !ok { + return + } + err = svc.Stream.ServeHLSPlaylist(c.Writer, c.Request, c.Param("id"), scope) if errors.Is(err, service.ErrTranscodeDisabled) { c.JSON(http.StatusConflict, gin.H{"error": "transcode disabled"}) return @@ -444,7 +448,11 @@ func embyVideoHLSSegmentHandler(svc *service.Container) gin.HandlerFunc { c.Status(http.StatusNotFound) return } - if err := svc.Stream.ServeHLSSegment(c.Writer, c.Request, c.Param("id"), c.Param("seg")); err != nil { + scope, ok := localHLSPlaybackScope(c) + if !ok { + return + } + if err := svc.Stream.ServeHLSSegment(c.Writer, c.Request, c.Param("id"), c.Param("seg"), scope); err != nil { c.Status(http.StatusNotFound) } } diff --git a/internal/handler/hls_session_test.go b/internal/handler/hls_session_test.go new file mode 100644 index 0000000..9619051 --- /dev/null +++ b/internal/handler/hls_session_test.go @@ -0,0 +1,37 @@ +package handler + +import ( + "net/http" + "net/http/httptest" + "testing" + + "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/middleware" + "github.com/truewhile/MeBox/internal/model" + "github.com/truewhile/MeBox/internal/service" +) + +func TestStopHLSChecksMediaVisibility(t *testing.T) { + router, svc, secret := newPlaybackScopeTestRouter(t) + svc.Transcoder = service.NewTranscoderService(svc.Cfg, zap.NewNop(), svc.Repo, nil) + router.DELETE("/api/hls/:id", middleware.AuthRequired(secret), stopTranscodeHandler(svc)) + req := httptest.NewRequest(http.MethodDelete, "/api/hls/nonexistent?session_id=player-a", nil) + req.Header.Set("Authorization", "Bearer "+signedTestToken(t, secret)) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusNotFound { + t.Fatalf("stop of a nonexistent/invisible media: status=%d body=%s, want 404", rec.Code, rec.Body.String()) + } + profile := &model.PlayProfile{UserID: "user-1", Name: "restricted", IsDefault: true, AllowedLibraryIDs: `["other-library"]`} + if err := svc.Repo.DB.Create(profile).Error; err != nil { + t.Fatal(err) + } + req = httptest.NewRequest(http.MethodDelete, "/api/hls/media-1?session_id=player-a", nil) + req.Header.Set("Authorization", "Bearer "+signedTestToken(t, secret)) + rec = httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusNotFound { + t.Fatalf("invisible existing media can be stopped: %d %s", rec.Code, rec.Body.String()) + } +} diff --git a/internal/handler/streaming.go b/internal/handler/streaming.go index 468f86f..a3dc164 100644 --- a/internal/handler/streaming.go +++ b/internal/handler/streaming.go @@ -27,7 +27,15 @@ func hlsPlaylistHandler(svc *service.Container) gin.HandlerFunc { if !enforceScopedPlaybackToken(c, m.ID) { return } - err = svc.Stream.ServeHLSPlaylist(c.Writer, c.Request, c.Param("id")) + scope, ok := localHLSPlaybackScope(c) + if !ok { + return + } + err = svc.Stream.ServeHLSPlaylist(c.Writer, c.Request, c.Param("id"), scope) + if errors.Is(err, service.ErrHLSGenerationSuperseded) { + c.JSON(http.StatusConflict, gin.H{"code": "hls_generation_superseded", "error": err.Error()}) + return + } if errors.Is(err, service.ErrMediaNotFound) { c.JSON(http.StatusNotFound, gin.H{"error": "not found"}) return @@ -62,7 +70,11 @@ func hlsSegmentHandler(svc *service.Container) gin.HandlerFunc { if !enforceScopedPlaybackToken(c, m.ID) { return } - err = svc.Stream.ServeHLSSegment(c.Writer, c.Request, c.Param("id"), c.Param("seg")) + scope, ok := localHLSPlaybackScope(c) + if !ok { + return + } + err = svc.Stream.ServeHLSSegment(c.Writer, c.Request, c.Param("id"), c.Param("seg"), scope) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) return @@ -72,11 +84,39 @@ func hlsSegmentHandler(svc *service.Container) gin.HandlerFunc { func stopTranscodeHandler(svc *service.Container) gin.HandlerFunc { return func(c *gin.Context) { - svc.Transcoder.StopJob(c.Param("id")) + m, err := findMediaForPlaybackEndpoint(c, svc, c.Param("id")) + if err != nil || m == nil || !mediaVisibleForRequest(c, svc, m) { + c.JSON(http.StatusNotFound, gin.H{"error": "media not found"}) + return + } + if !enforceScopedPlaybackToken(c, m.ID) { + return + } + scope, ok := localHLSPlaybackScope(c) + if !ok { + return + } + if svc.Transcoder != nil { + svc.Transcoder.StopPlayback(m.ID, scope) + } c.Status(http.StatusNoContent) } } +func localHLSPlaybackScope(c *gin.Context) (service.HLSPlaybackScope, bool) { + userID := currentUserID(c) + if userID == "" { + c.JSON(http.StatusUnauthorized, gin.H{"error": "missing playback owner"}) + return service.HLSPlaybackScope{}, false + } + scope, err := service.HLSPlaybackScopeForRequest(c.Request, userID) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return service.HLSPlaybackScope{}, false + } + return scope, true +} + func imageProxyHandler(svc *service.Container) gin.HandlerFunc { return func(c *gin.Context) { raw := c.Query("url") diff --git a/internal/service/cloud115_hls_authorization_test.go b/internal/service/cloud115_hls_authorization_test.go new file mode 100644 index 0000000..9d6403b --- /dev/null +++ b/internal/service/cloud115_hls_authorization_test.go @@ -0,0 +1,33 @@ +package service + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestCloud115HLSChildRejectsMismatchedSessionMediaBeforeFetch(t *testing.T) { + fetched := false + proxy := &Cloud115HLSProxy{sessions: map[string]*cloud115HLSSession{}} + proxy.client = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + fetched = true + return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"video/mp2t"}}, Body: io.NopCloser(strings.NewReader("private-segment")), Request: req}, nil + })} + proxy.sessions["session-b"] = &cloud115HLSSession{ + ID: "session-b", MediaID: "media-b", ExpiresAt: time.Now().Add(time.Hour), + entries: map[string]string{"segment": "https://cpats01.115.com/seg.ts"}, + } + for _, query := range []string{"?media_id=media-a", ""} { + fetched = false + req := httptest.NewRequest(http.MethodGet, "/api/cloud115/hls/session-b/segment"+query, nil) + rec := httptest.NewRecorder() + err := proxy.ServeChild(context.Background(), rec, req, "session-b", "segment") + if err == nil || fetched { + t.Errorf("query %q must reject the wrong/missing media before fetch; err=%v fetched=%v", query, err, fetched) + } + } +} diff --git a/internal/service/cloud115_hls_proxy.go b/internal/service/cloud115_hls_proxy.go index 02dab2a..c65f45a 100644 --- a/internal/service/cloud115_hls_proxy.go +++ b/internal/service/cloud115_hls_proxy.go @@ -20,8 +20,9 @@ import ( ) var ( - ErrCloud115HLSSessionNotFound = errors.New("115 hls session not found") - ErrCloud115HLSUpstreamExpired = errors.New("115 hls upstream expired") + ErrCloud115HLSSessionNotFound = errors.New("115 hls session not found") + ErrCloud115HLSUpstreamExpired = errors.New("115 hls upstream expired") + ErrCloud115HLSSessionMediaMismatch = errors.New("115 hls session media mismatch") ) const ( @@ -131,6 +132,9 @@ func (p *Cloud115HLSProxy) ServeChild(ctx context.Context, w http.ResponseWriter if !ok { return ErrCloud115HLSSessionNotFound } + if r == nil || strings.TrimSpace(r.URL.Query().Get("media_id")) != session.MediaID { + return ErrCloud115HLSSessionMediaMismatch + } resp, err := p.fetchUpstream(ctx, upstream, r.Header.Get("Range")) if err != nil { return err @@ -186,6 +190,15 @@ func (p *Cloud115HLSProxy) ServeChild(ctx context.Context, w http.ResponseWriter return err } +// SessionMediaID resolves the server-owned media identity before handler authorization. +func (p *Cloud115HLSProxy) SessionMediaID(sessionID, key string) (string, error) { + session, _, ok := p.lookup(sessionID, key) + if !ok { + return "", ErrCloud115HLSSessionNotFound + } + return session.MediaID, nil +} + func (p *Cloud115HLSProxy) newSession(mediaID string, definition int) *cloud115HLSSession { now := time.Now() return &cloud115HLSSession{ diff --git a/internal/service/cloud115_playback.go b/internal/service/cloud115_playback.go index af5834e..e7a3f3e 100644 --- a/internal/service/cloud115_playback.go +++ b/internal/service/cloud115_playback.go @@ -104,11 +104,13 @@ func (s *Cloud115PlaybackService) HLSProxy() *Cloud115HLSProxy { return s.proxy } -// PlaybackInfo 返回媒体在 direct/115-cloud/local 三种模式下的能力描述。 -// -// requestedDefinition 为空或 0 时默认选择 1080P；该档位不可用时返回 -// transcoding 状态（并只触发一次加速转码，避免轮询风暴）。 +// PlaybackInfo only reads direct/115-cloud/local playback capabilities. +// Cloud work is queued exclusively by StartTranscode, never by polling. func (s *Cloud115PlaybackService) PlaybackInfo(ctx context.Context, mediaID string, requestedDefinition int) (*PlaybackInfo, error) { + return s.playbackInfo(ctx, mediaID, requestedDefinition, false) +} + +func (s *Cloud115PlaybackService) playbackInfo(ctx context.Context, mediaID string, requestedDefinition int, startTranscode bool) (*PlaybackInfo, error) { if s == nil || s.repo == nil || s.repo.Media == nil { return nil, ErrMediaNotFound } @@ -179,13 +181,26 @@ func (s *Cloud115PlaybackService) PlaybackInfo(ctx context.Context, mediaID stri return info, nil } - info.Transcode = s.ensureCloudTranscode(ctx, account.ID, pickCode, selected, client) + if startTranscode { + info.Transcode = s.ensureCloudTranscode(ctx, account.ID, pickCode, selected, client) + } else { + info.Transcode = Cloud115TranscodeState{ + State: "idle", Definition: selected.ID, Message: "该清晰度尚未就绪，可请求 115 云端转码", RetryAfterSec: 5, + } + s.mu.Lock() + if last, ok := s.pushState[account.ID+":"+pickCode]; ok { + info.Transcode.State = "transcoding" + info.Transcode.Message = last.message + info.Transcode.StartedAt = last.at.Unix() + } + s.mu.Unlock() + } return info, nil } // StartTranscode 显式触发某个清晰度的云端转码，返回最新状态。 func (s *Cloud115PlaybackService) StartTranscode(ctx context.Context, mediaID string, definition int) (*PlaybackInfo, error) { - return s.PlaybackInfo(ctx, mediaID, definition) + return s.playbackInfo(ctx, mediaID, definition, true) } // ResolveCloud115URL 返回指定清晰度的 115 云端 m3u8 地址。 @@ -258,6 +273,9 @@ func (s *Cloud115PlaybackService) ensureCloudTranscode( } } } + // Reserve the in-flight attempt before releasing the lock. Two simultaneous + // explicit starts must not both submit a push while the first awaits 115. + s.pushState[key] = cloud115PushAttempt{at: now, message: "正在提交 115 云端转码请求…"} s.mu.Unlock() err := client.SubmitVideoPush(ctx, pickCode, "vip_push") diff --git a/internal/service/cloud115_playback_test.go b/internal/service/cloud115_playback_test.go index 2e1d451..0160cc7 100644 --- a/internal/service/cloud115_playback_test.go +++ b/internal/service/cloud115_playback_test.go @@ -16,7 +16,7 @@ import ( "github.com/truewhile/MeBox/internal/service/cloud115" ) -func TestCloud115PlaybackInfoTriggersTranscodeOnce(t *testing.T) { +func TestCloud115PlaybackInfoIsReadOnlyAndStartTriggersTranscodeOnce(t *testing.T) { var playCalls int32 var pushCalls int32 var ready int32 @@ -93,9 +93,16 @@ func TestCloud115PlaybackInfoTriggersTranscodeOnce(t *testing.T) { } svc := NewCloud115PlaybackService(nil, zap.NewNop(), repos, strm) - info, err := svc.PlaybackInfo(ctx, media.ID, 0) + if _, err := svc.PlaybackInfo(ctx, media.ID, 0); err != nil { + t.Fatalf("read playback info: %v", err) + } + if got := atomic.LoadInt32(&pushCalls); got != 0 { + t.Fatalf("read-only playback info triggered %d cloud transcode pushes", got) + } + + info, err := svc.StartTranscode(ctx, media.ID, 0) if err != nil { - t.Fatalf("first playback info: %v", err) + t.Fatalf("start transcode: %v", err) } if info.Provider != model.StrmProvider115 || info.DefaultQuality != "4" { t.Fatalf("unexpected playback info: %#v", info) diff --git a/internal/service/cloud115_transcode_concurrency_test.go b/internal/service/cloud115_transcode_concurrency_test.go new file mode 100644 index 0000000..9966779 --- /dev/null +++ b/internal/service/cloud115_transcode_concurrency_test.go @@ -0,0 +1,77 @@ +package service + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/service/cloud115" +) + +func TestCloud115ConcurrentStartsDoNotDuplicateInFlightPush(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + var releaseOnce sync.Once + var pushCalls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if pushCalls.Add(1) == 1 { + close(started) + } + <-release + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"state": true, "code": 0, "data": []any{}}) + })) + defer server.Close() + defer releaseOnce.Do(func() { close(release) }) + oldBase := cloud115.ProAPIBase + cloud115.ProAPIBase = server.URL + defer func() { cloud115.ProAPIBase = oldBase }() + client := cloud115.NewOpenClient("review-app", "access", "refresh") + svc := NewCloud115PlaybackService(nil, zap.NewNop(), nil, nil) + quality := PlaybackQuality{ID: "4", Label: "1080P", Source: "cloud"} + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + first := make(chan struct{}) + go func() { defer close(first); svc.ensureCloudTranscode(ctx, "account", "pick", quality, client) }() + select { + case <-started: + case <-ctx.Done(): + t.Fatal("first push did not reach the local server") + } + second := make(chan Cloud115TranscodeState, 1) + go func() { second <- svc.ensureCloudTranscode(ctx, "account", "pick", quality, client) }() + secondReceived := false + select { + case state := <-second: + secondReceived = true + if state.State != "transcoding" { + t.Errorf("in-flight request should be reported as transcoding: %#v", state) + } + case <-time.After(250 * time.Millisecond): + t.Error("second start issued another request instead of reusing the in-flight push") + } + releaseOnce.Do(func() { close(release) }) + select { + case <-first: + case <-ctx.Done(): + t.Fatal("first push did not finish") + } + // The duplicate path is awaited before restoring the package-wide fake API. + if !secondReceived { + select { + case <-second: + case <-ctx.Done(): + t.Fatal("second start did not finish") + } + } + if got := pushCalls.Load(); got != 1 { + t.Fatalf("pushes=%d, want 1", got) + } +} diff --git a/internal/service/hls_session.go b/internal/service/hls_session.go new file mode 100644 index 0000000..bce09bc --- /dev/null +++ b/internal/service/hls_session.go @@ -0,0 +1,109 @@ +package service + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "net/http" + "os" + "path/filepath" + "strings" + "time" + "unicode" +) + +var ErrInvalidHLSPlaybackScope = errors.New("invalid HLS playback scope") +var ErrHLSGenerationSuperseded = errors.New("HLS playback generation superseded") + +// HLSPlaybackScope separates a media's independent users and player instances. +// UserID must come from server authentication, never from a request parameter. +type HLSPlaybackScope struct { + UserID string + SessionID string +} + +func HLSPlaybackScopeForRequest(r *http.Request, userID string) (HLSPlaybackScope, error) { + scope := HLSPlaybackScope{UserID: strings.TrimSpace(userID)} + if scope.UserID == "" || r == nil { + return scope, ErrInvalidHLSPlaybackScope + } + for _, key := range []string{"session_id", "PlaySessionId", "playSessionId", "DeviceId", "deviceId"} { + if value := strings.TrimSpace(r.URL.Query().Get(key)); value != "" { + scope.SessionID = value + break + } + } + if scope.SessionID == "" { + if device := strings.TrimSpace(r.Header.Get("X-Emby-Device-Id")); device != "" { + scope.SessionID = device + } + } + if len(scope.SessionID) > 256 || strings.IndexFunc(scope.SessionID, unicode.IsControl) >= 0 { + return scope, ErrInvalidHLSPlaybackScope + } + // Clients without a session/device identifier remain compatible but only + // share a legacy task with the same authenticated user, never other users. + return scope, nil +} + +func (scope HLSPlaybackScope) jobKey(mediaID string) string { + if scope.UserID == "" && scope.SessionID == "" { + return mediaID // Existing internal callers/tests retain their legacy API. + } + sum := sha256.Sum256([]byte(scope.UserID + "\x00" + mediaID + "\x00" + scope.SessionID)) + return "player-" + hex.EncodeToString(sum[:]) +} + +func hlsPlaybackJobKey(mediaID string, scopes []HLSPlaybackScope) string { + if len(scopes) == 0 { + return mediaID + } + return scopes[0].jobKey(mediaID) +} + +func (job *hlsJob) key() string { + if job.jobKey != "" { + return job.jobKey + } + return job.mediaID +} + +func (t *TranscoderService) jobPlaylistPath(job *hlsJob) string { + dir := job.outputDir + if dir == "" { + dir = t.HLSDir(job.key()) + } + return filepath.Join(dir, "index.m3u8") +} + +// hlsOutputDir captures one immutable run directory. Even if a new run starts +// after this check, the request cannot accidentally open its replacement files. +func (t *TranscoderService) hlsOutputDir(key string, generation int64) (string, error) { + t.mu.Lock() + defer t.mu.Unlock() + job := t.jobs[key] + if job == nil { + return "", ErrMediaNotFound + } + if job.seekGen != generation { + return "", ErrHLSGenerationSuperseded + } + t.touchJobLocked(key) + if job.outputDir != "" { + return job.outputDir, nil + } + return t.HLSDir(key), nil +} + +func (t *TranscoderService) StopPlayback(mediaID string, scope HLSPlaybackScope) { + key := scope.jobKey(mediaID) + unlock := t.lockJobStart(key) + defer unlock() + t.mu.Lock() + job := t.detachJobLocked(key) + t.mu.Unlock() + waitJobExit(job, 12*time.Second) + // Keep cleanup under the same gate as start: a delayed stop must not wipe + // an output directory created by a concurrent replacement request. + _ = os.RemoveAll(t.HLSDir(key)) +} diff --git a/internal/service/hls_session_lifecycle_test.go b/internal/service/hls_session_lifecycle_test.go new file mode 100644 index 0000000..7dc6594 --- /dev/null +++ b/internal/service/hls_session_lifecycle_test.go @@ -0,0 +1,133 @@ +package service + +import ( + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" + "github.com/truewhile/MeBox/internal/model" + "github.com/truewhile/MeBox/internal/repository" +) + +func cachedScopedHLS(t *testing.T, scopes ...HLSPlaybackScope) (*StreamService, *TranscoderService, map[string]string) { + t.Helper() + cfg := &config.Config{} + cfg.Cache.CacheDir = t.TempDir() + cfg.Transcoder.Enabled = true + db := newServiceTestDB(t) + if err := db.Create(&model.Media{Base: model.Base{ID: "movie"}, Path: "unused-cached-source", DurationSec: 60}).Error; err != nil { + t.Fatal(err) + } + transcoder := NewTranscoderService(cfg, zap.NewNop(), repository.New(db), nil) + paths := map[string]string{} + for _, scope := range scopes { + key := scope.jobKey("movie") + dir := filepath.Join(transcoder.HLSDir(key), "run-cached") + if err := os.MkdirAll(dir, 0o750); err != nil { + t.Fatal(err) + } + playlist := filepath.Join(dir, "index.m3u8") + if err := os.WriteFile(playlist, []byte("#EXTM3U\n#EXTINF:4,\nseg_00000.ts\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "seg_00000.ts"), []byte(scope.UserID+"/"+scope.SessionID), 0o600); err != nil { + t.Fatal(err) + } + transcoder.jobs[key] = &hlsJob{mediaID: "movie", jobKey: key, outputDir: dir, subtitleStream: -1, startedAt: time.Now(), cancel: func() {}} + paths[key] = playlist + } + return NewStreamService(cfg, zap.NewNop(), repository.New(db), transcoder), transcoder, paths +} + +func TestHLSPlaybackEnsureReusesOnlyMatchingOwnerAndSession(t *testing.T) { + a := HLSPlaybackScope{UserID: "viewer", SessionID: "tab-a"} + b := HLSPlaybackScope{UserID: "viewer", SessionID: "tab-b"} + _, transcoder, paths := cachedScopedHLS(t, a, b) + for _, scope := range []HLSPlaybackScope{a, b} { + playlist, err := transcoder.EnsureJobFromSubtitleQuality(t.Context(), "movie", 0, 0, -1, "", scope) + if err != nil { + t.Fatal(err) + } + if playlist != paths[scope.jobKey("movie")] { + t.Fatalf("scope %#v reused another session's playlist %q", scope, playlist) + } + } +} + +func TestHLSPlaybackStopRemovesOnlyItsSessionOutput(t *testing.T) { + a := HLSPlaybackScope{UserID: "viewer", SessionID: "tab-a"} + b := HLSPlaybackScope{UserID: "viewer", SessionID: "tab-b"} + _, transcoder, paths := cachedScopedHLS(t, a, b) + transcoder.StopPlayback("movie", a) + if _, err := os.Stat(paths[a.jobKey("movie")]); !os.IsNotExist(err) { + t.Fatalf("stopped session output must be cleaned up, err=%v", err) + } + if _, err := os.Stat(paths[b.jobKey("movie")]); err != nil { + t.Fatalf("another player's output was removed: %v", err) + } +} + +func TestHLSPlaybackIdleCleanupRemovesExpiredSessionOutput(t *testing.T) { + a := HLSPlaybackScope{UserID: "viewer", SessionID: "tab-a"} + b := HLSPlaybackScope{UserID: "viewer", SessionID: "tab-b"} + _, transcoder, paths := cachedScopedHLS(t, a, b) + job := transcoder.jobs[a.jobKey("movie")] + job.lastAccess = time.Now().Add(-2 * time.Minute) + if !transcoder.stopIdleJob(job, time.Minute) { + t.Fatal("expired session was not stopped") + } + if _, err := os.Stat(paths[a.jobKey("movie")]); !os.IsNotExist(err) { + t.Fatalf("expired output was retained: %v", err) + } + if _, err := os.Stat(paths[b.jobKey("movie")]); err != nil { + t.Fatalf("active session output was removed: %v", err) + } +} + +func TestHLSPlaybackPropagatesHeaderDeviceIdentityToSegments(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/hls/movie/index.m3u8?token=test", nil) + req.Header.Set("X-Emby-Device-Id", "legacy-device") + scope, err := HLSPlaybackScopeForRequest(req, "viewer") + if err != nil { + t.Fatal(err) + } + svc, _, _ := cachedScopedHLS(t, scope) + rec := httptest.NewRecorder() + if err := svc.ServeHLSPlaylist(rec, req, "movie", scope); err != nil { + t.Fatal(err) + } + var segment string + for _, line := range strings.Split(rec.Body.String(), "\n") { + if strings.HasPrefix(line, "seg_") { + segment = line + break + } + } + u, err := url.Parse(segment) + if err != nil || u.Query().Get("session_id") == "" { + t.Fatalf("playlist must carry the header-only device scope to child URLs: %q err=%v", segment, err) + } + child := httptest.NewRequest(http.MethodGet, "/api/hls/movie/"+segment, nil) + childScope, err := HLSPlaybackScopeForRequest(child, "viewer") + if err != nil { + t.Fatal(err) + } + if childScope.jobKey("movie") != scope.jobKey("movie") { + t.Fatal("child request changed its owner's device scope") + } + childRec := httptest.NewRecorder() + if err := svc.ServeHLSSegment(childRec, child, "movie", "seg_00000.ts", childScope); err != nil { + t.Fatal(err) + } + if childRec.Body.String() != "viewer/"+scope.SessionID { + t.Fatalf("wrong scoped segment: %s", childRec.Body.String()) + } +} diff --git a/internal/service/hls_session_test.go b/internal/service/hls_session_test.go new file mode 100644 index 0000000..cbc930c --- /dev/null +++ b/internal/service/hls_session_test.go @@ -0,0 +1,97 @@ +package service + +import ( + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" +) + +func TestHLSPlaybackScopesIsolateSegmentReadsAndStop(t *testing.T) { + cfg := &config.Config{} + cfg.Cache.CacheDir = t.TempDir() + transcoder := NewTranscoderService(cfg, zap.NewNop(), nil, nil) + svc := NewStreamService(cfg, zap.NewNop(), nil, transcoder) + a := HLSPlaybackScope{UserID: "user-a", SessionID: "tab-a"} + b := HLSPlaybackScope{UserID: "user-b", SessionID: "tab-a"} + c := HLSPlaybackScope{UserID: "user-a", SessionID: "tab-c"} + for _, scope := range []HLSPlaybackScope{a, b, c} { + key := scope.jobKey("movie") + dir := filepath.Join(cfg.Cache.CacheDir, key, "run-current") + if err := os.MkdirAll(dir, 0o750); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "seg_00000.ts"), []byte(scope.UserID+"/"+scope.SessionID), 0o600); err != nil { + t.Fatal(err) + } + transcoder.jobs[key] = &hlsJob{mediaID: "movie", jobKey: key, outputDir: dir, seekGen: 100, cancel: func() {}} + } + for _, scope := range []HLSPlaybackScope{a, b, c} { + req := httptest.NewRequest(http.MethodGet, "/api/hls/movie/seg_00000.ts?_seek=100", nil) + rec := httptest.NewRecorder() + if err := svc.ServeHLSSegment(rec, req, "movie", "seg_00000.ts", scope); err != nil { + t.Fatal(err) + } + if got, want := rec.Body.String(), scope.UserID+"/"+scope.SessionID; got != want { + t.Fatalf("segment=%q, want %q", got, want) + } + } + transcoder.StopPlayback("movie", a) + if transcoder.jobs[a.jobKey("movie")] != nil { + t.Fatal("stopped scope is still active") + } + if transcoder.jobs[b.jobKey("movie")] == nil || transcoder.jobs[c.jobKey("movie")] == nil { + t.Fatal("stopping one player affected another user/tab") + } + // A delayed stop from a previous local-playback session must not affect a replacement. + transcoder.StopPlayback("movie", a) + if transcoder.jobs[c.jobKey("movie")] == nil { + t.Fatal("old stop cancelled a new session") + } +} + +func TestHLSPlaybackRejectsStaleSegmentGeneration(t *testing.T) { + cfg := &config.Config{} + cfg.Cache.CacheDir = t.TempDir() + transcoder := NewTranscoderService(cfg, zap.NewNop(), nil, nil) + svc := NewStreamService(cfg, zap.NewNop(), nil, transcoder) + scope := HLSPlaybackScope{UserID: "viewer", SessionID: "tab"} + key := scope.jobKey("movie") + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "seg_00000.ts"), []byte("new-generation"), 0o600); err != nil { + t.Fatal(err) + } + transcoder.jobs[key] = &hlsJob{mediaID: "movie", jobKey: key, outputDir: dir, seekGen: 200, cancel: func() {}} + req := httptest.NewRequest(http.MethodGet, "/api/hls/movie/seg_00000.ts?_seek=100", nil) + rec := httptest.NewRecorder() + if err := svc.ServeHLSSegment(rec, req, "movie", "seg_00000.ts", scope); err == nil { + t.Fatal("stale segment request read a replacement generation") + } + if rec.Body.Len() != 0 { + t.Fatal("stale request exposed replacement bytes") + } +} + +func TestHLSPlaybackScopeRequiresServerOwner(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/?session_id=tab&user_id=attacker-controlled", nil) + if _, err := HLSPlaybackScopeForRequest(req, ""); err == nil { + t.Fatal("missing authenticated owner was accepted") + } + a, err := HLSPlaybackScopeForRequest(req, "real-owner") + if err != nil { + t.Fatal(err) + } + if a.UserID != "real-owner" { + t.Fatal("client parameter supplied the owner") + } + req = httptest.NewRequest(http.MethodGet, "/?PlaySessionId=emby-session", nil) + b, err := HLSPlaybackScopeForRequest(req, "real-owner") + if err != nil || b.SessionID == "" { + t.Fatalf("Emby session must stay scoped: %#v %v", b, err) + } +} diff --git a/internal/service/stream_hls.go b/internal/service/stream_hls.go index 81416a8..44b7dec 100644 --- a/internal/service/stream_hls.go +++ b/internal/service/stream_hls.go @@ -17,7 +17,8 @@ import ( // Optional query `start` (seconds) restarts ffmpeg from that source offset so // the web player can scrub the full timeline without waiting for a full // head-to-tail transcode. -func (s *StreamService) ServeHLSPlaylist(w http.ResponseWriter, r *http.Request, mediaID string) error { +func (s *StreamService) ServeHLSPlaylist(w http.ResponseWriter, r *http.Request, mediaID string, scopes ...HLSPlaybackScope) error { + jobKey := hlsPlaybackJobKey(mediaID, scopes) // 「客户端直连解码」模式下宿主机不提供转码，HLS 一律拒绝， // 迫使播放器走 direct play 本地解码。 if s.directPlayOnly(r.Context()) { @@ -27,19 +28,23 @@ func (s *StreamService) ServeHLSPlaylist(w http.ResponseWriter, r *http.Request, seekGen := parseHLSSeekGen(r) subtitleStream := parseHLSSubtitleStream(r) quality := parseHLSQuality(r) - if _, err := s.transcoder.EnsureJobFromSubtitleQuality(r.Context(), mediaID, startSec, seekGen, subtitleStream, quality); err != nil { + if _, err := s.transcoder.EnsureJobFromSubtitleQuality(r.Context(), mediaID, startSec, seekGen, subtitleStream, quality, scopes...); err != nil { return err } - s.transcoder.TouchJob(mediaID) + s.transcoder.TouchJob(jobKey) readyTimeout := 45 * time.Second if startSec > 0.05 { // Mid-file HTTP seeks (esp. WMV) need longer before the first segment appears. readyTimeout = 120 * time.Second } - if !s.transcoder.WaitReady(r.Context(), mediaID, readyTimeout) { + if !s.transcoder.WaitReady(r.Context(), jobKey, readyTimeout) { return errors.New("hls playlist not ready") } - playlist := s.transcoder.PlaylistPath(mediaID) + outputDir, err := s.transcoder.hlsOutputDir(jobKey, seekGen) + if err != nil { + return err + } + playlist := filepath.Join(outputDir, "index.m3u8") f, err := os.Open(playlist) // #nosec G304 -- playlist path is generated under the transcoder cache directory for this media ID. if err != nil { return err @@ -49,12 +54,18 @@ func (s *StreamService) ServeHLSPlaylist(w http.ResponseWriter, r *http.Request, w.Header().Set("Content-Type", "application/vnd.apple.mpegurl") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Content-Disposition", "inline") - if r.URL.RawQuery != "" { + rawQuery := r.URL.RawQuery + if len(scopes) > 0 && scopes[0].SessionID != "" { + query := r.URL.Query() + query.Set("session_id", scopes[0].SessionID) + rawQuery = query.Encode() + } + if rawQuery != "" { data, err := io.ReadAll(f) if err != nil { return err } - playlist := appendQueryToHLSSegments(string(data), r.URL.RawQuery) + playlist := appendQueryToHLSSegments(string(data), rawQuery) _, err = io.WriteString(w, playlist) return err } @@ -171,19 +182,26 @@ func filterHLSSegmentQuery(rawQuery string) string { } // ServeHLSSegment writes a single .ts segment from the on-disk cache. -func (s *StreamService) ServeHLSSegment(w http.ResponseWriter, r *http.Request, mediaID, segment string) error { - s.transcoder.TouchJob(mediaID) +func (s *StreamService) ServeHLSSegment(w http.ResponseWriter, r *http.Request, mediaID, segment string, scopes ...HLSPlaybackScope) error { + jobKey := hlsPlaybackJobKey(mediaID, scopes) + outputDir, err := s.transcoder.hlsOutputDir(jobKey, parseHLSSeekGen(r)) + if err != nil { + if len(scopes) > 0 || parseHLSSeekGen(r) > 0 { + return err + } + outputDir = s.transcoder.HLSDir(jobKey) // Legacy internal static-file callers. + } // Only allow segments that look like seg_NNNNN.ts so we cannot be tricked // into reading arbitrary files via path traversal. if !strings.HasPrefix(segment, "seg_") || !strings.HasSuffix(segment, ".ts") { return errors.New("bad segment") } - full := filepath.Join(s.transcoder.HLSDir(mediaID), segment) + full := filepath.Join(outputDir, segment) abs, err := filepath.Abs(full) if err != nil { return err } - dir, _ := filepath.Abs(s.transcoder.HLSDir(mediaID)) + dir, _ := filepath.Abs(outputDir) if !pathWithin(abs, dir) { return errors.New("path escape") } diff --git a/internal/service/transcoder.go b/internal/service/transcoder.go index 357981a..ff3f0e2 100644 --- a/internal/service/transcoder.go +++ b/internal/service/transcoder.go @@ -1,7 +1,8 @@ // Package service — HLS on-demand transcoder. // // TranscoderService spawns ffmpeg processes that segment a source media file -// into HLS (.m3u8 + .ts). The output lives under cache.cache_dir/hls/. +// into HLS (.m3u8 + .ts). Each playback scope owns a cache directory and +// every ffmpeg run writes to its own run-* subdirectory. // The HTTP layer serves these files directly with a normal http.FileServer. // // Encoder selection (read once at startup from the config): @@ -15,9 +16,11 @@ // plus the kernel module loaded) // // Concurrency model: -// - Each Media has at most one active ffmpeg job. -// - jobs[mediaID] tracks the running goroutine + cancel func. -// - Calling Start while a job already exists is a no-op. +// - Each authenticated user/player scope has at most one active ffmpeg job. +// - jobs[jobKey] tracks the running goroutine + cancel func; mediaID remains +// the actual media identity for database access and activity reporting. +// - Legacy internal callers keep media-keyed behaviour. +// - Equivalent requests within one scope reuse the current output. // - When the playlist file appears on disk we consider the job "ready" // and unblock the HTTP handler that was waiting on it. package service @@ -49,9 +52,10 @@ type TranscoderService struct { mu sync.Mutex jobs map[string]*hlsJob - // startGates serializes EnsureJobFrom / StopJob per media so concurrent - // playlist hits cannot spawn multiple ffmpeg writers into one HLS dir. - startGates sync.Map // mediaID -> *sync.Mutex + // Start gates are reference-counted so random playback session keys do not + // accumulate after their last starter/stopper has left. + startGateMu sync.Mutex + startGates map[string]*hlsStartGate strmResolve func(ctx context.Context, raw string) (*StrmPlayResult, error) probe *FFprobeService @@ -66,6 +70,7 @@ type TranscoderService struct { // hlsJob holds the live state of one ffmpeg run. type hlsJob struct { + jobKey string mediaID string outputDir string cancel context.CancelFunc @@ -156,14 +161,19 @@ func (t *TranscoderService) transcodeConfig() *config.Config { } } -// HLSDir is the per-media directory that holds index.m3u8 + segment files. -func (t *TranscoderService) HLSDir(mediaID string) string { - return filepath.Join(t.cfg.Cache.CacheDir, "hls", mediaID) +// HLSDir is the cache root for one internal playback job key. +func (t *TranscoderService) HLSDir(jobKey string) string { + return filepath.Join(t.cfg.Cache.CacheDir, "hls", jobKey) } // PlaylistPath returns the absolute path of the m3u8 playlist for a media. -func (t *TranscoderService) PlaylistPath(mediaID string) string { - return filepath.Join(t.HLSDir(mediaID), "index.m3u8") +func (t *TranscoderService) PlaylistPath(jobKey string) string { + t.mu.Lock() + defer t.mu.Unlock() + if job := t.jobs[jobKey]; job != nil { + return t.jobPlaylistPath(job) + } + return filepath.Join(t.HLSDir(jobKey), "index.m3u8") } // EnsureJob makes sure a transcode is running for mediaID from the start of @@ -194,7 +204,8 @@ func (t *TranscoderService) EnsureJobFromSubtitle(ctx context.Context, mediaID s // EnsureJobFromSubtitleQuality is EnsureJobFromSubtitle with an explicit local // HLS quality profile. Empty qualityID keeps the global transcoder config. -func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, mediaID string, startSec float64, seekGen int64, subtitleStream int, qualityID string) (string, error) { +func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, mediaID string, startSec float64, seekGen int64, subtitleStream int, qualityID string, scopes ...HLSPlaybackScope) (string, error) { + jobKey := hlsPlaybackJobKey(mediaID, scopes) cfg := t.transcodeConfig() if !cfg.Transcoder.Enabled { return "", ErrTranscodeDisabled @@ -219,28 +230,31 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me return "", ErrMediaNotFound } - gate := t.mediaStartGate(mediaID) - gate.Lock() - defer gate.Unlock() + unlock := t.lockJobStart(jobKey) + defer unlock() t.mu.Lock() - if existing, ok := t.jobs[mediaID]; ok { + if existing, ok := t.jobs[jobKey]; ok { reusable := true + playlist := t.jobPlaylistPath(existing) if existing.completed { - info, err := os.Stat(t.PlaylistPath(mediaID)) + info, err := os.Stat(playlist) reusable = err == nil && info.Size() > 0 } if reusable && sameHLSConfiguration(existing, startSec, subtitleStream, qualityID) { - t.touchJobLocked(mediaID) + if seekGen > existing.seekGen { + existing.seekGen = seekGen + } + t.touchJobLocked(jobKey) t.mu.Unlock() - return t.PlaylistPath(mediaID), nil + return playlist, nil } if reusable && !shouldReplaceHLSJobConfigurationQuality(existing, startSec, seekGen, subtitleStream, qualityID) { - t.touchJobLocked(mediaID) + t.touchJobLocked(jobKey) t.mu.Unlock() - return t.PlaylistPath(mediaID), nil + return playlist, nil } - prev := t.detachJobLocked(mediaID) + prev := t.detachJobLocked(jobKey) t.mu.Unlock() waitJobExit(prev, 12*time.Second) } else { @@ -267,12 +281,23 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me return "", err } - outDir := t.HLSDir(mediaID) - // Wipe prior segments so a mid-file restart cannot serve stale early chunks. - // Only safe after the previous ffmpeg has exited (waited above / via gate). - if err := resetHLSDir(outDir); err != nil { + rootDir := t.HLSDir(jobKey) + // Runs use distinct directories even when Windows keeps an old segment + // open. A delayed request can never read a newer run's seg_00000.ts. + if err := resetHLSDir(rootDir); err != nil { return "", err } + outDir, err := os.MkdirTemp(rootDir, "run-") + if err != nil { + return "", err + } + registered := false + defer func() { + if !registered { + _ = os.RemoveAll(outDir) + _ = os.Remove(rootDir) + } + }() t.mu.Lock() if max := t.maxConcurrent(); max > 0 && t.runningJobCountLocked() >= max { @@ -288,6 +313,7 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me jobEncoder = "software (subtitle burn-in)" } job := &hlsJob{ + jobKey: jobKey, mediaID: mediaID, outputDir: outDir, cancel: cancel, @@ -302,7 +328,8 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me ffmpegPath: bin, config: cfg, } - t.jobs[mediaID] = job + t.jobs[jobKey] = job + registered = true t.mu.Unlock() helper.Go(t.log, "transcoder.monitorIdle", func() { t.monitorIdle(jobCtx, job) }) @@ -310,7 +337,7 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me defer close(job.done) t.runFFmpeg(jobCtx, job, input) }) - return t.PlaylistPath(mediaID), nil + return filepath.Join(outDir, "index.m3u8"), nil } // shouldReplaceHLSJob reports whether an incoming playlist request may cancel @@ -357,9 +384,33 @@ func sameHLSConfiguration(existing *hlsJob, startSec float64, subtitleStream int sameHLSStart(existing.startSec, startSec) } -func (t *TranscoderService) mediaStartGate(mediaID string) *sync.Mutex { - v, _ := t.startGates.LoadOrStore(mediaID, &sync.Mutex{}) - return v.(*sync.Mutex) +type hlsStartGate struct { + mu sync.Mutex + users int +} + +func (t *TranscoderService) lockJobStart(key string) func() { + t.startGateMu.Lock() + if t.startGates == nil { + t.startGates = make(map[string]*hlsStartGate) + } + gate := t.startGates[key] + if gate == nil { + gate = &hlsStartGate{} + t.startGates[key] = gate + } + gate.users++ + t.startGateMu.Unlock() + gate.mu.Lock() + return func() { + gate.mu.Unlock() + t.startGateMu.Lock() + gate.users-- + if gate.users == 0 { + delete(t.startGates, key) + } + t.startGateMu.Unlock() + } } // detachJobLocked cancels and removes a job from the map without waiting. diff --git a/internal/service/transcoder_ffmpeg_probe.go b/internal/service/transcoder_ffmpeg_probe.go index 9cc6c1c..366fd20 100644 --- a/internal/service/transcoder_ffmpeg_probe.go +++ b/internal/service/transcoder_ffmpeg_probe.go @@ -49,16 +49,18 @@ func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, input tr t.mu.Lock() // Only update the map entry if we are still the registered generation. - if cur, ok := t.jobs[job.mediaID]; ok && cur == job { + if cur, ok := t.jobs[job.key()]; ok && cur == job { if err == nil && ctx.Err() == nil { job.completed = true } else { - delete(t.jobs, job.mediaID) + delete(t.jobs, job.key()) } } t.mu.Unlock() if err != nil || ctx.Err() != nil { job.cancel() + _ = os.RemoveAll(job.outputDir) + _ = os.Remove(filepath.Dir(job.outputDir)) // Remove the scope root only if empty. } t.hub.Publish("transcode", map[string]any{ diff --git a/internal/service/transcoder_jobs.go b/internal/service/transcoder_jobs.go index 496ffdb..9f43fe9 100644 --- a/internal/service/transcoder_jobs.go +++ b/internal/service/transcoder_jobs.go @@ -26,16 +26,18 @@ func (t *TranscoderService) WaitReady(ctx context.Context, mediaID string, timeo job, ok := t.jobs[mediaID] var started time.Time var completed bool + var playlist string if ok { started = job.startedAt completed = job.completed + playlist = t.jobPlaylistPath(job) } t.mu.Unlock() if !ok { return false // A failed/exited job cannot become ready later. } if ok { - if info, err := os.Stat(t.PlaylistPath(mediaID)); err == nil { + if info, err := os.Stat(playlist); err == nil { // Allow a small clock skew; reject anything older than this job. if !info.ModTime().Before(started.Add(-2 * time.Second)) { t.mu.Lock() @@ -66,9 +68,8 @@ func (t *TranscoderService) WaitReady(ctx context.Context, mediaID string, timeo // briefly for it to exit so a subsequent EnsureJobFrom cannot race-write the // same HLS directory. func (t *TranscoderService) StopJob(mediaID string) { - gate := t.mediaStartGate(mediaID) - gate.Lock() - defer gate.Unlock() + unlock := t.lockJobStart(mediaID) + defer unlock() t.mu.Lock() prev := t.detachJobLocked(mediaID) @@ -170,7 +171,7 @@ func (t *TranscoderService) monitorIdle(ctx context.Context, job *hlsJob) { return case <-ticker.C: t.mu.Lock() - current, ok := t.jobs[job.mediaID] + current, ok := t.jobs[job.key()] if !ok || current != job { t.mu.Unlock() return @@ -185,11 +186,10 @@ func (t *TranscoderService) monitorIdle(ctx context.Context, job *hlsJob) { } func (t *TranscoderService) stopIdleJob(job *hlsJob, timeout time.Duration) bool { - gate := t.mediaStartGate(job.mediaID) - gate.Lock() - defer gate.Unlock() + unlock := t.lockJobStart(job.key()) + defer unlock() t.mu.Lock() - current := t.jobs[job.mediaID] + current := t.jobs[job.key()] if current != job { t.mu.Unlock() return true @@ -199,9 +199,12 @@ func (t *TranscoderService) stopIdleJob(job *hlsJob, timeout time.Duration) bool t.mu.Unlock() return false } - t.detachJobLocked(job.mediaID) + t.detachJobLocked(job.key()) t.mu.Unlock() waitJobExit(job, 12*time.Second) + if job.jobKey != "" { + _ = os.RemoveAll(t.HLSDir(job.key())) + } t.log.Info("transcode idle timeout", zap.String("media_id", job.mediaID), zap.Duration("idle_for", idleFor), zap.Duration("timeout", timeout)) return true diff --git a/web/src/api/client.ts b/web/src/api/client.ts index 4a0912b..a7f729e 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -175,7 +175,7 @@ export function streamURL(mediaId: string, options: { proxy?: boolean } = {}): s // hlsURL returns the m3u8 playlist URL fed into hls.js. // startSec > 0 asks the server to (re)start ffmpeg from that source offset. -export function hlsURL(mediaId: string, startSec = 0, subtitleStream?: number, quality?: string): string { +export function hlsURL(mediaId: string, startSec = 0, subtitleStream?: number, quality?: string, sessionID?: string): string { const safeStart = Math.max(0, Math.round(startSec * 1000) / 1000) // Always send start= (including 0) so the server can tell an intentional // restart-from-head apart from a missing query on a stale refresh. @@ -188,7 +188,8 @@ export function hlsURL(mediaId: string, startSec = 0, subtitleStream?: number, q ? `&subtitle=${encodeURIComponent(String(subtitleStream))}` : '' const qualityQuery = quality ? `&quality=${encodeURIComponent(quality)}` : '' - return `/api/hls/${encodeURIComponent(mediaId)}/index.m3u8?${tokenQuery()}${profileQuery()}${start}${subtitle}${qualityQuery}${bust}` + const session = sessionID ? `&session_id=${encodeURIComponent(sessionID)}` : '' + return `/api/hls/${encodeURIComponent(mediaId)}/index.m3u8?${tokenQuery()}${profileQuery()}${start}${subtitle}${qualityQuery}${bust}${session}` } // cloudHlsURL returns the MeBox-proxied 115 cloud HLS master playlist URL. @@ -202,8 +203,9 @@ export function cloudHlsURL(mediaId: string, definition: string): string { // Stop an on-demand HLS job. keepalive makes the request survive page // navigation/tab close, where an axios promise can be discarded by browsers. -export function stopHLSJob(mediaId: string): void { - const url = `/api/hls/${encodeURIComponent(mediaId)}?${tokenQuery()}${profileQuery()}` +export function stopHLSJob(mediaId: string, sessionID?: string): void { + const session = sessionID ? `&session_id=${encodeURIComponent(sessionID)}` : '' + const url = `/api/hls/${encodeURIComponent(mediaId)}?${tokenQuery()}${profileQuery()}${session}` void fetch(url, { method: 'DELETE', credentials: 'same-origin', diff --git a/web/src/pages/PlayerPage.tsx b/web/src/pages/PlayerPage.tsx index d90f70b..0dae777 100644 --- a/web/src/pages/PlayerPage.tsx +++ b/web/src/pages/PlayerPage.tsx @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useMemo, useRef, useState } from 'react' +import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react' import { useLocation, useNavigate, useParams } from 'react-router-dom' import type Hls from 'hls.js' import toast from 'react-hot-toast' @@ -41,7 +41,9 @@ import { pickPlayerMode, needsTranscodeForBrowser, isDirectStreamMedia, isStrmMe import { classifyDirectPlayError, directPlayFailureCopy, directStallTimeoutMs, DIRECT_SEEK_GRACE_MS } from './directPlayError' import type { DirectPlayFailureReason } from './directPlayError' import { canSeekDirectNow, DIRECT_SEEK_WAIT_MS } from './directPlaySeek' -import { applyPendingHlsSeek, attachNativeHls, bindPendingHlsSeek, hlsManifestTimeoutMs, playbackPositionSec } from './hlsPlayback' +import { applyPendingHlsSeek, attachNativeHls, bindPendingHlsSeek, hlsManifestTimeoutMs, playbackPositionSec, playbackTimeSec } from './hlsPlayback' +import { availableLocalQuality, defaultLocalQualityId, findPlaybackQualityById } from './playbackQuality' +import { createPlaybackRequestGuard } from './playbackRequestGuard' import { apiErrorMessage } from './StrmManagePage' import { PlayerTopBar } from './PlayerTopBar' import { PlayerVideoStage } from './PlayerVideoStage' @@ -118,6 +120,14 @@ export function PlayerPage() { const hlsRef = useRef(null) const lastSentRef = useRef(0) const progressSessionRef = useRef(null) + const cloudRequestGuardRef = useRef(createPlaybackRequestGuard()) + const currentMediaIdRef = useRef(id) + useLayoutEffect(() => { + const guard = cloudRequestGuardRef.current + currentMediaIdRef.current = id + guard.invalidate() + return () => guard.invalidate() + }, [id]) const directRetryRef = useRef(false) const retryingDirectRef = useRef(false) // 程序化续播 seek 后的宽限截止时间；期内直连误报不升到 HLS。 @@ -563,6 +573,7 @@ export function PlayerPage() { const setPlaybackMode = useCallback((next: PlayerMode) => { // 一旦显式决定过播放方式，URL 上的 ?mode= 就不再是权威（避免它把状态打回去）。 requestedModeRef.current = null + if (next === 'direct') cloudRequestGuardRef.current.invalidate() setMode(next) }, []) @@ -767,6 +778,11 @@ export function PlayerPage() { */ const handleDirectSeekStalled = useCallback( (target: number) => { + cloudRequestGuardRef.current.invalidate() + if (directOnly || isDirectStreamMedia(mediaRef.current)) { + toast.error('当前播放方式不提供 HLS 转码，请使用外部播放器定位') + return + } const cloudQuality = findPlaybackQualityById(playbackInfo, selectedQuality) ?? findPlaybackQualityById(playbackInfo, playbackInfo?.default_quality ?? '') @@ -780,9 +796,7 @@ export function PlayerPage() { toast('直连定位失败，已切到 115 云 HLS 跳转') return } - const localQuality = - findPlaybackQualityById(playbackInfo, selectedQuality) ?? - findPlaybackQualityById(playbackInfo, defaultLocalQualityId(playbackInfo)) + const localQuality = availableLocalQuality(playbackInfo, selectedQuality) if (localQuality?.available) { setSelectedQuality(localQuality.id) setHlsSource('local') @@ -795,7 +809,7 @@ export function PlayerPage() { } toast.error('直连定位失败，且当前没有可用的 HLS 转码档位') }, - [playbackInfo, selectedQuality, setPlaybackMode], + [directOnly, playbackInfo, selectedQuality, setPlaybackMode], ) // Load metadata and pick a default mode. @@ -1032,17 +1046,22 @@ export function PlayerPage() { const switchToLocalHLS = useCallback( (position?: number) => { + cloudRequestGuardRef.current.invalidate() pendingSeekRef.current = null setHlsSource('local') setCloudWaiting(false) - setSelectedQuality((prev) => { - if (playbackInfo?.local_qualities?.some((quality) => quality.id === prev)) return prev - return defaultLocalQualityId(playbackInfo) - }) + const quality = availableLocalQuality(playbackInfo, selectedQuality) + if (playbackInfo && !quality) { + setHlsUnavailable(true) + showPlayFailure('本地 HLS 转码不可用，请配置 ffmpeg 或使用外部播放器。') + setPlaybackMode('direct') + return + } + setSelectedQuality(quality?.id ?? defaultLocalQualityId(playbackInfo)) if (position !== undefined) setHlsStartSec(Math.max(0, position)) setPlaybackMode('hls') }, - [playbackInfo, setPlaybackMode], + [playbackInfo, selectedQuality, setPlaybackMode, showPlayFailure], ) /** @@ -1080,8 +1099,11 @@ export function PlayerPage() { const startCloudTranscode = useCallback( async (definition: number) => { if (!mediaId) return + const isCurrentRequest = cloudRequestGuardRef.current.begin() + const isCurrent = () => isCurrentRequest() && currentMediaIdRef.current === mediaId && !directOnlyRef.current try { const info = await mediaAPI.startCloudTranscode(mediaId, definition) + if (!isCurrent() || info.media_id !== mediaId) return setPlaybackInfo(info) if (info.transcode.state === 'ready') { setCloudWaiting(false) @@ -1094,6 +1116,7 @@ export function PlayerPage() { setCloudWaitMessage(info.transcode.message || '115 正在转码…') setCloudWaitStartedAt(Date.now()) } catch { + if (!isCurrent()) return switchToLocalHLS() toast.error('115 云端转码触发失败，已切换本地转码') } @@ -1105,6 +1128,7 @@ export function PlayerPage() { // 的自愈逻辑共用，避免两处各写一遍云端/本地选择。 const enterHlsPlayback = useCallback( (startSec: number, infoOverride?: PlaybackInfo | null) => { + cloudRequestGuardRef.current.invalidate() const position = Math.max(0, startSec) // 允许传入刚拿到的能力信息：从 URL 强制 HLS 时不想等到 state 更新后才选线路。 const info = infoOverride ?? playbackInfo @@ -1127,6 +1151,7 @@ export function PlayerPage() { } } else { switchToLocalHLS(position) + return } setPlaybackMode('hls') }, @@ -1165,7 +1190,7 @@ export function PlayerPage() { setCloudWaitMessage(info.transcode.message || '115 正在转码…') cloudPollTimerRef.current = setTimeout(poll, cloudRetryRef.current * 1000) } catch { - cloudPollTimerRef.current = setTimeout(poll, 5000) + if (!cancelled) cloudPollTimerRef.current = setTimeout(poll, 5000) } } cloudPollTimerRef.current = setTimeout(poll, cloudRetryRef.current * 1000) @@ -1338,6 +1363,7 @@ export function PlayerPage() { if (!currentMedia) return let cancelled = false let cleanupHlsResume: (() => void) | undefined + let stopLocalPlayback: (() => void) | undefined let onDirectMetadata: (() => void) | undefined teardownHls() @@ -1348,10 +1374,17 @@ export function PlayerPage() { teardownHls() return } + // Each wired local source owns its job, including seek/quality changes. + // Capture the ID in cleanup so a delayed DELETE cannot stop a newer run. + const localSessionID = hlsSource === 'local' ? newPlaybackProgressSession(mediaId).id : undefined + if (localSessionID) { + stopLocalPlayback = () => stopHLSJob(mediaId, localSessionID) + window.addEventListener('pagehide', stopLocalPlayback) + } const url = hlsSource === 'cloud' ? cloudHlsURL(mediaId, selectedQuality) - : hlsURL(mediaId, hlsStartSec, activeBurnedSubtitleStream, selectedQuality) + : hlsURL(mediaId, hlsStartSec, activeBurnedSubtitleStream, selectedQuality, localSessionID) void import('hls.js').then(({ default: HlsCtor }) => { if (cancelled || !ref.current) return if (HlsCtor.isSupported()) { @@ -1494,6 +1527,10 @@ export function PlayerPage() { if (onDirectMetadata) video.removeEventListener('loadedmetadata', onDirectMetadata) cleanupHlsResume?.() teardownHls() + if (stopLocalPlayback) { + window.removeEventListener('pagehide', stopLocalPlayback) + stopLocalPlayback() + } } }, [ activeBurnedSubtitleStream, @@ -1518,24 +1555,6 @@ export function PlayerPage() { vr360DirectProxy, ]) - // Stop host ffmpeg when leaving this HLS player. The keepalive request also - // survives route navigation while the component is being torn down. - useEffect(() => { - if (!mediaId || mode !== 'hls' || hlsSource !== 'local') return - return () => { - stopHLSJob(mediaId) - } - }, [hlsSource, mediaId, mode]) - - // React cleanup is not guaranteed when a tab/window closes. pagehide fires - // while the document can still dispatch a keepalive request. - useEffect(() => { - if (!mediaId || mode !== 'hls' || hlsSource !== 'local') return - const stopOnPageExit = () => stopHLSJob(mediaId) - window.addEventListener('pagehide', stopOnPageExit) - return () => window.removeEventListener('pagehide', stopOnPageExit) - }, [hlsSource, mediaId, mode]) - // 自动拉取已有的播放进度并恢复播放位置 useEffect(() => { if (!id) return @@ -1828,8 +1847,7 @@ export function PlayerPage() { const sync = () => { // HLS 转码时