feat: add Buffer, a strategy that batches calls instead of dropping them by xsahil03x · Pull Request #11 · GetStream/rate_limiter

xsahil03x · 2026-09-02T13:55:26Z

Closes #6.

What

Buffer collects the items passed to it and invokes your function once with all of them, instead of once per item. Debounce and Throttle keep only the last call's arguments and drop the rest; a buffer keeps every one.

final markRead = buffer<String>(
  (ids) => api.markAllRead(ids),
  const Duration(milliseconds: 500),
  maxSize: 25,
);

void onMessageSeen(String id) => markRead(id);

Twenty calls as the user scrolls, one request, nothing lost. Debouncing the same thing would send the last id and discard the other nineteen.

Ships with buffered() on functions taking a List and a top-level buffer(), matching how Debounce and Throttle are exposed.

Why these semantics

wait runs from the buffer opening, not as a quiet period, so a steady producer still drains on schedule. That is what buffer means in RxJS, Reactor and Akka, and it removes the need for a second duration to bound the wait.
One flush at a time. Otherwise a slow or retrying onFlush stacks concurrent requests against an endpoint that is already struggling, and await flush() in dispose returns while work is still outstanding.
maxSize caps what one flush carries; maxQueueSize caps the backlog behind a slow one. Unbounded by default, so nothing is dropped unless asked for, and onDrop reports it when it is.
Whichever call starts a flush owns its failure. One started by flush() lands on the returned future; one the buffer scheduled itself goes to onError along with its items, so they can be re-queued.

Verification

88 tests. dart analyze and dart format --set-exit-if-changed are clean on the package and the example, and lib/src/buffer.dart is also clean under stream-core-flutter's full lint config.

buffer_test.dart covers collecting, both caps and overflow policies, cancellation, and the one-flush-at-a-time behaviour. buffer_backoff_test.dart covers backOff as the flush action. A property test drives 200 seeded random operation sequences, asserting after each one that buffered items always have something scheduled to move them and that no flush ever exceeds maxSize.

🤖 Generated with Claude Code

`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) <noreply@anthropic.com>

codecov · 2026-09-02T13:57:11Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 100.00%. Comparing base (f48e3cf) to head (d0554c6).

Additional details and impacted files

@@            Coverage Diff             @@
##            master       #11    +/-   ##
==========================================
  Coverage   100.00%   100.00%            
==========================================
  Files            4         5     +1     
  Lines          100       205   +105     
==========================================
+ Hits           100       205   +105

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

Copilot

🟡 Changes recommended

Explicit flush failures can strand queued items, and combined capacity limits can drop an immediately dispatchable batch.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds a buffered rate-limiting strategy that batches calls while supporting size limits, overflow policies, errors, cancellation, and backoff.

Changes:

Adds Buffer, buffer(), and buffered().
Documents and demonstrates buffering.
Adds comprehensive behavior and backoff tests.

File summaries

File	Description
`lib/src/buffer.dart`	Implements buffering.
`lib/src/extension.dart`	Adds buffer APIs.
`lib/rate_limiter.dart`	Exports Buffer.
`test/buffer_test.dart`	Tests core behavior.
`test/buffer_backoff_test.dart`	Tests backoff integration.
`README.md`	Documents usage.
`example/lib/main.dart`	Demonstrates batching.
`CHANGELOG.md`	Records the feature.

Review details

Files reviewed: 8/8 changed files
Comments generated: 4
Review effort level: Balanced

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

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) <noreply@anthropic.com>

Copilot

🟡 Changes recommended

Invalid production limits and lost deadlines can cause infinite flushing or delayed batches.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Files reviewed: 8/8 changed files
Comments generated: 2
Review effort level: Balanced

`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) <noreply@anthropic.com>

Copilot

🟡 Changes recommended

Remainder deadlines can be delayed by synchronous flush work, and queue draining has quadratic scaling.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (3)

