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