Retry each flaky row independently

FxDart wins async

Requirement

Import six rows through a flaky endpoint: the even rows fail exactly once before succeeding. Give each row its own retry budget of two attempts, run up to three rows at a time, and print the results in source order with the attempt count that succeeded. The failure injection and per-row delays are deterministic and in the code; both versions must print the lines shown under Expected output.

Expected output

row 1 (alpha) imported on attempt 1
row 2 (bravo) imported on attempt 2
row 3 (charlie) imported on attempt 1
row 4 (delta) imported on attempt 2
row 5 (echo) imported on attempt 1
row 6 (foxtrot) imported on attempt 2

Side by side

RxDart

import 'package:rxdart/rxdart.dart';

/// Six import rows; the even rows hit a flaky endpoint that fails exactly
/// once before succeeding. Per-row delays make completion order differ
/// from source order under concurrency.
const rows = [
  (1, 'alpha'),
  (2, 'bravo'),
  (3, 'charlie'),
  (4, 'delta'),
  (5, 'echo'),
  (6, 'foxtrot'),
];
const importMs = [0, 70, 30, 50, 30, 20, 40]; // indexed by row id

final attemptsByRow = <int, int>{};

Future<String> importRow((int, String) row) async {
  final (id, name) = row;
  final attempt = attemptsByRow.update(id, (n) => n + 1, ifAbsent: () => 1);
  await Future<void>.delayed(Duration(milliseconds: importMs[id]));
  if (id.isEven && attempt == 1) throw StateError('endpoint reset on $name');
  return 'row $id ($name) imported on attempt $attempt';
}

Future<void> main() async {
  final results = await Stream.fromIterable(rows)
      // Each row gets its own retrying inner stream, 3 subscribed at a
      // time — but flatMap emits in COMPLETION order, so every result
      // must carry its row id for the sort back to source order.
      .flatMap(
          (row) => Rx.retry(() => Rx.fromCallable(() => importRow(row)), 1)
              .map((line) => (row.$1, line)),
          maxConcurrent: 3)
      .toList();

results.sort((a, b) => a.$1.compareTo(b.$1));
  for (final (_, line) in results) {
    print(line);
  }
}

FxDart

import 'package:fxdart/fxdart.dart';

final attemptsByRow = <int, int>{};

Future<void> main() async {
  // Two attempts per row, three rows in flight — each in-flight row
  // retries independently, and mapRetry under concurrent yields in
  // SOURCE order by construction.
  final results = await fx(rows)
      .toAsync()
      .mapRetry(2, importRow)
      .concurrent(3)
      .toList();

results.forEach(print);
}

Why they differ

Both sides express the resilience half the same way: a retry wrapper per row, so one flaky row re-runs while its neighbors sail through. RxDart spells it flatMap into a retrying inner stream per row with maxConcurrent: 3; FxDart spells it mapRetry(2, …) under concurrent(3), where each in-flight element carries its own independent budget.

The difference is what comes out the other end. flatMap merges inner streams in completion order — that is its contract — so with three rows in flight and unequal delays, the results arrive shuffled. Getting source order back means tagging every result with its row id and sorting after toList. The RxDart operator that would preserve order, concatMap, does it by giving up the concurrency — one row at a time. In the pull model, ordered concurrency is the native mode: concurrent(3) evaluates three pulls at once but yields them in source order by construction, so there is nothing to tag and nothing to sort.

Verdict FxDart — the ordering is the story. "N at a time, retried independently, in order" is one chain in the pull model and a merge-then-reorder workaround in the push model.

Benchmark

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

Async case: the headline scale is N = 10,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

RxDart 737 µs

FxDart 768 µs

Peak memory Tie

RxDart 17.2 MB

FxDart 16.7 MB

N = 10,000

Time Tie

RxDart 65.4 ms

FxDart 62.9 ms

Peak memory FxDart wins

RxDart 61.2 MB

FxDart 40.3 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.