Provide tool to ensure async sequencing.

[lrhn]

Too many tests depend on accidental timing of events.

It's easy to make a test which succeeds when you wait, say, precisely two microtasks before checking that something has happened, it's much harder to write code to ensure that something actually happens after something else. And the first test seems to work. Until it doesn't, because of some slight change to the microtask scheduler or internal implementation of futures or streams.

I propose that the test package provides a canonical and capable tool to ensure sequencing of events. Something like:

/// Sequence checker and enforcer.
///
/// Represents a sequence of checkpoints which must be reached in order.
/// The checkpoints are represented as consecutive integers.
///
/// Example use:
/// ```dart
/// var seq = Sequencer();
/// seq.at(0); // Trivially succeeds.
/// Future.delayed(Duration.zero, () {
///    seq.at(2);
/// });
/// Future.microtask(() {
///    seq.at(1);
/// });
/// await seq.wait(3);
/// ```
/// This example checks that a microtask gets executed before a timer 
/// (checkpoint 1 before checkpoint 2),
/// and then waits for those two checkpoints to have been reached.
///
/// Can also be used with dynamically computed checkpoint numbers
/// ```dart
/// var seq = Sequencer()
/// var i = 0;
/// await stream.forEach((_) {
///   seq.at(i++);
/// });
/// // Expect 10 events
/// seq.at(10);
/// ```
abstract class Sequencer {
  /// Creates a sequencer starting at [start].
  ///
  /// Defaults to starting at zero, so the first check must be `at(0)`
  /// or `wait(0)`. Any other integer can be used.
  Sequencer([int start = 0]);

  /// Reaches checkpoint [point]. 
  ///
  /// The [point] checkpoint must be the next expected checkpoint.
  /// If not, the [onError] is called. If omitted, an error is thrown.
  ///
  /// If successful, the next expected checkpoint becomes `point + 1`.
  void at(int point, [void Function()? onError]);

  /// Waits for checkpoint [point] to be the next expected checkpoint.
  ///
  /// Returns a future which completes at some point after the
  /// checkpoint previous to [point] being reached.
  /// The checkpoint [point] is considered reached when the 
  /// future completes, and the next expected checkpoint becomes
  /// `point + 1` at that point. That is, the following should work:
  /// ```dart
  /// seq.wait(1).then((_) => seq.at(2));
  /// ```
  ///
  /// The same [point] cannot be used as argument more than once.
  /// Waiting for the returned future more than once is fine.
  /// The [point] must not be before the next expected checkpoint.
  ///
  /// Can be called out-of-order, before the `point - 1` checkpoint
  /// was reached.
  Future<void> wait(int point);
}

with an implementation like:

class _Sequencer implements Sequencer {
  /// Next checkpoint to reach.
  int _next;

  /// Completers for futures returned by [wait].
  Map<int, Completer<void>> _waits = {};

  _Sequencer([this._next = 0]);

  void at(int point, [void Function()? onError]) {
    if (point == next) {
      _advance();
    } else {
      if (onError != null) {
        onError();
      } else {
        throw StateError("Expected checkpoint $_next, was $point");
      }
    }
  }

  Future<void>(int point) {
    if (point < _next) {
      throw StateError("Waiting for checkpoint $point, already at $_next");
    }
    if (_waits.containsKey(point)) {
      throw StateError("Already waited for checkpoint $point.");
    }
    var completer = Completer<void>(); // Asynchronous completer.
    if (point == _next) {
      completer.complete(null);
    } else {
      _waits[point] = completer;
    }
    return completer.future.whenComplete(_advance);
  }

  void _advance() {
    _waits.remove(++_next)?.complete(null);
  }
}

[eernstg]

Would it be feasible (and useful) to introduce a fuzzing tool into the task scheduler? For instance, it could use a random value in {0, 1} to make a choice for each microtask: If we get 1 then it runs, otherwise it is skipped. The intention is that it should enable behaviors that might be exhibited by different, but correct, schedulers.

Would it still be possible to use a Sequencer to annotate asynchronous programs? If not, then maybe Sequencer should actually build on a partial order rather than a total order, and it should only target the situations where that partial order is contradicted.

