diff --git a/go_backend/cover.go b/go_backend/cover.go index 87fc82a1..78397f9c 100644 --- a/go_backend/cover.go +++ b/go_backend/cover.go @@ -6,6 +6,8 @@ import ( "net/http" "regexp" "strings" + "sync" + "time" ) const ( @@ -52,6 +54,104 @@ func downloadCoverToMemory(coverURL string, maxQuality bool) ([]byte, error) { GoLog("[Cover] Final URL: %s", downloadURL) + data, err := fetchCoverCached(downloadURL) + if err != nil { + return nil, err + } + // Cached bytes are shared across goroutines and must never be mutated; + // hand callers their own copy. + return append([]byte(nil), data...), nil +} + +const ( + coverCacheMaxBytes = 24 * 1024 * 1024 + coverCacheTTL = 15 * time.Minute +) + +type coverCacheEntry struct { + data []byte + expiresAt time.Time +} + +type coverInflightCall struct { + wg sync.WaitGroup + data []byte + err error +} + +var ( + coverMu sync.Mutex + coverCache = map[string]*coverCacheEntry{} + coverCacheBytes int + coverInflight = map[string]*coverInflightCall{} + coverFetch = fetchCoverBytes +) + +// fetchCoverCached returns cover bytes for a final URL, collapsing concurrent +// requests for the same URL into a single fetch (singleflight) and caching +// results in memory for the duration of an album batch. The returned slice is +// shared; callers must copy before mutating. +func fetchCoverCached(downloadURL string) ([]byte, error) { + coverMu.Lock() + if e, ok := coverCache[downloadURL]; ok { + if time.Now().Before(e.expiresAt) { + data := e.data + coverMu.Unlock() + return data, nil + } + delete(coverCache, downloadURL) + coverCacheBytes -= len(e.data) + } + if call, ok := coverInflight[downloadURL]; ok { + coverMu.Unlock() + call.wg.Wait() + return call.data, call.err + } + call := &coverInflightCall{} + call.wg.Add(1) + coverInflight[downloadURL] = call + coverMu.Unlock() + + data, err := coverFetch(downloadURL) + call.data, call.err = data, err + if err == nil { + coverCachePut(downloadURL, data) + } + call.wg.Done() + + coverMu.Lock() + delete(coverInflight, downloadURL) + coverMu.Unlock() + + return data, err +} + +func coverCachePut(downloadURL string, data []byte) { + if len(data) == 0 || len(data) > coverCacheMaxBytes { + return + } + coverMu.Lock() + defer coverMu.Unlock() + if e, ok := coverCache[downloadURL]; ok { + coverCacheBytes -= len(e.data) + } + coverCache[downloadURL] = &coverCacheEntry{data: data, expiresAt: time.Now().Add(coverCacheTTL)} + coverCacheBytes += len(data) + for coverCacheBytes > coverCacheMaxBytes && len(coverCache) > 1 { + var oldestKey string + var oldest time.Time + first := true + for k, e := range coverCache { + if first || e.expiresAt.Before(oldest) { + oldest, oldestKey, first = e.expiresAt, k, false + } + } + coverCacheBytes -= len(coverCache[oldestKey].data) + delete(coverCache, oldestKey) + } +} + +func fetchCoverBytes(downloadURL string) ([]byte, error) { client := NewHTTPClientWithTimeout(DefaultTimeout) req, err := http.NewRequest("GET", downloadURL, nil) diff --git a/go_backend/cover_test.go b/go_backend/cover_test.go new file mode 100644 index 00000000..3c915e73 --- /dev/null +++ b/go_backend/cover_test.go @@ -0,0 +1,96 @@ +package gobackend + +import ( + "sync" + "sync/atomic" + "testing" + "time" +) + +func resetCoverCache() { + coverMu.Lock() + coverCache = map[string]*coverCacheEntry{} + coverInflight = map[string]*coverInflightCall{} + coverCacheBytes = 0 + coverMu.Unlock() +} + +func TestFetchCoverCachedSingleflight(t *testing.T) { + orig := coverFetch + defer func() { coverFetch = orig }() + resetCoverCache() + + var calls int32 + entered := make(chan struct{}) + release := make(chan struct{}) + coverFetch = func(string) ([]byte, error) { + if atomic.AddInt32(&calls, 1) == 1 { + close(entered) + } + <-release + return []byte("coverbytes"), nil + } + + const url = "https://cdn.example/cover_max.jpg" + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + if _, err := fetchCoverCached(url); err != nil { + t.Errorf("leader fetch error: %v", err) + } + }() + + <-entered // leader has registered inflight and is blocked in coverFetch + + wg.Add(1) + go func() { + defer wg.Done() + if _, err := fetchCoverCached(url); err != nil { + t.Errorf("follower fetch error: %v", err) + } + }() + + close(release) + wg.Wait() + + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("expected 1 fetch for concurrent requests, got %d", got) + } +} + +func TestFetchCoverCachedTTLExpiry(t *testing.T) { + orig := coverFetch + defer func() { coverFetch = orig }() + resetCoverCache() + + var calls int32 + coverFetch = func(string) ([]byte, error) { + atomic.AddInt32(&calls, 1) + return []byte("data"), nil + } + + const url = "https://cdn.example/ttl.jpg" + if _, err := fetchCoverCached(url); err != nil { + t.Fatalf("first fetch error: %v", err) + } + // second call served from cache + if _, err := fetchCoverCached(url); err != nil { + t.Fatalf("second fetch error: %v", err) + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("expected cache hit, got %d fetches", got) + } + + // expire the entry and confirm a refetch + coverMu.Lock() + coverCache[url].expiresAt = time.Now().Add(-time.Minute) + coverMu.Unlock() + + if _, err := fetchCoverCached(url); err != nil { + t.Fatalf("third fetch error: %v", err) + } + if got := atomic.LoadInt32(&calls); got != 2 { + t.Fatalf("expected refetch after TTL expiry, got %d fetches", got) + } +} diff --git a/go_backend/deezer.go b/go_backend/deezer.go index ab13ac58..8d78730e 100644 --- a/go_backend/deezer.go +++ b/go_backend/deezer.go @@ -1269,6 +1269,10 @@ func (c *DeezerClient) getJSON(ctx context.Context, endpoint string, dst any) er for attempt := 0; attempt <= deezerMaxRetries; attempt++ { if attempt > 0 { delay := deezerRetryDelay * time.Duration(1<<(attempt-1)) + var apiErr *deezerAPIError + if errors.As(lastErr, &apiErr) && apiErr.RetryAfter > 0 { + delay = apiErr.RetryAfter + } GoLog("[Deezer] Retry %d/%d after %v...\n", attempt, deezerMaxRetries, delay) time.Sleep(delay) } @@ -1292,6 +1296,7 @@ func (c *DeezerClient) getJSON(ctx context.Context, endpoint string, dst any) er type deezerAPIError struct { StatusCode int Body string + RetryAfter time.Duration } func (e *deezerAPIError) Error() string { @@ -1329,7 +1334,7 @@ func (c *DeezerClient) doGetJSON(ctx context.Context, endpoint string, dst any) } if resp.StatusCode != http.StatusOK { - return &deezerAPIError{StatusCode: resp.StatusCode, Body: string(body)} + return &deezerAPIError{StatusCode: resp.StatusCode, Body: string(body), RetryAfter: getRetryAfterDuration(resp)} } return json.Unmarshal(body, dst) diff --git a/go_backend/extension_runtime.go b/go_backend/extension_runtime.go index 4f783227..ea31cac4 100644 --- a/go_backend/extension_runtime.go +++ b/go_backend/extension_runtime.go @@ -1,6 +1,7 @@ package gobackend import ( + "context" "fmt" "net" "net/http" @@ -276,6 +277,49 @@ func (r *extensionRuntime) bindDownloadCancelContext(req *http.Request) *http.Re return req.WithContext(initDownloadCancel(itemID)) } +// downloadStallTimeout is how long a download may go without receiving a single +// byte before the stall watchdog aborts it. A dead radio mid-transfer otherwise +// blocks on Body.Read until the 24h client timeout with no error and no retry. +const downloadStallTimeout = 60 * time.Second + +// stallWatchdog cancels an in-flight download when no data arrives within +// timeout. It wraps the request context in a child cancel so firing it does NOT +// set the user-cancel flag (isDownloadCancelled stays false) — a stall is a +// distinct, retryable condition. Call reset() after every successful Read and +// stop() when the transfer ends. +type stallWatchdog struct { + cancel context.CancelFunc + timer *time.Timer + timeout time.Duration + stalled atomic.Bool +} + +func bindStallWatchdog(req *http.Request, timeout time.Duration) (*http.Request, *stallWatchdog) { + ctx, cancel := context.WithCancel(req.Context()) + w := &stallWatchdog{cancel: cancel, timeout: timeout} + w.timer = time.AfterFunc(timeout, func() { + w.stalled.Store(true) + cancel() + }) + return req.WithContext(ctx), w +} + +func (w *stallWatchdog) reset() { w.timer.Reset(w.timeout) } + +// stop halts the timer and releases the child context so a completed download +// leaks neither a pending timer nor a live cancel func. +func (w *stallWatchdog) stop() { + w.timer.Stop() + w.cancel() +} + +// stallError is returned when the watchdog fires. The message is deliberately +// free of "cancel" and worded to classify as retryable network failure, so the +// fallback layer retries instead of treating it as a user cancellation. +func (r *extensionRuntime) stallError() goja.Value { + return r.jsError("download stalled: no data received for %ds (network timeout)", int(downloadStallTimeout.Seconds())) +} + func newExtensionHTTPClient(ext *loadedExtension, jar http.CookieJar, timeout time.Duration, compressResponses bool) *http.Client { // Extension sandbox enforces HTTPS-only domains. Do not apply global // allow_http scheme downgrade here, because some extension APIs (e.g. diff --git a/go_backend/extension_runtime_file.go b/go_backend/extension_runtime_file.go index 97b284cb..b4e4177e 100644 --- a/go_backend/extension_runtime_file.go +++ b/go_backend/extension_runtime_file.go @@ -203,6 +203,8 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value { return r.jsError("%s", err.Error()) } req = r.bindDownloadCancelContext(req) + req, wd := bindStallWatchdog(req, downloadStallTimeout) + defer wd.stop() for k, v := range headers { req.Header.Set(k, v) @@ -213,6 +215,9 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value { resp, err := client.Do(req) if err != nil { + if wd.stalled.Load() { + return r.stallError() + } return r.jsError("%s", err.Error()) } defer resp.Body.Close() @@ -259,6 +264,7 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value { for { nr, er := resp.Body.Read(buf) if nr > 0 { + wd.reset() nw, ew := progressWriter.Write(buf[0:nr]) if nw < 0 || nr < nw { nw = 0 @@ -283,6 +289,9 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value { } if er != nil { if er != io.EOF { + if wd.stalled.Load() { + return r.stallError() + } return r.jsError("failed to read response: %v", er) } break @@ -332,6 +341,9 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full return r.jsError("chunked: probe request error: %v", err) } probeReq = r.bindDownloadCancelContext(probeReq) + probeReq, wd := bindStallWatchdog(probeReq, downloadStallTimeout) + defer wd.stop() + stallCtx := probeReq.Context() probeReq.Header.Set("User-Agent", ua) for k, v := range headers { if k != "Range" { // Don't copy any existing Range header @@ -342,6 +354,9 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full probeResp, err := client.Do(probeReq) if err != nil { + if wd.stalled.Load() { + return r.stallError() + } return r.jsError("chunked: probe error: %v", err) } io.Copy(io.Discard, probeResp.Body) @@ -420,7 +435,7 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full if err != nil { return r.jsError("chunked: request error at offset %d: %v", offset, err) } - chunkReq = r.bindDownloadCancelContext(chunkReq) + chunkReq = chunkReq.WithContext(stallCtx) chunkReq.Header.Set("User-Agent", ua) for k, v := range headers { if k != "Range" { @@ -431,6 +446,9 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full chunkResp, chunkErr = client.Do(chunkReq) if chunkErr != nil { + if wd.stalled.Load() { + return r.stallError() + } if retry < maxRetries-1 { time.Sleep(time.Duration(retry+1) * time.Second) continue @@ -459,6 +477,7 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full for { nr, er := chunkResp.Body.Read(buf) if nr > 0 { + wd.reset() nw, ew := progressWriter.Write(buf[0:nr]) if nw < 0 || nr < nw { nw = 0 @@ -487,6 +506,9 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full if er != nil { if er != io.EOF { chunkResp.Body.Close() + if wd.stalled.Load() { + return r.stallError() + } return r.jsError("failed to read chunk at offset %d: %v", offset, er) } break diff --git a/go_backend/extension_runtime_stall_test.go b/go_backend/extension_runtime_stall_test.go new file mode 100644 index 00000000..ec1694af --- /dev/null +++ b/go_backend/extension_runtime_stall_test.go @@ -0,0 +1,54 @@ +package gobackend + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// TestStallWatchdogCancelsOnNoData verifies the watchdog aborts a transfer that +// stops sending bytes, marks itself stalled (not user-cancelled), and does so +// after resetting on the initial byte. +func TestStallWatchdogCancelsOnNoData(t *testing.T) { + block := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", "1000") + w.WriteHeader(http.StatusOK) + w.Write([]byte("x")) + w.(http.Flusher).Flush() + <-block // simulate a dead radio mid-transfer: stop sending + })) + defer srv.Close() + defer close(block) + + req, _ := http.NewRequestWithContext(context.Background(), "GET", srv.URL, nil) + req, wd := bindStallWatchdog(req, 150*time.Millisecond) + defer wd.stop() + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do: %v", err) + } + defer resp.Body.Close() + + buf := make([]byte, 32) + var readErr error + for { + n, er := resp.Body.Read(buf) + if n > 0 { + wd.reset() + } + if er != nil { + readErr = er + break + } + } + if readErr == nil { + t.Fatal("expected read error from stall cancel") + } + if !wd.stalled.Load() { + t.Fatalf("watchdog did not mark stalled; err=%v", readErr) + } +} diff --git a/go_backend/httputil.go b/go_backend/httputil.go index 61657c27..15e95b20 100644 --- a/go_backend/httputil.go +++ b/go_backend/httputil.go @@ -65,14 +65,15 @@ var ( var sharedTransport = &http.Transport{ DialContext: (&net.Dialer{ - Timeout: 30 * time.Second, + Timeout: 10 * time.Second, KeepAlive: 30 * time.Second, }).DialContext, MaxIdleConns: 100, MaxIdleConnsPerHost: 10, MaxConnsPerHost: 20, - IdleConnTimeout: 90 * time.Second, + IdleConnTimeout: 60 * time.Second, TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 45 * time.Second, ExpectContinueTimeout: 1 * time.Second, DisableKeepAlives: false, ForceAttemptHTTP2: true, @@ -84,14 +85,15 @@ var sharedTransport = &http.Transport{ var extensionAPITransport = &http.Transport{ DialContext: (&net.Dialer{ - Timeout: 30 * time.Second, + Timeout: 10 * time.Second, KeepAlive: 30 * time.Second, }).DialContext, MaxIdleConns: 100, MaxIdleConnsPerHost: 10, MaxConnsPerHost: 20, - IdleConnTimeout: 90 * time.Second, + IdleConnTimeout: 60 * time.Second, TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 45 * time.Second, ExpectContinueTimeout: 1 * time.Second, DisableKeepAlives: false, ForceAttemptHTTP2: true, @@ -103,14 +105,15 @@ var extensionAPITransport = &http.Transport{ var metadataTransport = &http.Transport{ DialContext: (&net.Dialer{ - Timeout: 30 * time.Second, + Timeout: 10 * time.Second, KeepAlive: 30 * time.Second, }).DialContext, MaxIdleConns: 30, MaxIdleConnsPerHost: 5, MaxConnsPerHost: 10, - IdleConnTimeout: 90 * time.Second, + IdleConnTimeout: 60 * time.Second, TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: 45 * time.Second, ExpectContinueTimeout: 1 * time.Second, DisableKeepAlives: false, ForceAttemptHTTP2: true, @@ -276,7 +279,7 @@ func DoRequestWithRetry(client *http.Client, req *http.Request, config RetryConf if err != nil { lastErr = err - if CheckAndLogISPBlocking(err, reqCopy.URL.String(), "HTTP") { + if isHardConnectivityBlock(err) { return nil, WrapErrorWithISPCheck(err, reqCopy.URL.String(), "HTTP") } @@ -331,6 +334,9 @@ func DoRequestWithRetry(client *http.Client, req *http.Request, config RetryConf if resp.StatusCode >= 500 { resp.Body.Close() + if retryAfter := getRetryAfterDuration(resp); retryAfter > 0 { + delay = retryAfter + } lastErr = fmt.Errorf("server error: HTTP %d", resp.StatusCode) if attempt < config.MaxRetries { GoLog("[HTTP] Server error %d, retrying in %v...\n", resp.StatusCode, delay) @@ -346,9 +352,20 @@ func DoRequestWithRetry(client *http.Client, req *http.Request, config RetryConf return nil, fmt.Errorf("request failed after %d retries: %w", config.MaxRetries+1, lastErr) } +// jitterFloat returns a fraction in [0,1); overridable in tests for +// deterministic backoff assertions. +var jitterFloat = rand.Float64 + func calculateNextDelay(currentDelay time.Duration, config RetryConfig) time.Duration { nextDelay := time.Duration(float64(currentDelay) * config.BackoffFactor) - return min(nextDelay, config.MaxDelay) + capped := min(nextDelay, config.MaxDelay) + // Full jitter: spread retries between InitialDelay and the capped + // exponential ceiling to avoid synchronized thundering-herd retries. + if capped <= config.InitialDelay { + return capped + } + span := capped - config.InitialDelay + return config.InitialDelay + time.Duration(jitterFloat()*float64(span)) } // Returns 0 if the header is missing or invalid so callers can keep their @@ -524,6 +541,42 @@ func isTLSHandshakeOrResetError(err error) bool { return false } +// isHardConnectivityBlock reports transport failures that signal an active +// block (DNS not found, connection refused/reset, TLS/cert MITM) and should +// abort retries immediately. Timeouts and deadline-exceeded are treated as +// transient and excluded so they fall through to normal retry backoff. +func isHardConnectivityBlock(err error) bool { + if err == nil { + return false + } + + var urlErr *url.Error + if errors.As(err, &urlErr) && urlErr.Timeout() { + return false + } + + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) { + return dnsErr.IsNotFound + } + + var opErr *net.OpError + if errors.As(err, &opErr) { + if opErr.Timeout() { + return false + } + var errno syscall.Errno + if errors.As(opErr.Err, &errno) { + switch errno { + case syscall.ECONNREFUSED, syscall.ECONNRESET: + return true + } + } + } + + return isTLSHandshakeOrResetError(err) +} + func IsISPBlocking(err error, requestURL string) *ISPBlockingError { if err == nil { return nil diff --git a/go_backend/httputil_supplement_test.go b/go_backend/httputil_supplement_test.go index 0e388044..d600764a 100644 --- a/go_backend/httputil_supplement_test.go +++ b/go_backend/httputil_supplement_test.go @@ -112,9 +112,12 @@ func TestHTTPUtilityHelpers(t *testing.T) { if body, err := ReadResponseBody(&http.Response{Body: io.NopCloser(strings.NewReader("ok"))}); err != nil || string(body) != "ok" { t.Fatalf("ReadResponseBody = %q/%v", body, err) } + origJitter := jitterFloat + jitterFloat = func() float64 { return 1 } if calculateNextDelay(10*time.Millisecond, RetryConfig{BackoffFactor: 3, MaxDelay: 20 * time.Millisecond}) != 20*time.Millisecond { t.Fatal("calculateNextDelay mismatch") } + jitterFloat = origJitter if getRetryAfterDuration(&http.Response{Header: http.Header{"Retry-After": []string{"bad"}}}) != 0 { t.Fatal("invalid retry-after should be zero") } diff --git a/go_backend/httputil_utls.go b/go_backend/httputil_utls.go index 2e713ab3..3a62fcfc 100644 --- a/go_backend/httputil_utls.go +++ b/go_backend/httputil_utls.go @@ -4,27 +4,38 @@ package gobackend import ( "context" - "crypto/tls" "io" "net" "net/http" "net/url" "strings" + "sync" utls "github.com/refraction-networking/utls" "golang.org/x/net/http2" ) +// utlsSessionCache is shared by every uTLS handshake so TLS 1.3 tickets enable +// resumption (fewer round-trips) across requests and hosts. +var utlsSessionCache = utls.NewLRUClientSessionCache(0) + +// utlsTransport dials with a Chrome TLS fingerprint and pools one healthy HTTP/2 +// connection per host, re-dialing when it dies (e.g. after a network switch). type utlsTransport struct { dialer *net.Dialer + h2 *http2.Transport + mu sync.Mutex + conns map[string]*http2.ClientConn } func newUTLSTransport() *utlsTransport { return &utlsTransport{ dialer: &net.Dialer{ - Timeout: 30 * Second, + Timeout: 10 * Second, KeepAlive: 30 * Second, }, + h2: &http2.Transport{}, + conns: make(map[string]*http2.ClientConn), } } @@ -34,48 +45,94 @@ func (t *utlsTransport) RoundTrip(req *http.Request) (*http.Response, error) { } host := req.URL.Hostname() - port := t.getPort(req.URL) - addr := net.JoinHostPort(host, port) + addr := net.JoinHostPort(host, t.getPort(req.URL)) - conn, err := t.dialer.DialContext(req.Context(), "tcp", addr) + if cc := t.cachedConn(addr); cc != nil { + resp, err := cc.RoundTrip(req) + if err != nil { + t.invalidate(addr, cc) + } + return resp, err + } + + tlsConn, proto, err := t.dial(req.Context(), host, addr) if err != nil { return nil, err } + if proto != "h2" { + // HTTP/1.1: single-use conn closed once the body is drained. + transport := &http.Transport{ + DialTLSContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + return tlsConn, nil + }, + DisableKeepAlives: true, + } + return transport.RoundTrip(req) + } + + cc, err := t.h2.NewClientConn(tlsConn) + if err != nil { + tlsConn.Close() + return nil, err + } + cc = t.storeConn(addr, cc) + return cc.RoundTrip(req) +} + +// dial opens a TCP connection and completes the Chrome-fingerprint TLS handshake, +// returning the connection and negotiated ALPN protocol. +func (t *utlsTransport) dial(ctx context.Context, host, addr string) (*utls.UConn, string, error) { + conn, err := t.dialer.DialContext(ctx, "tcp", addr) + if err != nil { + return nil, "", err + } + opts := GetNetworkCompatibilityOptions() tlsConn := utls.UClient(conn, &utls.Config{ RootCAs: supplementalRootCAs(), InsecureSkipVerify: opts.InsecureTLS, ServerName: host, NextProtos: []string{"h2", "http/1.1"}, + ClientSessionCache: utlsSessionCache, }, utls.HelloChrome_Auto) if err := tlsConn.Handshake(); err != nil { conn.Close() - return nil, err + return nil, "", err } + return tlsConn, tlsConn.ConnectionState().NegotiatedProtocol, nil +} - negotiatedProto := tlsConn.ConnectionState().NegotiatedProtocol - - if negotiatedProto == "h2" { - h2Transport := &http2.Transport{ - DialTLSContext: func(ctx context.Context, network, addr string, cfg *tls.Config) (net.Conn, error) { - return tlsConn, nil - }, - AllowHTTP: false, - DisableCompression: false, - } - return h2Transport.RoundTrip(req) +func (t *utlsTransport) cachedConn(addr string) *http2.ClientConn { + t.mu.Lock() + defer t.mu.Unlock() + if cc := t.conns[addr]; cc != nil && cc.CanTakeNewRequest() { + return cc } + return nil +} - transport := &http.Transport{ - DialTLSContext: func(ctx context.Context, network, addr string) (net.Conn, error) { - return tlsConn, nil - }, - DisableKeepAlives: true, +func (t *utlsTransport) invalidate(addr string, cc *http2.ClientConn) { + t.mu.Lock() + if t.conns[addr] == cc { + delete(t.conns, addr) } + t.mu.Unlock() +} - return transport.RoundTrip(req) +// storeConn caches cc, but if a concurrent dial already cached a healthy conn for +// addr it discards the freshly built cc (no in-flight requests) and returns the +// existing one, avoiding a leaked connection. +func (t *utlsTransport) storeConn(addr string, cc *http2.ClientConn) *http2.ClientConn { + t.mu.Lock() + defer t.mu.Unlock() + if existing := t.conns[addr]; existing != nil && existing.CanTakeNewRequest() { + cc.Close() + return existing + } + t.conns[addr] = cc + return cc } func (t *utlsTransport) getPort(u *url.URL) string { diff --git a/go_backend/songlink.go b/go_backend/songlink.go index c9bbd93e..9543b080 100644 --- a/go_backend/songlink.go +++ b/go_backend/songlink.go @@ -9,6 +9,7 @@ import ( "net/url" "strings" "sync" + "time" ) type SongLinkClient struct { @@ -294,18 +295,111 @@ func (s *SongLinkClient) doSongLinkRequest(apiURL string) (map[string]songLinkPl return songLinkResp.LinksByPlatform, nil } +const ( + trackAvailabilityCacheTTL = 30 * time.Minute + trackAvailabilityNegCacheTTL = 5 * time.Minute + trackAvailabilityCacheMax = 500 +) + +type trackAvailabilityCacheEntry struct { + availability *TrackAvailability + err bool + expiresAt time.Time +} + +var ( + trackAvailabilityCacheMu sync.Mutex + trackAvailabilityCache = map[string]trackAvailabilityCacheEntry{} +) + +// CheckTrackAvailability resolves platform availability for a track. Results are +// cached in memory (keyed by region + spotifyID/ISRC) to spare the song.link +// path its 9 req/min budget. This is an extra layer beneath the Dart-side +// cached-invoke and is safe: lookups are idempotent. Negative results use a +// shorter TTL so transient failures recover quickly. func (s *SongLinkClient) CheckTrackAvailability(spotifyTrackID string, isrc string) (*TrackAvailability, error) { spotifyTrackID = strings.TrimSpace(spotifyTrackID) isrc = strings.ToUpper(strings.TrimSpace(isrc)) + var idKey string switch { case spotifyTrackID != "": - return s.checkTrackAvailabilityFromSpotify(spotifyTrackID) + idKey = "spotify:" + spotifyTrackID case isrc != "": - return s.checkTrackAvailabilityFromISRC(isrc) + idKey = "isrc:" + isrc default: return nil, fmt.Errorf("spotify track ID and ISRC are empty") } + key := GetSongLinkRegion() + "|" + idKey + + if cached, hit, cachedErr := trackAvailabilityCacheLookup(key); hit { + if cachedErr { + return nil, fmt.Errorf("track availability unavailable (cached)") + } + return cloneTrackAvailability(cached), nil + } + + var availability *TrackAvailability + var err error + switch { + case spotifyTrackID != "": + availability, err = s.checkTrackAvailabilityFromSpotify(spotifyTrackID) + default: + availability, err = s.checkTrackAvailabilityFromISRC(isrc) + } + + trackAvailabilityCacheStore(key, availability, err) + if err != nil { + return nil, err + } + return cloneTrackAvailability(availability), nil +} + +func cloneTrackAvailability(a *TrackAvailability) *TrackAvailability { + if a == nil { + return nil + } + c := *a + return &c +} + +func trackAvailabilityCacheLookup(key string) (*TrackAvailability, bool, bool) { + trackAvailabilityCacheMu.Lock() + defer trackAvailabilityCacheMu.Unlock() + e, ok := trackAvailabilityCache[key] + if !ok { + return nil, false, false + } + if time.Now().After(e.expiresAt) { + delete(trackAvailabilityCache, key) + return nil, false, false + } + return e.availability, true, e.err +} + +func trackAvailabilityCacheStore(key string, availability *TrackAvailability, err error) { + ttl := trackAvailabilityCacheTTL + if err != nil { + ttl = trackAvailabilityNegCacheTTL + } + trackAvailabilityCacheMu.Lock() + defer trackAvailabilityCacheMu.Unlock() + if _, exists := trackAvailabilityCache[key]; !exists && len(trackAvailabilityCache) >= trackAvailabilityCacheMax { + var oldestKey string + var oldest time.Time + first := true + for k, e := range trackAvailabilityCache { + if first || e.expiresAt.Before(oldest) { + oldest, oldestKey, first = e.expiresAt, k, false + } + } + delete(trackAvailabilityCache, oldestKey) + } + trackAvailabilityCache[key] = trackAvailabilityCacheEntry{ + availability: availability, + err: err != nil, + expiresAt: time.Now().Add(ttl), + } } func (s *SongLinkClient) checkTrackAvailabilityFromSpotify(spotifyTrackID string) (*TrackAvailability, error) { diff --git a/go_backend/songlink_test.go b/go_backend/songlink_test.go index 404a47c9..d950cb5c 100644 --- a/go_backend/songlink_test.go +++ b/go_backend/songlink_test.go @@ -4,7 +4,9 @@ import ( "io" "net/http" "strings" + "sync/atomic" "testing" + "time" ) type roundTripFunc func(*http.Request) (*http.Response, error) @@ -23,7 +25,14 @@ func TestGetRetryAfterDurationMissingHeaderReturnsZero(t *testing.T) { } } +func resetTrackAvailabilityCache() { + trackAvailabilityCacheMu.Lock() + trackAvailabilityCache = map[string]trackAvailabilityCacheEntry{} + trackAvailabilityCacheMu.Unlock() +} + func TestCheckTrackAvailabilityFromSpotifyViaResolveAPI(t *testing.T) { + resetTrackAvailabilityCache() origRetryConfig := songLinkRetryConfig defer func() { songLinkRetryConfig = origRetryConfig }() @@ -65,6 +74,7 @@ func TestCheckTrackAvailabilityFromSpotifyViaResolveAPI(t *testing.T) { } func TestCheckTrackAvailabilityFromSpotifyResolveAPIFailure(t *testing.T) { + resetTrackAvailabilityCache() origRetryConfig := songLinkRetryConfig songLinkRetryConfig = func() RetryConfig { return RetryConfig{MaxRetries: 0, InitialDelay: 0, MaxDelay: 0, BackoffFactor: 1} @@ -115,6 +125,7 @@ func TestCheckTrackAvailabilityFromSpotifyResolveAPIFailure(t *testing.T) { } func TestCheckTrackAvailabilityFromSpotifyViaResolveAPIMixedSongURLShapes(t *testing.T) { + resetTrackAvailabilityCache() origRetryConfig := songLinkRetryConfig defer func() { songLinkRetryConfig = origRetryConfig }() @@ -155,6 +166,61 @@ func TestCheckTrackAvailabilityFromSpotifyViaResolveAPIMixedSongURLShapes(t *tes } } +func TestCheckTrackAvailabilityCachesResult(t *testing.T) { + resetTrackAvailabilityCache() + origRetryConfig := songLinkRetryConfig + defer func() { songLinkRetryConfig = origRetryConfig }() + + var calls int32 + client := &SongLinkClient{ + client: &http.Client{ + Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + atomic.AddInt32(&calls, 1) + body := `{"success":true,"songUrls":{"Spotify":"https://open.spotify.com/track/cachedid","Deezer":"https://www.deezer.com/track/111"}}` + return &http.Response{ + StatusCode: 200, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(body)), + Request: req, + }, nil + }), + }, + } + + first, err := client.CheckTrackAvailability("cachedid", "") + if err != nil { + t.Fatalf("first CheckTrackAvailability() error = %v", err) + } + second, err := client.CheckTrackAvailability("cachedid", "") + if err != nil { + t.Fatalf("second CheckTrackAvailability() error = %v", err) + } + + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("expected 1 network call with caching, got %d", got) + } + if first == second { + t.Fatal("expected cache to return a distinct clone, got same pointer") + } + if second.DeezerID != "111" { + t.Fatalf("cached DeezerID = %q, want 111", second.DeezerID) + } +} + +func TestCheckTrackAvailabilityNegativeCacheTTL(t *testing.T) { + resetTrackAvailabilityCache() + + entry := trackAvailabilityCacheEntry{err: true, expiresAt: time.Now().Add(-time.Second)} + key := GetSongLinkRegion() + "|spotify:expiredneg" + trackAvailabilityCacheMu.Lock() + trackAvailabilityCache[key] = entry + trackAvailabilityCacheMu.Unlock() + + if _, hit, _ := trackAvailabilityCacheLookup(key); hit { + t.Fatal("expired negative entry should not be a cache hit") + } +} + func TestCheckAvailabilityFromDeezerUsesSongLink(t *testing.T) { origRetryConfig := songLinkRetryConfig songLinkRetryConfig = func() RetryConfig { diff --git a/lib/providers/download_queue_provider.dart b/lib/providers/download_queue_provider.dart index da9a7ff9..00220d79 100644 --- a/lib/providers/download_queue_provider.dart +++ b/lib/providers/download_queue_provider.dart @@ -196,6 +196,9 @@ class DownloadQueueNotifier extends Notifier { final Set _ensuredDirs = {}; int _idleProgressPollTick = 0; bool _networkPausedByWifiOnly = false; + List? _lastConnectivityResults; + DateTime _lastConnectionCleanupAt = DateTime.fromMillisecondsSinceEpoch(0); + static const _connectionCleanupDebounce = Duration(seconds: 2); String? _lastServiceTrackName; String? _lastServiceArtistName; String? _lastServiceStatus; @@ -1829,18 +1832,59 @@ class DownloadQueueNotifier extends Notifier { } final shouldResume = _networkPausedByWifiOnly && state.isPaused; - _stopConnectivityMonitoring(); + // Keep monitoring active in every mode while downloading so idle + // connections are still recycled when the network switches. + if (state.isProcessing) { + _networkPausedByWifiOnly = false; + _startConnectivityMonitoring(); + } else { + _stopConnectivityMonitoring(); + } if (shouldResume) { resumeQueue(); } } - void _handleConnectivityResults(List results) { - final settings = ref.read(settingsProvider); - if (settings.downloadNetworkMode != 'wifi_only') { - _handleDownloadNetworkModeChanged(settings.downloadNetworkMode); + /// Closes idle backend HTTP connections when the connectivity set changes + /// (e.g. WiFi <-> cellular). Stale sockets bound to the old interface would + /// otherwise be reused, stalling the first request after a network switch. + /// Applies to every download network mode. Fire-and-forget with a light + /// debounce so network flapping does not spam the bridge. + void _maybeCleanupOnNetworkChange(List results) { + final previous = _lastConnectivityResults; + final unchanged = + previous != null && _connectivitySetEquals(previous, results); + _lastConnectivityResults = List.unmodifiable(results); + if (previous == null || unchanged) return; + + final now = DateTime.now(); + if (now.difference(_lastConnectionCleanupAt) < _connectionCleanupDebounce) { return; } + _lastConnectionCleanupAt = now; + + _log.i('Network changed, closing idle backend connections'); + unawaited( + PlatformBridge.cleanupConnections().catchError((Object e) { + _log.w('Failed to clean up connections after network change: $e'); + }), + ); + } + + bool _connectivitySetEquals( + List a, + List b, + ) { + final setA = a.toSet(); + final setB = b.toSet(); + return setA.length == setB.length && setA.containsAll(setB); + } + + void _handleConnectivityResults(List results) { + _maybeCleanupOnNetworkChange(results); + + final settings = ref.read(settingsProvider); + if (settings.downloadNetworkMode != 'wifi_only') return; if (_hasWifiConnection(results)) { if (_networkPausedByWifiOnly && state.isPaused) { @@ -1921,11 +1965,11 @@ class DownloadQueueNotifier extends Notifier { state = state.copyWith(isProcessing: false, isPaused: true); return; } - _networkPausedByWifiOnly = false; - _startConnectivityMonitoring(); - } else { - _stopConnectivityMonitoring(); } + // Monitor in every mode so idle connections are recycled on a network + // switch, even though only wifi_only pauses the queue. + _networkPausedByWifiOnly = false; + _startConnectivityMonitoring(); if (await _tryProcessQueueWithAndroidNativeWorker(settings)) { return;