Previously missed (3) — in code that hasn't changed since the last review.

lib/src/buffer.dart:118

A growable List makes this FIFO quadratic for large backlogs: every chunk later removed by _take uses removeRange(0, take), shifting all remaining items, and dropOldest similarly shifts the capped queue on each overflow. With an unbounded default queue and a small maxSize, draining N items can move O(N²) elements. Use a queue/deque or a head index with periodic compaction so front removal is amortized O(1).
README.md:268
This repeats the snapshot guarantee, but a nonempty flush() recursively drains items that arrive while earlier chunks are in flight. It can therefore wait beyond the items held “right now,” and may not complete under continuous input; the usage guidance should expose that behavior.
lib/src/buffer.dart:156
This describes a snapshot drain, but the recursive call in flush() checks the live queue after each chunk and therefore includes items added after flush() was called. Under a continuous producer, that future may never complete. Document the drain-until-empty behavior so callers do not rely on the stated snapshot boundary.

Files reviewed: 8/8 changed files
Comments generated: 1
Review effort level: Balanced

`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) <noreply@anthropic.com>

Copilot

🟡 Changes recommended

Flush snapshot accounting can consume later arrivals indefinitely, and error-handler failures can be silently lost.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (3)

Previously missed (3) — in code that hasn't changed since the last review.

lib/src/buffer.dart:362

If onError itself throws, _invokeAndReport completes with that new error, but _startFlush consumes it in its private error handler followed by .ignore(). The handler failure is therefore silently lost, which can hide a failed requeue or recovery action. Report callback failures through the current zone while allowing the queue to continue.
README.md:183
Add the missing article: “into a buffered function.”
lib/src/buffer.dart:46
The “no caller is ever slowed down” guarantee is inaccurate because reaching maxSize invokes onFlush synchronously in the triggering call; any synchronous work before it returns a future blocks that caller. Describe the actual lack of backpressure from an already in-flight flush instead.

Files reviewed: 8/8 changed files
Comments generated: 2
Review effort level: Balanced

`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) <noreply@anthropic.com>

Copilot

🟡 Changes recommended

Flush snapshot accounting and scheduling contain unresolved correctness bugs.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (2)

lib/src/buffer.dart:215

A drain-started flush always takes up to _maxSize, even when fewer items remain from the snapshot. For example, with maxSize: 2, call flush() on [a, b, c], block [a, b], then append d; the next chunk is [c, d], so the explicit flush sends a post-snapshot item and owns its failure. Limit this call to min(_maxSize, target - _removed) so the documented snapshot boundary is preserved.

    return _startFlush(report: false).then((_) => _drainUntil(target));

lib/src/buffer.dart:324

_removed acts as a FIFO snapshot watermark, but this increment also counts dropNewest removals from the tail. If flush() is waiting on an in-flight call, later arrivals dropped by dropNewest can advance the target while the original snapshot items remain queued, causing flush() to complete before those items are handed over. Only removals from the queued prefix should advance drain progress, or the snapshot needs per-item sequence tracking.

    _removed += excess;

Files reviewed: 8/8 changed files
Comments generated: 1
Review effort level: Balanced

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) <noreply@anthropic.com>

Copilot

🟡 Changes recommended

Drop-newest overflow can incorrectly complete a flush while its snapshotted items remain buffered.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Files reviewed: 8/8 changed files
Comments generated: 1
Review effort level: Balanced

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) <noreply@anthropic.com>

Copilot

🟡 Changes recommended

Flush completion, deadline preservation, and partial iterable failures can currently produce incorrect behavior.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

lib/src/buffer.dart:361

Clearing _isDue for every chunk loses the deadline of a partial remainder after multiple full chunks. With maxSize: 2, enqueue [1,2,3,4,5] together and let the first flush run past wait: the original timer makes [3,4] run immediately, but this reset causes [5]—which arrived at the same time and is already overdue—to wait a fresh full interval. Preserve the first remaining item's original deadline across each chunk (which likely requires per-batch timing metadata).

