Rate-limited batch import

FxDart wins async

Requirement

Push nine ledger transactions (in the code below) to an import endpoint that accepts batches of three, one call at a time — strictly sequential, never overlapping. After each batch, record its size, its amount, and the running total imported so far; print the batch summaries in order, then prove the rate limit held via the max-in-flight counter (it must read 1).

In FxDart the whole policy is the chain: chunk(3) sets the batch size, concurrent(1) sets the pace, and scan threads the running total through the acknowledgments (drop(1) discards the scan seed). The endpoint itself simulates latency with delay and sums its batch with sumBy.

Expected output

importing 9 txns in batches of 3, one at a time:
  batch 1: 3 txns, $172.49 — running total $172.49
  batch 2: 3 txns, $403.65 — running total $576.14
  batch 3: 3 txns, $325.34 — running total $901.48
max batches in flight: 1

Side by side

Native Dart

import 'package:collection/collection.dart';

class Txn {
  final String id;
  final double amount;
  const Txn(this.id, this.amount);
}

const txns = [
  Txn('t-01', 120.00), Txn('t-02', 42.50), Txn('t-03', 9.99),
  Txn('t-04', 310.25), Txn('t-05', 75.00), Txn('t-06', 18.40),
  Txn('t-07', 220.10), Txn('t-08', 5.25), Txn('t-09', 99.99),
];

int inFlight = 0;
int maxInFlight = 0;

/// The rate-limited import endpoint: accepts one batch per call.
Future<(int, double)> importBatch(List<Txn> batch) async {
  inFlight++;
  if (inFlight > maxInFlight) maxInFlight = inFlight;
  await Future.delayed(const Duration(milliseconds: 15));
  final ack = (batch.length, batch.fold(0.0, (sum, t) => sum + t.amount));
  inFlight--;
  return ack;
}

Future<void> main() async {
  final report = <(int, int, double, double)>[];
  var running = 0.0;
  var n = 0;
  for (final batch in txns.slices(3)) {
    final (count, amount) = await importBatch(batch);
    n++;
    running += amount;
    report.add((n, count, amount, running));
  }
  print('importing ${txns.length} txns in batches of 3, one at a time:');
  for (final (n, count, amount, running) in report) {
    print('  batch $n: $count txns, \$${amount.toStringAsFixed(2)} '
        '— running total \$${running.toStringAsFixed(2)}');
  }
  print('max batches in flight: $maxInFlight');
}

FxDart

import 'package:fxdart/fxdart.dart';

class Txn {
  final String id;
  final double amount;
  const Txn(this.id, this.amount);
}

int inFlight = 0;
int maxInFlight = 0;

/// The rate-limited import endpoint: accepts one batch per call.
Future<(int, double)> importBatch(List<Txn> batch) async {
  inFlight++;
  if (inFlight > maxInFlight) maxInFlight = inFlight;
  final ack = await delay(const Duration(milliseconds: 15),
      (batch.length, fx(batch).sumBy((t) => t.amount).toDouble()));
  inFlight--;
  return ack;
}

Future<void> main() async {
  final report = await fx(txns)
      .chunk(3)
      .toAsync()
      .map(importBatch)
      .concurrent(1)
      .scan((acc, b) => (acc.$1 + 1, b.$1, b.$2, acc.$4 + b.$2),
          (0, 0, 0.0, 0.0))
      .drop(1) // drop the scan seed
      .toList();
  print('importing ${txns.length} txns in batches of 3, one at a time:');
  for (final (n, count, amount, running) in report) {
    print('  batch $n: $count txns, \$${amount.toStringAsFixed(2)} '
        '— running total \$${running.toStringAsFixed(2)}');
  }
  print('max batches in flight: $maxInFlight');
}

Why they differ

To be fair: a strictly sequential import is the one concurrency policy a plain for loop handles gracefully, and the native version reads fine — slices from package:collection even covers the batching. The running total, though, is already mutable state threaded by hand (running += amount next to n++), where scan makes it a declared step. And the loop's simplicity is a dead end: the day the endpoint allows two concurrent batches, the FxDart version changes 1 to 2, while the loop becomes the worker pool from the other async examples. The chain states the policy; the loop encodes it.

Benchmark

Apple M1 Max, 32 GB RAM · Dart 3.12.2 (AOT-compiled) · 2026-08-24

Async case: the headline scale is N = 100,000, not 1,000,000. Every element costs an event-loop turn on both sides, so a million real awaits would measure Dart's event loop for minutes — not the pipeline. Delays are zero-length and the example's concurrency limit is kept; what the bars compare is the pipeline machinery.

N = 100

Time Tie

Native Dart 156 µs

FxDart 177 µs

Peak memory Tie

Native Dart 16.5 MB

FxDart 16.8 MB

N = 10,000

Time Native wins

Native Dart 12.7 ms

FxDart 17.3 ms

Peak memory Native wins

Native Dart 24.0 MB

FxDart 27.5 MB

N = 100,000

Time Native wins

Native Dart 131.0 ms

FxDart 157.7 ms

Peak memory Tie

Native Dart 75.9 MB

FxDart 79.7 MB

Bars are medians of repeated timed iterations in fresh processes per side (small N is batched for timer resolution). Sides within 5% of each other — or within 0.6 ms, a difference no person can perceive — count as a tie; close relative races are re-measured up to 5 runs. In an app, anything under a few milliseconds is invisible to the user regardless of which bar is shorter. Memory is peak process RSS. The Dart VM and the dataset are identical on both sides, so the difference between the two bars is what the pipeline itself holds onto.