feat: wire source cache and manifest cache into playback pipeline

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,

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 != "" {
+	cacheKey := manifestCacheKey(targetURL, referer)
-		req.Header.Set("Range", rangeHeader)
+	if h.writeCachedPlaylist(c, cacheKey, targetURL, referer) {
-	}
+		return
-	if ifRangeHeader := c.GetHeader("If-Range"); ifRangeHeader != "" {
-		req.Header.Set("If-Range", ifRangeHeader)
 	}
+	copyProxyRangeHeaders(req, c)
 
 	resp, err := h.streamingClient.Do(req)
 	if err != nil {
-		if !errors.Is(err, context.Canceled) {
+		h.handleStreamRequestError(c, err)
-			observability.ErrorContext(c.Request.Context(), "proxy_stream_upstream_failed", "playback", "", map[string]any{"target_url": targetURL}, err)
-			recordPrivateGinError(c, err)
-		}
-		c.Status(http.StatusBadGateway)
 		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) {
+func readBoundedPlaylist(body io.Reader) ([]byte, error) {
-	body, err := io.ReadAll(io.LimitReader(resp.Body, netutil.MiB2))
+	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)
+		return nil, err
-		recordPrivateGinError(c, err)
-		c.Status(http.StatusBadGateway)
-		return
 	}
+	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)
+		safeErr := errors.New("stream playlist rewrite failed")
-		recordPrivateGinError(c, err)
+		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))
 }

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())
-}

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) {
+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) {
-	if res := s.resolveStreamResult(ctx, animeID, searchTitles, episode, requestedMode); res != nil {
+	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,15 +204,77 @@ 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 {
+func (s *playbackService) resolveStreamResult(ctx context.Context, animeID int, searchTitles []string, episode string, mode string, allowStale bool, forceRefresh bool) *domain.StreamResult {
-	for _, p := range s.providers {
+	key := newSourceCacheKey(animeID, episode, mode)
-		res, err := p.GetStreams(ctx, animeID, searchTitles, episode, mode)
+	stale, state := s.sourceCache.get(key, time.Now())
-		if err == nil && res != nil {
+	if !forceRefresh && state == sourceCacheFresh {
-			return res
+		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 {
@@ -201,7 +282,7 @@ func (s *playbackService) buildModeSource(res *domain.StreamResult) domain.ModeS
 	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,