0.2.0

CHANGELOG.md

@@ -1,3 +1,21 @@
+## 0.2.0
+
+This release is designed to preserve existing behavior while making caching and streaming faster and more robust.
+
+### Improvements
+
+* Improved streaming performance and efficiency when serving cached content.
+* Improved reliability when resuming partially downloaded files.
+* Improved handling of interrupted downloads, cache validation, and source changes.
+* Improved range request handling and cache integrity checks.
+* Improved cache lifecycle and cleanup behavior across platforms.
+
+### Fixes
+
+* Fixed edge cases that could cause incomplete or outdated cached data to be served.
+* Fixed several issues involving partial downloads, interrupted connections, and cache file transitions.
+* Fixed assorted range request and local cache server edge cases.
+
 ## 0.1.0
 
 This release significantly simplifies cache management. A new `getCacheUrl` API automates the full lifecycle of cache streams, eliminating the need to create or manage `HttpCacheStream` instances for most integrations.

benchmarker/README.md

@@ -145,8 +145,4 @@ flutter test
 
 The suite covers request distribution and statistics, drives real worker
 isolates against a local origin server, and runs all three benchmark types
-end-to-end through a real `HttpCacheManager`.
-
-> When testing against a **loopback** origin, address it as `localhost` rather
-> than `127.0.0.1`: `http_cache_stream` rejects source URLs whose host matches
-> the cache server's own host.
+end-to-end through a real `HttpCacheManager`.

benchmarker/lib/src/benchmark/benchmark_config.dart

@@ -194,8 +194,7 @@ class RangePlan {
       );
 
   @override
-  String toString() =>
-      'RangePlan(bytes $start-$end, window $windowSize, '
+  String toString() => 'RangePlan(bytes $start-$end, window $windowSize, '
       'sequential: $isSequential)';
 }
 
