Fix imported playlists failing every download; add coordinated exponential backoff

Packages/ContinuityCore/Sources/ContinuityCore/IngestBackoff.swift

@@ -0,0 +1,79 @@
+import Foundation
+
+/// Exponential-backoff schedules for the ingest pipeline (resolve → download → analyse).
+///
+/// Pure math so the policy is unit-testable without a network: callers pass the attempt number
+/// and a random fraction, and get back the delay to sleep before the next attempt.
+///
+/// **Equal jitter, not full jitter.** Full jitter (`random(0, window)`) can return ~0s, which is
+/// exactly wrong against a rate limiter — the retry lands inside the same throttle window and
+/// extends it. Equal jitter keeps half the window as a hard floor and randomises the other half,
+/// so retries still de-synchronise across concurrently-importing tracks without ever hot-looping.
+public enum IngestBackoff {
+
+    /// One backoff curve: `base · multiplier^(attempt-1)`, clamped to `cap`, then jittered.
+    public struct Policy: Sendable, Equatable {
+        /// Delay window after the first failure.
+        public let base: TimeInterval
+        /// Growth factor per additional failure.
+        public let multiplier: Double
+        /// Upper bound on the window, before jitter.
+        public let cap: TimeInterval
+        /// Share of the window that is randomised (0 = fixed delay, 1 = full jitter).
+        public let jitterFraction: Double
+
+        public init(base: TimeInterval, multiplier: Double = 2, cap: TimeInterval, jitterFraction: Double = 0.5) {
+            self.base = max(0, base)
+            self.multiplier = max(1, multiplier)
+            self.cap = max(0, cap)
+            self.jitterFraction = min(max(0, jitterFraction), 1)
+        }
+
+        /// Transient blips on a single scrape/extract call (YouTubeKit churn, a dropped connection).
+        /// Short — the user is watching an import progress.
+        public static let request = Policy(base: 0.8, cap: 12)
+
+        /// HTTP 429. A rate limit does not clear in a second, and retrying inside its window is
+        /// what turns a throttle into a longer throttle — so this starts far out and climbs hard.
+        public static let rateLimited = Policy(base: 6, cap: 120)
+
+        /// Whole-track re-attempt after its per-call retries were exhausted. Minutes-scale: the
+        /// point is to outlast a bot-detection window, not to re-hammer it.
+        public static let track = Policy(base: 20, cap: 900)
+
+        /// Process-wide cool-down armed when the source starts throttling us, shared by every
+        /// in-flight track so an import backs off as one client rather than N independent ones.
+        public static let sourceCooldown = Policy(base: 4, cap: 300)
+    }
+
+    /// Delay to wait *after* `attempt` has failed, before making attempt `attempt + 1`.
+    /// `attempt` is 1-based; anything below 1 is treated as the first attempt.
+    ///
+    /// `randomFraction` must be in `0...1` — injected so tests are deterministic.
+    public static func delay(
+        afterAttempt attempt: Int,
+        policy: Policy,
+        randomFraction: Double = Double.random(in: 0...1)
+    ) -> TimeInterval {
+        let window = window(afterAttempt: attempt, policy: policy)
+        let fixed = window * (1 - policy.jitterFraction)
+        let jittered = window * policy.jitterFraction * min(max(0, randomFraction), 1)
+        return fixed + jittered
+    }
+
+    /// The un-jittered delay window after `attempt` failures — `base · multiplier^(attempt-1)`,
+    /// clamped to `cap`. Exposed for callers that want to report "retrying in ~Ns".
+    public static func window(afterAttempt attempt: Int, policy: Policy) -> TimeInterval {
+        let exponent = Double(max(1, attempt) - 1)
+        // pow overflows to .infinity for large exponents; min() with a finite cap still works,
+        // but guard anyway so a runaway attempt counter can't produce NaN downstream.
+        let grown = policy.base * pow(policy.multiplier, exponent)
+        guard grown.isFinite else { return policy.cap }
+        return min(grown, policy.cap)
+    }
+
+    /// Whether `attempt` (1-based) was the last one allowed by a budget of `maxAttempts`.
+    public static func isFinalAttempt(_ attempt: Int, maxAttempts: Int) -> Bool {
+        attempt >= max(1, maxAttempts)
+    }
+}

