Concurrent price lookup with fallback

FxDart wins async

Requirement

Price a six-line order. Each SKU is looked up in the live pricing service — at most three lookups in flight — but the service is missing some SKUs and returns null for them; those lines fall back to the catalog list price carried on the item. Print each priced line in order (flagging fallbacks), the fallback count, the order total, and the maximum observed concurrency. All data is in the code below.

The FxDart chain does the lookup under concurrent(3), then a second map applies the fallback with plain ?? — the recovery policy is just another pipeline step downstream of the bounded fetch.

Expected output

priced 6 items, 3 lookups at a time:
  SKU-100 x2 @ $8.49
  SKU-205 x1 @ $24.50 (catalog fallback)
  SKU-317 x3 @ $3.99
  SKU-408 x1 @ $119.00
  SKU-512 x4 @ $2.80 (catalog fallback)
  SKU-620 x2 @ $14.20
fallbacks used: 2
order total: $212.05
max lookups in flight: 3

Side by side

Native Dart

class Item {
  final String sku;
  final int qty;
  final double listPrice; // catalog fallback
  const Item(this.sku, this.qty, this.listPrice);
}

const items = [
  Item('SKU-100', 2, 9.99), Item('SKU-205', 1, 24.50),
  Item('SKU-317', 3, 4.25), Item('SKU-408', 1, 129.00),
  Item('SKU-512', 4, 2.80), Item('SKU-620', 2, 15.75),
];

int inFlight = 0;
int maxInFlight = 0;

/// The live pricing service; some SKUs are missing from it (null).
Future<double?> lookupPrice(String sku) async {
  inFlight++;
  if (inFlight > maxInFlight) maxInFlight = inFlight;
  await Future.delayed(const Duration(milliseconds: 15));
  inFlight--;
  const live = {
    'SKU-100': 8.49, 'SKU-317': 3.99, 'SKU-408': 119.00, 'SKU-620': 14.20,
  };
  return live[sku];
}

/// Worker pool: [limit] workers share a cursor, results keep input order.
Future<List<(Item, double, bool)>> priceAll(List<Item> all, int limit) async {
  final results = List<(Item, double, bool)?>.filled(all.length, null);
  var next = 0;
  Future<void> worker() async {
    while (next < all.length) {
      final i = next++;
      final live = await lookupPrice(all[i].sku);
      results[i] = (all[i], live ?? all[i].listPrice, live == null);
    }
  }

await Future.wait([for (var i = 0; i < limit; i++) worker()]);
  return results.cast<(Item, double, bool)>();
}

Future<void> main() async {
  final priced = await priceAll(items, 3);
  print('priced ${priced.length} items, 3 lookups at a time:');
  for (final (it, price, fellBack) in priced) {
    print('  ${it.sku} x${it.qty} @ \$${price.toStringAsFixed(2)}'
        '${fellBack ? ' (catalog fallback)' : ''}');
  }
  print('fallbacks used: ${priced.where((r) => r.$3).length}');
  final total = priced.fold(0.0, (sum, r) => sum + r.$2 * r.$1.qty);
  print('order total: \$${total.toStringAsFixed(2)}');
  print('max lookups in flight: $maxInFlight');
}

FxDart

import 'package:fxdart/fxdart.dart';

class Item {
  final String sku;
  final int qty;
  final double listPrice; // catalog fallback
  const Item(this.sku, this.qty, this.listPrice);
}

const items = [
  Item('SKU-100', 2, 9.99), Item('SKU-205', 1, 24.50),
  Item('SKU-317', 3, 4.25), Item('SKU-408', 1, 129.00),
  Item('SKU-512', 4, 2.80), Item('SKU-620', 2, 15.75),
];

int inFlight = 0;
int maxInFlight = 0;

Future<void> main() async {
  final priced = await fx(items)
      .toAsync()
      .map((it) async => (it, await lookupPrice(it.sku)))
      .concurrent(3)
      .map((r) => (r.$1, r.$2 ?? r.$1.listPrice, r.$2 == null))
      .toList();
  print('priced ${priced.length} items, 3 lookups at a time:');
  for (final (it, price, fellBack) in priced) {
    print('  ${it.sku} x${it.qty} @ \$${price.toStringAsFixed(2)}'
        '${fellBack ? ' (catalog fallback)' : ''}');
  }
  print('fallbacks used: ${fx(priced).filter((r) => r.$3).size()}');
  final total = fx(priced).sumBy((r) => r.$2 * r.$1.qty);
  print('order total: \$${total.toStringAsFixed(2)}');
  print('max lookups in flight: $maxInFlight');
}

Why they differ

The fallback itself is easy in both versions — ?? is Dart. What plain Dart lacks is the step before it: three-at-a-time lookups with results in input order force the worker-pool idiom (shared cursor, pre-sized slots, Future.wait), and the fallback logic ends up buried inside the worker body where it is hardest to see and test. In the FxDart version fetch and recovery are two separate, visible stages of one chain — concurrent(3) owns the limit, the next map owns the policy — and the summary lines (filter + size for fallbacks, sumBy for the total) reuse the same vocabulary the rest of the site teaches.

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 374 µs

FxDart 395 µs

Peak memory Tie

Native Dart 16.5 MB

FxDart 17.1 MB

N = 10,000

Time Tie

Native Dart 34.9 ms

FxDart 34.2 ms

Peak memory FxDart wins

Native Dart 49.1 MB

FxDart 30.9 MB

N = 100,000

Time Tie

Native Dart 351.3 ms

FxDart 353.2 ms

Peak memory Tie

Native Dart 81.1 MB

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