Stream.fromIterable doesn't handle errors during iteration well

[kevmoo]

If an error is throw, it's not caught. The app just seems to exit with 0 without done being hit. Very weird.

The sample code at the bottom has the following output:

custom safe iteratorToStream
*** STARTING ***
*** STREAM CREATED ***
 DATA: 1
 DATA: 2
ERROR: no 3!
DONE!
*** DONE ***
true

new Stream.fromIterator
*** STARTING ***
*** STREAM CREATED ***
 DATA: 1
 DATA: 2
ERROR: no 3!

Note: no *** DONE ***, no true printed at the bottom. The app just exits (with status 0).

This is on -dev.61

import 'dart:async';

main() async {
  print('custom safe iteratorToStream');
  var completed = await _printStream(safeIteratorToStream);
  print(completed);
  print('');

  print('new Stream.fromIterator');
  completed =
      await _printStream((iterable) => new Stream.fromIterable(iterable));
  // THIS IS NEVER CALLED!
  print(completed);
}

Future<bool> _printStream(Stream makeStream(Iterable iterable)) async {
  print('*** STARTING ***');
  var stream = makeStream(_badIterable());
  print('*** STREAM CREATED ***');
  await stream.transform(_printingTransformer).drain();
  print('*** DONE ***');
  return true;
}

final _printingTransformer = new StreamTransformer<Object, Object>.fromHandlers(
    handleData: (data, sink) {
  print(' DATA: $data');
}, handleError: (error, stack, sink) {
  print('ERROR: $error');
}, handleDone: (sink) {
  print('DONE!');
  sink.close();
});

Stream safeIteratorToStream(Iterable source) {
  var controller = new StreamController();

  new Future(() {
    try {
      var iterator = source.iterator;
      while (iterator.moveNext()) {
        controller.add(iterator.current);
      }
    } catch (e, stack) {
      controller.addError(e, stack);
    } finally {
      controller.close();
    }
  });

  return controller.stream;
}

Iterable<int> _badIterable() sync* {
  yield 1;
  yield 2;
  throw 'no 3!';
}

CC @lrhn

[natebosch]

This seems working as intended. From the Doc comment on the constructor:

If iterating data throws an error, the stream ends immediately with that error. No done event will be sent (iteration is not complete), but no further data events will be generated either, since iteration cannot continue.

[natebosch]

This was changed here: 781e842#diff-5a9c317c5de11a5a7ab7a39c97e76a90R517

Before that it looks like the stream was closed on error. Nothing in the commit message or the review mention why this decision was made. @lrhn could maybe add some context on that.

[kevmoo]

Reopening – seems really weird that final print is not called, but there is no error and the exit code is still 0.

[natebosch]

https://codereview.chromium.org/14753009/diff/6001/sdk/lib/async/stream_impl.dart#newcode523

[kevmoo]

See this test case - https://gist.github.com/kevmoo/c51dbeedb9091052d36b193432fea87f

pkg:test just hangs!

@natebosch – thanks for your digging!

[kevmoo]

The "safe" behavior makes so much more sense to me. Having the Stream end w/ the first error is much easier to deal with...

[kevmoo]

@mraleph – FYI re our chat before lunch

[lrhn]

The issue of the VM terminating in the middle of an async method is not new (#23797).
If that's the only issue here, the finally not being executed when an async operation doesn't complete and the VM terminates anyway, then I'll just defer to that issue.

I'm willing to reconsider the behavior for fromIterable. We also don't catch errors thrown by the current getter (which can happen, especially with Iterable.cast which checks the type only when you access the element), so an overhaul is due.

The issue here is that sending a done event has a meaning - the stream is complete. It's not the same as there being no further events, although that's obviously implied.
On the other hand, most real code won't just catch and print the error and continue. It'll either listen with cancelOnError set or do something on an error (or not do something on an error and it becomes an uncaught async error terminating the program, great fun!)
In that case, adding a done event should not change anything, if the listener is already stopped anyway.

Which makes me ask: Is this an actual issue, separate from the VM terminating? That is, if the VM hung instead, would that be enough? Is this occurring in actual code, and not just toy examples?

If we change it, one option is to keep iterating after a throw. That allows each call to moveNext to throw, and be retried afterwards. If the next call succeeds, then there is no issue. If the next call returns false, then we'll get a done event automatically.
The danger is that repeated calls to moveNext will all throw, so you get an infinite stream of errors. That's a very real danger, so it probably disqualifies the approach.

The other alternative is to always send a done event after the error event. If code ignores the error, it might think that the input is actually complete, but then, if you ignore an error in your data source, you should probably not trust the result anyway.

[kevmoo]

I'd say error out after the first error – on current, moveNext, iterator access, anything.

By the law of least surprise, I'd assume that running over _badStream() should have ~identical behavior to running over new Stream.fromIterable(_badIterable()).

Stream<int> _badStream() async* {
  yield 1;
  yield 2;
  throw 'no 3!';
}

Iterable<int> _badIterable() sync* {
  yield 1;
  yield 2;
  throw 'no 3!';
}

[lrhn]

I'm trying an implementation where a throwing moveNext call ends iteration, but does send a done event after the error, and a throwing current just sends the error and keeps iterating.

That should let the example above have identical behavior - the throw in the sync* function shows up as a throwing moveNext call, and the throw in he async* will report an error and close the stream as well.