Packages/ContinuityCore/Sources/ContinuityCore/YouTubeSearch.swift

@@ -9,14 +9,35 @@ import Foundation
 /// changes the path. **No networking** — the app layer fetches the HTML; pinned by unit tests.
 public enum YouTubeSearch {
 
-    /// The video ID of the top result, or nil if none is found.
-    public static func firstVideoID(html: String) -> String? {
+    /// What a fetched search page actually told us. A bare `nil` video ID conflates two opposite
+    /// situations — "YouTube answered, nothing matched" (permanent) and "YouTube didn't send us a
+    /// results page at all" (transient: bot wall, consent interstitial, throttle page). Callers
+    /// must retry the second and must not retry the first.
+    public enum PageOutcome: Equatable, Sendable {
+        /// Top-ranked result.
+        case found(String)
+        /// A genuine results page that contained no usable video. Retrying won't change it.
+        case noResults
+        /// No parseable `ytInitialData` — this wasn't a results page. Worth retrying with backoff.
+        case unreadable
+    }
+
+    /// Classifies a fetched search results page.
+    public static func outcome(html: String) -> PageOutcome {
         guard let json = YouTubePlaylist.extractBracedObject(in: html, afterMarker: "ytInitialData"),
               let data = json.data(using: .utf8),
               let root = try? JSONSerialization.jsonObject(with: data) else {
-            return nil
+            return .unreadable
         }
-        return rankedFirstVideoID(root) ?? anyVideoID(root)
+        if let id = rankedFirstVideoID(root) ?? anyVideoID(root) { return .found(id) }
+        return .noResults
+    }
+
+    /// The video ID of the top result, or nil if none is found.
+    /// Prefer `outcome(html:)` when you need to tell "no match" from "no results page".
+    public static func firstVideoID(html: String) -> String? {
+        if case .found(let id) = outcome(html: html) { return id }
+        return nil
     }
 
     /// Walks the ordered results arrays and returns the first video's ID (rank-correct).

Packages/ContinuityCore/Tests/ContinuityCoreTests/IngestBackoffTests.swift

@@ -0,0 +1,107 @@
+import XCTest
+@testable import ContinuityCore
+
+final class IngestBackoffTests: XCTestCase {
+
+    private let policy = IngestBackoff.Policy(base: 1, multiplier: 2, cap: 10, jitterFraction: 0.5)
+
+    // MARK: Window growth
+
+    func testWindowDoublesPerAttempt() {
+        XCTAssertEqual(IngestBackoff.window(afterAttempt: 1, policy: policy), 1, accuracy: 1e-9)
+        XCTAssertEqual(IngestBackoff.window(afterAttempt: 2, policy: policy), 2, accuracy: 1e-9)
+        XCTAssertEqual(IngestBackoff.window(afterAttempt: 3, policy: policy), 4, accuracy: 1e-9)
+        XCTAssertEqual(IngestBackoff.window(afterAttempt: 4, policy: policy), 8, accuracy: 1e-9)
+    }
+
+    func testWindowClampsToCap() {
+        XCTAssertEqual(IngestBackoff.window(afterAttempt: 5, policy: policy), 10, accuracy: 1e-9)
+        XCTAssertEqual(IngestBackoff.window(afterAttempt: 50, policy: policy), 10, accuracy: 1e-9)
+    }
+
+    /// A runaway attempt counter overflows `pow` to infinity; the cap must still hold.
+    func testWindowSurvivesOverflowingExponent() {
+        let window = IngestBackoff.window(afterAttempt: 5000, policy: policy)
+        XCTAssertTrue(window.isFinite)
+        XCTAssertEqual(window, 10, accuracy: 1e-9)
+    }
+
+    func testAttemptBelowOneTreatedAsFirst() {
+        XCTAssertEqual(IngestBackoff.window(afterAttempt: 0, policy: policy), 1, accuracy: 1e-9)
+        XCTAssertEqual(IngestBackoff.window(afterAttempt: -7, policy: policy), 1, accuracy: 1e-9)
+    }
+
+    // MARK: Jitter
+
+    /// The point of equal jitter: never returns ~0, so a retry can't land inside the same
+    /// throttle window it was throttled by.
+    func testEqualJitterKeepsHalfTheWindowAsAFloor() {
+        for attempt in 1...6 {
+            let window = IngestBackoff.window(afterAttempt: attempt, policy: policy)
+            let low = IngestBackoff.delay(afterAttempt: attempt, policy: policy, randomFraction: 0)
+            let high = IngestBackoff.delay(afterAttempt: attempt, policy: policy, randomFraction: 1)
+            XCTAssertEqual(low, window / 2, accuracy: 1e-9)
+            XCTAssertEqual(high, window, accuracy: 1e-9)
+        }
+    }
+
+    func testDelayIsMonotonicInRandomFraction() {
+        let a = IngestBackoff.delay(afterAttempt: 3, policy: policy, randomFraction: 0.25)
+        let b = IngestBackoff.delay(afterAttempt: 3, policy: policy, randomFraction: 0.75)
+        XCTAssertLessThan(a, b)
+    }
+
+    func testRandomFractionIsClamped() {
+        let window = IngestBackoff.window(afterAttempt: 2, policy: policy)
+        XCTAssertEqual(IngestBackoff.delay(afterAttempt: 2, policy: policy, randomFraction: -5), window / 2, accuracy: 1e-9)
+        XCTAssertEqual(IngestBackoff.delay(afterAttempt: 2, policy: policy, randomFraction: 5), window, accuracy: 1e-9)
+    }
+
+    func testZeroJitterFractionGivesFixedDelay() {
+        let fixed = IngestBackoff.Policy(base: 2, multiplier: 2, cap: 100, jitterFraction: 0)
+        XCTAssertEqual(IngestBackoff.delay(afterAttempt: 3, policy: fixed, randomFraction: 0), 8, accuracy: 1e-9)
+        XCTAssertEqual(IngestBackoff.delay(afterAttempt: 3, policy: fixed, randomFraction: 1), 8, accuracy: 1e-9)
+    }
+
+    func testFullJitterFractionSpansWholeWindow() {
+        let full = IngestBackoff.Policy(base: 2, multiplier: 2, cap: 100, jitterFraction: 1)
+        XCTAssertEqual(IngestBackoff.delay(afterAttempt: 1, policy: full, randomFraction: 0), 0, accuracy: 1e-9)
+        XCTAssertEqual(IngestBackoff.delay(afterAttempt: 1, policy: full, randomFraction: 1), 2, accuracy: 1e-9)
+    }
+
+    // MARK: Policy invariants
+
+    func testPolicyClampsDegenerateInputs() {
+        let p = IngestBackoff.Policy(base: -1, multiplier: 0.2, cap: -3, jitterFraction: 4)
+        XCTAssertEqual(p.base, 0)
+        XCTAssertEqual(p.multiplier, 1)     // never shrink
+        XCTAssertEqual(p.cap, 0)
+        XCTAssertEqual(p.jitterFraction, 1)
+    }
+
+    /// Rate limits must start far enough out that the retry isn't inside the same window —
+    /// the old code retried a 429 after 1.5s, which is what poisoned whole playlist imports.
+    func testRateLimitPolicyStartsWellBeyondTheThrottleWindow() {
+        let first = IngestBackoff.delay(afterAttempt: 1, policy: .rateLimited, randomFraction: 0)
+        XCTAssertGreaterThanOrEqual(first, 3)
+        XCTAssertGreaterThan(IngestBackoff.Policy.rateLimited.cap, IngestBackoff.Policy.request.cap)
+    }
+
+    func testTrackPolicyOutlastsABotDetectionWindow() {
+        // Four whole-track attempts should span minutes, not seconds.
+        let total = (1...4).reduce(0.0) {
+            $0 + IngestBackoff.delay(afterAttempt: $1, policy: .track, randomFraction: 0)
+        }
+        XCTAssertGreaterThan(total, 120)
+    }
+
+    // MARK: Budget
+
+    func testIsFinalAttempt() {
+        XCTAssertFalse(IngestBackoff.isFinalAttempt(1, maxAttempts: 3))
+        XCTAssertFalse(IngestBackoff.isFinalAttempt(2, maxAttempts: 3))
+        XCTAssertTrue(IngestBackoff.isFinalAttempt(3, maxAttempts: 3))
+        XCTAssertTrue(IngestBackoff.isFinalAttempt(9, maxAttempts: 3))
+        XCTAssertTrue(IngestBackoff.isFinalAttempt(1, maxAttempts: 0))  // degenerate budget → one shot
+    }
+}

Packages/ContinuityCore/Tests/ContinuityCoreTests/YouTubeSearchTests.swift

@@ -34,4 +34,46 @@ final class YouTubeSearchTests: XCTestCase {
         XCTAssertNil(YouTubeSearch.firstVideoID(html: "ytInitialData = {\"contents\":{}};"))
         XCTAssertNil(YouTubeSearch.firstVideoID(html: "<html>nothing</html>"))
     }
+
+    // MARK: Page outcome
+    //
+    // The distinction that matters to ingest: a real-but-empty results page is permanent, while
+    // a page with no `ytInitialData` is a bot wall / consent interstitial and must be retried.
+    // Collapsing both into `nil` is what failed every song of a fresh Spotify import.
+
+    func testOutcomeFindsTopRankedVideo() {
+        XCTAssertEqual(YouTubeSearch.outcome(html: searchHTML), .found("kPa7bsKwL-c"))
+    }
+
+    func testOutcomeReportsNoResultsForRealButEmptyPage() {
+        XCTAssertEqual(YouTubeSearch.outcome(html: "ytInitialData = {\"contents\":{}};"), .noResults)
+    }
+
+    func testOutcomeReportsUnreadableForBotWall() {
+        XCTAssertEqual(YouTubeSearch.outcome(html: "<html>nothing</html>"), .unreadable)
+    }
+
+    /// The consent interstitial YouTube serves to unrecognised clients: a full HTML page, no
+    /// `ytInitialData` anywhere. Must classify as retryable, not "no such song".
+    func testOutcomeReportsUnreadableForConsentInterstitial() {
+        let consent = """
+        <!DOCTYPE html><html><head><title>Before you continue to YouTube</title></head>
+        <body><form action="https://consent.youtube.com/save"><button>Accept all</button></form></body></html>
+        """
+        XCTAssertEqual(YouTubeSearch.outcome(html: consent), .unreadable)
+    }
+
+    /// Truncated/garbled JSON is equally unreadable — don't mistake it for an empty result set.
+    func testOutcomeReportsUnreadableForMalformedInitialData() {
+        XCTAssertEqual(YouTubeSearch.outcome(html: "ytInitialData = {\"contents\": [ broken"), .unreadable)
+    }
+
+    func testFirstVideoIDStaysConsistentWithOutcome() {
+        for html in [searchHTML, "ytInitialData = {\"contents\":{}};", "<html>nothing</html>"] {
+            switch YouTubeSearch.outcome(html: html) {
+            case .found(let id): XCTAssertEqual(YouTubeSearch.firstVideoID(html: html), id)
+            case .noResults, .unreadable: XCTAssertNil(YouTubeSearch.firstVideoID(html: html))
+            }
+        }
+    }
 }

Packages/ContinuityKit/Sources/Ingest/AudioDownloader.swift

@@ -1,5 +1,6 @@
 import Foundation
 import Domain
+import ContinuityCore
 
 /// Downloads a resolved audio stream to the on-disk cache.
 ///
@@ -73,11 +74,14 @@ final class AudioDownloader: AudioFileDownloading {
 
             try handle.close()
 
+            // A truncated or empty body is a transient server-side symptom (throttling, dropped
+            // connection), not a permanent property of the video — classify it retryable so the
+            // track-level backoff gets another pass rather than failing the row outright.
             if let totalSize, offset < totalSize {
-                throw IngestError.downloadFailed("incomplete download: \(offset)/\(totalSize) bytes")
+                throw IngestError.network("incomplete download: \(offset)/\(totalSize) bytes")
             }
             if offset == 0 {
-                throw IngestError.downloadFailed("empty download")
+                throw IngestError.network("empty download")
             }
         } catch {
             try? handle.close()
@@ -91,35 +95,59 @@ final class AudioDownloader: AudioFileDownloading {
     /// must not append a whole-file body at a non-zero offset.
     private func fetchChunk(url: URL, from: Int, to: Int) async throws -> (Data, Int?, Bool) {
         var lastError: Error?
-        for attempt in 0..<maxRetriesPerChunk {
+        for attempt in 1...max(1, maxRetriesPerChunk) {
+            // Share the app-wide cool-down: if the source is throttling other tracks' resolves,
+            // pushing chunk requests through anyway just deepens it.
+            await IngestThrottle.shared.gate()
             do {
                 var request = URLRequest(url: url)
                 request.setValue("bytes=\(from)-\(to)", forHTTPHeaderField: "Range")
                 let (data, response) = try await URLSession.shared.data(for: request)
 
                 guard let http = response as? HTTPURLResponse else {
+                    await IngestThrottle.shared.noteSuccess()
                     return (data, nil, false)
                 }
                 switch http.statusCode {
                 case 206:
                     // Partial content — the normal ranged path.
+                    await IngestThrottle.shared.noteSuccess()
                     let total = Self.totalSize(fromContentRange: http.value(forHTTPHeaderField: "Content-Range"))
                     return (data, total, false)
                 case 200:
                     // Server ignored Range and sent the whole file in one shot.
+                    await IngestThrottle.shared.noteSuccess()
                     return (data, data.count, true)
                 case 416:
                     // Requested range not satisfiable — we've already read past the end.
+                    await IngestThrottle.shared.noteSuccess()
                     return (Data(), nil, false)
+                case 403, 410:
+                    // Signed googlevideo URL went stale (they're short-lived, and a throttled
+                    // client gets them invalidated early). Retrying this URL is guaranteed to
+                    // fail; bail out immediately so the caller can re-resolve for a fresh one.
+                    throw IngestError.streamURLExpired
+                case 429:
+                    throw IngestError.rateLimited
                 default:
-                    throw IngestError.downloadFailed("HTTP \(http.statusCode)")
+                    throw IngestError.network("range fetch HTTP \(http.statusCode)")
                 }
+            } catch let error as IngestError where error.needsFreshStreamURL {
+                throw error     // pointless to retry — needs a different URL
             } catch {
                 lastError = error
-                // Linear backoff before retrying this chunk.
-                try? await Task.sleep(nanoseconds: UInt64(200_000_000) * UInt64(attempt + 1))
+                let ingestError = error as? IngestError
+                await IngestThrottle.shared.noteThrottled(isRateLimit: ingestError?.isRateLimited ?? false)
+                guard !IngestBackoff.isFinalAttempt(attempt, maxAttempts: maxRetriesPerChunk) else { break }
+                try Task.checkCancellation()
+                // Exponential backoff + jitter, matching the resolve path. The old linear
+                // 0.2/0.4/0.6s schedule burned its whole budget inside a single throttle window.
+                let policy: IngestBackoff.Policy = (ingestError?.isRateLimited ?? false) ? .rateLimited : .request
+                let delay = IngestBackoff.delay(afterAttempt: attempt, policy: policy)
+                try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
             }
         }
+        if let ingestError = lastError as? IngestError { throw ingestError }
         throw IngestError.downloadFailed(String(describing: lastError ?? IngestError.downloadFailed("range fetch failed")))
     }
 

Packages/ContinuityKit/Sources/Ingest/IngestContracts.swift

@@ -28,18 +28,32 @@ public enum IngestError: Error, Sendable {
     case network(String)
     /// HTTP 429 from the source — a retry after a short delay may succeed.
     case rateLimited
+    /// The `googlevideo` URL was rejected (403/410): its signature expired, or the source
+    /// invalidated it mid-download. Retrying the same URL can never work — the video must be
+    /// re-resolved for a fresh one.
+    case streamURLExpired
     /// The source was reached and understood, but has no usable content
     /// (private, empty, region-locked, or deleted). Retrying won't help.
     case sourceUnavailable
 
     /// Whether retrying the same request with backoff could plausibly succeed. Distinguishes a
     /// transient blip (worth retrying, and not the user's fault) from a definitively empty source.
+    ///
+    /// `.streamURLExpired` is deliberately excluded: it needs a *different* URL, not another
+    /// attempt at this one. `process` handles it by re-resolving.
     var isRetryable: Bool {
         switch self {
         case .network, .rateLimited: return true
         default: return false
         }
     }
+
+    /// Whether a fresh resolve could plausibly fix this — i.e. the failure is about the URL we
+    /// used, not about the video or the connection.
+    var needsFreshStreamURL: Bool {
+        if case .streamURLExpired = self { return true }
+        return false
+    }
 }
 
 /// A resolved YouTube playlist: its videos (in order) plus the playlist's own title.