Test that reverting df2820 fixed #2275.

Make some instances of #2315 into errors instead of silent early
returns.

Specifically, observeChanges calls (with added or addedBefore callbacks)
from within another observeChanges callback on the same collection will
be unable to differentiate between initial and later added/addedBefore
calls, which is serious enough to be an error (see #2315 for details).

We don't currently think that the other effect of #2315 (where observe
callbacks triggered by insert/remove/update/resumeObservers will not
occur before those methods return, if they are called reentrantly) is
problematic enough to deserve this sort of error.

packages/minimongo/minimongo.js

@@ -339,21 +339,6 @@ _.extend(LocalCollection.Cursor.prototype, {
       query.movedBefore = wrapCallback(options.movedBefore);
     }
 
-    if (!options._suppress_initial && !self.collection.paused) {
-      // XXX unify ordered and unordered interface
-      var each = ordered
-            ? _.bind(_.each, null, query.results)
-            : _.bind(query.results.forEach, query.results);
-      each(function (doc) {
-        var fields = EJSON.clone(doc);
-
-        delete fields._id;
-        if (ordered)
-          query.addedBefore(doc._id, fields, null);
-        query.added(doc._id, fields);
-      });
-    }
-
     var handle = new LocalCollection.ObserveHandle;
     _.extend(handle, {
       collection: self.collection,
@@ -373,9 +358,30 @@ _.extend(LocalCollection.Cursor.prototype, {
         handle.stop();
       });
     }
-    // run the observe callbacks resulting from the initial contents
-    // before we leave the observe.
-    self.collection._observeQueue.drain();
+
+    if (!options._suppress_initial && !self.collection.paused) {
+      // XXX unify ordered and unordered interface
+      var each = ordered
+            ? _.bind(_.each, null, query.results)
+            : _.bind(query.results.forEach, query.results);
+      each(function (doc) {
+        var fields = EJSON.clone(doc);
+
+        delete fields._id;
+        if (ordered)
+          query.addedBefore(doc._id, fields, null);
+        query.added(doc._id, fields);
+      });
+
+      // run the observe callbacks resulting from the initial contents
+      // before we leave the observe.
+      if (self.collection._observeQueue.safeToRunTask()) {
+        self.collection._observeQueue.drain();
+      } else if (options.added || options.addedBefore) {
+        // See #2315.
+        throw Error("observeChanges called from an observe callback on the same collection cannot differentiate between initial and later adds");
+      }
+    }
 
     return handle;
   }

packages/minimongo/minimongo_tests.js

@@ -3086,3 +3086,51 @@ Tinytest.add("minimongo - $near operator tests", function (test) {
   handle.stop();
 });
 
+// See #2275.
+Tinytest.add("minimongo - fetch in observe", function (test) {
+  var coll = new LocalCollection;
+  var callbackInvoked = false;
+  var observe = coll.find().observeChanges({
+    added: function (id, fields) {
+      callbackInvoked = true;
+      test.equal(fields, {foo: 1});
+      var doc = coll.findOne({foo: 1});
+      test.isTrue(doc);
+      test.equal(doc.foo, 1);
+    }
+  });
+  test.isFalse(callbackInvoked);
+  var computation = Deps.autorun(function (computation) {
+    if (computation.firstRun) {
+      coll.insert({foo: 1});
+    }
+  });
+  test.isTrue(callbackInvoked);
+  observe.stop();
+  computation.stop();
+});
+
+Tinytest.add("minimongo - observe in observe", function (test) {
+  var coll = new LocalCollection;
+  coll.insert({foo: 2});
+
+  var observe1AddedCalled = false;
+  var observe1 = coll.find({foo: 1}).observeChanges({
+    added: function (id, fields) {
+      observe1AddedCalled = true;
+      test.equal(fields, {foo: 1});
+
+      // It would be even better if this didn't throw; see #2315.
+      test.throws(function () {
+        coll.find({foo: 2}).observeChanges({
+          added: function () {
+          }
+        });
+      });
+    }
+  });
+  test.isFalse(observe1AddedCalled);
+  coll.insert({foo: 1});
+  test.isTrue(observe1AddedCalled);
+  observe1.stop();
+});