From b24d6603949e1477ad7abecf3d899eb7d66436d5 Mon Sep 17 00:00:00 2001 From: Colton Date: Mon, 9 Nov 2026 18:27:28 -0500 Subject: [PATCH 1/8] CacheState carries filePosition, the bytes saved in the cache file --- lib/src/models/cache_state/cache_state.dart | 24 ++++++++++++++++----- test/unit/cache_state_test.dart | 17 ++++++++++++++- 2 files changed, 35 insertions(+), 6 deletions(-) diff --git a/lib/src/models/cache_state/cache_state.dart b/lib/src/models/cache_state/cache_state.dart index 48c7a7c..0f32764 100644 --- a/lib/src/models/cache_state/cache_state.dart +++ b/lib/src/models/cache_state/cache_state.dart @@ -3,8 +3,8 @@ sealed class CacheState { const factory CacheState.zero() = IncompleteCacheState.zero; - const factory CacheState.incomplete(int position, int? sourceLength) = - IncompleteCacheState; + const factory CacheState.incomplete(int position, int? sourceLength, + {int? filePosition}) = IncompleteCacheState; const factory CacheState.complete(int sourceLength) = CompleteCacheState; @@ -12,6 +12,10 @@ sealed class CacheState { /// For an active download, this may be ahead of the current read position. For a completed cache, this will match [sourceLength]. int get position; + /// Bytes saved in the cache file on disk. + /// For an active download, this may be behind [position] by the bytes it has yet to save. For a completed cache, this will match [sourceLength]. + int get filePosition; + /// Total source size in bytes, or null if the server hasn't reported it yet. int? get sourceLength; @@ -30,18 +34,24 @@ sealed class CacheState { @override String toString() => - 'CacheState(position: $position, sourceLength: $sourceLength, isComplete: $isComplete)'; + 'CacheState(position: $position, filePosition: $filePosition, sourceLength: $sourceLength, isComplete: $isComplete)'; } /// Cache state while a download is in progress, paused, or not yet started. final class IncompleteCacheState extends CacheState { - const IncompleteCacheState(this.position, this.sourceLength); + /// [filePosition] defaults to [position]: every byte saved. + const IncompleteCacheState(this.position, this.sourceLength, + {final int? filePosition}) + : filePosition = filePosition ?? position; const IncompleteCacheState.zero() : this(0, null); @override final int position; + @override + final int filePosition; + @override final int? sourceLength; @@ -73,11 +83,12 @@ final class IncompleteCacheState extends CacheState { if (identical(this, other)) return true; return other is IncompleteCacheState && position == other.position && + filePosition == other.filePosition && sourceLength == other.sourceLength; } @override - int get hashCode => Object.hash(position, sourceLength); + int get hashCode => Object.hash(position, filePosition, sourceLength); } /// Cache state once the file has been fully downloaded and renamed into place. @@ -90,6 +101,9 @@ final class CompleteCacheState extends CacheState { @override int get position => sourceLength; + @override + int get filePosition => sourceLength; + @override bool get isComplete => true; diff --git a/test/unit/cache_state_test.dart b/test/unit/cache_state_test.dart index 028225b..d91e801 100644 --- a/test/unit/cache_state_test.dart +++ b/test/unit/cache_state_test.dart @@ -6,6 +6,7 @@ void main() { test('has no position, length or progress', () { const s = CacheState.zero(); expect(s.position, 0); + expect(s.filePosition, 0); expect(s.sourceLength, isNull); expect(s.isComplete, isFalse); expect(s.progress, isNull); @@ -34,6 +35,15 @@ void main() { expect(s.remainingBytes, isNull); }); + test('every byte is saved unless a file position is given', () { + expect(const CacheState.incomplete(50, 100).filePosition, 50); + const s = CacheState.incomplete(50, 100, filePosition: 30); + expect(s.position, 50); + expect(s.filePosition, 30); + expect(s.progress, closeTo(0.5, 1e-9), + reason: 'progress follows the downloaded position'); + }); + test('zero source length means null progress', () { const s = CacheState.incomplete(0, 0); expect(s.progress, isNull); @@ -46,16 +56,21 @@ void main() { expect(s.isComplete, isTrue); expect(s.progress, 1.0); expect(s.position, 100); + expect(s.filePosition, 100); expect(s.remainingBytes, 0); }); }); group('equality', () { - test('incomplete states compare by position and length', () { + test('incomplete states compare by positions and length', () { expect(const CacheState.incomplete(10, 100), const CacheState.incomplete(10, 100)); expect(const CacheState.incomplete(10, 100), isNot(const CacheState.incomplete(20, 100))); + expect(const CacheState.incomplete(10, 100), + const CacheState.incomplete(10, 100, filePosition: 10)); + expect(const CacheState.incomplete(10, 100), + isNot(const CacheState.incomplete(10, 100, filePosition: 5))); }); test('complete states compare by source length', () { From 90beed4274f847066006a437abf58746056d3879 Mon Sep 17 00:00:00 2001 From: Colton Date: Mon, 9 Nov 2026 18:29:14 -0500 Subject: [PATCH 2/8] Reset a download's cache even while it waits between responses resetCache(), or any reset for an invalid cache, cancels the running download with the exception, and the download resets the cache when it ends on it. A download waiting for a response, or to retry, had no response for the exception to end, so it returned as if stopped and resumed the old partial cache. The download now ends on it either way, and a download cancelled for an invalid cache serves no more requests. --- .../cache_downloader/cache_downloader.dart | 16 +++++ test/e2e/reset_cache_test.dart | 64 +++++++++++++++++++ 2 files changed, 80 insertions(+) create mode 100644 test/e2e/reset_cache_test.dart diff --git a/lib/src/cache_stream/cache_downloader/cache_downloader.dart b/lib/src/cache_stream/cache_downloader/cache_downloader.dart index e03ec73..69b3169 100644 --- a/lib/src/cache_stream/cache_downloader/cache_downloader.dart +++ b/lib/src/cache_stream/cache_downloader/cache_downloader.dart @@ -27,6 +27,9 @@ class CacheDownloader { ///Headers received and validated CachedResponseHeaders? _validatedHeaders; + + ///The invalid cache [cancel] stopped the download for, if any. The download ends on it, and serves no more requests. + InvalidCacheException? _invalidCache; CacheDownloader._( final CacheMetadata cacheMetadata, final int startPosition, @@ -109,6 +112,11 @@ class CacheDownloader { } }, ); + //Cancelled for an invalid cache between responses (waiting for one, or to retry), the download + //returns as if stopped; it ends on the exception, as when a response is cancelled for it. + if (_invalidCache case final InvalidCacheException invalidCache) { + throw invalidCache; + } } on InvalidCacheException { _validatedHeaders = null; rethrow; @@ -144,7 +152,11 @@ class CacheDownloader { } /// Cancels the download and closes the stream. An optional [error] can be provided to indicate the reason for cancellation. + /// An [InvalidCacheException] ends [download] on it, and the download serves no more requests. Future cancel([Object? exception]) async { + if (exception is InvalidCacheException) { + _invalidCache ??= exception; + } _downloader.close(exception); _paused = false; return _completer.future; @@ -164,6 +176,7 @@ class CacheDownloader { bool processRequest(final StreamRequest request) { assert(!_paused); + if (isInvalidated) return false; if (request.start > downloadPosition) return false; final headers = _validatedHeaders; if (headers == null) return false; @@ -193,5 +206,8 @@ class CacheDownloader { int get filePosition => _sink.flushedBytes; Uri get sourceUrl => _downloader.sourceUrl; bool get isClosed => _completer.isCompleted; + + ///If [cancel] stopped the download for an invalid cache. + bool get isInvalidated => _invalidCache != null; bool get isPaused => _paused; } diff --git a/test/e2e/reset_cache_test.dart b/test/e2e/reset_cache_test.dart new file mode 100644 index 0000000..235f674 --- /dev/null +++ b/test/e2e/reset_cache_test.dart @@ -0,0 +1,64 @@ +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; + +import '../support/harness.dart'; + +/// Waits for [condition], polling, and fails if it does not hold within 5 s. +Future _until(bool Function() condition) async { + for (var i = 0; i < 500 && !condition(); i++) { + await Future.delayed(const Duration(milliseconds: 10)); + } + expect(condition(), isTrue); +} + +void main() { + late CacheTestHarness h; + + setUp(() async { + h = CacheTestHarness(); + await h.setUp(); + }); + + tearDown(() => h.tearDown()); + + test('resetCache while the download waits for the origin resets the cache', + () async { + final source = h.origin.url('/reset-waiting.mp3'); + final half = h.origin.payload.length ~/ 2; + + // A partial cache of the first half, as an interrupted download leaves it. + final firstGate = Completer(); + h.origin + ..responseBodyGate = firstGate + ..responseBodyGateAfterBytes = half; + final interrupted = h.manager.createStream(source); + interrupted.download().ignore(); + await _until(() => interrupted.cachePosition >= half); + await interrupted.dispose(force: true); + h.origin + ..responseBodyGate = null + ..responseBodyGateAfterBytes = null; + firstGate.complete(); + + // The resumed download waits for the origin's response when the cache is + // reset: no response is open for the reset to end. + final startGate = Completer(); + h.origin.responseStartGate = startGate; + final requestsBefore = h.origin.requestCount; + final stream = h.manager.createStream(source); + stream.download().ignore(); + await _until(() => h.origin.requestCount == requestsBefore + 1); + expect(h.origin.lastRangeHeader, 'bytes=$half-'); + + final reset = stream.resetCache(); + startGate.complete(); + await reset.timeout(const Duration(seconds: 5)); + + // The download starts over, from the start, not from the reset partial. + await _until(() => h.origin.requestCount == requestsBefore + 2); + expect(h.origin.lastRangeHeader, isNull); + + await stream.dispose(force: true); + }); +} From afb18e8f8798f597fed2216a77c92455c455611e Mon Sep 17 00:00:00 2001 From: Colton Date: Mon, 9 Nov 2026 18:31:45 -0500 Subject: [PATCH 3/8] Revalidate the cache files on every request _cacheFileState checks the cache files against the running download: a partial cache file missing bytes the download saved (deleted underneath, by the device's cache management, say) is an invalid cache, and refreshCacheState resets it. request() refreshes the cache state every time, so a cache whose files were deleted or changed is downloaded again rather than failing the response, and the ad hoc deletion check in request() goes. While a download runs, the state takes its position (ahead of the file by the bytes it has yet to save) and source length, so a refresh no longer drops the position of a download of unknown length to zero, and the partial file is renamed only once the download stops writing to it. A reset of a download that stopped writing (completing, and refreshing on the way) is done at once rather than waiting for the download, which waits for the refresh. --- .../cache_downloader/cache_downloader.dart | 3 + lib/src/cache_stream/http_cache_stream.dart | 214 +++++++++--------- test/e2e/cache_files_changed_test.dart | 122 ++++++++++ test/e2e/partial_deleted_test.dart | 70 ------ 4 files changed, 236 insertions(+), 173 deletions(-) create mode 100644 test/e2e/cache_files_changed_test.dart delete mode 100644 test/e2e/partial_deleted_test.dart diff --git a/lib/src/cache_stream/cache_downloader/cache_downloader.dart b/lib/src/cache_stream/cache_downloader/cache_downloader.dart index 69b3169..3351dff 100644 --- a/lib/src/cache_stream/cache_downloader/cache_downloader.dart +++ b/lib/src/cache_stream/cache_downloader/cache_downloader.dart @@ -204,6 +204,9 @@ class CacheDownloader { _validatedHeaders?.sourceLength ?? _resumeHeaders?.sourceLength; int get downloadPosition => _position; int get filePosition => _sink.flushedBytes; + + ///If the download may still write to the partial cache file. + bool get isWriting => !_sink.isClosed; Uri get sourceUrl => _downloader.sourceUrl; bool get isClosed => _completer.isCompleted; diff --git a/lib/src/cache_stream/http_cache_stream.dart b/lib/src/cache_stream/http_cache_stream.dart index 36f4d8a..79ff15d 100644 --- a/lib/src/cache_stream/http_cache_stream.dart +++ b/lib/src/cache_stream/http_cache_stream.dart @@ -108,34 +108,22 @@ class HttpCacheStream { await _ensureInit(); _checkDisposed(); + //Revalidated for every request: cache files deleted or changed underneath + //(by the device's cache management, say) reset the cache here, and the + //request waits for its range to download again, rather than its response + //failing to read them. + final state = await refreshCacheState(); final responseHeaders = _cachedResponseHeaders; final range = IntRange.validate(start, end, responseHeaders?.sourceLength); - if (responseHeaders != null && cacheState.isComplete) { - final verifiedCacheState = await refreshCacheState(); - if (verifiedCacheState.isComplete) { - return FileStreamResponse(range, files, responseHeaders); - } - } - - //Deleted while it downloads (by the device's cache management, say), the - //partial cache file takes the download's writes into a file no response can - //open, until the download ends. Reset, the download starts again, and the - //request waits for its range. - final activeDownloader = _cacheDownloader; - final cacheDeleted = activeDownloader != null && - !activeDownloader.isClosed && - activeDownloader.filePosition > 0 && - !await files.activeCacheFile().exists(); - if (cacheDeleted) { - await _resetCache(InvalidCacheSizeException( - sourceUrl, -1, responseHeaders?.sourceLength ?? 0)); + if (responseHeaders != null && state.isComplete) { + return FileStreamResponse(range, files, responseHeaders); } final rangeThreshold = config.rangeRequestSplitThreshold; if (rangeThreshold != null && range.start >= rangeThreshold && - (range.start - cachePosition) >= rangeThreshold) { + (range.start - state.position) >= rangeThreshold) { return RangeDownloadStreamResponse.construct(sourceUrl, range, config); } @@ -146,9 +134,7 @@ class HttpCacheStream { final streamRequest = StreamRequest.construct(range); final downloader = _cacheDownloader; - if (!cacheDeleted && - downloader != null && - downloader.processRequest(streamRequest)) { + if (downloader != null && downloader.processRequest(streamRequest)) { return streamRequest.response; //Request was processed immediately } else { _queuedRequests @@ -250,8 +236,9 @@ class HttpCacheStream { await downloader.download( onPosition: (position) { endError = null; //It moved on, past any error before - _updateCacheState( - CacheState.incomplete(position, downloader.sourceLength)); + _updateCacheState(CacheState.incomplete( + position, downloader.sourceLength, + filePosition: downloader.filePosition)); while (_queuedRequests.isNotEmpty && downloader.processRequest(_queuedRequests.first)) { _queuedRequests.removeAt(0); @@ -390,28 +377,32 @@ class HttpCacheStream { Future _resetCache(final InvalidCacheException exception) { final downloader = _cacheDownloader; if (downloader != null && !downloader.isClosed) { - return downloader.cancel( - exception); //Close the ongoing download, which will rethrow the exception and reset the cache - } else { - return _fileLock.synchronized(() async { - try { - _cachedResponseHeaders = null; - _updateCacheState(const CacheState.zero()); - if (exception is! CacheResetException) { - _addError(exception, closeRequests: false); - } - await files.delete(partialOnly: false); - } catch (e) { - _addError(e, closeRequests: false); - } finally { - if (_queuedRequests.isNotEmpty && !isDownloading && isRetained) { - //Restart download to fulfill pending requests - Timer.run(() => download() - .ignore()); //Use Timer.run to avoid calling download() within the lock - } - } - }); + //The download serves no more requests from the cache + final cancelled = downloader.cancel(exception); + if (downloader.isWriting) { + return cancelled; //It ends on the exception, and resets the cache + } + //It stopped writing, and completes (refreshing the cache state on the + //way, which may be what is resetting it): reset here, without waiting. } + return _fileLock.synchronized(() async { + try { + _cachedResponseHeaders = null; + _updateCacheState(const CacheState.zero()); + if (exception is! CacheResetException) { + _addError(exception, closeRequests: false); + } + await files.delete(partialOnly: false); + } catch (e) { + _addError(e, closeRequests: false); + } finally { + if (_queuedRequests.isNotEmpty && !isDownloading && isRetained) { + //Restart download to fulfill pending requests + Timer.run(() => download() + .ignore()); //Use Timer.run to avoid calling download() within the lock + } + } + }); } Future _setCachedResponseHeaders(CachedResponseHeaders headers) { @@ -430,84 +421,95 @@ class HttpCacheStream { }); } + /// Reads the cache state from the cache files, checked against the running download, if any, and emits it. + /// + /// Resets the cache when its files no longer hold what it recorded (see [InvalidCacheException]), and returns + /// [CacheState.zero]. Future refreshCacheState() async { - final state = await _fileLock.synchronized(_cacheFileState); - _updateCacheState(state); - return state; + try { + final state = await _fileLock.synchronized(_cacheFileState); + _updateCacheState(state); + return state; + } on InvalidCacheException catch (e) { + await _resetCache(e); + return const CacheState.zero(); + } } + /// The cache state the cache files hold, checked against the running download, if any. + /// + /// Throws an [InvalidCacheSizeException] when the files no longer hold what the cache recorded: a complete + /// cache file of the wrong size, a partial one longer than the source, or a partial one missing bytes the + /// download saved to it. Future _cacheFileState() async { assert(_fileLock.locked, - 'fileCacheState must be called within _fileLock.synchronized()'); + '_cacheFileState must be called within _fileLock.synchronized()'); + final downloader = _activeDownloader; + //Read before the file is: by then the file holds at least these bytes, while + //bytes saved during the stat may be missing from its size. + final savedBytes = downloader?.filePosition ?? 0; final sourceLength = _cachedResponseHeaders?.sourceLength; - if (sourceLength == null) return const CacheState.zero(); - - InvalidCacheException? cacheException; - try { + if (sourceLength != null) { final completeCacheStat = await files.complete.stat(); if (completeCacheStat.type == FileSystemEntityType.file) { InvalidCacheSizeException.validate( sourceUrl, completeCacheStat.size, sourceLength); return CacheState.complete(completeCacheStat.size); } - } catch (e) { - if (e is InvalidCacheException) { - await files.complete.delete().ignoreResult(); - cacheException = e; - } - _addError(e, closeRequests: false); + } else if (downloader == null) { + return const CacheState.zero(); //Nothing to resume without the length } - try { - final partialCacheStat = await files.partial.stat(); + final partialCacheStat = await files.partial.stat(); + final partialSize = partialCacheStat.type == FileSystemEntityType.file + ? partialCacheStat.size + : null; + + //Deleted or truncated while the download writes into it (by the device's + //cache management, say), the partial cache file lost bytes the download + //saved: no response can read them, and the download saves on into a file + //that no longer holds them (a deleted one, or at the wrong offset). + if (savedBytes > (partialSize ?? 0)) { + throw InvalidCacheSizeException(sourceUrl, partialSize ?? -1, savedBytes); + } - if (partialCacheStat.type == FileSystemEntityType.file) { - InvalidCacheSizeException.validate( - sourceUrl, partialCacheStat.size, sourceLength, - partial: true); - - if (partialCacheStat.size == sourceLength) { - try { - await files.partial.rename(files.complete - .path); //Rename the partial cache to the complete cache - return CacheState.complete(partialCacheStat.size); - } on FileSystemException catch (e) { - final completeCacheStat = await files.complete.stat(); - if (completeCacheStat.type == FileSystemEntityType.file && - completeCacheStat.size == sourceLength) { - return CacheState.complete(completeCacheStat - .size); //Renamed by another process, treat as complete. - } - //Rename can fail if the file is open by a response stream on Windows. - if (lastErrorOrNull is! FileSystemException) { - _addError(e, - closeRequests: - false); //Prevent spamming the error log with repeated rename failures - } + if (partialSize != null && sourceLength != null) { + InvalidCacheSizeException.validate(sourceUrl, partialSize, sourceLength, + partial: true); + + //Renamed once the download stops writing to it + if (partialSize == sourceLength && downloader?.isWriting != true) { + try { + await files.partial.rename(files + .complete.path); //Rename the partial cache to the complete cache + return CacheState.complete(partialSize); + } on FileSystemException catch (e) { + final completeCacheStat = await files.complete.stat(); + if (completeCacheStat.type == FileSystemEntityType.file && + completeCacheStat.size == sourceLength) { + return CacheState.complete(completeCacheStat + .size); //Renamed by another process, treat as complete. + } + //Rename can fail if the file is open by a response stream on Windows. + if (lastErrorOrNull is! FileSystemException) { + _addError(e, + closeRequests: + false); //Prevent spamming the error log with repeated rename failures } } - - return CacheState.incomplete(partialCacheStat.size, sourceLength); - } - } catch (e) { - if (e is InvalidCacheException) { - cacheException = e; } - _addError(e, closeRequests: false); } - if (cacheException != null && _cacheDownloader?.isClosed != false) { - _cachedResponseHeaders = - null; //Reset cached headers if the cache is invalid - await files.delete(partialOnly: false).ignoreResult(); - if (_queuedRequests.isNotEmpty && !isDownloading && isRetained) { - Timer.run(() => download().ignore()); - } - return const CacheState.zero(); + if (downloader != null) { + //The download is ahead of the file by the bytes it has yet to save. A + //partial cache file it did not resume holds none of its bytes until it + //saves them. + return CacheState.incomplete( + downloader.downloadPosition, sourceLength ?? downloader.sourceLength, + filePosition: downloader.filePosition); } - - return CacheState.incomplete(0, sourceLength); + return CacheState.incomplete(partialSize ?? 0, sourceLength); } void _updateCacheState(final CacheState cacheState) { @@ -571,7 +573,13 @@ class HttpCacheStream { /// Bytes currently available in the cache (downloaded or on disk). /// For an active download, this may be ahead of the current read position. For a completed cache, this will match [sourceLength]. int get cachePosition => - _cacheDownloader?.downloadPosition ?? cacheState.position; + _activeDownloader?.downloadPosition ?? cacheState.position; + + ///The download filling the cache, unless it was cancelled for an invalid cache. + CacheDownloader? get _activeDownloader { + final downloader = _cacheDownloader; + return downloader == null || downloader.isInvalidated ? null : downloader; + } /// If this [HttpCacheStream] is retained. /// diff --git a/test/e2e/cache_files_changed_test.dart b/test/e2e/cache_files_changed_test.dart new file mode 100644 index 0000000..6f8cbb3 --- /dev/null +++ b/test/e2e/cache_files_changed_test.dart @@ -0,0 +1,122 @@ +import 'dart:async'; +import 'dart:typed_data'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:http_cache_stream/http_cache_stream.dart'; +import 'package:http_cache_stream/src/etc/extensions/file_extensions.dart'; + +import '../support/harness.dart'; +import '../support/payload.dart'; + +/// Waits for [condition], polling, and fails if it does not hold within 5 s. +Future _until(bool Function() condition) async { + for (var i = 0; i < 500 && !condition(); i++) { + await Future.delayed(const Duration(milliseconds: 10)); + } + expect(condition(), isTrue); +} + +/// Cache files deleted or changed underneath a stream (by the device's cache +/// management, say). Each request revalidates them, and a cache they no longer +/// hold is reset and downloaded again. +void main() { + late CacheTestHarness h; + + setUp(() async { + h = CacheTestHarness(); + await h.setUp(); + }); + + tearDown(() => h.tearDown()); + + Future read(HttpCacheStream stream, int start, int end) async { + final response = await stream.request(start: start, end: end); + final builder = BytesBuilder(copy: false); + await for (final chunk in response.stream) { + builder.add(chunk); + } + return builder.takeBytes(); + } + + /// The [InvalidCacheSizeException]s [stream] reports, as their sizes. + List listenForInvalidSizes(HttpCacheStream stream) { + final sizes = []; + stream.cacheStateStream.listen(null, onError: (Object e) { + if (e is InvalidCacheSizeException) sizes.add(e.size); + }); + return sizes; + } + + test( + 'a range it had downloaded is read again after the partial file was ' + 'deleted', () async { + final half = h.origin.payload.length ~/ 2; + final bodyGate = Completer(); + h.origin + ..responseBodyGate = bodyGate + ..responseBodyGateAfterBytes = half; + final stream = h.manager.createStream(h.origin.url('/deleted.mp3')); + final invalidSizes = listenForInvalidSizes(stream); + final download = stream.download(); + await _until(() => stream.cachePosition >= half); + final partial = stream.files.partial; + await _until(() => (partial.lengthSyncOrNull() ?? 0) >= half); + await partial.delete(); + + // The download writes on into the deleted file, where no response can + // open what it holds. + final bytes = + await read(stream, 1000, 9000).timeout(const Duration(seconds: 10)); + expect(bytes, Uint8List.sublistView(h.origin.payload, 1000, 9000)); + expect(invalidSizes, contains(-1), reason: 'the deletion is reported'); + + bodyGate.complete(); + final file = await download.timeout(const Duration(seconds: 15)); + expect(Payload.hash(await file.readAsBytes()), h.payloadHash); + await stream.dispose(force: true); + }); + + test('a complete cache file of the wrong size is downloaded again', () async { + final stream = h.manager.createStream(h.origin.url('/truncated.mp3')); + final invalidSizes = listenForInvalidSizes(stream); + final file = await stream.download(); + await file.writeAsBytes(Uint8List(1000)); + + final bytes = + await read(stream, 1000, 9000).timeout(const Duration(seconds: 10)); + expect(bytes, Uint8List.sublistView(h.origin.payload, 1000, 9000)); + expect(invalidSizes, contains(1000)); + + final downloaded = + await stream.download().timeout(const Duration(seconds: 15)); + expect(Payload.hash(await downloaded.readAsBytes()), h.payloadHash); + await stream.dispose(force: true); + }); + + test( + 'refreshing the state of a download of unknown length keeps its ' + 'position', () async { + final half = h.origin.payload.length ~/ 2; + final bodyGate = Completer(); + h.origin + ..chunkedTransferEncoding = true + ..responseBodyGate = bodyGate + ..responseBodyGateAfterBytes = half; + final stream = h.manager.createStream(h.origin.url('/chunked.mp3')); + final download = stream.download(); + await _until(() => stream.cachePosition >= half); + await _until(() => (stream.files.partial.lengthSyncOrNull() ?? 0) >= half); + + // The cache files cannot tell how far a download of unknown length got. + final state = await stream.refreshCacheState(); + expect(state.position, half); + expect(state.filePosition, half); + expect(state.sourceLength, isNull); + expect(stream.cacheState, state); + + bodyGate.complete(); + final file = await download.timeout(const Duration(seconds: 15)); + expect(Payload.hash(await file.readAsBytes()), h.payloadHash); + await stream.dispose(force: true); + }); +} diff --git a/test/e2e/partial_deleted_test.dart b/test/e2e/partial_deleted_test.dart deleted file mode 100644 index d5b9c74..0000000 --- a/test/e2e/partial_deleted_test.dart +++ /dev/null @@ -1,70 +0,0 @@ -import 'dart:async'; -import 'dart:typed_data'; - -import 'package:flutter_test/flutter_test.dart'; -import 'package:http_cache_stream/http_cache_stream.dart'; -import 'package:http_cache_stream/src/etc/extensions/file_extensions.dart'; - -import '../support/harness.dart'; -import '../support/payload.dart'; - -/// The partial cache file deleted while it downloads (by the device's cache -/// management, say): the download writes on into the deleted file, where no -/// response can open what it holds. A request resets the cache, and its range -/// downloads again. -void main() { - late CacheTestHarness h; - - setUp(() async { - h = CacheTestHarness(); - await h.setUp(); - }); - - tearDown(() => h.tearDown()); - - Future read(HttpCacheStream stream, int start, int end) async { - final response = await stream.request(start: start, end: end); - final builder = BytesBuilder(copy: false); - await for (final chunk in response.stream) { - builder.add(chunk); - } - return builder.takeBytes(); - } - - test('a range it had downloaded is read again after the file was deleted', - () async { - final half = h.origin.payload.length ~/ 2; - final bodyGate = Completer(); - h.origin - ..responseBodyGate = bodyGate - ..responseBodyGateAfterBytes = half; - final stream = h.manager.createStream(h.origin.url('/deleted.mp3')); - final reachedHalf = Completer(); - final errors = []; - final subscription = stream.cacheStateStream.listen((state) { - if (state.position >= half && !reachedHalf.isCompleted) { - reachedHalf.complete(); - } - }, onError: errors.add); - final download = stream.download(); - await reachedHalf.future.timeout(const Duration(seconds: 5)); - final partial = stream.files.partial; - for (var i = 0; i < 100 && (partial.lengthSyncOrNull() ?? 0) < half; i++) { - await Future.delayed(const Duration(milliseconds: 10)); - } - await partial.delete(); - - final bytes = - await read(stream, 1000, 9000).timeout(const Duration(seconds: 10)); - expect(bytes, Uint8List.sublistView(h.origin.payload, 1000, 9000)); - expect(errors.whereType().map((e) => e.size), - contains(-1), - reason: 'the deletion is reported'); - - bodyGate.complete(); - final file = await download.timeout(const Duration(seconds: 15)); - expect(Payload.hash(await file.readAsBytes()), h.payloadHash); - await subscription.cancel(); - await stream.dispose(force: true); - }); -} From e5a6b1875d325bfade983079528e509be5e0a651 Mon Sep 17 00:00:00 2001 From: Colton Date: Mon, 9 Nov 2026 18:33:14 -0500 Subject: [PATCH 4/8] Start over a resumed download whose partial file changed before it wrote A download resuming a partial cache file opens it to append when it first saves. Deleted since (appending creates it again), truncated or written to, the file took the download's bytes at the wrong offset, and the cache resumed from it later held them as the content: a complete cache file of the right size with the wrong bytes. The sink checks the file ends where it resumes it, and a file that does not is an invalid cache, which resets. --- .../cache_downloader/buffered_io_sink.dart | 31 +++++++++- .../cache_downloader/cache_downloader.dart | 16 +++-- test/e2e/cache_files_changed_test.dart | 59 +++++++++++++++++++ test/io/buffered_io_sink_test.dart | 27 +++++++++ 4 files changed, 126 insertions(+), 7 deletions(-) diff --git a/lib/src/cache_stream/cache_downloader/buffered_io_sink.dart b/lib/src/cache_stream/cache_downloader/buffered_io_sink.dart index 456562e..769f7c9 100644 --- a/lib/src/cache_stream/cache_downloader/buffered_io_sink.dart +++ b/lib/src/cache_stream/cache_downloader/buffered_io_sink.dart @@ -7,6 +7,24 @@ import '../response_streams/partial_cache_feed.dart'; part 'buffered_io_sink_feed.dart'; +/// Thrown when the file a [BufferedIOSink] resumes does not end where the sink resumes it: it was deleted, +/// truncated or written to since the sink's earlier bytes were saved. +class SinkFileLengthException implements Exception { + final String path; + + /// The length of the file. + final int length; + + /// Where the sink resumes the file. + final int expected; + + const SinkFileLengthException(this.path, this.length, this.expected); + + @override + String toString() => + 'SinkFileLengthException: $path is $length bytes long, but the sink resumes it at $expected'; +} + /// An IO sink that supports adding data while flushing to disk asynchronously. class BufferedIOSink { //Maximum number of bytes to write in a single write operation. This prevents long writes from stalling position waiters. @@ -50,7 +68,18 @@ class BufferedIOSink { fileMode = FileMode.write; //Overwrite existing } - raf = _openedRAF = await file.open(mode: fileMode); + raf = await file.open(mode: fileMode); + if (fileMode == FileMode.append) { + //Appended at its end, the file must end where the sink resumes it. + //Deleted since (append creates it again), truncated or written to, + //it would take the sink's bytes at the wrong offset. + final length = await raf.length(); + if (length != flushedBytes) { + await raf.close(); + throw SinkFileLengthException(file.path, length, flushedBytes); + } + } + _openedRAF = raf; } while (_buffer.isNotEmpty) { diff --git a/lib/src/cache_stream/cache_downloader/cache_downloader.dart b/lib/src/cache_stream/cache_downloader/cache_downloader.dart index 3351dff..4cbe72c 100644 --- a/lib/src/cache_stream/cache_downloader/cache_downloader.dart +++ b/lib/src/cache_stream/cache_downloader/cache_downloader.dart @@ -101,14 +101,10 @@ class CacheDownloader { (_) { _downloader.resume(); }, - onError: (e) { - cancel(e); - }, + onError: _cancelOnFlushError, ); } else if (!_sink.isFlushing) { - _sink.flush().catchError((e) { - cancel(e); - }); + _sink.flush().catchError(_cancelOnFlushError); } }, ); @@ -151,6 +147,14 @@ class CacheDownloader { } } + ///Cancels the download on a failed save. A partial cache file that does not end where the download + ///resumes it (deleted, truncated or written to since) is an invalid cache: the download starts over. + void _cancelOnFlushError(final Object e) { + cancel(e is SinkFileLengthException + ? InvalidCacheSizeException(sourceUrl, e.length, e.expected) + : e); + } + /// Cancels the download and closes the stream. An optional [error] can be provided to indicate the reason for cancellation. /// An [InvalidCacheException] ends [download] on it, and the download serves no more requests. Future cancel([Object? exception]) async { diff --git a/test/e2e/cache_files_changed_test.dart b/test/e2e/cache_files_changed_test.dart index 6f8cbb3..25edcb4 100644 --- a/test/e2e/cache_files_changed_test.dart +++ b/test/e2e/cache_files_changed_test.dart @@ -47,6 +47,24 @@ void main() { return sizes; } + /// Leaves a partial cache of [source]'s first [bytes], as an interrupted + /// download does. + Future interruptAt(Uri source, int bytes) async { + final gate = Completer(); + h.origin + ..responseBodyGate = gate + ..responseBodyGateAfterBytes = bytes; + final stream = h.manager.createStream(source); + stream.download().ignore(); + await _until(() => stream.cachePosition >= bytes); + await stream.dispose(force: true); + h.origin + ..responseBodyGate = null + ..responseBodyGateAfterBytes = null; + gate.complete(); + expect(stream.files.partial.lengthSyncOrNull(), bytes); + } + test( 'a range it had downloaded is read again after the partial file was ' 'deleted', () async { @@ -76,6 +94,47 @@ void main() { await stream.dispose(force: true); }); + test( + 'a resumed download whose partial file was deleted before it wrote ' + 'saves no bytes at the wrong offset', () async { + final source = h.origin.url('/resume-deleted.mp3'); + final half = h.origin.payload.length ~/ 2; + final quarter = h.origin.payload.length ~/ 4; + await interruptAt(source, half); + + // The resumed download waits for the origin while its partial file is + // deleted, then saves a quarter and stops, before any request or refresh + // could see the file change. + final startGate = Completer(); + final bodyGate = Completer(); + h.origin + ..responseStartGate = startGate + ..responseBodyGate = bodyGate + ..responseBodyGateAfterBytes = quarter; + final requestsBefore = h.origin.requestCount; + final resumed = h.manager.createStream(source); + final invalidSizes = listenForInvalidSizes(resumed); + resumed.download().ignore(); + await _until(() => h.origin.requestCount > requestsBefore); + await resumed.files.partial.delete(); + startGate.complete(); + await _until( + () => (resumed.files.partial.lengthSyncOrNull() ?? 0) >= quarter); + await resumed.dispose(force: true); + expect(invalidSizes, contains(0), reason: 'the deletion is reported'); + h.origin + ..responseStartGate = null + ..responseBodyGate = null + ..responseBodyGateAfterBytes = null; + bodyGate.complete(); + + // What it saved is resumed from, and must be the content's start. + final next = h.manager.createStream(source); + final file = await next.download().timeout(const Duration(seconds: 15)); + expect(Payload.hash(await file.readAsBytes()), h.payloadHash); + await next.dispose(force: true); + }); + test('a complete cache file of the wrong size is downloaded again', () async { final stream = h.manager.createStream(h.origin.url('/truncated.mp3')); final invalidSizes = listenForInvalidSizes(stream); diff --git a/test/io/buffered_io_sink_test.dart b/test/io/buffered_io_sink_test.dart index 83e5a54..49411f9 100644 --- a/test/io/buffered_io_sink_test.dart +++ b/test/io/buffered_io_sink_test.dart @@ -71,6 +71,33 @@ void main() { expect(Payload.hash(written), Payload.hash(expected)); }); + test('resuming a file that does not end where the sink resumes it fails', + () async { + final saved = Payload.generate(50 * 1024, seed: 1); + for (final changed in [ + null, // deleted + Uint8List.sublistView(saved, 0, 1024), // truncated + Uint8List.fromList([...saved, 1, 2, 3]), // written to + ]) { + final file = tmp('changed.bin'); + if (changed != null) await file.writeAsBytes(changed); + if (changed == null && file.existsSync()) await file.delete(); + + final sink = BufferedIOSink(file, saved.length); + sink.add(Payload.generate(1024, seed: 2)); + await expectLater( + sink.flush(), + throwsA(isA() + .having((e) => e.length, 'length', changed?.length ?? 0) + .having((e) => e.expected, 'expected', saved.length)), + ); + expect(sink.flushedBytes, saved.length); + expect(await file.length(), changed?.length ?? 0, + reason: 'nothing is written to it'); + await sink.close().catchError((_) {}); + } + }); + test('waitForPosition completes once the target is flushed', () async { final data = Payload.generate(10 * 1024); final sink = BufferedIOSink(tmp('wait.bin'), 0); From aca897b02b243bbd7c664d53ae3754c64a1dcee7 Mon Sep 17 00:00:00 2001 From: Colton Date: Mon, 9 Nov 2026 18:33:37 -0500 Subject: [PATCH 5/8] CHANGELOG: revalidated cache files, filePosition and the reset fixes --- CHANGELOG.md | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3051048..828ca88 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,16 @@ ## Unreleased +### New Features + +* Added `CacheState.filePosition`, the bytes saved in the cache file. While a download runs it can be behind `position`, which counts the bytes downloaded. `CacheState.incomplete` takes it as an optional `filePosition`, which defaults to `position`. + ### Fixes -* Fixed a partial cache file deleted mid-download (by the device's cache management, say) failing every request for what it held until the download ended: the download wrote on into the deleted file. A request now resets the cache, and waits for its range to download again. +* Every request now revalidates the cache files against the running download, and resets a cache they no longer hold (deleted or changed underneath, by the device's cache management, say). The request waits for its range to download again. +* Fixed a partial cache file deleted mid-download failing every request for what it held until the download ended: the download wrote on into the deleted file. +* Fixed a resumed download whose partial cache file was deleted, truncated or written to before the download first saved saving its bytes at the wrong offset, which could leave a complete cache file of the right size with the wrong content. The download now starts over. +* Fixed `resetCache()`, and the reset of an invalid cache, being ignored while the download waited for a response or to retry: it resumed the old partial cache. +* Fixed `refreshCacheState()` during a download of unknown length (a chunked response) dropping the position to 0. While a download runs, the state takes its position. ## 0.3.1 From a645a936ed77509dec4ed522e55c3e08e7509948 Mon Sep 17 00:00:00 2001 From: Colton Date: Mon, 9 Nov 2026 18:43:40 -0500 Subject: [PATCH 6/8] Cancel the download a refresh found invalid before releasing the lock The reset came after the lock was released, so a refresh waiting for the lock could judge the same download again: two requests revalidating a deleted partial file reset the cache twice, and the late reset could delete the restarted download's file. The refresh that finds the cache invalid now cancels the download under the lock, so later refreshes see none to judge, and no request is served from it. --- lib/src/cache_stream/http_cache_stream.dart | 12 +++- test/e2e/cache_files_changed_test.dart | 64 ++++++++++++++++----- 2 files changed, 62 insertions(+), 14 deletions(-) diff --git a/lib/src/cache_stream/http_cache_stream.dart b/lib/src/cache_stream/http_cache_stream.dart index 79ff15d..1cad9f2 100644 --- a/lib/src/cache_stream/http_cache_stream.dart +++ b/lib/src/cache_stream/http_cache_stream.dart @@ -427,7 +427,17 @@ class HttpCacheStream { /// [CacheState.zero]. Future refreshCacheState() async { try { - final state = await _fileLock.synchronized(_cacheFileState); + final state = await _fileLock.synchronized(() async { + try { + return await _cacheFileState(); + } on InvalidCacheException catch (e) { + //Cancelled before the lock is released: a refresh waiting for it + //does not judge the download again (and reset its successor), and no + //request is served from it. + _activeDownloader?.cancel(e).ignore(); + rethrow; + } + }); _updateCacheState(state); return state; } on InvalidCacheException catch (e) { diff --git a/test/e2e/cache_files_changed_test.dart b/test/e2e/cache_files_changed_test.dart index 25edcb4..e91a792 100644 --- a/test/e2e/cache_files_changed_test.dart +++ b/test/e2e/cache_files_changed_test.dart @@ -38,15 +38,17 @@ void main() { return builder.takeBytes(); } - /// The [InvalidCacheSizeException]s [stream] reports, as their sizes. - List listenForInvalidSizes(HttpCacheStream stream) { - final sizes = []; - stream.cacheStateStream.listen(null, onError: (Object e) { - if (e is InvalidCacheSizeException) sizes.add(e.size); - }); - return sizes; + /// The errors [stream] reports. + List listenForErrors(HttpCacheStream stream) { + final errors = []; + stream.cacheStateStream.listen(null, onError: errors.add); + return errors; } + /// The [InvalidCacheSizeException]s in [errors], as their sizes. + List invalidSizes(List errors) => + [for (final e in errors.whereType()) e.size]; + /// Leaves a partial cache of [source]'s first [bytes], as an interrupted /// download does. Future interruptAt(Uri source, int bytes) async { @@ -74,7 +76,7 @@ void main() { ..responseBodyGate = bodyGate ..responseBodyGateAfterBytes = half; final stream = h.manager.createStream(h.origin.url('/deleted.mp3')); - final invalidSizes = listenForInvalidSizes(stream); + final errors = listenForErrors(stream); final download = stream.download(); await _until(() => stream.cachePosition >= half); final partial = stream.files.partial; @@ -86,7 +88,8 @@ void main() { final bytes = await read(stream, 1000, 9000).timeout(const Duration(seconds: 10)); expect(bytes, Uint8List.sublistView(h.origin.payload, 1000, 9000)); - expect(invalidSizes, contains(-1), reason: 'the deletion is reported'); + expect(invalidSizes(errors), contains(-1), + reason: 'the deletion is reported'); bodyGate.complete(); final file = await download.timeout(const Duration(seconds: 15)); @@ -113,7 +116,7 @@ void main() { ..responseBodyGateAfterBytes = quarter; final requestsBefore = h.origin.requestCount; final resumed = h.manager.createStream(source); - final invalidSizes = listenForInvalidSizes(resumed); + final errors = listenForErrors(resumed); resumed.download().ignore(); await _until(() => h.origin.requestCount > requestsBefore); await resumed.files.partial.delete(); @@ -121,7 +124,8 @@ void main() { await _until( () => (resumed.files.partial.lengthSyncOrNull() ?? 0) >= quarter); await resumed.dispose(force: true); - expect(invalidSizes, contains(0), reason: 'the deletion is reported'); + expect(invalidSizes(errors), contains(0), + reason: 'the deletion is reported'); h.origin ..responseStartGate = null ..responseBodyGate = null @@ -135,16 +139,50 @@ void main() { await next.dispose(force: true); }); + test('concurrent requests after the partial file was deleted reset it once', + () async { + final half = h.origin.payload.length ~/ 2; + final bodyGate = Completer(); + h.origin + ..responseBodyGate = bodyGate + ..responseBodyGateAfterBytes = half; + final stream = h.manager.createStream(h.origin.url('/deleted-twice.mp3')); + final errors = listenForErrors(stream); + final download = stream.download(); + await _until(() => stream.cachePosition >= half); + final partial = stream.files.partial; + await _until(() => (partial.lengthSyncOrNull() ?? 0) >= half); + final requestsBefore = h.origin.requestCount; + await partial.delete(); + + // Each request revalidates; one reset is due, not one for each. + final ranges = [(1000, 9000), (20000, 30000), (50000, 60000)]; + final reads = await Future.wait([ + for (final (start, end) in ranges) read(stream, start, end), + ]).timeout(const Duration(seconds: 10)); + for (final (i, (start, end)) in ranges.indexed) { + expect(reads[i], Uint8List.sublistView(h.origin.payload, start, end)); + } + expect(invalidSizes(errors), [-1]); + expect(h.origin.requestCount, requestsBefore + 1, + reason: 'the download starts over once'); + + bodyGate.complete(); + final file = await download.timeout(const Duration(seconds: 15)); + expect(Payload.hash(await file.readAsBytes()), h.payloadHash); + await stream.dispose(force: true); + }); + test('a complete cache file of the wrong size is downloaded again', () async { final stream = h.manager.createStream(h.origin.url('/truncated.mp3')); - final invalidSizes = listenForInvalidSizes(stream); + final errors = listenForErrors(stream); final file = await stream.download(); await file.writeAsBytes(Uint8List(1000)); final bytes = await read(stream, 1000, 9000).timeout(const Duration(seconds: 10)); expect(bytes, Uint8List.sublistView(h.origin.payload, 1000, 9000)); - expect(invalidSizes, contains(1000)); + expect(invalidSizes(errors), contains(1000)); final downloaded = await stream.download().timeout(const Duration(seconds: 15)); From 7c27aa3708ee04605f31b99f0defc77385052b79 Mon Sep 17 00:00:00 2001 From: Colton Date: Mon, 9 Nov 2026 18:44:31 -0500 Subject: [PATCH 7/8] End a download on an invalid cache found while it saves its last bytes A resumed download whose partial file was deleted, receiving the rest at once, finished its response before its first save found the file gone: the failed save was reported as a plain error, the download waited 5 s to retry, and a file written to since would have been resumed from. The download now ends on the invalid cache found by a save after its response, or by a cancel while it saves its last bytes, as it does during a response, and resets at once. A download counts as writing until its sink is closed and the file released, not from when the sink starts closing: a refresh in between neither renames a file still open (which fails on Windows) nor resets the cache under the download, which then completed with no headers. While writing, it ends on the reset's exception instead. Once it stopped writing, the state is read from the files. --- .../cache_downloader/cache_downloader.dart | 56 +++++++++++++------ lib/src/cache_stream/http_cache_stream.dart | 11 ++-- test/e2e/cache_files_changed_test.dart | 28 ++++++++++ 3 files changed, 73 insertions(+), 22 deletions(-) diff --git a/lib/src/cache_stream/cache_downloader/cache_downloader.dart b/lib/src/cache_stream/cache_downloader/cache_downloader.dart index 4cbe72c..6df9cc0 100644 --- a/lib/src/cache_stream/cache_downloader/cache_downloader.dart +++ b/lib/src/cache_stream/cache_downloader/cache_downloader.dart @@ -28,8 +28,12 @@ class CacheDownloader { ///Headers received and validated CachedResponseHeaders? _validatedHeaders; - ///The invalid cache [cancel] stopped the download for, if any. The download ends on it, and serves no more requests. + ///The invalid cache the download stopped for, if any: [cancel]led for it, or found by a save. The download + ///ends on it, and serves no more requests. InvalidCacheException? _invalidCache; + + ///If the download may still write to the partial cache file, until its sink is closed. + bool _writing = true; CacheDownloader._( final CacheMetadata cacheMetadata, final int startPosition, @@ -108,12 +112,8 @@ class CacheDownloader { } }, ); - //Cancelled for an invalid cache between responses (waiting for one, or to retry), the download - //returns as if stopped; it ends on the exception, as when a response is cancelled for it. - if (_invalidCache case final InvalidCacheException invalidCache) { - throw invalidCache; - } - } on InvalidCacheException { + } on InvalidCacheException catch (e) { + _invalidCache ??= e; _validatedHeaders = null; rethrow; } catch (e) { @@ -123,12 +123,28 @@ class CacheDownloader { // Post-download — flush remaining data and verify cache integrity try { await _sink.close( - flushBuffer: true, + flushBuffer: !isInvalidated, //An invalid cache keeps no more bytes isDone: _downloader .isDone, //If the source did not end, the feed is marked as aborted so readers do not treat it as an end of stream ); //Flushes all buffered data and closes the sink } catch (e) { - onError(e); + final error = _saveError(e); + if (error is InvalidCacheException) { + _invalidCache ??= error; + } else { + onError(error); + } + } finally { + _writing = false; + } + + //Cancelled for an invalid cache between responses (waiting for one, or + //to retry) or while saving its last bytes, or found invalid by a save that + //failed after the response ended: the download returned as if stopped, + //and ends on it, as when a response is cancelled for it. + if (_invalidCache case final InvalidCacheException invalidCache) { + _validatedHeaders = null; + throw invalidCache; } final sourceLength = _validatedHeaders?.sourceLength ?? @@ -141,20 +157,25 @@ class CacheDownloader { ///The sink is not closed on invalid cache exception, so we need to close it here await _sink.close(flushBuffer: false).ignoreResult(); } + _writing = false; if (!_completer.isCompleted) { _completer.complete(); } } } - ///Cancels the download on a failed save. A partial cache file that does not end where the download - ///resumes it (deleted, truncated or written to since) is an invalid cache: the download starts over. + ///Cancels the download on a failed save. void _cancelOnFlushError(final Object e) { - cancel(e is SinkFileLengthException - ? InvalidCacheSizeException(sourceUrl, e.length, e.expected) - : e); + //A block body: a flush future's error handler must return null. + cancel(_saveError(e)); } + ///A partial cache file that does not end where the download resumes it (deleted, truncated or written + ///to since) is an invalid cache: the download starts over. + Object _saveError(final Object e) => e is SinkFileLengthException + ? InvalidCacheSizeException(sourceUrl, e.length, e.expected) + : e; + /// Cancels the download and closes the stream. An optional [error] can be provided to indicate the reason for cancellation. /// An [InvalidCacheException] ends [download] on it, and the download serves no more requests. Future cancel([Object? exception]) async { @@ -209,12 +230,13 @@ class CacheDownloader { int get downloadPosition => _position; int get filePosition => _sink.flushedBytes; - ///If the download may still write to the partial cache file. - bool get isWriting => !_sink.isClosed; + ///If the download may still write to the partial cache file: until its sink is closed, and the file + ///released. + bool get isWriting => _writing; Uri get sourceUrl => _downloader.sourceUrl; bool get isClosed => _completer.isCompleted; - ///If [cancel] stopped the download for an invalid cache. + ///If the download stopped for an invalid cache: [cancel]led for it, or found by a save. bool get isInvalidated => _invalidCache != null; bool get isPaused => _paused; } diff --git a/lib/src/cache_stream/http_cache_stream.dart b/lib/src/cache_stream/http_cache_stream.dart index 1cad9f2..f6df66f 100644 --- a/lib/src/cache_stream/http_cache_stream.dart +++ b/lib/src/cache_stream/http_cache_stream.dart @@ -455,6 +455,7 @@ class HttpCacheStream { assert(_fileLock.locked, '_cacheFileState must be called within _fileLock.synchronized()'); final downloader = _activeDownloader; + final writer = downloader?.isWriting == true ? downloader : null; //Read before the file is: by then the file holds at least these bytes, while //bytes saved during the stat may be missing from its size. final savedBytes = downloader?.filePosition ?? 0; @@ -467,7 +468,7 @@ class HttpCacheStream { sourceUrl, completeCacheStat.size, sourceLength); return CacheState.complete(completeCacheStat.size); } - } else if (downloader == null) { + } else if (writer == null) { return const CacheState.zero(); //Nothing to resume without the length } @@ -489,7 +490,7 @@ class HttpCacheStream { partial: true); //Renamed once the download stops writing to it - if (partialSize == sourceLength && downloader?.isWriting != true) { + if (partialSize == sourceLength && writer == null) { try { await files.partial.rename(files .complete.path); //Rename the partial cache to the complete cache @@ -511,13 +512,13 @@ class HttpCacheStream { } } - if (downloader != null) { + if (writer != null) { //The download is ahead of the file by the bytes it has yet to save. A //partial cache file it did not resume holds none of its bytes until it //saves them. return CacheState.incomplete( - downloader.downloadPosition, sourceLength ?? downloader.sourceLength, - filePosition: downloader.filePosition); + writer.downloadPosition, sourceLength ?? writer.sourceLength, + filePosition: writer.filePosition); } return CacheState.incomplete(partialSize ?? 0, sourceLength); } diff --git a/test/e2e/cache_files_changed_test.dart b/test/e2e/cache_files_changed_test.dart index e91a792..fec1990 100644 --- a/test/e2e/cache_files_changed_test.dart +++ b/test/e2e/cache_files_changed_test.dart @@ -173,6 +173,34 @@ void main() { await stream.dispose(force: true); }); + test( + 'a resumed download whose partial file was deleted, receiving the rest ' + 'at once, starts over', () async { + final source = h.origin.url('/resume-rest.mp3'); + final saved = h.origin.payload.length - 1000; + await interruptAt(source, saved); + + // The rest arrives before the download's first save finds the file gone. + final startGate = Completer(); + h.origin.responseStartGate = startGate; + final requestsBefore = h.origin.requestCount; + final resumed = h.manager.createStream(source); + final errors = listenForErrors(resumed); + final download = resumed.download(); + await _until(() => h.origin.requestCount > requestsBefore); + await resumed.files.partial.delete(); + h.origin.responseStartGate = null; + startGate.complete(); + + final file = await download.timeout(const Duration(seconds: 15)); + expect(Payload.hash(await file.readAsBytes()), h.payloadHash); + expect(invalidSizes(errors), [0], reason: 'the deletion is reported'); + expect( + errors.whereType(), hasLength(errors.length), + reason: 'as an invalid cache, which resets, not a failed save'); + await resumed.dispose(force: true); + }); + test('a complete cache file of the wrong size is downloaded again', () async { final stream = h.manager.createStream(h.origin.url('/truncated.mp3')); final errors = listenForErrors(stream); From 501764c54654ff3ac27953259cb73c5a56bfe01c Mon Sep 17 00:00:00 2001 From: Colton Date: Mon, 9 Nov 2026 19:09:47 -0500 Subject: [PATCH 8/8] CHANGELOG and version bump to 0.3.2 --- CHANGELOG.md | 6 +++--- pubspec.yaml | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 828ca88..9331e54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,14 +1,14 @@ -## Unreleased +## 0.3.2 ### New Features -* Added `CacheState.filePosition`, the bytes saved in the cache file. While a download runs it can be behind `position`, which counts the bytes downloaded. `CacheState.incomplete` takes it as an optional `filePosition`, which defaults to `position`. +* Added `CacheState.filePosition`, the bytes saved in the cache file. While a download runs it can be behind `position`, which counts the bytes downloaded. `CacheState.incomplete` takes it as an optional `filePosition`, which defaults to `position`, and incomplete states compare by it too. ### Fixes * Every request now revalidates the cache files against the running download, and resets a cache they no longer hold (deleted or changed underneath, by the device's cache management, say). The request waits for its range to download again. * Fixed a partial cache file deleted mid-download failing every request for what it held until the download ended: the download wrote on into the deleted file. -* Fixed a resumed download whose partial cache file was deleted, truncated or written to before the download first saved saving its bytes at the wrong offset, which could leave a complete cache file of the right size with the wrong content. The download now starts over. +* Fixed a resumed download saving its bytes at the wrong offset when its partial cache file was deleted, truncated or written to before the download first wrote to it, which could leave a complete cache file of the right size with the wrong content. The download now starts over. * Fixed `resetCache()`, and the reset of an invalid cache, being ignored while the download waited for a response or to retry: it resumed the old partial cache. * Fixed `refreshCacheState()` during a download of unknown length (a chunked response) dropping the position to 0. While a download runs, the state takes its position. diff --git a/pubspec.yaml b/pubspec.yaml index f881abe..8c2c11c 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -1,6 +1,6 @@ name: http_cache_stream description: "Simultaneously download, cache, and stream remote content. Perfect for media players and any plugin that streams web content." -version: 0.3.1 +version: 0.3.2 homepage: https://github.com/Colton127/http_cache_stream repository: https://github.com/Colton127/http_cache_stream topics: