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

perf(download): reduce progress wakeups and harden retries

Browse Source
This commit is contained in:
zarzet
2026-08-29 23:47:58 +07:00
parent fd74240c24
commit f1fb01604d
12 changed files with 234 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
android/app/src/main/kotlin/com/zarz/spotiflac/DownloadService.kt
+1
View File
@@ -374,6 +374,7 @@ class DownloadService : Service() {
@Volatile private var nativeWorkerCancelRequested = false
internal var nativeWorkerDownloadNetworkMode = "any"
internal var nativeWorkerNetworkCallback: ConnectivityManager.NetworkCallback? = null
internal var nativeWorkerNetworkFallbackJob: Job? = null
internal val nativeWorkerWifiNetworks = mutableSetOf<Network>()
// Bumped every time a new native queue replaces the current one. A worker
// coroutine that observes a different generation than its own must stop
android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceNetwork.kt
+13
View File
@@ -94,6 +94,17 @@ internal fun DownloadService.configureNativeWorkerNetworkPolicy(settingsJson: St
"DownloadService",
"Failed to monitor Wi-Fi for native worker: ${e.message}",
)
// Callback registration can fail when the process/system callback quota is
// exhausted. Keep the Wi-Fi-only queue observable instead of leaving it
// paused forever with no event capable of resuming it.
nativeWorkerNetworkFallbackJob?.cancel()
nativeWorkerNetworkFallbackJob = serviceScope.launch {
while (NativeWorkerPolicy.requiresWifi(nativeWorkerDownloadNetworkMode)) {
delay(5_000)
refreshNativeWorkerNetworkPause()
if (nativeWorkerJob?.isActive != true) break
}
}
}
}
@@ -159,6 +170,8 @@ internal fun DownloadService.refreshNativeWorkerNetworkPause() {
}
internal fun DownloadService.unregisterNativeWorkerNetworkCallback() {
nativeWorkerNetworkFallbackJob?.cancel()
nativeWorkerNetworkFallbackJob = null
val callback = nativeWorkerNetworkCallback
nativeWorkerNetworkCallback = null
synchronized(nativeWorkerWifiNetworks) {
android/app/src/main/kotlin/com/zarz/spotiflac/DownloadServiceSnapshot.kt
+1 -1
View File
@@ -329,7 +329,7 @@ internal fun DownloadService.cancelNativeWorkerProgressCoordinator() {
private fun DownloadService.pollNativeWorkerProgress(generation: Long): Set<String> {
val sinceSeq = synchronized(nativeWorkerProgressLock) { nativeWorkerProgressSeq }
val raw = try {
Gobackend.getAllDownloadProgressDelta(sinceSeq)
Gobackend.waitForAllDownloadProgressDelta(sinceSeq, 5_000L)
} catch (_: Exception) {
return emptySet()
}
android/app/src/main/kotlin/com/zarz/spotiflac/MainActivity.kt
+9 -3
View File
@@ -66,7 +66,6 @@ class MainActivity: FlutterFragmentActivity() {
"com.zarz.spotiflac/download_progress_stream"
private val LIBRARY_SCAN_PROGRESS_STREAM_CHANNEL =
"com.zarz.spotiflac/library_scan_progress_stream"
private val DOWNLOAD_PROGRESS_STREAM_POLLING_INTERVAL_MS = 1200L
// A progress bar can't show sub-second granularity; 400ms halves the
// disk-read wakeups during a scan vs the previous 200ms.
private val LIBRARY_SCAN_PROGRESS_STREAM_POLLING_INTERVAL_MS = 400L
@@ -482,12 +481,17 @@ class MainActivity: FlutterFragmentActivity() {
while (isActive && downloadProgressEventSink === sink) {
try {
val payload = withContext(Dispatchers.IO) {
Gobackend.getAllDownloadProgressDelta(lastDownloadProgressSeq)
Gobackend.waitForAllDownloadProgressDelta(
lastDownloadProgressSeq,
15_000L,
)
}
if (!isActive || downloadProgressEventSink !== sink) break
if (payload.isNotEmpty() && payload != lastDownloadProgressPayload) {
updateDownloadProgressSeq(payload)
lastDownloadProgressPayload = payload
sink.success(parseJsonPayload(payload))
delay(250L)
}
} catch (e: Exception) {
android.util.Log.w(
@@ -495,7 +499,7 @@ class MainActivity: FlutterFragmentActivity() {
"Download progress stream poll failed: ${e.message}",
)
}
delay(DOWNLOAD_PROGRESS_STREAM_POLLING_INTERVAL_MS)
if (downloadProgressEventSink !== sink) break
}
}
}
@@ -517,6 +521,7 @@ class MainActivity: FlutterFragmentActivity() {
val initialPayload = withContext(Dispatchers.IO) {
readLibraryScanProgressJsonForStream()
}
if (!isActive || libraryScanProgressEventSink !== sink) return@launch
lastLibraryScanProgressPayload = initialPayload
sink.success(parseJsonPayload(initialPayload))
} catch (e: Exception) {
@@ -530,6 +535,7 @@ class MainActivity: FlutterFragmentActivity() {
val payload = withContext(Dispatchers.IO) {
readLibraryScanProgressJsonForStream()
}
if (!isActive || libraryScanProgressEventSink !== sink) break
if (payload != lastLibraryScanProgressPayload) {
lastLibraryScanProgressPayload = payload
sink.success(parseJsonPayload(payload))
go_backend/exports_download.go
+4
View File
@@ -433,6 +433,10 @@ func GetAllDownloadProgressDelta(sinceSeq int64) string {
return GetMultiProgressDelta(sinceSeq)
}
func WaitForAllDownloadProgressDelta(sinceSeq, timeoutMs int64) string {
return WaitForMultiProgressDelta(sinceSeq, timeoutMs)
}
func InitItemProgress(itemID string) {
StartItemProgress(itemID)
}
go_backend/progress.go
+34
View File
@@ -67,6 +67,7 @@ var (
multiProgressSeq int64
multiProgressReset int64
removedProgressSeq = make(map[string]int64)
multiProgressNotify = make(chan struct{})
)
func markMultiProgressDirtyLocked() {
@@ -75,6 +76,8 @@ func markMultiProgressDirtyLocked() {
func nextMultiProgressSeqLocked() int64 {
multiProgressSeq++
close(multiProgressNotify)
multiProgressNotify = make(chan struct{})
return multiProgressSeq
}
@@ -180,6 +183,37 @@ func GetMultiProgressDelta(sinceSeq int64) string {
return string(jsonBytes)
}
// WaitForMultiProgressDelta blocks without polling until the bridge revision
// advances or the bounded heartbeat expires. Each waiter observes the same
// close-only notification channel, so UI and native-worker consumers do not
// compete for events.
func WaitForMultiProgressDelta(sinceSeq, timeoutMs int64) string {
if timeoutMs <= 0 {
timeoutMs = 15_000
}
if timeoutMs > 60_000 {
timeoutMs = 60_000
}
timer := time.NewTimer(time.Duration(timeoutMs) * time.Millisecond)
defer timer.Stop()
for {
multiMu.RLock()
if sinceSeq < multiProgressSeq {
multiMu.RUnlock()
return GetMultiProgressDelta(sinceSeq)
}
notify := multiProgressNotify
multiMu.RUnlock()
select {
case <-notify:
continue
case <-timer.C:
return ""
}
}
}
func StartItemProgress(itemID string) {
multiMu.Lock()
defer multiMu.Unlock()
go_backend/progress_test.go
+45
View File
@@ -6,6 +6,51 @@ import (
"time"
)
func TestWaitForMultiProgressDeltaWakesOnRevision(t *testing.T) {
ClearAllItemProgress()
defer ClearAllItemProgress()
multiMu.RLock()
since := multiProgressSeq
multiMu.RUnlock()
result := make(chan string, 1)
go func() {
result <- WaitForMultiProgressDelta(since, 1_000)
}()
time.Sleep(10 * time.Millisecond)
StartItemProgress("wait-progress")
select {
case payload := <-result:
if payload == "" {
t.Fatal("waiter woke without a progress delta")
}
var delta MultiProgressDelta
if err := json.Unmarshal([]byte(payload), &delta); err != nil {
t.Fatalf("decode delta: %v", err)
}
if delta.Items["wait-progress"] == nil {
t.Fatalf("delta missing item: %#v", delta)
}
case <-time.After(time.Second):
t.Fatal("progress waiter did not wake")
}
}
func TestWaitForMultiProgressDeltaHeartbeatTimeout(t *testing.T) {
ClearAllItemProgress()
defer ClearAllItemProgress()
multiMu.RLock()
since := multiProgressSeq
multiMu.RUnlock()
startedAt := time.Now()
if payload := WaitForMultiProgressDelta(since, 20); payload != "" {
t.Fatalf("timeout payload = %q, want empty", payload)
}
if elapsed := time.Since(startedAt); elapsed < 15*time.Millisecond {
t.Fatalf("wait returned too early after %v", elapsed)
}
}
func TestItemTransferProgressReporterCoalescesHotPathUpdates(t *testing.T) {
ClearAllItemProgress()
defer ClearAllItemProgress()
ios/Runner/AppDelegate.swift
+24 -7
View File
@@ -12,13 +12,19 @@ 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 var libraryScanProgressTimer: DispatchSourceTimer?
private var libraryScanProgressEventSink: FlutterEventSink?
private var lastLibraryScanProgressPayload: String?
private var libraryScanProgressGeneration: UInt64 = 0
private var backendChannel: FlutterMethodChannel?
private var pendingSessionGrantEvents: [[String: Any]] = []
@@ -222,22 +228,28 @@ 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: streamQueue)
timer.schedule(deadline: .now(), repeating: .milliseconds(800))
let timer = DispatchSource.makeTimerSource(queue: downloadProgressQueue)
timer.schedule(deadline: .now(), repeating: .milliseconds(250))
timer.setEventHandler { [weak self] in
guard let self else { return }
let payload = GobackendGetAllDownloadProgressDelta(self.lastDownloadProgressSeq) as String? ?? ""
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
self?.downloadProgressEventSink?(self?.parseJsonPayload(payload))
guard let self, self.downloadProgressGeneration == generation else { return }
eventSink(self.parseJsonPayload(payload))
}
}
downloadProgressTimer = timer
@@ -245,6 +257,7 @@ import Gobackend
}
private func stopDownloadProgressStream() {
downloadProgressGeneration &+= 1
downloadProgressTimer?.setEventHandler {}
downloadProgressTimer?.cancel()
downloadProgressTimer = nil
@@ -255,20 +268,23 @@ import Gobackend
private func startLibraryScanProgressStream(_ eventSink: @escaping FlutterEventSink) {
stopLibraryScanProgressStream()
libraryScanProgressGeneration &+= 1
let generation = libraryScanProgressGeneration
libraryScanProgressEventSink = eventSink
lastLibraryScanProgressPayload = nil
let timer = DispatchSource.makeTimerSource(queue: streamQueue)
timer.schedule(deadline: .now(), repeating: .milliseconds(800))
timer.setEventHandler { [weak self] in
guard let self else { return }
guard let self, self.libraryScanProgressGeneration == generation else { return }
let payload = GobackendGetLibraryScanProgressJSON() as String? ?? "{}"
if payload == self.lastLibraryScanProgressPayload {
return
}
self.lastLibraryScanProgressPayload = payload
DispatchQueue.main.async { [weak self] in
self?.libraryScanProgressEventSink?(self?.parseJsonPayload(payload))
guard let self, self.libraryScanProgressGeneration == generation else { return }
eventSink(self.parseJsonPayload(payload))
}
}
libraryScanProgressTimer = timer
@@ -276,6 +292,7 @@ import Gobackend
}
private func stopLibraryScanProgressStream() {
libraryScanProgressGeneration &+= 1
libraryScanProgressTimer?.setEventHandler {}
libraryScanProgressTimer?.cancel()
libraryScanProgressTimer = nil
lib/models/download_item.dart
+14 -6
View File
@@ -3,6 +3,8 @@ import 'package:spotiflac_android/models/track.dart';
part 'download_item.g.dart';
const Object _downloadItemUnset = Object();
enum DownloadStatus {
queued,
downloading,
@@ -72,9 +74,9 @@ class DownloadItem {
double? speedMBps,
int? bytesReceived,
int? bytesTotal,
String? filePath,
String? error,
DownloadErrorType? errorType,
Object? filePath = _downloadItemUnset,
Object? error = _downloadItemUnset,
Object? errorType = _downloadItemUnset,
String? preparationStage,
DateTime? createdAt,
String? qualityOverride,
@@ -92,9 +94,15 @@ class DownloadItem {
speedMBps: speedMBps ?? this.speedMBps,
bytesReceived: bytesReceived ?? this.bytesReceived,
bytesTotal: bytesTotal ?? this.bytesTotal,
filePath: filePath ?? this.filePath,
error: error ?? this.error,
errorType: errorType ?? this.errorType,
filePath: identical(filePath, _downloadItemUnset)
? this.filePath
: filePath as String?,
error: identical(error, _downloadItemUnset)
? this.error
: error as String?,
errorType: identical(errorType, _downloadItemUnset)
? this.errorType
: errorType as DownloadErrorType?,
preparationStage: preparationStage ?? this.preparationStage,
createdAt: createdAt ?? this.createdAt,
qualityOverride: qualityOverride ?? this.qualityOverride,
lib/providers/download_queue_provider.dart
+38 -17
View File
@@ -1311,7 +1311,7 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
}
}
void retryItem(String id) {
Future<void> retryItem(String id) async {
final item = state.items.where((i) => i.id == id).firstOrNull;
if (item == null) {
_log.w('retryItem: Item not found: $id');
@@ -1328,11 +1328,18 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
// A cancel issued while the item never started leaves a pre-registered
// flag in the Go backend that the next attempt would consume and abort
// instantly; the user asked for a retry, so drop it first.
unawaited(
PlatformBridge.resetDownloadCancel(id).catchError((Object e) {
_log.w('Failed to reset cancel flag for $id: $e');
}),
);
try {
await PlatformBridge.resetDownloadCancel(id);
} catch (e) {
_log.w('Failed to reset cancel flag for $id: $e');
return;
}
final current = state.items.where((i) => i.id == id).firstOrNull;
if (current == null ||
(current.status != DownloadStatus.failed &&
current.status != DownloadStatus.skipped)) {
return;
}
_locallyCancelledItemIds.remove(id);
_verificationRetryGuard.clearItem(id);
_rateLimitRetriedItemIds.remove(id);
@@ -1347,6 +1354,8 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
status: DownloadStatus.queued,
progress: 0,
error: null,
errorType: null,
filePath: null,
);
}
return i;
@@ -1362,7 +1371,7 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
}
}
void retryAllFailed({bool networkOnly = false}) {
Future<void> retryAllFailed({bool networkOnly = false}) async {
final failedIds = state.items
.where(
(item) =>
@@ -1378,24 +1387,34 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
}
_log.i('Retrying ${failedIds.length} failed download(s)');
for (final id in failedIds) {
unawaited(
PlatformBridge.resetDownloadCancel(id).catchError((Object e) {
final resetResults = await Future.wait(
failedIds.map((id) async {
try {
await PlatformBridge.resetDownloadCancel(id);
return id;
} catch (e) {
_log.w('Failed to reset cancel flag for $id: $e');
}),
);
}
_locallyCancelledItemIds.removeAll(failedIds);
_pausePendingItemIds.removeAll(failedIds);
return null;
}
}),
);
final retryableIds = resetResults.whereType<String>().toSet();
if (retryableIds.isEmpty) return;
_locallyCancelledItemIds.removeAll(retryableIds);
_pausePendingItemIds.removeAll(retryableIds);
for (final item in state.items) {
if (!failedIds.contains(item.id)) continue;
if (!retryableIds.contains(item.id)) continue;
_purgeAlbumRgEntry(item.track);
}
final items = state.items
.map((item) {
if (!failedIds.contains(item.id)) return item;
if (!retryableIds.contains(item.id)) return item;
if (item.status != DownloadStatus.failed &&
item.status != DownloadStatus.skipped) {
return item;
}
return item.copyWith(
status: DownloadStatus.queued,
progress: 0,
@@ -1403,6 +1422,8 @@ class DownloadQueueNotifier extends Notifier<DownloadQueueState> {
bytesReceived: 0,
bytesTotal: 0,
error: null,
errorType: null,
filePath: null,
);
})
.toList(growable: false);
lib/utils/progress_stream_poller.dart
+38 -14
View File
@@ -56,8 +56,9 @@ class ProgressStreamPoller<T> {
StreamSubscription<T>? _sub;
bool _hasReceivedStreamEvent = false;
bool _usingStream = false;
bool _inFlight = false;
int? _inFlightGeneration;
int _errorCount = 0;
int _generation = 0;
bool get usingStream => _usingStream;
@@ -65,6 +66,8 @@ class ProgressStreamPoller<T> {
/// stream (falling back to polling on timeout/error); otherwise starts
/// polling immediately.
void start({required bool useStream}) {
_generation++;
final generation = _generation;
_timer?.cancel();
_bootstrapTimer?.cancel();
_bootstrapTimer = null;
@@ -74,22 +77,28 @@ class ProgressStreamPoller<T> {
_usingStream = false;
if (useStream) {
_attachStream();
_attachStream(generation);
return;
}
startPollingTimer();
startPollingTimer(generation: generation);
}
void _attachStream() {
void _attachStream(int generation) {
_sub = streamProvider().listen(
(progress) async {
if (generation != _generation) return;
_hasReceivedStreamEvent = true;
_usingStream = true;
_bootstrapTimer?.cancel();
_bootstrapTimer = null;
await _runGuarded(() async => progress, onStreamProcessingError);
await _runGuarded(
() async => progress,
onStreamProcessingError,
generation: generation,
);
},
onError: (Object error, StackTrace stackTrace) {
if (generation != _generation) return;
if (_usingStream) {
onStreamFailed(error);
}
@@ -98,59 +107,74 @@ class ProgressStreamPoller<T> {
_usingStream = false;
_bootstrapTimer?.cancel();
_bootstrapTimer = null;
startPollingTimer();
startPollingTimer(generation: generation);
},
cancelOnError: false,
);
_bootstrapTimer = Timer(bootstrapTimeout, () {
if (generation != _generation) return;
if (_hasReceivedStreamEvent) return;
onStreamTimeout();
_sub?.cancel();
_sub = null;
_usingStream = false;
startPollingTimer();
startPollingTimer(generation: generation);
});
}
/// (Re)starts the fallback polling timer.
void startPollingTimer() {
void startPollingTimer({int? generation}) {
final activeGeneration = generation ?? _generation;
if (activeGeneration != _generation) return;
_timer?.cancel();
_timer = Timer.periodic(pollingInterval, (_) async {
await _runGuarded(pollProvider, onPollError, gate: shouldPollTick);
await _runGuarded(
pollProvider,
onPollError,
gate: shouldPollTick,
generation: activeGeneration,
);
});
}
/// One-shot immediate fetch reusing the same in-flight guard and error
/// counter as the periodic poll, with its own error callback.
Future<void> pollOnce(void Function(Object error) onError) =>
_runGuarded(pollProvider, onError);
_runGuarded(pollProvider, onError, generation: _generation);
Future<void> _runGuarded(
Future<T> Function() fetch,
void Function(Object error) onError, {
bool Function()? gate,
required int generation,
}) async {
if (_inFlight) return;
_inFlight = true;
if (generation != _generation || _inFlightGeneration == generation) return;
_inFlightGeneration = generation;
try {
if (gate != null && !gate()) return;
final progress = await fetch();
if (generation != _generation) return;
await onProgress(progress);
if (generation != _generation) return;
_errorCount = 0;
} catch (e) {
if (generation != _generation) return;
_errorCount++;
if (_errorCount <= _errorLogThreshold) {
onError(e);
}
} finally {
_inFlight = false;
if (_inFlightGeneration == generation) {
_inFlightGeneration = null;
}
}
}
/// Cancels timers/subscription and resets guard/counter state, without
/// releasing the poller for reuse (matches each caller's `_stop*` reset).
void stop() {
_generation++;
_timer?.cancel();
_bootstrapTimer?.cancel();
_sub?.cancel();
@@ -158,7 +182,7 @@ class ProgressStreamPoller<T> {
_bootstrapTimer = null;
_sub = null;
_errorCount = 0;
_inFlight = false;
_inFlightGeneration = null;
_hasReceivedStreamEvent = false;
_usingStream = false;
}
test/models_and_utils_test.dart
+13 -3
View File
@@ -281,10 +281,10 @@ void main() {
);
});
test('polls one progress delta stream and preserves concurrent items', () {
test('waits on one progress delta stream and preserves concurrent items', () {
expect(
RegExp(
r'Gobackend\.getAllDownloadProgressDelta\(',
r'Gobackend\.waitForAllDownloadProgressDelta\(',
).allMatches(workerSnapshotSource),
hasLength(1),
);
@@ -726,6 +726,7 @@ void main() {
track: sampleTrack(),
service: 'qobuz',
createdAt: DateTime.utc(2026),
filePath: '/music/stale.flac',
error: 'raw backend failure',
);
@@ -746,7 +747,16 @@ void main() {
base.copyWith(errorType: DownloadErrorType.permission).errorMessage,
'Cannot write to folder, check storage permission',
);
expect(base.copyWith(error: null).errorMessage, 'raw backend failure');
expect(base.copyWith(error: null).errorMessage, isEmpty);
expect(base.copyWith().filePath, '/music/stale.flac');
expect(base.copyWith(filePath: null).filePath, isNull);
expect(
base
.copyWith(errorType: DownloadErrorType.network)
.copyWith(errorType: null)
.errorType,
isNull,
);
});
test('decodes json defaults and enums', () {