From 34853fd7fab0c4a3916a54277bf12b45d7fa2e7b Mon Sep 17 00:00:00 2001 From: sanylaxq Date: Fri, 11 Sep 2026 15:34:14 -0700 Subject: [PATCH] Fix launch watchdog and Downloads ghost rows after delete Device crash logs showed repeated 0x8BADF00D kills inside resumePreparation enqueue; suspend per-row job sorts during bulk resume/import and delay resume until after scene-create. Also clear ingest jobs/waiters on delete so Downloads cannot keep forever-waiting ghosts. Co-authored-by: Cursor --- .gitignore | 1 + App/Continuity/Views/LibraryView.swift | 1 + App/Continuity/Views/PlaylistDetailView.swift | 5 ++- App/Continuity/Views/RootView.swift | 5 +++ .../Sources/Ingest/ConcurrencyLimiter.swift | 42 ++++++++++++++----- .../Ingest/PreparationQueue+Import.swift | 6 +++ .../Ingest/PreparationQueue+Jobs.swift | 22 ++++++++-- .../Ingest/PreparationQueue+Sync.swift | 4 +- .../Sources/Ingest/PreparationQueue.swift | 35 +++++++++++++--- .../IngestTests/ConcurrencyLimiterTests.swift | 34 +++++++++++++-- 10 files changed, 129 insertions(+), 26 deletions(-) diff --git a/.gitignore b/.gitignore index c73a746..31cfc2d 100644 --- a/.gitignore +++ b/.gitignore @@ -20,3 +20,4 @@ xcuserdata/ /Cache/ *.m4a *.caf +.debug-logs/ diff --git a/App/Continuity/Views/LibraryView.swift b/App/Continuity/Views/LibraryView.swift index 500905b..8697e1e 100644 --- a/App/Continuity/Views/LibraryView.swift +++ b/App/Continuity/Views/LibraryView.swift @@ -96,6 +96,7 @@ struct LibraryView: View { private func delete(_ playlist: Playlist) { let trackIDs = Set(playlist.tracks.map(\.id)) let keys = playlist.tracks.map(\.stemKey) + prepQueue.handleTracksDeleted(trackIDs) player.handleDeleted(trackIDs: trackIDs) modelContext.delete(playlist) // cascade deletes its tracks try? modelContext.save() diff --git a/App/Continuity/Views/PlaylistDetailView.swift b/App/Continuity/Views/PlaylistDetailView.swift index 45961e2..4662fcb 100644 --- a/App/Continuity/Views/PlaylistDetailView.swift +++ b/App/Continuity/Views/PlaylistDetailView.swift @@ -73,7 +73,10 @@ struct PlaylistDetailView: View { /// model goes, then any cached files no other track shares. private func delete(_ track: Track) { let key = track.stemKey - player.handleDeleted(trackIDs: [track.id]) + let id = track.id + // Prep queue first — clears Downloads ghosts / ingest waiters before the model dies. + prepQueue.handleTracksDeleted([id]) + player.handleDeleted(trackIDs: [id]) modelContext.delete(track) playlist.touch() // membership changed → resort the library try? modelContext.save() diff --git a/App/Continuity/Views/RootView.swift b/App/Continuity/Views/RootView.swift index 458bfe2..d57441c 100644 --- a/App/Continuity/Views/RootView.swift +++ b/App/Continuity/Views/RootView.swift @@ -105,6 +105,11 @@ struct RootView: View { // a large unfinished import occupies the main actor until the first frame. restorePlaybackSession() await Task.yield() + // Scene-create watchdog is 10s wall-clock (`0x8BADF00D`). A large `.pending` + // library used to keep `resumePreparation` on the main actor through that + // window (enqueue → SwiftData notify per track). Give the scene a beat to + // finish creating, then resume in chunks that yield. + try? await Task.sleep(for: .milliseconds(300)) await prepQueue.resumePreparation(in: modelContext) // Launch-time polling pass over source-backed playlists (per-playlist opt-out). prepQueue.autoSyncIfNeeded(in: modelContext) diff --git a/Packages/ContinuityKit/Sources/Ingest/ConcurrencyLimiter.swift b/Packages/ContinuityKit/Sources/Ingest/ConcurrencyLimiter.swift index 40b2e5e..0c69cbd 100644 --- a/Packages/ContinuityKit/Sources/Ingest/ConcurrencyLimiter.swift +++ b/Packages/ContinuityKit/Sources/Ingest/ConcurrencyLimiter.swift @@ -5,7 +5,8 @@ import Foundation /// /// Waiters are FIFO among equal `priority` values. A later `bump` reorders still-waiting /// acquirers so a user-prioritized track jumps the ingest queue without cancelling anyone -/// already downloading. +/// already downloading. `cancel` drops waiters without granting a slot (returns `false` from +/// `acquire`) so a deleted track can leave the queue without inflating concurrency. /// /// Used by `PreparationQueue` so importing a large playlist doesn't fire dozens of simultaneous /// resolves/downloads (network throttling) or stem separations (each loads a ~158 MB model and @@ -15,7 +16,7 @@ actor ConcurrencyLimiter { let id: UUID var priority: Int let sequence: Int - let continuation: CheckedContinuation + let continuation: CheckedContinuation } private let limit: Int @@ -27,18 +28,22 @@ actor ConcurrencyLimiter { self.limit = max(1, limit) } - /// Suspends until a slot is available, then claims it. Pair with exactly one `release()`. - func acquire() async { + /// Suspends until a slot is available, then claims it. Pair with exactly one `release()` + /// when the return value is `true`. A `false` return means `cancel` dropped this waiter — + /// do not call `release()`. + @discardableResult + func acquire() async -> Bool { await acquire(id: UUID(), priority: 0) } - /// Identified acquire so a later `bump` can move this waiter ahead of equal/lower priority. - func acquire(id: UUID, priority: Int) async { + /// Identified acquire so a later `bump` / `cancel` can target this waiter. + @discardableResult + func acquire(id: UUID, priority: Int) async -> Bool { if active < limit { active += 1 - return + return true } - await withCheckedContinuation { continuation in + return await withCheckedContinuation { continuation in waiters.append(Waiter( id: id, priority: priority, @@ -48,7 +53,7 @@ actor ConcurrencyLimiter { nextSequence += 1 sortWaiters() } - // Resumed by `release()`, which hands over its slot without touching `active`. + // Resumed by `release()` with `true` (slot handoff) or `cancel` with `false`. } /// Raises still-waiting acquirers in `ids` to at least `priority`. No-op for holders @@ -65,13 +70,30 @@ actor ConcurrencyLimiter { if changed { sortWaiters() } } + /// Drops still-waiting acquirers in `ids` without granting a slot. Holders already running + /// finish normally; their callers should stop after seeing the model is gone. + func cancel(ids: Set) { + guard !ids.isEmpty else { return } + var cancelled: [Waiter] = [] + waiters.removeAll { waiter in + if ids.contains(waiter.id) { + cancelled.append(waiter) + return true + } + return false + } + for waiter in cancelled { + waiter.continuation.resume(returning: false) + } + } + /// Frees a slot, waking the highest-priority waiter (FIFO among ties). func release() { if waiters.isEmpty { active = max(0, active - 1) } else { let next = waiters.removeFirst() - next.continuation.resume() + next.continuation.resume(returning: true) } } diff --git a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Import.swift b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Import.swift index 0ab2d66..af83828 100644 --- a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Import.swift +++ b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Import.swift @@ -31,6 +31,7 @@ extension PreparationQueue { playlist.lastSyncedAt = Date() context.insert(playlist) + jobSortSuspended += 1 for (index, item) in resolved.items.enumerated() { let track = Track( title: item.title ?? "YouTube Video (\(item.videoID.prefix(6)))", @@ -47,6 +48,8 @@ extension PreparationQueue { context.insert(track) enqueue(track, in: context, saving: false) } + jobSortSuspended = max(0, jobSortSuspended - 1) + sortJobs() playlist.touch() // creation + initial tracks count as a content change try? context.save() return playlist @@ -76,6 +79,7 @@ extension PreparationQueue { playlist.lastSyncedAt = Date() context.insert(playlist) + jobSortSuspended += 1 for (index, spotifyTrack) in resolved.tracks.enumerated() { let track = Track( title: spotifyTrack.title, @@ -92,6 +96,8 @@ extension PreparationQueue { context.insert(track) enqueue(track, in: context, saving: false) } + jobSortSuspended = max(0, jobSortSuspended - 1) + sortJobs() playlist.touch() // creation + initial tracks count as a content change try? context.save() return playlist diff --git a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Jobs.swift b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Jobs.swift index d527de9..24605b0 100644 --- a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Jobs.swift +++ b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Jobs.swift @@ -6,6 +6,19 @@ extension PreparationQueue { static let songPriority = 100 static let playlistPriority = 50 + /// Drop Downloads rows, priority, and ingest waiters for tracks the UI or sync just deleted. + /// Call *before* destroying the `@Model`s so a mid-backoff retry doesn't leave a ghost. + public func handleTracksDeleted(_ ids: Set) { + guard !ids.isEmpty else { return } + for id in ids { + ingestAttempts[id] = nil + ingestPriority[id] = nil + retryScheduledTrackIDs.remove(id) + removeJob(id) + } + Task { await ingestLimiter.cancel(ids: ids) } + } + /// Jump `track` to the head of the ingest queue. Failed rows are re-enqueued. /// Already-ready tracks are a no-op — there's nothing to download. public func prioritize(_ track: Track, in context: ModelContext) { @@ -61,7 +74,7 @@ extension PreparationQueue { isPrioritized: prioritized )) } - sortJobs() + if jobSortSuspended == 0 { sortJobs() } } func updateJobProgress(_ id: UUID, bytes: Int, total: Int?) { @@ -79,13 +92,14 @@ extension PreparationQueue { } /// Active downloads first, then analysis, then the waiting queue. Prioritized rows float up - /// within a phase so "Download first" is visible at the top of the screen. - private func sortJobs() { + /// within a phase so "Download first" is visible at the top of each section — never above an + /// in-flight download we can't preempt. + func sortJobs() { ingestJobs.sort { a, b in - if a.isPrioritized != b.isPrioritized { return a.isPrioritized && !b.isPrioritized } let pa = Self.phaseOrder(a.phase) let pb = Self.phaseOrder(b.phase) if pa != pb { return pa < pb } + if a.isPrioritized != b.isPrioritized { return a.isPrioritized && !b.isPrioritized } return a.title.localizedCaseInsensitiveCompare(b.title) == .orderedAscending } } diff --git a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Sync.swift b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Sync.swift index f716731..0ccdd8f 100644 --- a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Sync.swift +++ b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue+Sync.swift @@ -203,7 +203,9 @@ extension PreparationQueue { /// dead model), then the models, then share-aware file cleanup. private func removeTracks(_ tracks: [Track], in context: ModelContext) { guard !tracks.isEmpty else { return } - onTracksDeleted?(Set(tracks.map(\.id))) + let ids = Set(tracks.map(\.id)) + handleTracksDeleted(ids) + onTracksDeleted?(ids) let keys = tracks.map(\.stemKey) for track in tracks { context.delete(track) } LibraryCleanup.removeOrphanedFiles(keys: keys, in: context) diff --git a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue.swift b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue.swift index 4f865a1..60c1893 100644 --- a/Packages/ContinuityKit/Sources/Ingest/PreparationQueue.swift +++ b/Packages/ContinuityKit/Sources/Ingest/PreparationQueue.swift @@ -52,6 +52,8 @@ public final class PreparationQueue { public internal(set) var ingestJobs: [IngestJob] = [] /// User-raised ingest priority per track. 100 = one song, 50 = whole playlist/album. var ingestPriority: [UUID: Int] = [:] + /// When > 0, `upsertJob` skips per-call sorts — resume/import batches sort once at the end. + var jobSortSuspended = 0 /// Production wiring — the app constructs the queue with no arguments. The parameterized /// initializer stays internal for dependency-injected tests within the module. @@ -125,7 +127,9 @@ public final class PreparationQueue { /// no stems yet. `.failed` tracks are left as-is for an explicit retry. /// /// Yields every 40 rows so a thousand-track library doesn't occupy the main actor for the - /// whole pass before the first frame can land. + /// whole pass before the first frame can land. Job-list sorts are suspended for the whole + /// walk — otherwise each `enqueue` → `upsertJob` → `sortJobs` is O(n²) during a large + /// unfinished import and trips the scene-create watchdog (`0x8BADF00D`). public func resumePreparation(in context: ModelContext) async { guard let tracks = try? context.fetch(FetchDescriptor()) else { return } // One directory listing per cache instead of up to five `fileExists` probes per track: @@ -134,6 +138,11 @@ public final class PreparationQueue { let cacheIndex = CacheIndex.snapshot() // Demo healing used to `save()` once per track; batched into a single save at the end. var needsSave = false + jobSortSuspended += 1 + defer { + jobSortSuspended = max(0, jobSortSuspended - 1) + sortJobs() + } for (i, track) in tracks.enumerated() { // Demo tracks have no source and play synthesized audio — there is nothing to ingest // or resume. Without this guard they'd be re-enqueued (they have no audio file), fail @@ -212,11 +221,19 @@ public final class PreparationQueue { } // Stay `.pending` until a limiter slot is actually ours — otherwise a 200-track import - // looks like 200 simultaneous downloads. `bump` can reorder this waiter meanwhile. - await ingestLimiter.acquire(id: trackID, priority: ingestPriority[trackID] ?? 0) - // Deleted while queued. + // looks like 200 simultaneous downloads. `bump` can reorder this waiter meanwhile; + // `cancel` (delete) resumes with false and no slot — do not `release()` in that case. + let acquired = await ingestLimiter.acquire(id: trackID, priority: ingestPriority[trackID] ?? 0) + guard acquired else { + ingestAttempts[trackID] = nil + ingestPriority[trackID] = nil + removeJob(trackID) + return + } + // Deleted while queued (or cancelled after a race with `handleTracksDeleted`). guard track.modelContext != nil else { ingestAttempts[trackID] = nil + ingestPriority[trackID] = nil removeJob(trackID) await ingestLimiter.release() return @@ -395,8 +412,14 @@ public final class PreparationQueue { // 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 + guard let fresh = try? context.fetch(descriptor).first else { + // Deleted while sleeping — drop the Downloads ghost and any leftover priority. + self.ingestAttempts[trackID] = nil + self.ingestPriority[trackID] = nil + self.removeJob(trackID) + return + } + guard fresh.prepState != .ready else { return } // healed by a manual retry meanwhile Logger.ingest.notice( "retrying \(label, privacy: .public) (attempt \(attempt + 1)/\(maxAttempts))" ) diff --git a/Packages/ContinuityKit/Tests/IngestTests/ConcurrencyLimiterTests.swift b/Packages/ContinuityKit/Tests/IngestTests/ConcurrencyLimiterTests.swift index 6b6d5ad..a2e30c4 100644 --- a/Packages/ContinuityKit/Tests/IngestTests/ConcurrencyLimiterTests.swift +++ b/Packages/ContinuityKit/Tests/IngestTests/ConcurrencyLimiterTests.swift @@ -13,14 +13,14 @@ final class ConcurrencyLimiterTests: XCTestCase { await limiter.acquire(id: UUID(), priority: 0) let first = Task { - await limiter.acquire(id: firstID, priority: 0) + XCTAssertTrue(await limiter.acquire(id: firstID, priority: 0)) order.append(firstID) await limiter.release() } await waitUntil(limiter, pending: 1) let second = Task { - await limiter.acquire(id: secondID, priority: 0) + XCTAssertTrue(await limiter.acquire(id: secondID, priority: 0)) order.append(secondID) await limiter.release() } @@ -43,14 +43,14 @@ final class ConcurrencyLimiterTests: XCTestCase { await limiter.acquire() let first = Task { - await limiter.acquire(id: firstID, priority: 0) + XCTAssertTrue(await limiter.acquire(id: firstID, priority: 0)) order.append(firstID) await limiter.release() } await waitUntil(limiter, pending: 1) let second = Task { - await limiter.acquire(id: secondID, priority: 0) + XCTAssertTrue(await limiter.acquire(id: secondID, priority: 0)) order.append(secondID) await limiter.release() } @@ -62,6 +62,32 @@ final class ConcurrencyLimiterTests: XCTestCase { XCTAssertEqual(order, [firstID, secondID]) } + /// Cancelled waiters wake with `false` and do not consume a slot. + func testCancelDropsWaiterWithoutGrantingSlot() async { + let limiter = ConcurrencyLimiter(limit: 1) + let cancelledID = UUID() + let survivorID = UUID() + + await limiter.acquire(id: UUID(), priority: 0) + + let cancelled = Task { + await limiter.acquire(id: cancelledID, priority: 0) + } + await waitUntil(limiter, pending: 1) + + let survivor = Task { + await limiter.acquire(id: survivorID, priority: 0) + } + await waitUntil(limiter, pending: 2) + + await limiter.cancel(ids: [cancelledID]) + XCTAssertFalse(await cancelled.value) + + await limiter.release() + XCTAssertTrue(await survivor.value) + await limiter.release() + } + private func waitUntil(_ limiter: ConcurrencyLimiter, pending: Int) async { for _ in 0..<10_000 { if await limiter.pendingCount >= pending { return }