introduce new Publisher for incremental delivery

yaacovCR · yaacovCR · commit fb36bff996c6 · 2023-01-03T16:55:17.000+02:00
Depends on #3784 The proposed new Publisher: 1. does not use the event loop to manage AsyncRecord dependencies 2. uses separate sets to store pending vs ready AsyncRecords 3. does not use `Promise.race` ## No event loop for managing AsyncRecord dependencies The current Publisher wraps every AsyncRecord result in a promise that only resolves after the parent AsyncRecord resolves. If multiple items within a stream are released for publishing because their parent has just been published, the stream cannot be in its entirety until after all of these promises unwind. The new Publisher keeps track of dependencies manually. When an AsyncRecord is pushed by the publisher, all of its dependent AsyncRecords are synchronously pushed, repeating as necessary, without using the event loop. ## Separate sets for pending vs ready AsyncRecords The current publisher inspects all pending AsyncRecords whenever any of them resolves. All that are completed are added to the response. The new Publisher moves AsyncRecords from the pending set to the ready set as they are pushed, so that on each call to next, only the ready set must be iterated. As a side-effect of this change, the incremental array is ordered by which items are ready for delivery first, and not by the initial document. This seems like a worthwhile tradeoff, and is still adherent to the spec, as far as I can tell. ## No `Promise.race` The old Publisher uses `Promise.race` as the trigger to determine whether payloads are ready. The new Publisher uses a single triggering promise that is triggered whenever an AsyncRecord is pushed, and then reset. This may be beneficial as the implementation of `Promise.race` within V8 has a known memory leak for long-running promises. (see https://bugs.chromium.org/p/v8/issues/detail?id=9858). An alternative would be to utilize @brainkim 's memory-safe version detailed in that issue.
diff --git a/src/execution/__tests__/defer-test.ts b/src/execution/__tests__/defer-test.ts
@@ -605,11 +605,6 @@ describe('Execute: defer directive', () => {
             data: { slowField: 'slow', friends: [{}, {}, {}] },
             path: ['hero'],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           { data: { name: 'Han' }, path: ['hero', 'friends', 0] },
           { data: { name: 'Leia' }, path: ['hero', 'friends', 1] },
           { data: { name: 'C-3PO' }, path: ['hero', 'friends', 2] },
@@ -653,11 +648,6 @@ describe('Execute: defer directive', () => {
             },
             path: ['hero'],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           { data: { name: 'Han' }, path: ['hero', 'friends', 0] },
           { data: { name: 'Leia' }, path: ['hero', 'friends', 1] },
           { data: { name: 'C-3PO' }, path: ['hero', 'friends', 2] },
diff --git a/src/execution/__tests__/stream-test.ts b/src/execution/__tests__/stream-test.ts
@@ -151,11 +151,10 @@ describe('Execute: stream directive', () => {
         hasNext: true,
       },
       {
-        incremental: [{ items: ['banana'], path: ['scalarList', 1] }],
-        hasNext: true,
-      },
-      {
-        incremental: [{ items: ['coconut'], path: ['scalarList', 2] }],
+        incremental: [
+          { items: ['banana'], path: ['scalarList', 1] },
+          { items: ['coconut'], path: ['scalarList', 2] },
+        ],
         hasNext: false,
       },
     ]);
@@ -173,15 +172,11 @@ describe('Execute: stream directive', () => {
         hasNext: true,
       },
       {
-        incremental: [{ items: ['apple'], path: ['scalarList', 0] }],
-        hasNext: true,
-      },
-      {
-        incremental: [{ items: ['banana'], path: ['scalarList', 1] }],
-        hasNext: true,
-      },
-      {
-        incremental: [{ items: ['coconut'], path: ['scalarList', 2] }],
+        incremental: [
+          { items: ['apple'], path: ['scalarList', 0] },
+          { items: ['banana'], path: ['scalarList', 1] },
+          { items: ['coconut'], path: ['scalarList', 2] },
+        ],
         hasNext: false,
       },
     ]);
@@ -230,11 +225,6 @@ describe('Execute: stream directive', () => {
             path: ['scalarList', 1],
             label: 'scalar-stream',
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: ['coconut'],
             path: ['scalarList', 2],
@@ -296,11 +286,6 @@ describe('Execute: stream directive', () => {
             items: [['banana', 'banana', 'banana']],
             path: ['scalarListList', 1],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [['coconut', 'coconut', 'coconut']],
             path: ['scalarListList', 2],
@@ -379,20 +364,10 @@ describe('Execute: stream directive', () => {
             items: [{ name: 'Luke', id: '1' }],
             path: ['friendList', 0],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ name: 'Han', id: '2' }],
             path: ['friendList', 1],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ name: 'Leia', id: '3' }],
             path: ['friendList', 2],
@@ -628,9 +603,6 @@ describe('Execute: stream directive', () => {
             path: ['friendList', 2],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -670,7 +642,7 @@ describe('Execute: stream directive', () => {
         }
       }
     `);
-    const result = await completeAsync(document, 3, {
+    const result = await completeAsync(document, 2, {
       async *friendList() {
         yield await Promise.resolve(friends[0]);
         yield await Promise.resolve(friends[1]);
@@ -699,10 +671,9 @@ describe('Execute: stream directive', () => {
               path: ['friendList', 2],
             },
           ],
-          hasNext: true,
+          hasNext: false,
         },
       },
-      { done: false, value: { hasNext: false } },
       { done: true, value: undefined },
     ]);
   });
@@ -930,11 +901,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ nonNullName: 'Han' }],
             path: ['friendList', 2],
@@ -1107,11 +1073,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ nonNullName: 'Han' }],
             path: ['friendList', 2],
@@ -1276,6 +1237,10 @@ describe('Execute: stream directive', () => {
       },
       {
         incremental: [
+          {
+            items: [{ name: 'Luke' }],
+            path: ['nestedObject', 'nestedFriendList', 0],
+          },
           {
             data: { scalarField: null },
             path: ['otherNestedObject'],
@@ -1287,10 +1252,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-          {
-            items: [{ name: 'Luke' }],
-            path: ['nestedObject', 'nestedFriendList', 0],
-          },
         ],
         hasNext: false,
       },
@@ -1397,9 +1358,6 @@ describe('Execute: stream directive', () => {
             ],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -1546,9 +1504,6 @@ describe('Execute: stream directive', () => {
             path: ['friendList', 2],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -1602,15 +1557,6 @@ describe('Execute: stream directive', () => {
             data: { scalarField: 'slow', nestedFriendList: [] },
             path: ['nestedObject'],
           },
-        ],
-        hasNext: true,
-      },
-      done: false,
-    });
-    const result3 = await iterator.next();
-    expectJSON(result3).toDeepEqual({
-      value: {
-        incremental: [
           {
             items: [{ name: 'Luke' }],
             path: ['nestedObject', 'nestedFriendList', 0],
@@ -1620,8 +1566,8 @@ describe('Execute: stream directive', () => {
       },
       done: false,
     });
-    const result4 = await iterator.next();
-    expectJSON(result4).toDeepEqual({
+    const result3 = await iterator.next();
+    expectJSON(result3).toDeepEqual({
       value: {
         incremental: [
           {
@@ -1633,13 +1579,13 @@ describe('Execute: stream directive', () => {
       },
       done: false,
     });
-    const result5 = await iterator.next();
-    expectJSON(result5).toDeepEqual({
+    const result4 = await iterator.next();
+    expectJSON(result4).toDeepEqual({
       value: { hasNext: false },
       done: false,
     });
-    const result6 = await iterator.next();
-    expectJSON(result6).toDeepEqual({
+    const result5 = await iterator.next();
+    expectJSON(result5).toDeepEqual({
       value: undefined,
       done: true,
     });
diff --git a/src/execution/execute.ts b/src/execution/execute.ts
@@ -2095,12 +2095,14 @@ class DeferredFragmentRecord {
   errors: Array<GraphQLError>;
   label: string | undefined;
   path: Array<string | number>;
-  promise: Promise<void>;
   data: ObjMap<unknown> | null;
   parentContext: AsyncPayloadRecord | undefined;
-  isCompleted: boolean;
-  _exeContext: ExecutionContext;
-  _resolve?: (arg: PromiseOrValue<ObjMap<unknown> | null>) => void;
+  _publisher: Publisher<
+    AsyncPayloadRecord,
+    IncrementalResult,
+    SubsequentIncrementalExecutionResult
+  >;
+
   constructor(opts: {
     label: string | undefined;
     path: Path | undefined;
@@ -2112,27 +2114,14 @@ class DeferredFragmentRecord {
     this.path = pathToArray(opts.path);
     this.parentContext = opts.parentContext;
     this.errors = [];
-    this._exeContext = opts.exeContext;
-    this._exeContext.publisher.add(this);
-    this.isCompleted = false;
+    this._publisher = opts.exeContext.publisher;
+    this._publisher.add(this);
     this.data = null;
-    this.promise = new Promise<ObjMap<unknown> | null>((resolve) => {
-      this._resolve = (promiseOrValue) => {
-        resolve(promiseOrValue);
-      };
-    }).then((data) => {
-      this.data = data;
-      this.isCompleted = true;
-    });
   }
 
   addData(data: ObjMap<unknown> | null) {
-    const parentData = this.parentContext?.promise;
-    if (parentData) {
-      this._resolve?.(parentData.then(() => data));
-      return;
-    }
-    this._resolve?.(data);
+    this.data = data;
+    this._publisher.complete(this);
   }
 }
 
@@ -2142,13 +2131,15 @@ class StreamRecord {
   label: string | undefined;
   path: Array<string | number>;
   items: Array<unknown> | null;
-  promise: Promise<void>;
   parentContext: AsyncPayloadRecord | undefined;
   iterator: AsyncIterator<unknown> | undefined;
   isCompletedIterator?: boolean;
-  isCompleted: boolean;
-  _exeContext: ExecutionContext;
-  _resolve?: (arg: PromiseOrValue<Array<unknown> | null>) => void;
+  _publisher: Publisher<
+    AsyncPayloadRecord,
+    IncrementalResult,
+    SubsequentIncrementalExecutionResult
+  >;
+
   constructor(opts: {
     label: string | undefined;
     path: Path | undefined;
@@ -2163,27 +2154,14 @@ class StreamRecord {
     this.parentContext = opts.parentContext;
     this.iterator = opts.iterator;
     this.errors = [];
-    this._exeContext = opts.exeContext;
-    this._exeContext.publisher.add(this);
-    this.isCompleted = false;
+    this._publisher = opts.exeContext.publisher;
+    this._publisher.add(this);
     this.items = null;
-    this.promise = new Promise<Array<unknown> | null>((resolve) => {
-      this._resolve = (promiseOrValue) => {
-        resolve(promiseOrValue);
-      };
-    }).then((items) => {
-      this.items = items;
-      this.isCompleted = true;
-    });
   }
 
   addItems(items: Array<unknown> | null) {
-    const parentData = this.parentContext?.promise;
-    if (parentData) {
-      this._resolve?.(parentData.then(() => items));
-      return;
-    }
-    this._resolve?.(items);
+    this.items = items;
+    this._publisher.complete(this);
   }
 
   setIsCompletedIterator() {
diff --git a/src/execution/publisher.ts b/src/execution/publisher.ts
