Skip to content

Commit f9d8950

Browse files
committed
feat: enhance search timeout handling and add origin probe for better diagnostics; improve SuperFlix error handling for dead streams
1 parent d7e6233 commit f9d8950

10 files changed

Lines changed: 467 additions & 34 deletions

File tree

internal/api/providers/dispatch.go

Lines changed: 63 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -23,20 +23,39 @@ var searchBreaker = netx.NewCircuitBreaker()
2323
// Aggregate search timing. searchAllTimeout is the hard ceiling for the whole
2424
// fan-out; stragglerGrace bounds how long we keep waiting for the remaining
2525
// sources once the first results are in — so a fast source isn't held hostage
26-
// by a slow one. Mirrors the ScraperManager engine's budgets. Per-source
27-
// breaker/timeout/tagging is handled inside each Source.Search (searchViaManager).
26+
// by a slow one. Mirrors the ScraperManager engine's budgets.
2827
const (
2928
searchAllTimeout = 15 * time.Second
3029
stragglerGrace = 5 * time.Second
30+
// originProbeBudget bounds the disambiguation HEAD issued after a per-source
31+
// search deadline (see searchOneWithTimeout / netx.EnrichTimeoutWithProbe).
32+
originProbeBudget = 3 * time.Second
3133
)
3234

35+
// perSourceSearchTimeout caps a single source's search so a wedged adapter
36+
// (the underlying clients still issue non-cancelable http.NewRequest calls)
37+
// cannot hold the fan-out open until searchAllTimeout and, more importantly,
38+
// trips that source's circuit breaker with an accurate, probe-enriched
39+
// diagnostic instead of a generic aggregate timeout. A var, not a const, so
40+
// tests can shorten it.
41+
var perSourceSearchTimeout = 12 * time.Second
42+
3343
// searchOne is a single source's search outcome in the fan-out.
3444
type searchOne struct {
3545
kind source.SourceKind
3646
results []*models.Anime
3747
err error
3848
}
3949

