mirror of
https://github.com/KRTirtho/spotube.git
synced 2026-08-05 19:59:51 +00:00
fix(server): race condition and cache corruption in playback routes
Two related bugs in ServerPlaybackRoutes:
1. _getSourcedTrack looked up the requested media in
audioPlayer.playlist.medias by URI, but MPV issues its HTTP request
before media_kit's Dart-side state has synced with the native
player, so the lookup threw "Bad state: No element" and the proxy
returned 500 ("Failed to open" in MPV). Now resolves the track
directly from playlist.tracks by trackId via firstWhereOrNull,
independent of the player's media list.
2. MPV opens multiple concurrent range requests for the same track
(buffering/seek/reconnect), each opening its own writeOnlyAppend
sink to the same .part cache file; the final rename would then fail
with a file-in-use error (or corrupt the cache from concurrent
appends). Added a _cachingInProgress guard so only one request
caches a given track at a time, cache only full requests
(contentRange.start == 0), write with truncate instead of append,
and clean up the in-progress marker/partial file in finally.
Fixes #3089
This commit is contained in:
parent
69a310c78f
commit
19adaf5e54
@ -2,6 +2,7 @@ import 'dart:async';
|
|||||||
import 'dart:io';
|
import 'dart:io';
|
||||||
import 'dart:math';
|
import 'dart:math';
|
||||||
|
|
||||||
|
import 'package:collection/collection.dart';
|
||||||
import 'package:dio/dio.dart' hide Response;
|
import 'package:dio/dio.dart' hide Response;
|
||||||
import 'package:dio/dio.dart' as dio_lib;
|
import 'package:dio/dio.dart' as dio_lib;
|
||||||
import 'package:flutter/foundation.dart';
|
import 'package:flutter/foundation.dart';
|
||||||
@ -42,6 +43,13 @@ class ServerPlaybackRoutes {
|
|||||||
AudioPlayerState get playlist => ref.read(audioPlayerProvider);
|
AudioPlayerState get playlist => ref.read(audioPlayerProvider);
|
||||||
final Dio dio;
|
final Dio dio;
|
||||||
|
|
||||||
|
/// Track IDs that are currently being written to the music cache.
|
||||||
|
/// MediaKit/MPV opens several concurrent range requests for the same track
|
||||||
|
/// (buffering, seeking, reconnects); without this guard every request would
|
||||||
|
/// open its own sink to the same `.part` file and the final `rename` would
|
||||||
|
/// fail with a file-in-use error on Windows.
|
||||||
|
final Set<String> _cachingInProgress = {};
|
||||||
|
|
||||||
ServerPlaybackRoutes(this.ref) : dio = Dio();
|
ServerPlaybackRoutes(this.ref) : dio = Dio();
|
||||||
|
|
||||||
Future<String> _getTrackCacheFilePath(SourcedTrack track) async {
|
Future<String> _getTrackCacheFilePath(SourcedTrack track) async {
|
||||||
@ -53,25 +61,19 @@ class ServerPlaybackRoutes {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
Future<SourcedTrack?> _getSourcedTrack(
|
Future<SourcedTrack?> _getSourcedTrack(String trackId) async {
|
||||||
Request request,
|
final track = playlist.tracks
|
||||||
String trackId,
|
.firstWhereOrNull((element) => element.id == trackId);
|
||||||
) async {
|
|
||||||
final track =
|
if (track == null) return null;
|
||||||
playlist.tracks.firstWhere((element) => element.id == trackId);
|
|
||||||
|
|
||||||
final activeSourcedTrack =
|
final activeSourcedTrack =
|
||||||
await ref.read(activeTrackSourcesProvider.future);
|
await ref.read(activeTrackSourcesProvider.future);
|
||||||
|
|
||||||
final media = audioPlayer.playlist.medias
|
|
||||||
.firstWhere((e) => e.uri == request.requestedUri.toString());
|
|
||||||
final spotubeMedia =
|
|
||||||
media is SpotubeMedia ? media : SpotubeMedia.media(media);
|
|
||||||
final sourcedTrack = activeSourcedTrack?.track.id == track.id
|
final sourcedTrack = activeSourcedTrack?.track.id == track.id
|
||||||
? activeSourcedTrack?.source
|
? activeSourcedTrack?.source
|
||||||
: await ref.read(
|
: await ref.read(
|
||||||
sourcedTrackProvider(spotubeMedia.track as SpotubeFullTrackObject)
|
sourcedTrackProvider(track as SpotubeFullTrackObject).future,
|
||||||
.future,
|
|
||||||
);
|
);
|
||||||
|
|
||||||
return sourcedTrack;
|
return sourcedTrack;
|
||||||
@ -103,11 +105,16 @@ class ServerPlaybackRoutes {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
String url = track.url ??
|
final swappedUrl0 = track.url ??
|
||||||
await ref
|
await ref
|
||||||
.read(sourcedTrackProvider(track.query).notifier)
|
.read(sourcedTrackProvider(track.query).notifier)
|
||||||
.swapWithNextSibling()
|
.swapWithNextSibling()
|
||||||
.then((track) => track.url!);
|
.then((t) => t.url);
|
||||||
|
if (swappedUrl0 == null) {
|
||||||
|
AppLogger.log.e("No playable URL for ${track.query.name}");
|
||||||
|
throw Exception("No playable URL for ${track.query.name}");
|
||||||
|
}
|
||||||
|
String url = swappedUrl0;
|
||||||
|
|
||||||
final options = Options(
|
final options = Options(
|
||||||
headers: {
|
headers: {
|
||||||
@ -156,11 +163,16 @@ class ServerPlaybackRoutes {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
String url = track.url ??
|
final swappedUrl1 = track.url ??
|
||||||
await ref
|
await ref
|
||||||
.read(sourcedTrackProvider(track.query).notifier)
|
.read(sourcedTrackProvider(track.query).notifier)
|
||||||
.swapWithNextSibling()
|
.swapWithNextSibling()
|
||||||
.then((track) => track.url!);
|
.then((t) => t.url);
|
||||||
|
if (swappedUrl1 == null) {
|
||||||
|
AppLogger.log.e("No playable URL for ${track.query.name}");
|
||||||
|
throw Exception("No playable URL for ${track.query.name}");
|
||||||
|
}
|
||||||
|
String url = swappedUrl1;
|
||||||
|
|
||||||
final options = Options(
|
final options = Options(
|
||||||
headers: {
|
headers: {
|
||||||
@ -214,9 +226,20 @@ class ServerPlaybackRoutes {
|
|||||||
"Headers: ${res.headers.map}",
|
"Headers: ${res.headers.map}",
|
||||||
);
|
);
|
||||||
|
|
||||||
if (!userPreferences.cacheMusic) {
|
final contentRange = res.headers.value("content-range") != null
|
||||||
|
? ContentRangeHeader.parse(res.headers.value("content-range") ?? "")
|
||||||
|
: ContentRangeHeader(0, 0, 0);
|
||||||
|
|
||||||
|
// Only cache full requests (range starting at 0) and only one writer per
|
||||||
|
// track at a time. Concurrent range requests just stream straight through.
|
||||||
|
final trackId = track.info.id;
|
||||||
|
final isFullRequest = contentRange.start == 0;
|
||||||
|
if (!userPreferences.cacheMusic ||
|
||||||
|
!isFullRequest ||
|
||||||
|
_cachingInProgress.contains(trackId)) {
|
||||||
return res;
|
return res;
|
||||||
}
|
}
|
||||||
|
_cachingInProgress.add(trackId);
|
||||||
|
|
||||||
final resStream = res.data!.stream.asBroadcastStream();
|
final resStream = res.data!.stream.asBroadcastStream();
|
||||||
|
|
||||||
@ -227,23 +250,25 @@ class ServerPlaybackRoutes {
|
|||||||
|
|
||||||
// Write the stream to the file based on the range
|
// Write the stream to the file based on the range
|
||||||
final partialCacheFileSink =
|
final partialCacheFileSink =
|
||||||
trackPartialCacheFile.openWrite(mode: FileMode.writeOnlyAppend);
|
trackPartialCacheFile.openWrite(mode: FileMode.writeOnly);
|
||||||
final contentRange = res.headers.value("content-range") != null
|
|
||||||
? ContentRangeHeader.parse(res.headers.value("content-range") ?? "")
|
|
||||||
: ContentRangeHeader(0, 0, 0);
|
|
||||||
|
|
||||||
resStream.listen(
|
resStream.listen(
|
||||||
(data) {
|
(data) {
|
||||||
partialCacheFileSink.add(data);
|
partialCacheFileSink.add(data);
|
||||||
},
|
},
|
||||||
onError: (e, stack) {
|
onError: (e, stack) async {
|
||||||
partialCacheFileSink.close();
|
await partialCacheFileSink.close();
|
||||||
|
_cachingInProgress.remove(trackId);
|
||||||
},
|
},
|
||||||
onDone: () async {
|
onDone: () async {
|
||||||
|
try {
|
||||||
await partialCacheFileSink.close();
|
await partialCacheFileSink.close();
|
||||||
|
|
||||||
final fileLength = await trackPartialCacheFile.length();
|
final fileLength = await trackPartialCacheFile.length();
|
||||||
if (fileLength != contentRange.total) return;
|
if (fileLength != contentRange.total) {
|
||||||
|
await trackPartialCacheFile.delete().catchError((_) => trackPartialCacheFile);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
await trackPartialCacheFile.rename(trackCacheFile.path);
|
await trackPartialCacheFile.rename(trackCacheFile.path);
|
||||||
|
|
||||||
@ -265,6 +290,11 @@ class ServerPlaybackRoutes {
|
|||||||
).catchError((e, stackTrace) {
|
).catchError((e, stackTrace) {
|
||||||
AppLogger.reportError(e, stackTrace);
|
AppLogger.reportError(e, stackTrace);
|
||||||
});
|
});
|
||||||
|
} catch (e, stack) {
|
||||||
|
AppLogger.reportError(e, stack);
|
||||||
|
} finally {
|
||||||
|
_cachingInProgress.remove(trackId);
|
||||||
|
}
|
||||||
},
|
},
|
||||||
cancelOnError: true,
|
cancelOnError: true,
|
||||||
);
|
);
|
||||||
@ -277,7 +307,7 @@ class ServerPlaybackRoutes {
|
|||||||
/// @head('/stream/<trackId>')
|
/// @head('/stream/<trackId>')
|
||||||
Future<Response> headStreamTrackId(Request request, String trackId) async {
|
Future<Response> headStreamTrackId(Request request, String trackId) async {
|
||||||
try {
|
try {
|
||||||
final sourcedTrack = await _getSourcedTrack(request, trackId);
|
final sourcedTrack = await _getSourcedTrack(trackId);
|
||||||
|
|
||||||
if (sourcedTrack == null) {
|
if (sourcedTrack == null) {
|
||||||
return Response.notFound("Track not found in the current queue");
|
return Response.notFound("Track not found in the current queue");
|
||||||
@ -301,7 +331,7 @@ class ServerPlaybackRoutes {
|
|||||||
/// @get('/stream/<trackId>')
|
/// @get('/stream/<trackId>')
|
||||||
Future<Response> getStreamTrackId(Request request, String trackId) async {
|
Future<Response> getStreamTrackId(Request request, String trackId) async {
|
||||||
try {
|
try {
|
||||||
final sourcedTrack = await _getSourcedTrack(request, trackId);
|
final sourcedTrack = await _getSourcedTrack(trackId);
|
||||||
|
|
||||||
if (sourcedTrack == null) {
|
if (sourcedTrack == null) {
|
||||||
return Response.notFound("Track not found in the current queue");
|
return Response.notFound("Track not found in the current queue");
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user