Working on mongo-livedata tests.

packages/meteor/fiber_helpers.js

@@ -160,17 +160,35 @@ SQp._scheduleRun = function () {
    *  For autoupdate, for example, swapping to just running it sync won't make a difference,
    *  but maybe there is another place that would? (DDP/Web)-Server?
    */
-  if (Meteor._isFibersEnabled) {
-    setImmediate(function() {
-      Meteor._runAsync(function() {
-        self._run();
-      });
-    });
-  } else {
-    Meteor._runAsync(() => {
-      self._run();
-    });
+  setImmediate(function() {
+    Meteor._runAsync(Meteor._isFibersEnabled ? self._run : self._runAsync, self);
+  });
+};
+
+SQp._runAsync = async function() {
+  var self = this;
+
+  if (!self._runningOrRunScheduled)
+    throw new Error("expected to be _runningOrRunScheduled");
+
+  if (self._taskHandles.isEmpty()) {
+    // Done running tasks! Don't immediately schedule another run, but
+    // allow future tasks to do so.
+    self._runningOrRunScheduled = false;
+    return;
   }
+  var taskHandle = self._taskHandles.shift();
+
+  // Run the task.
+  try {
+    await taskHandle.task();
+  } catch (err) {
+    Meteor._debug("Exception in queued task", err);
+  }
+
+  // Soon, run the next task, if there is any.
+  self._runningOrRunScheduled = false;
+  self._scheduleRun();
 };
 
 SQp._run = function () {

packages/mongo/mongo_driver.js

@@ -479,14 +479,27 @@ MongoConnection.prototype._update = function (collection_name, selector, mod,
   // non-object modifier in that they don't crash, they are not
   // meaningful operations and do not do anything. Defensively throw an
   // error here.
-  if (!mod || typeof mod !== 'object')
-    throw new Error("Invalid modifier. Modifier must be an object.");
+  if (!mod || typeof mod !== 'object') {
+    const error = new Error("Invalid modifier. Modifier must be an object.");
+
+    if (callback) {
+      return callback(error);
+    } else {
+      throw error;
+    }
+  }
 
   if (!(LocalCollection._isPlainObject(mod) &&
         !EJSON._isCustomType(mod))) {
-    throw new Error(
-      "Only plain objects may be used as replacement" +
+    const error = new Error(
+        "Only plain objects may be used as replacement" +
         " documents in MongoDB");
+
+    if (callback) {
+      return callback(error);
+    } else {
+      throw error;
+    }
   }
 
   if (!options) options = {};

packages/mongo/observe_multiplex.js

@@ -185,7 +185,7 @@ _.extend(ObserveMultiplexer.prototype, {
   },
   _applyCallback: function (callbackName, args) {
     var self = this;
-    self._queue.queueTask(function () {
+    self._queue.queueTask(async function () {
       // If we stopped in the meantime, do nothing.
       if (!self._handles)
         return;
@@ -205,15 +205,17 @@ _.extend(ObserveMultiplexer.prototype, {
       // can continue until these are done. (But we do have to be careful to not
       // use a handle that got removed, because removeHandle does not use the
       // queue; thus, we iterate over an array of keys that we control.)
-      _.each(_.keys(self._handles), function (handleId) {
+      const toAwait = Object.keys(self._handles).map(async (handleId) => {
         var handle = self._handles && self._handles[handleId];
         if (!handle)
           return;
         var callback = handle['_' + callbackName];
         // clone arguments so that callbacks can mutate their arguments
-        callback && callback.apply(null,
-          handle.nonMutatingCallbacks ? args : EJSON.clone(args));
+        callback && await callback.apply(null,
+            handle.nonMutatingCallbacks ? args : EJSON.clone(args));
       });
+
+      await Promise.all(toAwait);
     });
   },
 

packages/test-helpers/async_multi.js

@@ -195,3 +195,16 @@ pollUntil = function (expect, f, timeout, step, noFail) {
     step
   );
 };
+
+runAndThrowIfNeeded = async (fn, test, shouldErrorOut) => {
+  let err, result;
+  try {
+    result = await fn();
+  } catch (e) {
+    err = e;
+  }
+
+  test[shouldErrorOut ? "isTrue" : "isFalse"](err);
+
+  return result;
+};

packages/test-helpers/package.js

@@ -28,7 +28,7 @@ Package.onUse(function (api) {
     'SeededRandom', 'clickElement', 'blurElement',
     'focusElement', 'simulateEvent', 'getStyleProperty', 'canonicalizeHtml',
     'renderToDiv', 'clickIt',
-    'withCallbackLogger', 'testAsyncMulti', 'simplePoll',
+    'withCallbackLogger', 'testAsyncMulti', 'simplePoll', 'runAndThrowIfNeeded',
     'makeTestConnection', 'DomUtils']);
 
   api.addFiles('try_all_permutations.js');

packages/tinytest/tinytest.js

@@ -240,6 +240,52 @@ export class TestCaseResults {
       });
   }
 
+  /**
+   * Same as throw, but accepts an async function as a parameter.
+   * @param f
+   * @param expected
+   * @param message
+   * @returns {Promise<void>}
+   */
+  async throwsAsync(f, expected, message) {
+    var actual, predicate;
+
+    if (expected === undefined) {
+      predicate = function (actual) {
+        return true;
+      };
+    } else if (typeof expected === "string") {
+      predicate = function (actual) {
+        return typeof actual.message === "string" &&
+            actual.message.indexOf(expected) !== -1;
+      };
+    } else if (expected instanceof RegExp) {
+      predicate = function (actual) {
+        return expected.test(actual.message);
+      };
+    } else if (typeof expected === 'function') {
+      predicate = expected;
+    } else {
+      throw new Error('expected should be a string, regexp, or predicate function');
+    }
+
+    try {
+      await f();
+    } catch (exception) {
+      actual = exception;
+    }
+
+    if (actual && predicate(actual))
+      this.ok();
+    else
+      this.fail({
+        type: "throws",
+        message: (actual ?
+            "wrong error thrown: " + actual.message :
+            "did not throw an error as expected") + (message ? ": " + message : ""),
+      });
+  }
+
   isTrue(v, msg) {
     if (v)
       this.ok();