From 27dcfc2fe91dc4e192be9f74821a5a1988540548 Mon Sep 17 00:00:00 2001 From: jason jia Date: Sat, 17 Oct 2026 01:30:04 +0000 Subject: [PATCH] fix(playback): reuse completed HLS output and preserve resume positions --- internal/service/transcoder.go | 73 ++++-- internal/service/transcoder_ffmpeg_probe.go | 90 ++++--- internal/service/transcoder_jobs.go | 79 ++++-- internal/service/transcoder_lifecycle_test.go | 228 ++++++++++++++++++ web/src/api/playback.ts | 4 +- web/src/pages/PlayerPage.tsx | 108 ++++----- web/src/pages/hlsPlayback.test.ts | 58 +++++ web/src/pages/hlsPlayback.ts | 43 ++++ 8 files changed, 557 insertions(+), 126 deletions(-) create mode 100644 internal/service/transcoder_lifecycle_test.go create mode 100644 web/src/pages/hlsPlayback.test.ts create mode 100644 web/src/pages/hlsPlayback.ts diff --git a/internal/service/transcoder.go b/internal/service/transcoder.go index 4c0e7d6..357981a 100644 --- a/internal/service/transcoder.go +++ b/internal/service/transcoder.go @@ -59,6 +59,9 @@ type TranscoderService struct { availMu sync.Mutex availCheckedAt time.Time availValue bool + availKey string + ffmpegMu sync.Mutex + ffmpegCache ffmpegValidationCache } // hlsJob holds the live state of one ffmpeg run. @@ -83,6 +86,11 @@ type hlsJob struct { quality string // done is closed when the ffmpeg goroutine fully exits (after process death). done chan struct{} + // Completed outputs stay reusable until idle eviction without occupying a + // running-process slot. The binary and config are fixed for this generation. + completed bool + ffmpegPath string + config *config.Config } var ( @@ -117,24 +125,37 @@ func (t *TranscoderService) Available() bool { if t == nil || t.cfg == nil { return false } - if !t.cfg.Transcoder.Enabled { + cfg := t.transcodeConfig() + if !cfg.Transcoder.Enabled { return false } + encoder := ffmpegArgSettingsFromConfig(cfg, nil).encoder + key := cfg.App.FFmpegPath + "\x00" + encoder t.availMu.Lock() defer t.availMu.Unlock() - if !t.availCheckedAt.IsZero() && time.Since(t.availCheckedAt) < transcodeAvailabilityTTL { - return t.availValue + if !t.availValue && t.availKey == key && !t.availCheckedAt.IsZero() && time.Since(t.availCheckedAt) < transcodeAvailabilityTTL { + return false } - _, err := t.resolveFFmpegPath() + _, err := t.resolveFFmpegPathFor(context.Background(), cfg.App.FFmpegPath, encoder) available := err == nil if !available && t.log != nil { t.log.Debug("local HLS renditions unavailable", zap.Error(err)) } t.availValue = available t.availCheckedAt = time.Now() + t.availKey = key return available } +func (t *TranscoderService) transcodeConfig() *config.Config { + config.RuntimeMu.RLock() + defer config.RuntimeMu.RUnlock() + return &config.Config{ + App: config.AppConfig{FFmpegPath: t.cfg.App.FFmpegPath, VAAPIDevice: t.cfg.App.VAAPIDevice}, + Transcoder: t.cfg.Transcoder, + } +} + // 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) @@ -174,7 +195,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) { - if !t.cfg.Transcoder.Enabled { + cfg := t.transcodeConfig() + if !cfg.Transcoder.Enabled { return "", ErrTranscodeDisabled } if startSec < 0 { @@ -203,12 +225,17 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me t.mu.Lock() if existing, ok := t.jobs[mediaID]; ok { - if sameHLSConfiguration(existing, startSec, subtitleStream, qualityID) { + reusable := true + if existing.completed { + info, err := os.Stat(t.PlaylistPath(mediaID)) + reusable = err == nil && info.Size() > 0 + } + if reusable && sameHLSConfiguration(existing, startSec, subtitleStream, qualityID) { t.touchJobLocked(mediaID) t.mu.Unlock() return t.PlaylistPath(mediaID), nil } - if !shouldReplaceHLSJobConfigurationQuality(existing, startSec, seekGen, subtitleStream, qualityID) { + if reusable && !shouldReplaceHLSJobConfigurationQuality(existing, startSec, seekGen, subtitleStream, qualityID) { t.touchJobLocked(mediaID) t.mu.Unlock() return t.PlaylistPath(mediaID), nil @@ -230,7 +257,13 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me } input.Quality = quality t.maybeFillDuration(ctx, m, input) - if _, err := t.resolveFFmpegPath(); err != nil { + settings := ffmpegArgSettingsFromConfig(cfg, quality) + jobEncoder := settings.encoder + if settings.noScale || subtitleStream >= 0 { + jobEncoder = "" + } + bin, err := t.resolveFFmpegPathFor(ctx, cfg.App.FFmpegPath, jobEncoder) + if err != nil { return "", err } @@ -242,29 +275,15 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me } t.mu.Lock() - if existing, ok := t.jobs[mediaID]; ok { - if sameHLSConfiguration(existing, startSec, subtitleStream, qualityID) { - t.touchJobLocked(mediaID) - t.mu.Unlock() - return t.PlaylistPath(mediaID), nil - } - if !shouldReplaceHLSJobConfigurationQuality(existing, startSec, seekGen, subtitleStream, qualityID) { - t.touchJobLocked(mediaID) - t.mu.Unlock() - return t.PlaylistPath(mediaID), nil - } - prev := t.detachJobLocked(mediaID) - t.mu.Unlock() - waitJobExit(prev, 12*time.Second) - t.mu.Lock() - } - if max := t.maxConcurrent(); max > 0 && len(t.jobs) >= max { + if max := t.maxConcurrent(); max > 0 && t.runningJobCountLocked() >= max { t.mu.Unlock() return "", ErrTranscodeBusy } jobCtx, cancel := context.WithCancel(context.Background()) - jobEncoder := t.effectiveEncoder() + if jobEncoder == "" { + jobEncoder = "software" + } if subtitleStream >= 0 { jobEncoder = "software (subtitle burn-in)" } @@ -280,6 +299,8 @@ func (t *TranscoderService) EnsureJobFromSubtitleQuality(ctx context.Context, me subtitleStream: subtitleStream, quality: qualityID, done: make(chan struct{}), + ffmpegPath: bin, + config: cfg, } t.jobs[mediaID] = job t.mu.Unlock() diff --git a/internal/service/transcoder_ffmpeg_probe.go b/internal/service/transcoder_ffmpeg_probe.go index dee811b..9cc6c1c 100644 --- a/internal/service/transcoder_ffmpeg_probe.go +++ b/internal/service/transcoder_ffmpeg_probe.go @@ -11,29 +11,17 @@ import ( "time" "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" ) func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, input transcodeInput) { - bin, err := t.resolveFFmpegPath() - if err != nil { - t.log.Warn("ffmpeg unavailable", zap.String("media_id", job.mediaID), zap.Error(err)) - t.mu.Lock() - delete(t.jobs, job.mediaID) - t.mu.Unlock() - t.hub.Publish("transcode", map[string]any{ - "media_id": job.mediaID, - "status": "error", - "error": err.Error(), - }) - return - } - playlist := filepath.Join(job.outputDir, "index.m3u8") segments := filepath.Join(job.outputDir, "seg_%05d.ts") - args := buildFFmpegArgsForInput(t.cfg, input, playlist, segments) + args := buildFFmpegArgsForInput(job.config, input, playlist, segments) - cmd := exec.CommandContext(ctx, bin, args...) // #nosec G204 -- bin is resolved by resolveFFmpegPath and args are passed without a shell. + cmd := exec.CommandContext(ctx, job.ffmpegPath, args...) // #nosec G204 -- the binary was validated before creating this job. setFFmpegSysProcAttr(cmd) cmd.Stderr = os.Stderr @@ -50,7 +38,8 @@ func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, input tr "start_sec": input.StartSec, }) - if err := cmd.Run(); err != nil && !errors.Is(ctx.Err(), context.Canceled) { + err := cmd.Run() + if err != nil && !errors.Is(ctx.Err(), context.Canceled) { t.log.Warn("ffmpeg exited", zap.String("media_id", job.mediaID), zap.Float64("start_sec", input.StartSec), @@ -59,11 +48,18 @@ func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, input tr } t.mu.Lock() - // Only drop the map entry if we are still the registered generation. + // Only update the map entry if we are still the registered generation. if cur, ok := t.jobs[job.mediaID]; ok && cur == job { - delete(t.jobs, job.mediaID) + if err == nil && ctx.Err() == nil { + job.completed = true + } else { + delete(t.jobs, job.mediaID) + } } t.mu.Unlock() + if err != nil || ctx.Err() != nil { + job.cancel() + } t.hub.Publish("transcode", map[string]any{ "media_id": job.mediaID, @@ -73,19 +69,53 @@ func (t *TranscoderService) runFFmpeg(ctx context.Context, job *hlsJob, input tr } func (t *TranscoderService) resolveFFmpegPath() (string, error) { - t.mu.Lock() - configuredPath := strings.TrimSpace(t.cfg.App.FFmpegPath) - t.mu.Unlock() + cfg := t.transcodeConfig() + return t.resolveFFmpegPathFor(context.Background(), cfg.App.FFmpegPath, ffmpegArgSettingsFromConfig(cfg, nil).encoder) +} + +type ffmpegValidationCache struct { + configuredPath string + path string + encoder string + size int64 + modTime time.Time + checkedAt time.Time +} + +func (t *TranscoderService) resolveFFmpegPathFor(ctx context.Context, configuredPath, encoder string) (string, error) { + t.ffmpegMu.Lock() + defer t.ffmpegMu.Unlock() + if err := ctx.Err(); err != nil { + return "", err + } + configuredPath = strings.TrimSpace(configuredPath) + cached := t.ffmpegCache + if cached.path != "" && cached.encoder == encoder && + (configuredPath == cached.configuredPath || configuredPath == cached.path) && + time.Since(cached.checkedAt) < transcodeAvailabilityTTL { + if info, err := os.Stat(cached.path); err == nil && info.Size() == cached.size && info.ModTime().Equal(cached.modTime) { + return cached.path, nil + } + } var lastErr error for _, bin := range executableCandidates(configuredPath, "ffmpeg") { - if err := validateFFmpegForTranscode(context.Background(), bin, t.effectiveEncoder()); err != nil { + if err := validateFFmpegForTranscode(ctx, bin, encoder); err != nil { + if ctx.Err() != nil { + return "", ctx.Err() + } lastErr = err continue } - t.mu.Lock() - t.cfg.App.FFmpegPath = bin - t.mu.Unlock() + if info, err := os.Stat(bin); err == nil { + t.ffmpegCache = ffmpegValidationCache{configuredPath: configuredPath, path: bin, encoder: encoder, + size: info.Size(), modTime: info.ModTime(), checkedAt: time.Now()} + } + config.RuntimeMu.Lock() + if strings.TrimSpace(t.cfg.App.FFmpegPath) == configuredPath { + t.cfg.App.FFmpegPath = bin + } + config.RuntimeMu.Unlock() return bin, nil } if lastErr != nil { @@ -95,10 +125,11 @@ func (t *TranscoderService) resolveFFmpegPath() (string, error) { } func (t *TranscoderService) effectiveEncoder() string { - if !t.cfg.Transcoder.HardwareAccel { + cfg := t.transcodeConfig() + if !cfg.Transcoder.HardwareAccel { return "" } - return normalizedHardwareEncoder(t.cfg.Transcoder.Encoder) + return normalizedHardwareEncoder(cfg.Transcoder.Encoder) } func normalizedHardwareEncoder(encoder string) string { @@ -160,6 +191,7 @@ func hasFFmpegListEntry(output, name string) bool { // HumanFFmpegProfile is exposed for the admin UI / settings view. func (t *TranscoderService) HumanFFmpegProfile() string { + cfg := t.transcodeConfig() return fmt.Sprintf("ffmpeg=%s, encoder=%s, output=%s", - t.cfg.App.FFmpegPath, t.cfg.Transcoder.Encoder, filepath.Join(t.cfg.Cache.CacheDir, "hls")) + cfg.App.FFmpegPath, cfg.Transcoder.Encoder, filepath.Join(t.cfg.Cache.CacheDir, "hls")) } diff --git a/internal/service/transcoder_jobs.go b/internal/service/transcoder_jobs.go index daedb3e..496ffdb 100644 --- a/internal/service/transcoder_jobs.go +++ b/internal/service/transcoder_jobs.go @@ -6,41 +6,58 @@ import ( "time" "go.uber.org/zap" + + "github.com/truewhile/MeBox/internal/config" ) // WaitReady blocks (with a deadline) until the playlist file shows up on // disk for the *current* job generation. Stale playlists left behind by a // failed RemoveAll / still-exiting ffmpeg must not unblock a mid-file restart. func (t *TranscoderService) WaitReady(ctx context.Context, mediaID string, timeout time.Duration) bool { - deadline := time.Now().Add(timeout) + timer := time.NewTimer(timeout) + defer timer.Stop() + ticker := time.NewTicker(250 * time.Millisecond) + defer ticker.Stop() for { + if ctx.Err() != nil { + return false + } t.mu.Lock() job, ok := t.jobs[mediaID] var started time.Time + var completed bool if ok { started = job.startedAt + completed = job.completed } 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 { // Allow a small clock skew; reject anything older than this job. if !info.ModTime().Before(started.Add(-2 * time.Second)) { t.mu.Lock() - if j, exists := t.jobs[mediaID]; exists { + if j, exists := t.jobs[mediaID]; exists && j == job { j.playlistOK = true + t.mu.Unlock() + return true } t.mu.Unlock() - return true } } } - if time.Now().After(deadline) || ctx.Err() != nil { - return false + if completed { + return false // A completed job cannot regenerate a missing playlist. } select { case <-ctx.Done(): return false - case <-time.After(250 * time.Millisecond): + case <-timer.C: + return false + case <-ticker.C: + case <-job.done: } } } @@ -101,6 +118,9 @@ func (t *TranscoderService) Active() []ActiveJob { defer t.mu.Unlock() out := make([]ActiveJob, 0, len(t.jobs)) for _, j := range t.jobs { + if j.completed { + continue + } out = append(out, ActiveJob{ MediaID: j.mediaID, Encoder: j.encoder, @@ -111,7 +131,19 @@ func (t *TranscoderService) Active() []ActiveJob { return out } +func (t *TranscoderService) runningJobCountLocked() int { + n := 0 + for _, job := range t.jobs { + if !job.completed { + n++ + } + } + return n +} + func (t *TranscoderService) maxConcurrent() int { + config.RuntimeMu.RLock() + defer config.RuntimeMu.RUnlock() if t.cfg.Transcoder.MaxConcurrent <= 0 { return 1 } @@ -119,6 +151,8 @@ func (t *TranscoderService) maxConcurrent() int { } func (t *TranscoderService) idleTimeout() time.Duration { + config.RuntimeMu.RLock() + defer config.RuntimeMu.RUnlock() if t.cfg.Transcoder.IdleTimeoutSeconds <= 0 { return 120 * time.Second } @@ -137,21 +171,38 @@ func (t *TranscoderService) monitorIdle(ctx context.Context, job *hlsJob) { case <-ticker.C: t.mu.Lock() current, ok := t.jobs[job.mediaID] - if !ok { + if !ok || current != job { t.mu.Unlock() return } idleFor := time.Since(current.lastAccess) t.mu.Unlock() - if idleFor >= timeout { - t.log.Info("transcode idle timeout", - zap.String("media_id", job.mediaID), - zap.Duration("idle_for", idleFor), - zap.Duration("timeout", timeout), - ) - t.StopJob(job.mediaID) + if idleFor >= timeout && t.stopIdleJob(job, timeout) { return } } } } + +func (t *TranscoderService) stopIdleJob(job *hlsJob, timeout time.Duration) bool { + gate := t.mediaStartGate(job.mediaID) + gate.Lock() + defer gate.Unlock() + t.mu.Lock() + current := t.jobs[job.mediaID] + if current != job { + t.mu.Unlock() + return true + } + idleFor := time.Since(job.lastAccess) + if idleFor < timeout { + t.mu.Unlock() + return false + } + t.detachJobLocked(job.mediaID) + t.mu.Unlock() + waitJobExit(job, 12*time.Second) + 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/internal/service/transcoder_lifecycle_test.go b/internal/service/transcoder_lifecycle_test.go new file mode 100644 index 0000000..4adb7cb --- /dev/null +++ b/internal/service/transcoder_lifecycle_test.go @@ -0,0 +1,228 @@ +package service + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "runtime" + "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 TestHLSCompletedTranscodeStaysReadyAndIsReused(t *testing.T) { + bin, err := exec.LookPath("ffmpeg") + if err != nil { + t.Skip("ffmpeg is required for the real HLS lifecycle test") + } + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + source := filepath.Join(t.TempDir(), "short.mp4") + cmd := exec.CommandContext(ctx, bin, "-hide_banner", "-loglevel", "error", "-f", "lavfi", "-i", "testsrc=size=64x64:rate=10", "-t", "1", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-threads", "1", source) + if output, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("generate source: %v: %s", err, output) + } + cfg := &config.Config{} + cfg.App.FFmpegPath = bin + cfg.Cache.CacheDir = t.TempDir() + cfg.Transcoder.Enabled = true + cfg.Transcoder.MaxConcurrent = 1 + cfg.Transcoder.MaxHeight = 64 + cfg.Transcoder.SegmentSeconds = 1 + cfg.Transcoder.Threads = 1 + db := newServiceTestDB(t) + media := model.Media{Path: source, DurationSec: 1} + if err := db.Create(&media).Error; err != nil { + t.Fatal(err) + } + svc := NewTranscoderService(cfg, zap.NewNop(), repository.New(db), NewHub(zap.NewNop())) + t.Cleanup(svc.StopAll) + playlist, err := svc.EnsureJob(ctx, media.ID) + if err != nil { + t.Fatal(err) + } + svc.mu.Lock() + job := svc.jobs[media.ID] + svc.mu.Unlock() + if job == nil { + t.Fatal("transcode lost its job before the playlist could be read") + } + select { + case <-job.done: + case <-ctx.Done(): + t.Fatal("short transcode did not finish") + } + data, err := os.ReadFile(playlist) + if err != nil || !strings.Contains(string(data), "#EXT-X-ENDLIST") { + t.Fatalf("transcode did not produce a complete playlist: err=%v, data=%s", err, data) + } + if !svc.WaitReady(ctx, media.ID, 100*time.Millisecond) { + t.Fatal("completed HLS playlist is incorrectly reported as not ready") + } + if len(svc.Active()) != 0 { + t.Fatal("completed transcode still occupies an active job slot") + } + if _, err := svc.EnsureJob(ctx, media.ID); err != nil { + t.Fatal(err) + } + svc.mu.Lock() + reused := svc.jobs[media.ID] == job + svc.mu.Unlock() + if !reused { + t.Fatal("refreshing a completed playlist started a second transcode") + } + other := model.Media{Path: source + ".second", DurationSec: 1} + if err := os.WriteFile(other.Path, mustReadTestFile(t, source), 0o600); err != nil { + t.Fatal(err) + } + if err := db.Create(&other).Error; err != nil { + t.Fatal(err) + } + if _, err := svc.EnsureJob(ctx, other.ID); err != nil { + t.Fatalf("completed cache must not consume MaxConcurrent: %v", err) + } +} + +func mustReadTestFile(t *testing.T, path string) []byte { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + return data +} + +func TestHLSWaitReadyReturnsImmediatelyForMissingJob(t *testing.T) { + svc := NewTranscoderService(&config.Config{}, zap.NewNop(), nil, NewHub(zap.NewNop())) + started := time.Now() + if svc.WaitReady(t.Context(), "failed", time.Second) { + t.Fatal("missing job must not be ready") + } + if elapsed := time.Since(started); elapsed >= 500*time.Millisecond { + t.Fatalf("missing/failed job waited for the readiness deadline: %v", elapsed) + } +} + +func countingFFmpeg(t *testing.T) string { + t.Helper() + if runtime.GOOS == "windows" { + t.Skip("test ffmpeg uses a POSIX shell") + } + bin := filepath.Join(t.TempDir(), "ffmpeg") + script := `#!/bin/sh +printf '%s\n' "$*" >> "$0.calls" +case " $* " in + *" -encoders "*) printf ' V..... libx264\n V..... h264_nvenc\n' ;; + *" -muxers "*) printf ' E hls\n' ;; +esac +` + if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + return bin +} + +func TestFFmpegValidationIsCachedAndInvalidatedByEncoder(t *testing.T) { + bin := countingFFmpeg(t) + cfg := &config.Config{} + cfg.App.FFmpegPath = bin + svc := NewTranscoderService(cfg, zap.NewNop(), nil, NewHub(zap.NewNop())) + for range 3 { + if _, err := svc.resolveFFmpegPath(); err != nil { + t.Fatal(err) + } + } + calls := strings.Count(string(mustReadTestFile(t, bin+".calls")), "\n") + if calls != 2 { + t.Fatalf("repeated resolution spawned ffmpeg %d times, want one encoder/muxer probe pair", calls) + } + ApplyRuntimeSetting(cfg, "transcoder.hardware_accel", "true") + ApplyRuntimeSetting(cfg, "transcoder.encoder", "nvenc") + if _, err := svc.resolveFFmpegPath(); err != nil { + t.Fatal(err) + } + calls = strings.Count(string(mustReadTestFile(t, bin+".calls")), "\n") + if calls != 4 { + t.Fatalf("encoder change must revalidate capabilities, calls=%d", calls) + } + changed := time.Now().Add(time.Second) + if err := os.Chtimes(bin, changed, changed); err != nil { + t.Fatal(err) + } + if _, err := svc.resolveFFmpegPath(); err != nil { + t.Fatal(err) + } + calls = strings.Count(string(mustReadTestFile(t, bin+".calls")), "\n") + if calls != 6 { + t.Fatalf("binary replacement must invalidate validation, calls=%d", calls) + } +} + +func TestHLSIdleCleanupPreservesReplacementAndRecentActivity(t *testing.T) { + svc := NewTranscoderService(&config.Config{}, zap.NewNop(), nil, NewHub(zap.NewNop())) + done := make(chan struct{}) + close(done) + cancelled := false + old := &hlsJob{mediaID: "m", lastAccess: time.Now().Add(-time.Hour)} + current := &hlsJob{mediaID: "m", lastAccess: time.Now(), done: done, cancel: func() { cancelled = true }} + svc.jobs["m"] = current + if !svc.stopIdleJob(old, time.Minute) || cancelled || svc.jobs["m"] != current { + t.Fatal("an old watchdog must leave the replacement generation intact") + } + if svc.stopIdleJob(current, time.Minute) || cancelled { + t.Fatal("recent segment access must prevent idle cancellation") + } + current.lastAccess = time.Now().Add(-time.Hour) + if !svc.stopIdleJob(current, time.Minute) || !cancelled || svc.jobs["m"] != nil { + t.Fatal("an idle generation must release its resources") + } +} + +func TestHLSOriginalQualityValidatesSoftwareEncoder(t *testing.T) { + bin := countingFFmpeg(t) + softwareOnly := strings.ReplaceAll(string(mustReadTestFile(t, bin)), ` V..... h264_nvenc\n`, "") + if err := os.WriteFile(bin, []byte(softwareOnly), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", t.TempDir()) + t.Chdir(t.TempDir()) + cfg := &config.Config{} + cfg.App.FFmpegPath = bin + cfg.Transcoder.Enabled = true + cfg.Transcoder.HardwareAccel = true + cfg.Transcoder.Encoder = "nvenc" + cfg.Cache.CacheDir = t.TempDir() + db := newServiceTestDB(t) + source := filepath.Join(t.TempDir(), "movie.mkv") + if err := os.WriteFile(source, []byte("source"), 0o600); err != nil { + t.Fatal(err) + } + media := model.Media{Path: source, DurationSec: 60} + if err := db.Create(&media).Error; err != nil { + t.Fatal(err) + } + svc := NewTranscoderService(cfg, zap.NewNop(), repository.New(db), NewHub(zap.NewNop())) + t.Cleanup(svc.StopAll) + if _, err := svc.EnsureJobFromSubtitleQuality(t.Context(), media.ID, 0, 1, -1, "source"); err != nil { + t.Fatalf("source quality uses software and must not require NVENC: %v", err) + } + svc.mu.Lock() + job := svc.jobs[media.ID] + svc.mu.Unlock() + select { + case <-job.done: + case <-time.After(time.Second): + t.Fatal("fake ffmpeg did not finish") + } + calls := string(mustReadTestFile(t, bin+".calls")) + if strings.Count(calls, "\n") != 3 || !strings.Contains(calls, "-c:v libx264") { + t.Fatalf("startup must probe once and encode with libx264, calls=%s", calls) + } +} diff --git a/web/src/api/playback.ts b/web/src/api/playback.ts index add8ced..2138b8e 100644 --- a/web/src/api/playback.ts +++ b/web/src/api/playback.ts @@ -42,9 +42,9 @@ function publicOriginHeader() { } export const playbackAPI = { - getResume: (mediaId: string) => + getResume: (mediaId: string, signal?: AbortSignal) => api - .get<{ position_ms: number; duration_ms: number; completed: boolean }>(`/playback/${mediaId}/resume`) + .get<{ position_ms: number; duration_ms: number; completed: boolean }>(`/playback/${encodeURIComponent(mediaId)}/resume`, { signal }) .then((r) => r.data), // 片头/片尾片段：播放开始后再调用，服务端可能需要几秒去外部数据库取数， diff --git a/web/src/pages/PlayerPage.tsx b/web/src/pages/PlayerPage.tsx index c3650b2..17a1d16 100644 --- a/web/src/pages/PlayerPage.tsx +++ b/web/src/pages/PlayerPage.tsx @@ -40,6 +40,7 @@ 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 { apiErrorMessage } from './StrmManagePage' import { PlayerTopBar } from './PlayerTopBar' import { PlayerVideoStage } from './PlayerVideoStage' @@ -992,14 +993,15 @@ export function PlayerPage() { }, []) const switchToLocalHLS = useCallback( - (position = 0) => { + (position?: number) => { + pendingSeekRef.current = null setHlsSource('local') setCloudWaiting(false) setSelectedQuality((prev) => { if (playbackInfo?.local_qualities?.some((quality) => quality.id === prev)) return prev return defaultLocalQualityId(playbackInfo) }) - if (position > 2) setHlsStartSec(position) + if (position !== undefined) setHlsStartSec(Math.max(0, position)) setPlaybackMode('hls') }, [playbackInfo, setPlaybackMode], @@ -1076,10 +1078,10 @@ export function PlayerPage() { : undefined if (preferred?.source === 'cloud') { setHlsSource('cloud') + // 云 HLS 是整段时间轴，等待云转码期间也要保留目标位置。 + pendingSeekRef.current = position > 2 ? position : null if (preferred.available) { setCloudWaiting(false) - // 云 HLS 是整段时间轴：切过去后按当前位置跳一下，别让用户从头看。 - pendingSeekRef.current = position > 2 ? position : null } else { setCloudWaiting(true) setCloudWaitMessage(preferred.note || `正在等待 115 转码 ${preferred.label}…`) @@ -1144,8 +1146,9 @@ export function PlayerPage() { if (!playbackInfo || !initialHlsRequestRef.current) return if (directOnly || !mediaId) return initialHlsRequestRef.current = false - enterHlsPlayback(0, playbackInfo) - }, [directOnly, enterHlsPlayback, mediaId, playbackInfo]) + const position = Math.max(resumePosition, playbackPositionSec(mode, hlsSource, hlsStartSec, ref.current?.currentTime || 0)) + enterHlsPlayback(position, playbackInfo) + }, [directOnly, enterHlsPlayback, hlsSource, hlsStartSec, mediaId, mode, playbackInfo, resumePosition]) // 直连播放只发现外挂字幕；只有 HLS 模式需要探测可烧录的内嵌字幕。 useEffect(() => { @@ -1294,6 +1297,8 @@ export function PlayerPage() { const currentMedia = mediaRef.current if (!currentMedia) return let cancelled = false + let cleanupHlsResume: (() => void) | undefined + let onDirectMetadata: (() => void) | undefined teardownHls() const video = ref.current @@ -1310,13 +1315,14 @@ export function PlayerPage() { void import('hls.js').then(({ default: HlsCtor }) => { if (cancelled || !ref.current) return if (HlsCtor.isSupported()) { + cleanupHlsResume = bindPendingHlsSeek(video, pendingSeekRef) const hls = new HlsCtor({ enableWorker: true, lowLatencyMode: false, // The server waits up to 45s for the first segment. Slow two-core // hosts and remote STRM sources regularly need more than hls.js's // 10s default, which otherwise aborts a healthy transcode. - manifestLoadingTimeOut: 60_000, + manifestLoadingTimeOut: hlsManifestTimeoutMs(hlsSource, hlsStartSec), // 115 云 HLS 的 master.m3u8 每次都要现调 115 接口（实测 0.5–1.3s）， // 还可能撞上 409「正在转码」。只重试 1 次的话，一次抖动就判死刑， // 直接掉到本地转码。 @@ -1330,15 +1336,7 @@ export function PlayerPage() { hls.loadSource(url) hls.attachMedia(video) hls.on(HlsCtor.Events.MANIFEST_PARSED, () => { - if (pendingSeekRef.current !== null) { - const target = pendingSeekRef.current - pendingSeekRef.current = null - try { - video.currentTime = target - } catch { - // ignore - } - } + applyPendingHlsSeek(video, pendingSeekRef) void video.play().catch(() => undefined) // .strm 入库时常缺 duration；转码启动时会补探测，这里刷新一次给进度条总时长。 if (durationSec > 0) return @@ -1388,8 +1386,7 @@ export function PlayerPage() { } catch { // ignore } - video.src = url - void video.play().catch(() => undefined) + cleanupHlsResume = attachNativeHls(video, url, pendingSeekRef) } else { setHlsUnavailable(true) showPlayFailure('当前浏览器不支持 HLS，正在尝试直接播放。') @@ -1421,23 +1418,20 @@ export function PlayerPage() { ) pendingDirectStartRef.current = 0 if (resumeAt > 0) { - video.addEventListener( - 'loadedmetadata', - () => { - // STRM/115 直链要等 seekable 覆盖目标位置再写 currentTime（否则会被 - // 钳回 0），并顺带打开误报宽限期，避免这次跳转被当成播放失败又弹回 HLS。 - if (isStrmMedia(currentMedia)) { - seekStrmDirectTo(resumeAt) - return - } - try { - video.currentTime = resumeAt - } catch { - // ignore - } - }, - { once: true }, - ) + onDirectMetadata = () => { + // STRM/115 直链要等 seekable 覆盖目标位置再写 currentTime（否则会被 + // 钳回 0），并顺带打开误报宽限期，避免这次跳转被当成播放失败又弹回 HLS。 + if (isStrmMedia(currentMedia)) { + seekStrmDirectTo(resumeAt) + return + } + try { + video.currentTime = resumeAt + } catch { + // ignore + } + } + video.addEventListener('loadedmetadata', onDirectMetadata, { once: true }) } video.src = url void video.play().catch(() => undefined) @@ -1457,6 +1451,8 @@ export function PlayerPage() { return () => { cancelled = true video.removeEventListener('playing', onPlaying) + if (onDirectMetadata) video.removeEventListener('loadedmetadata', onDirectMetadata) + cleanupHlsResume?.() teardownHls() } }, [ @@ -1505,19 +1501,22 @@ export function PlayerPage() { if (!id) return setResumePosition(0) setInitialSeekDone(false) + const controller = new AbortController() playbackAPI - .getResume(id) + .getResume(id, controller.signal) .then((progress) => { + if (controller.signal.aborted) return if (progress.position_ms > 2000 && !progress.completed) { setResumePosition(progress.position_ms / 1000) } }) .catch(() => undefined) + return () => controller.abort() }, [id]) useEffect(() => { if (!resumePosition || initialSeekDone) return - if (mode === 'hls') { + if (mode === 'hls' && hlsSource === 'local') { // Restart transcode near the resume point instead of seeking a short partial playlist. if (Math.abs(hlsStartSec - resumePosition) > 2) { setHlsStartSec(resumePosition) @@ -1536,6 +1535,7 @@ export function PlayerPage() { let seekStarted = false let onPlaying: (() => void) | undefined let onCanPlay: (() => void) | undefined + let onSeeked: (() => void) | undefined const notifyResume = () => { const m = Math.floor(resumePosition / 60) @@ -1552,7 +1552,7 @@ export function PlayerPage() { return } seekStarted = true - if (isStrmMedia(mediaRef.current)) { + if (mode === 'direct' && isStrmMedia(mediaRef.current)) { seekStrmDirectTo(resumePosition, { timeoutMs: 6000, onSettled: () => { @@ -1579,7 +1579,8 @@ export function PlayerPage() { directSeekGraceUntilRef.current = Date.now() + DIRECT_SEEK_GRACE_MS directRetryRef.current = false clearFallbackTimer() - const onSeeked = () => { + onSeeked = () => { + if (cancelled) return setInitialSeekDone(true) notifyResume() } @@ -1592,7 +1593,7 @@ export function PlayerPage() { } // STRM/云盘：等真正起播后再续播跳转，避免 302 未完成就 seek。 - const deferForStrm = isStrmMedia(mediaRef.current) + const deferForStrm = mode === 'direct' && isStrmMedia(mediaRef.current) if (deferForStrm) { onPlaying = () => { performResumeSeek() @@ -1615,8 +1616,9 @@ export function PlayerPage() { cancelled = true if (onPlaying) video.removeEventListener('playing', onPlaying) if (onCanPlay) video.removeEventListener('canplay', onCanPlay) + if (onSeeked) video.removeEventListener('seeked', onSeeked) } - }, [resumePosition, initialSeekDone, mode, hlsStartSec, clearFallbackTimer, seekStrmDirectTo]) + }, [resumePosition, initialSeekDone, mode, hlsSource, hlsStartSec, clearFallbackTimer, seekStrmDirectTo]) // 使用 ref 实时同步进度计算所需的状态，避免每次 hlsStartSec 改变都触发 cleanup 并误上报旧进度 const hlsStartSecRef = useRef(hlsStartSec) @@ -2223,28 +2225,22 @@ export function PlayerPage() { toast.error('图片字幕需要开启 HLS 转码后才能显示') return } + const position = playbackPositionSec(mode, hlsSource, hlsStartSec, ref.current?.currentTime || 0) if (nextTrack?.delivery === 'burn') { // PGS 等图片字幕必须走本地 FFmpeg 烧录，不能交给云端 HLS。 - switchToLocalHLS(ref.current?.currentTime || 0) - } - const burnChanged = - oldTrack?.delivery === 'burn' || nextTrack?.delivery === 'burn' - if (burnChanged && mode === 'hls' && ref.current) { - setHlsStartSec(hlsStartSec + (ref.current.currentTime || 0)) + switchToLocalHLS(position) + } else if (oldTrack?.delivery === 'burn' && mode === 'hls') { + setHlsStartSec(position) } setSubtitleIndex(index) - if (nextTrack?.delivery === 'burn' && mode !== 'hls' && !directOnly && !isDirectStream) { - setHlsStartSec(ref.current?.currentTime || 0) - setPlaybackMode('hls') - } - }, [directOnly, hlsStartSec, isDirectStream, mode, setPlaybackMode, subs, subtitleIndex, switchToLocalHLS]) + }, [directOnly, hlsSource, hlsStartSec, isDirectStream, mode, subs, subtitleIndex, switchToLocalHLS]) const selectPlaybackQuality = useCallback( (quality: PlaybackQuality) => { const video = ref.current const localTime = video?.currentTime || 0 // 云 HLS 的 currentTime 本来就在绝对时间轴上；本地 HLS 的 t=0 对应 hlsStartSec。 - const position = mode === 'hls' && hlsSource === 'local' ? hlsStartSec + localTime : localTime + const position = playbackPositionSec(mode, hlsSource, hlsStartSec, localTime) if (quality.source === 'original') { setCloudWaiting(false) setHlsSource('local') @@ -2256,12 +2252,13 @@ export function PlayerPage() { if (quality.source === 'cloud') { setSelectedQuality(quality.id) setHlsSource('cloud') + setHlsStartSec(position) + pendingSeekRef.current = position > 2 ? position : null // 用户手动选了云端档位：这是一次明确的重新尝试，重置自动重连的次数窗口。 cloudReloadsRef.current = [] if (quality.available) { setCloudWaiting(false) setCloudWaitMessage('') - if (position > 2) pendingSeekRef.current = position setPlaybackMode('hls') } else { setCloudWaiting(true) @@ -2281,7 +2278,8 @@ export function PlayerPage() { toast.error(quality.note || '当前无法进行 HLS 转码，请在「系统设置 → 常规」安装 ffmpeg 后重试') return } - if (position > 2) setHlsStartSec(position) + pendingSeekRef.current = null + setHlsStartSec(position) setPlaybackMode('hls') }, [hlsSource, hlsStartSec, mode, setPlaybackMode, startCloudTranscode], diff --git a/web/src/pages/hlsPlayback.test.ts b/web/src/pages/hlsPlayback.test.ts new file mode 100644 index 0000000..25606b0 --- /dev/null +++ b/web/src/pages/hlsPlayback.test.ts @@ -0,0 +1,58 @@ +import { strict as assert } from 'node:assert' +import { applyPendingHlsSeek, attachNativeHls, bindPendingHlsSeek, hlsManifestTimeoutMs, playbackPositionSec } from './hlsPlayback.ts' + +class FakeVideo extends EventTarget { + src = '' + rejectSeek = false + position = 0 + playCalls = 0 + get currentTime() { return this.position } + set currentTime(value: number) { + if (this.rejectSeek) throw new Error('metadata not ready') + this.position = value + } + play() { this.playCalls += 1; return Promise.resolve() } +} + +const video = new FakeVideo() +const pending = { current: 420 as number | null } +const detach = attachNativeHls(video, '/cloud/master.m3u8', pending) +assert.equal(video.src, '/cloud/master.m3u8') +assert.equal(video.playCalls, 1) +assert.equal(video.currentTime, 0, 'seek must wait for the new source metadata') +video.rejectSeek = true +video.dispatchEvent(new Event('loadedmetadata')) +assert.equal(pending.current, 420, 'a rejected seek must survive until canplay') +video.rejectSeek = false +video.dispatchEvent(new Event('canplay')) +assert.equal(video.currentTime, 420, 'native HLS must restore the absolute cloud position') +assert.equal(pending.current, null) +pending.current = 600 +detach() +video.dispatchEvent(new Event('loadedmetadata')) +assert.equal(video.currentTime, 420, 'a detached source must not apply late metadata events') +assert.equal(pending.current, 600) + +const msePending = { current: 120 as number | null } +video.rejectSeek = true +assert.equal(applyPendingHlsSeek(video, msePending), false) +assert.equal(msePending.current, 120) +video.rejectSeek = false +assert.equal(applyPendingHlsSeek(video, msePending), true) +assert.equal(msePending.current, null) +const lateMsePending = { current: 900 as number | null } +video.rejectSeek = true +const detachMse = bindPendingHlsSeek(video, lateMsePending) +assert.equal(applyPendingHlsSeek(video, lateMsePending), false) +video.rejectSeek = false +video.dispatchEvent(new Event('loadedmetadata')) +assert.equal(video.currentTime, 900, 'hls.js seeks rejected during manifest parsing must retry after metadata') +assert.equal(lateMsePending.current, null) +detachMse() +assert.ok(hlsManifestTimeoutMs('local', 1800) > 120_000, 'deep seeks must outlast the server readiness deadline') +assert.equal(hlsManifestTimeoutMs('local', 0), 60_000) +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) +console.log('hlsPlayback.test.ts ok') diff --git a/web/src/pages/hlsPlayback.ts b/web/src/pages/hlsPlayback.ts new file mode 100644 index 0000000..a73f3a8 --- /dev/null +++ b/web/src/pages/hlsPlayback.ts @@ -0,0 +1,43 @@ +type HlsVideo = Pick +type PendingSeek = { current: number | null } + +/** Local mid-file restarts may wait 120 seconds on the server before a playlist exists. */ +export function hlsManifestTimeoutMs(source: 'cloud' | 'local', startSec: number): number { + return source === 'local' && startSec > 0.05 ? 135_000 : 60_000 +} + +/** Keep a pending seek until the media element accepts it; metadata can arrive later. */ +export function applyPendingHlsSeek(video: Pick, pending: PendingSeek): boolean { + if (pending.current === null) return true + if (!Number.isFinite(pending.current) || pending.current < 0) return false + try { + video.currentTime = pending.current + pending.current = null + return true + } catch { + return false + } +} + +export function bindPendingHlsSeek(video: HlsVideo, pending: PendingSeek): () => void { + const restore = () => { applyPendingHlsSeek(video, pending) } + video.addEventListener('loadedmetadata', restore) + video.addEventListener('canplay', restore) + return () => { + video.removeEventListener('loadedmetadata', restore) + video.removeEventListener('canplay', restore) + } +} + +/** Native HLS needs the same seek restoration as hls.js, with removable listeners. */ +export function attachNativeHls(video: HlsVideo, url: string, pending: PendingSeek): () => void { + const detach = bindPendingHlsSeek(video, pending) + video.src = url + void video.play().catch(() => undefined) + return detach +} + +/** Only locally transcoded HLS uses a relative timeline starting at startSec. */ +export function playbackPositionSec(mode: 'direct' | 'hls', source: 'cloud' | 'local', startSec: number, currentTime: number): number { + return mode === 'hls' && source === 'local' ? startSec + currentTime : currentTime +}