mirror of
https://github.com/zarzet/SpotiFLAC-Mobile.git
synced 2026-09-27 20:02:01 +02:00
perf(network): mobile stability and data-usage fixes from network audit
Stability: - transient timeouts no longer classified as ISP blocking; they fall through to retry backoff (hard blocks - DNS/RST/cert - still abort) - connectivity change now closes idle Go connections in every network mode (debounced), so pooled sockets from the old interface are not reused after a wifi/cellular handoff - download body reads get a 60s stall watchdog that cancels and surfaces a retryable network error instead of hanging; distinct from user cancellation - ResponseHeaderTimeout 45s on all transports; IdleConnTimeout 90->60s; dial timeout 30->10s; retry backoff gains full jitter; Retry-After honored on 5xx and Deezer 429 Data usage / speed: - cover downloads deduplicated with singleflight plus a 24MB/15min in-memory cache keyed by final URL (album batches fetched the same 1800px cover once per track before) - song.link availability cached (30min positive / 5min negative) in front of the 9-req/min rate limiter - uTLS Cloudflare path now pools one HTTP/2 connection per host with a shared TLS session cache instead of a full handshake per request Deezer artist track-count N+1 kept: counts feed the discography download UI and are already amortized by the artist TTL cache.
This commit is contained in:
@@ -6,6 +6,8 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"regexp"
|
"regexp"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@@ -52,6 +54,104 @@ func downloadCoverToMemory(coverURL string, maxQuality bool) ([]byte, error) {
|
|||||||
|
|
||||||
GoLog("[Cover] Final URL: %s", downloadURL)
|
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)
|
client := NewHTTPClientWithTimeout(DefaultTimeout)
|
||||||
|
|
||||||
req, err := http.NewRequest("GET", downloadURL, nil)
|
req, err := http.NewRequest("GET", downloadURL, nil)
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1269,6 +1269,10 @@ func (c *DeezerClient) getJSON(ctx context.Context, endpoint string, dst any) er
|
|||||||
for attempt := 0; attempt <= deezerMaxRetries; attempt++ {
|
for attempt := 0; attempt <= deezerMaxRetries; attempt++ {
|
||||||
if attempt > 0 {
|
if attempt > 0 {
|
||||||
delay := deezerRetryDelay * time.Duration(1<<(attempt-1))
|
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)
|
GoLog("[Deezer] Retry %d/%d after %v...\n", attempt, deezerMaxRetries, delay)
|
||||||
time.Sleep(delay)
|
time.Sleep(delay)
|
||||||
}
|
}
|
||||||
@@ -1292,6 +1296,7 @@ func (c *DeezerClient) getJSON(ctx context.Context, endpoint string, dst any) er
|
|||||||
type deezerAPIError struct {
|
type deezerAPIError struct {
|
||||||
StatusCode int
|
StatusCode int
|
||||||
Body string
|
Body string
|
||||||
|
RetryAfter time.Duration
|
||||||
}
|
}
|
||||||
|
|
||||||
func (e *deezerAPIError) Error() string {
|
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 {
|
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)
|
return json.Unmarshal(body, dst)
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package gobackend
|
package gobackend
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -276,6 +277,49 @@ func (r *extensionRuntime) bindDownloadCancelContext(req *http.Request) *http.Re
|
|||||||
return req.WithContext(initDownloadCancel(itemID))
|
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 {
|
func newExtensionHTTPClient(ext *loadedExtension, jar http.CookieJar, timeout time.Duration, compressResponses bool) *http.Client {
|
||||||
// Extension sandbox enforces HTTPS-only domains. Do not apply global
|
// Extension sandbox enforces HTTPS-only domains. Do not apply global
|
||||||
// allow_http scheme downgrade here, because some extension APIs (e.g.
|
// allow_http scheme downgrade here, because some extension APIs (e.g.
|
||||||
|
|||||||
@@ -203,6 +203,8 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value {
|
|||||||
return r.jsError("%s", err.Error())
|
return r.jsError("%s", err.Error())
|
||||||
}
|
}
|
||||||
req = r.bindDownloadCancelContext(req)
|
req = r.bindDownloadCancelContext(req)
|
||||||
|
req, wd := bindStallWatchdog(req, downloadStallTimeout)
|
||||||
|
defer wd.stop()
|
||||||
|
|
||||||
for k, v := range headers {
|
for k, v := range headers {
|
||||||
req.Header.Set(k, v)
|
req.Header.Set(k, v)
|
||||||
@@ -213,6 +215,9 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value {
|
|||||||
|
|
||||||
resp, err := client.Do(req)
|
resp, err := client.Do(req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
if wd.stalled.Load() {
|
||||||
|
return r.stallError()
|
||||||
|
}
|
||||||
return r.jsError("%s", err.Error())
|
return r.jsError("%s", err.Error())
|
||||||
}
|
}
|
||||||
defer resp.Body.Close()
|
defer resp.Body.Close()
|
||||||
@@ -259,6 +264,7 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value {
|
|||||||
for {
|
for {
|
||||||
nr, er := resp.Body.Read(buf)
|
nr, er := resp.Body.Read(buf)
|
||||||
if nr > 0 {
|
if nr > 0 {
|
||||||
|
wd.reset()
|
||||||
nw, ew := progressWriter.Write(buf[0:nr])
|
nw, ew := progressWriter.Write(buf[0:nr])
|
||||||
if nw < 0 || nr < nw {
|
if nw < 0 || nr < nw {
|
||||||
nw = 0
|
nw = 0
|
||||||
@@ -283,6 +289,9 @@ func (r *extensionRuntime) fileDownload(call goja.FunctionCall) goja.Value {
|
|||||||
}
|
}
|
||||||
if er != nil {
|
if er != nil {
|
||||||
if er != io.EOF {
|
if er != io.EOF {
|
||||||
|
if wd.stalled.Load() {
|
||||||
|
return r.stallError()
|
||||||
|
}
|
||||||
return r.jsError("failed to read response: %v", er)
|
return r.jsError("failed to read response: %v", er)
|
||||||
}
|
}
|
||||||
break
|
break
|
||||||
@@ -332,6 +341,9 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full
|
|||||||
return r.jsError("chunked: probe request error: %v", err)
|
return r.jsError("chunked: probe request error: %v", err)
|
||||||
}
|
}
|
||||||
probeReq = r.bindDownloadCancelContext(probeReq)
|
probeReq = r.bindDownloadCancelContext(probeReq)
|
||||||
|
probeReq, wd := bindStallWatchdog(probeReq, downloadStallTimeout)
|
||||||
|
defer wd.stop()
|
||||||
|
stallCtx := probeReq.Context()
|
||||||
probeReq.Header.Set("User-Agent", ua)
|
probeReq.Header.Set("User-Agent", ua)
|
||||||
for k, v := range headers {
|
for k, v := range headers {
|
||||||
if k != "Range" { // Don't copy any existing Range header
|
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)
|
probeResp, err := client.Do(probeReq)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
if wd.stalled.Load() {
|
||||||
|
return r.stallError()
|
||||||
|
}
|
||||||
return r.jsError("chunked: probe error: %v", err)
|
return r.jsError("chunked: probe error: %v", err)
|
||||||
}
|
}
|
||||||
io.Copy(io.Discard, probeResp.Body)
|
io.Copy(io.Discard, probeResp.Body)
|
||||||
@@ -420,7 +435,7 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return r.jsError("chunked: request error at offset %d: %v", offset, err)
|
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)
|
chunkReq.Header.Set("User-Agent", ua)
|
||||||
for k, v := range headers {
|
for k, v := range headers {
|
||||||
if k != "Range" {
|
if k != "Range" {
|
||||||
@@ -431,6 +446,9 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full
|
|||||||
|
|
||||||
chunkResp, chunkErr = client.Do(chunkReq)
|
chunkResp, chunkErr = client.Do(chunkReq)
|
||||||
if chunkErr != nil {
|
if chunkErr != nil {
|
||||||
|
if wd.stalled.Load() {
|
||||||
|
return r.stallError()
|
||||||
|
}
|
||||||
if retry < maxRetries-1 {
|
if retry < maxRetries-1 {
|
||||||
time.Sleep(time.Duration(retry+1) * time.Second)
|
time.Sleep(time.Duration(retry+1) * time.Second)
|
||||||
continue
|
continue
|
||||||
@@ -459,6 +477,7 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full
|
|||||||
for {
|
for {
|
||||||
nr, er := chunkResp.Body.Read(buf)
|
nr, er := chunkResp.Body.Read(buf)
|
||||||
if nr > 0 {
|
if nr > 0 {
|
||||||
|
wd.reset()
|
||||||
nw, ew := progressWriter.Write(buf[0:nr])
|
nw, ew := progressWriter.Write(buf[0:nr])
|
||||||
if nw < 0 || nr < nw {
|
if nw < 0 || nr < nw {
|
||||||
nw = 0
|
nw = 0
|
||||||
@@ -487,6 +506,9 @@ func (r *extensionRuntime) fileDownloadChunked(client *http.Client, urlStr, full
|
|||||||
if er != nil {
|
if er != nil {
|
||||||
if er != io.EOF {
|
if er != io.EOF {
|
||||||
chunkResp.Body.Close()
|
chunkResp.Body.Close()
|
||||||
|
if wd.stalled.Load() {
|
||||||
|
return r.stallError()
|
||||||
|
}
|
||||||
return r.jsError("failed to read chunk at offset %d: %v", offset, er)
|
return r.jsError("failed to read chunk at offset %d: %v", offset, er)
|
||||||
}
|
}
|
||||||
break
|
break
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
+61
-8
@@ -65,14 +65,15 @@ var (
|
|||||||
|
|
||||||
var sharedTransport = &http.Transport{
|
var sharedTransport = &http.Transport{
|
||||||
DialContext: (&net.Dialer{
|
DialContext: (&net.Dialer{
|
||||||
Timeout: 30 * time.Second,
|
Timeout: 10 * time.Second,
|
||||||
KeepAlive: 30 * time.Second,
|
KeepAlive: 30 * time.Second,
|
||||||
}).DialContext,
|
}).DialContext,
|
||||||
MaxIdleConns: 100,
|
MaxIdleConns: 100,
|
||||||
MaxIdleConnsPerHost: 10,
|
MaxIdleConnsPerHost: 10,
|
||||||
MaxConnsPerHost: 20,
|
MaxConnsPerHost: 20,
|
||||||
IdleConnTimeout: 90 * time.Second,
|
IdleConnTimeout: 60 * time.Second,
|
||||||
TLSHandshakeTimeout: 10 * time.Second,
|
TLSHandshakeTimeout: 10 * time.Second,
|
||||||
|
ResponseHeaderTimeout: 45 * time.Second,
|
||||||
ExpectContinueTimeout: 1 * time.Second,
|
ExpectContinueTimeout: 1 * time.Second,
|
||||||
DisableKeepAlives: false,
|
DisableKeepAlives: false,
|
||||||
ForceAttemptHTTP2: true,
|
ForceAttemptHTTP2: true,
|
||||||
@@ -84,14 +85,15 @@ var sharedTransport = &http.Transport{
|
|||||||
|
|
||||||
var extensionAPITransport = &http.Transport{
|
var extensionAPITransport = &http.Transport{
|
||||||
DialContext: (&net.Dialer{
|
DialContext: (&net.Dialer{
|
||||||
Timeout: 30 * time.Second,
|
Timeout: 10 * time.Second,
|
||||||
KeepAlive: 30 * time.Second,
|
KeepAlive: 30 * time.Second,
|
||||||
}).DialContext,
|
}).DialContext,
|
||||||
MaxIdleConns: 100,
|
MaxIdleConns: 100,
|
||||||
MaxIdleConnsPerHost: 10,
|
MaxIdleConnsPerHost: 10,
|
||||||
MaxConnsPerHost: 20,
|
MaxConnsPerHost: 20,
|
||||||
IdleConnTimeout: 90 * time.Second,
|
IdleConnTimeout: 60 * time.Second,
|
||||||
TLSHandshakeTimeout: 10 * time.Second,
|
TLSHandshakeTimeout: 10 * time.Second,
|
||||||
|
ResponseHeaderTimeout: 45 * time.Second,
|
||||||
ExpectContinueTimeout: 1 * time.Second,
|
ExpectContinueTimeout: 1 * time.Second,
|
||||||
DisableKeepAlives: false,
|
DisableKeepAlives: false,
|
||||||
ForceAttemptHTTP2: true,
|
ForceAttemptHTTP2: true,
|
||||||
@@ -103,14 +105,15 @@ var extensionAPITransport = &http.Transport{
|
|||||||
|
|
||||||
var metadataTransport = &http.Transport{
|
var metadataTransport = &http.Transport{
|
||||||
DialContext: (&net.Dialer{
|
DialContext: (&net.Dialer{
|
||||||
Timeout: 30 * time.Second,
|
Timeout: 10 * time.Second,
|
||||||
KeepAlive: 30 * time.Second,
|
KeepAlive: 30 * time.Second,
|
||||||
}).DialContext,
|
}).DialContext,
|
||||||
MaxIdleConns: 30,
|
MaxIdleConns: 30,
|
||||||
MaxIdleConnsPerHost: 5,
|
MaxIdleConnsPerHost: 5,
|
||||||
MaxConnsPerHost: 10,
|
MaxConnsPerHost: 10,
|
||||||
IdleConnTimeout: 90 * time.Second,
|
IdleConnTimeout: 60 * time.Second,
|
||||||
TLSHandshakeTimeout: 10 * time.Second,
|
TLSHandshakeTimeout: 10 * time.Second,
|
||||||
|
ResponseHeaderTimeout: 45 * time.Second,
|
||||||
ExpectContinueTimeout: 1 * time.Second,
|
ExpectContinueTimeout: 1 * time.Second,
|
||||||
DisableKeepAlives: false,
|
DisableKeepAlives: false,
|
||||||
ForceAttemptHTTP2: true,
|
ForceAttemptHTTP2: true,
|
||||||
@@ -276,7 +279,7 @@ func DoRequestWithRetry(client *http.Client, req *http.Request, config RetryConf
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
lastErr = err
|
lastErr = err
|
||||||
|
|
||||||
if CheckAndLogISPBlocking(err, reqCopy.URL.String(), "HTTP") {
|
if isHardConnectivityBlock(err) {
|
||||||
return nil, WrapErrorWithISPCheck(err, reqCopy.URL.String(), "HTTP")
|
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 {
|
if resp.StatusCode >= 500 {
|
||||||
resp.Body.Close()
|
resp.Body.Close()
|
||||||
|
if retryAfter := getRetryAfterDuration(resp); retryAfter > 0 {
|
||||||
|
delay = retryAfter
|
||||||
|
}
|
||||||
lastErr = fmt.Errorf("server error: HTTP %d", resp.StatusCode)
|
lastErr = fmt.Errorf("server error: HTTP %d", resp.StatusCode)
|
||||||
if attempt < config.MaxRetries {
|
if attempt < config.MaxRetries {
|
||||||
GoLog("[HTTP] Server error %d, retrying in %v...\n", resp.StatusCode, delay)
|
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)
|
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 {
|
func calculateNextDelay(currentDelay time.Duration, config RetryConfig) time.Duration {
|
||||||
nextDelay := time.Duration(float64(currentDelay) * config.BackoffFactor)
|
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
|
// 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
|
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 {
|
func IsISPBlocking(err error, requestURL string) *ISPBlockingError {
|
||||||
if err == nil {
|
if err == nil {
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -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" {
|
if body, err := ReadResponseBody(&http.Response{Body: io.NopCloser(strings.NewReader("ok"))}); err != nil || string(body) != "ok" {
|
||||||
t.Fatalf("ReadResponseBody = %q/%v", body, err)
|
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 {
|
if calculateNextDelay(10*time.Millisecond, RetryConfig{BackoffFactor: 3, MaxDelay: 20 * time.Millisecond}) != 20*time.Millisecond {
|
||||||
t.Fatal("calculateNextDelay mismatch")
|
t.Fatal("calculateNextDelay mismatch")
|
||||||
}
|
}
|
||||||
|
jitterFloat = origJitter
|
||||||
if getRetryAfterDuration(&http.Response{Header: http.Header{"Retry-After": []string{"bad"}}}) != 0 {
|
if getRetryAfterDuration(&http.Response{Header: http.Header{"Retry-After": []string{"bad"}}}) != 0 {
|
||||||
t.Fatal("invalid retry-after should be zero")
|
t.Fatal("invalid retry-after should be zero")
|
||||||
}
|
}
|
||||||
|
|||||||
+80
-23
@@ -4,27 +4,38 @@ package gobackend
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"crypto/tls"
|
|
||||||
"io"
|
"io"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
|
|
||||||
utls "github.com/refraction-networking/utls"
|
utls "github.com/refraction-networking/utls"
|
||||||
"golang.org/x/net/http2"
|
"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 {
|
type utlsTransport struct {
|
||||||
dialer *net.Dialer
|
dialer *net.Dialer
|
||||||
|
h2 *http2.Transport
|
||||||
|
mu sync.Mutex
|
||||||
|
conns map[string]*http2.ClientConn
|
||||||
}
|
}
|
||||||
|
|
||||||
func newUTLSTransport() *utlsTransport {
|
func newUTLSTransport() *utlsTransport {
|
||||||
return &utlsTransport{
|
return &utlsTransport{
|
||||||
dialer: &net.Dialer{
|
dialer: &net.Dialer{
|
||||||
Timeout: 30 * Second,
|
Timeout: 10 * Second,
|
||||||
KeepAlive: 30 * 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()
|
host := req.URL.Hostname()
|
||||||
port := t.getPort(req.URL)
|
addr := net.JoinHostPort(host, t.getPort(req.URL))
|
||||||
addr := net.JoinHostPort(host, port)
|
|
||||||
|
|
||||||
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 {
|
if err != nil {
|
||||||
return nil, err
|
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()
|
opts := GetNetworkCompatibilityOptions()
|
||||||
tlsConn := utls.UClient(conn, &utls.Config{
|
tlsConn := utls.UClient(conn, &utls.Config{
|
||||||
RootCAs: supplementalRootCAs(),
|
RootCAs: supplementalRootCAs(),
|
||||||
InsecureSkipVerify: opts.InsecureTLS,
|
InsecureSkipVerify: opts.InsecureTLS,
|
||||||
ServerName: host,
|
ServerName: host,
|
||||||
NextProtos: []string{"h2", "http/1.1"},
|
NextProtos: []string{"h2", "http/1.1"},
|
||||||
|
ClientSessionCache: utlsSessionCache,
|
||||||
}, utls.HelloChrome_Auto)
|
}, utls.HelloChrome_Auto)
|
||||||
|
|
||||||
if err := tlsConn.Handshake(); err != nil {
|
if err := tlsConn.Handshake(); err != nil {
|
||||||
conn.Close()
|
conn.Close()
|
||||||
return nil, err
|
return nil, "", err
|
||||||
}
|
}
|
||||||
|
return tlsConn, tlsConn.ConnectionState().NegotiatedProtocol, nil
|
||||||
|
}
|
||||||
|
|
||||||
negotiatedProto := tlsConn.ConnectionState().NegotiatedProtocol
|
func (t *utlsTransport) cachedConn(addr string) *http2.ClientConn {
|
||||||
|
t.mu.Lock()
|
||||||
if negotiatedProto == "h2" {
|
defer t.mu.Unlock()
|
||||||
h2Transport := &http2.Transport{
|
if cc := t.conns[addr]; cc != nil && cc.CanTakeNewRequest() {
|
||||||
DialTLSContext: func(ctx context.Context, network, addr string, cfg *tls.Config) (net.Conn, error) {
|
return cc
|
||||||
return tlsConn, nil
|
|
||||||
},
|
|
||||||
AllowHTTP: false,
|
|
||||||
DisableCompression: false,
|
|
||||||
}
|
|
||||||
return h2Transport.RoundTrip(req)
|
|
||||||
}
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
transport := &http.Transport{
|
func (t *utlsTransport) invalidate(addr string, cc *http2.ClientConn) {
|
||||||
DialTLSContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
|
t.mu.Lock()
|
||||||
return tlsConn, nil
|
if t.conns[addr] == cc {
|
||||||
},
|
delete(t.conns, addr)
|
||||||
DisableKeepAlives: true,
|
|
||||||
}
|
}
|
||||||
|
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 {
|
func (t *utlsTransport) getPort(u *url.URL) string {
|
||||||
|
|||||||
+96
-2
@@ -9,6 +9,7 @@ import (
|
|||||||
"net/url"
|
"net/url"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
type SongLinkClient struct {
|
type SongLinkClient struct {
|
||||||
@@ -294,18 +295,111 @@ func (s *SongLinkClient) doSongLinkRequest(apiURL string) (map[string]songLinkPl
|
|||||||
return songLinkResp.LinksByPlatform, nil
|
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) {
|
func (s *SongLinkClient) CheckTrackAvailability(spotifyTrackID string, isrc string) (*TrackAvailability, error) {
|
||||||
spotifyTrackID = strings.TrimSpace(spotifyTrackID)
|
spotifyTrackID = strings.TrimSpace(spotifyTrackID)
|
||||||
isrc = strings.ToUpper(strings.TrimSpace(isrc))
|
isrc = strings.ToUpper(strings.TrimSpace(isrc))
|
||||||
|
|
||||||
|
var idKey string
|
||||||
switch {
|
switch {
|
||||||
case spotifyTrackID != "":
|
case spotifyTrackID != "":
|
||||||
return s.checkTrackAvailabilityFromSpotify(spotifyTrackID)
|
idKey = "spotify:" + spotifyTrackID
|
||||||
case isrc != "":
|
case isrc != "":
|
||||||
return s.checkTrackAvailabilityFromISRC(isrc)
|
idKey = "isrc:" + isrc
|
||||||
default:
|
default:
|
||||||
return nil, fmt.Errorf("spotify track ID and ISRC are empty")
|
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) {
|
func (s *SongLinkClient) checkTrackAvailabilityFromSpotify(spotifyTrackID string) (*TrackAvailability, error) {
|
||||||
|
|||||||
@@ -4,7 +4,9 @@ import (
|
|||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
type roundTripFunc func(*http.Request) (*http.Response, error)
|
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) {
|
func TestCheckTrackAvailabilityFromSpotifyViaResolveAPI(t *testing.T) {
|
||||||
|
resetTrackAvailabilityCache()
|
||||||
origRetryConfig := songLinkRetryConfig
|
origRetryConfig := songLinkRetryConfig
|
||||||
defer func() { songLinkRetryConfig = origRetryConfig }()
|
defer func() { songLinkRetryConfig = origRetryConfig }()
|
||||||
|
|
||||||
@@ -65,6 +74,7 @@ func TestCheckTrackAvailabilityFromSpotifyViaResolveAPI(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestCheckTrackAvailabilityFromSpotifyResolveAPIFailure(t *testing.T) {
|
func TestCheckTrackAvailabilityFromSpotifyResolveAPIFailure(t *testing.T) {
|
||||||
|
resetTrackAvailabilityCache()
|
||||||
origRetryConfig := songLinkRetryConfig
|
origRetryConfig := songLinkRetryConfig
|
||||||
songLinkRetryConfig = func() RetryConfig {
|
songLinkRetryConfig = func() RetryConfig {
|
||||||
return RetryConfig{MaxRetries: 0, InitialDelay: 0, MaxDelay: 0, BackoffFactor: 1}
|
return RetryConfig{MaxRetries: 0, InitialDelay: 0, MaxDelay: 0, BackoffFactor: 1}
|
||||||
@@ -115,6 +125,7 @@ func TestCheckTrackAvailabilityFromSpotifyResolveAPIFailure(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestCheckTrackAvailabilityFromSpotifyViaResolveAPIMixedSongURLShapes(t *testing.T) {
|
func TestCheckTrackAvailabilityFromSpotifyViaResolveAPIMixedSongURLShapes(t *testing.T) {
|
||||||
|
resetTrackAvailabilityCache()
|
||||||
origRetryConfig := songLinkRetryConfig
|
origRetryConfig := songLinkRetryConfig
|
||||||
defer func() { songLinkRetryConfig = origRetryConfig }()
|
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) {
|
func TestCheckAvailabilityFromDeezerUsesSongLink(t *testing.T) {
|
||||||
origRetryConfig := songLinkRetryConfig
|
origRetryConfig := songLinkRetryConfig
|
||||||
songLinkRetryConfig = func() RetryConfig {
|
songLinkRetryConfig = func() RetryConfig {
|
||||||
|
|||||||
@@ -196,6 +196,9 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
|
|||||||
final Set<String> _ensuredDirs = {};
|
final Set<String> _ensuredDirs = {};
|
||||||
int _idleProgressPollTick = 0;
|
int _idleProgressPollTick = 0;
|
||||||
bool _networkPausedByWifiOnly = false;
|
bool _networkPausedByWifiOnly = false;
|
||||||
|
List<ConnectivityResult>? _lastConnectivityResults;
|
||||||
|
DateTime _lastConnectionCleanupAt = DateTime.fromMillisecondsSinceEpoch(0);
|
||||||
|
static const _connectionCleanupDebounce = Duration(seconds: 2);
|
||||||
String? _lastServiceTrackName;
|
String? _lastServiceTrackName;
|
||||||
String? _lastServiceArtistName;
|
String? _lastServiceArtistName;
|
||||||
String? _lastServiceStatus;
|
String? _lastServiceStatus;
|
||||||
@@ -1829,18 +1832,59 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
final shouldResume = _networkPausedByWifiOnly && state.isPaused;
|
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) {
|
if (shouldResume) {
|
||||||
resumeQueue();
|
resumeQueue();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
void _handleConnectivityResults(List<ConnectivityResult> results) {
|
/// Closes idle backend HTTP connections when the connectivity set changes
|
||||||
final settings = ref.read(settingsProvider);
|
/// (e.g. WiFi <-> cellular). Stale sockets bound to the old interface would
|
||||||
if (settings.downloadNetworkMode != 'wifi_only') {
|
/// otherwise be reused, stalling the first request after a network switch.
|
||||||
_handleDownloadNetworkModeChanged(settings.downloadNetworkMode);
|
/// Applies to every download network mode. Fire-and-forget with a light
|
||||||
|
/// debounce so network flapping does not spam the bridge.
|
||||||
|
void _maybeCleanupOnNetworkChange(List<ConnectivityResult> results) {
|
||||||
|
final previous = _lastConnectivityResults;
|
||||||
|
final unchanged =
|
||||||
|
previous != null && _connectivitySetEquals(previous, results);
|
||||||
|
_lastConnectivityResults = List<ConnectivityResult>.unmodifiable(results);
|
||||||
|
if (previous == null || unchanged) return;
|
||||||
|
|
||||||
|
final now = DateTime.now();
|
||||||
|
if (now.difference(_lastConnectionCleanupAt) < _connectionCleanupDebounce) {
|
||||||
return;
|
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<ConnectivityResult> a,
|
||||||
|
List<ConnectivityResult> b,
|
||||||
|
) {
|
||||||
|
final setA = a.toSet();
|
||||||
|
final setB = b.toSet();
|
||||||
|
return setA.length == setB.length && setA.containsAll(setB);
|
||||||
|
}
|
||||||
|
|
||||||
|
void _handleConnectivityResults(List<ConnectivityResult> results) {
|
||||||
|
_maybeCleanupOnNetworkChange(results);
|
||||||
|
|
||||||
|
final settings = ref.read(settingsProvider);
|
||||||
|
if (settings.downloadNetworkMode != 'wifi_only') return;
|
||||||
|
|
||||||
if (_hasWifiConnection(results)) {
|
if (_hasWifiConnection(results)) {
|
||||||
if (_networkPausedByWifiOnly && state.isPaused) {
|
if (_networkPausedByWifiOnly && state.isPaused) {
|
||||||
@@ -1921,11 +1965,11 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
|
|||||||
state = state.copyWith(isProcessing: false, isPaused: true);
|
state = state.copyWith(isProcessing: false, isPaused: true);
|
||||||
return;
|
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)) {
|
if (await _tryProcessQueueWithAndroidNativeWorker(settings)) {
|
||||||
return;
|
return;
|
||||||
|
|||||||
Reference in New Issue
Block a user