diff --git a/go_backend/dns_doh.go b/go_backend/dns_doh.go index b353cf58..4850acf1 100644 --- a/go_backend/dns_doh.go +++ b/go_backend/dns_doh.go @@ -43,6 +43,11 @@ const ( dohCacheMinTTL = time.Minute dohCacheMaxTTL = 30 * time.Minute dohCacheErrorTTL = 30 * time.Second + // Match Go's net.Dialer fallback cadence: give the preferred address family + // a brief head start, then race the remaining vetted answers. We cannot hand + // the hostname back to net.Dialer because the socket must stay pinned to an + // address that already passed the private-network filter. + happyEyeballsFallbackDelay = 300 * time.Millisecond ) type dohCacheEntry struct { @@ -50,6 +55,13 @@ type dohCacheEntry struct { expiresAt time.Time } +type resolvedDialResult struct { + conn net.Conn + err error +} + +type dialContextFunc func(context.Context, string, string) (net.Conn, error) + var ( dohMu sync.Mutex dohCache = map[string]dohCacheEntry{} @@ -102,18 +114,180 @@ func dialResolvedIPs( ips []net.IP, initialErr error, ) (net.Conn, error) { - lastErr := initialErr - for _, ip := range ips { - conn, dialErr := dialer.DialContext(ctx, network, net.JoinHostPort(ip.String(), port)) - if dialErr == nil { - return conn, nil + ordered := interleaveDialIPs(ips, network) + return raceResolvedIPs( + ctx, + network, + host, + port, + ordered, + initialErr, + happyEyeballsFallbackDelay, + dialer.DialContext, + ) +} + +func raceResolvedIPs( + ctx context.Context, + network string, + host string, + port string, + ordered []net.IP, + initialErr error, + fallbackDelay time.Duration, + dial dialContextFunc, +) (net.Conn, error) { + if len(ordered) == 0 { + if initialErr != nil { + return nil, initialErr } - lastErr = dialErr + return nil, fmt.Errorf("no dialable address for %s", host) } - if lastErr == nil { - lastErr = fmt.Errorf("no dialable address for %s", host) + + raceCtx, cancel := context.WithCancel(ctx) + defer cancel() + results := make(chan resolvedDialResult, len(ordered)) + started := 0 + finished := 0 + lastErr := initialErr + + startNext := func() bool { + if started >= len(ordered) { + return false + } + ip := ordered[started] + started++ + go func() { + conn, err := dial( + raceCtx, + network, + net.JoinHostPort(ip.String(), port), + ) + // The channel is sized for every possible attempt, so each goroutine + // can always report exactly once. The winner path drains and closes any + // late successful connections after cancelling the race. + results <- resolvedDialResult{conn: conn, err: err} + }() + return true } - return nil, lastErr + + startNext() + timer := time.NewTimer(fallbackDelay) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + cancel() + drainDialResults(results, started-finished) + return nil, ctx.Err() + case result := <-results: + finished++ + if result.err == nil && result.conn != nil { + cancel() + drainDialResults(results, started-finished) + return result.conn, nil + } + if result.err != nil { + lastErr = result.err + } + if finished == len(ordered) { + if lastErr == nil { + lastErr = fmt.Errorf("no dialable address for %s", host) + } + return nil, lastErr + } + // A fast refusal should not wait for the fallback timer when there is + // no other connection attempt currently in flight. + if finished == started && startNext() { + resetTimer(timer, fallbackDelay) + } + case <-timer.C: + if startNext() && started < len(ordered) { + timer.Reset(fallbackDelay) + } + } + } +} + +func drainDialResults(results <-chan resolvedDialResult, count int) { + if count <= 0 { + return + } + go func() { + for i := 0; i < count; i++ { + result := <-results + if result.conn != nil { + result.conn.Close() + } + } + }() +} + +func resetTimer(timer *time.Timer, delay time.Duration) { + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(delay) +} + +// interleaveDialIPs preserves the resolver's preferred family while ensuring +// the first fallback uses the other family. This avoids waiting through every +// unreachable IPv6 address before trying IPv4 (and vice versa). +func interleaveDialIPs(ips []net.IP, network string) []net.IP { + ordered := make([]net.IP, 0, len(ips)) + var v4, v6 []net.IP + for _, ip := range ips { + if ip == nil { + continue + } + if ip.To4() != nil { + if network != "tcp6" && network != "udp6" { + v4 = append(v4, ip) + } + } else if network != "tcp4" && network != "udp4" { + v6 = append(v6, ip) + } + } + if len(v4) == 0 { + return append(ordered, v6...) + } + if len(v6) == 0 { + return append(ordered, v4...) + } + + firstV4 := false + for _, ip := range ips { + if ip == nil { + continue + } + firstV4 = ip.To4() != nil + break + } + for len(v4) > 0 || len(v6) > 0 { + if firstV4 { + if len(v4) > 0 { + ordered = append(ordered, v4[0]) + v4 = v4[1:] + } + if len(v6) > 0 { + ordered = append(ordered, v6[0]) + v6 = v6[1:] + } + } else { + if len(v6) > 0 { + ordered = append(ordered, v6[0]) + v6 = v6[1:] + } + if len(v4) > 0 { + ordered = append(ordered, v4[0]) + v4 = v4[1:] + } + } + } + return ordered } // dohResolve resolves host over DoH, IPv4 first. Failures are negative-cached diff --git a/go_backend/dns_doh_security_test.go b/go_backend/dns_doh_security_test.go index 5c83850f..33cf9cbc 100644 --- a/go_backend/dns_doh_security_test.go +++ b/go_backend/dns_doh_security_test.go @@ -2,6 +2,7 @@ package gobackend import ( "context" + "errors" "net" "strings" "testing" @@ -37,3 +38,92 @@ func TestFilterDialableIPsDropsEveryPrivateAnswer(t *testing.T) { t.Fatalf("unexpected filtered addresses: %v", filtered) } } + +func TestInterleaveDialIPsAlternatesAddressFamilies(t *testing.T) { + ordered := interleaveDialIPs([]net.IP{ + net.ParseIP("2001:db8::1"), + net.ParseIP("2001:db8::2"), + net.ParseIP("192.0.2.1"), + net.ParseIP("192.0.2.2"), + }, "tcp") + want := []string{"2001:db8::1", "192.0.2.1", "2001:db8::2", "192.0.2.2"} + if len(ordered) != len(want) { + t.Fatalf("ordered addresses = %v, want %v", ordered, want) + } + for i, ip := range ordered { + if ip.String() != want[i] { + t.Fatalf("ordered[%d] = %s, want %s", i, ip, want[i]) + } + } +} + +func TestRaceResolvedIPsFallsBackWithoutWaitingForPreferredFamilyTimeout(t *testing.T) { + preferredStarted := make(chan struct{}) + clientPeerClosed := make(chan struct{}) + dial := func(ctx context.Context, _ string, address string) (net.Conn, error) { + host, _, err := net.SplitHostPort(address) + if err != nil { + return nil, err + } + if net.ParseIP(host).To4() == nil { + close(preferredStarted) + <-ctx.Done() + return nil, ctx.Err() + } + client, peer := net.Pipe() + go func() { + <-ctx.Done() + peer.Close() + close(clientPeerClosed) + }() + return client, nil + } + + startedAt := time.Now() + conn, err := raceResolvedIPs( + context.Background(), + "tcp", + "dual-stack.example", + "443", + []net.IP{net.ParseIP("2001:db8::1"), net.ParseIP("192.0.2.1")}, + nil, + 10*time.Millisecond, + dial, + ) + if err != nil { + t.Fatalf("raceResolvedIPs returned error: %v", err) + } + defer conn.Close() + if elapsed := time.Since(startedAt); elapsed > 100*time.Millisecond { + t.Fatalf("fallback took %v, want <100ms", elapsed) + } + select { + case <-preferredStarted: + default: + t.Fatal("preferred address family was not attempted first") + } + select { + case <-clientPeerClosed: + case <-time.After(time.Second): + t.Fatal("losing dial was not cancelled") + } +} + +func TestRaceResolvedIPsReturnsLastErrorAfterFastFailures(t *testing.T) { + wantErr := errors.New("refused") + conn, err := raceResolvedIPs( + context.Background(), + "tcp", + "failed.example", + "443", + []net.IP{net.ParseIP("192.0.2.1"), net.ParseIP("192.0.2.2")}, + nil, + time.Second, + func(context.Context, string, string) (net.Conn, error) { + return nil, wantErr + }, + ) + if conn != nil || !errors.Is(err, wantErr) { + t.Fatalf("result = (%v, %v), want (nil, %v)", conn, err, wantErr) + } +} diff --git a/go_backend/extension_runtime.go b/go_backend/extension_runtime.go index f339e905..dadacd83 100644 --- a/go_backend/extension_runtime.go +++ b/go_backend/extension_runtime.go @@ -486,7 +486,10 @@ func newExtensionHTTPClient(ext *loadedExtension, jar http.CookieJar, timeout ti GoLog("[Extension:%s] Redirect blocked: domain '%s' not in allowed list\n", ext.ID, domain) return &RedirectBlockedError{Domain: domain} } - if isPrivateIP(domain) { + // The transport resolves and pins every redirect target before dialing. + // Reject literals/local aliases here without doing a second, uncancellable + // DNS lookup on the redirect path. + if isPrivateIPLiteralOrLocal(domain) { GoLog("[Extension:%s] Redirect blocked: private IP '%s'\n", ext.ID, domain) return &RedirectBlockedError{Domain: domain, IsPrivate: true} } @@ -521,15 +524,10 @@ func isPrivateIP(host string) bool { if hostLower == "" { return false } - - if hostLower == "localhost" || strings.HasSuffix(hostLower, ".local") { + if isPrivateIPLiteralOrLocal(hostLower) { return true } - if ip := net.ParseIP(hostLower); ip != nil { - return isPrivateIPAddr(ip) - } - if cached, ok := getPrivateIPCache(hostLower); ok { return cached } @@ -548,6 +546,28 @@ func isPrivateIP(host string) bool { return isPrivate } +// isPrivateIPLiteralOrLocal performs the validation that does not require DNS. +// Extension HTTP requests use this before dispatch; dialWithDoHFallback remains +// the authoritative hostname check because it filters and pins the exact DNS +// answers used by the socket, closing the rebinding window without a duplicate +// lookup. +func isPrivateIPLiteralOrLocal(host string) bool { + if allowPrivateNetworkAccess.Load() { + return false + } + host = strings.ToLower(strings.TrimSpace(host)) + if host == "" { + return false + } + if host == "localhost" || strings.HasSuffix(host, ".local") { + return true + } + if ip := net.ParseIP(host); ip != nil { + return isPrivateIPAddr(ip) + } + return false +} + func getPrivateIPCache(host string) (bool, bool) { now := time.Now() diff --git a/go_backend/extension_runtime_http.go b/go_backend/extension_runtime_http.go index 988f0eba..32d0b811 100644 --- a/go_backend/extension_runtime_http.go +++ b/go_backend/extension_runtime_http.go @@ -63,7 +63,10 @@ func (r *extensionRuntime) validateDomain(urlStr string) error { return fmt.Errorf("invalid URL: hostname is required") } - if isPrivateIP(domain) { + // Hostname answers are filtered and pinned by transportDialContext. Avoid a + // second net.LookupIP here: it was uncancellable and the answer was discarded + // before the transport resolved the same host again. + if isPrivateIPLiteralOrLocal(domain) { return fmt.Errorf("network access denied: private/local network '%s' not allowed", domain) } diff --git a/go_backend/httputil.go b/go_backend/httputil.go index e3981f55..b7b11b0f 100644 --- a/go_backend/httputil.go +++ b/go_backend/httputil.go @@ -52,6 +52,11 @@ const ( DefaultMaxRetries = 3 DefaultRetryDelay = 1 * time.Second Second = time.Second + // Error responses are diagnostic data, not media payloads. Keeping this + // bounded prevents a hostile intermediary from turning a retry/status path + // into a large allocation while still allowing small bodies to be drained + // for HTTP keep-alive reuse. + maxRetryResponseBodyBytes = int64(64 << 10) ) type NetworkCompatibilityOptions struct { @@ -310,7 +315,7 @@ func DoRequestWithRetry(client *http.Client, req *http.Request, config RetryConf } if resp.StatusCode == 429 { - resp.Body.Close() + drainAndCloseResponseBody(resp.Body, maxRetryResponseBodyBytes) retryAfter := getRetryAfterDuration(resp) if retryAfter > 0 { delay = retryAfter @@ -327,8 +332,11 @@ func DoRequestWithRetry(client *http.Client, req *http.Request, config RetryConf } if resp.StatusCode == 403 || resp.StatusCode == 451 { - body, _ := io.ReadAll(resp.Body) + body, _ := io.ReadAll(io.LimitReader(resp.Body, maxRetryResponseBodyBytes+1)) resp.Body.Close() + if int64(len(body)) > maxRetryResponseBodyBytes { + body = body[:maxRetryResponseBodyBytes] + } bodyStr := strings.ToLower(string(body)) ispBlockingIndicators := []string{ @@ -352,7 +360,7 @@ func DoRequestWithRetry(client *http.Client, req *http.Request, config RetryConf } if resp.StatusCode >= 500 { - resp.Body.Close() + drainAndCloseResponseBody(resp.Body, maxRetryResponseBodyBytes) if retryAfter := getRetryAfterDuration(resp); retryAfter > 0 { delay = retryAfter } @@ -373,6 +381,16 @@ func DoRequestWithRetry(client *http.Client, req *http.Request, config RetryConf return nil, fmt.Errorf("request failed after %d retries: %w", config.MaxRetries+1, lastErr) } +func drainAndCloseResponseBody(body io.ReadCloser, limit int64) { + if body == nil { + return + } + if limit > 0 { + _, _ = io.Copy(io.Discard, io.LimitReader(body, limit)) + } + _ = body.Close() +} + // sleepRetry waits out a retry delay, aborting early when the request context // is cancelled so a cancelled download never sits in a backoff sleep. func sleepRetry(ctx context.Context, d time.Duration) error { diff --git a/go_backend/httputil_retry_test.go b/go_backend/httputil_retry_test.go index 3676cea3..3b0c6ae8 100644 --- a/go_backend/httputil_retry_test.go +++ b/go_backend/httputil_retry_test.go @@ -1,6 +1,7 @@ package gobackend import ( + "bytes" "context" "io" "net/http" @@ -49,3 +50,62 @@ func TestRetryHardening(t *testing.T) { t.Fatal("sleepRetry did not abort promptly on cancel") } } + +func TestRetryDrainsSmallFailureBodyForConnectionReuse(t *testing.T) { + failedBody := bytes.NewBufferString("temporary failure") + attempts := 0 + client := &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + attempts++ + if attempts == 1 { + return &http.Response{ + StatusCode: http.StatusServiceUnavailable, + Header: make(http.Header), + Body: io.NopCloser(failedBody), + Request: req, + }, nil + } + return &http.Response{ + StatusCode: http.StatusNoContent, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader("")), + Request: req, + }, nil + })} + resp, err := DoRequestWithRetry( + client, + mustNewRequest(t, "https://example.com/reuse"), + RetryConfig{MaxRetries: 1}, + ) + if err != nil || resp.StatusCode != http.StatusNoContent { + t.Fatalf("DoRequestWithRetry = %#v/%v", resp, err) + } + resp.Body.Close() + if failedBody.Len() != 0 { + t.Fatalf("retry body retained %d unread bytes", failedBody.Len()) + } +} + +func TestRetryCapsInspectedForbiddenBody(t *testing.T) { + huge := strings.Repeat("x", int(maxRetryResponseBodyBytes)+1024) + client := &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusForbidden, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(huge)), + Request: req, + }, nil + })} + resp, err := DoRequestWithRetry( + client, + mustNewRequest(t, "https://example.com/capped"), + RetryConfig{MaxRetries: 0}, + ) + if err != nil { + t.Fatalf("DoRequestWithRetry returned error: %v", err) + } + body, readErr := io.ReadAll(resp.Body) + resp.Body.Close() + if readErr != nil || int64(len(body)) != maxRetryResponseBodyBytes { + t.Fatalf("capped body length = %d/%v, want %d", len(body), readErr, maxRetryResponseBodyBytes) + } +} diff --git a/go_backend/ratelimit.go b/go_backend/ratelimit.go index 91078c86..be0159f7 100644 --- a/go_backend/ratelimit.go +++ b/go_backend/ratelimit.go @@ -1,6 +1,7 @@ package gobackend import ( + "context" "sync" "time" ) @@ -21,31 +22,43 @@ func NewRateLimiter(maxRequests int, window time.Duration) *RateLimiter { } func (r *RateLimiter) WaitForSlot() { - r.mu.Lock() - defer r.mu.Unlock() + _ = r.WaitForSlotContext(context.Background()) - now := time.Now() +} - r.cleanOldTimestamps(now) - - if len(r.timestamps) < r.maxRequests { - r.timestamps = append(r.timestamps, now) - return +// WaitForSlotContext reserves exactly one slot, rechecking the window after +// every wake-up. Multiple waiters may wake together, but only the first one +// that reacquires the mutex can consume the newly available slot. +func (r *RateLimiter) WaitForSlotContext(ctx context.Context) error { + if ctx == nil { + ctx = context.Background() } - - oldestTimestamp := r.timestamps[0] - waitUntil := oldestTimestamp.Add(r.window) - waitDuration := waitUntil.Sub(now) - - if waitDuration > 0 { - r.mu.Unlock() - time.Sleep(waitDuration) + for { r.mu.Lock() + now := time.Now() + r.cleanOldTimestamps(now) + if len(r.timestamps) < r.maxRequests { + r.timestamps = append(r.timestamps, now) + r.mu.Unlock() + return nil + } - r.cleanOldTimestamps(time.Now()) + waitDuration := r.timestamps[0].Add(r.window).Sub(now) + r.mu.Unlock() + if waitDuration <= 0 { + continue + } + + timer := time.NewTimer(waitDuration) + select { + case <-ctx.Done(): + if !timer.Stop() { + <-timer.C + } + return ctx.Err() + case <-timer.C: + } } - - r.timestamps = append(r.timestamps, time.Now()) } func (r *RateLimiter) cleanOldTimestamps(now time.Time) { diff --git a/go_backend/ratelimit_test.go b/go_backend/ratelimit_test.go new file mode 100644 index 00000000..ef32b54a --- /dev/null +++ b/go_backend/ratelimit_test.go @@ -0,0 +1,66 @@ +package gobackend + +import ( + "context" + "errors" + "sync" + "testing" + "time" +) + +func TestRateLimiterRechecksCapacityAfterConcurrentWait(t *testing.T) { + const ( + waiters = 4 + window = 25 * time.Millisecond + ) + limiter := NewRateLimiter(1, window) + if !limiter.TryAcquire() { + t.Fatal("failed to consume initial slot") + } + + started := make(chan struct{}) + admitted := make(chan time.Time, waiters) + var ready sync.WaitGroup + ready.Add(waiters) + for i := 0; i < waiters; i++ { + go func() { + ready.Done() + <-started + limiter.WaitForSlot() + admitted <- time.Now() + }() + } + ready.Wait() + start := time.Now() + close(started) + + previous := start + for i := 0; i < waiters; i++ { + select { + case admittedAt := <-admitted: + if gap := admittedAt.Sub(previous); gap < window/2 { + t.Fatalf("waiters %d and %d admitted only %v apart", i, i+1, gap) + } + previous = admittedAt + case <-time.After(time.Second): + t.Fatal("timed out waiting for rate-limiter admission") + } + } +} + +func TestRateLimiterWaitCanBeCancelled(t *testing.T) { + limiter := NewRateLimiter(1, time.Hour) + if !limiter.TryAcquire() { + t.Fatal("failed to consume initial slot") + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + startedAt := time.Now() + err := limiter.WaitForSlotContext(ctx) + if !errors.Is(err, context.Canceled) { + t.Fatalf("WaitForSlotContext error = %v, want context.Canceled", err) + } + if elapsed := time.Since(startedAt); elapsed > 100*time.Millisecond { + t.Fatalf("cancelled wait took %v", elapsed) + } +} diff --git a/go_backend/songlink.go b/go_backend/songlink.go index a31289ab..fe22f900 100644 --- a/go_backend/songlink.go +++ b/go_backend/songlink.go @@ -9,10 +9,16 @@ import ( "strings" "sync" "time" + + "golang.org/x/sync/singleflight" ) type SongLinkClient struct { - client *http.Client + client *http.Client + requestFlight singleflight.Group + resolutionFlight singleflight.Group + availabilityFlight singleflight.Group + platformLinksFlight singleflight.Group } type songLinkPlatformLink struct { @@ -102,6 +108,16 @@ func (s *SongLinkClient) resolveTrackPlatforms(inputURL string) (map[string]song // SongLink is rate-limited or unavailable. IDHS accepts the same source URL and // returns a smaller but still useful set of verified platform links. func (s *SongLinkClient) resolveTrackPlatformsWithIDHS(inputURL string) (map[string]songLinkPlatformLink, error) { + value, err, _ := s.resolutionFlight.Do(inputURL, func() (any, error) { + return s.resolveTrackPlatformsWithIDHSUncoalesced(inputURL) + }) + if err != nil { + return nil, err + } + return cloneSongLinkPlatformLinks(value.(map[string]songLinkPlatformLink)), nil +} + +func (s *SongLinkClient) resolveTrackPlatformsWithIDHSUncoalesced(inputURL string) (map[string]songLinkPlatformLink, error) { links, songLinkErr := s.resolveTrackPlatforms(inputURL) if songLinkErr == nil { return links, nil @@ -157,8 +173,6 @@ func (s *SongLinkClient) resolveTrackPlatformsByPlatform(platform, entityType, e // songLinkByTargetURL calls the SongLink API with a target URL. func (s *SongLinkClient) songLinkByTargetURL(targetURL string) (map[string]songLinkPlatformLink, error) { - songLinkRateLimiter.WaitForSlot() - apiURL := fmt.Sprintf("%s?url=%s&userCountry=%s", songLinkBaseURL(), url.QueryEscape(targetURL), @@ -169,8 +183,6 @@ func (s *SongLinkClient) songLinkByTargetURL(targetURL string) (map[string]songL // songLinkByPlatform calls the SongLink API with platform + type + id (for non-Spotify platforms). func (s *SongLinkClient) songLinkByPlatform(platform, entityType, entityID string) (map[string]songLinkPlatformLink, error) { - songLinkRateLimiter.WaitForSlot() - apiURL := fmt.Sprintf("%s?platform=%s&type=%s&id=%s&userCountry=%s", songLinkBaseURL(), url.QueryEscape(platform), @@ -183,6 +195,20 @@ func (s *SongLinkClient) songLinkByPlatform(platform, entityType, entityID strin // doSongLinkRequest calls the SongLink API and parses the response. func (s *SongLinkClient) doSongLinkRequest(apiURL string) (map[string]songLinkPlatformLink, error) { + value, err, _ := s.requestFlight.Do(apiURL, func() (any, error) { + return s.doSongLinkRequestUncoalesced(apiURL) + }) + if err != nil { + return nil, err + } + return cloneSongLinkPlatformLinks(value.(map[string]songLinkPlatformLink)), nil +} + +func (s *SongLinkClient) doSongLinkRequestUncoalesced(apiURL string) (map[string]songLinkPlatformLink, error) { + // Reserve the rate-limit slot inside the singleflight owner. Waiters for an + // identical URL share this request without consuming the remaining budget. + songLinkRateLimiter.WaitForSlot() + req, err := http.NewRequest("GET", apiURL, nil) if err != nil { return nil, fmt.Errorf("failed to create SongLink request: %w", err) @@ -221,6 +247,17 @@ func (s *SongLinkClient) doSongLinkRequest(apiURL string) (map[string]songLinkPl return songLinkResp.LinksByPlatform, nil } +func cloneSongLinkPlatformLinks(links map[string]songLinkPlatformLink) map[string]songLinkPlatformLink { + if links == nil { + return nil + } + cloned := make(map[string]songLinkPlatformLink, len(links)) + for platform, link := range links { + cloned[platform] = link + } + return cloned +} + const ( trackAvailabilityCacheTTL = 30 * time.Minute trackAvailabilityNegCacheTTL = 5 * time.Minute @@ -265,20 +302,31 @@ func (s *SongLinkClient) CheckTrackAvailability(spotifyTrackID string, isrc stri return cloneTrackAvailability(cached), nil } - var availability *TrackAvailability - var err error - switch { - case spotifyTrackID != "": - availability, err = s.checkTrackAvailabilityFromSpotify(spotifyTrackID) - default: - availability, err = s.checkTrackAvailabilityFromISRC(isrc) - } + value, err, _ := s.availabilityFlight.Do(key, func() (any, error) { + // Another caller may have populated the cache while this caller was + // waiting to become the singleflight owner. + if cached, hit, cachedErr := trackAvailabilityCacheLookup(key); hit { + if cachedErr { + return nil, fmt.Errorf("track availability unavailable (cached)") + } + return cached, nil + } - trackAvailabilityCacheStore(key, availability, err) + var availability *TrackAvailability + var resolveErr error + switch { + case spotifyTrackID != "": + availability, resolveErr = s.checkTrackAvailabilityFromSpotify(spotifyTrackID) + default: + availability, resolveErr = s.checkTrackAvailabilityFromISRC(isrc) + } + trackAvailabilityCacheStore(key, availability, resolveErr) + return availability, resolveErr + }) if err != nil { return nil, err } - return cloneTrackAvailability(availability), nil + return cloneTrackAvailability(value.(*TrackAvailability)), nil } const trackPlatformLinksCacheMax = 200 @@ -319,12 +367,21 @@ func (s *SongLinkClient) GetTrackPlatformLinks(spotifyTrackID string, isrc strin return links, nil } - links, err := s.fetchTrackPlatformLinks(spotifyTrackID, isrc) - trackPlatformLinksCacheStore(key, links, err) + value, err, _ := s.platformLinksFlight.Do(key, func() (any, error) { + if links, hit, cachedErr := trackPlatformLinksCacheLookup(key); hit { + if cachedErr { + return nil, fmt.Errorf("track platform links unavailable (cached)") + } + return links, nil + } + links, fetchErr := s.fetchTrackPlatformLinks(spotifyTrackID, isrc) + trackPlatformLinksCacheStore(key, links, fetchErr) + return links, fetchErr + }) if err != nil { return nil, err } - return links, nil + return cloneStringMap(value.(map[string]string)), nil } func (s *SongLinkClient) fetchTrackPlatformLinks(spotifyTrackID string, isrc string) (map[string]string, error) { diff --git a/go_backend/songlink_test.go b/go_backend/songlink_test.go index c82107e6..2739e2b8 100644 --- a/go_backend/songlink_test.go +++ b/go_backend/songlink_test.go @@ -4,11 +4,143 @@ import ( "io" "net/http" "strings" + "sync" "sync/atomic" "testing" "time" ) +func TestSongLinkIdenticalRequestsAreCoalesced(t *testing.T) { + origRateLimiter := songLinkRateLimiter + // A single slot proves duplicate waiters join singleflight before reserving + // rate-limit capacity. Reserving first would block 15 workers for an hour. + songLinkRateLimiter = NewRateLimiter(1, time.Hour) + defer func() { songLinkRateLimiter = origRateLimiter }() + + var calls int32 + release := make(chan struct{}) + client := &SongLinkClient{client: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + atomic.AddInt32(&calls, 1) + <-release + body := `{"linksByPlatform":{"deezer":{"url":"https://www.deezer.com/track/123"}}}` + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(body)), + Request: req, + }, nil + })}} + + const workers = 16 + start := make(chan struct{}) + errs := make(chan error, workers) + var wg sync.WaitGroup + wg.Add(workers) + for range workers { + go func() { + defer wg.Done() + <-start + links, err := client.resolveTrackPlatforms("https://open.spotify.com/track/coalesced") + if err == nil && links["deezer"].URL == "" { + err = io.ErrUnexpectedEOF + } + errs <- err + }() + } + close(start) + + deadline := time.Now().Add(time.Second) + for atomic.LoadInt32(&calls) == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + time.Sleep(20 * time.Millisecond) + close(release) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatalf("coalesced request failed: %v", err) + } + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("network calls = %d, want 1", got) + } +} + +func TestSongLinkIdenticalFallbacksShareOneIDHSRequest(t *testing.T) { + origSongLinkLimiter := songLinkRateLimiter + origIDHSLimiter := idhsRateLimiter + origIDHSClient := NewIDHSClient() + songLinkRateLimiter = NewRateLimiter(1, time.Hour) + idhsRateLimiter = NewRateLimiter(1, time.Hour) + defer func() { + songLinkRateLimiter = origSongLinkLimiter + idhsRateLimiter = origIDHSLimiter + globalIDHSClient = origIDHSClient + }() + + var songLinkCalls atomic.Int32 + var idhsCalls atomic.Int32 + idhsStarted := make(chan struct{}) + releaseIDHS := make(chan struct{}) + globalIDHSClient = &IDHSClient{client: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + if idhsCalls.Add(1) == 1 { + close(idhsStarted) + } + <-releaseIDHS + return &http.Response{ + StatusCode: http.StatusOK, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader( + `{"type":"song","links":[{"type":"deezer","url":"https://www.deezer.com/track/123"}]}`, + )), + Request: req, + }, nil + })}} + client := &SongLinkClient{client: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + songLinkCalls.Add(1) + return &http.Response{ + StatusCode: http.StatusUnauthorized, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{}`)), + Request: req, + }, nil + })}} + + const workers = 16 + start := make(chan struct{}) + errs := make(chan error, workers) + for range workers { + go func() { + <-start + links, err := client.resolveTrackPlatformsWithIDHS("https://open.spotify.com/track/fallback-coalesced") + if err == nil && links["deezer"].URL == "" { + err = io.ErrUnexpectedEOF + } + errs <- err + }() + } + close(start) + select { + case <-idhsStarted: + case <-time.After(time.Second): + t.Fatal("IDHS fallback did not start") + } + time.Sleep(20 * time.Millisecond) + close(releaseIDHS) + for range workers { + if err := <-errs; err != nil { + t.Fatalf("coalesced fallback failed: %v", err) + } + } + if got := songLinkCalls.Load(); got != 1 { + t.Fatalf("SongLink calls = %d, want 1", got) + } + if got := idhsCalls.Load(); got != 1 { + t.Fatalf("IDHS calls = %d, want 1", got) + } +} + type roundTripFunc func(*http.Request) (*http.Response, error) func (fn roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {