`Stream.first` hangs for `combineLatest + switchMap` with open-ended `async*` streams

[saibotma]

Awaiting stream.first hangs for a stream built from combineLatest and
switchMap over open-ended async* streams.

This looks very similar to the RxDart issue below, but for stream_transform:

• Stream.first hangs for combineLatest + switchMap with open-ended async* streams ReactiveX/rxdart#804

Minimal reproduction

import 'dart:async';
import 'dart:io';

import 'package:stream_transform/stream_transform.dart';

Future<void> main() async {
  final stream = _emitOnceAndNeverClose('left').combineLatest(
    Stream.value('right').switchMap(_emitOnceAndNeverClose),
    (left, right) => '$left|$right',
  );

  final value = await stream.first.timeout(const Duration(milliseconds: 200));
  stdout.writeln(value);
}

Stream<T> _emitOnceAndNeverClose<T>(T value) async* {
  yield value;
  await Completer<void>().future;
}

How to run

dart run repro.dart

Expected behavior

The program should complete and print:

left|right

Actual behavior

The program times out:

Unhandled exception:
TimeoutException after 0:00:00.200000: Future not completed

Without the timeout, it hangs indefinitely.

What I verified

• switchMap is required. Replacing the switchMap(...) branch with a direct
  _emitOnceAndNeverClose('right') stream makes the problem disappear.
• The non-switchMap branch also matters. Replacing
  _emitOnceAndNeverClose('left') with Stream.value('left') makes the
  problem disappear.
• Stream.value('right').switchMap(_emitOnceAndNeverClose) alone works.
• combineLatest without switchMap works.

Additional context

We ran into this in a Flutter app where the source streams were long-lived
database watchers from Drift (watchSingleOrNull()), wrapped in async*
syncers. Those streams are intentionally open-ended until canceled.

The failure happened when we composed those watcher streams with
combineLatest and switchMap, then awaited .first to get one initial
bootstrap value. The minimal repro above removes Drift entirely and still
reproduces, so this does not appear to be Drift-specific.

Environment

• stream_transform: 2.1.1
• Dart SDK: 3.12.0-113.1.beta
• Flutter: 3.42.0-0.0.pre
• Platform: macos_arm64

Related context

This seems especially relevant because stream_transform's changelog already
mentions cancellation behavior in switchLatest / switchMap:

• https://pub.dev/packages/stream_transform/changelog

And the current implementation of both switchMap and combineLatest
explicitly waits for cancel() futures during teardown, which seems related to
this hang.

[lrhn]

This is possibly expected behavior.

The first cancels the stream after the first event.
That cancel call propagates down to the _emitOnceAndNeverClose streams. Calling StreamSubscription.cancel on the subscriptions of those streams never completes if they have reached the await. (It waits for control to leave the function body, which it never does.)

Since first waits for the cancel call to complete, it also never completes.

Now, this should only happen if something allows the stream to continue (doesn't pause it when receiving the event), because cancelling does work if it's done at the yield. When it leaves that yield, it becomes impossible to cancel the stream.

Or it's the switchMap stream which fails to cancel its stream.

In any case: Don't make streams that can't be cancelled.

A (simpler?) example could be:

import 'dart:async';

Future<void> main() async {
  var stream = delayedStream(
    _emitOnceAndNeverClose('right'),
    Duration(milliseconds: 25),
  );
  var first = await stream.first.timeout(Duration(seconds: 5));
  print("Done: $first");
}

Stream<T> _emitOnceAndNeverClose<T>(T value) async* {
  yield value;
  await Completer<void>().future;
}

// Delayes the events of a stream
Stream<T> delayedStream<T>(Stream<T> source, Duration duration) =>
    source.transform(
      StreamTransformer.fromHandlers(
        handleData: (value, sink) {
          Timer(duration, () {
            sink.add(value);
          });
        },
        handleError: (error, stack, sink) {
          Timer(duration, () {
            sink.addError(error, stack);
          });
        },
        handleDone: (sink) {
          Timer(duration, () {
            sink.close();
          });
        },
      ),
    );

All that is needed is that the _emitOnceAndNeverClose is cancelled after the yield has completed.

[saibotma]

"In any case: Don't make streams that can't be cancelled." agree on that. I actually had that issue in a more complex scenario and just tried to reproduce a simple example. In my real world scenario the underlying streams were streams coming from a drift database and I think they should cancel correctly. Will try to build a better example.

[lrhn]

If you change the _emitOnceAndNeverClose to:

Stream<T> _emitOnceAndNeverClose<T>(T value) {
  return (StreamController<T>()..add(value)).stream;
}

which is a canonical "stream with one event and not closing", then the problem goes away.

That suggests that the real problem is the async* function that cannot be cancelled.