CalvinBackup/SpotiFLAC-Mobile
1
0
Fork 0
mirror of https://github.com/zarzet/SpotiFLAC-Mobile.git synced 2026-09-16 06:45:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(ios): isolate download progress subscriptions

Browse Source 
Give every subscription its own cursor, payload, and worker queue so an old blocked waiter cannot advance a replacement listener. Serialize lifecycle handling and guard cancellation.

Add a restart regression with an injected waiter; verified with Thread Sanitizer.
This commit is contained in:
zarzet
2026-09-06 14:03:49 +07:00
parent 49c3c66f15
commit 857d146c55
4 changed files with 133 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ios/Runner.xcodeproj/project.pbxproj
+4
View File
@@ -7,6 +7,7 @@
objects = {
/* Begin PBXBuildFile section */
D08A594730CF000100000001 /* DownloadProgressSubscription.swift in Sources */ = {isa = PBXBuildFile; fileRef = D08A594730CF000100000002 /* DownloadProgressSubscription.swift */; };
1498D2341E8E89220040F4C2 /* GeneratedPluginRegistrant.m in Sources */ = {isa = PBXBuildFile; fileRef = 1498D2331E8E89220040F4C2 /* GeneratedPluginRegistrant.m */; };
331C808B294A63AB00263BE5 /* RunnerTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 331C807B294A618700263BE5 /* RunnerTests.swift */; };
3B3967161E833CAA004F5970 /* AppFrameworkInfo.plist in Resources */ = {isa = PBXBuildFile; fileRef = 3B3967151E833CAA004F5970 /* AppFrameworkInfo.plist */; };
@@ -47,6 +48,7 @@
/* End PBXCopyFilesBuildPhase section */
/* Begin PBXFileReference section */
D08A594730CF000100000002 /* DownloadProgressSubscription.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = DownloadProgressSubscription.swift; sourceTree = "<group>"; };
0448E6A8F05C4FAF8461F7C9 /* libPods-RunnerTests.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-RunnerTests.a"; sourceTree = BUILT_PRODUCTS_DIR; };
1498D2321E8E86230040F4C2 /* GeneratedPluginRegistrant.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = GeneratedPluginRegistrant.h; sourceTree = "<group>"; };
1498D2331E8E89220040F4C2 /* GeneratedPluginRegistrant.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; path = GeneratedPluginRegistrant.m; sourceTree = "<group>"; };
@@ -162,6 +164,7 @@
1498D2321E8E86230040F4C2 /* GeneratedPluginRegistrant.h */,
1498D2331E8E89220040F4C2 /* GeneratedPluginRegistrant.m */,
74858FAE1ED2DC5600515810 /* AppDelegate.swift */,
D08A594730CF000100000002 /* DownloadProgressSubscription.swift */,
A11CE0022F00000000000001 /* ExtensionCallbackParser.swift */,
74858FAD1ED2DC5600515810 /* Runner-Bridging-Header.h */,
);
@@ -397,6 +400,7 @@
buildActionMask = 2147483647;
files = (
74858FAF1ED2DC5600515810 /* AppDelegate.swift in Sources */,
D08A594730CF000100000001 /* DownloadProgressSubscription.swift in Sources */,
A11CE0012F00000000000001 /* ExtensionCallbackParser.swift in Sources */,
1498D2341E8E89220040F4C2 /* GeneratedPluginRegistrant.m in Sources */,
);
ios/Runner/AppDelegate.swift
+5 -55
View File
@@ -12,15 +12,9 @@ import Gobackend
private let LARGE_JSON_RESULT_FILE_KEY = "__json_file"
private let LARGE_JSON_RESULT_FILE_THRESHOLD_BYTES = 256 * 1024
private let streamQueue = DispatchQueue(label: "com.zarz.spotiflac.progress_stream", qos: .utility)
private let downloadProgressQueue = DispatchQueue(
label: "com.zarz.spotiflac.download_progress_stream",
qos: .utility
)
private var downloadProgressTimer: DispatchSourceTimer?
private var downloadProgressEventSink: FlutterEventSink?
private var lastDownloadProgressPayload: String?
private var lastDownloadProgressSeq: Int64 = 0
private var downloadProgressGeneration: UInt64 = 0
private let downloadProgressSubscription = DownloadProgressSubscription { sequence, timeout in
GobackendWaitForAllDownloadProgressDelta(sequence, timeout) as String? ?? ""
}
private var libraryScanProgressTimer: DispatchSourceTimer?
private var libraryScanProgressEventSink: FlutterEventSink?
private var lastLibraryScanProgressPayload: String?
@@ -236,43 +230,11 @@ import Gobackend
}
private func startDownloadProgressStream(_ eventSink: @escaping FlutterEventSink) {
stopDownloadProgressStream()
downloadProgressGeneration &+= 1
let generation = downloadProgressGeneration
downloadProgressEventSink = eventSink
lastDownloadProgressPayload = nil
lastDownloadProgressSeq = 0
let timer = DispatchSource.makeTimerSource(queue: downloadProgressQueue)
timer.schedule(deadline: .now(), repeating: .milliseconds(250))
timer.setEventHandler { [weak self] in
guard let self, self.downloadProgressGeneration == generation else { return }
let payload = GobackendWaitForAllDownloadProgressDelta(
self.lastDownloadProgressSeq,
15_000
) as String? ?? ""
if payload.isEmpty || payload == self.lastDownloadProgressPayload {
return
}
self.updateDownloadProgressSeq(payload)
self.lastDownloadProgressPayload = payload
DispatchQueue.main.async { [weak self] in
guard let self, self.downloadProgressGeneration == generation else { return }
eventSink(self.parseJsonPayload(payload))
}
}
downloadProgressTimer = timer
timer.resume()
downloadProgressSubscription.start(eventSink)
}
private func stopDownloadProgressStream() {
downloadProgressGeneration &+= 1
downloadProgressTimer?.setEventHandler {}
downloadProgressTimer?.cancel()
downloadProgressTimer = nil
downloadProgressEventSink = nil
lastDownloadProgressPayload = nil
lastDownloadProgressSeq = 0
downloadProgressSubscription.stop()
}
private func startLibraryScanProgressStream(_ eventSink: @escaping FlutterEventSink) {
@@ -320,18 +282,6 @@ import Gobackend
}
}
private func updateDownloadProgressSeq(_ payload: String) {
guard let data = payload.data(using: .utf8) else { return }
do {
if let obj = try JSONSerialization.jsonObject(with: data, options: [.fragmentsAllowed]) as? [String: Any],
let seq = obj["seq"] as? NSNumber,
seq.int64Value > lastDownloadProgressSeq {
lastDownloadProgressSeq = seq.int64Value
}
} catch {
}
}
private func bridgeJsonResult(_ payload: String) -> Any {
if payload.utf8.count < LARGE_JSON_RESULT_FILE_THRESHOLD_BYTES {
return payload
ios/Runner/DownloadProgressSubscription.swift
+82
View File
@@ -0,0 +1,82 @@
import Foundation
/// Each listener owns its cursor and pending waiter. Lifecycle and delivery run
/// on the main queue; a cancelled wait may finish, but can only touch its own state.
final class DownloadProgressSubscription {
typealias Waiter = (Int64, Int64) -> String
private final class Session {
let queue = DispatchQueue(label: "com.zarz.spotiflac.download_progress_subscription", qos: .utility)
private let lock = NSLock()
private var cancelled = false
// Accessed only on this session's queue.
var sequence: Int64 = 0
var lastPayload: String?
func cancel() {
lock.lock()
cancelled = true
lock.unlock()
}
var isCancelled: Bool {
lock.lock()
defer { lock.unlock() }
return cancelled
}
}
private let waiter: Waiter
private let interval: TimeInterval
private var current: Session?
init(interval: TimeInterval = 0.25, waiter: @escaping Waiter) {
self.interval = interval
self.waiter = waiter
}
func start(_ receive: @escaping (Any) -> Void) {
dispatchPrecondition(condition: .onQueue(.main))
stop()
let session = Session()
current = session
poll(session, receive: receive)
}
func stop() {
dispatchPrecondition(condition: .onQueue(.main))
current?.cancel()
current = nil
}
deinit { current?.cancel() }
private func poll(_ session: Session, receive: @escaping (Any) -> Void) {
let waiter = self.waiter
session.queue.async { [weak self] in
guard !session.isCancelled else { return }
let payload = waiter(session.sequence, 15_000)
guard !session.isCancelled else { return }
if !payload.isEmpty && payload != session.lastPayload,
let data = payload.data(using: .utf8),
let object = try? JSONSerialization.jsonObject(with: data, options: [.fragmentsAllowed]),
let delta = object as? [String: Any],
let sequence = delta["seq"] as? NSNumber {
session.sequence = max(session.sequence, sequence.int64Value)
session.lastPayload = payload
DispatchQueue.main.async { [weak self] in
guard let self, self.current === session, !session.isCancelled else { return }
receive(object)
}
}
// Schedule from main so subscription identity is never accessed
// concurrently, and stop/relisten does not wait for an old Go call.
DispatchQueue.main.async { [weak self] in
guard let self, self.current === session, !session.isCancelled else { return }
DispatchQueue.main.asyncAfter(deadline: .now() + self.interval) { [weak self] in
guard let self, self.current === session, !session.isCancelled else { return }
self.poll(session, receive: receive)
}
}
}
}
}
ios/RunnerTests/RunnerTests.swift
+42
View File
@@ -3,6 +3,48 @@ import XCTest
@testable import Runner
class RunnerTests: XCTestCase {
func testProgressRestartDoesNotAcceptCancelledWaiterState() {
let oldStarted = expectation(description: "old waiter started")
let replacementDelivered = expectation(description: "replacement snapshot")
let replacementAdvanced = expectation(description: "replacement owns cursor")
let oldReleased = expectation(description: "old waiter returned")
let releaseOld = DispatchSemaphore(value: 0)
let lock = NSLock()
var calls = 0
var invalidCursors = [Int64]()
let stream = DownloadProgressSubscription(interval: 0.01) { sequence, _ in
lock.lock()
calls += 1
let call = calls
if (call <= 2 && sequence != 0) || (call > 2 && sequence != 2) {
invalidCursors.append(sequence)
}
lock.unlock()
if call == 1 {
oldStarted.fulfill()
_ = releaseOld.wait(timeout: .now() + 3)
oldReleased.fulfill()
return "{\"seq\":99,\"items\":{}}"
}
if call == 2 { return "{\"seq\":2,\"reset\":true,\"items\":{}}" }
if call == 3 { replacementAdvanced.fulfill() }
return ""
}
stream.start { _ in XCTFail("Cancelled listener received an event") }
wait(for: [oldStarted], timeout: 2)
stream.stop()
stream.start { event in
XCTAssertEqual((event as? [String: Any])?["seq"] as? Int, 2)
replacementDelivered.fulfill()
releaseOld.signal()
}
wait(for: [replacementDelivered, oldReleased, replacementAdvanced], timeout: 2)
stream.stop()
lock.lock()
XCTAssertTrue(invalidCursors.isEmpty, "Unexpected cursors: \(invalidCursors)")
lock.unlock()
}
func testParsesOAuthCallback() {
let route = ExtensionCallbackParser.parse(
URL(string: "spotiflac://callback?code=auth-code&state=spotify-web")!