Reintentar cada fila inestable por separado

Gana FxDart async

Requisito

Importa seis filas a través de un endpoint inestable: las filas pares fallan exactamente una vez antes de tener éxito. Da a cada fila su propio presupuesto de reintentos de dos intentos, ejecuta hasta tres filas a la vez, e imprime los resultados en orden de origen con el número del intento que tuvo éxito. La inyección de fallos y los retardos por fila son deterministas y están en el código; las dos versiones deben imprimir las líneas que aparecen bajo Salida esperada.

Salida esperada

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

Lado a lado

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);
}

Por qué difieren

Ambos lados expresan la mitad de resiliencia igual: un envoltorio de retry por fila, de modo que una fila inestable se re-ejecuta mientras sus vecinas pasan sin tropiezos. RxDart lo escribe flatMap hacia un stream interno con reintentos por fila con maxConcurrent: 3; FxDart lo escribe mapRetry(2, …) bajo concurrent(3), donde cada elemento en vuelo lleva su propio presupuesto independiente.

La diferencia es lo que sale por el otro extremo. flatMap fusiona los streams internos en orden de terminación — ese es su contrato — así que con tres filas en vuelo y retardos desiguales, los resultados llegan barajados. Recuperar el orden de origen significa etiquetar cada resultado con su id de fila y ordenar después de toList. El operador de RxDart que sí conservaría el orden, concatMap, lo hace renunciando a la concurrencia — una fila a la vez. En el modelo pull, la concurrencia ordenada es el modo nativo: concurrent(3) evalúa tres pulls a la vez pero los entrega en orden de origen por construcción, así que no hay nada que etiquetar ni nada que ordenar.

Veredicto FxDart — el orden es la historia. «N a la vez, reintentadas por separado, en orden» es una sola cadena en el modelo pull y un rodeo de fusionar-y-reordenar en el modelo push.

Benchmark

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

Caso async: la escala principal es N = 10,000, no 1,000,000. Cada elemento cuesta una vuelta del event loop en ambos lados, así que un millón de awaits reales mediría el event loop de Dart durante minutos — no el pipeline. Los retardos son de longitud cero y se conserva el límite de concurrencia del ejemplo; lo que comparan las barras es la maquinaria del pipeline.

N = 100

Tiempo Empate

RxDart 737 µs

FxDart 768 µs

Memoria pico Empate

RxDart 17.2 MB

FxDart 16.7 MB

N = 10,000

Tiempo Empate

RxDart 65.4 ms

FxDart 62.9 ms

Memoria pico Gana FxDart

RxDart 61.2 MB

FxDart 40.3 MB

Las barras son medianas de iteraciones cronometradas repetidas en procesos nuevos por lado (los N pequeños se agrupan por resolución del temporizador). Dos lados a menos del 5% entre sí — o a menos de 0.6 ms, una diferencia que nadie puede percibir — cuentan como empate; las carreras relativas ajustadas se vuelven a medir hasta 5 veces. En una app, cualquier cosa por debajo de unos pocos milisegundos es invisible para el usuario, gane la barra que gane. La memoria es el RSS pico del proceso. La VM de Dart y el dataset son idénticos en ambos lados, así que la diferencia entre las dos barras es lo que retiene el pipeline en sí.