Files reviewed: 8/8 changed files
Comments generated: 2
Review effort level: Balanced

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) <noreply@anthropic.com>

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) <noreply@anthropic.com>

Copilot

🟡 Changes recommended

Queue deadline tracking has correctness issues, and one claimed scheduling invariant test is tautological.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

test/buffer_test.dart:1407

This assertion cannot verify that buffered items have work scheduled: Buffer.isPending is defined as _pending.isNotEmpty || _running != null (lib/src/buffer.dart:155), so it is unconditionally true inside length > 0. The property test would still pass if timer/pump scheduling were removed entirely. Exercise eventual progress with fake time or add a test-visible scheduling invariant instead.

Files reviewed: 8/8 changed files
Comments generated: 3
Review effort level: Balanced

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) <noreply@anthropic.com>

xsahil03x requested a balanced review from Copilot September 2, 2026 14:00

Copilot started reviewing on behalf of xsahil03x September 2, 2026 14:00 View session

Copilot AI reviewed Sep 2, 2026

View reviewed changes

Comment thread lib/src/buffer.dart

Comment thread lib/src/buffer.dart Outdated

Comment thread README.md Outdated

Comment thread lib/src/buffer.dart Outdated

xsahil03x requested a balanced review from Copilot September 2, 2026 14:19

Copilot started reviewing on behalf of xsahil03x September 2, 2026 14:19 View session

Copilot AI reviewed Sep 2, 2026

View reviewed changes

Comment thread lib/src/buffer.dart Outdated

Comment thread lib/src/buffer.dart Outdated

xsahil03x requested a balanced review from Copilot September 2, 2026 14:34

Copilot AI reviewed Sep 2, 2026

View reviewed changes

Comment thread lib/src/buffer.dart

xsahil03x requested a balanced review from Copilot September 2, 2026 15:02

Copilot started reviewing on behalf of xsahil03x September 2, 2026 15:03 View session

Copilot AI reviewed Sep 2, 2026

View reviewed changes

Comment thread lib/src/buffer.dart Outdated

Comment thread README.md Outdated

xsahil03x requested a balanced review from Copilot September 2, 2026 15:22

Copilot started reviewing on behalf of xsahil03x September 2, 2026 15:22 View session

Copilot AI reviewed Sep 2, 2026

View reviewed changes

Comment thread lib/src/buffer.dart Outdated

xsahil03x requested a balanced review from Copilot September 2, 2026 15:36

Copilot started reviewing on behalf of xsahil03x September 2, 2026 15:37 View session

Copilot AI reviewed Sep 2, 2026

View reviewed changes

Comment thread lib/src/buffer.dart Outdated

xsahil03x requested a balanced review from Copilot September 2, 2026 16:01

Copilot started reviewing on behalf of xsahil03x September 2, 2026 16:01 View session

Copilot AI reviewed Sep 2, 2026

View reviewed changes

Comment thread lib/src/buffer.dart Outdated

Comment thread lib/src/buffer.dart Outdated

xsahil03x and others added 2 commits September 2, 2026 18:21

xsahil03x requested a balanced review from Copilot September 2, 2026 16:43

Copilot started reviewing on behalf of xsahil03x September 2, 2026 16:43 View session

Copilot AI reviewed Sep 2, 2026

View reviewed changes

Comment thread lib/src/buffer.dart Outdated

Comment thread lib/src/buffer.dart Outdated

Comment thread lib/src/buffer.dart Outdated

Uh oh!

Conversation

xsahil03x commented Sep 2, 2026

What

Why these semantics

Verification

Uh oh!

codecov Bot commented Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Suppressed comments (3)

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Suppressed comments (3)

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Suppressed comments (2)

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Suppressed comments (1)

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Suppressed comments (1)

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

codecov Bot commented Sep 2, 2026 •

edited

Loading