Proposal: keyed in-flight request coalescing (`SingleFlight<K,V>`) in package:async

[bthnkucuk]

Motivation

A common need in Dart apps — especially Flutter clients — is request
coalescing: when N concurrent calls arrive for the same key while one is
already in flight, run the underlying work once and share the result with
every caller. Real cases we hit in production:

• Token refresh: a burst of 401s from parallel API calls must trigger a
  single refresh, not N.
• TTS / image / file download: bursty UI events (rapid scroll, repeated
  taps) fan out to the same endpoint within the same event-loop tick.
• Cache stampede: a cache miss observed by N concurrent readers should
  result in one origin fetch, not N.

This is the classic "single-flight" pattern, popularized by Go's
golang.org/x/sync/singleflight
(originally from groupcache). It's a small primitive but ubiquitous in
backend and client code.

Gap in current Dart options

• package:async ships AsyncCache and AsyncMemoizer, both of which
  dedupe in-flight calls — but for a single, fixed key:

  • AsyncMemoizer: one-shot, never resets.
  • AsyncCache: one key per instance, caches the result for a TTL.

  Neither covers the keyed + no-caching + multi-round use case (e.g. one
  SingleFlight<String, User> shared across fetchUser('42'),
  fetchUser('57'), …).

• Pub.dev's singleflight exists
  but appears unmaintained:

  • Last update 19 months ago, 0 likes, ~2.4k total downloads.
  • Source repo (github.com/bachue/dart-singleflight) returns 404.
  • Unverified uploader.
  • API: String-only keys, no Forget, depends on the mutex package
    (which is unnecessary on Dart's single-threaded event loop).

So there's a real gap and no maintained third-party fills it. package:async
is the natural home — it already groups async coordination primitives
(AsyncCache, AsyncMemoizer, CancelableOperation, StreamGroup, etc.).

Proposed API

class SingleFlight<K, V> {
  /// Runs [compute] for [key], deduplicating concurrent calls.
  /// Returns the value plus a `shared` flag — `true` if at least one other
  /// caller waited on the same in-flight execution.
  /// Mirrors Go's `Do(key, fn) (val, err, shared)`.
  Future<SingleFlightResult<V>> run(K key, Future<V> Function() compute);

  /// Drops [key] from the in-flight map without canceling the running call.
  /// Already-waiting callers still receive that flight's result; subsequent
  /// callers start a fresh flight. Mirrors Go's `Forget(key)`.
  void forget(K key);

  /// Keys currently in flight, for telemetry/debugging.
  Iterable<K> get inFlightKeys;
}

class SingleFlightResult<V> {
  final V value;
  final bool shared;
}

## Reference implementation

A working version is in use in our app monorepo. ~80 LOC, no extra deps,
passes 77 tests including error propagation, mid-flight `forget`,
record-key dedup, and 50-way concurrent fan-out.

/// In-flight request coalescing — Dart port of Go's
/// `golang.org/x/sync/singleflight`.
class SingleFlight<K, V> {
  final _ongoing = <K, _Call<V>>{};

  Future<SingleFlightResult<V>> run(
    K key,
    Future<V> Function() compute,
  ) async {
    final existing = _ongoing[key];
    if (existing != null) {
      existing.dups++;
      final value = await existing.future;
      return SingleFlightResult._(value, shared: true);
    }

    final call = _Call<V>(compute());
    _ongoing[key] = call;
    try {
      final value = await call.future;
      return SingleFlightResult._(value, shared: call.dups > 0);
    } finally {
      // Only release the slot if it still points to *this* call —
      // forget() may have replaced it with a fresh flight already.
      if (identical(_ongoing[key], call)) _ongoing.remove(key);
    }
  }

  void forget(K key) => _ongoing.remove(key);

  Iterable<K> get inFlightKeys => _ongoing.keys;
}

class SingleFlightResult<V> {
  const SingleFlightResult._(this.value, {required this.shared});
  final V value;
  final bool shared;
}

class _Call<V> {
  _Call(this.future);
  final Future<V> future;
  int dups = 0;
}

[lrhn]

I'm not familiar with the pattern.

Why have the shared boolean?
I can see how you should only dispose a value if its not shared, but then it's not actionable of it is shared. One bit is to too little information to be useful.
If you had a way to create independent result objects for each caller, knowing the total share count, then you could choose a strategy that fits.

How would this differ from a Map<K, AsyncCache<V>>, using putIfAbsent to add a transient cache, and remove to forget it?
That doesn't let us easily see if an ephemeral cache has a value or not, but if not caring about inFlightKeys (which I'd drop, introspection is bad for abstraction) and shared (which feels insufficient anyway), it's really just:

