diff --git a/CHANGELOG.md b/CHANGELOG.md index 3051048..9331e54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,16 @@ -## 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`, and incomplete states compare by it too. ### 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 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. ## 0.3.1 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 e03ec73..6df9cc0 100644 --- a/lib/src/cache_stream/cache_downloader/cache_downloader.dart +++ b/lib/src/cache_stream/cache_downloader/cache_downloader.dart @@ -27,6 +27,13 @@ class CacheDownloader { ///Headers received and validated CachedResponseHeaders? _validatedHeaders; + + ///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, @@ -98,18 +105,15 @@ class CacheDownloader { (_) { _downloader.resume(); }, - onError: (e) { - cancel(e); - }, + onError: _cancelOnFlushError, ); } else if (!_sink.isFlushing) { - _sink.flush().catchError((e) { - cancel(e); - }); + _sink.flush().catchError(_cancelOnFlushError); } }, ); - } on InvalidCacheException { + } on InvalidCacheException catch (e) { + _invalidCache ??= e; _validatedHeaders = null; rethrow; } catch (e) { @@ -119,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 ?? @@ -137,14 +157,31 @@ 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. + void _cancelOnFlushError(final Object 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 { + if (exception is InvalidCacheException) { + _invalidCache ??= exception; + } _downloader.close(exception); _paused = false; return _completer.future; @@ -164,6 +201,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; @@ -191,7 +229,14 @@ 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: until its sink is closed, and the file + ///released. + bool get isWriting => _writing; Uri get sourceUrl => _downloader.sourceUrl; bool get isClosed => _completer.isCompleted; + + ///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 36f4d8a..f6df66f 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,106 @@ 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(() 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) { + 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; + 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; 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 (writer == 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 && writer == null) { + 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 (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( + writer.downloadPosition, sourceLength ?? writer.sourceLength, + filePosition: writer.filePosition); } - - return CacheState.incomplete(0, sourceLength); + return CacheState.incomplete(partialSize ?? 0, sourceLength); } void _updateCacheState(final CacheState cacheState) { @@ -571,7 +584,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/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/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: diff --git a/test/e2e/cache_files_changed_test.dart b/test/e2e/cache_files_changed_test.dart new file mode 100644 index 0000000..fec1990 --- /dev/null +++ b/test/e2e/cache_files_changed_test.dart @@ -0,0 +1,247 @@ +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 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 { + 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 { + 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 errors = listenForErrors(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(errors), 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 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 errors = listenForErrors(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(errors), 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('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 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); + 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(errors), 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); - }); -} 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); + }); +} 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); 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', () {