upgrade executor to non-duplicating incremental delivery format

includes new options to allow for prior branching format

Author: yaacovCR

.changeset/fifty-bobcats-jog.md

@@ -0,0 +1,41 @@
+---
+'@graphql-tools/executor': major
+'@graphql-tools/utils': minor
+---
+
+Upgrade to non-duplicating Incremental Delivery format
+
+## Description
+
+GraphQL Incremental Delivery is moving to a [new response format without duplication](https://github.com/graphql/defer-stream-wg/discussions/69).
+
+This PR updates the executor within graphql-tools to avoid any duplication of fields as per the new format, a BREAKING CHANGE, released in graphql-js `v17.0.0-alpha.3`. The original version of incremental delivery was released in graphql-js `v17.0.0-alpha.2`.
+
+The new format also includes new `pending` and `completed` entries where the `pending` entries assign `ids` to `defer` and `stream` entries, and the `completed` entries are sent as deferred fragments or streams complete. In the new format, the `path` and `label` are only sent along with the `id` within the `pending` entries. Also, incremental errors (i.e. errors that bubble up to a position that has already been sent) are sent within the `errors` field on `completed` entries, rather than as `incremental` entries with `data` or `items` set to `null`. The missing `path` and `label` fields and different mechanism for reporting incremental errors are also a BREAKING CHANGE.
+
+Along with the new format, the GraphQL Working Group has also decided to disable incremental delivery support for subscriptions (1) to gather more information about use cases and (2) explore how to interleaving the incremental response streams generated from different source events into one overall subscription response stream. This is also a BREAKING CHANGE.
+
+Library users can explicitly opt in to the older format by call `execute` with the following option:
+
+```ts
+const result = await execute({
+  ...,
+  incrementalPreset: 'v17.0.0-alpha.2',
+});
+```
+
+The default value for `incrementalPreset` when omitted is `'v17.0.0-alpha.3'`, which enables the new behaviors described above. The new behaviors can also be disabled granularly as follows:
+
+```ts
+const result = await execute({
+  ...,
+  deferWithoutDuplication: false,
+  useIncrementalNotifications: false,
+  errorOnSubscriptionWithIncrementalDelivery: false,
+});
+```
+
+Setting `deferWithoutDuplication` to `false` will re-enable deduplication according to the older format.
+Setting `useIncrementalNotifications` to `false` will (1) omit the `pending` entries, (2) send `path` and `label` on every `incremental` entry, (3) omit `completed` entries, and (4) send incremental errors within `incremental` entries along with a `data` or `items` field set to `null`.
+Setting `errorOnSubscriptionWithIncrementalDelivery` to `false` will re-enable the use of incremental delivery with subscriptions.
+```

packages/delegate/src/defaultMergedResolver.ts

@@ -5,8 +5,8 @@ import {
   responsePathAsArray,
   SelectionSetNode,
 } from 'graphql';
-import { getResponseKeyFromInfo, isPromise } from '@graphql-tools/utils';
-import { createDeferred, DelegationPlanLeftOver, getPlanLeftOverFromParent } from './leftOver.js';
+import { createDeferred, getResponseKeyFromInfo, isPromise } from '@graphql-tools/utils';
+import { DelegationPlanLeftOver, getPlanLeftOverFromParent } from './leftOver.js';
 import {
   getSubschema,
   getUnpathedErrors,

packages/delegate/src/leftOver.ts

@@ -1,23 +1,8 @@
 import { FieldNode } from 'graphql';
+import { Deferred } from '@graphql-tools/utils';
 import { Subschema } from './Subschema.js';
 import { DelegationPlanBuilder, ExternalObject } from './types.js';
 
-export type Deferred<T = unknown> = PromiseWithResolvers<T>;
-
-// TODO: Remove this after Node 22
-export function createDeferred<T>(): Deferred<T> {
-  if (Promise.withResolvers) {
-    return Promise.withResolvers();
-  }
-  let resolve: (value: T | PromiseLike<T>) => void;
-  let reject: (error: unknown) => void;
-  const promise = new Promise<T>((_resolve, _reject) => {
-    resolve = _resolve;
-    reject = _reject;
-  });
-  return { promise, resolve: resolve!, reject: reject! };
-}
-
 export interface DelegationPlanLeftOver {
   unproxiableFieldNodes: Array<FieldNode>;
   nonProxiableSubschemas: Array<Subschema>;

packages/executor/package.json

@@ -62,6 +62,7 @@
     "value-or-promise": "^1.0.12"
   },
   "devDependencies": {
+    "@types/dlv": "^1.1.4",
     "cross-inspect": "1.0.1",
     "graphql": "^16.6.0"
   },

packages/executor/src/execution/AccumulatorMap.ts

@@ -0,0 +1,17 @@
+/**
+ * ES6 Map with additional `add` method to accumulate items.
+ */
+export class AccumulatorMap<K, T> extends Map<K, Array<T>> {
+  get [Symbol.toStringTag]() {
+    return 'AccumulatorMap';
+  }
+
+  add(key: K, item: T): void {
+    const group = this.get(key);
+    if (group === undefined) {
+      this.set(key, [item]);
+    } else {
+      group.push(item);
+    }
+  }
+}

packages/executor/src/execution/BoxedPromiseOrValue.ts

@@ -0,0 +1,25 @@
+import { isPromise } from '@graphql-tools/utils';
+import type { MaybePromise } from '@graphql-tools/utils';
+
+/**
+ * A BoxedPromiseOrValue is a container for a value or promise where the value
+ * will be updated when the promise resolves.
+ *
+ * A BoxedPromiseOrValue may only be used with promises whose possible
+ * rejection has already been handled, otherwise this will lead to unhandled
+ * promise rejections.
+ *
+ * @internal
+ * */
+export class BoxedPromiseOrValue<T> {
+  value: MaybePromise<T>;
+
+  constructor(value: MaybePromise<T>) {
+    this.value = value;
+    if (isPromise(value)) {
+      value.then(resolved => {
+        this.value = resolved;
+      });
+    }
+  }
+}

packages/executor/src/execution/DeferredFragments.ts

@@ -0,0 +1,106 @@
+import { Path } from '@graphql-tools/utils';
+import { DeferUsage } from './collectFields.js';
+import { PendingExecutionGroup, StreamRecord, SuccessfulExecutionGroup } from './types.js';
+
+export type DeliveryGroup = DeferredFragmentRecord | StreamRecord;
+
+/** @internal */
+export class DeferredFragmentRecord {
+  path: Path | undefined;
+  label: string | undefined;
+  id?: string | undefined;
+  parentDeferUsage: DeferUsage | undefined;
+  pendingExecutionGroups: Set<PendingExecutionGroup>;
+  successfulExecutionGroups: Set<SuccessfulExecutionGroup>;
+  children: Set<DeliveryGroup>;
+  pending: boolean;
+  fns: Array<() => void>;
+
+  constructor(
+    path: Path | undefined,
+    label: string | undefined,
+    parentDeferUsage: DeferUsage | undefined,
+  ) {
+    this.path = path;
+    this.label = label;
+    this.parentDeferUsage = parentDeferUsage;
+    this.pendingExecutionGroups = new Set();
+    this.successfulExecutionGroups = new Set();
+    this.children = new Set();
+    this.pending = false;
+    this.fns = [];
+  }
+
+  onPending(fn: () => void): void {
+    this.fns.push(fn);
+  }
+
+  setAsPending(): void {
+    this.pending = true;
+    for (const fn of this.fns) {
+      fn();
+    }
+  }
+}
+
+export function isDeferredFragmentRecord(
+  deliveryGroup: DeliveryGroup,
+): deliveryGroup is DeferredFragmentRecord {
+  return deliveryGroup instanceof DeferredFragmentRecord;
+}
+
+/**
+ * @internal
+ */
+export class DeferredFragmentFactory {
+  private _rootDeferredFragments = new Map<DeferUsage, DeferredFragmentRecord>();
+
+  get(deferUsage: DeferUsage, path: Path | undefined): DeferredFragmentRecord {
+    const deferUsagePath = this._pathAtDepth(path, deferUsage.depth);
+    let deferredFragmentRecords: Map<DeferUsage, DeferredFragmentRecord> | undefined;
+    if (deferUsagePath === undefined) {
+      deferredFragmentRecords = this._rootDeferredFragments;
+    } else {
+      // A doubly nested Map<Path, Map<DeferUsage, DeferredFragmentRecord>>
+      // could be used, but could leak memory in long running operations.
+      // A WeakMap could be used instead. The below implementation is
+      // WeakMap-Like, saving the Map on the Path object directly.
+      // Alternatively, memory could be reclaimed manually, taking care to
+      // also reclaim memory for nested DeferredFragmentRecords if the parent
+      // is removed secondary to an error.
+      deferredFragmentRecords = (
+        deferUsagePath as unknown as {
+          deferredFragmentRecords: Map<DeferUsage, DeferredFragmentRecord>;
+        }
+      ).deferredFragmentRecords;
+      if (deferredFragmentRecords === undefined) {
+        deferredFragmentRecords = new Map();
+        (
+          deferUsagePath as unknown as {
+            deferredFragmentRecords: Map<DeferUsage, DeferredFragmentRecord>;
+          }
+        ).deferredFragmentRecords = deferredFragmentRecords;
+      }
+    }
+    let deferredFragmentRecord = deferredFragmentRecords.get(deferUsage);
+    if (deferredFragmentRecord === undefined) {
+      const { label, parentDeferUsage } = deferUsage;
+      deferredFragmentRecord = new DeferredFragmentRecord(deferUsagePath, label, parentDeferUsage);
+      deferredFragmentRecords.set(deferUsage, deferredFragmentRecord);
+    }
+    return deferredFragmentRecord;
+  }
+
+  private _pathAtDepth(path: Path | undefined, depth: number): Path | undefined {
+    if (depth === 0) {
+      return;
+    }
+    const stack: Array<Path> = [];
+    let currentPath = path;
+    while (currentPath !== undefined) {
+      stack.unshift(currentPath);
+      currentPath = currentPath.prev;
+    }
+    return stack[depth - 1];
+  }
+}