@@ -248,7 +247,10 @@ class BenchmarkConfig {
     required int? totalRequests,
   }) {
     final uri = Uri.tryParse(url.trim());
-    if (url.trim().isEmpty || uri == null || !uri.hasScheme || uri.host.isEmpty) {
+    if (url.trim().isEmpty ||
+        uri == null ||
+        !uri.hasScheme ||
+        uri.host.isEmpty) {
       return 'Enter a valid absolute source URL.';
     }
     if (uri.scheme != 'http' && uri.scheme != 'https') {

benchmarker/lib/src/benchmark/source_probe.dart

@@ -28,7 +28,8 @@ Future<SourceInfo> probeSource(
   try {
     try {
       final response = await httpClient.head(url).timeout(timeout);
-      if (_isSuccess(response.statusCode) && (response.contentLength ?? 0) > 0) {
+      if (_isSuccess(response.statusCode) &&
+          (response.contentLength ?? 0) > 0) {
         return SourceInfo(
           contentLength: response.contentLength,
           acceptsRanges: _advertisesRanges(response.headers),

benchmarker/lib/src/ui/widgets/config_panel.dart

@@ -263,8 +263,9 @@ class _RangeSelector extends StatelessWidget {
               ],
               selected: {form.rangeMode},
               showSelectedIcon: false,
-              onSelectionChanged:
-                  isBusy ? null : (selection) => form.rangeMode = selection.first,
+              onSelectionChanged: isBusy
+                  ? null
+                  : (selection) => form.rangeMode = selection.first,
             ),
           ),
         ),

benchmarker/lib/src/ui/widgets/stats_panel.dart

@@ -351,8 +351,7 @@ class _TileGrid extends StatelessWidget {
           spacing: spacing,
           runSpacing: spacing,
           children: [
-            for (final tile in tiles)
-              SizedBox(width: tileWidth, child: tile),
+            for (final tile in tiles) SizedBox(width: tileWidth, child: tile),
           ],
         );
       },
@@ -426,10 +425,9 @@ class _TimingTable extends StatelessWidget {
         child: Text(
           text,
           textAlign: leading ? TextAlign.left : TextAlign.right,
-          style: (header
-                  ? theme.textTheme.labelSmall
-                  : theme.textTheme.bodySmall)
-              ?.copyWith(
+          style:
+              (header ? theme.textTheme.labelSmall : theme.textTheme.bodySmall)
+                  ?.copyWith(
             color: header ? theme.colorScheme.onSurfaceVariant : null,
             fontFeatures: const [FontFeature.tabularFigures()],
           ),

benchmarker/lib/src/util/formatting.dart

@@ -57,5 +57,4 @@ String formatTimestamp(DateTime time) {
       '${formatClockTime(local)}';
 }
 
-String _pad(int value, [int width = 2]) =>
-    value.toString().padLeft(width, '0');
+String _pad(int value, [int width = 2]) => value.toString().padLeft(width, '0');

benchmarker/test/benchmark_config_test.dart

@@ -31,8 +31,8 @@ void main() {
     test('always sums to the total request count', () {
       for (var concurrency = 1; concurrency <= 16; concurrency++) {
         for (var total = concurrency; total <= 200; total += 7) {
-          final distribution =
-              _config(concurrency: concurrency, total: total).requestDistribution();
+          final distribution = _config(concurrency: concurrency, total: total)
+              .requestDistribution();
           expect(distribution, hasLength(concurrency));
           expect(distribution.reduce((a, b) => a + b), total);
           expect(distribution.every((count) => count > 0), isTrue);
@@ -100,7 +100,8 @@ void main() {
 
     test('windows start where the previous one ended', () {
       const requestCount = 7;
-      final plan = RangePlan.sequential(const ByteRange(4096, 20479), requestCount);
+      final plan =
+          RangePlan.sequential(const ByteRange(4096, 20479), requestCount);
 
       // Every window but the last starts directly after its predecessor; the
       // last slides back to end on the final byte, so it may overlap.
@@ -138,7 +139,8 @@ void main() {
 
     test('the windows cover the whole selected range', () {
       for (final requestCount in [1, 2, 3, 7, 16, 100]) {
-        final plan = RangePlan.sequential(const ByteRange(500, 1499), requestCount);
+        final plan =
+            RangePlan.sequential(const ByteRange(500, 1499), requestCount);
         expect(plan.windowFor(0).start, 500);
         expect(plan.windowFor(requestCount - 1).end, 1499);
       }

benchmarker/test/benchmark_controller_test.dart

@@ -58,10 +58,7 @@ void main() {
         await request.response.close();
       }
     }());
-    // Addressed as `localhost` rather than the bound `127.0.0.1` so the source
-    // host differs from the cache server's host; otherwise http_cache_stream
-    // treats the source URL as an already-encoded cache URL.
-    sourceUrl = Uri.parse('http://localhost:${origin.port}/payload.bin');
+    sourceUrl = Uri.parse('http://127.0.0.1:${origin.port}/payload.bin');
 
     cacheDir = await Directory.systemTemp.createTemp('benchmarker_test');
     await HttpCacheManager.init(
@@ -213,10 +210,12 @@ void main() {
     // Every window is the same size and together they cover the payload once.
     expect(stats.totalBytes, payload.length);
     expect(stats.avgBytesPerRequest, plan.windowSize.toDouble());
-    expect(receivedRanges..sort(), [
-      for (var sequence = 0; sequence < 8; sequence++)
-        plan.windowFor(sequence).header,
-    ]..sort());
+    expect(
+        receivedRanges..sort(),
+        [
+          for (var sequence = 0; sequence < 8; sequence++)
+            plan.windowFor(sequence).header,
+        ]..sort());
     expect(
       controller.logs.map((entry) => entry.message),
       contains(contains('Sequential windows: 8 ×')),

benchmarker/test/benchmark_report_test.dart

@@ -108,14 +108,16 @@ void main() {
 
       expect(json['run_id'], 3);
       expect(json['source_url'], 'https://example.com/file.bin');
-      expect(json['target_url'], 'http://127.0.0.1:4612/https/example.com/f.bin');
+      expect(
+          json['target_url'], 'http://127.0.0.1:4612/https/example.com/f.bin');
       expect(json['cache_type'], 'preCached');
       expect(json['cache_type_label'], 'Pre-cached');
       expect(json['status'], 'Finished');
       expect(json['build_mode'], 'release');
       expect(json['started_at'], _startedAt.toIso8601String());
       expect(json['ended_at'], _endedAt.toIso8601String());
-      expect(json['wall_duration_us'], const Duration(seconds: 5).inMicroseconds);
+      expect(
+          json['wall_duration_us'], const Duration(seconds: 5).inMicroseconds);
       expect(json['concurrency'], 2);
       expect(json['http_client'], kHttpClientOptions.first.label);
 
@@ -155,8 +157,8 @@ void main() {
     });
 
     test('renders without a config', () {
-      final json = jsonDecode(buildJsonReport(_result()))
-          as Map<String, Object?>;
+      final json =
+          jsonDecode(buildJsonReport(_result())) as Map<String, Object?>;
 
       expect(json['source_url'], isNull);
       expect((json['requests']! as Map<String, Object?>)['completed'], 4);

benchmarker/test/config_panel_test.dart

@@ -45,7 +45,8 @@ void main() {
       (tester) async {
     await pumpPanel(tester);
 
-    RangeSlider slider() => tester.widget<RangeSlider>(find.byType(RangeSlider));
+    RangeSlider slider() =>
+        tester.widget<RangeSlider>(find.byType(RangeSlider));
     ButtonSegment<RangeMode> segment(RangeMode mode) => tester
         .widget<SegmentedButton<RangeMode>>(
           find.byType(SegmentedButton<RangeMode>),

lib/http_cache_stream.dart

@@ -25,6 +25,7 @@ export 'src/models/cache_files/cache_files.dart';
 export 'src/models/cache_state/cache_state.dart';
 export 'src/models/exceptions/http_exceptions.dart';
 export 'src/models/exceptions/invalid_cache_exceptions.dart';
+export 'src/models/exceptions/partial_cache_feed_exceptions.dart';
 export 'src/models/exceptions/state_errors.dart';
 export 'src/models/exceptions/stream_response_exceptions.dart';
 export 'src/models/http_range/http_range.dart';

lib/src/cache_manager/http_cache_manager.dart

@@ -72,7 +72,9 @@ class HttpCacheManager {
 
     ///Remove when stream is disposed
     cacheStream.future.onComplete(() {
-      _streams.remove(requestKey);
+      if (identical(_streams[requestKey], cacheStream)) {
+        _streams.remove(requestKey);
+      }
     });
 
     if (_onStreamCreated case final streamCreatedCallback?) {

lib/src/cache_server/keep_alive_server.dart

@@ -67,7 +67,6 @@ class KeepAliveServer {
         if (_closed) return;
 
         final prevServer = _server;
-        _serverSubscription?.cancel();
 
         _server = await HttpServer.bind(address, port, shared: true);
         _forwardEvents(_server);

lib/src/cache_server/local_cache_server.dart

@@ -80,11 +80,17 @@ class LocalCacheServer {
 
   Uri encodeSourceUrl(Uri sourceUrl) {
     if (sourceUrl.host == serverUri.host) {
-      if (!validateCacheUrl(sourceUrl)) {
-        throw ArgumentError(
-            'Invalid source URL: $sourceUrl. The host matches the cache server host but the URL is not a valid cache URL.');
+      if (validateCacheUrl(sourceUrl)) {
+        return sourceUrl; // Already encoded for this server.
+      }
+      // A cache server may be assigned a different port between runs. Decode a
+      // URL produced by an earlier instance before encoding it for this one.
+      // Requiring the cache server's scheme and a different port lets regular
+      // source URLs hosted by another local server pass through unchanged.
+      if (sourceUrl.scheme == serverUri.scheme &&
+          sourceUrl.port != serverUri.port) {
+        sourceUrl = decodeSourceUrl(sourceUrl) ?? sourceUrl;
       }
-      return sourceUrl; //Already encoded
     }
 
     final defaultPort = switch (sourceUrl.scheme) {

lib/src/cache_stream/cache_downloader/buffered_io_sink.dart

@@ -2,17 +2,28 @@ import 'dart:async';
 import 'dart:io';
 import 'dart:typed_data';
 
+import '../../models/exceptions/partial_cache_feed_exceptions.dart';
+import '../response_streams/partial_cache_feed.dart';
+
+part 'buffered_io_sink_feed.dart';
+
 /// 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.
+  static const int _maxWriteSize = 256 * 1024; // 256 KB
+
   final File file;
   BufferedIOSink(this.file, int initialPosition)
-      : _flushedBytes = initialPosition;
+      : _flushedBytes = initialPosition {
+    _feed = BufferedIOSinkFeed._(this);
+  }
+
   int _flushedBytes;
   final _buffer = BytesBuilder(copy: false);
   RandomAccessFile? _openedRAF;
   bool _isClosed = false;
   Future<void>? _flushFuture;
-  final List<({int position, Completer<void> completer})> _positionWaiters = [];
+  late final BufferedIOSinkFeed _feed;
 
   void add(List<int> data) {
     if (_isClosed) {
@@ -44,56 +55,40 @@ class BufferedIOSink {
 
         while (_buffer.isNotEmpty) {
           final bytes = _buffer.takeBytes();
-          await raf.writeFrom(bytes, 0, bytes.length);
-          _flushedBytes += bytes.length;
-          _notifyPositionWaiters();
+          for (int start = 0; start < bytes.length; start += _maxWriteSize) {
+            final int uncappedEnd = start + _maxWriteSize;
+            final int end =
+                uncappedEnd < bytes.length ? uncappedEnd : bytes.length;
+            await raf.writeFrom(bytes, start, end);
+            _flushedBytes += end - start;
+            _feed._notifyPositionWaiters();
+          }
         }
         _flushFuture = null;
       } catch (e) {
-        _failPositionWaiters(e);
+        _feed._failPositionWaiters(e);
         rethrow;
       }
     }();
   }
 
-  /// Returns a [Future] that completes once [flushedBytes] reaches or exceeds [minFlushedBytes].
+  /// Returns a [PositionWaiter] that completes once [flushedBytes] reaches or exceeds [minFlushedBytes].
   /// Completes immediately if the position is already reached.
-  /// Fails if the sink is closed or a flush error occurs before the position is reached.
-  Future<void> waitForPosition(int minFlushedBytes,
-      [Duration timeout = const Duration(seconds: 30)]) {
-    if (_flushedBytes >= minFlushedBytes) return Future.value();
-    if (_isClosed) {
-      return Future.error(StateError(
-          'BufferedIOSink closed before reaching position $minFlushedBytes'));
-    }
-    final completer = Completer<void>();
-    _positionWaiters.add((position: minFlushedBytes, completer: completer));
-    return completer.future.timeout(timeout, onTimeout: () {
-      _positionWaiters.removeWhere((w) => w.completer == completer);
-      throw TimeoutException(
-          'Timeout while waiting for flushedBytes to reach $minFlushedBytes',
-          timeout);
-    });
-  }
+  /// Fails if the sink is closed, a flush error occurs before the position is reached, or the waiter is cancelled.
+  PositionWaiter waitForPosition(int minFlushedBytes) =>
+      _feed.waitForPosition(minFlushedBytes);
 
-  void _notifyPositionWaiters() {
-    if (_positionWaiters.isEmpty) return;
-    for (int i = _positionWaiters.length - 1; i >= 0; i--) {
-      if (_flushedBytes >= _positionWaiters[i].position) {
-        _positionWaiters.removeAt(i).completer.complete();
-      }
-    }
-  }
-
-  void _failPositionWaiters(Object error) {
-    if (_positionWaiters.isEmpty) return;
-    for (final w in _positionWaiters) {
-      w.completer.completeError(error);
-    }
-    _positionWaiters.clear();
-  }
-
-  Future<void> close({final bool flushBuffer = true}) async {
+  /// Closes the sink, resolving any waiters that can no longer be satisfied.
+  ///
+  /// Set [isDone] when the producer reached the end of its content. The feed is
+  /// then left unfailed, so readers treat [flushedBytes] as the true end of the
+  /// content. When [isDone] is false the download was aborted, and the feed
+  /// fails with [PartialCacheAbortedException] so readers do not mistake the
+  /// truncated content for an end of stream.
+  Future<void> close({
+    final bool flushBuffer = true,
+    final bool isDone = false,
+  }) async {
     if (_isClosed) return;
     _isClosed = true;
 
@@ -103,17 +98,23 @@ class BufferedIOSink {
       }
       await flush(); //Even if !flushBuffer, ongoing flush must complete before RAF can be closed
     } finally {
-      _failPositionWaiters(StateError('BufferedIOSink closed'));
       _buffer.clear();
-      if (_openedRAF case final RandomAccessFile raf) {
-        _openedRAF = null;
-        await raf.close();
+      try {
+        if (_openedRAF case final RandomAccessFile raf) {
+          _openedRAF = null;
+          await raf.close();
+        }
+      } finally {
+        _feed._close(
+          failure: isDone ? null : PartialCacheAbortedException(_flushedBytes),
+        );
       }
     }
   }
 
   int get bufferSize => _buffer.length;
   int get flushedBytes => _flushedBytes;
+  PartialCacheFeed get feed => _feed;
   bool get flushed => _buffer.isEmpty && !isFlushing;
   bool get isFlushing => _flushFuture != null;
   bool get isClosed => _isClosed;