From 3e89fd48909431a3780158172ee9a9e6ecdab3be Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 15:54:54 +0200 Subject: [PATCH 01/10] feat: add Buffer, a strategy that batches calls instead of dropping them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Debounce` and `Throttle` keep only the last call's arguments and discard the rest. `Buffer` keeps every item and invokes the wrapped function once with all of them, so a burst of per-item calls collapses into one batch rather than one survivor. It flushes `wait` after the first item lands — measured from the buffer opening rather than as a quiet period, so a steady producer still drains on schedule without needing a second duration to bound it — or as soon as it holds `maxSize` items. Only one flush runs at a time, which keeps a slow or retrying `onFlush` from stacking up concurrent requests, and makes `await flush()` in `dispose` a real drain. `maxQueueSize` then caps the backlog that builds up behind a slow flush, shedding the excess per `OverflowPolicy` and reporting it to `onDrop`; it is unbounded by default, so nothing is dropped unless asked for. Whichever call starts a flush owns its failure: one started by `flush()` lands on the returned future, and one the buffer scheduled itself goes to `onError` along with the items it was carrying, so they can be re-queued. Comes with `buffered()` on functions taking a `List` and a top-level `buffer()` lambda, matching how `Debounce` and `Throttle` are exposed. Closes #6. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 6 + README.md | 91 +++ example/lib/main.dart | 18 + lib/rate_limiter.dart | 1 + lib/src/buffer.dart | 312 +++++++++ lib/src/extension.dart | 46 ++ test/buffer_backoff_test.dart | 228 +++++++ test/buffer_test.dart | 1211 +++++++++++++++++++++++++++++++++ 8 files changed, 1913 insertions(+) create mode 100644 lib/src/buffer.dart create mode 100644 test/buffer_backoff_test.dart create mode 100644 test/buffer_test.dart diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a86d55..12c98c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +## Unreleased + +* Added `Buffer`, which collects the items passed to it and invokes your + function once with all of them, instead of once per item. +* Added `buffered()` on functions taking a `List`, and a top-level `buffer()`. + ## [1.1.1] - (14-08-2026) * Relaxed the `clock` constraint to `^1.1.1` so packages that depend on diff --git a/README.md b/README.md index bd918f0..9c3240e 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ _Rate limiting_ is a strategy for limiting an action. It puts a cap on how often - [Debounce](#debounce) - [Throttle](#throttle) - [BackOff](#backoff) + - [Buffer](#buffer) - [Pending](#pending) - [Flush](#flush) - [Cancellation](#cancellation) @@ -166,11 +167,96 @@ final response = backOff( ); ``` +### Buffer +A _buffered function_ collects the items passed to it and invokes your function **once** with all of them, rather than once per item. Where debounce and throttle keep only the last call's arguments and drop the rest, a buffer keeps every one — so it batches work instead of shedding it. + +The buffer is flushed once `wait` has passed since the first item landed in it, or as soon as it holds `maxSize` items, whichever comes first. Only one flush runs at a time: while your function is working, arriving items collect for the next one, which goes out the moment the current one finishes. By default nothing is dropped and no caller is ever slowed down — pass `maxQueueSize` to cap the buffer and shed the excess instead. + +#### Usage +1. Creating from scratch +```dart +final markRead = buffer((ids) { + print('Marking ${ids.length} messages read'); + return api.markAllRead(ids); +}, const Duration(milliseconds: 500), maxSize: 25); +``` +2. Converting an existing function into buffered function +```dart +Future markAllRead(List ids) => api.markAllRead(ids); + +final markRead = markAllRead.buffered( + const Duration(milliseconds: 500), + maxSize: 25, +); +``` + +#### Example +Marking messages read as the user scrolls calls `markRead` once per message. Debouncing it would send only the last id and lose the other nineteen; buffering sends one request carrying all twenty. +```dart +void onMessageSeen(String id) { + markRead(id); +} +``` + +Passing `Duration.zero` batches everything queued up in the current event loop turn, which is how a data loader collapses a screen's worth of lookups into one request. +```dart +final loadUsers = buffer( + (ids) => api.getUsers(ids), + Duration.zero, +); +``` + +A producer faster than `onFlush` can grow the buffer without bound. `maxQueueSize` caps it, `overflow` picks which end goes, and `onDrop` reports what that cost. +```dart +final trackEvent = buffer( + (events) => analytics.send(events), + const Duration(seconds: 5), + maxQueueSize: 10000, + overflow: OverflowPolicy.dropOldest, // or dropNewest, to keep what is waiting + onDrop: (events) => log.warning('dropped ${events.length} events'), +); +``` + +The two caps work on different things and compose: `maxSize` limits what any one flush carries, `maxQueueSize` limits the backlog that builds up behind a flush still running. + +Because no caller is waiting on a scheduled flush, failures go to `onError` instead — which is also handed the items, so they can be re-queued rather than lost. +```dart +final markRead = buffer( + (ids) => api.markAllRead(ids), + const Duration(milliseconds: 500), + onError: (error, stackTrace, ids) => retryQueue.addAll(ids), +); +``` + +Handing them back to the buffer retries them on the next flush: +```dart +late final markRead = buffer( + (ids) => api.markAllRead(ids), + const Duration(milliseconds: 500), + onError: (error, stackTrace, ids) => markRead.addAll(ids), +); +``` +Retried items go to the back of the buffer, so their order is not preserved, and nothing spaces the attempts out. For that, put [backOff](#backoff) inside the flush instead: +```dart +final markRead = buffer( + (ids) => backOff( + () => api.markAllRead(ids), + maxAttempts: 4, + retryIf: (error, attempt) => error is SocketException, + ), + const Duration(milliseconds: 500), + // Reached only once backoff has run out of attempts. + onError: (error, stackTrace, ids) => log.warning('gave up on ${ids.length}'), +); +``` +Every attempt re-sends the same batch, `onError` fires once at the end rather than per attempt, and `await markRead.flush()` waits the retries out. Pick one of the two though — requeuing through `onError` *and* retrying with `backOff` compounds the two schedules. + ### Pending Used to check if the there are functions still remaining to get invoked. ```dart final pending = debouncedFunction.isPending; final pending = throttledFunction.isPending; +final pending = bufferedFunction.isPending; ``` ### Flush @@ -178,6 +264,9 @@ Used to immediately invoke all the remaining delayed functions. ```dart final result = debouncedFunction.flush(); final result = throttledFunction.flush(); +// A buffer's flush covers the items it holds right now, and completes once +// your function does, so it can be awaited before going away. +await bufferedFunction.flush(); ``` ### Cancellation @@ -185,4 +274,6 @@ Used to cancel all the remaining delayed functions. ```dart debouncedFunction.cancel(); throttledFunction.cancel(); +// Discards the items collected so far, without invoking your function. +bufferedFunction.cancel(); ``` diff --git a/example/lib/main.dart b/example/lib/main.dart index 91a8488..3117784 100644 --- a/example/lib/main.dart +++ b/example/lib/main.dart @@ -36,4 +36,22 @@ void main() { for (var i = 0; i < 10000; i++) { throttledFunction(); } + + // prints how many values it was handed at once + void printBatchSize(List values) { + print('got ${values.length} values'); + } + + // collects the values it is given and invokes `func` once with all of them, + // instead of once per value + final bufferedFunction = printBatchSize.buffered( + const Duration(milliseconds: 100), + maxSize: 1000, + ); + + // bufferedFunction prints 10 times, once per full buffer, even though + // invoked 10000 times — and no value is dropped + for (var i = 0; i < 10000; i++) { + bufferedFunction(i); + } } diff --git a/lib/rate_limiter.dart b/lib/rate_limiter.dart index 36a3ff8..be1d141 100644 --- a/lib/rate_limiter.dart +++ b/lib/rate_limiter.dart @@ -1,6 +1,7 @@ library rate_limiter; export 'src/backoff.dart'; +export 'src/buffer.dart'; export 'src/debounce.dart'; export 'src/throttle.dart'; export 'src/extension.dart'; diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart new file mode 100644 index 0000000..5203ff5 --- /dev/null +++ b/lib/src/buffer.dart @@ -0,0 +1,312 @@ +import 'dart:async'; + +/// Invoked with every item a [Buffer] collected, in the order they arrived. +typedef BufferFlushCallback = FutureOr Function(List items); + +/// Invoked when a flush a [Buffer] started itself fails, with the items that +/// flush was carrying so they can be re-queued rather than lost. +typedef BufferErrorCallback = void Function( + Object error, + StackTrace stackTrace, + List items, +); + +/// Invoked with the items a [Buffer] dropped to stay within `maxQueueSize`. +typedef BufferDropCallback = void Function(List items); + +/// What a [Buffer] gives up when it is holding `maxQueueSize` items and +/// another one arrives. +enum OverflowPolicy { + /// Drops the items that have been waiting longest, keeping the newest. + dropOldest, + + /// Drops the items that just arrived, keeping the ones already waiting. + dropNewest, +} + +/// Creates a buffered function that collects the items passed to it and +/// invokes `onFlush` **once** with all of them, instead of once per item. +/// +/// The buffer is flushed `wait` after the first item lands in it, or as soon +/// as it holds `maxSize` items, whichever comes first. The buffered function +/// comes with a [Buffer.cancel] method to discard the collected items and a +/// [Buffer.flush] method to invoke `onFlush` immediately. +/// +/// Only one flush runs at a time. Items arriving while `onFlush` is working +/// collect for the next one, which goes out as soon as the current one +/// finishes rather than waiting all over again — so `wait` is how long items +/// sit for company, never a queue behind the flush ahead of them. +/// +/// By default nothing is dropped and no caller is ever slowed down: this is +/// not a capacity buffer. `maxSize` caps what any one flush carries, and +/// `maxQueueSize` caps the backlog that builds up behind a slow one, shedding +/// the excess per `overflow`. Where `Debounce` and `Throttle` keep only the +/// arguments of the last call and discard the rest, a [Buffer] keeps them all. +/// +/// Some examples: +/// +/// Collapse a burst of per-item calls into a single request. +/// ```dart +/// final markRead = Buffer( +/// (ids) => api.markAllRead(ids), +/// const Duration(milliseconds: 500), +/// maxSize: 25, +/// ); +/// +/// void onMessageSeen(String id) => markRead(id); +/// ``` +/// +/// Batch everything queued up in the current event loop turn, the way a +/// data loader does, by waiting for no time at all. +/// ```dart +/// final loadUsers = Buffer( +/// (ids) => api.getUsers(ids), +/// Duration.zero, +/// ); +/// ``` +/// +/// Keep a runaway producer from growing the buffer without bound, and log +/// what that costs. +/// ```dart +/// final trackEvent = Buffer( +/// (events) => analytics.send(events), +/// const Duration(seconds: 5), +/// maxQueueSize: 10000, +/// onDrop: (events) => log.warning('dropped ${events.length} events'), +/// ); +/// ``` +/// +/// Send what is left before going away. +/// ```dart +/// Future dispose() => markRead.flush(); +/// ``` +class Buffer { + /// Creates a new instance of [Buffer]. + Buffer( + this._onFlush, + Duration wait, { + int? maxSize, + int? maxQueueSize, + OverflowPolicy overflow = OverflowPolicy.dropOldest, + BufferErrorCallback? onError, + BufferDropCallback? onDrop, + }) : assert( + maxSize == null || maxSize > 0, + 'maxSize must be greater than 0', + ), + assert( + maxQueueSize == null || maxQueueSize > 0, + 'maxQueueSize must be greater than 0', + ), + _wait = wait, + _maxSize = maxSize, + _maxQueueSize = maxQueueSize, + _overflow = overflow, + _onError = onError, + _onDrop = onDrop; + + final BufferFlushCallback _onFlush; + final Duration _wait; + final int? _maxSize; + final int? _maxQueueSize; + final OverflowPolicy _overflow; + final BufferErrorCallback? _onError; + final BufferDropCallback? _onDrop; + + final _items = []; + Timer? _timer; + + // Settles when the flush running right now finishes, however it finishes. + // Never carries its error, so a failed flush cannot wedge the queue. + Future? _inFlight; + + // Set when a buffer came due while a flush was running, so the wait is not + // served twice over. + var _isDue = false; + + /// The number of items waiting to get flushed. + /// + /// Counts what is still held. Items handed to `onFlush` are gone from here + /// even while that flush is running. + int get length => _items.length; + + /// True if there are items waiting to get flushed, or a flush is still + /// running. + bool get isPending => _timer != null || _inFlight != null; + + /// Adds [item] to the buffer, arming the flush if it is the first one in. + void call(T item) { + _items.add(item); + _collect(); + } + + /// Adds every item in [items] to the buffer. + /// + /// With a `maxSize` set, this hands over one full buffer at a time, starting + /// the next only once the one before it has finished. With a `maxQueueSize` + /// set, the excess is dropped as one group rather than an item at a time. + void addAll(Iterable items) { + _items.addAll(items); + _collect(); + } + + /// Invokes `onFlush` with the items collected so far, and keeps going until + /// everything held when this was called has been handed over. + /// + /// If a flush is already running, this waits for it too, so awaiting the + /// result drains the buffer — which is what makes it safe to call from + /// `dispose`. On a buffer holding nothing it waits out the running flush and + /// no more, leaving items that arrive afterwards to their own window. + /// + /// Whichever call starts a flush owns its failure: one started here lands on + /// the returned future, and one the buffer started itself goes to `onError`. + /// So a drain that waits out a flush already running can complete normally + /// even though that flush failed — `onError` was told instead. + Future flush() { + if (_inFlight case final inFlight?) { + // Already empty, so the flush running is all there is left to wait for. + // Draining past it would cut short the window of items that have only + // just arrived, and enrol this caller in sending them. + if (_items.isEmpty) return inFlight; + + return inFlight.then((_) => flush()); + } + + if (_items.isEmpty) return Future.value(); + + return _startFlush(report: false).then((_) => flush()); + } + + /// Discards the collected items without invoking `onFlush`. + /// + /// A flush already running is left alone; its items were handed over before + /// this was called. + void cancel() { + _timer?.cancel(); + _timer = null; + _isDue = false; + _items.clear(); + } + + // Shared by `call` and `addAll` so a bulk add costs one pass, and so a + // group that overflows is reported to `onDrop` in one piece. + void _collect() { + if (_maxQueueSize case final maxQueueSize? + when _items.length > maxQueueSize) { + _dropDownTo(maxQueueSize); + } + + _pump(); + } + + bool get _isFull { + if (_maxSize case final maxSize?) return _items.length >= maxSize; + return false; + } + + // Starts a flush if one is due and none is running, otherwise arms the wait. + // Re-entered when a flush settles, so the buffer drains one flush at a time. + void _pump() { + if (_items.isEmpty) return; + + if (_isDue || _isFull) { + // One at a time. Whoever is flushing pumps again on the way out, so + // these go as soon as it is done rather than waiting all over again. + if (_inFlight != null) return; + + _startFlush(report: true); + return; + } + + // Armed even behind a running flush: the wait is how long these items are + // willing to sit for company, not a queue behind the flush ahead of them. + _timer ??= Timer(_wait, _onDue); + } + + void _onDue() { + _timer = null; + _isDue = true; + _pump(); + } + + // Hands the next batch over and claims the queue while it runs. + // + // The claim is staked before `onFlush` is invoked, because `onFlush` can add + // items synchronously and those have to queue behind this flush rather than + // start a second one. + // + // With `report`, a failure goes to `onError`; without it, the failure is + // left on the returned future for whoever asked for the flush. + Future _startFlush({required bool report}) { + final items = _take(_maxSize); + + // A completer rather than the flush itself: `onFlush` is invoked + // synchronously below, so there is no future to claim the queue with until + // after the window this is closing. + final settled = Completer(); + _inFlight = settled.future; + + final flushing = report ? _invokeAndReport(items) : _invoke(items); + + // `whenComplete` runs whichever way it ends, and `ignore` takes the error + // off this derived future, so a failure cannot wedge the queue. + flushing.whenComplete(() { + _inFlight = null; + settled.complete(); + + // Only the scheduled path pumps. An explicit flush drives its own drain, + // which is what keeps every chunk it sends answerable to its caller. + if (report) _pump(); + }).ignore(); + + return flushing; + } + + void _dropDownTo(int maxQueueSize) { + final excess = _items.length - maxQueueSize; + final from = switch (_overflow) { + OverflowPolicy.dropOldest => 0, + OverflowPolicy.dropNewest => maxQueueSize, + }; + + final dropped = _items.sublist(from, from + excess); + _items.removeRange(from, from + excess); + + _onDrop?.call(dropped); + } + + // Takes up to `count` items out of the buffer, disarming the wait. + // + // Taken before `onFlush` is invoked, so anything added while it runs + // collects into the next buffer instead of joining this one. + List _take(int? count) { + _timer?.cancel(); + _timer = null; + _isDue = false; + + final take = + (count == null || count > _items.length) ? _items.length : count; + final items = _items.sublist(0, take); + _items.removeRange(0, take); + return items; + } + + // Hands the items over, turning whatever `onFlush` returns into a future so + // the buffer can tell when it is done. + Future _invoke(List items) async => _onFlush(items); + + // A flush the buffer started itself has no caller to hand a failure back to, + // so it goes to `onError`. Without one it is reported to the zone, as an + // unhandled asynchronous error would be. + Future _invokeAndReport(List items) async { + try { + await _invoke(items); + } catch (error, stackTrace) { + if (_onError case final onError?) { + onError(error, stackTrace, items); + return; + } + Zone.current.handleUncaughtError(error, stackTrace); + } + } +} diff --git a/lib/src/extension.dart b/lib/src/extension.dart index d85fc37..e45b855 100644 --- a/lib/src/extension.dart +++ b/lib/src/extension.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'backoff.dart'; +import 'buffer.dart'; import 'debounce.dart'; import 'throttle.dart'; @@ -24,6 +25,31 @@ extension BackOffExtension on FutureOr Function() { ).call(); } +/// Useful rate limiter extensions for [Function] class. +/// +/// Deliberately not [BufferFlushCallback], which returns `FutureOr`: a +/// plain `void` function is a subtype of this, so both shapes resolve here. +extension BufferExtension on void Function(List items) { + /// Converts this into a [Buffer] function. + Buffer buffered( + Duration wait, { + int? maxSize, + int? maxQueueSize, + OverflowPolicy overflow = OverflowPolicy.dropOldest, + BufferErrorCallback? onError, + BufferDropCallback? onDrop, + }) => + Buffer( + this, + wait, + maxSize: maxSize, + maxQueueSize: maxQueueSize, + overflow: overflow, + onError: onError, + onDrop: onDrop, + ); +} + /// Useful rate limiter extensions for [Function] class. extension RateLimit on Function { /// Converts this into a [Debounce] function. @@ -89,6 +115,26 @@ Debounce debounce( maxWait: maxWait, ); +/// TopLevel lambda to create [Buffer] functions. +Buffer buffer( + BufferFlushCallback onFlush, + Duration wait, { + int? maxSize, + int? maxQueueSize, + OverflowPolicy overflow = OverflowPolicy.dropOldest, + BufferErrorCallback? onError, + BufferDropCallback? onDrop, +}) => + Buffer( + onFlush, + wait, + maxSize: maxSize, + maxQueueSize: maxQueueSize, + overflow: overflow, + onError: onError, + onDrop: onDrop, + ); + /// TopLevel lambda to create [Throttle] functions. Throttle throttle( Function func, diff --git a/test/buffer_backoff_test.dart b/test/buffer_backoff_test.dart new file mode 100644 index 0000000..7227842 --- /dev/null +++ b/test/buffer_backoff_test.dart @@ -0,0 +1,228 @@ +import 'package:fake_async/fake_async.dart'; +import 'package:rate_limiter/rate_limiter.dart'; +import 'package:test/test.dart'; + +import 'utils.dart'; + +class _Unreachable implements Exception {} + +class _Rejected implements Exception {} + +void main() { + group('buffer with a backed-off flush', () { + // randomizationFactor is 0 so the retry delays land where the comments say + // they do: 200ms * 2^attempt, meaning 400ms, then 800ms, then 1600ms. + Buffer build({ + required Future Function(List ids) send, + void Function(Object, StackTrace, List)? onError, + int maxAttempts = 4, + }) { + return buffer( + (ids) => backOff( + () => send(ids), + maxAttempts: maxAttempts, + delayFactor: 200.toDuration(), + randomizationFactor: 0, + retryIf: (error, attempt) => error is _Unreachable, + ), + 500.toDuration(), + onError: onError, + ); + } + + test('should retry the whole batch without telling onError', () { + fakeAsync((async) { + final attempts = >[]; + Object? reported; + + final markRead = build( + send: (ids) async { + attempts.add(ids); + if (attempts.length == 1) throw _Unreachable(); + }, + onError: (e, s, ids) => reported = e, + ); + + markRead('a'); + markRead('b'); + + async.elapse(500.toDuration()); + expect(attempts, [ + ['a', 'b'], + ]); + + async.elapse(400.toDuration()); + expect( + attempts, + [ + ['a', 'b'], + ['a', 'b'], + ], + reason: 'the same batch went out again, intact'); + expect(reported, isNull, reason: 'it succeeded, so nobody was told'); + }); + }); + + test('should hand the items to onError only once backoff gives up', () { + fakeAsync((async) { + var attempts = 0; + Object? reported; + List? handedBack; + + final markRead = build( + maxAttempts: 3, + send: (ids) async { + attempts++; + throw _Unreachable(); + }, + onError: (e, s, ids) { + reported = e; + handedBack = ids; + }, + ); + + markRead('a'); + + async.elapse(500.toDuration()); + expect(reported, isNull, reason: 'still retrying'); + + async.elapse(1200.toDuration()); + expect(attempts, 3); + expect(reported, isA<_Unreachable>()); + expect(handedBack, ['a'], reason: 'so they can be re-queued'); + }); + }); + + test('should go straight to onError for what retryIf rejects', () { + fakeAsync((async) { + var attempts = 0; + Object? reported; + + final markRead = build( + send: (ids) async { + attempts++; + throw _Rejected(); + }, + onError: (e, s, ids) => reported = e, + ); + + markRead('a'); + async.elapse(500.toDuration()); + + expect(attempts, 1); + expect(reported, isA<_Rejected>()); + }); + }); + + test('should hold new items until the retries are done', () { + fakeAsync((async) { + final attempts = >[]; + + final markRead = build( + send: (ids) async { + attempts.add(ids); + if (attempts.length < 3) throw _Unreachable(); + }, + onError: (e, s, ids) {}, + ); + + markRead('a'); + async.elapse(500.toDuration()); // attempt one of 'a' fails + + markRead('b'); + async.elapse(500.toDuration()); // 'b' comes due at t=1000, but waits + + expect( + attempts, + [ + ['a'], + ['a'], + ], + reason: 'only the retry of a has gone out'); + + // The chain runs to t=1700: 400ms then 800ms between its attempts. + async.elapse(700.toDuration()); + + expect( + attempts, + [ + ['a'], + ['a'], + ['a'], + ['b'], + ], + reason: "'b' went the moment the chain freed the queue"); + }); + }); + + test('should keep new items out of a batch being retried', () { + fakeAsync((async) { + final attempts = >[]; + + final markRead = build( + send: (ids) async { + attempts.add(ids); + if (attempts.length == 1) throw _Unreachable(); + }, + onError: (e, s, ids) {}, + ); + + markRead('a'); + async.elapse(500.toDuration()); + + markRead('b'); + async.elapse(400.toDuration()); // the retry of 'a' lands here + + expect(attempts[1], ['a'], + reason: "'b' cannot join a batch already handed over"); + }); + }); + + test('should let flush wait out the retries', () { + fakeAsync((async) { + var attempts = 0; + var drained = false; + + final markRead = build( + send: (ids) async { + attempts++; + if (attempts == 1) throw _Unreachable(); + }, + onError: (e, s, ids) {}, + ); + + markRead('a'); + markRead.flush().then((_) => drained = true).ignore(); + + async.flushMicrotasks(); + expect(drained, isFalse, reason: 'the first attempt has failed'); + + async.elapse(400.toDuration()); + + expect(attempts, 2); + expect(drained, isTrue, reason: 'flush completed once backoff did'); + }); + }); + + test('should report a retry that runs out through flush to its caller', () { + fakeAsync((async) { + Object? caught; + + final markRead = build( + maxAttempts: 2, + send: (ids) async => throw _Unreachable(), + ); + + markRead('a'); + markRead.flush().then((_) {}, onError: (Object e) { + caught = e; + }).ignore(); + + async.elapse(400.toDuration()); + + expect(caught, isA<_Unreachable>(), + reason: 'the caller asked, so the caller is told'); + }); + }); + }); +} diff --git a/test/buffer_test.dart b/test/buffer_test.dart new file mode 100644 index 0000000..40a5aca --- /dev/null +++ b/test/buffer_test.dart @@ -0,0 +1,1211 @@ +import 'dart:async'; +import 'dart:math'; + +import 'package:fake_async/fake_async.dart'; +import 'package:rate_limiter/rate_limiter.dart'; +import 'package:test/test.dart'; + +import 'utils.dart'; + +void main() { + group('buffer', () { + test('should invoke onFlush once with everything collected', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + buffered('b'); + buffered('c'); + + expect(flushes, isEmpty); + expect(buffered.length, 3); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b', 'c'], + ]); + }); + }); + + test('should measure the wait from the first item, not the last', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + async.elapse(20.toDuration()); + + // A later item rides the deadline already set; it does not push it out + // the way another debounced call would. + buffered('b'); + async.elapse(12.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ]); + }); + }); + + test('should open a new buffer for the items after a flush', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + async.elapse(32.toDuration()); + + buffered('b'); + expect(buffered.isPending, isTrue); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a'], + ['b'], + ]); + }); + }); + + test('should flush as soon as it holds maxSize items', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 2, + ); + + buffered('a'); + buffered('b'); + + // Full, so it went without waiting out the rest of the window. + expect(flushes, [ + ['a', 'b'], + ]); + expect(buffered.isPending, isTrue, + reason: 'the flush is still running'); + + async.flushMicrotasks(); + expect(buffered.isPending, isFalse); + + buffered('c'); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ['c'], + ]); + }); + }); + + test('should batch a burst from a single event loop turn', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, Duration.zero); + + buffered('a'); + buffered('b'); + buffered('c'); + + expect(flushes, isEmpty, reason: 'the burst has not finished yet'); + + async.elapse(Duration.zero); + + expect(flushes, [ + ['a', 'b', 'c'], + ]); + }); + }); + + test('should fill as many buffers as addAll has items for', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 2, + ); + + buffered.addAll([1, 2, 3, 4, 5]); + + // One at a time: the second buffer waits for the first to finish. + expect(flushes, [ + [1, 2], + ]); + + async.flushMicrotasks(); + expect(flushes, [ + [1, 2], + [3, 4], + ]); + expect(buffered.length, 1, reason: 'the remainder waits its turn'); + + async.elapse(32.toDuration()); + + expect(flushes, [ + [1, 2], + [3, 4], + [5], + ]); + }); + }); + + test('should fill the buffer already open when addAll arrives', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 2, + ); + + buffered(1); + buffered.addAll([2, 3, 4]); + async.flushMicrotasks(); + + expect(flushes, [ + [1, 2], + [3, 4], + ]); + expect(buffered.length, 0); + expect(buffered.isPending, isFalse); + }); + }); + + test('should keep collecting after a flush fails', () { + fakeAsync((async) { + final flushes = >[]; + var failNextFlush = true; + + final buffered = buffer( + (items) { + flushes.add(items); + if (failNextFlush) { + failNextFlush = false; + throw Exception('flush failed'); + } + }, + 32.toDuration(), + maxSize: 2, + onError: (e, s, items) {}, + ); + + buffered('a'); + buffered('b'); + + expect(flushes, [ + ['a', 'b'], + ]); + expect(buffered.length, 0); + + buffered('c'); + expect( + buffered.isPending, + isTrue, + reason: 'a failed flush must not wedge the buffer', + ); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ['c'], + ]); + }); + }); + + test('should flush every item when maxSize is one', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 1, + ); + + buffered('a'); + buffered('b'); + async.flushMicrotasks(); + + expect(flushes, [ + ['a'], + ['b'], + ]); + expect(buffered.isPending, isFalse); + }); + }); + + test('should drop the oldest items once maxQueueSize is reached', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 3, + onDrop: drops.add, + ); + + buffered.addAll(['a', 'b', 'c']); + buffered('d'); + + expect(drops, [ + ['a'], + ]); + expect(buffered.length, 3); + + async.elapse(32.toDuration()); + + expect( + flushes, + [ + ['b', 'c', 'd'], + ], + reason: 'the dropped item never reached onFlush'); + }); + }); + + test('should drop what just arrived when told to keep the oldest', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 3, + overflow: OverflowPolicy.dropNewest, + onDrop: drops.add, + ); + + buffered.addAll(['a', 'b', 'c']); + buffered('d'); + + expect(drops, [ + ['d'], + ]); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b', 'c'], + ]); + }); + }); + + test('should report an overflowing group as one drop', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 3, + onDrop: drops.add, + ); + + buffered('a'); + buffered.addAll(['b', 'c', 'd', 'e']); + + // Reported in one piece, rather than once per item over-the-line. + expect(drops, [ + ['a', 'b'], + ]); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['c', 'd', 'e'], + ]); + }); + }); + + test('should keep the waiting items when dropping the newest', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 2, + overflow: OverflowPolicy.dropNewest, + onDrop: drops.add, + ); + + buffered('a'); + buffered.addAll(['b', 'c', 'd']); + + // 'a' has been waiting, so it survives an incoming group too big to + // fit alongside it. + expect(drops, [ + ['c', 'd'], + ]); + + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ]); + }); + }); + + test('should keep only the newest when a group dwarfs maxQueueSize', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 2, + onDrop: drops.add, + ); + + buffered.addAll([1, 2, 3, 4, 5]); + + expect(drops, [ + [1, 2, 3], + ]); + expect(buffered.length, 2); + + async.elapse(32.toDuration()); + + expect(flushes, [ + [4, 5], + ]); + }); + }); + + test('should drop without an onDrop to report to', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxQueueSize: 2, + ); + + buffered.addAll(['a', 'b', 'c']); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['b', 'c'], + ]); + }); + }); + + test('should discard the collected items on cancel', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + buffered('b'); + buffered.cancel(); + + expect(buffered.length, 0); + expect(buffered.isPending, isFalse); + + // Past the deadline, so `cancel` is what stopped it. + async.elapse(128.toDuration()); + + expect(flushes, isEmpty); + }); + }); + + test('should never have two flushes running at once', () { + fakeAsync((async) { + final started = >[]; + final blockers = >[]; + var running = 0; + var mostAtOnce = 0; + + final buffered = buffer( + (items) { + started.add(items); + running++; + mostAtOnce = running > mostAtOnce ? running : mostAtOnce; + final blocker = Completer(); + blockers.add(blocker); + return blocker.future.whenComplete(() => running--); + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); + + buffered('b'); + async.elapse(32.toDuration()); + + buffered('c'); + async.elapse(32.toDuration()); + + expect( + started, + [ + ['a'], + ], + reason: "'b' and 'c' waited behind the flush carrying 'a'"); + expect(mostAtOnce, 1); + expect(buffered.isPending, isTrue); + + for (final blocker in blockers.toList()) { + blocker.complete(); + } + async.elapse(32.toDuration()); + + expect( + started, + [ + ['a'], + ['b', 'c'], + ], + reason: 'the two that waited went together'); + expect(mostAtOnce, 1); + }); + }); + + test('should not let onFlush start a second flush from inside itself', () { + fakeAsync((async) { + final started = >[]; + var running = 0; + var mostAtOnce = 0; + var reentered = false; + + late final Buffer buffered; + buffered = buffer( + (items) { + started.add(items); + running++; + mostAtOnce = running > mostAtOnce ? running : mostAtOnce; + + if (!reentered) { + reentered = true; + // Re-entrant, and maxSize of one makes it due immediately. The + // claim on the queue has to be staked before we get here. + buffered('b'); + } + + return Future.delayed(10.toDuration(), () => running--); + }, + 32.toDuration(), + maxSize: 1, + ); + + buffered('a'); + + expect( + started, + [ + ['a'], + ], + reason: "'b' has to wait its turn"); + expect(mostAtOnce, 1); + + async.elapse(100.toDuration()); + + expect(started, [ + ['a'], + ['b'], + ]); + expect(mostAtOnce, 1); + }); + }); + + test('should go straight out when a wait was served during a flush', () { + fakeAsync((async) { + final started = >[]; + final blocker = Completer(); + + final buffered = buffer( + (items) { + started.add(items); + return started.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); // flush of 'a' starts and blocks + + buffered('b'); + async.elapse(32.toDuration()); // 'b' comes due, but has to wait + + expect(started, [ + ['a'], + ]); + + blocker.complete(); + async.flushMicrotasks(); + + // No second wait: 'b' was already overdue when the flush finished. + expect(started, [ + ['a'], + ['b'], + ]); + }); + }); + + test('should report itself as pending while a flush is running', () { + fakeAsync((async) { + final blocker = Completer(); + + final buffered = buffer( + (items) => blocker.future, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); + + expect(buffered.length, 0, reason: 'handed over'); + expect(buffered.isPending, isTrue, reason: 'but not done'); + + blocker.complete(); + async.flushMicrotasks(); + + expect(buffered.isPending, isFalse); + }); + }); + + test('should let flush await the one already running', () { + fakeAsync((async) { + final started = >[]; + final blocker = Completer(); + var drained = false; + + final buffered = buffer( + (items) { + started.add(items); + return started.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); // flush of 'a' starts and blocks + + buffered('b'); + buffered.flush().then((_) => drained = true).ignore(); + async.flushMicrotasks(); + + expect(drained, isFalse, reason: 'the flush carrying a is still going'); + expect(started, [ + ['a'], + ]); + + blocker.complete(); + async.flushMicrotasks(); + + expect(started, [ + ['a'], + ['b'], + ]); + expect(drained, isTrue, reason: 'and everything after it went too'); + }); + }); + + test('should drain every full buffer on flush', () { + fakeAsync((async) { + final flushes = >[]; + var drained = false; + + final buffered = buffer( + flushes.add, + 32.toDuration(), + maxSize: 2, + ); + + buffered.addAll([1, 2, 3, 4, 5]); + buffered.flush().then((_) => drained = true).ignore(); + async.flushMicrotasks(); + + expect( + flushes, + [ + [1, 2], + [3, 4], + [5], + ], + reason: 'flush keeps going until nothing is held'); + expect(drained, isTrue); + expect(buffered.isPending, isFalse); + }); + }); + + test('should keep the queue moving when a flush fails', () { + fakeAsync((async) { + final started = >[]; + final blocker = Completer(); + + final buffered = buffer( + (items) { + started.add(items); + return started.length == 1 + ? blocker.future.then((_) => throw Exception('flush failed')) + : null; + }, + 32.toDuration(), + onError: (e, s, items) {}, + ); + + buffered('a'); + async.elapse(32.toDuration()); + + buffered('b'); + blocker.complete(); + async.elapse(32.toDuration()); + + expect( + started, + [ + ['a'], + ['b'], + ], + reason: 'a failure must not wedge what is queued behind it'); + expect(buffered.isPending, isFalse); + }); + }); + + test('should wait out the running flush and no more when empty', () { + fakeAsync((async) { + final blocker = Completer(); + final flushes = >[]; + var drained = false; + + final buffered = buffer( + (items) { + flushes.add(items); + return flushes.length == 1 ? blocker.future : null; + }, + 1000.toDuration(), + ); + + buffered(1); + async.elapse(1000.toDuration()); // the flush of 1 starts and blocks + expect(buffered.length, 0, reason: 'handed over already'); + + buffered.flush().then((_) => drained = true).ignore(); + + // Arrives after the drain was asked for, with its window still to run. + buffered(2); + + blocker.complete(); + async.flushMicrotasks(); + + expect(drained, isTrue); + expect( + flushes, + [ + [1], + ], + reason: 'the window of 2 was not cut short by an unrelated flush'); + expect(buffered.length, 1); + }); + }); + + test('should answer to flush for every chunk flush itself sent', () { + fakeAsync((async) { + final blocker = Completer(); + var calls = 0; + Object? caughtByFlush; + Object? caughtByOnError; + + final buffered = buffer( + (items) { + calls++; + if (calls == 1) return blocker.future; + throw Exception('the second chunk failed'); + }, + 32.toDuration(), + maxSize: 2, + onError: (e, s, items) => caughtByOnError = e, + ); + + buffered('a'); + buffered.flush().then((_) {}, onError: (Object e) { + caughtByFlush = e; + }).ignore(); + + // Arriving mid-drain, so the drain picks them up as its second chunk. + buffered('b'); + buffered('c'); + + blocker.complete(); + async.elapse(32.toDuration()); + + expect(calls, 2); + expect(caughtByFlush, isNotNull, reason: 'the drain sent it'); + expect(caughtByOnError, isNull, + reason: 'so onError has nothing to say'); + }); + }); + + test('should leave a flush it only waited for to onError', () { + fakeAsync((async) { + var calls = 0; + Object? caughtByFlush; + Object? caughtByOnError; + var drained = false; + + final buffered = buffer( + (items) { + calls++; + if (calls == 2) throw Exception('the second chunk failed'); + }, + 32.toDuration(), + maxSize: 2, + onError: (e, s, items) => caughtByOnError = e, + ); + + // Two full buffers, so the buffer schedules both chunks itself and + // flush only waits them out. + buffered.addAll([1, 2, 3, 4]); + buffered.flush().then((_) { + drained = true; + }, onError: (Object e) { + caughtByFlush = e; + }).ignore(); + + async.elapse(32.toDuration()); + + expect(calls, 2); + expect(caughtByOnError, isNotNull, reason: 'the buffer sent it'); + expect(caughtByFlush, isNull); + expect(drained, isTrue, + reason: 'the buffer did drain, and it is empty'); + expect(buffered.length, 0); + }); + }); + + test('should collect again after being cancelled', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + expect(buffered.isPending, isFalse); + buffered.cancel(); + + buffered('a'); + buffered.cancel(); + + buffered('b'); + async.elapse(32.toDuration()); + + expect( + flushes, + [ + ['b'], + ], + reason: 'cancel drops what is held, it does not stop the buffer'); + }); + }); + + test('should leave a flush already on its way out alone on cancel', () { + fakeAsync((async) { + final flushes = >[]; + final completer = Completer(); + + final buffered = buffer( + (items) { + flushes.add(items); + return completer.future; + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); + + buffered('b'); + buffered.cancel(); + completer.complete(); + async.elapse(32.toDuration()); + + expect( + flushes, + [ + ['a'], + ], + reason: 'the in-flight items were already handed over'); + }); + }); + + test('should invoke onFlush immediately on flush', () { + fakeAsync((async) { + final flushes = >[]; + + final buffered = buffer(flushes.add, 32.toDuration()); + + buffered('a'); + buffered.flush().ignore(); + async.flushMicrotasks(); + + expect(flushes, [ + ['a'], + ]); + expect(buffered.isPending, isFalse); + + // The timer it left behind would flush an empty buffer, or worse, + // report itself as pending again. + async.elapse(128.toDuration()); + + expect(flushes, [ + ['a'], + ]); + }); + }); + + test('should complete flush once onFlush does', () { + fakeAsync((async) { + final completer = Completer(); + var flushed = false; + + final buffered = buffer( + (items) => completer.future, + 32.toDuration(), + ); + + buffered('a'); + buffered.flush().then((_) => flushed = true).ignore(); + async.elapse(128.toDuration()); + + expect(flushed, isFalse, reason: 'onFlush has not finished'); + + completer.complete(); + async.flushMicrotasks(); + + expect(flushed, isTrue); + }); + }); + + test('should do nothing when flushing an empty buffer', () { + fakeAsync((async) { + var callCount = 0; + + final buffered = + buffer((items) => callCount++, 32.toDuration()); + + buffered.flush().ignore(); + buffered.addAll(const []); + async.elapse(128.toDuration()); + + expect(callCount, 0); + expect(buffered.isPending, isFalse); + }); + }); + + test('should collect what arrives while onFlush is running', () { + fakeAsync((async) { + final flushes = >[]; + final completer = Completer(); + + final buffered = buffer( + (items) { + flushes.add(items); + return completer.future; + }, + 32.toDuration(), + ); + + buffered('a'); + async.elapse(32.toDuration()); + + // Joining the buffer being flushed would hand these to a request + // already on its way out. + buffered('b'); + completer.complete(); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a'], + ['b'], + ]); + }); + }); + + test('should hand a scheduled failure to onError', () { + fakeAsync((async) { + final error = Exception('flush failed'); + Object? caughtError; + List? caughtItems; + + final buffered = buffer( + (items) => throw error, + 32.toDuration(), + onError: (e, s, items) { + caughtError = e; + caughtItems = items; + }, + ); + + buffered('a'); + buffered('b'); + async.elapse(32.toDuration()); + + expect(caughtError, error); + // Handed back so they can be re-queued rather than lost. + expect(caughtItems, ['a', 'b']); + }); + }); + + test('should give a failure to the caller that asked for the flush', () { + fakeAsync((async) { + final error = Exception('flush failed'); + Object? caughtByFlush; + Object? caughtByOnError; + + final buffered = buffer( + (items) async => throw error, + 32.toDuration(), + onError: (e, s, items) => caughtByOnError = e, + ); + + buffered('a'); + buffered.flush().onError((e, s) => caughtByFlush = e); + async.flushMicrotasks(); + + expect(caughtByFlush, error); + expect( + caughtByOnError, + isNull, + reason: 'the caller was told, so onError has nothing to report', + ); + }); + }); + + test('should report a scheduled failure with no onError to the zone', () { + final error = Exception('flush failed'); + final caught = Completer(); + + return runZonedGuarded(() async { + final buffered = buffer( + (items) async => throw error, + Duration.zero, + ); + + buffered('a'); + + expect(await caught.future, error); + }, (e, s) { + if (!caught.isCompleted) caught.complete(e); + }); + }); + + test('should convert an existing function into a buffered one', () { + fakeAsync((async) { + final flushes = >[]; + + void markAllRead(List ids) => flushes.add(ids); + + final buffered = markAllRead.buffered(32.toDuration(), maxSize: 3); + + buffered('a'); + buffered('b'); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a', 'b'], + ]); + }); + }); + + test('should carry every option through the extension', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + + void markAllRead(List ids) => flushes.add(ids); + + final buffered = markAllRead.buffered( + 32.toDuration(), + maxQueueSize: 2, + overflow: OverflowPolicy.dropNewest, + onDrop: drops.add, + ); + + buffered.addAll(['a', 'b', 'c']); + async.elapse(32.toDuration()); + + expect(drops, [ + ['c'], + ]); + expect(flushes, [ + ['a', 'b'], + ]); + }); + }); + + test('should hold its invariants whatever order it is driven in', () { + // Two of these are relied on by the implementation: `isPending` reads + // the timer rather than counting items, and `flush` hands over the whole + // buffer without re-checking `maxSize`. + for (var seed = 0; seed < 200; seed++) { + final rng = Random(seed); + final byMaxSize = rng.nextBool(); + final limit = 1 + rng.nextInt(4); + + fakeAsync((async) { + final flushes = >[]; + final buffered = Buffer( + flushes.add, + 10.toDuration(), + maxSize: byMaxSize ? limit : null, + maxQueueSize: byMaxSize ? null : limit, + ); + + void checkInvariants(String op) { + final where = 'seed $seed, after $op'; + if (buffered.length > 0) { + expect( + buffered.isPending, + isTrue, + reason: '$where: buffered items with nothing to move them', + ); + } + // `maxSize` bounds each flush, not the buffer: items pile up + // behind a running flush. Only `maxQueueSize` caps what is held. + if (!byMaxSize) { + expect( + buffered.length, + lessThanOrEqualTo(limit), + reason: '$where: held more than maxQueueSize allows', + ); + } + } + + var next = 0; + for (var step = 0; step < 30; step++) { + switch (rng.nextInt(5)) { + case 0: + buffered(next++); + checkInvariants('call'); + case 1: + buffered.addAll(List.generate(rng.nextInt(6), (_) => next++)); + checkInvariants('addAll'); + case 2: + buffered.flush().ignore(); + checkInvariants('flush'); + case 3: + buffered.cancel(); + checkInvariants('cancel'); + case 4: + async.elapse(rng.nextInt(20).toDuration()); + checkInvariants('elapse'); + } + } + + for (final flush in flushes) { + expect(flush, isNotEmpty, reason: 'seed $seed: flushed nothing'); + if (byMaxSize) { + expect( + flush.length, + lessThanOrEqualTo(limit), + reason: 'seed $seed: handed over more than maxSize', + ); + } + } + }); + } + }); + + test('should reject a maxSize that can never fill', () { + expect( + () => Buffer((items) {}, Duration.zero, maxSize: 0), + throwsA(isA()), + ); + }); + + test('should reject a maxQueueSize that can never hold anything', () { + expect( + () => Buffer((items) {}, Duration.zero, maxQueueSize: 0), + throwsA(isA()), + ); + }); + + test('should cap the batch and the backlog independently', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + final blocker = Completer(); + + // maxSize caps what one flush carries; maxQueueSize caps what may pile + // up behind it. Serialized flushes are what make both reachable. + final buffered = buffer( + (items) { + flushes.add(items); + return blocker.future; + }, + 32.toDuration(), + maxSize: 2, + maxQueueSize: 3, + onDrop: drops.add, + ); + + buffered.addAll([1, 2]); + + expect( + flushes, + [ + [1, 2], + ], + reason: 'full, so one flush carrying maxSize items', + ); + + // These pile up behind the running flush, and that backlog is what + // maxQueueSize caps. + buffered.addAll([3, 4, 5, 6]); + + expect(buffered.length, 3); + expect(drops, [ + [3], + ]); + + blocker.complete(); + async.elapse(32.toDuration()); + + expect(flushes, [ + [1, 2], + [4, 5], + [6], + ]); + }); + }); + + test('should take an asynchronous onFlush in either form', () { + fakeAsync((async) { + final flushes = >[]; + + Future markAllRead(List ids) async => flushes.add(ids); + + final fromExtension = markAllRead.buffered(32.toDuration()); + final fromLambda = buffer(markAllRead, 32.toDuration()); + + fromExtension('a'); + fromLambda('b'); + async.elapse(32.toDuration()); + + expect(flushes, [ + ['a'], + ['b'], + ]); + }); + }); + }); +} From 5200a7e4cb79a387e3babb74d60907639370b1a0 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 16:09:01 +0200 Subject: [PATCH 02/10] fix: keep the queue moving when an explicit flush fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Only the scheduled path pumped on the way out, so an explicit `flush()` that failed left anything queued behind it stranded — its own recursive drain stops at the first error, and nothing else was left to move the items. With `maxSize: 2` and a full batch arriving in one `addAll` during the flush, `length` stayed at 2 while `isPending` reported false, and no amount of elapsed time ever sent them. The error path now pumps, handing what is left back to the buffer's own schedule. Hand off what can go immediately before measuring the backlog against `maxQueueSize`. A group arriving all at once with nothing running was capped against a batch that was never going to sit in the backlog: with `maxSize: 2` and `maxQueueSize: 3`, `addAll([1..6])` dropped three items and sent `[4, 5]`, where taking `[1, 2]` for the batch first leaves only `[3]` to drop. Correct two docs that promised the next batch goes out the moment the running flush finishes. It also has to have come due, so an item arriving 10ms into a 500ms window still waits that window out even when the flush ahead of it ends at 20ms. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 2 +- lib/src/buffer.dart | 44 ++++++++++++++++++++-------- test/buffer_test.dart | 68 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 101 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 9c3240e..c2e6a1e 100644 --- a/README.md +++ b/README.md @@ -170,7 +170,7 @@ final response = backOff( ### Buffer A _buffered function_ collects the items passed to it and invokes your function **once** with all of them, rather than once per item. Where debounce and throttle keep only the last call's arguments and drop the rest, a buffer keeps every one — so it batches work instead of shedding it. -The buffer is flushed once `wait` has passed since the first item landed in it, or as soon as it holds `maxSize` items, whichever comes first. Only one flush runs at a time: while your function is working, arriving items collect for the next one, which goes out the moment the current one finishes. By default nothing is dropped and no caller is ever slowed down — pass `maxQueueSize` to cap the buffer and shed the excess instead. +The buffer is flushed once `wait` has passed since the first item landed in it, or as soon as it holds `maxSize` items, whichever comes first. Only one flush runs at a time: while your function is working, arriving items collect for the next one, which goes out once it has come due *and* the running one has finished — whichever is later. By default nothing is dropped and no caller is ever slowed down — pass `maxQueueSize` to cap the buffer and shed the excess instead. #### Usage 1. Creating from scratch diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart index 5203ff5..d4987a2 100644 --- a/lib/src/buffer.dart +++ b/lib/src/buffer.dart @@ -33,9 +33,10 @@ enum OverflowPolicy { /// [Buffer.flush] method to invoke `onFlush` immediately. /// /// Only one flush runs at a time. Items arriving while `onFlush` is working -/// collect for the next one, which goes out as soon as the current one -/// finishes rather than waiting all over again — so `wait` is how long items -/// sit for company, never a queue behind the flush ahead of them. +/// collect for the next one, which goes out once it has come due — `wait` +/// after its own first item landed, or on reaching `maxSize` — and the running +/// flush has finished, whichever is later. A batch that came due while a flush +/// was running does not then serve its wait a second time. /// /// By default nothing is dropped and no caller is ever slowed down: this is /// not a capacity buffer. `maxSize` caps what any one flush carries, and @@ -191,12 +192,17 @@ class Buffer { // Shared by `call` and `addAll` so a bulk add costs one pass, and so a // group that overflows is reported to `onDrop` in one piece. void _collect() { + // Hands off what can go right now, so a group arriving all at once is not + // capped against a batch that was never going to sit in the backlog. + _pump(); + if (_maxQueueSize case final maxQueueSize? when _items.length > maxQueueSize) { _dropDownTo(maxQueueSize); - } - _pump(); + // The backlog shrank, so whatever is left of it needs its own wait. + _pump(); + } } bool get _isFull { @@ -248,16 +254,30 @@ class Buffer { final flushing = report ? _invokeAndReport(items) : _invoke(items); - // `whenComplete` runs whichever way it ends, and `ignore` takes the error - // off this derived future, so a failure cannot wedge the queue. - flushing.whenComplete(() { + void release() { _inFlight = null; settled.complete(); + } - // Only the scheduled path pumps. An explicit flush drives its own drain, - // which is what keeps every chunk it sends answerable to its caller. - if (report) _pump(); - }).ignore(); + // `ignore` takes the error off this derived future, so a failure cannot + // wedge the queue. + flushing.then( + (_) { + release(); + + // Only the scheduled path pumps on the way out. An explicit flush + // drives its own drain, which is what keeps every chunk it sends + // answerable to its caller. + if (report) _pump(); + }, + onError: (Object _, StackTrace __) { + release(); + + // An explicit drain stops at its first failure, so without this the + // items queued behind it would sit with nothing left to move them. + _pump(); + }, + ).ignore(); return flushing; } diff --git a/test/buffer_test.dart b/test/buffer_test.dart index 40a5aca..d262ef6 100644 --- a/test/buffer_test.dart +++ b/test/buffer_test.dart @@ -682,6 +682,74 @@ void main() { }); }); + test('should keep the queue moving when an explicit flush fails', () { + fakeAsync((async) { + final blocker = Completer(); + final flushes = >[]; + + final buffered = buffer( + (items) { + flushes.add(items); + return flushes.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxSize: 2, + onError: (e, s, items) {}, + ); + + buffered('a'); + buffered.flush().then((_) {}, onError: (Object _) {}).ignore(); + + // One call, so this goes straight to the full branch and never arms a + // wait of its own. The failed drain is all that could move it. + buffered.addAll(['b', 'c']); + + blocker.completeError(Exception('flush failed')); + async.flushMicrotasks(); + + expect( + flushes, + [ + ['a'], + ['b', 'c'], + ], + reason: 'a failed drain must not strand what queued behind it'); + expect(buffered.length, 0); + expect(buffered.isPending, isFalse); + }); + }); + + test('should send what it can before capping the backlog', () { + fakeAsync((async) { + final flushes = >[]; + final drops = >[]; + final blocker = Completer(); + + final buffered = buffer( + (items) { + flushes.add(items); + return blocker.future; + }, + 32.toDuration(), + maxSize: 2, + maxQueueSize: 3, + onDrop: drops.add, + ); + + buffered.addAll([1, 2, 3, 4, 5, 6]); + + // Nothing was running, so [1, 2] takes the batch and only what is + // genuinely backlog gets measured against maxQueueSize. + expect(flushes, [ + [1, 2], + ]); + expect(drops, [ + [3], + ]); + expect(buffered.length, 3); + }); + }); + test('should wait out the running flush and no more when empty', () { fakeAsync((async) { final blocker = Completer(); From 8293fb7cb3c60c7307a9758ead889624a0335f75 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 16:28:52 +0200 Subject: [PATCH 03/10] fix: validate the limits at runtime and keep a remainder's deadline `maxSize` and `maxQueueSize` were only asserted, and asserts are stripped in release. A `maxSize` of zero is worse than a crash there: every buffer looks full while each flush takes nothing out of it, so it spins sending empty batches and never sends what it holds. Measured with asserts off, it reached the probe's cap of 50 empty flushes with the item still queued. Both limits are now checked and throw `ArgumentError`, which fails loudly at construction instead. A remainder left behind by a full batch also lost the deadline measured from when its own items arrived. With `maxSize: 2` and `wait: 500ms`, `addAll([1, 2, 3])` sent `[1, 2]` and left `3` with no timer at all, so a flush running until 2000ms pushed `3` out at 2500ms rather than 2000ms when it had been overdue since 500ms. `_pump` now arms the wait for whatever a batch leaves behind, and skips arming only when the buffer is already overdue and simply waiting for its turn. Co-Authored-By: Claude Opus 5 (1M context) --- lib/src/buffer.dart | 42 ++++++++++++++++++------------------- test/buffer_test.dart | 48 ++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 66 insertions(+), 24 deletions(-) diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart index d4987a2..58a8018 100644 --- a/lib/src/buffer.dart +++ b/lib/src/buffer.dart @@ -91,21 +91,22 @@ class Buffer { OverflowPolicy overflow = OverflowPolicy.dropOldest, BufferErrorCallback? onError, BufferDropCallback? onDrop, - }) : assert( - maxSize == null || maxSize > 0, - 'maxSize must be greater than 0', - ), - assert( - maxQueueSize == null || maxQueueSize > 0, - 'maxQueueSize must be greater than 0', - ), - _wait = wait, - _maxSize = maxSize, - _maxQueueSize = maxQueueSize, + }) : _wait = wait, + _maxSize = _checkPositive(maxSize, 'maxSize'), + _maxQueueSize = _checkPositive(maxQueueSize, 'maxQueueSize'), _overflow = overflow, _onError = onError, _onDrop = onDrop; + // Checked rather than asserted, because asserts are stripped in release and + // a `maxSize` of zero is worse than a crash there: every buffer looks full + // while each flush takes nothing out of it, so it spins sending empty + // batches and never sends the items it holds. + static int? _checkPositive(int? limit, String name) { + if (limit == null || limit > 0) return limit; + throw ArgumentError.value(limit, name, 'must be greater than zero'); + } + final BufferFlushCallback _onFlush; final Duration _wait; final int? _maxSize; @@ -213,19 +214,18 @@ class Buffer { // Starts a flush if one is due and none is running, otherwise arms the wait. // Re-entered when a flush settles, so the buffer drains one flush at a time. void _pump() { - if (_items.isEmpty) return; - - if (_isDue || _isFull) { - // One at a time. Whoever is flushing pumps again on the way out, so - // these go as soon as it is done rather than waiting all over again. - if (_inFlight != null) return; - + // One at a time. Whoever is flushing pumps again on the way out, so a + // batch that came due meanwhile goes then rather than waiting afresh. + if (_inFlight == null && (_isDue || _isFull)) { _startFlush(report: true); - return; } - // Armed even behind a running flush: the wait is how long these items are - // willing to sit for company, not a queue behind the flush ahead of them. + // Nothing left to time, or already overdue and only waiting for its turn. + if (_items.isEmpty || _isDue) return; + + // Armed behind a running flush, and for a remainder the batch just sent + // left behind: the wait runs from when these items arrived, not from + // whenever the flush ahead of them happens to finish. _timer ??= Timer(_wait, _onDue); } diff --git a/test/buffer_test.dart b/test/buffer_test.dart index d262ef6..548cb6f 100644 --- a/test/buffer_test.dart +++ b/test/buffer_test.dart @@ -527,6 +527,46 @@ void main() { }); }); + test('should hold a remainder to its own deadline, not the flush ahead', + () { + fakeAsync((async) { + final blocker = Completer(); + final flushes = >[]; + + final buffered = buffer( + (items) { + flushes.add(items); + return flushes.length == 1 ? blocker.future : null; + }, + 500.toDuration(), + maxSize: 2, + ); + + // [1, 2] fills a batch and goes; 3 stays behind with its own window + // running from now. + buffered.addAll([1, 2, 3]); + + expect(flushes, [ + [1, 2], + ]); + + // The batch ahead of it takes far longer than that window. + async.elapse(2000.toDuration()); + expect(flushes.length, 1, reason: 'still waiting for the queue'); + + blocker.complete(); + async.flushMicrotasks(); + + expect( + flushes, + [ + [1, 2], + [3], + ], + reason: '3 was overdue, so it went as soon as its turn came'); + }); + }); + test('should go straight out when a wait was served during a flush', () { fakeAsync((async) { final started = >[]; @@ -1193,17 +1233,19 @@ void main() { } }); + // Thrown rather than asserted: asserts are stripped in release, and a + // maxSize of zero leaves the buffer spinning on empty flushes there. test('should reject a maxSize that can never fill', () { expect( () => Buffer((items) {}, Duration.zero, maxSize: 0), - throwsA(isA()), + throwsA(isA()), ); }); test('should reject a maxQueueSize that can never hold anything', () { expect( - () => Buffer((items) {}, Duration.zero, maxQueueSize: 0), - throwsA(isA()), + () => Buffer((items) {}, Duration.zero, maxQueueSize: -1), + throwsA(isA()), ); }); From 44f13c356edd267df82e359dac79064a72da2773 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 16:59:53 +0200 Subject: [PATCH 04/10] fix: bound an explicit drain, and stop the queue scaling quadratically MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `flush()` documented a snapshot — everything held when it was called — but its recursion re-read the live queue after every batch, so it also sent whatever arrived while it waited. Under a producer that never stops, that future never completes, which is a poor property for the one call `dispose` is told to await. It now takes the measure of the buffer once and hands over that many items, leaving later arrivals to their own window. Items arriving mid-drain therefore answer to `onError` rather than to the caller, which is the same rule as before: whichever call starts a flush owns its failure. Arm a remainder's wait before invoking `onFlush` rather than after. The callback runs synchronously up to its first await, so a second of synchronous work in there used to push a remainder's window out by that whole second even though its deadline had already passed. Hold the items in a `ListQueue`. Taking a batch off the front of a list shifts everything behind it, so draining N items in batches of K moved O(N^2/K) elements; front and back removal are now O(1) each. Draining 80k items in batches of 10 measures flat against 20k and 40k. Co-Authored-By: Claude Opus 5 (1M context) --- lib/src/buffer.dart | 84 ++++++++++++++++++++++++++++++------------- test/buffer_test.dart | 16 +++++---- 2 files changed, 68 insertions(+), 32 deletions(-) diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart index 58a8018..d8b7b1b 100644 --- a/lib/src/buffer.dart +++ b/lib/src/buffer.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:collection'; /// Invoked with every item a [Buffer] collected, in the order they arrived. typedef BufferFlushCallback = FutureOr Function(List items); @@ -115,7 +116,10 @@ class Buffer { final BufferErrorCallback? _onError; final BufferDropCallback? _onDrop; - final _items = []; + // A queue rather than a list, so taking a batch off the front and shedding + // the oldest are both cheap. Draining N items through a list would move + // O(N^2) elements, since every removal shifts everything behind it. + final _items = ListQueue(); Timer? _timer; // Settles when the flush running right now finishes, however it finishes. @@ -165,18 +169,12 @@ class Buffer { /// So a drain that waits out a flush already running can complete normally /// even though that flush failed — `onError` was told instead. Future flush() { - if (_inFlight case final inFlight?) { - // Already empty, so the flush running is all there is left to wait for. - // Draining past it would cut short the window of items that have only - // just arrived, and enrol this caller in sending them. - if (_items.isEmpty) return inFlight; - - return inFlight.then((_) => flush()); - } + // Already empty, so the flush running is all there is left to wait for. + // Draining past it would cut short the window of items that have only + // just arrived, and enrol this caller in sending them. + if (_items.isEmpty) return _inFlight ?? Future.value(); - if (_items.isEmpty) return Future.value(); - - return _startFlush(report: false).then((_) => flush()); + return _drain(_items.length); } /// Discards the collected items without invoking `onFlush`. @@ -190,6 +188,27 @@ class Buffer { _items.clear(); } + // Hands over [remaining] items, a batch at a time, waiting out anything + // already running first. + // + // Counted rather than draining until empty, so a caller of `flush` is not + // made to send whatever arrives while it waits — under a producer that never + // stops, that future would never complete. + Future _drain(int remaining) { + if (remaining <= 0 || _items.isEmpty) return Future.value(); + + if (_inFlight case final inFlight?) { + return inFlight.then((_) => _drain(remaining)); + } + + final sending = switch (_maxSize) { + final maxSize? when maxSize < _items.length => maxSize, + _ => _items.length, + }; + + return _startFlush(report: false).then((_) => _drain(remaining - sending)); + } + // Shared by `call` and `addAll` so a bulk add costs one pass, and so a // group that overflows is reported to `onDrop` in one piece. void _collect() { @@ -217,15 +236,22 @@ class Buffer { // One at a time. Whoever is flushing pumps again on the way out, so a // batch that came due meanwhile goes then rather than waiting afresh. if (_inFlight == null && (_isDue || _isFull)) { + // Arms the remainder itself, before the callback gets a chance to run. _startFlush(report: true); + return; } - // Nothing left to time, or already overdue and only waiting for its turn. + _armWait(); + } + + // Starts the wait for whatever is held, so it runs from when those items + // arrived rather than from whenever the flush ahead of them finishes. + // + // Does nothing when the buffer is empty, or when it is already overdue and + // only waiting for its turn at the queue. + void _armWait() { if (_items.isEmpty || _isDue) return; - // Armed behind a running flush, and for a remainder the batch just sent - // left behind: the wait runs from when these items arrived, not from - // whenever the flush ahead of them happens to finish. _timer ??= Timer(_wait, _onDue); } @@ -252,6 +278,11 @@ class Buffer { final settled = Completer(); _inFlight = settled.future; + // Timed before the callback runs: `onFlush` may work synchronously for a + // while, and a remainder's wait should run from now rather than from + // whenever that work returns. + _armWait(); + final flushing = report ? _invokeAndReport(items) : _invoke(items); void release() { @@ -284,13 +315,18 @@ class Buffer { void _dropDownTo(int maxQueueSize) { final excess = _items.length - maxQueueSize; - final from = switch (_overflow) { - OverflowPolicy.dropOldest => 0, - OverflowPolicy.dropNewest => maxQueueSize, - }; - final dropped = _items.sublist(from, from + excess); - _items.removeRange(from, from + excess); + final dropped = switch (_overflow) { + OverflowPolicy.dropOldest => List.generate( + excess, + (_) => _items.removeFirst(), + ), + // Taken off the back, then put back in the order they arrived. + OverflowPolicy.dropNewest => List.generate( + excess, + (_) => _items.removeLast(), + ).reversed.toList(), + }; _onDrop?.call(dropped); } @@ -306,9 +342,7 @@ class Buffer { final take = (count == null || count > _items.length) ? _items.length : count; - final items = _items.sublist(0, take); - _items.removeRange(0, take); - return items; + return List.generate(take, (_) => _items.removeFirst()); } // Hands the items over, turning whatever `onFlush` returns into a future so diff --git a/test/buffer_test.dart b/test/buffer_test.dart index 548cb6f..cc116ea 100644 --- a/test/buffer_test.dart +++ b/test/buffer_test.dart @@ -827,18 +827,19 @@ void main() { }); }); - test('should answer to flush for every chunk flush itself sent', () { + test('should not enrol a flush in what arrives mid-drain', () { fakeAsync((async) { final blocker = Completer(); var calls = 0; Object? caughtByFlush; Object? caughtByOnError; + var drained = false; final buffered = buffer( (items) { calls++; if (calls == 1) return blocker.future; - throw Exception('the second chunk failed'); + throw Exception('the later batch failed'); }, 32.toDuration(), maxSize: 2, @@ -846,11 +847,12 @@ void main() { ); buffered('a'); - buffered.flush().then((_) {}, onError: (Object e) { + buffered.flush().then((_) => drained = true, onError: (Object e) { caughtByFlush = e; }).ignore(); - // Arriving mid-drain, so the drain picks them up as its second chunk. + // Arriving after the drain took its measure, so these are the + // buffer's own work rather than the caller's. buffered('b'); buffered('c'); @@ -858,9 +860,9 @@ void main() { async.elapse(32.toDuration()); expect(calls, 2); - expect(caughtByFlush, isNotNull, reason: 'the drain sent it'); - expect(caughtByOnError, isNull, - reason: 'so onError has nothing to say'); + expect(drained, isTrue, reason: 'the one item it took went out'); + expect(caughtByFlush, isNull, reason: 'the caller did not send those'); + expect(caughtByOnError, isNotNull, reason: 'so the buffer answers'); }); }); From 1facea36b328cb0c34a0206bd9694c52c29b717b Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 17:12:51 +0200 Subject: [PATCH 05/10] fix: measure a drain against what leaves the buffer, not its own batches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `flush()` counted only the batches it sent itself, but a flush the buffer schedules is started from the completion handler and therefore gets to the queue first. So the drain kept waiting while the scheduler carried its snapshot away, then sent whatever had arrived since — the very thing the snapshot was meant to prevent. Measured with a self-refilling callback, a two-item drain followed the producer through forty flushes, and would not have stopped at all had the producer not. It now counts every item that leaves the buffer, sent or shed or discarded, so the snapshot is satisfied whoever hands it over. Report an `onError` that throws. It ran inside the flush's catch block, so its own failure rode out on the flush future and was consumed by the private handler that keeps the queue moving — silently losing whatever requeue or logging the handler was there to do. It now goes to the zone. Two docs claimed no caller is ever slowed down. A call that reaches `maxSize` hands the batch over itself, so synchronous work in `onFlush` runs before that call returns; the guarantee is the absence of backpressure from a flush already running, not the absence of any work. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 4 +-- lib/src/buffer.dart | 55 ++++++++++++++++++++++++++--------------- test/buffer_test.dart | 57 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 95 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index c2e6a1e..a75388d 100644 --- a/README.md +++ b/README.md @@ -170,7 +170,7 @@ final response = backOff( ### Buffer A _buffered function_ collects the items passed to it and invokes your function **once** with all of them, rather than once per item. Where debounce and throttle keep only the last call's arguments and drop the rest, a buffer keeps every one — so it batches work instead of shedding it. -The buffer is flushed once `wait` has passed since the first item landed in it, or as soon as it holds `maxSize` items, whichever comes first. Only one flush runs at a time: while your function is working, arriving items collect for the next one, which goes out once it has come due *and* the running one has finished — whichever is later. By default nothing is dropped and no caller is ever slowed down — pass `maxQueueSize` to cap the buffer and shed the excess instead. +The buffer is flushed once `wait` has passed since the first item landed in it, or as soon as it holds `maxSize` items, whichever comes first. Only one flush runs at a time: while your function is working, arriving items collect for the next one, which goes out once it has come due *and* the running one has finished — whichever is later. By default nothing is dropped and no call ever waits on a flush already running — pass `maxQueueSize` to cap the buffer and shed the excess instead. A call that reaches `maxSize` hands its batch over itself, so synchronous work in your function runs before that call returns. #### Usage 1. Creating from scratch @@ -180,7 +180,7 @@ final markRead = buffer((ids) { return api.markAllRead(ids); }, const Duration(milliseconds: 500), maxSize: 25); ``` -2. Converting an existing function into buffered function +2. Converting an existing function into a buffered function ```dart Future markAllRead(List ids) => api.markAllRead(ids); diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart index d8b7b1b..d89fc22 100644 --- a/lib/src/buffer.dart +++ b/lib/src/buffer.dart @@ -39,8 +39,10 @@ enum OverflowPolicy { /// flush has finished, whichever is later. A batch that came due while a flush /// was running does not then serve its wait a second time. /// -/// By default nothing is dropped and no caller is ever slowed down: this is -/// not a capacity buffer. `maxSize` caps what any one flush carries, and +/// By default nothing is dropped and no call ever waits on a flush already +/// running: this is not a capacity buffer. A call that reaches `maxSize` does +/// hand its batch over itself, so synchronous work in `onFlush` runs before +/// that call returns. `maxSize` caps what any one flush carries, and /// `maxQueueSize` caps the backlog that builds up behind a slow one, shedding /// the excess per `overflow`. Where `Debounce` and `Throttle` keep only the /// arguments of the last call and discard the rest, a [Buffer] keeps them all. @@ -130,6 +132,10 @@ class Buffer { // served twice over. var _isDue = false; + // Counts every item that has left the buffer, whether it was sent, shed or + // discarded, so a drain can tell when the batch it measured has gone. + var _removed = 0; + /// The number of items waiting to get flushed. /// /// Counts what is still held. Items handed to `onFlush` are gone from here @@ -174,7 +180,7 @@ class Buffer { // just arrived, and enrol this caller in sending them. if (_items.isEmpty) return _inFlight ?? Future.value(); - return _drain(_items.length); + return _drainUntil(_removed + _items.length); } /// Discards the collected items without invoking `onFlush`. @@ -185,28 +191,28 @@ class Buffer { _timer?.cancel(); _timer = null; _isDue = false; + _removed += _items.length; _items.clear(); } - // Hands over [remaining] items, a batch at a time, waiting out anything - // already running first. + // Hands items over a batch at a time until everything the buffer held when + // [target] was measured has left it, waiting out anything already running. // - // Counted rather than draining until empty, so a caller of `flush` is not - // made to send whatever arrives while it waits — under a producer that never - // stops, that future would never complete. - Future _drain(int remaining) { - if (remaining <= 0 || _items.isEmpty) return Future.value(); + // Measured against what has left rather than against its own batches: a + // flush the buffer schedules itself carries part of the same items, and + // since that one is started from the completion handler it gets there first. + // Counting only its own would leave a drain following a steady producer for + // as long as one kept feeding it. + Future _drainUntil(int target) { + if (_removed >= target) return Future.value(); if (_inFlight case final inFlight?) { - return inFlight.then((_) => _drain(remaining)); + return inFlight.then((_) => _drainUntil(target)); } - final sending = switch (_maxSize) { - final maxSize? when maxSize < _items.length => maxSize, - _ => _items.length, - }; + if (_items.isEmpty) return Future.value(); - return _startFlush(report: false).then((_) => _drain(remaining - sending)); + return _startFlush(report: false).then((_) => _drainUntil(target)); } // Shared by `call` and `addAll` so a bulk add costs one pass, and so a @@ -315,6 +321,7 @@ class Buffer { void _dropDownTo(int maxQueueSize) { final excess = _items.length - maxQueueSize; + _removed += excess; final dropped = switch (_overflow) { OverflowPolicy.dropOldest => List.generate( @@ -342,6 +349,7 @@ class Buffer { final take = (count == null || count > _items.length) ? _items.length : count; + _removed += take; return List.generate(take, (_) => _items.removeFirst()); } @@ -356,11 +364,20 @@ class Buffer { try { await _invoke(items); } catch (error, stackTrace) { - if (_onError case final onError?) { - onError(error, stackTrace, items); + final onError = _onError; + if (onError == null) { + Zone.current.handleUncaughtError(error, stackTrace); return; } - Zone.current.handleUncaughtError(error, stackTrace); + + try { + onError(error, stackTrace, items); + } catch (handlerError, handlerStackTrace) { + // Reported rather than swallowed: a handler that fails has taken the + // requeue or the logging down with it, which is worse than the flush + // failing in the first place. + Zone.current.handleUncaughtError(handlerError, handlerStackTrace); + } } } } diff --git a/test/buffer_test.dart b/test/buffer_test.dart index cc116ea..3d3610b 100644 --- a/test/buffer_test.dart +++ b/test/buffer_test.dart @@ -827,6 +827,44 @@ void main() { }); }); + test('should not follow a producer past the batch it measured', () { + fakeAsync((async) { + final blocker = Completer(); + final sent = >[]; + var drained = false; + + final buffered = buffer( + (items) { + sent.add(items); + return sent.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxSize: 2, + ); + + buffered.addAll(['a', 'b']); // goes at once, and blocks + buffered.addAll(['c', 'd']); // what the drain takes the measure of + + buffered.flush().then((_) => drained = true).ignore(); + + // Arrives after that measure, and on its own is not a full batch, so + // only a drain that overreached would send it. + buffered('e'); + + blocker.complete(); + async.flushMicrotasks(); + + // The buffer schedules [c, d] itself from the completion handler, so + // the drain never sends a batch of its own — it is done all the same. + expect(sent, [ + ['a', 'b'], + ['c', 'd'], + ]); + expect(drained, isTrue, reason: 'its batch is gone, whoever sent it'); + expect(buffered.length, 1, reason: 'e is the buffer\'s own work'); + }); + }); + test('should not enrol a flush in what arrives mid-drain', () { fakeAsync((async) { final blocker = Completer(); @@ -1118,6 +1156,25 @@ void main() { }); }); + test('should report an onError that fails itself to the zone', () { + final caught = Completer(); + + return runZonedGuarded(() async { + final buffered = buffer( + (items) async => throw Exception('flush failed'), + Duration.zero, + // Takes the requeue down with it, so it cannot be swallowed. + onError: (e, s, items) => throw StateError('the handler failed too'), + ); + + buffered('a'); + + expect(await caught.future, isA()); + }, (e, s) { + if (!caught.isCompleted) caught.complete(e); + }); + }); + test('should convert an existing function into a buffered one', () { fakeAsync((async) { final flushes = >[]; From dcae6b8e41ffd99b654b4a7aa9ff4026fad96f69 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 17:31:14 +0200 Subject: [PATCH 06/10] fix: hand back what came due behind a drain when it finishes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A batch started by `flush()` deliberately does not pump on the way out, so that its failures stay answerable to the caller who asked for it. But the drain did not pump when it *stopped* either, so anything that came due while a slow batch was running was left with no timer, no flush in flight and nobody to move it: `length` stayed above zero while `isPending` reported false, and ten seconds of elapsed time sent nothing. `_drainUntil` now pumps at its base case, which is the point where ownership goes back to the buffer. The invariant test never caught this, or the two strand bugs before it, because its flush callback was synchronous — nothing could come due while one was running. Half its seeds now flush slowly, which reproduces this failure at seed 26. Co-Authored-By: Claude Opus 5 (1M context) --- lib/src/buffer.dart | 10 +++++++--- test/buffer_test.dart | 42 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart index d89fc22..f81246d 100644 --- a/lib/src/buffer.dart +++ b/lib/src/buffer.dart @@ -204,14 +204,18 @@ class Buffer { // Counting only its own would leave a drain following a steady producer for // as long as one kept feeding it. Future _drainUntil(int target) { - if (_removed >= target) return Future.value(); + if (_removed >= target || _items.isEmpty) { + // Hands back anything that arrived while this was draining. A batch of + // its own does not pump on the way out, to keep its failures answerable + // here, so without this those items would sit with nothing to move them. + _pump(); + return Future.value(); + } if (_inFlight case final inFlight?) { return inFlight.then((_) => _drainUntil(target)); } - if (_items.isEmpty) return Future.value(); - return _startFlush(report: false).then((_) => _drainUntil(target)); } diff --git a/test/buffer_test.dart b/test/buffer_test.dart index 3d3610b..4076645 100644 --- a/test/buffer_test.dart +++ b/test/buffer_test.dart @@ -827,6 +827,39 @@ void main() { }); }); + test('should hand back what came due behind a slow drain', () { + fakeAsync((async) { + final blocker = Completer(); + final sent = >[]; + + final buffered = buffer( + (items) { + sent.add(items); + return sent.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + ); + + buffered('a'); + buffered.flush().ignore(); // takes a, and blocks + + buffered('b'); + async.elapse(50.toDuration()); // b comes due while the drain runs + + blocker.complete(); + async.flushMicrotasks(); + + // The drain is finished, so what it was not measuring has to go back + // under the buffer's own ownership rather than sit unowned. + expect(sent, [ + ['a'], + ['b'], + ]); + expect(buffered.length, 0); + expect(buffered.isPending, isFalse); + }); + }); + test('should not follow a producer past the batch it measured', () { fakeAsync((async) { final blocker = Completer(); @@ -1228,10 +1261,17 @@ void main() { final byMaxSize = rng.nextBool(); final limit = 1 + rng.nextInt(4); + // Half the seeds flush slowly, so items come due while a flush is + // still running. Every strand bug so far has hidden in that window. + final slowFlush = rng.nextBool(); + fakeAsync((async) { final flushes = >[]; final buffered = Buffer( - flushes.add, + (items) { + flushes.add(items); + return slowFlush ? Future.delayed(25.toDuration()) : null; + }, 10.toDuration(), maxSize: byMaxSize ? limit : null, maxQueueSize: byMaxSize ? null : limit, From 3c7cce7a3415ff619debc1e44770637d22cf4db8 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 17:44:41 +0200 Subject: [PATCH 07/10] fix: track the queue by sequence, so shedding cannot fake a drain Counting removals let `dropNewest` satisfy a snapshot it never touched. Measured: with `maxQueueSize: 2` and a flush blocked, queue `[a, b]`, call `flush()`, then add `[c, d]`; shedding `c` and `d` advanced the count to the target and the drain reported success with `[a, b]` still buffered and unsent. `await flush()` lied about the one thing it exists to promise. The two ends of the queue now carry sequence numbers. Sending and shedding the oldest move the head, which is what can settle a snapshot; shedding the newest moves the tail back instead, and so settles nothing taken before those items arrived. `cancel()` moves the head to the tail, since discarding does dispose of what a drain was waiting for. Co-Authored-By: Claude Opus 5 (1M context) --- lib/src/buffer.dart | 43 ++++++++++++++++++++++++++----------------- test/buffer_test.dart | 38 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 17 deletions(-) diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart index f81246d..24f5a4a 100644 --- a/lib/src/buffer.dart +++ b/lib/src/buffer.dart @@ -132,9 +132,12 @@ class Buffer { // served twice over. var _isDue = false; - // Counts every item that has left the buffer, whether it was sent, shed or - // discarded, so a drain can tell when the batch it measured has gone. - var _removed = 0; + // Sequence numbers for the two ends of the queue, so a drain can tell + // whether the items it measured have really gone rather than just counting + // removals. Shedding the newest moves the tail back, which must not satisfy + // a snapshot taken before those items even arrived. + var _head = 0; + var _tail = 0; /// The number of items waiting to get flushed. /// @@ -149,6 +152,7 @@ class Buffer { /// Adds [item] to the buffer, arming the flush if it is the first one in. void call(T item) { _items.add(item); + _tail++; _collect(); } @@ -158,7 +162,9 @@ class Buffer { /// the next only once the one before it has finished. With a `maxQueueSize` /// set, the excess is dropped as one group rather than an item at a time. void addAll(Iterable items) { + final held = _items.length; _items.addAll(items); + _tail += _items.length - held; _collect(); } @@ -180,7 +186,7 @@ class Buffer { // just arrived, and enrol this caller in sending them. if (_items.isEmpty) return _inFlight ?? Future.value(); - return _drainUntil(_removed + _items.length); + return _drainUntil(_tail); } /// Discards the collected items without invoking `onFlush`. @@ -191,7 +197,7 @@ class Buffer { _timer?.cancel(); _timer = null; _isDue = false; - _removed += _items.length; + _head = _tail; _items.clear(); } @@ -204,7 +210,7 @@ class Buffer { // Counting only its own would leave a drain following a steady producer for // as long as one kept feeding it. Future _drainUntil(int target) { - if (_removed >= target || _items.isEmpty) { + if (_head >= target || _items.isEmpty) { // Hands back anything that arrived while this was draining. A batch of // its own does not pump on the way out, to keep its failures answerable // here, so without this those items would sit with nothing to move them. @@ -325,19 +331,22 @@ class Buffer { void _dropDownTo(int maxQueueSize) { final excess = _items.length - maxQueueSize; - _removed += excess; - final dropped = switch (_overflow) { - OverflowPolicy.dropOldest => List.generate( - excess, - (_) => _items.removeFirst(), - ), - // Taken off the back, then put back in the order they arrived. - OverflowPolicy.dropNewest => List.generate( + final List dropped; + switch (_overflow) { + case OverflowPolicy.dropOldest: + // Off the front, so a drain waiting on these can stop waiting. + _head += excess; + dropped = List.generate(excess, (_) => _items.removeFirst()); + case OverflowPolicy.dropNewest: + // Off the back, so these arrived last and cannot belong to a snapshot + // taken before them. Put back in the order they came in. + _tail -= excess; + dropped = List.generate( excess, (_) => _items.removeLast(), - ).reversed.toList(), - }; + ).reversed.toList(); + } _onDrop?.call(dropped); } @@ -353,7 +362,7 @@ class Buffer { final take = (count == null || count > _items.length) ? _items.length : count; - _removed += take; + _head += take; return List.generate(take, (_) => _items.removeFirst()); } diff --git a/test/buffer_test.dart b/test/buffer_test.dart index 4076645..4baf755 100644 --- a/test/buffer_test.dart +++ b/test/buffer_test.dart @@ -827,6 +827,44 @@ void main() { }); }); + test('should not let shedding the newest satisfy a drain', () { + fakeAsync((async) { + final blocker = Completer(); + final sent = >[]; + var drained = false; + + final buffered = buffer( + (items) { + sent.add(items); + return sent.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxQueueSize: 2, + overflow: OverflowPolicy.dropNewest, + ); + + buffered('z'); + async.elapse(32.toDuration()); // the flush of z starts, and blocks + + buffered.addAll(['a', 'b']); + buffered.flush().then((_) => drained = true).ignore(); + + // Over the cap, so these are shed. They arrived after the drain took + // its measure, so shedding them settles nothing it was waiting for. + buffered.addAll(['c', 'd']); + + blocker.complete(); + async.flushMicrotasks(); + + expect(sent, [ + ['z'], + ['a', 'b'], + ]); + expect(drained, isTrue, reason: 'and only once its own items went'); + expect(buffered.length, 0); + }); + }); + test('should hand back what came due behind a slow drain', () { fakeAsync((async) { final blocker = Completer(); From e8a032eecc4fecdd28d088f595eedf08621735e5 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 18:21:52 +0200 Subject: [PATCH 08/10] fix: outlast the carrier, keep a remainder's deadline, guard addAll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three from the same review, all measured before and after. A drain could finish while the flush carrying its batch was still running: the scheduled flush started from the completion handler advances the head to the target, so the drain saw its work as done and returned while `onFlush` was still going. That is precisely the guarantee `dispose` rests on. A running flush now records the sequence it starts from, and a drain waits for it when it overlaps. A remainder lost its deadline once a full batch went ahead of it. With `maxSize: 2` and five items arriving together, the third batch waited a fresh window from when the second finished — sent at t=132 where its items had been due since t=32. The window now belongs to the items rather than to the batch that just left, and survives until the buffer empties. `addAll` walked the iterable straight into the queue, so one that failed part way left those items in, uncounted and unscheduled. It is walked into a list first, and a list is passed through without a copy since it cannot fail part way. Co-Authored-By: Claude Opus 5 (1M context) --- lib/src/buffer.dart | 58 ++++++++++++++++++++------- test/buffer_test.dart | 92 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 136 insertions(+), 14 deletions(-) diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart index 24f5a4a..0da7c86 100644 --- a/lib/src/buffer.dart +++ b/lib/src/buffer.dart @@ -1,6 +1,8 @@ import 'dart:async'; import 'dart:collection'; +import 'package:clock/clock.dart'; + /// Invoked with every item a [Buffer] collected, in the order they arrived. typedef BufferFlushCallback = FutureOr Function(List items); @@ -128,8 +130,16 @@ class Buffer { // Never carries its error, so a failed flush cannot wedge the queue. Future? _inFlight; - // Set when a buffer came due while a flush was running, so the wait is not - // served twice over. + // The sequence of the first item that flush is carrying, so a drain can tell + // whether it is holding part of the batch that drain measured. + var _inFlightFrom = 0; + + // The window of the batch now collecting: when it comes due, and whether + // its timer has already fired. Both outlive a chunk going out, because a + // remainder is as old as the items it is made of and a full batch leaving + // ahead of it must not buy it a fresh window. Cleared together, and only + // once the buffer is empty. + DateTime? _dueAt; var _isDue = false; // Sequence numbers for the two ends of the queue, so a drain can tell @@ -150,11 +160,7 @@ class Buffer { bool get isPending => _timer != null || _inFlight != null; /// Adds [item] to the buffer, arming the flush if it is the first one in. - void call(T item) { - _items.add(item); - _tail++; - _collect(); - } + void call(T item) => addAll([item]); /// Adds every item in [items] to the buffer. /// @@ -162,9 +168,13 @@ class Buffer { /// the next only once the one before it has finished. With a `maxQueueSize` /// set, the excess is dropped as one group rather than an item at a time. void addAll(Iterable items) { - final held = _items.length; - _items.addAll(items); - _tail += _items.length - held; + // Walked before the queue is touched: an iterable that throws part way + // would otherwise leave those items in, uncounted and unscheduled. A list + // cannot fail part way, so it needs no copy of its own. + final incoming = items is List ? items : List.of(items); + + _items.addAll(incoming); + _tail += incoming.length; _collect(); } @@ -196,6 +206,7 @@ class Buffer { void cancel() { _timer?.cancel(); _timer = null; + _dueAt = null; _isDue = false; _head = _tail; _items.clear(); @@ -211,6 +222,12 @@ class Buffer { // as long as one kept feeding it. Future _drainUntil(int target) { if (_head >= target || _items.isEmpty) { + // The batch has left the buffer, but a flush still running may be the + // one carrying it, and this has to outlast that to be worth awaiting. + if (_inFlight case final inFlight? when _inFlightFrom < target) { + return inFlight.then((_) => _drainUntil(target)); + } + // Hands back anything that arrived while this was draining. A batch of // its own does not pump on the way out, to keep its failures answerable // here, so without this those items would sit with nothing to move them. @@ -266,9 +283,12 @@ class Buffer { // Does nothing when the buffer is empty, or when it is already overdue and // only waiting for its turn at the queue. void _armWait() { - if (_items.isEmpty || _isDue) return; + if (_items.isEmpty || _isDue || _timer != null) return; - _timer ??= Timer(_wait, _onDue); + // Timed against the batch's own deadline, which a chunk leaving ahead of + // it does not reset, so a remainder is not given a fresh window. + final dueAt = _dueAt ??= clock.now().add(_wait); + _timer = Timer(dueAt.difference(clock.now()), _onDue); } void _onDue() { @@ -286,6 +306,7 @@ class Buffer { // With `report`, a failure goes to `onError`; without it, the failure is // left on the returned future for whoever asked for the flush. Future _startFlush({required bool report}) { + final from = _head; final items = _take(_maxSize); // A completer rather than the flush itself: `onFlush` is invoked @@ -293,6 +314,7 @@ class Buffer { // after the window this is closing. final settled = Completer(); _inFlight = settled.future; + _inFlightFrom = from; // Timed before the callback runs: `onFlush` may work synchronously for a // while, and a remainder's wait should run from now rather than from @@ -358,12 +380,20 @@ class Buffer { List _take(int? count) { _timer?.cancel(); _timer = null; - _isDue = false; final take = (count == null || count > _items.length) ? _items.length : count; _head += take; - return List.generate(take, (_) => _items.removeFirst()); + final taken = List.generate(take, (_) => _items.removeFirst()); + + // The window belongs to the items, not to the batch that just left it, so + // it is only over once there are none of them. + if (_items.isEmpty) { + _dueAt = null; + _isDue = false; + } + + return taken; } // Hands the items over, turning whatever `onFlush` returns into a future so diff --git a/test/buffer_test.dart b/test/buffer_test.dart index 4baf755..17b92c2 100644 --- a/test/buffer_test.dart +++ b/test/buffer_test.dart @@ -827,6 +827,89 @@ void main() { }); }); + test('should outlast the flush that carries what it measured', () { + fakeAsync((async) { + final first = Completer(); + final second = Completer(); + final started = >[]; + var drained = false; + + final buffered = buffer( + (items) { + started.add(items); + return started.length == 1 ? first.future : second.future; + }, + 32.toDuration(), + maxSize: 2, + ); + + buffered.addAll(['a', 'b']); // goes at once, and blocks + buffered.addAll(['c', 'd']); // the snapshot + buffered.flush().then((_) => drained = true).ignore(); + + // Its completion handler starts [c, d], so the snapshot has left the + // buffer — but into a flush that has not finished. + first.complete(); + async.flushMicrotasks(); + + expect(started, [ + ['a', 'b'], + ['c', 'd'], + ]); + expect(drained, isFalse, reason: 'awaiting it must mean it is sent'); + + second.complete(); + async.flushMicrotasks(); + + expect(drained, isTrue); + }); + }); + + test('should keep a remainder to its deadline across full batches', () { + fakeAsync((async) { + final blocker = Completer(); + final sentAt = []; + + final buffered = buffer( + (items) { + sentAt.add(async.elapsed.inMilliseconds); + return sentAt.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxSize: 2, + ); + + // All five arrive together, so all five are due at t=32. + buffered.addAll([1, 2, 3, 4, 5]); + async.elapse(100.toDuration()); + + blocker.complete(); + async.elapse(500.toDuration()); + + // The odd one out is overdue by the time its turn comes, so it goes + // then rather than starting a window of its own. + expect(sentAt, [0, 100, 100]); + }); + }); + + test('should leave the buffer alone when an iterable fails part way', () { + fakeAsync((async) { + final flushes = >[]; + final buffered = buffer(flushes.add, 32.toDuration()); + + expect( + () => buffered.addAll(_failsAfter(3)), + throwsA(isA()), + ); + + expect(buffered.length, 0, reason: 'nothing half-added'); + expect(buffered.isPending, isFalse); + + async.elapse(128.toDuration()); + expect(flushes, isEmpty); + }); + }); + test('should not let shedding the newest satisfy a drain', () { fakeAsync((async) { final blocker = Completer(); @@ -1456,3 +1539,12 @@ void main() { }); }); } + +// Yields [count] values and then fails, the way a lazy source backed by a +// stream or a database cursor can. +Iterable _failsAfter(int count) sync* { + for (var i = 0; i < count; i++) { + yield i; + } + throw StateError('iteration failed'); +} From 013854e32777dc2aa8133367da3810234a46cf4e Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 18:39:50 +0200 Subject: [PATCH 09/10] refactor: let the queue carry what the flags were tracking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every bug this class has had came from state that had to be kept in step by hand: a deadline that must survive a batch but not an empty buffer, two counters where only one end may satisfy a drain, a future and the range it carries held apart. Reading it required holding all of that at once, which is why the same class of bug kept reappearing on a different path each time. Each item now carries the sequence it arrived at and the moment its batch comes due, so the queue answers both awkward questions itself. Taking from the front moves the oldest sequence on; shedding from the back cannot, which is what a drain relies on and is now impossible to get wrong. A remainder keeps the deadline of the items it is made of because those items are still holding it. `_take` and `_dropDownTo`, where the last three bugs lived, no longer touch scheduling state at all. The running flush and the sequence it starts from became one record, since two fields that must agree cannot if there is only one. `isPending` reads the queue rather than the timer, which makes the invariant the property test checks — items held means something will move them — true by construction rather than by maintenance. Eight mutable fields become five, and the two hardest invariants stop being invariants. All 98 tests pass untouched, and every failure found in review re-checked against the rewrite. Draining 80k items in batches of ten measures as before, so carrying the metadata per item costs nothing worth having back. Co-Authored-By: Claude Opus 5 (1M context) --- lib/src/buffer.dart | 190 +++++++++++++++++++++----------------------- 1 file changed, 91 insertions(+), 99 deletions(-) diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart index 0da7c86..eaf544b 100644 --- a/lib/src/buffer.dart +++ b/lib/src/buffer.dart @@ -120,44 +120,39 @@ class Buffer { final BufferErrorCallback? _onError; final BufferDropCallback? _onDrop; - // A queue rather than a list, so taking a batch off the front and shedding - // the oldest are both cheap. Draining N items through a list would move - // O(N^2) elements, since every removal shifts everything behind it. - final _items = ListQueue(); + // A queue, so taking a batch off the front and shedding from either end are + // all cheap. Draining N items through a list would move O(N^2) elements, + // since every removal from the front shifts everything behind it. + // + // Each item carries when it arrived and when its batch comes due, which is + // what lets the queue answer both of the awkward questions on its own: + // taking from the front moves the oldest sequence on, shedding from the back + // cannot, and a remainder keeps the deadline of the items it is made of + // rather than being given a fresh one. + final _pending = ListQueue<_Pending>(); + var _nextSeq = 0; + Timer? _timer; - // Settles when the flush running right now finishes, however it finishes. - // Never carries its error, so a failed flush cannot wedge the queue. - Future? _inFlight; - - // The sequence of the first item that flush is carrying, so a drain can tell - // whether it is holding part of the batch that drain measured. - var _inFlightFrom = 0; - - // The window of the batch now collecting: when it comes due, and whether - // its timer has already fired. Both outlive a chunk going out, because a - // remainder is as old as the items it is made of and a full batch leaving - // ahead of it must not buy it a fresh window. Cleared together, and only - // once the buffer is empty. - DateTime? _dueAt; - var _isDue = false; - - // Sequence numbers for the two ends of the queue, so a drain can tell - // whether the items it measured have really gone rather than just counting - // removals. Shedding the newest moves the tail back, which must not satisfy - // a snapshot taken before those items even arrived. - var _head = 0; - var _tail = 0; + // Set when the wait for the oldest item has run out, and cleared only once + // the buffer is empty. A batch leaving does not un-expire the wait of what + // is left behind, because those items are every bit as old. + var _waitIsUp = false; + + // The flush running right now, and the sequence it starts from, so a drain + // can tell whether it is the one carrying what that drain measured. One + // field, because the two can never disagree if there is only one of them. + _Running? _running; /// The number of items waiting to get flushed. /// /// Counts what is still held. Items handed to `onFlush` are gone from here /// even while that flush is running. - int get length => _items.length; + int get length => _pending.length; /// True if there are items waiting to get flushed, or a flush is still /// running. - bool get isPending => _timer != null || _inFlight != null; + bool get isPending => _pending.isNotEmpty || _running != null; /// Adds [item] to the buffer, arming the flush if it is the first one in. void call(T item) => addAll([item]); @@ -173,8 +168,11 @@ class Buffer { // cannot fail part way, so it needs no copy of its own. final incoming = items is List ? items : List.of(items); - _items.addAll(incoming); - _tail += incoming.length; + final dueAt = clock.now().add(_wait); + for (final item in incoming) { + _pending.add((item: item, seq: _nextSeq++, dueAt: dueAt)); + } + _collect(); } @@ -194,9 +192,9 @@ class Buffer { // Already empty, so the flush running is all there is left to wait for. // Draining past it would cut short the window of items that have only // just arrived, and enrol this caller in sending them. - if (_items.isEmpty) return _inFlight ?? Future.value(); + if (_pending.isEmpty) return _running?.settled ?? Future.value(); - return _drainUntil(_tail); + return _drainThrough(_pending.last.seq); } /// Discards the collected items without invoking `onFlush`. @@ -206,26 +204,23 @@ class Buffer { void cancel() { _timer?.cancel(); _timer = null; - _dueAt = null; - _isDue = false; - _head = _tail; - _items.clear(); + _waitIsUp = false; + _pending.clear(); } - // Hands items over a batch at a time until everything the buffer held when - // [target] was measured has left it, waiting out anything already running. + // Hands batches over until the item that arrived at [through] has left the + // buffer, and the flush carrying it has finished. // - // Measured against what has left rather than against its own batches: a - // flush the buffer schedules itself carries part of the same items, and - // since that one is started from the completion handler it gets there first. - // Counting only its own would leave a drain following a steady producer for - // as long as one kept feeding it. - Future _drainUntil(int target) { - if (_head >= target || _items.isEmpty) { - // The batch has left the buffer, but a flush still running may be the - // one carrying it, and this has to outlast that to be worth awaiting. - if (_inFlight case final inFlight? when _inFlightFrom < target) { - return inFlight.then((_) => _drainUntil(target)); + // Measured by sequence rather than by counting batches, because the buffer + // schedules flushes of its own from a completion handler and so gets to the + // queue first: counting only its own batches would leave a drain following + // a steady producer for as long as one kept feeding it. + Future _drainThrough(int through) { + if (_pending.isEmpty || _pending.first.seq > through) { + // Gone from the buffer, but a flush still running may be the one + // carrying it, and this has to outlast that to be worth awaiting. + if (_running case final running? when running.from <= through) { + return running.settled.then((_) => _drainThrough(through)); } // Hands back anything that arrived while this was draining. A batch of @@ -235,11 +230,11 @@ class Buffer { return Future.value(); } - if (_inFlight case final inFlight?) { - return inFlight.then((_) => _drainUntil(target)); + if (_running case final running?) { + return running.settled.then((_) => _drainThrough(through)); } - return _startFlush(report: false).then((_) => _drainUntil(target)); + return _startFlush(report: false).then((_) => _drainThrough(through)); } // Shared by `call` and `addAll` so a bulk add costs one pass, and so a @@ -250,7 +245,7 @@ class Buffer { _pump(); if (_maxQueueSize case final maxQueueSize? - when _items.length > maxQueueSize) { + when _pending.length > maxQueueSize) { _dropDownTo(maxQueueSize); // The backlog shrank, so whatever is left of it needs its own wait. @@ -259,17 +254,19 @@ class Buffer { } bool get _isFull { - if (_maxSize case final maxSize?) return _items.length >= maxSize; + if (_maxSize case final maxSize?) return _pending.length >= maxSize; return false; } // Starts a flush if one is due and none is running, otherwise arms the wait. - // Re-entered when a flush settles, so the buffer drains one flush at a time. + // Re-entered when a flush settles, so the buffer drains one at a time. void _pump() { + if (_pending.isEmpty) return; + // One at a time. Whoever is flushing pumps again on the way out, so a // batch that came due meanwhile goes then rather than waiting afresh. - if (_inFlight == null && (_isDue || _isFull)) { - // Arms the remainder itself, before the callback gets a chance to run. + if (_running == null && (_waitIsUp || _isFull)) { + // Arms whatever is left over itself, before the callback can run. _startFlush(report: true); return; } @@ -277,23 +274,21 @@ class Buffer { _armWait(); } - // Starts the wait for whatever is held, so it runs from when those items - // arrived rather than from whenever the flush ahead of them finishes. + // Waits out the oldest item's deadline, which is carried by the item rather + // than by the batch, so what is left behind is not given a fresh window. // - // Does nothing when the buffer is empty, or when it is already overdue and - // only waiting for its turn at the queue. + // Nothing to arm if the buffer is empty, if a wait is already running, or if + // one has already run out and is only waiting its turn at the queue. void _armWait() { - if (_items.isEmpty || _isDue || _timer != null) return; + if (_pending.isEmpty || _timer != null || _waitIsUp) return; - // Timed against the batch's own deadline, which a chunk leaving ahead of - // it does not reset, so a remainder is not given a fresh window. - final dueAt = _dueAt ??= clock.now().add(_wait); - _timer = Timer(dueAt.difference(clock.now()), _onDue); + final due = _pending.first.dueAt.difference(clock.now()); + _timer = Timer(due, _onDue); } void _onDue() { _timer = null; - _isDue = true; + _waitIsUp = true; _pump(); } @@ -306,25 +301,23 @@ class Buffer { // With `report`, a failure goes to `onError`; without it, the failure is // left on the returned future for whoever asked for the flush. Future _startFlush({required bool report}) { - final from = _head; + final from = _pending.first.seq; final items = _take(_maxSize); // A completer rather than the flush itself: `onFlush` is invoked // synchronously below, so there is no future to claim the queue with until // after the window this is closing. final settled = Completer(); - _inFlight = settled.future; - _inFlightFrom = from; + _running = (settled: settled.future, from: from); // Timed before the callback runs: `onFlush` may work synchronously for a - // while, and a remainder's wait should run from now rather than from - // whenever that work returns. + // while, and a remainder should not be waiting on that work to return. _armWait(); final flushing = report ? _invokeAndReport(items) : _invoke(items); void release() { - _inFlight = null; + _running = null; settled.complete(); } @@ -335,7 +328,7 @@ class Buffer { release(); // Only the scheduled path pumps on the way out. An explicit flush - // drives its own drain, which is what keeps every chunk it sends + // drives its own drain, which is what keeps every batch it sends // answerable to its caller. if (report) _pump(); }, @@ -352,46 +345,38 @@ class Buffer { } void _dropDownTo(int maxQueueSize) { - final excess = _items.length - maxQueueSize; + final excess = _pending.length - maxQueueSize; - final List dropped; + final List<_Pending> shed; switch (_overflow) { case OverflowPolicy.dropOldest: - // Off the front, so a drain waiting on these can stop waiting. - _head += excess; - dropped = List.generate(excess, (_) => _items.removeFirst()); + shed = List.generate(excess, (_) => _pending.removeFirst()); case OverflowPolicy.dropNewest: - // Off the back, so these arrived last and cannot belong to a snapshot - // taken before them. Put back in the order they came in. - _tail -= excess; - dropped = List.generate( + // Off the back, so these arrived last, which is why shedding them can + // never settle a drain measured before they turned up. Handed over in + // the order they came in. + shed = List.generate( excess, - (_) => _items.removeLast(), + (_) => _pending.removeLast(), ).reversed.toList(); } - _onDrop?.call(dropped); + _onDrop?.call([for (final pending in shed) pending.item]); } - // Takes up to `count` items out of the buffer, disarming the wait. - // - // Taken before `onFlush` is invoked, so anything added while it runs - // collects into the next buffer instead of joining this one. + // Takes up to `count` items off the front, disarming the wait. Taken before + // `onFlush` is invoked, so anything added while it runs collects into the + // next batch instead of joining this one. List _take(int? count) { _timer?.cancel(); _timer = null; final take = - (count == null || count > _items.length) ? _items.length : count; - _head += take; - final taken = List.generate(take, (_) => _items.removeFirst()); - - // The window belongs to the items, not to the batch that just left it, so - // it is only over once there are none of them. - if (_items.isEmpty) { - _dueAt = null; - _isDue = false; - } + (count == null || count > _pending.length) ? _pending.length : count; + final taken = List.generate(take, (_) => _pending.removeFirst().item); + + // No items left to be waiting for, so the next batch starts a fresh wait. + if (_pending.isEmpty) _waitIsUp = false; return taken; } @@ -424,3 +409,10 @@ class Buffer { } } } + +// One item, with the sequence it arrived at and the moment its batch is due. +typedef _Pending = ({T item, int seq, DateTime dueAt}); + +// A flush in progress: a future that settles either way, and the sequence of +// the first item it is carrying. +typedef _Running = ({Future settled, int from}); From d0554c6e75bd56b70a14ce744970be6ddca63b97 Mon Sep 17 00:00:00 2001 From: Sahil Kumar Date: Wed, 2 Sep 2026 18:57:00 +0200 Subject: [PATCH 10/10] fix: hand the wait to whoever is at the front of the queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Carrying the deadline on the items fixed which deadline applies, but the timer and the expired-wait flag were still left over from whichever item used to be at the front. Two ways that showed: Shedding the oldest left its timer armed. Add `a`, wait half its window, then overflow with `b, c`: measured at t=48 for a 32ms window, they went at t=32 on a timer belonging to an item that had already been dropped. Taking a batch left the flag set. With a flush running, let `1` come due, then add `2, 3`: `[1, 2]` rightly goes as soon as the queue frees, but `3` inherited a wait that had run out for `1` and went at t=72 instead of its own t=104. Both are now one rule: any change at the front rewinds the wait, cancels the timer and re-reads whether the new head is overdue. `addAll` builds its entries before touching the queue or the sequence, so the atomicity it promises holds for any iterable rather than only the ones that are not lists. Shedding hands the items back for `onDrop` to be called once scheduling is settled, since that callback may throw. The invariant test asserted `isPending` where items are held, which became true by definition when `isPending` started reading the queue — it would have passed with the scheduling removed outright. It now stops feeding the buffer and requires it to drain, which fails at seed 3 against exactly that mutation. Co-Authored-By: Claude Opus 5 (1M context) --- lib/src/buffer.dart | 50 +++++++++++++++++---------- test/buffer_test.dart | 79 +++++++++++++++++++++++++++++++++++++------ 2 files changed, 101 insertions(+), 28 deletions(-) diff --git a/lib/src/buffer.dart b/lib/src/buffer.dart index eaf544b..154bb22 100644 --- a/lib/src/buffer.dart +++ b/lib/src/buffer.dart @@ -163,16 +163,18 @@ class Buffer { /// the next only once the one before it has finished. With a `maxQueueSize` /// set, the excess is dropped as one group rather than an item at a time. void addAll(Iterable items) { - // Walked before the queue is touched: an iterable that throws part way - // would otherwise leave those items in, uncounted and unscheduled. A list - // cannot fail part way, so it needs no copy of its own. - final incoming = items is List ? items : List.of(items); - final dueAt = clock.now().add(_wait); - for (final item in incoming) { - _pending.add((item: item, seq: _nextSeq++, dueAt: dueAt)); + + // Built before the queue is touched, and before the sequence moves on: + // any iterable can fail part way through, and one that does must not + // leave items behind uncounted and unscheduled. + final incoming = <_Pending>[]; + for (final item in items) { + incoming.add((item: item, seq: _nextSeq + incoming.length, dueAt: dueAt)); } + _nextSeq += incoming.length; + _pending.addAll(incoming); _collect(); } @@ -246,10 +248,13 @@ class Buffer { if (_maxQueueSize case final maxQueueSize? when _pending.length > maxQueueSize) { - _dropDownTo(maxQueueSize); + final shed = _shedDownTo(maxQueueSize); - // The backlog shrank, so whatever is left of it needs its own wait. + // The backlog shrank, so whatever is left of it needs its own wait — + // settled before `onDrop`, which is free to throw. _pump(); + + _onDrop?.call(shed); } } @@ -274,6 +279,16 @@ class Buffer { _armWait(); } + // The wait belongs to whoever is at the front, so a head that has left takes + // its wait with it: the timer was armed for its deadline, and a wait that + // ran out for it says nothing about an item that arrived later. + void _rewindWait() { + _timer?.cancel(); + _timer = null; + _waitIsUp = + _pending.isNotEmpty && !clock.now().isBefore(_pending.first.dueAt); + } + // Waits out the oldest item's deadline, which is carried by the item rather // than by the batch, so what is left behind is not given a fresh window. // @@ -344,39 +359,38 @@ class Buffer { return flushing; } - void _dropDownTo(int maxQueueSize) { + // Sheds down to [maxQueueSize] and hands back what went, for the caller to + // report once scheduling has been settled. + List _shedDownTo(int maxQueueSize) { final excess = _pending.length - maxQueueSize; final List<_Pending> shed; switch (_overflow) { case OverflowPolicy.dropOldest: shed = List.generate(excess, (_) => _pending.removeFirst()); + _rewindWait(); case OverflowPolicy.dropNewest: // Off the back, so these arrived last, which is why shedding them can - // never settle a drain measured before they turned up. Handed over in - // the order they came in. + // never settle a drain measured before they turned up, and why the + // wait at the front is untouched. Handed over in arrival order. shed = List.generate( excess, (_) => _pending.removeLast(), ).reversed.toList(); } - _onDrop?.call([for (final pending in shed) pending.item]); + return [for (final pending in shed) pending.item]; } // Takes up to `count` items off the front, disarming the wait. Taken before // `onFlush` is invoked, so anything added while it runs collects into the // next batch instead of joining this one. List _take(int? count) { - _timer?.cancel(); - _timer = null; - final take = (count == null || count > _pending.length) ? _pending.length : count; final taken = List.generate(take, (_) => _pending.removeFirst().item); - // No items left to be waiting for, so the next batch starts a fresh wait. - if (_pending.isEmpty) _waitIsUp = false; + _rewindWait(); return taken; } diff --git a/test/buffer_test.dart b/test/buffer_test.dart index 17b92c2..903662a 100644 --- a/test/buffer_test.dart +++ b/test/buffer_test.dart @@ -865,6 +865,58 @@ void main() { }); }); + test('should not let a shed item take the wait of what replaced it', () { + fakeAsync((async) { + final sentAt = []; + + final buffered = buffer( + (items) => sentAt.add(async.elapsed.inMilliseconds), + 32.toDuration(), + maxQueueSize: 2, + ); + + buffered('a'); // due at t=32 + async.elapse(16.toDuration()); + + // Sheds 'a', so the wait armed for it belongs to nobody. What is left + // arrived at t=16 and is due at t=48. + buffered.addAll(['b', 'c']); + async.elapse(200.toDuration()); + + expect(sentAt, [48]); + }); + }); + + test('should not hand a later item an expired wait when batching', () { + fakeAsync((async) { + final blocker = Completer(); + final sentAt = []; + + final buffered = buffer( + (items) { + sentAt.add(async.elapsed.inMilliseconds); + return sentAt.length == 1 ? blocker.future : null; + }, + 32.toDuration(), + maxSize: 2, + ); + + buffered(0); + async.elapse(32.toDuration()); // sends [0], and blocks + + buffered(1); // due at t=64 + async.elapse(40.toDuration()); // t=72, so 1 is overdue + + buffered.addAll([2, 3]); // due at t=104 + blocker.complete(); + async.elapse(500.toDuration()); + + // [1, 2] goes as soon as the queue frees, since 1 was overdue. 3 was + // not, and must not inherit the wait that ran out for 1. + expect(sentAt, [32, 72, 104]); + }); + }); + test('should keep a remainder to its deadline across full batches', () { fakeAsync((async) { final blocker = Completer(); @@ -1374,9 +1426,10 @@ void main() { }); test('should hold its invariants whatever order it is driven in', () { - // Two of these are relied on by the implementation: `isPending` reads - // the timer rather than counting items, and `flush` hands over the whole - // buffer without re-checking `maxSize`. + // What this is really for: whatever order the buffer is driven in, it + // drains once you stop feeding it, and it never hands over more than it + // was told to. Asserting `isPending` here would prove nothing, since it + // reads the queue and so is true by definition wherever items are held. for (var seed = 0; seed < 200; seed++) { final rng = Random(seed); final byMaxSize = rng.nextBool(); @@ -1400,13 +1453,7 @@ void main() { void checkInvariants(String op) { final where = 'seed $seed, after $op'; - if (buffered.length > 0) { - expect( - buffered.isPending, - isTrue, - reason: '$where: buffered items with nothing to move them', - ); - } + // `maxSize` bounds each flush, not the buffer: items pile up // behind a running flush. Only `maxQueueSize` caps what is held. if (!byMaxSize) { @@ -1439,6 +1486,18 @@ void main() { } } + // Nothing more goes in, so everything held has to find its way out. + // A buffer that never armed a timer, or that dropped a wait when the + // head moved, is stuck here rather than merely late. + async.elapse(const Duration(minutes: 1)); + + expect( + buffered.length, + 0, + reason: 'seed $seed: stopped feeding it and it never drained', + ); + expect(buffered.isPending, isFalse, reason: 'seed $seed'); + for (final flush in flushes) { expect(flush, isNotEmpty, reason: 'seed $seed: flushed nothing'); if (byMaxSize) {