diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c64b617..68d0e08b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,8 @@ - **ISO completion diagnostics** — strict per-packet validation now records a bounded snapshot of packet lengths, transfer slots, and the selected wire format on the first mismatch, without accepting packets more loosely. This is diagnostic groundwork for #354, not a compatibility workaround. ### Network & SMB +- **SMB sync resilience** — an unreadable or vanished child folder no longer aborts the whole library scan; directory loops are depth-capped and duplicate paths are skipped. Folder art larger than the server's single-READ limit now loads through the pipelined reader. +- **SMB playback robustness** — a failing progress callback can no longer cancel a shared transfer or start a duplicate download, cache accounting failures no longer discard a completed file, a disappearing cache entry is treated as a miss, a changed remote file invalidates its cached bytes on resync, and a failed next-track download reports cleanly instead of throwing in the background. - **Atomic SMB downloads** — playback, next-track prefetch, and gapless preparation for the same file share one transfer. Downloads publish only after the native completion event, a consistent byte count, and a matching on-disk length; short, cancelled, and incomplete transfers delete their staging file and stay retryable. - **Cache safety** — SMB entries move to a new `smb-v2:` key so files left by older builds are bypassed without clearing app data, and staging and `.part` files are excluded from lookup, eviction, and the cache cap. - **Privacy-safe failures** — SMB errors resolve to allowlisted stage/code pairs and localized copy; hostnames, account names, shares, and paths are never echoed into logs or banners. URL parsing decodes escapes once and rejects embedded credentials, unsupported schemes, query/fragment data, and out-of-range ports; Test Connection probes the configured root. diff --git a/docs/SMB_VALIDATION.md b/docs/SMB_VALIDATION.md index e520d03a..141d23f8 100644 --- a/docs/SMB_VALIDATION.md +++ b/docs/SMB_VALIDATION.md @@ -7,13 +7,26 @@ guest fallback, or change to signing/encryption policy. Test Connection now open the share and lists the configured root folder once; it does not guarantee read permission for every file or successful decoding/output. +Library sync treats the configured root as mandatory but skips an unreadable or +vanished child directory (common on Windows shares: `System Volume +Information`, per-user folders, or a folder deleted mid-scan) instead of +aborting the whole sync. Directory identities are de-duplicated and recursion is +depth-capped so junction/DFS loops cannot run forever. Cover-art reads use the +pipelined reader, so images larger than the server's single-READ limit still +load. + Downloads use a same-directory hidden staging file, verify the server's byte count, then close and atomically rename it. Dart also requires the completion event and checks the published file length. Playback/prefetch for the same -server/file share one download. Older SMB cache entries are bypassed using a new -SMB-only cache key, without clearing app data or other protocols' cache entries. -Normal errors/cancellation clean up staging files; force-killing the process can -leave an unused hidden staging file, which is never considered playable. +server/file share one download. A resync invalidates cached bytes when the +listed file size changes, so a replaced file cannot keep playing from the old +cache. Older SMB cache entries are bypassed using a new SMB-only cache key, +without clearing app data or other protocols' cache entries. Normal +errors/cancellation clean up staging files; force-killing the process can leave +an unused hidden staging file, which is never considered playable. Progress +listeners are isolated from the transfer, so a failing UI callback cannot cancel +a download other callers are waiting on, and the shared-transfer entry is +released when the transfer finishes rather than when its first waiter leaves. Listing-only songs can still show `00:00` before their first download. After a successful download, metadata extraction asynchronously fills missing duration, @@ -33,7 +46,8 @@ flutter test Tests cover interrupted/short downloads and retry, absent/invalid completion, on-disk length validation, same-key concurrency with progress, original bytes, legacy-cache bypass, LRU accounting, staging-file exclusion, URL encoding, -configured-root probing, safe diagnostics and metadata preservation. +configured-root probing, safe diagnostics, metadata preservation, partial +directory walks, and progress-listener isolation. ## Real-device checklist (not replaced by unit tests) @@ -47,7 +61,9 @@ posting addresses, share names or credentials. in their separate fields, never in the URL. 2. Confirm correct credentials work. Wrong credentials, denied shares, missing roots and denied file reads should fail at the corresponding stage. A readable - root with a denied track can legitimately pass Test Connection. + root with a denied track can legitimately pass Test Connection. A denied + child folder should be skipped during sync while sibling folders still + import; a denied configured root must still fail the sync. 3. Play uncached FLAC/MP3 and a large hi-res/DSD original supported by the selected output. Compare the same file played locally in Flick. Confirm duration/format update after caching, and remain after library sync. diff --git a/lib/services/network_cache_service.dart b/lib/services/network_cache_service.dart index 5622930c..20e81a3d 100644 --- a/lib/services/network_cache_service.dart +++ b/lib/services/network_cache_service.dart @@ -58,9 +58,16 @@ class NetworkCacheService { final file = File(p.join(serverDir.path, fileName)); if (!await file.exists()) return null; + // The file can be evicted (or removed by Android's cache manager) between + // the existence check and the LRU touch. Treat that as a miss so callers + // re-download instead of handing playback a path that no longer exists. final now = DateTime.now(); - if (file.lastModifiedSync() != now) { - await file.setLastModified(now); + try { + if (file.lastModifiedSync() != now) { + await file.setLastModified(now); + } + } on FileSystemException { + return null; } return file.path; } @@ -123,6 +130,28 @@ class NetworkCacheService { Future completeDownload(String path) => _evictIfOverCap(protect: File(path)); + /// Drop cached copies for (server, remoteId), across all extensions. Used + /// when a remote file changes so the next play cannot reuse stale bytes. + /// Staging files are left alone: they belong to an in-flight downloader and + /// are never lookup candidates. + Future remove(int remoteServerId, String remoteId) async { + final serverDir = Directory( + p.join((await _root()).path, '$remoteServerId'), + ); + if (!await serverDir.exists()) return; + + final prefix = '${_hash(remoteServerId, remoteId)}.'; + await for (final entry in serverDir.list()) { + if (entry is! File) continue; + if (!p.basename(entry.path).startsWith(prefix)) continue; + try { + await entry.delete(); + } on FileSystemException { + /* Another caller may have removed or locked it; best effort. */ + } + } + } + // ponytail: full O(n) size scan per eviction; track running totals only // if a server with a huge cache ever shows up on profile. Future _evictIfOverCap({File? protect}) async { diff --git a/lib/services/player_service.dart b/lib/services/player_service.dart index 7fee7aba..07058937 100644 --- a/lib/services/player_service.dart +++ b/lib/services/player_service.dart @@ -5606,7 +5606,7 @@ class PlayerService { } } - final nextPath = await _resolveRustPath(nextSong); + final nextPath = await _resolveGaplessNextPath(nextSong); if (nextPath != null) { await _rustAudioService.setReplayGainDefault( await _computeReplayGainDbFor(nextSong), @@ -5630,6 +5630,26 @@ class PlayerService { } } + /// Resolve the next local path without letting an unawaited gapless + /// preparation task throw. Network sources (notably SMB) download on this + /// path; a failed transfer must leave the current track untouched and be + /// reported like any other SMB playback failure. + Future _resolveGaplessNextPath(Song nextSong) async { + try { + return await _resolveRustPath(nextSong); + } on SmbFailure catch (failure) { + _debugLog( + '[crossfade] next-track SMB prepare failed: ' + '${failure.stage}/${failure.code}', + ); + _reportSmbPlaybackFailure(failure); + return null; + } catch (e) { + _debugLog('[crossfade] next-track path resolve failed: $e'); + return null; + } + } + /// Resolves the next song for the just_audio crossfade engine. Mirrors the /// linear-next rule used by the Rust gapless path (shuffle is honoured /// because [_playlist] is pre-shuffled at play time). Returns null when there diff --git a/lib/services/sources/smb_service.dart b/lib/services/sources/smb_service.dart index 8f48996a..41d8b4ed 100644 --- a/lib/services/sources/smb_service.dart +++ b/lib/services/sources/smb_service.dart @@ -276,13 +276,33 @@ class SmbService implements NetworkSourceService { final key = (server.id, remoteId); final job = _downloads.putIfAbsent(key, () => _SmbDownload()); if (onProgress != null) job.listeners.add(onProgress); - job.future ??= _download(server, remoteId, extension, job); + // Tie map cleanup to the transfer itself, not to whichever caller happens + // to stop waiting first. A caller can bail out (e.g. a UI progress + // callback throws, or the screen is disposed) while the shared transfer is + // still running; dropping the map entry there would let the next caller + // start a second download for the same file. + job.future ??= _download(server, remoteId, extension, job).whenComplete(() { + if (identical(_downloads[key], job)) _downloads.remove(key); + }); try { - if (onProgress != null && job.progress != null) onProgress(job.progress!); + final progress = job.progress; + if (onProgress != null && progress != null) { + _notifyProgress(onProgress, progress); + } return await job.future!; } finally { if (onProgress != null) job.listeners.remove(onProgress); - if (identical(_downloads[key], job)) _downloads.remove(key); + } + } + + /// Progress callbacks belong to the UI, never to the transfer. Letting one + /// throw would abort the shared download and delete a file other callers are + /// waiting on. + void _notifyProgress(void Function(double) listener, double progress) { + try { + listener(progress); + } catch (_) { + /* A broken progress listener must not fail playback. */ } } @@ -295,6 +315,9 @@ class SmbService implements NetworkSourceService { // Old versions wrote interrupted transfers to the final filename. A new // SMB-only key bypasses those entries without wiping other protocols/data. final cacheId = 'smb-v2:$remoteId'; + // Callers may omit the extension (prefetch) or pass a different case. + // Derive it from the remote name so every caller resolves the same file. + final cacheExtension = _cacheExtension(extension, remoteId); String? destPath; final clock = Stopwatch()..start(); var received = 0; @@ -303,11 +326,15 @@ class SmbService implements NetworkSourceService { final cached = await _cache.getPath( server.id, cacheId, - extension: extension, + extension: cacheExtension, ); if (cached != null) return cached; final s = _parseServer(server); - destPath = await _cache.pathFor(server.id, cacheId, extension: extension); + destPath = await _cache.pathFor( + server.id, + cacheId, + extension: cacheExtension, + ); var complete = false; await for (final p in _downloader( host: s.host, @@ -327,14 +354,21 @@ class SmbService implements NetworkSourceService { if (total > 0) { job.progress = received / total; for (final listener in List.of(job.listeners)) { - listener(job.progress!); + _notifyProgress(listener, job.progress!); } } } if (!complete || await File(destPath).length() != total) { throw const SmbFailure('download', 'incomplete'); } - await _cache.completeDownload(destPath); + // The download is already complete and playable. Cache accounting is + // best-effort: a failure to evict old entries must not turn a successful + // transfer into a failed playback. + try { + await _cache.completeDownload(destPath); + } catch (_) { + AppLog.instance.add('SMB stage=cache code=storage result=failed'); + } AppLog.instance.add( 'SMB stage=download result=ok received=$received total=$total elapsed_ms=${clock.elapsedMilliseconds}', ); @@ -373,6 +407,15 @@ class SmbService implements NetworkSourceService { final user = server.username ?? ''; final pass = _password(server); + // Remember the sizes of already-listed rows so a replaced file invalidates + // its cached bytes as well as its metadata. SMB listings expose size only. + final previousSizes = {}; + for (final song in await _repo.getSongsByRemoteServer(server.id)) { + final remoteId = song.remoteId; + final size = song.fileSize; + if (remoteId != null && size != null) previousSizes[remoteId] = size; + } + // Collect file entries grouped by parent directory (for per-folder // cover/audio classification), yielding progress as the walk streams in. final byDir = >{}; @@ -430,6 +473,20 @@ class SmbService implements NetworkSourceService { ), ) .toList(); + // Drop bytes cached for a file whose size changed so playback cannot + // serve the previous version. Do this before persisting the new size: + // if the process stops in between, the next sync still sees the old + // listing size and repeats the cleanup instead of accepting stale bytes. + for (final e in songEntries) { + final previous = previousSizes[e.path]; + if (previous != null && previous != e.size.toInt()) { + try { + await _cache.remove(server.id, 'smb-v2:${e.path}'); + } catch (_) { + AppLog.instance.add('SMB stage=cache code=storage result=failed'); + } + } + } await _repo.upsertSmbSongs(entities); syncedRemoteIds.addAll(entities.map((e) => e.remoteId!)); songsFound += entities.length; @@ -498,10 +555,34 @@ class SmbService implements NetworkSourceService { String? _extension(String name) { final dot = name.lastIndexOf('.'); - if (dot <= 0) return null; + if (dot <= 0 || dot == name.length - 1) return null; return name.substring(dot + 1).toLowerCase(); } + /// Prefer the caller's extension, but normalize it and fall back to the + /// remote filename. This keeps `.bin` out of the cache when a caller omits + /// the hint, and lets callers that pass `FLAC` share the file cached as + /// `flac` instead of downloading it twice. + String? _cacheExtension(String? extension, String remoteId) { + String? clean(String? value) { + var candidate = value?.trim().toLowerCase(); + if (candidate == null) return null; + candidate = candidate.replaceFirst(RegExp(r'^\.+'), ''); + if (candidate.isEmpty || + candidate == 'unknown' || + !RegExp(r'^[a-z0-9]+$').hasMatch(candidate)) { + return null; + } + return candidate; + } + + final provided = clean(extension); + if (provided != null) return provided; + final slash = remoteId.lastIndexOf('/'); + final name = slash >= 0 ? remoteId.substring(slash + 1) : remoteId; + return clean(_extension(name)); + } + String _lastPathSegment(String path) { var p = path; while (p.endsWith('/') && p.length > 1) { diff --git a/rust/src/api/smb_api.rs b/rust/src/api/smb_api.rs index f44697db..7ac42dda 100644 --- a/rust/src/api/smb_api.rs +++ b/rust/src/api/smb_api.rs @@ -1,15 +1,41 @@ use crate::frb_generated::StreamSink; -use std::collections::VecDeque; +use std::collections::{HashSet, VecDeque}; use std::fs::{File, OpenOptions}; use std::io::Write; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; -// Only allowlisted stage/code pairs cross the bridge. smb2 errors can contain -// hostnames, account names and remote paths; never attach them as error sources. -fn failure(stage: &str, error: smb2::Error) -> anyhow::Error { +/// Refuse to recurse indefinitely through junctions/DFS links that point back +/// into the share. Real music libraries are far shallower than this. +const MAX_DIRECTORY_DEPTH: usize = 64; + +/// A child directory disappearing or being unreadable must not fail the whole +/// library sync. The configured root still fails loudly: that is the address +/// the user typed and the one they need to fix. +fn skippable_subdirectory_error(error: &smb2::Error) -> bool { use smb2::ErrorKind; - let code = match error.kind() { + matches!( + error.kind(), + ErrorKind::AccessDenied + | ErrorKind::NotFound + | ErrorKind::NotADirectory + | ErrorKind::InvalidName + | ErrorKind::Unsupported + ) +} + +/// The root is the address the user typed, so its failure must propagate. Only +/// errors from directories discovered below it are safe to skip. +fn should_skip_directory_error(root_listed: bool, error: &smb2::Error) -> bool { + root_listed && skippable_subdirectory_error(error) +} + +/// Only allowlisted stage/code pairs cross the bridge. smb2 errors can contain +/// hostnames, account names and remote paths; never attach them as error sources. +fn error_code(error: &smb2::Error) -> &'static str { + use smb2::ErrorKind; + match error.kind() { ErrorKind::AuthRequired => "auth", ErrorKind::SigningRequired => "signing", ErrorKind::AccessDenied => "denied", @@ -18,8 +44,31 @@ fn failure(stage: &str, error: smb2::Error) -> anyhow::Error { ErrorKind::ConnectionLost | ErrorKind::SessionExpired => "connection", ErrorKind::Io => "connection", _ => "protocol", - }; - anyhow::anyhow!("SMB stage={} code={}", stage, code) + } +} + +fn failure(stage: &str, error: smb2::Error) -> anyhow::Error { + anyhow::anyhow!("SMB stage={} code={}", stage, error_code(&error)) +} + +/// Shared client configuration. `smb2::connect`'s shorthand is a five-second, +/// no-reconnect client; a share on a waking NAS, or a Wi-Fi roam between +/// tracks, should recover in place instead of failing the whole transfer. +fn client_config(host: &str, port: u16, user: &str, pass: &str) -> smb2::ClientConfig { + smb2::ClientConfig { + addr: addr(host, port), + username: user.to_string(), + password: pass.to_string(), + timeout: Duration::from_secs(15), + auto_reconnect: true, + ..Default::default() + } +} + +async fn connect(host: &str, port: u16, user: &str, pass: &str) -> anyhow::Result { + smb2::SmbClient::connect(client_config(host, port, user, pass)) + .await + .map_err(|e| failure("connect", e)) } /// Same-directory staging file. Drop cleans up on error, stream cancellation, @@ -114,9 +163,7 @@ pub async fn smb_ping( pass: String, root_path: String, ) -> anyhow::Result<()> { - let mut client = smb2::connect(&addr(&host, port), &user, &pass) - .await - .map_err(|e| failure("connect", e))?; + let mut client = connect(&host, port, &user, &pass).await?; let mut tree = client .connect_share(&share) .await @@ -131,6 +178,12 @@ pub async fn smb_ping( /// BFS walk of `root_path` (relative to the share) pushing one [SmbEntry] per /// file or directory. `root_path` of "" lists the share root. One connection, /// reused across directories. +/// +/// A child directory that is unreadable or vanished during the walk is skipped +/// instead of aborting the scan; Windows shares commonly contain things like +/// `System Volume Information` that even a valid account cannot list. Failing +/// to list the configured root itself is still an error, since that is the +/// address the user needs to correct. pub async fn smb_list_share( host: String, port: u16, @@ -140,22 +193,31 @@ pub async fn smb_list_share( root_path: String, sink: StreamSink, ) -> anyhow::Result<()> { - let mut client = smb2::connect(&addr(&host, port), &user, &pass) - .await - .map_err(|e| failure("connect", e))?; + let mut client = connect(&host, port, &user, &pass).await?; let mut tree = client .connect_share(&share) .await .map_err(|e| failure("share", e))?; - let mut queue: VecDeque = VecDeque::new(); - queue.push_back(root_path); + let mut queue: VecDeque<(String, usize)> = VecDeque::new(); + let mut visited: HashSet = HashSet::new(); + visited.insert(root_path.clone()); + queue.push_back((root_path, 0)); + let mut root_listed = false; + let mut skipped = 0_u32; - while let Some(dir) = queue.pop_front() { - let entries = client - .list_directory(&mut tree, &dir) - .await - .map_err(|e| failure("root", e))?; + while let Some((dir, depth)) = queue.pop_front() { + let entries = match client.list_directory(&mut tree, &dir).await { + Ok(entries) => entries, + Err(e) if should_skip_directory_error(root_listed, &e) => { + skipped += 1; + continue; + } + Err(e) => { + return Err(failure(if root_listed { "read" } else { "root" }, e)); + } + }; + root_listed = true; for e in entries { if e.name == "." || e.name == ".." { continue; @@ -173,14 +235,27 @@ pub async fn smb_list_share( }) .map_err(|_| anyhow::anyhow!("SMB stage=root code=cancelled"))?; if e.is_directory { - queue.push_back(child_path); + if depth >= MAX_DIRECTORY_DEPTH || !visited.insert(child_path.clone()) { + skipped += 1; + continue; + } + queue.push_back((child_path, depth + 1)); } } } + if skipped > 0 { + // Numeric only: never log share, directory or file names. + log::info!("SMB stage=root result=partial skipped={skipped}"); + } Ok(()) } /// Read a full (small) file into memory — cover art and tiny sidecars. +/// +/// Uses the pipelined reader rather than `read_file`: the latter is a single +/// READ and fails with `FileTooLargeForSingleRead` when the file exceeds the +/// server's MaxReadSize (as small as 64 KiB on some NAS firmware). Folder art +/// routinely exceeds that, which made SMB covers fail on those servers. pub async fn smb_read_file( host: String, port: u16, @@ -189,15 +264,13 @@ pub async fn smb_read_file( user: String, pass: String, ) -> anyhow::Result> { - let mut client = smb2::connect(&addr(&host, port), &user, &pass) - .await - .map_err(|e| failure("connect", e))?; + let mut client = connect(&host, port, &user, &pass).await?; let mut tree = client .connect_share(&share) .await .map_err(|e| failure("share", e))?; let bytes = client - .read_file(&mut tree, &path) + .read_file_pipelined(&mut tree, &path) .await .map_err(|e| failure("read", e))?; Ok(bytes) @@ -216,9 +289,7 @@ pub async fn smb_download_file( dest_path: String, sink: StreamSink, ) -> anyhow::Result<()> { - let mut client = smb2::connect(&addr(&host, port), &user, &pass) - .await - .map_err(|e| failure("connect", e))?; + let mut client = connect(&host, port, &user, &pass).await?; let tree = client .connect_share(&share) .await @@ -261,10 +332,14 @@ pub async fn smb_download_file( #[cfg(test)] mod tests { - use super::{addr, failure, PendingDownload}; + use super::{ + addr, client_config, failure, should_skip_directory_error, skippable_subdirectory_error, + PendingDownload, + }; use std::io::Write; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; + use std::time::Duration; fn test_dir() -> PathBuf { static ID: AtomicU64 = AtomicU64::new(0); @@ -356,4 +431,42 @@ mod tests { assert_eq!(addr("192.168.1.10", 445), "192.168.1.10:445"); assert_eq!(addr("nas.local", 1445), "nas.local:1445"); } + + #[test] + fn unreadable_child_directories_are_skippable_but_connection_loss_is_not() { + use smb2::types::status::NtStatus; + use smb2::types::Command; + + let denied = smb2::Error::Protocol { + status: NtStatus::ACCESS_DENIED, + command: Command::QueryDirectory, + }; + assert!(skippable_subdirectory_error(&denied)); + assert!(!should_skip_directory_error(false, &denied)); + assert!(should_skip_directory_error(true, &denied)); + + let vanished = smb2::Error::Protocol { + status: NtStatus::OBJECT_NAME_NOT_FOUND, + command: Command::Create, + }; + assert!(skippable_subdirectory_error(&vanished)); + assert!(should_skip_directory_error(true, &vanished)); + + assert!(!skippable_subdirectory_error(&smb2::Error::Disconnected)); + assert!(!skippable_subdirectory_error(&smb2::Error::Timeout)); + assert!(!should_skip_directory_error( + true, + &smb2::Error::Disconnected + )); + } + + #[test] + fn client_config_keeps_credentials_and_allows_reconnect() { + let config = client_config("nas.local", 1445, "user", "pass"); + assert_eq!(config.addr, "nas.local:1445"); + assert_eq!(config.username, "user"); + assert_eq!(config.password, "pass"); + assert!(config.auto_reconnect); + assert_eq!(config.timeout, Duration::from_secs(15)); + } } diff --git a/test/services/network_cache_service_test.dart b/test/services/network_cache_service_test.dart index 81207408..6e7a03d9 100644 --- a/test/services/network_cache_service_test.dart +++ b/test/services/network_cache_service_test.dart @@ -114,4 +114,25 @@ void main() { expect(await staging.exists(), isTrue); expect(await oldStylePart.exists(), isTrue); }); + + test( + 'remove drops every extension for one song and leaves staging alone', + () async { + final flac = await service.stash(1, 'songA', [1], extension: 'flac'); + final dsf = await service.stash(1, 'songA', [2], extension: 'dsf'); + final other = await service.stash(1, 'songB', [3], extension: 'flac'); + final dest = await service.pathFor(1, 'songA', extension: 'flac'); + final staging = File( + '${File(dest).parent.path}/.${File(dest).uri.pathSegments.last}.1.part', + ); + await staging.writeAsBytes([9]); + + await service.remove(1, 'songA'); + + expect(await File(flac).exists(), isFalse); + expect(await File(dsf).exists(), isFalse); + expect(await staging.exists(), isTrue); + expect(await File(other).exists(), isTrue); + }, + ); } diff --git a/test/services/smb_service_download_test.dart b/test/services/smb_service_download_test.dart index 064e44e9..895363d2 100644 --- a/test/services/smb_service_download_test.dart +++ b/test/services/smb_service_download_test.dart @@ -87,6 +87,71 @@ void main() { }, ); + test( + 'a throwing progress callback cannot abort or duplicate the transfer', + () async { + final half = Completer(); + final first = service.stream( + server, + 'song.flac', + extension: 'flac', + onProgress: (value) { + if (value == 0.5 && !half.isCompleted) half.complete(); + }, + ); + final dest = await started.future; + await File(dest).writeAsBytes([1, 2, 3, 4]); + + progress.add(update(2, 4)); + await half.future; + + // Joining after progress already exists exercises the catch-up path; + // the completion event exercises the live listener path. Neither may + // escape or cancel the shared transfer. + final broken = service.stream( + server, + 'song.flac', + extension: 'flac', + onProgress: (_) => throw StateError('broken UI callback'), + ); + final alsoJoined = service.stream( + server, + 'song.flac', + extension: 'flac', + ); + + progress.add(update(4, 4, complete: true)); + await progress.close(); + + expect(await first, dest); + expect(await broken, dest); + expect(await alsoJoined, dest); + expect(calls, 1); + expect(await File(dest).readAsBytes(), [1, 2, 3, 4]); + }, + ); + + test( + 'cache extension is normalized and derived from the remote path', + () async { + final first = service.stream(server, 'album/song.FLAC'); + final dest = await started.future; + expect(dest, endsWith('.flac')); + await File(dest).writeAsBytes([1]); + progress.add(update(1, 1, complete: true)); + await progress.close(); + expect(await first, dest); + + // A later caller with an explicit (differently cased/spaced) hint must + // hit the same cached file instead of downloading it again. + expect( + await service.stream(server, 'album/song.FLAC', extension: ' FLAC '), + dest, + ); + expect(calls, 1); + }, + ); + for (final terminal in [false, true]) { test( 'rejects ${terminal ? 'wrong byte count' : 'missing completion'} and retries',