diff --git a/packages/relay-runtime/store/RelayModernStore.js b/packages/relay-runtime/store/RelayModernStore.js
index 112f6d1d89da5..32b60a6c1e0ec 100644
--- a/packages/relay-runtime/store/RelayModernStore.js
+++ b/packages/relay-runtime/store/RelayModernStore.js
@@ -13,6 +13,7 @@
 'use strict';
 
 const DataChecker = require('./DataChecker');
+const RelayFeatureFlags = require('../util/RelayFeatureFlags');
 const RelayModernRecord = require('./RelayModernRecord');
 const RelayOptimisticRecordSource = require('./RelayOptimisticRecordSource');
 const RelayProfiler = require('../util/RelayProfiler');
@@ -20,13 +21,12 @@ const RelayReader = require('./RelayReader');
 const RelayReferenceMarker = require('./RelayReferenceMarker');
 const RelayStoreReactFlightUtils = require('./RelayStoreReactFlightUtils');
 const RelayStoreSubscriptions = require('./RelayStoreSubscriptions');
+const RelayStoreSubscriptionsUsingMapByID = require('./RelayStoreSubscriptionsUsingMapByID');
 const RelayStoreUtils = require('./RelayStoreUtils');
 
 const deepFreeze = require('../util/deepFreeze');
 const defaultGetDataID = require('./defaultGetDataID');
-const hasOverlappingIDs = require('./hasOverlappingIDs');
 const invariant = require('invariant');
-const recycleNodesInto = require('../util/recycleNodesInto');
 const resolveImmediate = require('../util/resolveImmediate');
 
 const {ROOT_ID, ROOT_TYPE} = require('./RelayStoreUtils');
@@ -47,6 +47,7 @@ import type {
   SingularReaderSelector,
   Snapshot,
   Store,
+  StoreSubscriptions,
   UpdatedRecords,
 } from './RelayStoreTypes';
 
@@ -100,7 +101,7 @@ class RelayModernStore implements Store {
     |},
   >;
   _shouldScheduleGC: boolean;
