From 0abee703a1ba14035e2c6b44d71cfce15cd2a279 Mon Sep 17 00:00:00 2001 From: mkelvers Date: Sun, 5 Jul 2026 23:08:39 +0200 Subject: [PATCH] feat: wire source cache and manifest cache into playback pipeline --- internal/playback/handler/handler.go | 8 +- internal/playback/handler/proxy_stream.go | 94 ++++++++++++++----- internal/playback/service.go | 32 +------ internal/playback/watch_data.go | 107 +++++++++++++++++++--- 4 files changed, 175 insertions(+), 66 deletions(-) diff --git a/internal/playback/handler/handler.go b/internal/playback/handler/handler.go index 7a366566..f819d72a 100644 --- a/internal/playback/handler/handler.go +++ b/internal/playback/handler/handler.go @@ -22,6 +22,7 @@ type PlaybackHandler struct { proxyClient *http.Client streamingClient *http.Client subtitleCache *subtitleCache + manifestCache *manifestCache } func NewPlaybackHandler(svc domain.PlaybackService, animeSvc domain.AnimePlaybackService) *PlaybackHandler { @@ -31,6 +32,7 @@ func NewPlaybackHandler(svc domain.PlaybackService, animeSvc domain.AnimePlaybac proxyClient: proxytarget.NewClient(60 * time.Second), streamingClient: proxytarget.NewStreamingClient(), subtitleCache: newSubtitleCache(10*time.Minute, 256), + manifestCache: newManifestCache(manifestCacheMaxEntries), } } @@ -120,10 +122,14 @@ func (h *PlaybackHandler) HandleEpisodeData(c *gin.Context) { } mode := c.DefaultQuery("mode", "sub") + ctx := c.Request.Context() + if c.Query("refresh") == "1" { + ctx = domain.WithPlaybackSourceRefresh(ctx) + } userID := server.CurrentUserID(c) - data, err := h.svc.BuildWatchData(c.Request.Context(), animeID, []string{}, episode, mode, userID) + data, err := h.svc.BuildWatchData(ctx, animeID, []string{}, episode, mode, userID) if err != nil { server.RespondError( c, diff --git a/internal/playback/handler/proxy_stream.go b/internal/playback/handler/proxy_stream.go index 74ffa19e..1e0a0992 100644 --- a/internal/playback/handler/proxy_stream.go +++ b/internal/playback/handler/proxy_stream.go @@ -8,6 +8,7 @@ import ( errlog "mal/pkg" netutil "mal/pkg/net" "net/http" + "time" "github.com/gin-gonic/gin" ) @@ -17,68 +18,111 @@ func (h *PlaybackHandler) HandleProxyStream(c *gin.Context) { if !ok { return } - req, err := newProxyRequest(c.Request.Context(), targetURL, referer) if err != nil { c.Status(http.StatusBadGateway) return } - if rangeHeader := c.GetHeader("Range"); rangeHeader != "" { - req.Header.Set("Range", rangeHeader) - } - if ifRangeHeader := c.GetHeader("If-Range"); ifRangeHeader != "" { - req.Header.Set("If-Range", ifRangeHeader) + cacheKey := manifestCacheKey(targetURL, referer) + if h.writeCachedPlaylist(c, cacheKey, targetURL, referer) { + return } + copyProxyRangeHeaders(req, c) resp, err := h.streamingClient.Do(req) if err != nil { - if !errors.Is(err, context.Canceled) { - observability.ErrorContext(c.Request.Context(), "proxy_stream_upstream_failed", "playback", "", map[string]any{"target_url": targetURL}, err) - recordPrivateGinError(c, err) - } - c.Status(http.StatusBadGateway) + h.handleStreamRequestError(c, err) return } defer func() { errlog.Log("failed to close proxy stream response body", resp.Body.Close()) }() + h.writeStreamResponse(c, resp, cacheKey, targetURL, referer) +} + +func (h *PlaybackHandler) writeCachedPlaylist(c *gin.Context, cacheKey string, targetURL string, referer string) bool { + if c.GetHeader("Range") != "" { + return false + } + cached, found := h.manifestCache.get(cacheKey, time.Now()) + if !found { + return false + } + observability.Info("playback_manifest_cache_hit", "playback", "", nil) + h.writeProxyPlaylist(c, cached.status, cached.headers, cached.body, targetURL, referer) + return true +} + +func copyProxyRangeHeaders(req *http.Request, c *gin.Context) { + if rangeHeader := c.GetHeader("Range"); rangeHeader != "" { + req.Header.Set("Range", rangeHeader) + } + if ifRangeHeader := c.GetHeader("If-Range"); ifRangeHeader != "" { + req.Header.Set("If-Range", ifRangeHeader) + } +} + +func (h *PlaybackHandler) handleStreamRequestError(c *gin.Context, err error) { + if !errors.Is(err, context.Canceled) { + safeErr := errors.New("stream upstream request failed") + observability.ErrorContext(c.Request.Context(), "proxy_stream_upstream_failed", "playback", "", nil, safeErr) + recordPrivateGinError(c, safeErr) + } + c.Status(http.StatusBadGateway) +} + +func (h *PlaybackHandler) writeStreamResponse(c *gin.Context, resp *http.Response, cacheKey string, targetURL string, referer string) { if isHLSPlaylistResponse(targetURL, resp.Header) { - h.writeProxyPlaylist(c, resp, targetURL, referer) + observability.Info("playback_manifest_cache_miss", "playback", "", nil) + body, readErr := readBoundedPlaylist(resp.Body) + if readErr != nil { + safeErr := errors.New("stream playlist read failed") + observability.ErrorContext(c.Request.Context(), "proxy_stream_playlist_read_failed", "playback", "", nil, safeErr) + recordPrivateGinError(c, safeErr) + c.Status(http.StatusBadGateway) + return + } + h.manifestCache.set(cacheKey, resp.StatusCode, resp.Header, body, time.Now()) + h.writeProxyPlaylist(c, resp.StatusCode, resp.Header, body, targetURL, referer) return } copyProxyHeaders(c.Writer.Header(), resp.Header) c.Status(resp.StatusCode) - copyProxyResponseBody(c, resp.Body, targetURL) + copyProxyResponseBody(c, resp.Body) } -func copyProxyResponseBody(c *gin.Context, body io.Reader, targetURL string) { +func copyProxyResponseBody(c *gin.Context, body io.Reader) { n, err := io.Copy(c.Writer, body) if err == nil || errors.Is(err, context.Canceled) || c.Request.Context().Err() != nil { return } - observability.WarnContext(c.Request.Context(), "proxy_stream_copy_failed", "playback", "", map[string]any{"target_url": targetURL, "bytes_copied": n}, err) + observability.WarnContext(c.Request.Context(), "proxy_stream_copy_failed", "playback", "", map[string]any{"bytes_copied": n}, err) } -func (h *PlaybackHandler) writeProxyPlaylist(c *gin.Context, resp *http.Response, targetURL string, referer string) { - body, err := io.ReadAll(io.LimitReader(resp.Body, netutil.MiB2)) +func readBoundedPlaylist(body io.Reader) ([]byte, error) { + data, err := io.ReadAll(io.LimitReader(body, netutil.MiB2+1)) if err != nil { - observability.ErrorContext(c.Request.Context(), "proxy_stream_playlist_read_failed", "playback", "", map[string]any{"target_url": targetURL}, err) - recordPrivateGinError(c, err) - c.Status(http.StatusBadGateway) - return + return nil, err } + if len(data) > manifestCacheMaxBodySize { + return nil, errors.New("upstream playlist exceeds size limit") + } + return data, nil +} +func (h *PlaybackHandler) writeProxyPlaylist(c *gin.Context, status int, headers http.Header, body []byte, targetURL string, referer string) { rewritten, err := h.rewriteHLSPlaylist(string(body), targetURL, referer) if err != nil { - observability.ErrorContext(c.Request.Context(), "proxy_stream_playlist_rewrite_failed", "playback", "", map[string]any{"target_url": targetURL}, err) - recordPrivateGinError(c, err) + safeErr := errors.New("stream playlist rewrite failed") + observability.ErrorContext(c.Request.Context(), "proxy_stream_playlist_rewrite_failed", "playback", "", nil, safeErr) + recordPrivateGinError(c, safeErr) c.Status(http.StatusBadGateway) return } - copyProxyHeaders(c.Writer.Header(), resp.Header) + copyProxyHeaders(c.Writer.Header(), headers) c.Writer.Header().Del("Content-Length") - c.Data(resp.StatusCode, "application/vnd.apple.mpegurl", []byte(rewritten)) + c.Data(status, "application/vnd.apple.mpegurl", []byte(rewritten)) } diff --git a/internal/playback/service.go b/internal/playback/service.go index 1703261b..c1dffdda 100644 --- a/internal/playback/service.go +++ b/internal/playback/service.go @@ -2,16 +2,15 @@ package playback import ( - "context" "fmt" "mal/integrations/jikan" "mal/internal/domain" - "mal/internal/observability" "mal/internal/playback/proxytarget" - errlog "mal/pkg" netutil "mal/pkg/net" "net/http" "time" + + "golang.org/x/sync/singleflight" ) type playbackService struct { @@ -22,6 +21,8 @@ type playbackService struct { httpClient *http.Client proxyTokenKey string proxyTokens *proxyTokenStore + sourceCache *sourceCache + sourceFlight singleflight.Group auditSvc domain.AuditService } @@ -37,6 +38,7 @@ func NewPlaybackService(repo domain.PlaybackRepository, providers []domain.Provi httpClient: netutil.NewClient(), proxyTokenKey: string(proxyTokenKey), proxyTokens: newProxyTokenStore(), + sourceCache: newSourceCache(defaultSourceCacheTTL, defaultSourceCacheStaleTTL, defaultSourceCacheMaxEntries), } } @@ -63,27 +65,3 @@ func (s *playbackService) ResolveProxyToken(token string, scope string) (string, } return target.targetURL, target.referer, nil } - -func (s *playbackService) warmStreamURL(targetURL, referer string) { - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - - req, err := http.NewRequestWithContext(ctx, http.MethodGet, targetURL, nil) - if err != nil { - return - } - if referer != "" { - req.Header.Set("Referer", referer) - } - req.Header.Set("User-Agent", netutil.Firefox121) - - resp, err := s.httpClient.Do(req) - if err != nil { - if resp != nil { - errlog.Close(resp.Body, "failed to close warm stream error response body") - } - observability.LogJSON(observability.LogLevelWarn, "warm_stream_failed", "playback", err.Error(), map[string]any{"url": targetURL}, nil) - return - } - errlog.Log("failed to close warm stream response body", resp.Body.Close()) -} diff --git a/internal/playback/watch_data.go b/internal/playback/watch_data.go index 4f66ed9d..9a434f44 100644 --- a/internal/playback/watch_data.go +++ b/internal/playback/watch_data.go @@ -2,6 +2,7 @@ package playback import ( "context" + "errors" "fmt" "sort" "strconv" @@ -13,6 +14,16 @@ import ( "mal/internal/observability" ) +const sourceResolutionTimeout = 15 * time.Second + +type sourceResolutionResult struct { + result *domain.StreamResult + err error + duration time.Duration + shared bool + completed bool +} + const episodeAvailabilityUncertainWarning = "Episode availability may be incomplete or out of date. Continue only if you understand that the episode list and audio availability may be uncertain." func (s *playbackService) BuildWatchData(ctx context.Context, animeID int, titleCandidates []string, episode string, mode string, userID string) (domain.WatchPageData, error) { @@ -36,7 +47,15 @@ func (s *playbackService) BuildWatchData(ctx context.Context, animeID int, title // mode fallback mode, from := resolveMode(episode, mode, eps.Episodes) - modeSources, result, resolvedMode, switchedFrom := s.resolveModeSources(ctx, animeID, searchTitles, episode, mode) + modeSources, result, resolvedMode, switchedFrom := s.resolveModeSources( + ctx, + animeID, + searchTitles, + episode, + mode, + !anime.Airing, + domain.PlaybackSourceRefreshRequested(ctx), + ) if resolvedMode != "" { mode = resolvedMode } @@ -62,7 +81,6 @@ func (s *playbackService) BuildWatchData(ctx context.Context, animeID int, title return pageData, fmt.Errorf("no streams found for mode %s", mode) } - go s.warmStreamURL(result.URL, result.Referer) return pageData, nil } @@ -165,15 +183,16 @@ func resolveMode(episode string, requestedMode string, episodes []domain.Canonic return requestedMode, "" } -func (s *playbackService) resolveModeSources(ctx context.Context, animeID int, searchTitles []string, episode string, requestedMode string) (map[string]domain.ModeSource, *domain.StreamResult, string, string) { - if res := s.resolveStreamResult(ctx, animeID, searchTitles, episode, requestedMode); res != nil { +func (s *playbackService) resolveModeSources(ctx context.Context, animeID int, searchTitles []string, episode string, requestedMode string, allowStale bool, forceRefresh bool) (map[string]domain.ModeSource, *domain.StreamResult, string, string) { + requestedMode = normalizeSourceMode(requestedMode) + if res := s.resolveStreamResult(ctx, animeID, searchTitles, episode, requestedMode, allowStale, forceRefresh); res != nil { return map[string]domain.ModeSource{ requestedMode: s.buildModeSource(res), }, res, requestedMode, "" } for _, fallbackMode := range fallbackModes(requestedMode) { - res := s.resolveStreamResult(ctx, animeID, searchTitles, episode, fallbackMode) + res := s.resolveStreamResult(ctx, animeID, searchTitles, episode, fallbackMode, allowStale, forceRefresh) if res == nil { continue } @@ -185,23 +204,85 @@ func (s *playbackService) resolveModeSources(ctx context.Context, animeID int, s return map[string]domain.ModeSource{}, nil, requestedMode, "" } -func (s *playbackService) resolveStreamResult(ctx context.Context, animeID int, searchTitles []string, episode string, mode string) *domain.StreamResult { - for _, p := range s.providers { - res, err := p.GetStreams(ctx, animeID, searchTitles, episode, mode) - if err == nil && res != nil { - return res - } +func (s *playbackService) resolveStreamResult(ctx context.Context, animeID int, searchTitles []string, episode string, mode string, allowStale bool, forceRefresh bool) *domain.StreamResult { + key := newSourceCacheKey(animeID, episode, mode) + stale, state := s.sourceCache.get(key, time.Now()) + if !forceRefresh && state == sourceCacheFresh { + observability.Info("playback_source_cache_hit", "playback", "", map[string]any{"anime_id": animeID, "episode": episode, "mode": key.mode}) + return stale } + observability.Info("playback_source_cache_miss", "playback", "", map[string]any{"anime_id": animeID, "episode": episode, "mode": key.mode, "forced": forceRefresh}) + resolved := s.waitForSourceResult(ctx, key, animeID, searchTitles, forceRefresh) + if !resolved.completed { + return nil + } + observability.Info("playback_source_resolution", "playback", "", map[string]any{ + "anime_id": animeID, "episode": episode, "mode": key.mode, + "duration_ms": resolved.duration.Milliseconds(), "shared": resolved.shared, + }) + if resolved.err == nil { + return cloneStreamResult(resolved.result) + } + if allowStale && state == sourceCacheStale && stale != nil { + observability.Warn("playback_source_cache_stale_hit", "playback", "", map[string]any{"anime_id": animeID, "episode": episode, "mode": key.mode}, errors.New("provider source refresh failed")) + return stale + } + observability.Warn("playback_source_resolution_failed", "playback", "", map[string]any{"anime_id": animeID, "episode": episode, "mode": key.mode}, errors.New("provider source resolution failed")) return nil } +func (s *playbackService) waitForSourceResult(ctx context.Context, key sourceCacheKey, animeID int, searchTitles []string, forceRefresh bool) sourceResolutionResult { + startedAt := time.Now() + resultCh := s.sourceFlight.DoChan(key.flightKey(), func() (any, error) { + return s.resolveSource(key, animeID, searchTitles, forceRefresh) + }) + select { + case <-ctx.Done(): + return sourceResolutionResult{err: ctx.Err(), duration: time.Since(startedAt)} + case resolved := <-resultCh: + result, _ := resolved.Val.(*domain.StreamResult) + return sourceResolutionResult{ + result: result, err: resolved.Err, shared: resolved.Shared, + duration: time.Since(startedAt), completed: true, + } + } +} + +func (s *playbackService) resolveSource(key sourceCacheKey, animeID int, searchTitles []string, forceRefresh bool) (*domain.StreamResult, error) { + if !forceRefresh { + if cached, state := s.sourceCache.get(key, time.Now()); state == sourceCacheFresh { + return cached, nil + } + } + resolveCtx, cancel := context.WithTimeout(context.Background(), sourceResolutionTimeout) + defer cancel() + + var lastErr error + for _, provider := range s.providers { + result, err := provider.GetStreams(resolveCtx, animeID, searchTitles, key.episode, key.mode) + if err == nil && result != nil { + if s.sourceCache.set(key, result, time.Now()) { + observability.Info("playback_source_cache_eviction", "playback", "", nil) + } + return cloneStreamResult(result), nil + } + if err != nil { + lastErr = err + } + } + if lastErr == nil { + lastErr = errors.New("no provider returned a stream") + } + return nil, lastErr +} + func (s *playbackService) buildModeSource(res *domain.StreamResult) domain.ModeSource { subtitles := make([]domain.SubtitleItem, 0, len(res.Subtitles)) for _, sub := range res.Subtitles { token, err := s.SignProxyToken(sub.URL, res.Referer, "subtitle") if err != nil { - observability.LogJSON(observability.LogLevelWarn, "sign_subtitle_token_failed", "playback", err.Error(), map[string]any{"url": sub.URL}, nil) + observability.Warn("sign_subtitle_token_failed", "playback", "", nil, err) } subtitles = append(subtitles, domain.SubtitleItem{ Lang: sub.Label, @@ -211,7 +292,7 @@ func (s *playbackService) buildModeSource(res *domain.StreamResult) domain.ModeS streamToken, err := s.SignProxyToken(res.URL, res.Referer, "stream") if err != nil { - observability.LogJSON(observability.LogLevelWarn, "sign_stream_token_failed", "playback", err.Error(), map[string]any{"url": res.URL}, nil) + observability.Warn("sign_stream_token_failed", "playback", "", nil, err) } return domain.ModeSource{ Token: streamToken,