(So if we use sets of integers, {} comes before {1} and before {2}, which are both before {1, 2}; so it would be ok to encounter {2} before {1}, but having {1,2} before {2} is a violation.)

[lrhn]

I'm sure this can be made more complicated. Like, saying at(4, count: 2) and requiring precisely two calls of that (both arguments matching) to progress to 5. That one is probably simple. Have to make the onError named too, but that's OK.

I think I'd prefer that to using sets, those are too hard to properly specify for users, especially when you get into larger cases. Would {3} include {1, 2}, or do you have to write {1, 2, 3} - that would be inconvenient.

I'm not particularly worried about checking for an arbitrary partial ordering is being followed using a single sequencer. I'd use separate sequencers for the individual parts and shared sequence-points with count > 1 for joining those individual computations.

In practice, the most common use case is probably a seq.at(0) inside some sub-computation and one await seq.wait(1); later.

/// Sequence checker and enforcer.
///
/// Represents a sequence of checkpoints which must be reached in order.
/// The checkpoints are represented as consecutive integers.
///
/// Example use:
/// ```dart
///  var seq = _Sequencer();
///  seq.at(0);
///  Future.delayed(Duration.zero, () {
///    seq.at(3);
///  });
///  Future.microtask(() {
///    seq.at(1, count: 2);
///  });
///  Future.microtask(() {
///    seq.at(1, count: 2);
///  });
///  await seq.wait(2);
///  print("microtasks done");
///  await seq.wait(4);
///  print("timers done");
/// ```
/// This example checks that a microtask gets executed before a timer 
/// (checkpoint 1 before checkpoint 2),
/// and then waits for those two checkpoints to have been reached.
///
/// Can also be used with dynamically computed checkpoint numbers
/// ```dart
/// var seq = Sequencer()
/// var i = 0;
/// await stream.forEach((_) {
///   seq.at(i++);
/// });
/// // Expect 10 events
/// seq.at(10);
/// ```
abstract class Sequencer {
  /// Creates a sequencer starting at [start].
  ///
  /// Defaults to starting at zero, so the first check must be `at(0)`
  /// or `wait(0)`. Any other integer can be used.
  factory Sequencer([int start]) = _Sequencer;

  /// Reaches checkpoint [point]. 
  ///
  /// The [point] checkpoint must be the next expected checkpoint.
  /// If not, the [onError] is called. If omitted, an error is thrown.
  /// 
  /// If `count` is provided, it must be greater than zero.
  /// It then takes that many calls to [at], 
  /// which *must* all have the same [point] and [count],
  /// before moving on to the next checkpoint.
  ///
  /// If successful, the next expected checkpoint becomes `point + 1`,
  /// but only after [count] calls, if [count] is greater than one.
  void at(int point, {int count = 1, void Function()? onError});

  /// Waits for checkpoint [point] to be the next expected checkpoint.
  ///
  /// Returns a future which completes at some point after the
  /// checkpoint previous to [point] being reached.
  /// The checkpoint [point] is considered reached when the 
  /// future completes, and the next expected checkpoint becomes
  /// `point + 1` at that point. That is, the following should work:
  /// ```dart
  /// seq.wait(1).then((_) => seq.at(2));
  /// ```
  ///
  /// The same [point] cannot be used as argument more than once.
  /// Waiting for the returned future more than once is fine.
  /// The [point] must not be before the next expected checkpoint.
  ///
  /// Can be called out-of-order, before the `point - 1` checkpoint
  /// was reached.
  Future<void> wait(int point);
}

class _Sequencer implements Sequencer {
  /// Next checkpoint to reach.
  int _next;
  /// How many times to hit the next checkpoint before it counts.
  int _nextCount = 1;
  /// How may times the next checkpoint has already been hit.
  int _currentCount = 0;

  /// Completers for futures returned by [wait].
  final Map<int, Completer<void>> _waits = {};

  _Sequencer([this._next = 0]);