import 'package:async/async.dart';
extension type SingleFlight<K, V>._(Map<K, AsyncCache<V>> _) {
  SingleFlight() : this._({});
  Future<V> run(K key, Future<V> Function() function) => 
    _.putIfAbsent(key, AsyncCache<V>.ephemeral).fetch(function);
  void forget(K key) { _.remove(key); }
}

which is so simple that you could, and should, just create a map and use it directly in your own abstraction, no library type needed.

Or you can have a map and store the future in the map until it's completed, that allows checking for active keys:

extension type SingleFlight<K, V>._(Map<K, Future<V>> _) {
  SingleFlight(): this._({});
  Future<V> run(K key, Future<V> Function() function) => 
      _.putIfAbsent(key, () {
        var future = function();
        future = future.whenComplete(() {
          // Check if forgotten and recreated.
          if (identical(_[key], future)) _.remove(key);
        });
        return future;
     });
  void forget(K key) { _.remove(key); }
  Iterable<K> get inFlightKeys => _.keys;
}

Takes slightly more effort to track a shared bit, but I don't see its purpose, so I don't really know what to optimize it for.
It's basically what you suggest, just using a record type for returning two values.

import 'dart:async';
extension type SingleFlight<K, V>._(Map<K, _Call<V>> _) {
  SingleFlight(): this._({});
  Future<(V, {bool shared})> run(K key, Future<V> Function() function) =>
      _.update(key, (value) => value..shared = true, ifAbsent: () {
        var future = function();
        future = future.whenComplete(() {
          // Check if forgotten and recreated.
          if (identical(_[key]?.future, future)) _.remove(key);
        });
        return _Call(future);
      }).result.future;
  void forget(K key) { _.remove(key); }
  Iterable<K> get inFlightKeys => _.keys;
}

final class _Call<V> {
  final Future<V> future;
  bool shared = false;
  final Completer<(V, {bool shared})> result = Completer.sync();
  _Call(this.future) {
    future.then((v) => result.complete((v, shared: shared)),
       onError: (Object e, StackTrace s) => result.completeError(e, s));;
  }
}

Now, if there was a "share counter" so you could release the value when you no longer use it, and the user whose release returns true was the last owner and should dispose the value, then _that _might be useful and actionable.

class Share {
  _ShareCounter? _counter;
  Share._(this._counter);
  /// Whether this share has already been released.
  bool get isReleased => _counter == null;
  /// Released the share.
  ///
  /// If result is `true`, this was the last share of the
  /// value, and the caller is responsible for disposing
  /// the value, if necessary.
  /// The result is `false` if the share has already been released.
  bool release() {
    if (_counter case var counter?) {
      _counter = null; // Can only release once.
      return counter.release();
    }
    return false;
  }
}
class _ShareCounter {
  int useCount;
  _ShareCounter(this.useCount): assert(useCount > 0);
  bool release() {
    assert(useCount > 0);
    return --useCount == 0;
  }
  void increment() {  
    useCount++;
  }
}

Then every call to SingleFlight.run gets its own Share on the same ShareCounter, and whoever releases last draws the short straw and must dispose the value.
(Nothing enforced, just good citizenship to follow the protocol.)

extension type SingleFlight<K, V>._(Map<K, _Call<V>> _) {
  SingleFlight(): this._({});
  Future<(V, Share)> run(K key, Future<V> Function() function) =>
      _.update(key, (value) => value..uses.increment(), ifAbsent: () {
        var future = function();
        future = future.whenComplete(() {
          // Check if forgotten and recreated.
          if (identical(_[key]?.future, future)) _.remove(key);
        });
        return _Call(future);
      }).result;
  void forget(K key) { _.remove(key); }
  Iterable<K> get inFlightKeys => _.keys;
}

final class _Call<V> {
  final Future<V> future;
  final _ShareCounter uses = _ShareCounter(1);
  _Call(this.future);
  Future<(V, Share)> get result => future.then((v) => (v, Share._(uses)));
}

It'd be wasteful to use a complicated share-count tracking implementation for something that doesn't care about disposing anyway.
It'd be wasteful to care about supporting inFlightKeys if it's not needed.

This is a case where the there are valid designs, but there are multiple valid designs for different use-cases, and building one intopackage:async might help some people, but it may be overkill for others, or fail to actually be enough for almost similar use-cases.

We could have a family of operations, so you could pick the feature-set you need for your particular use-case, but then there may be requests for even more variations.

The simplest solution might be enough for most people, but you don't need a library for a map of ephemeral async caches.
Adding more features feels speculative to me, because I don't actually know when it'll be enough, or too much.

I guess that's the issue. I don't have a good grasp of the expected user-cases, and near-by use-cases. If I did, I might feel more confident that we could cover the important ones well. That's why I'm reluctant to put any particular specific design into the package. I can't say if it's sufficient, and if not, what would be.