50+
// activeSearcher is a source selected for the fan-out: its Searchable behavior,
51+
// its kind (for breaker keying and display), and its optional homepage probe
52+
// URL (empty for opaque/browser-gated sources).
53+
type activeSearcher struct {
54+
sr source.Searchable
55+
kind source.SourceKind
56+
probeURL string
57+
}
58+
4059
// init wires SearchAll into the api package's search seam so
4160
// api.SearchAnimeEnhanced dispatches through the Model B registry without
4261
// importing providers (which would cycle).
@@ -68,14 +87,14 @@ func SearchAll(ctx context.Context, query string, kinds ...source.SourceKind) ([
6887
want[k] = true
6988
}
7089

71-
var searchers []source.Searchable
72-
var names []source.SourceKind
90+
var searchers []activeSearcher
7391
for _, s := range source.ActiveSources() {
7492
sr, ok := s.(source.Searchable)
7593
if !ok {
7694
continue
7795
}
78-
kind := s.Describe().Kind
96+
desc := s.Describe()
97+
kind := desc.Kind
7998
if len(want) > 0 && !want[kind] {
8099
continue
81100
}
@@ -85,8 +104,7 @@ func SearchAll(ctx context.Context, query string, kinds ...source.SourceKind) ([
85104
util.Warn("search source skipped (circuit open)", "source", kind, "retry_after", retry.Round(time.Second), "diagnostic", diag.UserMessage())
86105
continue
87106
}
88-
searchers = append(searchers, sr)
89-
names = append(names, kind)
107+
searchers = append(searchers, activeSearcher{sr: sr, kind: kind, probeURL: desc.ProbeURL})
90108
}
91109
if len(searchers) == 0 {
92110
return nil, fmt.Errorf("no searchable source available for query %q", query)
@@ -97,13 +115,12 @@ func SearchAll(ctx context.Context, query string, kinds ...source.SourceKind) ([
97115

98116
resultChan := make(chan searchOne, len(searchers))
99117
var wg sync.WaitGroup
100-
for i, sr := range searchers {
118+
for _, a := range searchers {
101119
wg.Add(1)
102-
go func(sr source.Searchable, kind source.SourceKind) {
120+
go func(a activeSearcher) {
103121
defer wg.Done()
104-
res, err := sr.Search(ctx, query)
105-
resultChan <- searchOne{kind: kind, results: res, err: err}
106-
}(sr, names[i])
122+
resultChan <- searchOneWithTimeout(ctx, a, query)
123+
}(a)
107124
}
108125
go func() { wg.Wait(); close(resultChan) }()
109126

@@ -146,6 +163,40 @@ func SearchAll(ctx context.Context, query string, kinds ...source.SourceKind) ([
146163
}
147164
}
148165

166+
// searchOneWithTimeout runs a single source's Search under its own deadline,
167+
// derived from the fan-out context so the aggregate ceiling still applies.
168+
//
169+
// The underlying adapter clients still issue non-cancelable http.NewRequest
170+
// calls, so a wedged source's goroutine may outlive this call — we abandon it
171+
// (the buffered result channel absorbs a late send) rather than block. On
172+
// timeout we synthesize a per-source error and, when the source exposes a
173+
// homepage ProbeURL, upgrade it via netx.EnrichTimeoutWithProbe: a quick HEAD
174+
// distinguishes "the site's origin is down (5xx / Cloudflare)" from an opaque
175+
// hang, so the breaker opens with an actionable diagnostic.
176+
func searchOneWithTimeout(parent context.Context, a activeSearcher, query string) searchOne {
177+
sctx, cancel := context.WithTimeout(parent, perSourceSearchTimeout)
178+
defer cancel()
179+
180+
type outcome struct {
181+
results []*models.Anime
182+
err error
183+
}
184+
done := make(chan outcome, 1)
185+
go func() {
186+
res, err := a.sr.Search(sctx, query)
187+
done <- outcome{results: res, err: err}
188+
}()
189+
190+
select {
191+
case o := <-done:
192+
return searchOne{kind: a.kind, results: o.results, err: o.err}
193+
case <-sctx.Done():
194+
base := fmt.Errorf("%s search timed out after %v", sourceDisplayName(a.kind), perSourceSearchTimeout)
195+
err := netx.EnrichTimeoutWithProbe(parent, sourceDisplayName(a.kind), "search", a.probeURL, base, originProbeBudget)
196+
return searchOne{kind: a.kind, err: err}
197+
}
198+
}
199+
149200
func finishSearch(query string, all []*models.Anime, errs []error) ([]*models.Anime, error) {
150201
if len(all) == 0 {
151202
if len(errs) > 0 {

internal/api/providers/dispatch_test.go

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,16 @@ package providers
22

33
import (
44
"context"
5+
"errors"
6+
"net/http"
7+
"net/http/httptest"
58
"sync/atomic"
69
"testing"
710
"time"
811

912
"github.com/alvarorichard/Goanime/internal/api/source"
1013
"github.com/alvarorichard/Goanime/internal/models"
14+
"github.com/alvarorichard/Goanime/internal/scraper/netx"
1115
"github.com/stretchr/testify/assert"
1216
"github.com/stretchr/testify/require"
1317
)
@@ -163,6 +167,48 @@ func TestSearchAll_AllFailReturnsError(t *testing.T) {
163167
assert.Contains(t, err.Error(), "all sources failed")
164168
}
165169

170+
// hangingSearchSource models the real adapters: its Search ignores ctx and
171+
// blocks until released, so the only thing that ends the wait is the
172+
// per-source deadline enforced by searchOneWithTimeout.
173+
type hangingSearchSource struct {
174+
epStubSource
175+
release chan struct{}
176+
}
177+
178+
func (s *hangingSearchSource) Search(context.Context, string) ([]*models.Anime, error) {
179+
<-s.release
180+
return nil, nil
181+
}
182+
183+
func TestSearchOneWithTimeout_EnrichesWithOriginProbe(t *testing.T) {
184+
// Mutates the package-level perSourceSearchTimeout — not parallel.
185+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
186+
w.WriteHeader(522) // Cloudflare origin down
187+
}))
188+
t.Cleanup(srv.Close)
189+
190+
prev := perSourceSearchTimeout
191+
perSourceSearchTimeout = 20 * time.Millisecond
192+
t.Cleanup(func() { perSourceSearchTimeout = prev })
193+
194+
stub := &hangingSearchSource{
195+
epStubSource: epStubSource{desc: source.Descriptor{Kind: source.Goyabu, Priority: 1}},
196+
release: make(chan struct{}),
197+
}
198+
t.Cleanup(func() { close(stub.release) }) // let the abandoned goroutine exit
199+
200+
got := searchOneWithTimeout(context.Background(), activeSearcher{
201+
sr: stub,
202+
kind: source.Goyabu,
203+
probeURL: srv.URL,
204+
}, "naruto")
205+
206+
require.Error(t, got.err)
207+
var diag *netx.SourceDiagnostic
208+
require.True(t, errors.As(got.err, &diag), "timeout must be enriched into a *netx.SourceDiagnostic")
209+
assert.Equal(t, 522, diag.StatusCode)
210+
}
211+
166212
func TestSearchAll_NoSearchableSource(t *testing.T) {
167213
// Swaps the global registry with a non-searchable source — not parallel.
168214
restore := source.SwapRegistryForTesting(&epStubSource{desc: source.Descriptor{Kind: "plain", Priority: 1}})

internal/api/providers/source_providers.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,7 @@ func (p *animeFireProvider) Describe() source.Descriptor {
161161
Explicit: []string{"Animefire.io", "AnimeFire"},
162162
Tags: []string{"[animefire]"},
163163
URLMatchers: []string{"animefire"},
164+
ProbeURL: "https://animefire.io",
164165
}
165166
}
166167

@@ -239,6 +240,7 @@ func (p *goyabuProvider) Describe() source.Descriptor {
239240
Explicit: []string{"Goyabu"},
240241
Tags: []string{"[goyabu]"},
241242
URLMatchers: []string{"goyabu"},
243+
ProbeURL: "https://goyabu.io",
242244
}
243245
}
244246

internal/api/source/source.go

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,14 @@ type Descriptor struct {
2626
// or fragile sources that shouldn't ship live. Independent of the always-
2727
// available GOANIME_DISABLED_SOURCES kill-switch.
2828
DefaultDisabled bool
29+
30+
// ProbeURL is the source's public homepage, used only to disambiguate a
31+
// search timeout: after a per-source deadline the dispatcher issues a quick
32+
// HEAD against it (netx.EnrichTimeoutWithProbe) so a 5xx / Cloudflare-origin
33+
// failure is reported as "site down" instead of a generic timeout. Leave
34+
// empty for opaque APIs (AllAnime GraphQL) or browser-gated sources
35+
// (SuperFlix) where a homepage probe is not meaningful.
36+
ProbeURL string
2937
}
3038

3139
// matchNonExplicit checks all match criteria except the explicit Source field.
Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,152 @@
1+
// ===========================================================================
2+
// blogger_proxy_readiness_test.go — Regression tests for two Blogger-proxy bugs
3+
//
4+
// Issue observed: 2026-07-23 (Goyabu / Naruto Shippuden, debug log)
5+
// The FIRST play of an episode failed with:
6+
// "mpv exited before IPC socket was ready: exit status 2 ... (no stderr)"
7+
// Hint: bundled mpv may be missing DLLs
8+
// The second attempt (next episode) played fine. mpv was healthy — the hint
9+
// was misleading.
10+
//
11+
// Root causes (player/scraper.go):
12+
// 1. Redirect handling. The proxy's upstream client was built with
13+
// .NotFollowRedirects(); when the signed googlevideo URL 302-redirected to
14+
// a CDN node, surf surfaced that as a "net/http: use last response" error,
15+
// and the handler turned it into HTTP 502.
16+
// 2. Readiness masking. The readiness probe treated ANY transport-level
17+
// success as ready — a 502 (headErr==nil) counted as ready — so mpv was
18+
// launched against a proxy serving 502 and exited 2.
19+
//
20+
// Fixes:
21+
// 1. proxyClient.CheckRedirect = nil (follow redirects, net/http default).
22+
// 2. Readiness rejects non-2xx upstream status; on the deadline it returns an
23+
// error so extractBloggerVideoURL's retry loop re-resolves a fresh CDN host
24+
// instead of handing mpv a dead stream.
25+
//
26+
// Function tested: startBloggerProxyServer (the network-free seam split out of
27+
// startBloggerProxy so the forwarding + readiness gating are testable).
28+
// ===========================================================================
29+
30+
package player
31+
32+
import (
33+
"io"
34+
"net/http"
35+
"net/http/httptest"
36+
"testing"
37+
"time"
38+
39+
"github.com/stretchr/testify/assert"
40+
"github.com/stretchr/testify/require"
41+
)
42+
43+
// okVideoHandler answers HEAD with 200 and GET with the given body — a stand-in
44+
// for a healthy googlevideo CDN node.
45+
func okVideoHandler(body string) http.HandlerFunc {
46+
return func(w http.ResponseWriter, r *http.Request) {
47+
w.Header().Set("Content-Type", "video/mp4")
48+
if r.Method == http.MethodHead {
49+
w.WriteHeader(http.StatusOK)
50+
return
51+
}
52+
w.WriteHeader(http.StatusOK)
53+
_, _ = io.WriteString(w, body)
54+
}
55+
}
56+
57+
// getBloggerProxyBody performs a GET against the proxy and returns the body.
58+
func getBloggerProxyBody(t *testing.T, proxyURL string) string {
59+
t.Helper()
60+
resp, err := http.Get(proxyURL) //nolint:noctx // local proxy, test-only
61+
require.NoError(t, err)
62+
defer func() { _ = resp.Body.Close() }()
63+
b, err := io.ReadAll(resp.Body)
64+
require.NoError(t, err)
65+
return string(b)
66+
}
67+
68+
func TestStartBloggerProxyServer(t *testing.T) {
69+
// Mutates the bloggerReadiness* package vars and the bloggerProxy global —
70+
// not parallel, and subtests run sequentially so they don't fight over the
71+
// single shared proxy.
72+
prevTimeout, prevInterval := bloggerReadinessTimeout, bloggerReadinessInterval
73+
bloggerReadinessTimeout = 300 * time.Millisecond
74+
bloggerReadinessInterval = 10 * time.Millisecond
75+
t.Cleanup(func() {
76+
bloggerReadinessTimeout, bloggerReadinessInterval = prevTimeout, prevInterval
77+
StopBloggerProxy()
78+
})
79+
80+
const videoBody = "FAKE-MP4-BYTES"
81+
82+
t.Run("healthy 200 upstream is served", func(t *testing.T) {
83+
up := httptest.NewServer(okVideoHandler(videoBody))
84+
t.Cleanup(up.Close)
85+
t.Cleanup(StopBloggerProxy)
86+
87+
proxyURL, err := startBloggerProxyServer(up.URL, &http.Client{})
88+
require.NoError(t, err)
89+
assert.Equal(t, videoBody, getBloggerProxyBody(t, proxyURL))
90+
})
91+
92+
t.Run("follows an upstream redirect (regression: use-last-response -> 502)", func(t *testing.T) {
93+
final := httptest.NewServer(okVideoHandler(videoBody))
94+
t.Cleanup(final.Close)
95+
redir := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
96+
http.Redirect(w, r, final.URL, http.StatusFound)
97+
}))
98+
t.Cleanup(redir.Close)
99+
t.Cleanup(StopBloggerProxy)
100+
101+
// A default client (CheckRedirect==nil) follows the redirect — this is
102+
// the production fix. Before it, the proxy 502'd on the redirect.
103+
proxyURL, err := startBloggerProxyServer(redir.URL, &http.Client{})
104+
require.NoError(t, err, "a redirect-following client must reach the real 200 video")
105+
assert.Equal(t, videoBody, getBloggerProxyBody(t, proxyURL))
106+
})
107+
108+
t.Run("non-following client never yields a 2xx (the old bug shape)", func(t *testing.T) {
109+
final := httptest.NewServer(okVideoHandler(videoBody))
110+
t.Cleanup(final.Close)
111+
redir := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
112+
http.Redirect(w, r, final.URL, http.StatusFound)
113+
}))
114+
t.Cleanup(redir.Close)
115+
t.Cleanup(StopBloggerProxy)
116+
117+
// Models the removed .NotFollowRedirects(): the proxy forwards the 302
118+
// (never the final 200), so readiness never observes a 2xx and fails.
119+
noFollow := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error {
120+
return http.ErrUseLastResponse
121+
}}
122+
_, err := startBloggerProxyServer(redir.URL, noFollow)
123+
require.Error(t, err, "without redirect-follow the proxy never serves a 2xx video")
124+
})
125+
126+
t.Run("rejects a 502 upstream instead of masking it", func(t *testing.T) {
127+
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
128+
w.WriteHeader(http.StatusBadGateway)
129+
}))
130+
t.Cleanup(up.Close)
131+
t.Cleanup(StopBloggerProxy)
132+
133+
_, err := startBloggerProxyServer(up.URL, &http.Client{})
134+
require.Error(t, err, "a 502-serving upstream must not be reported as ready")
135+
assert.Contains(t, err.Error(), "502")
136+
})
137+
138+
t.Run("accepts a 206 partial-content upstream", func(t *testing.T) {
139+
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
140+
w.Header().Set("Content-Type", "video/mp4")
141+
w.WriteHeader(http.StatusPartialContent)
142+
if r.Method == http.MethodGet {
143+
_, _ = io.WriteString(w, videoBody)
144+
}
145+
}))
146+
t.Cleanup(up.Close)
147+
t.Cleanup(StopBloggerProxy)
148+
149+
_, err := startBloggerProxyServer(up.URL, &http.Client{})
150+
require.NoError(t, err, "206 is a valid streaming status and must count as ready")
151+
})
152+
}

0 commit comments

Comments
 (0)