  void at(int point, {int count = 1, void Function()? onError}) {
    if (count < 1) {
      throw RangeError.value(count, "count", "Must be greater than zero");
    }
    if (point == _next) {  
      if (_currentCount > 0) {
        if (count != _nextCount) {
          throw ArgumentError.value(count, "count", "Expected $_nextCount"); 
        }
      } else {
        _nextCount = count;
      }
      if (++_currentCount >= _nextCount) _advance();
    } else {
      if (onError != null) {
        onError();
      } else {
        throw StateError("Expected checkpoint $_next, was $point");
      }
    }
  }

  Future<void> wait(int point) {
    if (point < _next) {
      throw StateError("Waiting for checkpoint $point, already at $_next");
    }
    if (_waits.containsKey(point)) {
      throw StateError("Already waited for checkpoint $point.");
    }
    var completer = Completer<void>(); // Asynchronous completer.
    if (point == _next) {
      if (_currentCount > 0) {
        throw StateError("Cannot use 'at' and 'wait' with the same point number");
      }
      completer.complete(null);
    } else {
      _waits[point] = completer;
    }
    return completer.future.whenComplete(_advance);
  }

  void _advance() {
    _nextCount = 1;
    _currentCount = 0;
    _waits.remove(++_next)?.complete(null);
  }
}

[lrhn]

A much lower abstraction would be a simple:

/// A synchronization point.
///
/// The synchronization point completes its [done] future 
/// when enough [trigger] invocations have happened.
/// The default is one. 
class SyncPoint {
  /// Creates a synchronization point requiring [triggers] trigger calls.
  factory SyncPoint([int triggers]) = _SyncPoint;
  /// Perform one trigger call.
  /// 
  /// If enough trigger calls have happened, the [done] future completes.
  void trigger();

  /// The future which gets completed when enough triggers have occurred.
  Future<void> get done;

  /// Set to true when enough triggers have happened. 
  ///
  /// This happens before the [done] future completes.
  bool get isDone;
}

class _SyncPoint implements SyncPoint {
  int _count;
  Completer<void>? _completer;
  _SyncPoint([this._count = 1]) {
    if (_count < 0) {
      throw RangeError.value(_count, "triggers", "Must be greater than zero";
    }
  }
  void trigger() {
    if (_count > 0 && --_count == 0) {
      _completer?.complete(null);
    }
  }
  Future<void> get _done {
    var completer = _completer;
    if (completer == null) {
      _completer = completer = Completer<void>();
      if (_count == 0) completer.complete();
    }
    return completer.future;
  }
  bool get isDone => _count == 0;
}

Then you just use multiple of these:

var sync2 = SyncPoint(2);
var sync4 = SyncPoint();
Future.delayed(Duration.zero, () {
   assert(sync2.isDone);
   sync4.trigger();
});
Future.microtask(() {
   sync2.trigger();
});
Future.microtask(() {
   sync2.trigger();
});
await sync2.done();
print("microtasks done");
await sync4.done();
print("timers done");

This is isimpler, very close to just being a Completer, but that's basically what people aren't doing today (they already have completers, they don't use them like this). It also doesn't offer an easy way to check synchronous sequence points. You can use isDone, but it's more complicated. The advantage of Sequencer is that it handles failures for you, so you only need to declaratively say what should be true.

[eernstg]

@lrhn wrote:

I'm sure this can be made more complicated.

Granted. :grin:

I think I'd prefer that [this is, generalizations like at(4, count: 2)] to using sets

I wasn't actually expecting anyone to use sets of integers, they just happen to be a fully general way to specify an arbitrary partial order (based on the normal 'subset' or 'superset' relation).

It looks like Sequencer is the most convenient abstraction.

I was just worried that it might be an overspecification: There could be a correct scheduler which will maintain certain required ordering constraints, but apart from that it could schedule any activity arbitrarily (say, because it's running really fast if we do certain specific things). If that's true then anything you'd write using Sequencer might report a run-time failure, because the expectations are too specific.

[jakemac53]

This seems to me like an api that is possibly better suited to its own package? It seems like it might not quite be general enough (it only handles an exact linear sequence of events).

You can just use manual completers today (and that is what people should be doing instead of waiting X milliseconds). People should be pushing back during code review on tests that wait arbitrary amounts of time or micro tasks.

[natebosch]

I agree that this should be a separate package. That would also make it easier to iterate on an API.