Compare commits

...

2 Commits

Author SHA1 Message Date
truewhile 74487359bf 优 2026-10-21 13:21:13 +08:00
truewhile 55854a0adb 优化播放逻辑 2026-10-21 12:02:11 +08:00
29 changed files with 1254 additions and 133 deletions
BIN
View File
Binary file not shown.
@@ -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.
+24 -2
View File
@@ -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,
+10 -2
View File
@@ -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)
}
}
+37
View File
@@ -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())
}
}
+43 -3
View File
@@ -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")
@@ -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)
}
}
}
+15 -2
View File
@@ -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{
+24 -6
View File
@@ -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")
+10 -3
View File
@@ -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)
@@ -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)
}
}
+109
View File
@@ -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))
}
@@ -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())
}
}
+97
View File
@@ -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)
}
}
+29 -11
View File
@@ -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")
}
+89 -32
View File
@@ -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/<id>.
// 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 {
@@ -258,6 +272,9 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me
input.Quality = quality
t.maybeFillDuration(ctx, m, input)
settings := ffmpegArgSettingsFromConfig(cfg, quality)
// 「原画」档位遇到浏览器可解的 H.264 源时只换封装,避免慢机器上重编码
// 追不上播放(两核软件编码实测 0.1~0.2 倍速)。其它档位要改分辨率,仍重编码。
input.CopyVideo = t.remuxEligible(ctx, m, quality, subtitleStream)
jobEncoder := settings.encoder
if settings.noScale || subtitleStream >= 0 {
jobEncoder = ""
@@ -267,12 +284,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 {
@@ -287,7 +315,11 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me
if subtitleStream >= 0 {
jobEncoder = "software (subtitle burn-in)"
}
if input.CopyVideo {
jobEncoder = remuxEncoderLabel
}
job := &hlsJob{
jobKey: jobKey,
mediaID: mediaID,
outputDir: outDir,
cancel: cancel,
@@ -302,7 +334,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 +343,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 +390,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.
+33 -4
View File
@@ -16,6 +16,9 @@ type transcodeInput struct {
StartSec float64
SubtitleStream *int
Quality *LocalHLSQuality
// CopyVideo 为 true 时视频流只换封装(-c:v copy),不重编码。
// 仅当源视频是浏览器可解的老三样(H.264/8bit/4:2:0)且不需要缩放时启用。
CopyVideo bool
}
type ffmpegArgSettings struct {
@@ -52,6 +55,7 @@ func buildFFmpegArgsForInput(cfg *config.Config, input transcodeInput, playlist,
// 原画档位不做硬件缩放的兼容处理,直接走软件编码,保持源分辨率。
settings.encoder = ""
}
copyVideo := input.CopyVideo && input.SubtitleStream == nil
video := ffmpegVideoPlanForSettings(settings)
if input.SubtitleStream != nil {
// Bitmap subtitles must be composited in software. Keeping CUDA/QSV/
@@ -60,13 +64,17 @@ func buildFFmpegArgsForInput(cfg *config.Config, input transcodeInput, playlist,
settings.encoder = ""
video = ffmpegVideoPlanForSettings(settings)
}
if copyVideo {
// 直通:不做缩放、不挂滤镜、不选编码器,整条视频流原样塞进 TS。
video = ffmpegVideoPlan{codec: "copy"}
}
// Mid-file restarts must not use -re: output -ss would otherwise crawl to the
// seek point at 1x wall-clock before emitting the first HLS segment.
realtime := settings.realtime && input.StartSec <= 0.05
args := baseFFmpegArgs(video.preInput, realtime)
args = appendInputAndVideoArgs(args, input, settings, video)
args = appendOutputHLSArgs(args, settings, segments, playlist)
args = appendInputAndVideoArgs(args, input, settings, video, copyVideo)
args = appendOutputHLSArgs(args, settings, segments, playlist, copyVideo)
return args
}
@@ -161,7 +169,7 @@ func baseFFmpegArgs(preInput string, realtime bool) []string {
return args
}
func appendInputAndVideoArgs(args []string, input transcodeInput, settings ffmpegArgSettings, video ffmpegVideoPlan) []string {
func appendInputAndVideoArgs(args []string, input transcodeInput, settings ffmpegArgSettings, video ffmpegVideoPlan, copyVideo bool) []string {
args = append(args, ffmpegHTTPInputArgs(input)...)
ss := ""
if input.StartSec > 0.05 {
@@ -172,6 +180,11 @@ func appendInputAndVideoArgs(args []string, input transcodeInput, settings ffmpe
// scrubbing; CDNs with Range support jump via demuxer seek instead.
if ss != "" {
args = append(args, "-ss", ss)
if copyVideo {
// 复制视频流时 ffmpeg 不能解码,精确 seek 会把关键帧之前的分组丢掉,
// 输出从 P 帧开始(花屏)。退到前一个关键帧,最多提前一两秒。
args = append(args, "-noaccurate_seek")
}
}
args = append(args, "-i", input.Source)
if input.SubtitleStream != nil {
@@ -197,7 +210,23 @@ func appendInputAndVideoArgs(args []string, input transcodeInput, settings ffmpe
return args
}
func appendOutputHLSArgs(args []string, settings ffmpegArgSettings, segments, playlist string) []string {
func appendOutputHLSArgs(args []string, settings ffmpegArgSettings, segments, playlist string, copyVideo bool) []string {
if copyVideo {
// 视频是复制进来的,编码参数一律不适用;只保留音频转 AAC 与切片参数。
// -avoid_negative_ts 让带 B 帧的源在 -ss 之后不会出现负时间戳。
return append(args,
"-avoid_negative_ts", "make_zero",
"-c:a", "aac",
"-ar", "48000",
"-b:a", "128k",
"-ac", "2",
"-f", "hls",
"-hls_time", fmt.Sprintf("%d", settings.segmentSeconds),
"-hls_list_size", "0",
"-hls_segment_filename", segments,
playlist,
)
}
return append(args,
"-pix_fmt", "yuv420p",
"-b:v", settings.bitrate,
+4 -2
View File
@@ -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{
+13 -10
View File
@@ -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
+102
View File
@@ -0,0 +1,102 @@
package service
import (
"context"
"encoding/json"
"strings"
"github.com/truewhile/MeBox/internal/model"
)
// 蓝光原盘(m2ts,40 Mbps 级别)在两核小机器上重编码只有 0.1~0.2 倍速,HLS
// 永远追不上播放,播放器等不到切片就报「转码失败」。当源视频本身就是浏览器
// 能解的 H.264 / 8bit / 4:2:0,且用户选的是「原画」档位(不改分辨率)时,
// 这里改走只换封装(-c:v copy)的直通路径:视频不重编码,音频转 AAC,
// CPU 占用接近于零,起播时间从几十秒降到几秒。
//
// 直通只对「原画」生效。1080P/720P/480P 档位意味着要改分辨率,必须重新编码;
// 那类请求仍然按原来的 libx264 路径处理(慢,但码率可控)。
// remuxEncoderLabel 是直通任务在日志与任务面板里的标签。
const remuxEncoderLabel = "remux"
// remuxEligible 判断这次请求能否复制视频流而不是重编码。
func (t *TranscoderService) remuxEligible(ctx context.Context, m *model.Media, quality *LocalHLSQuality, subtitleStream int) bool {
if t == nil || t.repo == nil || m == nil {
return false
}
// 烧录图形字幕要把字幕叠加进画面,必须解码+重编码。
if subtitleStream >= 0 {
return false
}
if !remuxAppliesToQuality(quality) {
return false
}
probe, err := t.repo.MediaProbe.Get(ctx, m.ID)
if err != nil || probe == nil {
// 没有探测缓存就无法确认档次/像素格式,保守走重编码。
return false
}
return remuxCopyableHeight(probe) > 0
}
// remuxAppliesToQuality 只放行「原画」档位(Height 为 0,即不缩放)。
// quality 为 nil 表示调用方没有指定档位(走全局 max_height),同样不直通,
// 否则会绕过配置的分辨率上限给出一个远超预期的码流。
func remuxAppliesToQuality(quality *LocalHLSQuality) bool {
return quality != nil && quality.Height <= 0
}
// remuxCopyableHeight 在源视频可直接复制进 HLS 时返回它的高度,否则返回 0。
func remuxCopyableHeight(probe *model.MediaProbe) int {
if probe == nil || probe.Height <= 0 {
return 0
}
if !strings.EqualFold(strings.TrimSpace(probe.VideoCodec), "h264") {
return 0
}
if !browserSafeH264Payload(probe.Payload) {
return 0
}
return probe.Height
}
// probePayloadStreams 只解析落库 payload 里与直通判断有关的字段。
type probePayloadStreams struct {
Streams []struct {
Type string `json:"type"`
Profile string `json:"profile"`
PixFmt string `json:"pix_fmt"`
} `json:"streams"`
}
// browserSafeH264Payload 只放行 8bit 4:2:0 的 Baseline/Main/High。
// Hi10P(High 10)和 4:2:2 母带浏览器解不出来,复制过去只会黑屏或无声。
func browserSafeH264Payload(payload string) bool {
payload = strings.TrimSpace(payload)
if payload == "" {
// 老数据没有 payload:已经有 codec + 分辨率,按 8bit 常规情况处理。
return true
}
var parsed probePayloadStreams
if err := json.Unmarshal([]byte(payload), &parsed); err != nil {
return false
}
for _, stream := range parsed.Streams {
if !strings.EqualFold(strings.TrimSpace(stream.Type), "video") {
continue
}
switch strings.ToLower(strings.TrimSpace(stream.Profile)) {
case "", "constrained baseline", "baseline", "main", "high":
default:
return false
}
switch strings.ToLower(strings.TrimSpace(stream.PixFmt)) {
case "", "yuv420p", "yuvj420p":
default:
return false
}
return true
}
return true
}
+170
View File
@@ -0,0 +1,170 @@
package service
import (
"strings"
"testing"
"github.com/truewhile/MeBox/internal/config"
"github.com/truewhile/MeBox/internal/model"
)
// 直通档位必须真的把视频流复制过去:一旦又带上 libx264 / -vf / 码率参数,
// 慢机器上的「原画」就还是几十秒起播。
func TestBuildFFmpegArgsRemuxCopiesVideoStream(t *testing.T) {
cfg := &config.Config{}
cfg.Transcoder.SegmentSeconds = 4
cfg.Transcoder.Realtime = true
cfg.Transcoder.Threads = 2
cfg.Transcoder.MaxHeight = 720
quality, ok := LocalHLSQualityByID("source")
if !ok {
t.Fatal("source quality missing")
}
args := buildFFmpegArgsForInput(cfg, transcodeInput{
Source: "https://cdn.example/x.m2ts",
Headers: map[string]string{"User-Agent": "MeBox-115-GoClient/1.0"},
StartSec: 52.4,
Quality: &quality,
CopyVideo: true,
}, "/cache/hls/k/run-1/index.m3u8", "/cache/hls/k/run-1/seg_%05d.ts")
joined := strings.Join(args, " ")
for _, want := range []string{
"-c:v copy",
"-c:a aac",
"-ar 48000",
"-b:a 128k",
"-ac 2",
"-avoid_negative_ts make_zero",
"-ss 52.400",
"-noaccurate_seek",
"-hls_time 4",
"-hls_segment_filename /cache/hls/k/run-1/seg_%05d.ts",
} {
if !strings.Contains(joined, want) {
t.Errorf("remux args missing %q: %s", want, joined)
}
}
for _, unwanted := range []string{
"libx264",
"-vf ",
"-b:v",
"-maxrate",
"-bufsize",
"-force_key_frames",
"-preset",
"-threads",
"-pix_fmt",
} {
if strings.Contains(joined, unwanted) {
t.Errorf("remux args must not contain %q: %s", unwanted, joined)
}
}
// 中途起播不能加 -re,否则会先按 1 倍速爬到目标位置才出第一个切片。
if strings.Contains(joined, " -re ") {
t.Errorf("mid-file remux must not use -re: %s", joined)
}
}
// 烧录图形字幕要把字幕叠进画面,必须解码重编码,不能被直通标志绕过。
func TestBuildFFmpegArgsRemuxIgnoredForSubtitleBurnIn(t *testing.T) {
cfg := &config.Config{}
cfg.Transcoder.MaxHeight = 720
cfg.Transcoder.SegmentSeconds = 4
quality, _ := LocalHLSQualityByID("source")
subtitle := 4
args := buildFFmpegArgsForInput(cfg, transcodeInput{
Source: "/media/ep.mkv",
Quality: &quality,
CopyVideo: true,
SubtitleStream: &subtitle,
}, "/o/x.m3u8", "/o/seg_%05d.ts")
joined := strings.Join(args, " ")
if strings.Contains(joined, "-c:v copy") {
t.Fatalf("subtitle burn-in must not copy the video stream: %s", joined)
}
if !strings.Contains(joined, "libx264") || !strings.Contains(joined, "-filter_complex") {
t.Fatalf("subtitle burn-in should re-encode: %s", joined)
}
if strings.Contains(joined, "-noaccurate_seek") {
t.Fatalf("encoded path should keep accurate seek: %s", joined)
}
}
// 只有「原画」档位(不缩放)可以直通;其它档位要改分辨率。
func TestRemuxAppliesToQuality(t *testing.T) {
if remuxAppliesToQuality(nil) {
t.Fatal("nil quality must not remux: it falls back to the configured max height")
}
source, _ := LocalHLSQualityByID("source")
if !remuxAppliesToQuality(&source) {
t.Fatal("source quality should remux")
}
for _, id := range []string{"1080", "720", "480"} {
quality, ok := LocalHLSQualityByID(id)
if !ok {
t.Fatalf("quality %s missing", id)
}
if remuxAppliesToQuality(&quality) {
t.Fatalf("quality %s must re-encode", id)
}
}
}
// 只有 8bit 4:2:0 的 H.264 能安全复制;Hi10P / 4:2:2 / HEVC 复制过去浏览器黑屏。
func TestRemuxCopyableHeight(t *testing.T) {
const blurayPayload = `{"container":"mpegts","duration_sec":1451,"bit_rate":42609934,"streams":[{"index":0,"type":"video","codec":"h264","profile":"High","width":1920,"height":1080,"pix_fmt":"yuv420p","frame_rate":"23.976"},{"index":1,"type":"audio","codec":"pcm_bluray","channels":2}]}`
cases := []struct {
name string
probe *model.MediaProbe
want int
}{
{
name: "bluray h264 8bit 420",
probe: &model.MediaProbe{VideoCodec: "h264", Height: 1080, Payload: blurayPayload},
want: 1080,
},
{
name: "no payload falls back to codec and height",
probe: &model.MediaProbe{VideoCodec: "h264", Height: 720},
want: 720,
},
{
name: "hevc cannot be copied into browser hls",
probe: &model.MediaProbe{VideoCodec: "hevc", Height: 1080, Payload: `{"streams":[{"type":"video","profile":"Main 10","pix_fmt":"yuv420p10le"}]}`},
want: 0,
},
{
name: "hi10p h264 is unplayable in browsers",
probe: &model.MediaProbe{VideoCodec: "h264", Height: 1080, Payload: `{"streams":[{"type":"video","profile":"High 10","pix_fmt":"yuv420p10le"}]}`},
want: 0,
},
{
name: "4:2:2 h264 is unplayable in browsers",
probe: &model.MediaProbe{VideoCodec: "h264", Height: 1080, Payload: `{"streams":[{"type":"video","profile":"High 4:2:2","pix_fmt":"yuv422p"}]}`},
want: 0,
},
{
name: "missing probe",
probe: nil,
want: 0,
},
{
name: "unknown height",
probe: &model.MediaProbe{VideoCodec: "h264", Height: 0, Payload: blurayPayload},
want: 0,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := remuxCopyableHeight(tc.probe); got != tc.want {
t.Fatalf("remuxCopyableHeight() = %d, want %d", got, tc.want)
}
})
}
}
+6 -4
View File
@@ -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',
+54 -51
View File
@@ -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<Hls | null>(null)
const lastSentRef = useRef(0)
const progressSessionRef = useRef<PlaybackProgressSession | null>(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 转码时 <video>.currentTime 是相对转码起点的,必须把起点加回来。这里与
// 进度上报用的算法保持一致,否则转码场景下整段判断都会错位。
const offset = modeRef.current === 'hls' ? hlsStartSecRef.current : 0
const absolute = offset + (video.currentTime || 0)
const absolute = playbackPositionSec(modeRef.current, hlsSourceRef.current, hlsStartSecRef.current, video.currentTime || 0)
const next = resolveActiveSkip(absolute, skipSegments, {
hasNextEpisode: Boolean(nextEpisode),
excludedKinds: dismissedSkipKinds,
@@ -1854,8 +1872,7 @@ export function PlayerPage() {
selfSkipSeekRef.current = false
return
}
const offset = modeRef.current === 'hls' ? hlsStartSecRef.current : 0
const absolute = offset + (video.currentTime || 0)
const absolute = playbackPositionSec(modeRef.current, hlsSourceRef.current, hlsStartSecRef.current, video.currentTime || 0)
const hit = skipSegments.find(
(segment) =>
(segment.kind === 'intro' || segment.kind === 'recap') &&
@@ -2169,7 +2186,7 @@ export function PlayerPage() {
return
}
selfSkipSeekRef.current = true
video.currentTime = Math.max(0, target - hlsStartSec)
video.currentTime = playbackTimeSec(mode, hlsSource, hlsStartSec, target)
return
}
if (isStrmMedia(mediaRef.current)) {
@@ -2277,6 +2294,7 @@ export function PlayerPage() {
const selectPlaybackQuality = useCallback(
(quality: PlaybackQuality) => {
cloudRequestGuardRef.current.invalidate()
const video = ref.current
const localTime = video?.currentTime || 0
// 云 HLS 的 currentTime 本来就在绝对时间轴上;本地 HLS 的 t=0 对应 hlsStartSec。
@@ -2659,21 +2677,6 @@ function formatEpisodeDisplay(ep: Media, siblings: Media[]): string {
return episodeLabel || mediaTitle || title || '未命名'
}
function findPlaybackQualityById(info: PlaybackInfo | null, id: string): PlaybackQuality | undefined {
if (!info || !id) return undefined
return [...(info.cloud_qualities ?? []), ...(info.local_qualities ?? [])].find(
(quality) => quality.id === id,
)
}
function defaultLocalQualityId(info: PlaybackInfo | null): string {
const qualities = info?.local_qualities ?? []
for (const id of ['1080', '720', '480', 'source']) {
if (qualities.some((quality) => quality.id === id)) return id
}
return qualities[0]?.id ?? '720'
}
function looksLikeSeriesTitle(ep: Media, title: string, siblings: Media[]): boolean {
const normalized = normalizeEpisodeTitle(title)
if (!normalized) return true
+5 -1
View File
@@ -1,5 +1,5 @@
import { strict as assert } from 'node:assert'
import { applyPendingHlsSeek, attachNativeHls, bindPendingHlsSeek, hlsManifestTimeoutMs, playbackPositionSec } from './hlsPlayback.ts'
import { applyPendingHlsSeek, attachNativeHls, bindPendingHlsSeek, hlsManifestTimeoutMs, playbackPositionSec, playbackTimeSec } from './hlsPlayback.ts'
class FakeVideo extends EventTarget {
src = ''
@@ -55,4 +55,8 @@ assert.equal(hlsManifestTimeoutMs('cloud', 1800), 60_000)
assert.equal(playbackPositionSec('hls', 'cloud', 420, 600), 600, 'cloud subtitle/quality switches must not add an old local offset')
assert.equal(playbackPositionSec('hls', 'local', 420, 60), 480, 'local subtitle switches preserve the source position')
assert.equal(playbackPositionSec('direct', 'local', 420, 60), 60)
assert.equal(playbackTimeSec('hls', 'cloud', 600, 640), 640, 'cloud skip/undo uses the absolute target, not target minus a former local offset')
assert.equal(playbackTimeSec('hls', 'local', 600, 640), 40)
assert.equal(playbackTimeSec('direct', 'local', 600, 640), 640)
assert.equal(playbackTimeSec('hls', 'local', 600, 100), 0, 'targets before a local start clamp to zero')
console.log('hlsPlayback.test.ts ok')
+5
View File
@@ -41,3 +41,8 @@ export function attachNativeHls(video: HlsVideo, url: string, pending: PendingSe
export function playbackPositionSec(mode: 'direct' | 'hls', source: 'cloud' | 'local', startSec: number, currentTime: number): number {
return mode === 'hls' && source === 'local' ? startSec + currentTime : currentTime
}
/** Convert an absolute source target to the current media element's timeline. */
export function playbackTimeSec(mode: 'direct' | 'hls', source: 'cloud' | 'local', startSec: number, absoluteSec: number): number {
return Math.max(0, mode === 'hls' && source === 'local' ? absoluteSec - startSec : absoluteSec)
}
+19
View File
@@ -0,0 +1,19 @@
import { strict as assert } from 'node:assert'
import { availableLocalQuality, findPlaybackQualityById } from './playbackQuality.ts'
import type { PlaybackInfo } from '../types/playback.ts'
const info: PlaybackInfo = {
media_id: 'm', provider: 'cloud115', default_quality: '4', fallback: [],
transcode: { state: 'idle' },
cloud_qualities: [{ id: '4', label: '1080P', source: 'cloud', available: false }],
local_qualities: [
{ id: 'source', label: '原画', source: 'local', available: false },
{ id: '1080', label: '1080P', source: 'local', available: true },
{ id: '720', label: '720P', source: 'local', available: true },
],
}
assert.equal(availableLocalQuality(info, '4')?.id, '1080', 'pending cloud quality must not mask the available local fallback')
assert.equal(availableLocalQuality(info, '720')?.id, '720', 'preserve an available local preference')
assert.equal(availableLocalQuality({ ...info, local_qualities: info.local_qualities.map(q => ({ ...q, available: false })) }, '4'), undefined)
assert.equal(findPlaybackQualityById(info, '4')?.source, 'cloud')
console.log('playbackQuality.test.ts ok')
+24
View File
@@ -0,0 +1,24 @@
import type { PlaybackInfo, PlaybackQuality } from '../types/playback'
export function findPlaybackQualityById(info: PlaybackInfo | null, id: string): PlaybackQuality | undefined {
if (!info || !id) return undefined
return [...(info.cloud_qualities ?? []), ...(info.local_qualities ?? [])].find(
(quality) => quality.id === id,
)
}
export function defaultLocalQualityId(info: PlaybackInfo | null): string {
const qualities = info?.local_qualities ?? []
for (const id of ['1080', '720', '480', 'source']) {
if (qualities.some((quality) => quality.id === id)) return id
}
return qualities[0]?.id ?? '720'
}
/** A cloud option may exist without being ready; it is never a local fallback. */
export function availableLocalQuality(info: PlaybackInfo | null, preferred = ''): PlaybackQuality | undefined {
const available = info?.local_qualities.filter((quality) => quality.source === 'local' && quality.available) ?? []
return available.find((quality) => quality.id === preferred) ??
['1080', '720', '480', 'source'].map((id) => available.find((quality) => quality.id === id)).find(Boolean) ??
available[0]
}
@@ -0,0 +1,30 @@
import { strict as assert } from 'node:assert'
import { createPlaybackRequestGuard } from './playbackRequestGuard.ts'
const guard = createPlaybackRequestGuard()
const first = guard.begin()
assert.equal(first(), true)
const second = guard.begin()
assert.equal(first(), false, 'a newer quality request supersedes the earlier request')
assert.equal(second(), true)
guard.invalidate()
assert.equal(second(), false, 'media changes, unmount and direct/local switches cancel a pending cloud result')
let resolveRequest: (value: string) => void = () => {}
const request = new Promise<string>(resolve => { resolveRequest = resolve })
const current = guard.begin()
let applied = 'episode-B'
const settled = request.then(value => { if (current()) applied = value })
guard.invalidate()
resolveRequest('episode-A')
await settled
assert.equal(applied, 'episode-B', 'a delayed response cannot overwrite the next media')
let rejected: (error: Error) => void = () => {}
const failed = new Promise<never>((_, reject) => { rejected = reject })
const activeFailure = guard.begin()
let fallbackCount = 0
const caught = failed.catch(() => { if (activeFailure()) fallbackCount += 1 })
guard.invalidate()
rejected(new Error('old request failed'))
await caught
assert.equal(fallbackCount, 0, 'a stale rejection cannot force a local fallback')
console.log('playbackRequestGuard.test.ts ok')
+13
View File
@@ -0,0 +1,13 @@
/** Invalidates both successful and failed responses after a newer playback intent. */
export function createPlaybackRequestGuard() {
let generation = 0
return {
begin(): () => boolean {
const requested = ++generation
return () => requested === generation
},
invalidate(): void {
generation += 1
},
}
}