holdGC for incremental responses in QueryExecutor

Reviewed By: josephsavona

Differential Revision: D22714580

fbshipit-source-id: a0dc921d025e1d65260aa1e5416f0f85af1e5eca

Author: tyao1

packages/relay-runtime/store/RelayModernQueryExecutor.js

@@ -57,7 +57,7 @@ import type {
   NormalizationSplitOperation,
   NormalizationSelectableNode,
 } from '../util/NormalizationNode';
-import type {DataID, Variables} from '../util/RelayRuntimeTypes';
+import type {DataID, Variables, Disposable} from '../util/RelayRuntimeTypes';
 import type {GetDataID} from './RelayResponseNormalizer';
 import type {NormalizationOptions} from './RelayResponseNormalizer';
 
@@ -136,6 +136,7 @@ class Executor {
   _store: Store;
   _subscriptions: Map<number, Subscription>;
   _updater: ?SelectorStoreUpdater;
+  _retainDisposable: ?Disposable;
   +_isClientPayload: boolean;
 
   constructor({
@@ -223,6 +224,10 @@ class Executor {
     }
     this._incrementalResults.clear();
     this._completeOperationTracker();
+    if (this._retainDisposable) {
+      this._retainDisposable.dispose();
+      this._retainDisposable = null;
+    }
   }
 
   _updateActiveState(): void {
@@ -436,6 +441,9 @@ class Executor {
       const updatedOwners = this._publishQueue.run(this._operation);
       this._updateOperationTracker(updatedOwners);
       this._processPayloadFollowups(payloadFollowups);
+      if (this._incrementalPayloadsPending && !this._retainDisposable) {
+        this._retainDisposable = this._store.retain(this._operation);
+      }
     }
 
     if (incrementalResponses.length > 0) {

packages/relay-runtime/store/__tests__/RelayModernEnvironment-ExecuteWithStreamedConnection-test.js

@@ -83,16 +83,6 @@ describe('execute() fetches a @stream-ed @connection', () => {
             }
           }
         }
-
-        fragment FeedEdgeFragment on NewsFeedEdge {
-          cursor
-          node {
-            id
-            feedback {
-              id
-            }
-          }
-        }
       `));
     variables = {enableStream: true, after: null};
     operation = createOperationDescriptor(query, variables);
@@ -1268,4 +1258,135 @@ describe('execute() fetches a @stream-ed @connection', () => {
       },
     });
   });
+
+  it('does not garbage collect the server connection when a pagination query is in flight', () => {
+    environment.retain(operation);
+    const initialSnapshot = environment.lookup(selector);
+    callback = jest.fn();
+    environment.subscribe(initialSnapshot, callback);
+
+    // populate the first "page" of results (one item) using the query
+    // with streaming disabled
+    variables = {enableStream: false, after: null};
+    operation = createOperationDescriptor(query, variables);
+    environment.execute({operation}).subscribe({});
+    dataSource.next({
+      data: {
+        viewer: {
+          newsFeed: {
+            edges: [
+              {
+                cursor: 'cursor-1',
+                node: {
+                  __typename: 'Story',
+                  id: '1',
+                  feedback: {
+                    id: 'feedback-1',
+                    actors: [
+                      {
+                        id: 'actor-1',
+                        __typename: 'User',
+                        name: 'Alice',
+                      },
+                    ],
+                  },
+                },
+              },
+            ],
+            pageInfo: {
+              hasNextPage: true,
+              endCursor: 'cursor-1',
+            },
+          },
+        },
+      },
+    });
+    dataSource.complete();
+    jest.runAllTimers();
+    next.mockClear();
+    callback.mockClear();
+
+    // Start pagination query
+    const paginationOperation = createOperationDescriptor(query, {
+      enableStream: true,
+      after: 'cursor-1',
+    });
+    environment.execute({operation: paginationOperation}).subscribe(callbacks);
+    dataSource.next({
+      data: {
+        viewer: {
+          newsFeed: {
+            edges: [],
+          },
+        },
+      },
+    });
+    jest.runAllTimers();
+    next.mockClear();
+    callback.mockClear();
+
+    // Triggers a GC
+    store.__gc();
+    jest.runAllTimers();
+
+    // Second edge should be appended correctly
+    dataSource.next({
+      data: {
+        cursor: 'cursor-2',
+        node: {
+          __typename: 'Story',
+          id: '2',
+          feedback: {
+            id: 'feedback-2',
+            actors: [
+              {
+                id: 'actor-2',
+                __typename: 'User',
+                name: 'Bob',
+              },
+            ],
+          },
+        },
+      },
+      label: 'FeedFragment$stream$RelayModernEnvironment_newsFeed',
+      path: ['viewer', 'newsFeed', 'edges', 0],
+    });
+    expect(error.mock.calls.map(call => call[0].stack)).toEqual([]);
+    expect(next).toBeCalledTimes(1);
+    expect(callback).toBeCalledTimes(1);
+    const snapshot = callback.mock.calls[0][0];
+    expect(snapshot.isMissingData).toBe(false);
+    expect(snapshot.data).toEqual({
+      newsFeed: {
+        edges: [
+          {
+            cursor: 'cursor-1',
+            node: {
+              __typename: 'Story',
+              id: '1',
+              feedback: {
+                id: 'feedback-1',
+                actors: [{id: 'actor-1', name: 'ALICE'}],
+              },
+            },
+          },
+          {
+            cursor: 'cursor-2',
+            node: {
+              __typename: 'Story',
+              id: '2',
+              feedback: {
+                id: 'feedback-2',
+                actors: [{id: 'actor-2', name: 'BOB'}],
+              },
+            },
+          },
+        ],
+        pageInfo: {
+          endCursor: 'cursor-1',
+          hasNextPage: true,
+        },
+      },
+    });
+  });
 });