Don't try to open the Realm for notifications on the source queue

This happened to work when no migration was needed but hits some data races and
isn't something we can make work reliably.

Realm/RLMCollection.mm

@@ -434,15 +434,21 @@ - (void)invalidate {
         return token;
     }
 
-    RLMRealm *targetRealm = [RLMRealm realmWithConfiguration:realm.configuration queue:queue error:nil];
     RLMThreadSafeReference *tsr = [RLMThreadSafeReference referenceWithThreadConfined:collection];
-    token->_realm = targetRealm;
+    token->_realm = realm;
+    RLMRealmConfiguration *config = realm.configuration;
     dispatch_async(queue, ^{
         std::lock_guard<std::mutex> lock(token->_mutex);
         if (!token->_realm) {
             return;
         }
-        RLMCollection *collection = [targetRealm resolveThreadSafeReference:tsr];
+        NSError *error;
+        RLMRealm *realm = token->_realm = [RLMRealm realmWithConfiguration:config queue:queue error:&error];
+        if (!realm) {
+            block(nil, nil, error);
+            return;
+        }
+        RLMCollection *collection = [realm resolveThreadSafeReference:tsr];
         token->_token = RLMGetBackingCollection(collection).add_notification_callback(CollectionCallbackWrapper{block, collection, skip});
     });
     return token;

Realm/RLMObjectBase.mm

@@ -616,14 +616,31 @@ - (void)invalidate {
     _object = {};
 }
 
-- (void)addNotificationBlock:(RLMObjectNotificationCallback)block object:(RLMObjectBase *)obj {
+- (void)addNotificationBlock:(RLMObjectNotificationCallback)block
+         threadSafeReference:(RLMThreadSafeReference *)tsr
+                      config:(RLMRealmConfiguration *)config
+                       queue:(dispatch_queue_t)queue {
     std::lock_guard<std::mutex> lock(_mutex);
     if (!_realm) {
         // Token was invalidated before we got this far
         return;
     }
 
+    NSError *error;
+    RLMRealm *realm = _realm = [RLMRealm realmWithConfiguration:config queue:queue error:&error];
+    if (!realm) {
+        block(nil, nil, nil, error);
+        return;
+    }
+    RLMObjectBase *obj = [realm resolveThreadSafeReference:tsr];
+
+    _object = realm::Object(obj->_realm->_realm, *obj->_info->objectSchema, obj->_row);
+    _token = _object.add_notification_callback(ObjectChangeCallbackWrapper{block, obj});
+}
+
+- (void)addNotificationBlock:(RLMObjectNotificationCallback)block object:(RLMObjectBase *)obj {
     _object = realm::Object(obj->_realm->_realm, *obj->_info->objectSchema, obj->_row);
+    _realm = obj->_realm;
     _token = _object.add_notification_callback(ObjectChangeCallbackWrapper{block, obj});
 }
 
@@ -641,14 +658,13 @@ - (void)addNotificationBlock:(RLMObjectNotificationCallback)block object:(RLMObj
         return token;
     }
 
-    RLMRealm *targetRealm = [RLMRealm realmWithConfiguration:obj->_realm.configuration queue:queue error:nil];
     RLMThreadSafeReference *tsr = [RLMThreadSafeReference referenceWithThreadConfined:(id)obj];
     auto token = [[RLMObjectNotificationToken alloc] init];
     token->_realm = obj->_realm;
+    RLMRealmConfiguration *config = obj->_realm.configuration;
     dispatch_async(queue, ^{
         @autoreleasepool {
-            RLMObjectBase *obj = [targetRealm resolveThreadSafeReference:tsr];
-            [token addNotificationBlock:block object:obj];
+            [token addNotificationBlock:block threadSafeReference:tsr config:config queue:queue];
         }
     });
     return token;

Realm/RLMRealm.mm

@@ -492,6 +492,9 @@ + (instancetype)realmWithConfiguration:(RLMRealmConfiguration *)configuration
             else {
                 config.scheduler = realm::util::Scheduler::make_dispatch((__bridge void *)queue);
             }
+            if (!config.scheduler->is_on_thread()) {
+                throw RLMException(@"Realm opened from incorrect dispatch queue.");
+            }
         }
         realm->_realm = Realm::get_shared_realm(config);
     }

