fix(scope): Scope.runAsync in digest and flush

lib/core/scope.dart

@@ -270,9 +270,10 @@ class Scope {
     } catch (e, s) {
       rootScope._exceptionHandler(e, s);
     } finally {
-      rootScope.._transitionState(RootScope.STATE_APPLY, null)
-               ..digest()
-               ..flush();
+      rootScope.._transitionState(RootScope.STATE_APPLY, null);
+      do {
+        rootScope..digest()..flush();
+      } while (rootScope._runAsyncHead != null);
     }
   }
 
@@ -613,6 +614,7 @@ class RootScope extends Scope {
   {
     _zone.onTurnDone = apply;
     _zone.onError = (e, s, ls) => _exceptionHandler(e, s);
+    _zone.onScheduleMicrotask = runAsync;
   }
 
   RootScope get rootScope => this;
@@ -648,15 +650,8 @@ class RootScope extends Scope {
       ChangeLog changeLog;
       _scopeStats.digestStart();
       do {
-        while (_runAsyncHead != null) {
-          try {
-            _runAsyncHead.fn();
-          } catch (e, s) {
-            _exceptionHandler(e, s);
-          }
-          _runAsyncHead = _runAsyncHead._next;
-        }
-        _runAsyncTail = null;
+
+        int asyncCount = _runAsyncFns();
 
         digestTTL--;
         count = rootWatchGroup.detectChanges(
@@ -672,7 +667,7 @@ class RootScope extends Scope {
             digestLog = [];
             changeLog = (e, c, p) => digestLog.add('$e: $c <= $p');
           } else {
-            log.add(digestLog.join(', '));
+            log.add("${asyncCount > 0 ? 'async:$asyncCount' : ''}${digestLog.join(', ')}");
             digestLog.clear();
           }
         }
@@ -681,7 +676,7 @@ class RootScope extends Scope {
                 'Last $LOG_COUNT iterations:\n${log.join('\n')}';
         }
         _scopeStats.digestLoop(count);
-      } while (count > 0);
+      } while (count > 0 || _runAsyncHead != null);
     } finally {
       _scopeStats.digestEnd();
       _transitionState(STATE_DIGEST, null);
@@ -756,6 +751,9 @@ class RootScope extends Scope {
 
   // QUEUES
   void runAsync(fn()) {
+    if (_state == STATE_FLUSH_ASSERT) {
+      throw "Scheduling microtasks not allowed in $state state.";
+    }
     var chain = new _FunctionChain(fn);
     if (_runAsyncHead == null) {
       _runAsyncHead = _runAsyncTail = chain;
@@ -764,6 +762,21 @@ class RootScope extends Scope {
     }
   }
 
+  _runAsyncFns() {
+    var count = 0;
+    while (_runAsyncHead != null) {
+      try {
+        count++;
+        _runAsyncHead.fn();
+      } catch (e, s) {
+        _exceptionHandler(e, s);
+      }
+      _runAsyncHead = _runAsyncHead._next;
+    }
+    _runAsyncTail = null;
+    return count;
+  }
+
   void domWrite(fn()) {
     var chain = new _FunctionChain(fn);
     if (_domWriteHead == null) {

lib/core/zone.dart

@@ -60,6 +60,7 @@ class VmTurnZone {
   /// an "inner" [Zone], which is a child of the outer [Zone].
   async.Zone _innerZone;
 
+
   /**
    * Associates with this
    *

test/core/scope_spec.dart

@@ -1227,6 +1227,20 @@ void main() {
       });
 
 
+      it(r'should detect infinite digest through runAsync', (RootScope rootScope) {
+        rootScope.context['value'] = () { rootScope.runAsync(() {}); return 'a'; };
+        rootScope.watch('value()', (_, __) {});
+
+        expect(() {
+          rootScope.digest();
+        }).toThrow('Model did not stabilize in 5 digests. '
+        'Last 3 iterations:\n'
+        'async:1\n'
+        'async:1\n'
+        'async:1');
+      });
+
+
       it(r'should always call the watchr with newVal and oldVal equal on the first run',
       inject((RootScope rootScope) {
         var log = [];
@@ -1446,6 +1460,105 @@ void main() {
       });
     });
 
+    describe('microtask processing', () {
+      beforeEach((VmTurnZone zone, RootScope scope, Logger log) {
+        var onTurnDone = zone.onTurnDone;
+        zone.onTurnDone = () {
+          log('[');
+          onTurnDone();
+          log(']');
+        };
+        var onScheduleMicrotask = zone.onScheduleMicrotask;
+        zone.onScheduleMicrotask = (fn) {
+          log('(');
+          try {
+            onScheduleMicrotask(fn);
+          } catch (e) {
+            log('CATCH: $e');
+          }
+          log(')');
+        };
+      });
+
+      it('should schedule apply after future resolution',
+        async((Logger log, VmTurnZone zone, RootScope scope) {
+          Completer completer;
+          zone.run(() {
+            completer = new Completer();
+            completer.future.then((value) {
+              log('then($value)');
+            });
+          });
+
+          scope.runAsync(() => log('before'));
+          log.clear();
+          completer.complete('OK'); // this one causes APPLY which processe 'before'
+          // This one schedules work but apply already run so it does not execute.
+          scope.runAsync(() => log('NOT_EXECUTED'));
+
+          expect(log).toEqual(['(', ')', '[', 'before', 'then(OK)', ']']);
+        })
+      );
+
+      it('should schedule microtask to runAsync queue during digest',
+        async((Logger log, VmTurnZone zone, RootScope scope) {
+          Completer completer;
+          zone.run(() {
+            completer = new Completer();
+            completer.future.
+              then((value) {
+                scope.runAsync(() => log('in(${scope.state})'));
+                return new Future.value(value);
+              }).
+              then((value) {
+                log('then($value)');
+              });
+          });
+          log.clear();
+          completer.complete('OK');
+          expect(log).toEqual(['(', ')', '[', '(', ')', 'in(digest)', 'then(OK)', ']']);
+        })
+      );
+
+      it('should allow microtasks in flush phase, but will trigger another digest',
+        async((Logger log, VmTurnZone zone, RootScope scope) {
+          scope.watch('g()', (_, __) {});
+          scope.context['g'] = () {
+            log('!');
+            return 0;
+          };
+
+          zone.run(() {
+            scope.domWrite(() {
+              log('domWriteA');
+              return new Future.value(null).then((_) => scope.domWrite(() => log('domWriteB')));
+            });
+          });
+          expect(log).toEqual(
+              ['[', '!', '!', 'domWriteA', '(', ')', '!', '!', 'domWriteB', /* assert */'!', ']']);
+        })
+      );
+
+      it('should allow creation of Completers in flush phase',
+        async((Logger log, VmTurnZone zone, RootScope scope) {
+          Completer completer;
+          zone.run(() {
+            scope.domWrite(() {
+              log('new Completer');
+              completer = new Completer();
+              completer.future.then((value) {
+                log('then($value)');
+              });
+            });
+          });
+          log('=');
+          completer.complete('OK');
+          log(';');
+          expect(log).toEqual(
+              ['[', 'new Completer', ']', '=', '(', ')', '[', 'then(OK)', ']', ';']);
+        })
+      );
+    });
 
     describe('domRead/domWrite', () {
       beforeEachModule((Module module) {