From 0562593919e9b454dcaf14b3a495e5c24d437e27 Mon Sep 17 00:00:00 2001 From: sanylax_macbook Date: Tue, 21 Jul 2026 02:39:52 -0400 Subject: [PATCH] Fix imported playlists failing every download; add coordinated exponential backoff MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A freshly imported playlist enqueues every track at once, and each ran its own retry loop with no awareness of the others. When YouTube throttled the burst, all in-flight tracks retried into the same throttle window — extending it — then exhausted their ~2s budget (0.7s, 1.4s) within seconds of each other. The retries were the amplifier, not the cure. Failure was then terminal: process() marked the track `.failed`, and resumePreparation skips `.failed`, so one transient window killed the whole playlist until each row was tapped by hand. Spotify hit a second, worse path. YouTubeSearch.firstVideoID returns nil both when a real results page has no match and when the page carried no ytInitialData at all — i.e. a consent interstitial or bot wall, which is exactly what a 50-track burst provokes. Both collapsed into the non-retryable `.noPlayableStream`, failing the track permanently. Spotify tracks are the only ones that hit that endpoint (YouTube tracks already have a video ID) and make double the requests, hence "almost always Spotify, sometimes YouTube". - IngestBackoff (ContinuityCore): pure backoff math with equal jitter — full jitter can return ~0s, the wrong answer against a rate limiter. Four tuned policies (request / rateLimited / track / sourceCooldown). - IngestThrottle: process-wide cool-down so the app backs off as one client. Any throttle signal arms it; every other in-flight request waits it out. 250ms minimum spacing paces the initial burst. A stray success clears the streak but never yanks an armed cool-down back, which would oscillate. - Retry: gates on the shared throttle, real exponential backoff, 429s on their own slower curve. Bounded at 4 attempts since it holds an ingestLimiter slot. - PreparationQueue.process: a transient failure now stays `.pending` and reschedules on a ~5 minute curve plus any remaining shared cool-down instead of dying. Only non-retryable errors or an exhausted budget reach `.failed`. Re-fetches the Track by ID after the sleep rather than pinning a @Model for minutes. Adds retryFailedTracks(in:playlist:) for a bulk retry. - YouTubeSearch.PageOutcome splits .noResults (permanent) from .unreadable (bot wall → retryable). - AudioDownloader: exponential chunk backoff, 403/410 → new .streamURLExpired, which process handles by re-resolving once for a fresh signed URL. Truncated and empty bodies reclassified retryable. ContinuityCore: 187 tests pass (19 new). App target builds for the simulator. Co-Authored-By: Claude Opus 4.8 --- .../ContinuityCore/IngestBackoff.swift | 79 +++++++++++ .../ContinuityCore/YouTubeSearch.swift | 29 +++- .../IngestBackoffTests.swift | 107 +++++++++++++++ .../YouTubeSearchTests.swift | 42 ++++++ .../Sources/Ingest/AudioDownloader.swift | 40 +++++- .../Sources/Ingest/IngestContracts.swift | 14 ++ .../Sources/Ingest/IngestThrottle.swift | 92 +++++++++++++ .../Sources/Ingest/PreparationQueue.swift | 127 +++++++++++++++++- .../ContinuityKit/Sources/Ingest/Retry.swift | 52 ++++--- .../Ingest/YouTubeSearchResolver.swift | 15 ++- 10 files changed, 567 insertions(+), 30 deletions(-) create mode 100644 Packages/ContinuityCore/Sources/ContinuityCore/IngestBackoff.swift create mode 100644 Packages/ContinuityCore/Tests/ContinuityCoreTests/IngestBackoffTests.swift create mode 100644 Packages/ContinuityKit/Sources/Ingest/IngestThrottle.swift diff --git a/Packages/ContinuityCore/Sources/ContinuityCore/IngestBackoff.swift b/Packages/ContinuityCore/Sources/ContinuityCore/IngestBackoff.swift new file mode 100644 index 0000000..c7cff63 --- /dev/null +++ b/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) + } +} diff --git a/Packages/ContinuityCore/Sources/ContinuityCore/YouTubeSearch.swift b/Packages/ContinuityCore/Sources/ContinuityCore/YouTubeSearch.swift index c27452b..f77b54d 100644 --- a/Packages/ContinuityCore/Sources/ContinuityCore/YouTubeSearch.swift +++ b/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). diff --git a/Packages/ContinuityCore/Tests/ContinuityCoreTests/IngestBackoffTests.swift b/Packages/ContinuityCore/Tests/ContinuityCoreTests/IngestBackoffTests.swift new file mode 100644 index 0000000..f03344d --- /dev/null +++ b/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 + } +} diff --git a/Packages/ContinuityCore/Tests/ContinuityCoreTests/YouTubeSearchTests.swift b/Packages/ContinuityCore/Tests/ContinuityCoreTests/YouTubeSearchTests.swift index 3df96e8..7467544 100644 --- a/Packages/ContinuityCore/Tests/ContinuityCoreTests/YouTubeSearchTests.swift +++ b/Packages/ContinuityCore/Tests/ContinuityCoreTests/YouTubeSearchTests.swift @@ -34,4 +34,46 @@ final class YouTubeSearchTests: XCTestCase { XCTAssertNil(YouTubeSearch.firstVideoID(html: "ytInitialData = {\"contents\":{}};")) XCTAssertNil(YouTubeSearch.firstVideoID(html: "nothing")) } + + // 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: "nothing"), .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 = """ + Before you continue to YouTube +
+ """ + 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\":{}};", "nothing"] { + switch YouTubeSearch.outcome(html: html) { + case .found(let id): XCTAssertEqual(YouTubeSearch.firstVideoID(html: html), id) + case .noResults, .unreadable: XCTAssertNil(YouTubeSearch.firstVideoID(html: html)) + } + } + } } diff --git a/Packages/ContinuityKit/Sources/Ingest/AudioDownloader.swift b/Packages/ContinuityKit/Sources/Ingest/AudioDownloader.swift index a5a1eb6..8369c3c 100644 --- a/Packages/ContinuityKit/Sources/Ingest/AudioDownloader.swift +++ b/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.. now else { break } + let wait = readyAt.timeIntervalSince(now) + // `openAt` is only ever pushed out by other tasks, so re-checking after the sleep + // converges; the spacing term can only move by `minimumSpacing`. + try? await Task.sleep(nanoseconds: UInt64(wait * 1_000_000_000)) + if Task.isCancelled { break } + } + lastReleasedAt = Date() + } + + /// The source answered normally — clear the streak so the next blip starts from `base` again. + /// + /// Deliberately does **not** clear `openAt`: during a rate limit some requests still succeed + /// (a cached page, a different endpoint), and letting one of those cancel an armed cool-down + /// would release the whole import back into the throttle and oscillate. An armed cool-down + /// always expires on its own schedule. + func noteSuccess() { + failureStreak = 0 + } + + /// The source throttled or refused us. Arms (or extends) the shared cool-down. + /// + /// `isRateLimit` distinguishes an explicit 429 — which needs a much longer window than a + /// dropped connection — from a generic transient network failure. + func noteThrottled(isRateLimit: Bool) { + failureStreak += 1 + let policy: IngestBackoff.Policy = isRateLimit ? .rateLimited : .sourceCooldown + let delay = IngestBackoff.delay(afterAttempt: failureStreak, policy: policy) + let candidate = Date().addingTimeInterval(delay) + // Never pull the gate in: overlapping failures should compound, not reset each other. + if candidate > openAt { + openAt = candidate + Logger.ingest.notice( + "source cool-down \(delay, format: .fixed(precision: 1))s (streak \(self.failureStreak), rateLimit \(isRateLimit))" + ) + } + } + + /// How long callers must currently wait — used to decide whether a whole-track retry is worth + /// scheduling now or should be pushed past the cool-down. + func remainingCooldown() -> TimeInterval { + max(0, openAt.timeIntervalSince(Date())) + } + + /// Test seam: forget all throttle state. + func reset() { + failureStreak = 0 + openAt = .distantPast + lastReleasedAt = .distantPast + } +} diff --git a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue.swift b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue.swift index f110a86..14e4297 100644 --- a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue.swift +++ b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue.swift @@ -69,12 +69,38 @@ public final class PreparationQueue { /// /// Safe to call from the UI: it persists the pending state immediately so the row's /// badge updates, then detaches the resolve/download work into its own `Task`. + /// + /// User-initiated calls (tapping a failed row) reset the track's backoff — an explicit retry + /// means "try now", not "resume the curve where it left off". public func enqueue(_ track: Track, in context: ModelContext) { + ingestAttempts[track.id] = nil + retryScheduledTrackIDs.remove(track.id) + enqueueInternal(track, in: context) + } + + /// Enqueue without clearing the retry budget — used by the automatic backoff so a track that + /// keeps failing keeps climbing its curve instead of looping at `base` forever. + func enqueueInternal(_ track: Track, in context: ModelContext) { track.prepState = .pending try? context.save() Task { await process(track, in: context) } } + /// Re-enqueues every `.failed` track (optionally only within one playlist) with a fresh + /// backoff budget. Surfaces as the "Retry all" affordance after an import that hit a + /// sustained throttle. + public func retryFailedTracks(in context: ModelContext, playlist: Playlist? = nil) { + let candidates: [Track] + if let playlist { + candidates = playlist.tracks + } else { + candidates = (try? context.fetch(FetchDescriptor())) ?? [] + } + for track in candidates where track.prepState == .failed && !track.isDemo { + enqueue(track, in: context) + } + } + /// Resumes preparation for a persisted library at launch: re-enqueues tracks that were /// interrupted mid-ingest (e.g. the app was killed partway through a large import) or whose /// downloaded audio went missing, and finishes stem separation for tracks that have audio but @@ -143,6 +169,7 @@ public final class PreparationQueue { private func process(_ track: Track, in context: ModelContext) async { track.prepState = .preparing try? context.save() + retryScheduledTrackIDs.remove(track.id) // A track needs either a direct video ID (YouTube) or a search query (Spotify-sourced). guard track.youtubeVideoID != nil || track.searchQuery != nil else { @@ -154,6 +181,7 @@ public final class PreparationQueue { // Gate the network/CPU-heavy stage so a playlist import doesn't run all tracks at once. await ingestLimiter.acquire() var prepared = false + var failure: Error? do { // Resolve the video ID: use the known one, or find it on YouTube from the search query. let id: String @@ -167,7 +195,17 @@ public final class PreparationQueue { } let resolved = try await resolver.resolveAudio(videoID: id) - let fileURL = try await downloader.downloadAudio(resolved) + let fileURL: URL + do { + fileURL = try await downloader.downloadAudio(resolved) + } catch let error as IngestError where error.needsFreshStreamURL { + // Signed `googlevideo` URLs are short-lived, and a throttled client gets them + // invalidated early — so a queued track's URL can be dead by the time its turn + // comes. Retrying the dead URL can never work; re-resolve for a fresh one. + Logger.ingest.notice("stream URL expired for \(id, privacy: .public) — re-resolving") + let refreshed = try await resolver.resolveAudio(videoID: id) + fileURL = try await downloader.downloadAudio(refreshed) + } // The track could have been deleted while we were off the main actor; don't write to // (or resurrect) a dead model. if track.modelContext != nil { @@ -204,12 +242,27 @@ public final class PreparationQueue { "prep failed for \(label, privacy: .public): \(String(describing: error), privacy: .public)" ) prepared = false + failure = error } await ingestLimiter.release() // Don't touch a track that was deleted while we worked. - if track.modelContext != nil { - track.prepState = prepared ? .ready : .failed + guard track.modelContext != nil else { + ingestAttempts[track.id] = nil + try? context.save() + return + } + + if prepared { + ingestAttempts[track.id] = nil + track.prepState = .ready + } else if let failure, scheduleRetry(track, after: failure, in: context) { + // Stays `.pending`, not `.failed`: the row keeps its in-progress badge, and if the + // app is killed before the retry fires, `resumePreparation` picks it up at launch. + track.prepState = .pending + } else { + ingestAttempts[track.id] = nil + track.prepState = .failed } try? context.save() @@ -223,6 +276,74 @@ public final class PreparationQueue { } } + // MARK: - Track-level retry + + /// Consecutive failed ingest attempts per track, in memory (like `syncBackoff` for playlists). + /// Deliberately not persisted: a relaunch is itself a fresh start, and a `.pending` track is + /// re-enqueued by `resumePreparation` anyway. + var ingestAttempts: [UUID: Int] = [:] + + /// Tracks with a backoff retry already queued, so overlapping failures can't stack timers. + var retryScheduledTrackIDs: Set = [] + + /// Whole-track attempts before a track is finally marked `.failed`. With the `.track` curve + /// this spans roughly five minutes — long enough to outlast the bot-detection window that a + /// fresh playlist import trips. + static let maxIngestAttempts = 5 + + /// Schedules a backed-off re-attempt for a track whose ingest just failed, and reports whether + /// one was queued (`false` → the caller should mark the track `.failed`). + /// + /// This is the fix for the reported bug: a freshly imported playlist fires N resolves at once, + /// YouTube throttles, and every track used to burn its few seconds of retries inside the same + /// throttle window and land in `.failed` permanently — the whole playlist dead, recoverable + /// only by tapping each row. Now a transient failure keeps the track alive on an exponential + /// curve, and the shared `IngestThrottle` cool-down is added on top so the retry lands *after* + /// the source has stopped throttling rather than during. + private func scheduleRetry(_ track: Track, after error: Error, in context: ModelContext) -> Bool { + // Only transient failures earn a retry. A private/deleted video or a query that matches + // nothing will fail identically forever — surfacing that immediately is the honest answer. + guard let ingestError = error as? IngestError, + ingestError.isRetryable || ingestError.needsFreshStreamURL else { return false } + guard !retryScheduledTrackIDs.contains(track.id) else { return true } + + let attempt = (ingestAttempts[track.id] ?? 0) + 1 + ingestAttempts[track.id] = attempt + guard !IngestBackoff.isFinalAttempt(attempt, maxAttempts: Self.maxIngestAttempts) else { + return false + } + + let trackID = track.id + retryScheduledTrackIDs.insert(trackID) + let label = track.youtubeVideoID ?? track.searchQuery ?? track.title + + let maxAttempts = Self.maxIngestAttempts + Task { [weak self] in + // Wait out the curve, then whatever remains of the app-wide cool-down — a per-track + // delay alone would still let 50 tracks resume simultaneously into a live throttle. + let backoff = IngestBackoff.delay(afterAttempt: attempt, policy: .track) + try? await Task.sleep(nanoseconds: UInt64(backoff * 1_000_000_000)) + let cooldown = await IngestThrottle.shared.remainingCooldown() + if cooldown > 0 { + try? await Task.sleep(nanoseconds: UInt64(cooldown * 1_000_000_000)) + } + guard let self else { return } + guard self.retryScheduledTrackIDs.remove(trackID) != nil else { return } // superseded + // Re-fetch rather than capture the `@Model` across a minutes-long sleep — holding one + // (and its context) alive that long is a known memory-pressure pattern here. This also + // naturally drops tracks deleted while we waited. + var descriptor = FetchDescriptor(predicate: #Predicate { $0.id == trackID }) + descriptor.fetchLimit = 1 + guard let fresh = try? context.fetch(descriptor).first, + fresh.prepState != .ready else { return } // healed by a manual retry meanwhile + Logger.ingest.notice( + "retrying \(label, privacy: .public) (attempt \(attempt + 1)/\(maxAttempts))" + ) + self.enqueueInternal(fresh, in: context) + } + return true + } + /// Whether the track still shows the "YouTube Video (abc123)" placeholder from a bare-ID add. private static func hasPlaceholderMetadata(_ track: Track) -> Bool { track.youtubeVideoID != nil && track.title.hasPrefix("YouTube Video (") diff --git a/Packages/ContinuityKit/Sources/Ingest/Retry.swift b/Packages/ContinuityKit/Sources/Ingest/Retry.swift index 3412f45..6ca81f7 100644 --- a/Packages/ContinuityKit/Sources/Ingest/Retry.swift +++ b/Packages/ContinuityKit/Sources/Ingest/Retry.swift @@ -1,35 +1,55 @@ import Foundation +import ContinuityCore -/// Retries a transient-failing async operation with exponential backoff + jitter. +/// Retries a transient-failing async operation with exponential backoff + jitter, coordinated +/// across the whole app by `IngestThrottle`. /// -/// Only `IngestError`s that report themselves `isRetryable` (network blips, rate limits) are -/// retried; a definitively-empty source or any other error propagates immediately. The scraped -/// endpoints (Spotify embed, YouTube playlist page) fail transiently often enough that a single -/// blip shouldn't surface to the user as "playlist unavailable". +/// Only `IngestError`s that report themselves `isRetryable` (network blips, rate limits, expired +/// stream URLs) are retried; a definitively-empty source or any other error propagates +/// immediately. The scraped endpoints (Spotify embed, YouTube playlist/search pages, YouTubeKit +/// extraction) fail transiently often enough that a single blip shouldn't surface to the user as +/// "playlist unavailable". +/// +/// Two things matter here beyond "sleep and try again": +/// - **The gate is shared.** Every attempt waits on `IngestThrottle` first, so N concurrently +/// importing tracks back off together instead of each discovering the same rate limit alone. +/// - **The delay outlasts the throttle.** The previous schedule (0.7s then 1.4s) retried inside +/// the same window that had just rejected us, which extended the throttle and failed the track. enum Retry { static func run( - maxAttempts: Int = 3, - baseDelay: TimeInterval = 0.7, + // Bounded deliberately: this loop holds an `ingestLimiter` slot while it sleeps, so a + // long rate-limit curve here would stall the import. Past this budget the *track-level* + // backoff in `PreparationQueue.process` takes over, which retries without holding a slot. + maxAttempts: Int = 4, + policy: IngestBackoff.Policy = .request, _ operation: () async throws -> T ) async throws -> T { var attempt = 1 while true { + await IngestThrottle.shared.gate() do { - return try await operation() - } catch let error as IngestError where error.isRetryable && attempt < maxAttempts { - // Exponential backoff (0.7s, 1.4s, …) with up to 30% jitter so retries don't - // synchronize. Rate limits get a longer floor — they rarely clear in <1s. - let floor = (error.isRateLimited ? 1.5 : baseDelay) - let delay = floor * pow(2, Double(attempt - 1)) - let jitter = Double.random(in: 0...(delay * 0.3)) - try? await Task.sleep(nanoseconds: UInt64((delay + jitter) * 1_000_000_000)) + let value = try await operation() + await IngestThrottle.shared.noteSuccess() + return value + } catch let error as IngestError where error.isRetryable { + // Tell the shared gate first — even on the final attempt, so sibling tracks that + // are still going learn about the throttle from this failure. + await IngestThrottle.shared.noteThrottled(isRateLimit: error.isRateLimited) + guard !IngestBackoff.isFinalAttempt(attempt, maxAttempts: maxAttempts) else { throw error } + try Task.checkCancellation() + + // Rate limits escalate on their own, much slower curve; other transients use the + // caller's policy. The shared cool-down armed above is additive on top of this. + let effective = error.isRateLimited ? IngestBackoff.Policy.rateLimited : policy + let delay = IngestBackoff.delay(afterAttempt: attempt, policy: effective) + try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000)) attempt += 1 } } } } -private extension IngestError { +extension IngestError { var isRateLimited: Bool { if case .rateLimited = self { return true } return false diff --git a/Packages/ContinuityKit/Sources/Ingest/YouTubeSearchResolver.swift b/Packages/ContinuityKit/Sources/Ingest/YouTubeSearchResolver.swift index 25ccbba..e04b5f1 100644 --- a/Packages/ContinuityKit/Sources/Ingest/YouTubeSearchResolver.swift +++ b/Packages/ContinuityKit/Sources/Ingest/YouTubeSearchResolver.swift @@ -25,7 +25,20 @@ final class YouTubeSearchResolver: YouTubeSearching { // Spotify playlist imports fire many searches; a single 429/5xx/blip shouldn't fail a track. return try await Retry.run { let html = try await self.fetchSearchHTML(url) - return YouTubeSearch.firstVideoID(html: html) + switch YouTubeSearch.outcome(html: html) { + case .found(let id): + return id + case .noResults: + // A real results page with nothing in it — the query genuinely has no match. + return nil + case .unreadable: + // HTTP 200 but no `ytInitialData`: a consent interstitial or bot wall, which is + // what a burst of import searches provokes. This used to collapse into the same + // `nil` as `.noResults` and fail the track permanently — the main reason a fresh + // Spotify import failed every song. It's transient: retry it, and let the shared + // throttle back the whole import off. + throw IngestError.network("YouTube search page had no ytInitialData (bot wall?)") + } } }