-  _storeSubscriptions: RelayStoreSubscriptions;
+  _storeSubscriptions: StoreSubscriptions;
   _updatedRecordIDs: UpdatedRecords;
 
   constructor(
@@ -143,7 +144,10 @@ class RelayModernStore implements Store {
     this._releaseBuffer = [];
     this._roots = new Map();
     this._shouldScheduleGC = false;
-    this._storeSubscriptions = new RelayStoreSubscriptions();
+    this._storeSubscriptions =
+      RelayFeatureFlags.ENABLE_STORE_SUBSCRIPTIONS_REFACTOR === true
+        ? new RelayStoreSubscriptionsUsingMapByID()
+        : new RelayStoreSubscriptions();
     this._updatedRecordIDs = {};
 
     initializeRecordSource(this._recordSource);
diff --git a/packages/relay-runtime/store/RelayStoreSubscriptions.js b/packages/relay-runtime/store/RelayStoreSubscriptions.js
index e2ad6326c53c0..68b9b7fb97b68 100644
--- a/packages/relay-runtime/store/RelayStoreSubscriptions.js
+++ b/packages/relay-runtime/store/RelayStoreSubscriptions.js
@@ -24,17 +24,18 @@ import type {
   RecordSource,
   RequestDescriptor,
   Snapshot,
+  StoreSubscriptions,
   UpdatedRecords,
 } from './RelayStoreTypes';
 
-export type Subscription = {|
+type Subscription = {|
   callback: (snapshot: Snapshot) => void,
   snapshot: Snapshot,
   stale: boolean,
   backup: ?Snapshot,
 |};
 
-class RelayStoreSubscriptions {
+class RelayStoreSubscriptions implements StoreSubscriptions {
   _subscriptions: Set<Subscription>;
 
   constructor() {
@@ -119,8 +120,14 @@ class RelayStoreSubscriptions {
     });
   }
 
-  // Returns the owner (RequestDescriptor) if the subscription was affected by the
-  // latest update, or null if it was not affected.
+  /**
+   * Notifies the callback for the subscription if the data for the associated
+   * snapshot has changed.
+   * Additionally, updates the subscription snapshot with the latest snapshot,
+   * and marks it as not stale.
+   * Returns the owner (RequestDescriptor) if the subscription was affected by the
+   * latest update, or null if it was not affected.
+   */
   _updateSubscription(
     source: RecordSource,
     subscription: Subscription,
diff --git a/packages/relay-runtime/store/RelayStoreSubscriptionsUsingMapByID.js b/packages/relay-runtime/store/RelayStoreSubscriptionsUsingMapByID.js
new file mode 100644
index 0000000000000..74a3d4111dfea
--- /dev/null
+++ b/packages/relay-runtime/store/RelayStoreSubscriptionsUsingMapByID.js
@@ -0,0 +1,259 @@
+/**
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ *
+ * @flow strict-local
+ * @format
+ */
+
+// flowlint ambiguous-object-type:error
+
+'use strict';
+
+const RelayReader = require('./RelayReader');
+
+const deepFreeze = require('../util/deepFreeze');
+const recycleNodesInto = require('../util/recycleNodesInto');
+
+import type {DataID, Disposable} from '../util/RelayRuntimeTypes';
+import type {
+  RecordMap,
+  RecordSource,
+  RequestDescriptor,
+  Snapshot,
+  StoreSubscriptions,
+  UpdatedRecords,
+} from './RelayStoreTypes';
+
+type Subscription = {|
+  backup: ?Snapshot,
+  callback: (snapshot: Snapshot) => void,
+  notifiedRevision: number,
+  snapshot: Snapshot,
+  snapshotRevision: number,
+|};
+
+class RelayStoreSubscriptionsUsingMapByID implements StoreSubscriptions {
+  _notifiedRevision: number;
+  _snapshotRevision: number;
+  _subscriptionsByDataId: Map<DataID, Set<Subscription>>;
+  _staleSubscriptions: Set<Subscription>;
+
+  constructor() {
+    this._notifiedRevision = 0;
+    this._snapshotRevision = 0;
+    this._subscriptionsByDataId = new Map();
+    this._staleSubscriptions = new Set();
+  }
+
+  subscribe(
+    snapshot: Snapshot,
+    callback: (snapshot: Snapshot) => void,
+  ): Disposable {
+    const subscription = {
+      backup: null,
+      callback,
+      notifiedRevision: this._notifiedRevision,
+      snapshotRevision: this._snapshotRevision,
+      snapshot,
+    };
+    const dispose = () => {
+      for (const dataId in snapshot.seenRecords) {
+        const subscriptionsForDataId = this._subscriptionsByDataId.get(dataId);
+        if (subscriptionsForDataId != null) {
+          subscriptionsForDataId.delete(subscription);
+          if (subscriptionsForDataId.size === 0) {
+            this._subscriptionsByDataId.delete(dataId);
+          }
+        }
+      }
+    };
+
+    for (const dataId in snapshot.seenRecords) {
+      const subscriptionsForDataId = this._subscriptionsByDataId.get(dataId);
+      if (subscriptionsForDataId != null) {
+        subscriptionsForDataId.add(subscription);
+      } else {
+        this._subscriptionsByDataId.set(dataId, new Set([subscription]));
+      }
+    }
+
+    return {dispose};
+  }
+
+  snapshotSubscriptions(source: RecordSource) {
+    this._snapshotRevision++;
+    this._subscriptionsByDataId.forEach(subscriptions => {
+      subscriptions.forEach(subscription => {
+        if (subscription.snapshotRevision === this._snapshotRevision) {
+          return;
+        }
+        subscription.snapshotRevision = this._snapshotRevision;
+
+        // Backup occurs after writing a new "final" payload(s) and before (re)applying
+        // optimistic changes. Each subscription's `snapshot` represents what was *last
+        // published to the subscriber*, which notably may include previous optimistic
+        // updates. Therefore a subscription can be in any of the following states:
+        // - stale=true: This subscription was restored to a different value than
+        //   `snapshot`. That means this subscription has changes relative to its base,
+        //   but its base has changed (we just applied a final payload): recompute
+        //   a backup so that we can later restore to the state the subscription
+        //   should be in.
+        // - stale=false: This subscription was restored to the same value than
+        //   `snapshot`. That means this subscription does *not* have changes relative
+        //   to its base, so the current `snapshot` is valid to use as a backup.
+        if (!this._staleSubscriptions.has(subscription)) {
+          subscription.backup = subscription.snapshot;
+          return;
+        }
+        const snapshot = subscription.snapshot;
+        const backup = RelayReader.read(source, snapshot.selector);
+        const nextData = recycleNodesInto(snapshot.data, backup.data);
+        (backup: $FlowFixMe).data = nextData; // backup owns the snapshot and can safely mutate
+        subscription.backup = backup;
+      });
+    });
+  }
+
+  restoreSubscriptions() {
+    this._snapshotRevision++;
+    this._subscriptionsByDataId.forEach(subscriptions => {
+      subscriptions.forEach(subscription => {
+        if (subscription.snapshotRevision === this._snapshotRevision) {
+          return;
+        }
+        subscription.snapshotRevision = this._snapshotRevision;
+
+        const backup = subscription.backup;
+        subscription.backup = null;
+        if (backup) {
+          if (backup.data !== subscription.snapshot.data) {
+            this._staleSubscriptions.add(subscription);
+          }
+          const prevSeenRecords = subscription.snapshot.seenRecords;
+          subscription.snapshot = {
+            data: subscription.snapshot.data,
+            isMissingData: backup.isMissingData,
+            seenRecords: backup.seenRecords,
+            selector: backup.selector,
+            missingRequiredFields: backup.missingRequiredFields,
+          };
+          this._updateSubscriptionsMap(subscription, prevSeenRecords);
+        } else {
+          this._staleSubscriptions.add(subscription);
+        }
+      });
+    });
+  }
+
+  updateSubscriptions(
+    source: RecordSource,
+    updatedRecordIDs: UpdatedRecords,
+    updatedOwners: Array<RequestDescriptor>,
+  ) {
+    this._notifiedRevision++;
+    Object.keys(updatedRecordIDs).forEach(updatedRecordId => {
+      const subcriptionsForDataId = this._subscriptionsByDataId.get(
+        updatedRecordId,
+      );
+      if (subcriptionsForDataId == null) {
+        return;
+      }
+      subcriptionsForDataId.forEach(subscription => {
+        if (subscription.notifiedRevision === this._notifiedRevision) {
+          return;
+        }
+        const owner = this._updateSubscription(source, subscription, false);
+        if (owner != null) {
+          updatedOwners.push(owner);
+        }
+      });
+    });
+    this._staleSubscriptions.forEach(subscription => {
+      if (subscription.notifiedRevision === this._notifiedRevision) {
+        return;
+      }
+      const owner = this._updateSubscription(source, subscription, true);
+      if (owner != null) {
+        updatedOwners.push(owner);
+      }
+    });
+    this._staleSubscriptions.clear();
+  }
+
+  /**
+   * Notifies the callback for the subscription if the data for the associated
+   * snapshot has changed.
+   * Additionally, updates the subscription snapshot with the latest snapshot,
+   * amarks it as not stale, and updates the subscription tracking for any
+   * any new ids observed in the latest data snapshot.
+   * Returns the owner (RequestDescriptor) if the subscription was affected by the
+   * latest update, or null if it was not affected.
+   */
+  _updateSubscription(
+    source: RecordSource,
+    subscription: Subscription,
+    stale: boolean,
+  ): ?RequestDescriptor {
+    const {backup, callback, snapshot} = subscription;
+    let nextSnapshot: Snapshot =
+      stale && backup != null
+        ? backup
+        : RelayReader.read(source, snapshot.selector);
+    const nextData = recycleNodesInto(snapshot.data, nextSnapshot.data);
+    nextSnapshot = ({
+      data: nextData,
+      isMissingData: nextSnapshot.isMissingData,
+      seenRecords: nextSnapshot.seenRecords,
+      selector: nextSnapshot.selector,
+      missingRequiredFields: nextSnapshot.missingRequiredFields,
+    }: Snapshot);
+    if (__DEV__) {
+      deepFreeze(nextSnapshot);
+    }
+
+    const prevSeenRecords = subscription.snapshot.seenRecords;
+    subscription.snapshot = nextSnapshot;
+    subscription.notifiedRevision = this._notifiedRevision;
+    this._updateSubscriptionsMap(subscription, prevSeenRecords);
+
+    if (nextSnapshot.data !== snapshot.data) {
+      callback(nextSnapshot);
+      return snapshot.selector.owner;
+    }
+  }
+
+  /**
+   * Updates the Map that tracks subscriptions by id.
+   * Given an updated subscription and the records that where seen
+   * on the previous subscription snapshot, updates our tracking
+   * to track the subscription for the newly and no longer seen ids.
+   */
+  _updateSubscriptionsMap(
+    subscription: Subscription,
+    prevSeenRecords: RecordMap,
+  ) {
+    for (const dataId in prevSeenRecords) {
+      const subscriptionsForDataId = this._subscriptionsByDataId.get(dataId);
+      if (subscriptionsForDataId != null) {
+        subscriptionsForDataId.delete(subscription);
+        if (subscriptionsForDataId.size === 0) {
+          this._subscriptionsByDataId.delete(dataId);
+        }
+      }
+    }
+
+    for (const dataId in subscription.snapshot.seenRecords) {
+      const subscriptionsForDataId = this._subscriptionsByDataId.get(dataId);
+      if (subscriptionsForDataId != null) {
+        subscriptionsForDataId.add(subscription);
+      } else {
+        this._subscriptionsByDataId.set(dataId, new Set([subscription]));
+      }
+    }
+  }
+}
+
+module.exports = RelayStoreSubscriptionsUsingMapByID;
diff --git a/packages/relay-runtime/store/RelayStoreTypes.js b/packages/relay-runtime/store/RelayStoreTypes.js
index f1e536a6c8a68..e7f462a8e1b58 100644
--- a/packages/relay-runtime/store/RelayStoreTypes.js
+++ b/packages/relay-runtime/store/RelayStoreTypes.js
@@ -333,6 +333,40 @@ export interface Store {
   ): Disposable;
 }
 
+export interface StoreSubscriptions {
+  /**
+   * Subscribe to changes to the results of a selector. The callback is called
+   * when `updateSubscriptions()` is called *and* records have been published that affect the
+   * selector results relative to the last update.
+   */
+  subscribe(
+    snapshot: Snapshot,
+    callback: (snapshot: Snapshot) => void,
+  ): Disposable;
+
+  /**
+   * Record a backup/snapshot of the current state of the subscriptions.
+   * This state can be restored with restore().
+   */
+  snapshotSubscriptions(source: RecordSource): void;
+
+  /**
+   * Reset the state of the subscriptions to the point that snapshot() was last called.
+   */
+  restoreSubscriptions(): void;
+
+  /**
+   * Notifies each subscription if the snapshot for the subscription selector has changed.
+   * Mutates the updatedOwners array with any owners (RequestDescriptors) associated
+   * with the subscriptions that were notifed; i.e. the owners affected by the changes.
+   */
+  updateSubscriptions(
+    source: RecordSource,
+    updatedRecordIDs: UpdatedRecords,
+    updatedOwners: Array<RequestDescriptor>,
+  ): void;
+}
+
 /**
  * A type that accepts a callback and schedules it to run at some future time.
  * By convention, implementations should not execute the callback immediately.
diff --git a/packages/relay-runtime/store/__tests__/RelayModernStore-WithSubscriptionsUsingMapByID-test.js b/packages/relay-runtime/store/__tests__/RelayModernStore-WithSubscriptionsUsingMapByID-test.js
new file mode 100644
index 0000000000000..32839151c43fe
--- /dev/null
+++ b/packages/relay-runtime/store/__tests__/RelayModernStore-WithSubscriptionsUsingMapByID-test.js
@@ -0,0 +1,714 @@
+/**
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ *
+ * @format
+ * @flow strict-local
+ * @emails oncall+relay
+ */
+
+// flowlint ambiguous-object-type:error
+
+'use strict';
+
+const RelayFeatureFlags = require('../../util/RelayFeatureFlags');
+const RelayModernRecord = require('../RelayModernRecord');
+const RelayModernStore = require('../RelayModernStore');
+const RelayOptimisticRecordSource = require('../RelayOptimisticRecordSource');
+const RelayRecordSourceMapImpl = require('../RelayRecordSourceMapImpl');
+
+const {getRequest} = require('../../query/GraphQLTag');
+const {
+  createOperationDescriptor,
+} = require('../RelayModernOperationDescriptor');
+const {createReaderSelector} = require('../RelayModernSelector');
+const {INVALIDATED_AT_KEY, REF_KEY} = require('../RelayStoreUtils');
+const {generateAndCompile} = require('relay-test-utils-internal');
+
+function assertIsDeeplyFrozen(value: ?{...} | ?$ReadOnlyArray<{...}>) {
+  if (!value) {
+    throw new Error(
+      'Expected value to be a non-null object or array of objects',
+    );
+  }
+  expect(Object.isFrozen(value)).toBe(true);
+  if (Array.isArray(value)) {
+    value.forEach(item => assertIsDeeplyFrozen(item));
+  } else if (typeof value === 'object' && value !== null) {
+    for (const key in value) {
+      assertIsDeeplyFrozen(value[key]);
+    }
+  }
+}
+
+[
+  [data => new RelayRecordSourceMapImpl(data), 'Map'],
+  [
+    data =>
+      RelayOptimisticRecordSource.create(new RelayRecordSourceMapImpl(data)),
+    'Optimistic',
+  ],
+].forEach(([getRecordSourceImplementation, ImplementationName]) => {
+  beforeEach(() => {
+    RelayFeatureFlags.ENABLE_STORE_SUBSCRIPTIONS_REFACTOR = true;
+  });
+
+  afterEach(() => {
+    RelayFeatureFlags.ENABLE_STORE_SUBSCRIPTIONS_REFACTOR = false;
+  });
+
+  describe(`Relay Store with ${ImplementationName} Record Source`, () => {
+    describe('notify/publish/subscribe', () => {
+      let UserQuery;
+      let UserFragment;
+      let data;
+      let source;
+      let store;
+      let logEvents;
+
+      beforeEach(() => {
+        data = {
+          '4': {
+            __id: '4',
+            id: '4',
+            __typename: 'User',
+            name: 'Zuck',
+            'profilePicture(size:32)': {[REF_KEY]: 'client:1'},
+            emailAddresses: ['a@b.com'],
+          },
+          'client:1': {
+            __id: 'client:1',
+            uri: 'https://photo1.jpg',
+          },
+          'client:root': {
+            __id: 'client:root',
+            __typename: '__Root',
+            'node(id:"4")': {__ref: '4'},
+            me: {__ref: '4'},
+          },
+        };
+        logEvents = [];
+        source = getRecordSourceImplementation(data);
+        store = new RelayModernStore(source, {
+          log: event => {
+            logEvents.push(event);
+          },
+        });
+        ({UserFragment, UserQuery} = generateAndCompile(`
+          fragment UserFragment on User {
+            name
+            profilePicture(size: $size) {
+              uri
+            }
+            emailAddresses
+          }
+
+          query UserQuery($size: Int) {
+            me {
+              ...UserFragment
+            }
+          }
+        `));
+      });
+
+      it('calls subscribers whose data has changed since previous notify', () => {
+        // subscribe(), publish(), notify() -> subscriber called
+        const owner = createOperationDescriptor(UserQuery, {});
+        const selector = createReaderSelector(
+          UserFragment,
+          '4',
+          {size: 32},
+          owner.request,
+        );
+        const snapshot = store.lookup(selector);
+        const callback = jest.fn();
+        store.subscribe(snapshot, callback);
+        // Publish a change to profilePicture.uri
+        const nextSource = getRecordSourceImplementation({
+          'client:1': {
+            __id: 'client:1',
+            uri: 'https://photo2.jpg',
+          },
+        });
+        store.publish(nextSource);
+        expect(callback).not.toBeCalled();
+        store.notify();
+        expect(callback.mock.calls.length).toBe(1);
+        expect(callback.mock.calls[0][0]).toEqual({
+          ...snapshot,
+          data: {
+            name: 'Zuck',
+            profilePicture: {
+              uri: 'https://photo2.jpg', // new uri
+            },
+            emailAddresses: ['a@b.com'],
+          },
+          seenRecords: {
+            '4': {
+              __id: '4',
+              id: '4',
+              __typename: 'User',
+              name: 'Zuck',
+              'profilePicture(size:32)': {[REF_KEY]: 'client:1'},
+              emailAddresses: ['a@b.com'],
+            },
+            'client:1': {
+              ...data['client:1'],
+              uri: 'https://photo2.jpg',
+            },
+          },
+        });
+      });
+
+      it('calls subscribers and reads data with fragment owner if one is available in subscription snapshot', () => {
+        // subscribe(), publish(), notify() -> subscriber called
+        ({UserQuery, UserFragment} = generateAndCompile(`
+          query UserQuery($size: Float!) {
+            me {
+              ...UserFragment
+            }
+          }
+
+          fragment UserFragment on User {
+            name
+            profilePicture(size: $size) {
+              uri
+            }
+            emailAddresses
+          }
+        `));
+        const queryNode = getRequest(UserQuery);
+        const owner = createOperationDescriptor(queryNode, {size: 32});
+        const selector = createReaderSelector(
+          UserFragment,
+          '4',
+          {size: 32},
+          owner.request,
+        );
+        const snapshot = store.lookup(selector);
+        expect(snapshot.selector).toBe(selector);
+
+        const callback = jest.fn();
+        store.subscribe(snapshot, callback);
+        // Publish a change to profilePicture.uri
+        const nextSource = getRecordSourceImplementation({
+          'client:1': {
+            __id: 'client:1',
+            uri: 'https://photo2.jpg',
+          },
+        });
+        store.publish(nextSource);
+        expect(callback).not.toBeCalled();
+        store.notify();
+        expect(callback.mock.calls.length).toBe(1);
+        expect(callback.mock.calls[0][0]).toEqual({
+          ...snapshot,
+          data: {
+            name: 'Zuck',
+            profilePicture: {
+              uri: 'https://photo2.jpg', // new uri
+            },
+            emailAddresses: ['a@b.com'],
+          },
+          seenRecords: {
+            '4': {
+              __id: '4',
+              id: '4',
+              __typename: 'User',
+              name: 'Zuck',
+              'profilePicture(size:32)': {[REF_KEY]: 'client:1'},
+              emailAddresses: ['a@b.com'],
+            },
+            'client:1': {
+              ...data['client:1'],
+              uri: 'https://photo2.jpg',
+            },
+          },
+        });
+        expect(callback.mock.calls[0][0].selector).toBe(selector);
+      });
+
+      it('vends deeply-frozen objects', () => {
+        const owner = createOperationDescriptor(UserQuery, {});
+        const selector = createReaderSelector(
+          UserFragment,
+          '4',
+          {size: 32},
+          owner.request,
+        );
+        const snapshot = store.lookup(selector);
+        const callback = jest.fn();
+        store.subscribe(snapshot, callback);
+        // Publish a change to profilePicture.uri
+        const nextSource = getRecordSourceImplementation({
+          'client:1': {
+            __id: 'client:1',
+            uri: 'https://photo2.jpg',
+          },
+        });
+        store.publish(nextSource);
+        store.notify();
+        expect(callback.mock.calls.length).toBe(1);
+        const nextSnapshot = callback.mock.calls[0][0];
+        expect(Object.isFrozen(nextSnapshot)).toBe(true);
+        assertIsDeeplyFrozen(nextSnapshot.data);
+        assertIsDeeplyFrozen(nextSnapshot.selector.variables);
+      });
+
+      it('calls affected subscribers only once', () => {
+        // subscribe(), publish(), publish(), notify() -> subscriber called once
+        const owner = createOperationDescriptor(UserQuery, {});
+        const selector = createReaderSelector(
+          UserFragment,
+          '4',
+          {size: 32},
+          owner.request,
+        );
+        const snapshot = store.lookup(selector);
+        const callback = jest.fn();
+        store.subscribe(snapshot, callback);
+        // Publish a change to profilePicture.uri
+        let nextSource = getRecordSourceImplementation({
+          '4': {
+            __id: '4',
+            __typename: 'User',
+            name: 'Mark',
+            emailAddresses: ['a@b.com', 'c@d.net'],
+          },
+          'client:1': {
+            __id: 'client:1',
+            uri: 'https://photo2.jpg',
+          },
+        });
+        store.publish(nextSource);
+        nextSource = getRecordSourceImplementation({
+          'client:1': {
+            __id: 'client:1',
+            uri: 'https://photo3.jpg',
+          },
+        });
+        store.publish(nextSource);
+        expect(callback).not.toBeCalled();
+        store.notify();
+        expect(callback.mock.calls.length).toBe(1);
+        expect(callback.mock.calls[0][0]).toEqual({
+          ...snapshot,
+          data: {
+            name: 'Mark',
+            profilePicture: {
+              uri: 'https://photo3.jpg', // most recent uri
+            },
+            emailAddresses: ['a@b.com', 'c@d.net'],
+          },
+          seenRecords: {
+            '4': {
+              ...data['4'],
+              name: 'Mark',
+              emailAddresses: ['a@b.com', 'c@d.net'],
+            },
+            'client:1': {
+              ...data['client:1'],
+              uri: 'https://photo3.jpg',
+            },
+          },
+        });
+      });
+
+      it('notifies subscribers and sets updated value for isMissingData', () => {
+        data = {
+          '4': {
+            __id: '4',
+            id: '4',
+            __typename: 'User',
+            name: 'Zuck',
+            'profilePicture(size:32)': {[REF_KEY]: 'client:1'},
+          },
+          'client:1': {
+            __id: 'client:1',
+            uri: 'https://photo1.jpg',
+          },
+        };
+        source = getRecordSourceImplementation(data);
+        store = new RelayModernStore(source);
+        const owner = createOperationDescriptor(UserQuery, {});
+        const selector = createReaderSelector(
+          UserFragment,
+          '4',
+          {size: 32},
+          owner.request,
+        );
+        const snapshot = store.lookup(selector);
+        expect(snapshot.isMissingData).toEqual(true);
+
+        const callback = jest.fn();
+        // Record does not exist when subscribed
+        store.subscribe(snapshot, callback);
+        const nextSource = getRecordSourceImplementation({
+          '4': {
+            __id: '4',
+            __typename: 'User',
+            emailAddresses: ['a@b.com'],
+          },
+        });
+        store.publish(nextSource);
+        store.notify();
+        expect(callback.mock.calls.length).toBe(1);
+        expect(callback.mock.calls[0][0]).toEqual({
+          ...snapshot,
+          missingRequiredFields: null,
+          isMissingData: false,
+          data: {
+            name: 'Zuck',
+            profilePicture: {
+              uri: 'https://photo1.jpg',
+            },
+            emailAddresses: ['a@b.com'],
+          },
+          seenRecords: {
+            '4': {
+              ...data['4'],
+              emailAddresses: ['a@b.com'],
+            },
+            'client:1': {
+              ...data['client:1'],
+            },
+          },
+        });
+      });
+
+      it('notifies subscribers of changes to unfetched records', () => {
+        const owner = createOperationDescriptor(UserQuery, {});
+        const selector = createReaderSelector(
+          UserFragment,
+          '842472',
+          {
+            size: 32,
+          },
+          owner.request,
+        );
+        const snapshot = store.lookup(selector);
+        const callback = jest.fn();
+        // Record does not exist when subscribed
+        store.subscribe(snapshot, callback);
+        const nextSource = getRecordSourceImplementation({
+          '842472': {
+            __id: '842472',
+            __typename: 'User',
+            name: 'Joe',
+          },
+        });
+        store.publish(nextSource);
+        store.notify();
+        expect(callback.mock.calls.length).toBe(1);
+        expect(callback.mock.calls[0][0]).toEqual({
+          ...snapshot,
+          data: {
+            name: 'Joe',
+            profilePicture: undefined,
+          },
+          missingRequiredFields: null,
+          isMissingData: true,
+          seenRecords: nextSource.toJSON(),
+        });
+      });
+
+      it('notifies subscribers of changes to deleted records', () => {
+        const owner = createOperationDescriptor(UserQuery, {});
+        const selector = createReaderSelector(
+          UserFragment,
+          '842472',
+          {
+            size: 32,
+          },
+          owner.request,
+        );
+        // Initially delete the record
+        source.delete('842472');
+        const snapshot = store.lookup(selector);
+        const callback = jest.fn();
+        // Record does not exist when subscribed
+        store.subscribe(snapshot, callback);
+        // Create it again
+        const nextSource = getRecordSourceImplementation({
+          '842472': {
+            __id: '842472',
+            __typename: 'User',
+            name: 'Joe',
+          },
+        });
+        store.publish(nextSource);
+        store.notify();
+        expect(callback.mock.calls.length).toBe(1);
+        expect(callback.mock.calls[0][0]).toEqual({
+          ...snapshot,
+          data: {
+            name: 'Joe',
+            profilePicture: undefined,
+          },
+          missingRequiredFields: null,
+          isMissingData: true,
+          seenRecords: nextSource.toJSON(),
+        });
+      });
+
+      it('does not call subscribers whose data has not changed', () => {
+        // subscribe(), publish() -> subscriber *not* called
+        const owner = createOperationDescriptor(UserQuery, {});
+        const selector = createReaderSelector(
+          UserFragment,
+          '4',
+          {size: 32},
+          owner.request,
+        );
+        const snapshot = store.lookup(selector);
+        const callback = jest.fn();
+        store.subscribe(snapshot, callback);
+        // Publish a change to profilePicture.uri
+        const nextSource = getRecordSourceImplementation({
+          '842472': {
+            __id: '842472',
+            __typename: 'User',
+            name: 'Joe',
+          },
+        });
+        store.publish(nextSource);
+        store.notify();
+        expect(callback).not.toBeCalled();
+      });
+
+      it('does not notify disposed subscribers', () => {
+        // subscribe(), publish(), dispose(), notify() -> subscriber *not* called
+        const owner = createOperationDescriptor(UserQuery, {});
+        const selector = createReaderSelector(
+          UserFragment,
+          '4',
+          {size: 32},
+          owner.request,
+        );
+        const snapshot = store.lookup(selector);
+        const callback = jest.fn();
+        const {dispose} = store.subscribe(snapshot, callback);
+        // Publish a change to profilePicture.uri
+        const nextSource = getRecordSourceImplementation({
+          'client:1': {
+            __id: 'client:1',
+            uri: 'https://photo2.jpg',
+          },
+        });
+        store.publish(nextSource);
+        dispose();
+        store.notify();
+        expect(callback).not.toBeCalled();
+      });
+
+      it('throws if source records are modified', () => {
+        const zuck = source.get('4');
+        expect(zuck).toBeTruthy();
+        expect(() => {
+          // $FlowFixMe[incompatible-call]
+          RelayModernRecord.setValue(zuck, 'pet', 'Beast');
+        }).toThrow(TypeError);
+      });
+
+      it('throws if published records are modified', () => {
+        // Create and publish a source with a new record
+        const nextSource = getRecordSourceImplementation();
+        const beast = RelayModernRecord.create('beast', 'Pet');
+        nextSource.set('beast', beast);
+        store.publish(nextSource);
+        expect(() => {
+          RelayModernRecord.setValue(beast, 'name', 'Beast');
+        }).toThrow(TypeError);
+      });
+
+      it('throws if updated records are modified', () => {
+        // Create and publish a source with a record of the same id
+        const nextSource = getRecordSourceImplementation();
+        const beast = RelayModernRecord.create('beast', 'Pet');
+        nextSource.set('beast', beast);
+        const zuck = RelayModernRecord.create('4', 'User');
+        RelayModernRecord.setLinkedRecordID(zuck, 'pet', 'beast');
+        nextSource.set('4', zuck);
+        store.publish(nextSource);
+
+        // Cannot modify merged record
+        expect(() => {
+          const mergedRecord = source.get('4');
+          expect(mergedRecord).toBeTruthy();
+          // $FlowFixMe[incompatible-call]
+          RelayModernRecord.setValue(mergedRecord, 'pet', null);
+        }).toThrow(TypeError);
+        // Cannot modify the published record, even though it isn't in the store
+        // This is for consistency because it is non-deterinistic if published
+        // records will be merged into a new object or used as-is.
+        expect(() => {
+          RelayModernRecord.setValue(zuck, 'pet', null);
+        }).toThrow(TypeError);
+      });
+
+      describe('with data invalidation', () => {
+        it('correctly invalidates store when store is globally invalidated', () => {
+          const owner = createOperationDescriptor(UserQuery, {
+            id: '4',
+            size: 32,
+          });
+          const selector = createReaderSelector(
+            UserFragment,
+            '4',
+            {size: 32},
+            owner.request,
+          );
+          const snapshot = store.lookup(selector);
+          const callback = jest.fn();
+          store.subscribe(snapshot, callback);
+          // Publish a change to profilePicture.uri
+          const nextSource = getRecordSourceImplementation({
+            'client:1': {
+              __id: 'client:1',
+              uri: 'https://photo2.jpg',
+            },
+          });
+          store.publish(
+            nextSource,
+            new Set(), // indicate that no individual ids were invalidated
+          );
+          store.notify(
+            owner,
+            true, // indicate that store should be globally invalidated
+          );
+          // Results are asserted in earlier tests
+
+          expect(store.check(owner)).toEqual({status: 'stale'});
+        });
+
+        it('correctly invalidates individual records', () => {
+          const owner = createOperationDescriptor(UserQuery, {
+            id: '4',
+            size: 32,
+          });
+          const selector = createReaderSelector(
+            UserFragment,
+            '4',
+            {size: 32},
+            owner.request,
+          );
+          const snapshot = store.lookup(selector);
+          const callback = jest.fn();
+          store.subscribe(snapshot, callback);
+          // Publish a change to profilePicture.uri
+          const nextSource = getRecordSourceImplementation({
+            'client:1': {
+              __id: 'client:1',
+              uri: 'https://photo2.jpg',
+            },
+          });
+          store.publish(
+            nextSource,
+            new Set(['client:1']), // indicate that this id was invalidated
+          );
+          store.notify(owner, false);
+          // Results are asserted in earlier tests
+
+          const record = store.getSource().get('client:1');
+          if (!record) {
+            throw new Error('Expected to find record with id client:1');
+          }
+          expect(record[INVALIDATED_AT_KEY]).toEqual(1);
+          expect(store.check(owner)).toEqual({status: 'stale'});
+        });
+
+        it("correctly invalidates records even when they weren't modified in the source being published", () => {
+          const owner = createOperationDescriptor(UserQuery, {
+            id: '4',
+            size: 32,
+          });
+          const selector = createReaderSelector(
+            UserFragment,
+            '4',
+            {size: 32},
+            owner.request,
+          );
+          const snapshot = store.lookup(selector);
+          const callback = jest.fn();
+          store.subscribe(snapshot, callback);
+          // Publish a change to profilePicture.uri
+          const nextSource = getRecordSourceImplementation({
+            'client:1': {
+              __id: 'client:1',
+              uri: 'https://photo2.jpg',
+            },
+          });
+          store.publish(
+            nextSource,
+            new Set(['4']), // indicate that this id was invalidated
+          );
+          store.notify(owner, false);
+          // Results are asserted in earlier tests
+
+          const record = store.getSource().get('4');
+          if (!record) {
+            throw new Error('Expected to find record with id "4"');
+          }
+          expect(record[INVALIDATED_AT_KEY]).toEqual(1);
+          expect(store.check(owner)).toEqual({status: 'stale'});
+        });
+      });
+
+      it('emits log events for publish and notify', () => {
+        const owner = createOperationDescriptor(UserQuery, {});
+        const selector = createReaderSelector(
+          UserFragment,
+          '4',
+          {size: 32},
+          owner.request,
+        );
+        const snapshot = store.lookup(selector);
+        const callback = jest.fn(nextSnapshot => {
+          logEvents.push({
+            kind: 'test_only_callback',
+            data: nextSnapshot.data,
+          });
+        });
+        store.subscribe(snapshot, callback);
+
+        const nextSource = getRecordSourceImplementation({
+          'client:1': {
+            __id: 'client:1',
+            uri: 'https://photo2.jpg',
+          },
+        });
+        store.publish(nextSource);
+        expect(logEvents).toEqual([
+          {name: 'store.publish', source: nextSource, optimistic: false},
+        ]);
+        expect(callback).toBeCalledTimes(0);
+        logEvents.length = 0;
+        store.notify();
+        expect(logEvents).toEqual([
+          {
+            name: 'store.notify.start',
+          },
+          // callbacks occur after notify.start...
+          {
+            // not a real LogEvent, this is for testing only
+            kind: 'test_only_callback',
+            data: {
+              emailAddresses: ['a@b.com'],
+              name: 'Zuck',
+              profilePicture: {uri: 'https://photo2.jpg'},
+            },
+          },
+          // ...and before notify.complete
+          {
+            name: 'store.notify.complete',
+            updatedRecordIDs: {'client:1': true},
+            invalidatedRecordIDs: new Set(),
+          },
+        ]);
+        expect(callback).toBeCalledTimes(1);
+      });
+    });
+  });
+});
diff --git a/packages/relay-runtime/util/RelayFeatureFlags.js b/packages/relay-runtime/util/RelayFeatureFlags.js
index ea0ce11a16fdc..4a09f316bb4fc 100644
--- a/packages/relay-runtime/util/RelayFeatureFlags.js
+++ b/packages/relay-runtime/util/RelayFeatureFlags.js
@@ -21,6 +21,7 @@ type FeatureFlags = {|
   ENABLE_REQUIRED_DIRECTIVES: boolean | string,
   ENABLE_GETFRAGMENTIDENTIFIER_OPTIMIZATION: boolean,
   ENABLE_FRIENDLY_QUERY_NAME_GQL_URL: boolean,
+  ENABLE_STORE_SUBSCRIPTIONS_REFACTOR: boolean,
 |};
 
 const RelayFeatureFlags: FeatureFlags = {
@@ -32,6 +33,7 @@ const RelayFeatureFlags: FeatureFlags = {
   ENABLE_REQUIRED_DIRECTIVES: false,
   ENABLE_GETFRAGMENTIDENTIFIER_OPTIMIZATION: false,
   ENABLE_FRIENDLY_QUERY_NAME_GQL_URL: false,
+  ENABLE_STORE_SUBSCRIPTIONS_REFACTOR: false,
 };
 
 module.exports = RelayFeatureFlags;