Realm/Tests/ArrayPropertyTests.m

@@ -315,6 +315,7 @@ - (void)testUnmanagedPrimitive {
 
     [obj.intObj addObject:@5];
     XCTAssertEqualObjects(obj.intObj.firstObject, @5);
+    [realm cancelWriteTransaction];
 }
 
 - (void)testReplaceObjectAtIndexInUnmanagedArray {

Realm/Tests/AsyncTests.mm

@@ -843,14 +843,14 @@ - (void)testAddNotificationBlockFromWrongThread {
 
 - (void)testAddNotificationBlockFromWrongQueue {
     auto queue = dispatch_queue_create("background queue", DISPATCH_QUEUE_SERIAL);
-    RLMRealm *realm = [RLMRealm defaultRealmForQueue:queue];
     __block RLMResults *results;
-    dispatch_sync(queue, ^{ results = [IntObject allObjectsInRealm:realm]; });
-    [self dispatchAsyncAndWait:^{
-        XCTAssertThrows([results addNotificationBlock:^(RLMResults *results, RLMCollectionChange *change, NSError *error) {
-            XCTFail(@"should not be called");
-        }]);
-    }];
+    dispatch_sync(queue, ^{
+        RLMRealm *realm = [RLMRealm defaultRealmForQueue:queue];
+        results = [IntObject allObjectsInRealm:realm];
+    });
+    XCTAssertThrows([results addNotificationBlock:^(RLMResults *results, RLMCollectionChange *change, NSError *error) {
+        XCTFail(@"should not be called");
+    }]);
 }
 
 - (void)testRemoveNotificationBlockFromWrongThread {
@@ -995,11 +995,11 @@ - (void)testInitialResultDiscardsChanges {
 
 - (void)testNotificationDeliveryToQueue {
     RLMRealm *realm = [RLMRealm defaultRealm];
-    RLMRealm *bgRealm = [RLMRealm defaultRealmForQueue:self.bgQueue];
     __block RLMNotificationToken *token;
     dispatch_semaphore_t sema = dispatch_semaphore_create(0);
 
     [self dispatchAsync:^{
+        RLMRealm *bgRealm = [RLMRealm defaultRealmForQueue:self.bgQueue];
         token = [[IntObject allObjectsInRealm:bgRealm] addNotificationBlock:^(RLMResults *results, RLMCollectionChange *, NSError *) {
             XCTAssertNotNil(results);
             XCTAssertNoThrow(results.count);

Realm/Tests/NotificationTests.m

@@ -1079,6 +1079,9 @@ - (void)testMultipleObjectNotifiers {
         XCTFail(@"notification block for wrong object called");
     }];
 
+    // Ensure initial notification is processed so that the change can report previousValue
+    [_obj.realm transactionWithBlock:^{}];
+
     [self dispatchAsync:^{
         RLMRealm *realm = [RLMRealm defaultRealm];
         AllTypesObject *obj = [[AllTypesObject allObjectsInRealm:realm] firstObject];

Realm/Tests/ObjectCreationTests.mm

@@ -1416,6 +1416,7 @@ - (void)testAddOrUpdateWithoutPKThrows {
 
     RLMAssertThrowsWithReason([realm addOrUpdateObject:[DogObject new]],
                               @"'DogObject' does not have a primary key");
+    [realm cancelWriteTransaction];
 }
 
 - (void)testAddOrUpdateUpdatesExistingItemWithSamePK {

Realm/Tests/RealmTests.mm

@@ -1532,8 +1532,10 @@ - (void)testQueueBoundRealmCaching {
 
     RLMRealm *mainThreadRealm1 = [RLMRealm defaultRealm];
     RLMRealm *mainQueueRealm1 = [RLMRealm defaultRealmForQueue:dispatch_get_main_queue()];
-    RLMRealm *q1Realm1 = [RLMRealm defaultRealmForQueue:q1];
-    RLMRealm *q2Realm1 = [RLMRealm defaultRealmForQueue:q2];
+    __block RLMRealm *q1Realm1;
+    __block RLMRealm *q2Realm1;
+    dispatch_sync(q1, ^{ q1Realm1 = [RLMRealm defaultRealmForQueue:q1]; });
+    dispatch_sync(q2, ^{ q2Realm1 = [RLMRealm defaultRealmForQueue:q2]; });
 
     XCTAssertEqual(mainQueueRealm1, mainThreadRealm1);
 
@@ -1545,8 +1547,10 @@ - (void)testQueueBoundRealmCaching {
 
     RLMRealm *mainThreadRealm2 = [RLMRealm defaultRealm];
     RLMRealm *mainQueueRealm2 = [RLMRealm defaultRealmForQueue:dispatch_get_main_queue()];
-    RLMRealm *q1Realm2 = [RLMRealm defaultRealmForQueue:q1];
-    RLMRealm *q2Realm2 = [RLMRealm defaultRealmForQueue:q2];
+    __block RLMRealm *q1Realm2;
+    __block RLMRealm *q2Realm2;
+    dispatch_sync(q1, ^{ q1Realm2 = [RLMRealm defaultRealmForQueue:q1]; });
+    dispatch_sync(q2, ^{ q2Realm2 = [RLMRealm defaultRealmForQueue:q2]; });
 
     XCTAssertEqual(mainThreadRealm1, mainThreadRealm2);
     XCTAssertEqual(mainQueueRealm1, mainQueueRealm2);
@@ -1572,18 +1576,26 @@ - (void)testQueueBoundRealmCaching {
 
 - (void)testQueueValidation {
     XCTAssertNoThrow([RLMRealm defaultRealmForQueue:dispatch_get_main_queue()]);
-    XCTAssertNoThrow([RLMRealm defaultRealmForQueue:self.bgQueue]);
-    XCTAssertThrows([RLMRealm defaultRealmForQueue:dispatch_get_global_queue(0, 0)]);
-    XCTAssertThrows([RLMRealm defaultRealmForQueue:dispatch_queue_create("concurrent queue", DISPATCH_QUEUE_CONCURRENT)]);
+    RLMAssertThrowsWithReason([RLMRealm defaultRealmForQueue:self.bgQueue],
+                              @"Realm opened from incorrect dispatch queue.");
+    RLMAssertThrowsWithReasonMatching([RLMRealm defaultRealmForQueue:dispatch_get_global_queue(0, 0)],
+                              @"Invalid queue '.*' \\(.*\\): Realms can only be confined to serial queues or the main queue.");
+    RLMAssertThrowsWithReason([RLMRealm defaultRealmForQueue:dispatch_queue_create("concurrent queue", DISPATCH_QUEUE_CONCURRENT)],
+                              @"Invalid queue 'concurrent queue' (OS_dispatch_queue_concurrent): Realms can only be confined to serial queues or the main queue.");
+    dispatch_sync(self.bgQueue, ^{
+        XCTAssertNoThrow([RLMRealm defaultRealmForQueue:self.bgQueue]);
+    });
 }
 
 - (void)testQueueChecking {
     auto q1 = dispatch_queue_create("queue 1", DISPATCH_QUEUE_SERIAL);
     auto q2 = dispatch_queue_create("queue 2", DISPATCH_QUEUE_SERIAL);
 
     RLMRealm *mainRealm = [RLMRealm defaultRealmForQueue:dispatch_get_main_queue()];
-    RLMRealm *q1Realm = [RLMRealm defaultRealmForQueue:q1];
-    RLMRealm *q2Realm = [RLMRealm defaultRealmForQueue:q2];
+    __block RLMRealm *q1Realm;
+    __block RLMRealm *q2Realm;
+    dispatch_sync(q1, ^{ q1Realm = [RLMRealm defaultRealmForQueue:q1]; });
+    dispatch_sync(q2, ^{ q2Realm = [RLMRealm defaultRealmForQueue:q2]; });
 
     XCTAssertNoThrow([mainRealm refresh]);
     RLMAssertThrowsWithReason([q1Realm refresh], @"thread");