From b370b08058312e96432ae37205e9ecc54eaf0b57 Mon Sep 17 00:00:00 2001 From: Pavel Glac Date: Mon, 22 Jun 2026 09:20:20 +0000 Subject: [PATCH 01/18] early execution and merge --- .../src/__tests__/executeWithSchema.test.ts | 730 ++++++++++++++++++ .../supermassive/src/executeWithSchema.ts | 4 + .../supermassive/src/executeWithoutSchema.ts | 146 +++- .../src/jsutils/getObjectAtPath.ts | 22 + packages/supermassive/src/types.ts | 2 + 5 files changed, 890 insertions(+), 14 deletions(-) create mode 100644 packages/supermassive/src/__tests__/executeWithSchema.test.ts create mode 100644 packages/supermassive/src/jsutils/getObjectAtPath.ts diff --git a/packages/supermassive/src/__tests__/executeWithSchema.test.ts b/packages/supermassive/src/__tests__/executeWithSchema.test.ts new file mode 100644 index 000000000..7a2c3b778 --- /dev/null +++ b/packages/supermassive/src/__tests__/executeWithSchema.test.ts @@ -0,0 +1,730 @@ +import { parse } from "graphql"; +import { executeWithSchema } from "../executeWithSchema"; + +describe("executeWithSchema - @defer behavior", () => { + function createDeferred() { + let resolve: (value: T) => void; + const promise = new Promise((innerResolve) => { + resolve = innerResolve; + }); + return { promise, resolve: resolve! }; + } + + const definitions = parse(` + type Query { obj: Obj } + type Obj { + critical: String + deferred: String + } + `); + + const document = parse(` + { + obj { + critical + ... on Obj @defer { + deferred + } + } + } + `); + + const nestedDefinitions = parse(` + type Query { obj: Obj } + type Obj { + critical: String + nested: Nested + } + type Nested { + deferred: String + } + `); + + const nestedDocument = parse(` + { + obj { + critical + ... on Obj @defer { + nested { + deferred + } + } + } + } + `); + + const mutationDefinitions = parse(` + type Query { noop: String } + type Mutation { + critical: String + deferred: String + } + `); + + const mutationDocument = parse(` + mutation { + critical + ... on Mutation @defer { + deferred + } + } + `); + + type Resolver = () => string | Promise; + + function executeTestQuery(Obj: { critical: Resolver; deferred: Resolver }) { + return Promise.resolve( + executeWithSchema({ + document, + definitions, + enableEarlyExecution: true, + enableDeferredMerge: true, + resolvers: { + Query: { + obj: () => ({}), + }, + Obj, + }, + }), + ); + } + + function executeNestedTestQuery(Obj: { + critical: Resolver; + deferred: Resolver; + }) { + return Promise.resolve( + executeWithSchema({ + document: nestedDocument, + definitions: nestedDefinitions, + enableEarlyExecution: true, + enableDeferredMerge: true, + resolvers: { + Query: { + obj: () => ({}), + }, + Obj: { + critical: Obj.critical, + nested: () => ({}), + }, + Nested: { + deferred: Obj.deferred, + }, + }, + }), + ); + } + + function executeTestMutation(Mutation: { + critical: Resolver; + deferred: Resolver; + }) { + return Promise.resolve( + executeWithSchema({ + document: mutationDocument, + definitions: mutationDefinitions, + enableEarlyExecution: true, + enableDeferredMerge: true, + resolvers: { + Mutation, + }, + }), + ); + } + + test("returns the initial response as soon as critical fields are ready", async () => { + const critical = createDeferred(); + const deferred = createDeferred(); + + const resultPromise = executeTestQuery({ + critical: () => critical.promise, + deferred: () => deferred.promise, + }); + + critical.resolve("critical"); + const result = await Promise.race([ + resultPromise, + new Promise<"blocked">((resolve) => setTimeout(resolve, 0, "blocked")), + ]); + + if (result === "blocked") { + throw new Error("Initial response waited for deferred field"); + } + + expect(result).toMatchObject({ + initialResult: { + data: { + obj: { + critical: "critical", + }, + }, + hasNext: true, + }, + }); + + if (!("initialResult" in result)) { + throw new Error("Expected an incremental result"); + } + + const subsequentResultPromise = result.subsequentResults.next(); + deferred.resolve("deferred"); + + await expect(subsequentResultPromise).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + deferred: "deferred", + }, + path: ["obj"], + }, + ], + hasNext: false, + }, + done: false, + }); + }); + + test("includes deferred fields in the initial response when they complete before the critical fields", async () => { + const critical = createDeferred(); + + const resultPromise = executeTestQuery({ + deferred: () => Promise.resolve("deferred"), + critical: () => critical.promise, + }); + + setTimeout(() => critical.resolve("critical"), 0); + + await expect(resultPromise).resolves.toEqual({ + data: { + obj: { + critical: "critical", + deferred: "deferred", + }, + }, + }); + }); + + test("surfaces deferred errors at the top level when piggybacked onto the initial response", async () => { + const critical = createDeferred(); + + const resultPromise = executeTestQuery({ + deferred: () => { + throw new Error("Deferred boom"); + }, + critical: () => critical.promise, + }); + + setTimeout(() => critical.resolve("critical"), 0); + + const result = await resultPromise; + + expect(result).toMatchObject({ + data: { + obj: { + critical: "critical", + deferred: null, + }, + }, + }); + expect("initialResult" in (result as object)).toBe(false); + expect( + (result as { errors?: ReadonlyArray<{ message: string }> }).errors, + ).toHaveLength(1); + expect( + (result as { errors: ReadonlyArray<{ message: string }> }).errors[0] + .message, + ).toBe("Deferred boom"); + }); + + test("folds a synchronous deferred field into the initial response instead of streaming it", async () => { + const result = await executeTestQuery({ + critical: () => "critical", + deferred: () => "deferred", + }); + + // Both fields resolve synchronously, so the deferred fragment is already + // complete by the time the response is built. The optimization folds it + // into the initial data rather than emitting a redundant incremental + // payload, producing a single (non-incremental) result. + expect(result).toEqual({ + data: { + obj: { + critical: "critical", + deferred: "deferred", + }, + }, + }); + expect("initialResult" in (result as object)).toBe(false); + }); + + test("delivers a synchronously-completed deferred field incrementally when merge is disabled", async () => { + // Early execution starts the deferred fragment eagerly, but with merge + // disabled it is never folded into the initial response (mirroring + // graphql-js's `enableEarlyExecution`): it is always delivered as a + // separate incremental payload. + const result = await executeWithSchema({ + document, + definitions, + enableEarlyExecution: true, + enableDeferredMerge: false, + resolvers: { + Query: { + obj: () => ({}), + }, + Obj: { + critical: () => "critical", + deferred: () => "deferred", + }, + }, + }); + + expect(result).toMatchObject({ + initialResult: { + data: { + obj: { + critical: "critical", + }, + }, + hasNext: true, + }, + }); + + if (!("initialResult" in result)) { + throw new Error("Expected an incremental result"); + } + + expect(result.initialResult.data).toEqual({ + obj: { critical: "critical" }, + }); + + await expect(result.subsequentResults.next()).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + deferred: "deferred", + }, + path: ["obj"], + }, + ], + hasNext: false, + }, + done: false, + }); + }); + + test("keeps legacy defer behavior when early execution is disabled", async () => { + const result = await executeWithSchema({ + document, + definitions, + enableEarlyExecution: false, + enableDeferredMerge: false, + resolvers: { + Query: { + obj: () => ({}), + }, + Obj: { + critical: () => "critical", + deferred: () => "deferred", + }, + }, + }); + + expect(result).toMatchObject({ + initialResult: { + data: { + obj: { + critical: "critical", + }, + }, + hasNext: true, + }, + }); + + if (!("initialResult" in result)) { + throw new Error("Expected an incremental result"); + } + + await expect(result.subsequentResults.next()).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + deferred: "deferred", + }, + path: ["obj"], + }, + ], + hasNext: false, + }, + done: false, + }); + }); + + test("includes nested deferred field in the initial response when it settles before critical", async () => { + const critical = createDeferred(); + + const resultPromise = executeNestedTestQuery({ + deferred: () => Promise.resolve("deferred"), + critical: () => critical.promise, + }); + + setTimeout(() => critical.resolve("critical"), 0); + + await expect(resultPromise).resolves.toEqual({ + data: { + obj: { + critical: "critical", + nested: { + deferred: "deferred", + }, + }, + }, + }); + }); + + test("streams nested deferred field when critical fields are ready first", async () => { + const deferred = createDeferred(); + + const result = await executeNestedTestQuery({ + critical: () => "critical", + deferred: () => deferred.promise, + }); + + expect(result).toMatchObject({ + initialResult: { + data: { + obj: { + critical: "critical", + }, + }, + hasNext: true, + }, + }); + + if (!("initialResult" in result)) { + throw new Error("Expected an incremental result"); + } + + const subsequentResultPromise = result.subsequentResults.next(); + deferred.resolve("deferred"); + + await expect(subsequentResultPromise).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + nested: { + deferred: "deferred", + }, + }, + path: ["obj"], + }, + ], + hasNext: false, + }, + done: false, + }); + }); + + test("includes deferred mutation fields in the initial response when they settle before critical", async () => { + const critical = createDeferred(); + + const resultPromise = executeTestMutation({ + critical: () => critical.promise, + deferred: () => Promise.resolve("deferred"), + }); + + setTimeout(() => critical.resolve("critical"), 0); + + await expect(resultPromise).resolves.toEqual({ + data: { + critical: "critical", + deferred: "deferred", + }, + }); + }); + + const twoDeferDefinitions = parse(` + type Query { obj: Obj } + type Obj { + critical: String + deferredEarly: String + deferredLate: String + } + `); + + const twoDeferDocument = parse(` + { + obj { + critical + ... on Obj @defer(label: "early") { + deferredEarly + } + ... on Obj @defer(label: "late") { + deferredLate + } + } + } + `); + + function executeTwoDeferQuery(Obj: { + critical: Resolver; + deferredEarly: Resolver; + deferredLate: Resolver; + }) { + return Promise.resolve( + executeWithSchema({ + document: twoDeferDocument, + definitions: twoDeferDefinitions, + enableEarlyExecution: true, + enableDeferredMerge: true, + resolvers: { + Query: { + obj: () => ({}), + }, + Obj, + }, + }), + ); + } + + test("folds the deferred fragment that settled before critical and streams the one that settled after", async () => { + const critical = createDeferred(); + const late = createDeferred(); + + const resultPromise = executeTwoDeferQuery({ + // Resolves synchronously, so this fragment completes before critical. + deferredEarly: () => "early", + // Stays pending past the initial response, so this fragment streams. + deferredLate: () => late.promise, + critical: () => critical.promise, + }); + + setTimeout(() => critical.resolve("critical"), 0); + + const result = await resultPromise; + + // The early fragment is folded into the initial response; the late one is + // still pending, so the overall result is incremental. + expect(result).toMatchObject({ + initialResult: { + data: { + obj: { + critical: "critical", + deferredEarly: "early", + }, + }, + hasNext: true, + }, + }); + + if (!("initialResult" in result)) { + throw new Error("Expected an incremental result"); + } + + expect(result.initialResult.data).toEqual({ + obj: { critical: "critical", deferredEarly: "early" }, + }); + + const subsequentResultPromise = result.subsequentResults.next(); + late.resolve("late"); + + await expect(subsequentResultPromise).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + deferredLate: "late", + }, + path: ["obj"], + }, + ], + hasNext: false, + }, + done: false, + }); + }); + + const ifDocument = parse(` + query ($shouldDefer: Boolean!) { + obj { + critical + ... on Obj @defer(if: $shouldDefer) { + deferred + } + } + } + `); + + function executeIfTestQuery( + Obj: { critical: Resolver; deferred: Resolver }, + shouldDefer: boolean, + ) { + return Promise.resolve( + executeWithSchema({ + document: ifDocument, + definitions, + enableEarlyExecution: true, + enableDeferredMerge: true, + variableValues: { shouldDefer }, + resolvers: { + Query: { + obj: () => ({}), + }, + Obj, + }, + }), + ); + } + + test("streams the deferred field when @defer(if: true) and critical is ready first", async () => { + const deferred = createDeferred(); + + const result = await executeIfTestQuery( + { + critical: () => "critical", + deferred: () => deferred.promise, + }, + true, + ); + + expect(result).toMatchObject({ + initialResult: { + data: { + obj: { + critical: "critical", + }, + }, + hasNext: true, + }, + }); + + if (!("initialResult" in result)) { + throw new Error("Expected an incremental result"); + } + + const subsequentResultPromise = result.subsequentResults.next(); + deferred.resolve("deferred"); + + await expect(subsequentResultPromise).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + deferred: "deferred", + }, + path: ["obj"], + }, + ], + hasNext: false, + }, + done: false, + }); + }); + + test("does not defer the field when @defer(if: false), even if it is slow", async () => { + const deferred = createDeferred(); + + const resultPromise = executeIfTestQuery( + { + critical: () => "critical", + deferred: () => deferred.promise, + }, + false, + ); + + // With `if: false` the fragment is not deferred, so the field is critical + // and the response must wait for it instead of streaming it. + const raced = await Promise.race([ + resultPromise, + new Promise<"pending">((resolve) => setTimeout(resolve, 0, "pending")), + ]); + + expect(raced).toBe("pending"); + + deferred.resolve("deferred"); + + await expect(resultPromise).resolves.toEqual({ + data: { + obj: { + critical: "critical", + deferred: "deferred", + }, + }, + }); + }); + + test("returns a single non-incremental result when @defer(if: false)", async () => { + const result = await executeIfTestQuery( + { + critical: () => "critical", + deferred: () => "deferred", + }, + false, + ); + + expect(result).toEqual({ + data: { + obj: { + critical: "critical", + deferred: "deferred", + }, + }, + }); + expect("initialResult" in (result as object)).toBe(false); + }); + + // Schema where the critical field is non-null: when it errors, the null + // bubbles up and destroys the whole `obj` branch the deferred fragment is + // anchored to. + const nonNullCriticalDefinitions = parse(` + type Query { obj: Obj } + type Obj { + critical: String! + deferred: String + } + `); + + function executeNonNullCriticalQuery(Obj: { + critical: Resolver; + deferred: Resolver; + }) { + return Promise.resolve( + executeWithSchema({ + document, + definitions: nonNullCriticalDefinitions, + resolvers: { + Query: { + obj: () => ({}), + }, + Obj, + }, + }), + ); + } + + test("does not fold a synchronously-completed deferred fragment when a non-null sibling error nulls its parent", async () => { + const result = await executeNonNullCriticalQuery({ + critical: () => { + throw new Error("Critical boom"); + }, + deferred: () => "deferred", + }); + + // The non-null `critical` error bubbles up and nulls `obj`. The deferred + // fragment completed synchronously, but its anchor object is gone, so it + // must not be folded into the (now null) branch. + expect(result).toMatchObject({ + data: { + obj: null, + }, + }); + expect("initialResult" in (result as object)).toBe(false); + const errors = (result as { errors?: ReadonlyArray<{ message: string }> }) + .errors; + expect(errors).toHaveLength(1); + expect(errors?.[0].message).toBe("Critical boom"); + }); +}); diff --git a/packages/supermassive/src/executeWithSchema.ts b/packages/supermassive/src/executeWithSchema.ts index 2b5d242a8..eed97b305 100644 --- a/packages/supermassive/src/executeWithSchema.ts +++ b/packages/supermassive/src/executeWithSchema.ts @@ -16,6 +16,8 @@ export function executeWithSchema({ typeResolver, fieldExecutionHooks, enablePerEventContext, + enableEarlyExecution, + enableDeferredMerge, }: ExecutionWithSchemaArgs): PromiseOrValue { const extracted = extractMinimalViableSchemaForRequestDocument( buildASTSchema(definitions), @@ -36,5 +38,7 @@ export function executeWithSchema({ typeResolver, fieldExecutionHooks, enablePerEventContext, + enableEarlyExecution, + enableDeferredMerge, }); } diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index 340c4523c..4d613348a 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -13,8 +13,10 @@ import { collectSubfields as _collectSubfields, FieldGroup, GroupedFieldSet, + PatchFields, } from "./collectFields"; import { devAssert } from "./jsutils/devAssert"; +import { getObjectAtPath } from "./jsutils/getObjectAtPath"; import { inspect } from "./jsutils/inspect"; import { invariant } from "./jsutils/invariant"; import { isIterableObject } from "./jsutils/isIterableObject"; @@ -122,6 +124,8 @@ export interface ExecutionContext { fieldExecutionHooks?: ExecutionHooks; subsequentPayloads: Set; enablePerEventContext: boolean; + enableEarlyExecution: boolean; + enableDeferredMerge: boolean; } /** @@ -193,6 +197,8 @@ function buildExecutionContext( subscribeFieldResolver, fieldExecutionHooks, enablePerEventContext, + enableEarlyExecution, + enableDeferredMerge, } = args; assertValidExecutionArguments(document, variableValues); @@ -262,6 +268,8 @@ function buildExecutionContext( fieldExecutionHooks, subsequentPayloads: new Set(), enablePerEventContext: enablePerEventContext ?? true, + enableEarlyExecution: enableEarlyExecution ?? false, + enableDeferredMerge: enableDeferredMerge ?? false, }; } @@ -328,6 +336,7 @@ function executeOperation( path, groupedFieldSet, undefined, + exeContext.enableEarlyExecution ? patches : undefined, ); result = buildResponse(exeContext, result); break; @@ -338,6 +347,7 @@ function executeOperation( rootValue, path, groupedFieldSet, + exeContext.enableEarlyExecution ? patches : undefined, ); result = buildResponse(exeContext, result); break; @@ -359,16 +369,18 @@ function executeOperation( ); } - for (const patch of patches) { - const { label, groupedFieldSet: patchGroupedFieldSet } = patch; - executeDeferredFragment( - exeContext, - rootTypeName, - rootValue, - patchGroupedFieldSet, - label, - path, - ); + if (!exeContext.enableEarlyExecution) { + for (const patch of patches) { + const { label, groupedFieldSet: patchGroupedFieldSet } = patch; + executeDeferredFragment( + exeContext, + rootTypeName, + rootValue, + patchGroupedFieldSet, + label, + path, + ); + } } return result; @@ -398,6 +410,10 @@ function buildResponse( const hooks = exeContext.fieldExecutionHooks; try { + if (exeContext.enableDeferredMerge) { + includeCompletedDeferredFragmentsInResult(exeContext, data); + } + const initialResult = exeContext.errors.length === 0 ? { data } @@ -442,7 +458,18 @@ function executeFieldsSerially( sourceValue: unknown, path: Path | undefined, groupedFieldSet: GroupedFieldSet, + patches?: Array, ): PromiseOrValue> { + if (exeContext.enableEarlyExecution) { + startExecutingPatches( + exeContext, + parentTypeName, + sourceValue, + path, + patches, + undefined, + ); + } return promiseReduce( groupedFieldSet, (results, [responseName, fieldGroup]) => { @@ -486,6 +513,7 @@ function executeFields( path: Path | undefined, groupedFieldSet: GroupedFieldSet, incrementalDataRecord: IncrementalDataRecord | undefined, + patches?: Array, ): PromiseOrValue> { const results = Object.create(null); let containsPromise = false; @@ -519,6 +547,15 @@ function executeFields( throw error; } + startExecutingPatches( + exeContext, + parentTypeName, + sourceValue, + path, + patches, + incrementalDataRecord, + ); + // If there are no promises, we can just return the object if (!containsPromise) { return results; @@ -530,6 +567,64 @@ function executeFields( return promiseForObject(results); } +function startExecutingPatches( + exeContext: ExecutionContext, + parentTypeName: string, + sourceValue: unknown, + path: Path | undefined, + patches: Array | undefined, + incrementalDataRecord: IncrementalDataRecord | undefined, +): void { + if (!patches) { + return; + } + + for (const { label, groupedFieldSet: patchGroupedFieldSet } of patches) { + executeDeferredFragment( + exeContext, + parentTypeName, + sourceValue, + patchGroupedFieldSet, + label, + path, + incrementalDataRecord, + ); + } +} + +function includeCompletedDeferredFragmentsInResult( + exeContext: ExecutionContext, + result: ObjMap | null, +): void { + if (result == null) { + return; + } + + for (const incrementalDataRecord of exeContext.subsequentPayloads) { + if ( + !isDeferredFragmentRecord(incrementalDataRecord) || + !incrementalDataRecord.isCompleted + ) { + continue; + } + + const target = getObjectAtPath(result, incrementalDataRecord.path); + if (!target) { + continue; + } + + if (incrementalDataRecord.data != null) { + Object.assign(target, incrementalDataRecord.data); + } + + if (incrementalDataRecord.errors.length > 0) { + exeContext.errors.push(...incrementalDataRecord.errors); + } + + exeContext.subsequentPayloads.delete(incrementalDataRecord); + } +} + /** * Implements the "Executing field" section of the spec * In particular, this function figures out the value that the field returns by @@ -2003,6 +2098,18 @@ function collectAndExecuteSubfields( const { groupedFieldSet: subGroupedFieldSet, patches: subPatches } = collectSubfields(exeContext, { name: returnTypeName }, fieldGroup); + if (exeContext.enableEarlyExecution) { + return executeFields( + exeContext, + returnTypeName, + result, + path, + subGroupedFieldSet, + incrementalDataRecord, + subPatches, + ); + } + const subFields = executeFields( exeContext, returnTypeName, @@ -2875,6 +2982,12 @@ function isStreamItemsRecord( return incrementalDataRecord.type === "stream"; } +function isDeferredFragmentRecord( + incrementalDataRecord: IncrementalDataRecord, +): incrementalDataRecord is DeferredFragmentRecord { + return incrementalDataRecord.type === "defer"; +} + class DeferredFragmentRecord { type: "defer"; errors: Array; @@ -2906,10 +3019,12 @@ class DeferredFragmentRecord { this._resolve = (promiseOrValue) => { resolve(promiseOrValue); }; - }).then((data) => { - this.data = data; - this.isCompleted = true; - }); + }).then((data) => this.complete(data)); + } + + private complete(data: ObjMap | null): void { + this.data = data; + this.isCompleted = true; } addData(data: PromiseOrValue | null>) { @@ -2918,6 +3033,9 @@ class DeferredFragmentRecord { this._resolve?.(parentData.then(() => data)); return; } + if (!isPromise(data)) { + this.complete(data); + } this._resolve?.(data); } } diff --git a/packages/supermassive/src/jsutils/getObjectAtPath.ts b/packages/supermassive/src/jsutils/getObjectAtPath.ts new file mode 100644 index 000000000..a1b1d4501 --- /dev/null +++ b/packages/supermassive/src/jsutils/getObjectAtPath.ts @@ -0,0 +1,22 @@ +import type { ObjMap } from "./ObjMap"; + +/** + * Walks `result` following `path` and returns the nested value, but only when + * that value is a non-null object (a valid `Object.assign` target). Returns + * `undefined` if any segment is missing or the resolved value is not an object. + */ +export function getObjectAtPath( + result: ObjMap, + path: ReadonlyArray, +): ObjMap | undefined { + let target = result; + for (const key of path) { + const next = target[key]; + if (next === null || typeof next !== "object") { + return undefined; + } + target = next as ObjMap; + } + + return target; +} diff --git a/packages/supermassive/src/types.ts b/packages/supermassive/src/types.ts index 457134685..935181441 100644 --- a/packages/supermassive/src/types.ts +++ b/packages/supermassive/src/types.ts @@ -257,6 +257,8 @@ export interface CommonExecutionArgs { subscribeFieldResolver?: Maybe>; fieldExecutionHooks?: ExecutionHooks; enablePerEventContext?: boolean; + enableEarlyExecution?: boolean; + enableDeferredMerge?: boolean; } export type ExecutionWithoutSchemaArgs = CommonExecutionArgs & { schemaFragment: SchemaFragment; From 281e265576b56a75b77823810e341432143b4134 Mon Sep 17 00:00:00 2001 From: Pavel Glac Date: Mon, 22 Jun 2026 09:22:13 +0000 Subject: [PATCH 02/18] fixies --- .../supermassive/src/executeWithoutSchema.ts | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index 4d613348a..51aacdbb7 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -518,6 +518,17 @@ function executeFields( const results = Object.create(null); let containsPromise = false; + if (exeContext.enableEarlyExecution) { + startExecutingPatches( + exeContext, + parentTypeName, + sourceValue, + path, + patches, + incrementalDataRecord, + ); + } + try { for (const [responseName, fieldGroup] of groupedFieldSet) { const fieldPath = addPath(path, responseName, parentTypeName); @@ -547,15 +558,6 @@ function executeFields( throw error; } - startExecutingPatches( - exeContext, - parentTypeName, - sourceValue, - path, - patches, - incrementalDataRecord, - ); - // If there are no promises, we can just return the object if (!containsPromise) { return results; From cc14f5be7477ce871cf8967db513a850c0725843 Mon Sep 17 00:00:00 2001 From: Pavel Glac Date: Mon, 22 Jun 2026 09:22:31 +0000 Subject: [PATCH 03/18] Change files --- ...-supermassive-b119265a-cda6-437b-ad83-e893facef588.json | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 change/@graphitation-supermassive-b119265a-cda6-437b-ad83-e893facef588.json diff --git a/change/@graphitation-supermassive-b119265a-cda6-437b-ad83-e893facef588.json b/change/@graphitation-supermassive-b119265a-cda6-437b-ad83-e893facef588.json new file mode 100644 index 000000000..7aa033ac9 --- /dev/null +++ b/change/@graphitation-supermassive-b119265a-cda6-437b-ad83-e893facef588.json @@ -0,0 +1,7 @@ +{ + "type": "minor", + "comment": "early execution and merge", + "packageName": "@graphitation/supermassive", + "email": "pavelglac@gmail.com", + "dependentChangeType": "patch" +} From cf7dfc9a5d470c1e72efde0988f3d353682182aa Mon Sep 17 00:00:00 2001 From: Pavel Glac Date: Mon, 22 Jun 2026 11:51:41 +0000 Subject: [PATCH 04/18] remove the fetch --- .azure-devops/graphitation-release.yml | 1 - 1 file changed, 1 deletion(-) diff --git a/.azure-devops/graphitation-release.yml b/.azure-devops/graphitation-release.yml index 8d9676411..367a146c9 100644 --- a/.azure-devops/graphitation-release.yml +++ b/.azure-devops/graphitation-release.yml @@ -55,7 +55,6 @@ extends: - script: | git config user.email "gql-svc@microsoft.com" git config user.name "Graphitation Service Account" - git fetch --depth=2 displayName: Configure git for release - script: | releaseBranch="origin/${BUILD_SOURCEBRANCH#refs/heads/}" From dc9a49994a588ba6d65f8ec4745ed803ec595005 Mon Sep 17 00:00:00 2001 From: pavelglac Date: Fri, 26 Jun 2026 13:07:20 +0200 Subject: [PATCH 05/18] Batch incremental payload patches Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/__tests__/executeWithSchema.test.ts | 171 ++++++++++++++++++ .../supermassive/src/executeWithSchema.ts | 2 + .../supermassive/src/executeWithoutSchema.ts | 49 ++++- packages/supermassive/src/types.ts | 1 + 4 files changed, 219 insertions(+), 4 deletions(-) diff --git a/packages/supermassive/src/__tests__/executeWithSchema.test.ts b/packages/supermassive/src/__tests__/executeWithSchema.test.ts index 7a2c3b778..bec9e11e3 100644 --- a/packages/supermassive/src/__tests__/executeWithSchema.test.ts +++ b/packages/supermassive/src/__tests__/executeWithSchema.test.ts @@ -727,4 +727,175 @@ describe("executeWithSchema - @defer behavior", () => { expect(errors).toHaveLength(1); expect(errors?.[0].message).toBe("Critical boom"); }); + + const messageListDefinitions = parse(` + type Query { messages: [Message!]! } + type Message { + id: String! + name: String + } + `); + + const messageListDocument = parse(` + { + messages { + id + ... on Message @defer { + name + } + } + } + `); + + test("batches deferred patches that complete within the batching interval", async () => { + const firstName = createDeferred(); + const secondName = createDeferred(); + const thirdName = createDeferred(); + + const result = await Promise.resolve( + executeWithSchema({ + document: messageListDocument, + definitions: messageListDefinitions, + enableEarlyExecution: true, + enableDeferredMerge: true, + enableIncrementalPayloadBatching: true, + resolvers: { + Query: { + messages: () => [ + { id: "1", name: firstName.promise }, + { id: "2", name: secondName.promise }, + { id: "3", name: thirdName.promise }, + ], + }, + }, + }), + ); + + expect(result).toMatchObject({ + initialResult: { + data: { + messages: [{ id: "1" }, { id: "2" }, { id: "3" }], + }, + hasNext: true, + }, + }); + + if (!("initialResult" in result)) { + throw new Error("Expected an incremental result"); + } + + const subsequentResultPromise = result.subsequentResults.next(); + firstName.resolve("Ada"); + setTimeout(() => secondName.resolve("Grace"), 0); + setTimeout(() => thirdName.resolve("Linus"), 0); + + await expect(subsequentResultPromise).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + name: "Ada", + }, + path: ["messages", 0], + }, + { + data: { + name: "Grace", + }, + path: ["messages", 1], + }, + { + data: { + name: "Linus", + }, + path: ["messages", 2], + }, + ], + hasNext: false, + }, + done: false, + }); + }); + + test("emits deferred patches individually by default", async () => { + const firstName = createDeferred(); + const secondName = createDeferred(); + const thirdName = createDeferred(); + + const result = await Promise.resolve( + executeWithSchema({ + document: messageListDocument, + definitions: messageListDefinitions, + enableEarlyExecution: true, + enableDeferredMerge: true, + resolvers: { + Query: { + messages: () => [ + { id: "1", name: firstName.promise }, + { id: "2", name: secondName.promise }, + { id: "3", name: thirdName.promise }, + ], + }, + }, + }), + ); + + if (!("initialResult" in result)) { + throw new Error("Expected an incremental result"); + } + + const firstSubsequentResultPromise = result.subsequentResults.next(); + firstName.resolve("Ada"); + + await expect(firstSubsequentResultPromise).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + name: "Ada", + }, + path: ["messages", 0], + }, + ], + hasNext: true, + }, + done: false, + }); + + const secondSubsequentResultPromise = result.subsequentResults.next(); + secondName.resolve("Grace"); + + await expect(secondSubsequentResultPromise).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + name: "Grace", + }, + path: ["messages", 1], + }, + ], + hasNext: true, + }, + done: false, + }); + + const thirdSubsequentResultPromise = result.subsequentResults.next(); + thirdName.resolve("Linus"); + + await expect(thirdSubsequentResultPromise).resolves.toMatchObject({ + value: { + incremental: [ + { + data: { + name: "Linus", + }, + path: ["messages", 2], + }, + ], + hasNext: false, + }, + done: false, + }); + }); }); diff --git a/packages/supermassive/src/executeWithSchema.ts b/packages/supermassive/src/executeWithSchema.ts index eed97b305..d4f9808d6 100644 --- a/packages/supermassive/src/executeWithSchema.ts +++ b/packages/supermassive/src/executeWithSchema.ts @@ -18,6 +18,7 @@ export function executeWithSchema({ enablePerEventContext, enableEarlyExecution, enableDeferredMerge, + enableIncrementalPayloadBatching, }: ExecutionWithSchemaArgs): PromiseOrValue { const extracted = extractMinimalViableSchemaForRequestDocument( buildASTSchema(definitions), @@ -40,5 +41,6 @@ export function executeWithSchema({ enablePerEventContext, enableEarlyExecution, enableDeferredMerge, + enableIncrementalPayloadBatching, }); } diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index 51aacdbb7..7456a13ac 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -82,6 +82,8 @@ const collectSubfields = memoize3( ) => _collectSubfields(exeContext, returnTypeName.name, fieldGroup), ); +const SUBSEQUENT_PAYLOAD_BATCH_INTERVAL_MS = 5; + /** * Terminology * @@ -126,6 +128,7 @@ export interface ExecutionContext { enablePerEventContext: boolean; enableEarlyExecution: boolean; enableDeferredMerge: boolean; + enableIncrementalPayloadBatching: boolean; } /** @@ -199,6 +202,7 @@ function buildExecutionContext( enablePerEventContext, enableEarlyExecution, enableDeferredMerge, + enableIncrementalPayloadBatching, } = args; assertValidExecutionArguments(document, variableValues); @@ -270,6 +274,8 @@ function buildExecutionContext( enablePerEventContext: enablePerEventContext ?? true, enableEarlyExecution: enableEarlyExecution ?? false, enableDeferredMerge: enableDeferredMerge ?? false, + enableIncrementalPayloadBatching: + enableIncrementalPayloadBatching ?? false, }; } @@ -2903,6 +2909,41 @@ function getCompletedIncrementalResults( return incrementalResults; } +function waitForCompletedSubsequentPayload( + exeContext: ExecutionContext, +): Promise { + return Promise.race( + Array.from(exeContext.subsequentPayloads).map((p) => p.promise), + ); +} + +function waitForBatchInterval(): Promise<"timeout"> { + return new Promise((resolve) => + setTimeout(resolve, SUBSEQUENT_PAYLOAD_BATCH_INTERVAL_MS, "timeout"), + ); +} + +async function getCompletedIncrementalResultsWithBatching( + exeContext: ExecutionContext, +): Promise> { + const incremental = getCompletedIncrementalResults(exeContext); + + while (incremental.length && exeContext.subsequentPayloads.size > 0) { + const result = await Promise.race([ + waitForCompletedSubsequentPayload(exeContext).then(() => "payload"), + waitForBatchInterval(), + ]); + + if (result === "timeout") { + break; + } + + incremental.push(...getCompletedIncrementalResults(exeContext)); + } + + return incremental; +} + function yieldSubsequentPayloads( exeContext: ExecutionContext, ): AsyncGenerator { @@ -2915,16 +2956,16 @@ function yieldSubsequentPayloads( return { value: undefined, done: true }; } - await Promise.race( - Array.from(exeContext.subsequentPayloads).map((p) => p.promise), - ); + await waitForCompletedSubsequentPayload(exeContext); if (isDone) { // a different call to next has exhausted all payloads return { value: undefined, done: true }; } - const incremental = getCompletedIncrementalResults(exeContext); + const incremental = exeContext.enableIncrementalPayloadBatching + ? await getCompletedIncrementalResultsWithBatching(exeContext) + : getCompletedIncrementalResults(exeContext); const hasNext = exeContext.subsequentPayloads.size > 0; if (!incremental.length && hasNext) { diff --git a/packages/supermassive/src/types.ts b/packages/supermassive/src/types.ts index 935181441..2857c54a7 100644 --- a/packages/supermassive/src/types.ts +++ b/packages/supermassive/src/types.ts @@ -259,6 +259,7 @@ export interface CommonExecutionArgs { enablePerEventContext?: boolean; enableEarlyExecution?: boolean; enableDeferredMerge?: boolean; + enableIncrementalPayloadBatching?: boolean; } export type ExecutionWithoutSchemaArgs = CommonExecutionArgs & { schemaFragment: SchemaFragment; From 5abfe918d167b857dd275d85fa8b39676adaa62c Mon Sep 17 00:00:00 2001 From: pavelglac Date: Fri, 26 Jun 2026 13:28:54 +0200 Subject: [PATCH 06/18] Fix incremental batching formatting Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- packages/supermassive/src/executeWithoutSchema.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index 7456a13ac..51241a313 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -274,8 +274,7 @@ function buildExecutionContext( enablePerEventContext: enablePerEventContext ?? true, enableEarlyExecution: enableEarlyExecution ?? false, enableDeferredMerge: enableDeferredMerge ?? false, - enableIncrementalPayloadBatching: - enableIncrementalPayloadBatching ?? false, + enableIncrementalPayloadBatching: enableIncrementalPayloadBatching ?? false, }; } From 08660fc177bd5467669e308cf8af18a2d77bf573 Mon Sep 17 00:00:00 2001 From: pavelglac Date: Mon, 29 Jun 2026 13:23:51 +0200 Subject: [PATCH 07/18] Wait one tick before deferred merge Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/__tests__/executeWithSchema.test.ts | 29 ++++++++++----- .../supermassive/src/executeWithoutSchema.ts | 37 +++++++++++++++++++ 2 files changed, 57 insertions(+), 9 deletions(-) diff --git a/packages/supermassive/src/__tests__/executeWithSchema.test.ts b/packages/supermassive/src/__tests__/executeWithSchema.test.ts index bec9e11e3..465070496 100644 --- a/packages/supermassive/src/__tests__/executeWithSchema.test.ts +++ b/packages/supermassive/src/__tests__/executeWithSchema.test.ts @@ -132,7 +132,7 @@ describe("executeWithSchema - @defer behavior", () => { ); } - test("returns the initial response as soon as critical fields are ready", async () => { + test("returns the initial response without waiting for deferred fields beyond the merge tick", async () => { const critical = createDeferred(); const deferred = createDeferred(); @@ -142,14 +142,7 @@ describe("executeWithSchema - @defer behavior", () => { }); critical.resolve("critical"); - const result = await Promise.race([ - resultPromise, - new Promise<"blocked">((resolve) => setTimeout(resolve, 0, "blocked")), - ]); - - if (result === "blocked") { - throw new Error("Initial response waited for deferred field"); - } + const result = await resultPromise; expect(result).toMatchObject({ initialResult: { @@ -185,6 +178,24 @@ describe("executeWithSchema - @defer behavior", () => { }); }); + test("includes deferred fields in the initial response when they complete on the merge tick", async () => { + const result = await executeTestQuery({ + critical: () => "critical", + deferred: () => + new Promise((resolve) => setTimeout(resolve, 0, "deferred")), + }); + + expect(result).toEqual({ + data: { + obj: { + critical: "critical", + deferred: "deferred", + }, + }, + }); + expect("initialResult" in (result as object)).toBe(false); + }); + test("includes deferred fields in the initial response when they complete before the critical fields", async () => { const critical = createDeferred(); diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index 51241a313..728621077 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -402,6 +402,7 @@ function executeOperation( function buildResponse( exeContext: ExecutionContext, data: PromiseOrValue | null>, + didWaitForDeferredMerge?: boolean, ): PromiseOrValue { if (isPromise(data)) { return data.then( @@ -416,6 +417,13 @@ function buildResponse( const hooks = exeContext.fieldExecutionHooks; try { if (exeContext.enableDeferredMerge) { + if ( + shouldWaitForDeferredMerge(exeContext, data, didWaitForDeferredMerge) + ) { + return waitForNextTick().then(() => + buildResponse(exeContext, data, true), + ); + } includeCompletedDeferredFragmentsInResult(exeContext, data); } @@ -453,6 +461,35 @@ function buildResponse( } } +function shouldWaitForDeferredMerge( + exeContext: ExecutionContext, + result: ObjMap | null, + didWaitForDeferredMerge: boolean | undefined, +): boolean { + if (didWaitForDeferredMerge || result == null) { + return false; + } + + let hasDeferredFragment = false; + for (const incrementalDataRecord of exeContext.subsequentPayloads) { + if (!isDeferredFragmentRecord(incrementalDataRecord)) { + continue; + } + + if (incrementalDataRecord.isCompleted) { + return false; + } + + hasDeferredFragment = true; + } + + return hasDeferredFragment; +} + +function waitForNextTick(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + /** * Implements the "Executing selection sets" section of the spec * for fields that must be executed serially. From dea81b5a9f3b8a464712140a4322ea2d0d9638fd Mon Sep 17 00:00:00 2001 From: pavelglac Date: Tue, 30 Jun 2026 15:41:46 +0200 Subject: [PATCH 08/18] Batch incremental payloads by queue Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/__tests__/executeWithSchema.test.ts | 6 ++--- .../supermassive/src/executeWithoutSchema.ts | 26 ++----------------- 2 files changed, 5 insertions(+), 27 deletions(-) diff --git a/packages/supermassive/src/__tests__/executeWithSchema.test.ts b/packages/supermassive/src/__tests__/executeWithSchema.test.ts index 465070496..4a1c6eeba 100644 --- a/packages/supermassive/src/__tests__/executeWithSchema.test.ts +++ b/packages/supermassive/src/__tests__/executeWithSchema.test.ts @@ -758,7 +758,7 @@ describe("executeWithSchema - @defer behavior", () => { } `); - test("batches deferred patches that complete within the batching interval", async () => { + test("batches deferred patches that complete in the same microtask queue", async () => { const firstName = createDeferred(); const secondName = createDeferred(); const thirdName = createDeferred(); @@ -797,8 +797,8 @@ describe("executeWithSchema - @defer behavior", () => { const subsequentResultPromise = result.subsequentResults.next(); firstName.resolve("Ada"); - setTimeout(() => secondName.resolve("Grace"), 0); - setTimeout(() => thirdName.resolve("Linus"), 0); + secondName.resolve("Grace"); + thirdName.resolve("Linus"); await expect(subsequentResultPromise).resolves.toMatchObject({ value: { diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index 728621077..e79e2e2c3 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -82,8 +82,6 @@ const collectSubfields = memoize3( ) => _collectSubfields(exeContext, returnTypeName.name, fieldGroup), ); -const SUBSEQUENT_PAYLOAD_BATCH_INTERVAL_MS = 5; - /** * Terminology * @@ -2953,31 +2951,11 @@ function waitForCompletedSubsequentPayload( ); } -function waitForBatchInterval(): Promise<"timeout"> { - return new Promise((resolve) => - setTimeout(resolve, SUBSEQUENT_PAYLOAD_BATCH_INTERVAL_MS, "timeout"), - ); -} - async function getCompletedIncrementalResultsWithBatching( exeContext: ExecutionContext, ): Promise> { - const incremental = getCompletedIncrementalResults(exeContext); - - while (incremental.length && exeContext.subsequentPayloads.size > 0) { - const result = await Promise.race([ - waitForCompletedSubsequentPayload(exeContext).then(() => "payload"), - waitForBatchInterval(), - ]); - - if (result === "timeout") { - break; - } - - incremental.push(...getCompletedIncrementalResults(exeContext)); - } - - return incremental; + await Promise.resolve(); + return getCompletedIncrementalResults(exeContext); } function yieldSubsequentPayloads( From 2e1210cb87037754db98a554bec0da610d9422cd Mon Sep 17 00:00:00 2001 From: pavelglac Date: Wed, 1 Jul 2026 11:07:17 +0200 Subject: [PATCH 09/18] Use timed incremental payload batching Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/__tests__/executeWithSchema.test.ts | 10 ++-- .../supermassive/src/executeWithoutSchema.ts | 53 +++++++++++++++---- 2 files changed, 49 insertions(+), 14 deletions(-) diff --git a/packages/supermassive/src/__tests__/executeWithSchema.test.ts b/packages/supermassive/src/__tests__/executeWithSchema.test.ts index 4a1c6eeba..3fa209e12 100644 --- a/packages/supermassive/src/__tests__/executeWithSchema.test.ts +++ b/packages/supermassive/src/__tests__/executeWithSchema.test.ts @@ -132,7 +132,7 @@ describe("executeWithSchema - @defer behavior", () => { ); } - test("returns the initial response without waiting for deferred fields beyond the merge tick", async () => { + test("returns the initial response without waiting for deferred fields beyond the merge timeout", async () => { const critical = createDeferred(); const deferred = createDeferred(); @@ -178,7 +178,7 @@ describe("executeWithSchema - @defer behavior", () => { }); }); - test("includes deferred fields in the initial response when they complete on the merge tick", async () => { + test("includes deferred fields in the initial response when they complete within the merge timeout", async () => { const result = await executeTestQuery({ critical: () => "critical", deferred: () => @@ -758,7 +758,7 @@ describe("executeWithSchema - @defer behavior", () => { } `); - test("batches deferred patches that complete in the same microtask queue", async () => { + test("batches deferred patches that complete within the batching timeout", async () => { const firstName = createDeferred(); const secondName = createDeferred(); const thirdName = createDeferred(); @@ -797,8 +797,8 @@ describe("executeWithSchema - @defer behavior", () => { const subsequentResultPromise = result.subsequentResults.next(); firstName.resolve("Ada"); - secondName.resolve("Grace"); - thirdName.resolve("Linus"); + setTimeout(() => secondName.resolve("Grace"), 0); + setTimeout(() => thirdName.resolve("Linus"), 0); await expect(subsequentResultPromise).resolves.toMatchObject({ value: { diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index e79e2e2c3..ac6f80a34 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -82,6 +82,8 @@ const collectSubfields = memoize3( ) => _collectSubfields(exeContext, returnTypeName.name, fieldGroup), ); +const INCREMENTAL_PAYLOAD_BATCH_TIMEOUT_MS = 10; + /** * Terminology * @@ -418,7 +420,7 @@ function buildResponse( if ( shouldWaitForDeferredMerge(exeContext, data, didWaitForDeferredMerge) ) { - return waitForNextTick().then(() => + return waitForDeferredMerge(exeContext).then(() => buildResponse(exeContext, data, true), ); } @@ -474,18 +476,28 @@ function shouldWaitForDeferredMerge( continue; } - if (incrementalDataRecord.isCompleted) { - return false; + if (!incrementalDataRecord.isCompleted) { + hasDeferredFragment = true; + break; } - - hasDeferredFragment = true; } return hasDeferredFragment; } -function waitForNextTick(): Promise { - return new Promise((resolve) => setTimeout(resolve, 0)); +function waitForDeferredMerge(exeContext: ExecutionContext): Promise { + const pendingDeferredFragments = Array.from(exeContext.subsequentPayloads) + .filter( + (incrementalDataRecord) => + isDeferredFragmentRecord(incrementalDataRecord) && + !incrementalDataRecord.isCompleted, + ) + .map((incrementalDataRecord) => incrementalDataRecord.promise); + + return Promise.race([ + Promise.all(pendingDeferredFragments).then(() => undefined), + waitForIncrementalPayloadBatchTimeout(), + ]); } /** @@ -2954,8 +2966,31 @@ function waitForCompletedSubsequentPayload( async function getCompletedIncrementalResultsWithBatching( exeContext: ExecutionContext, ): Promise> { - await Promise.resolve(); - return getCompletedIncrementalResults(exeContext); + const incremental = getCompletedIncrementalResults(exeContext); + + if (incremental.length && exeContext.subsequentPayloads.size > 0) { + await waitForSubsequentPayloadBatch(exeContext); + incremental.push(...getCompletedIncrementalResults(exeContext)); + } + + return incremental; +} + +function waitForSubsequentPayloadBatch( + exeContext: ExecutionContext, +): Promise { + return Promise.race([ + Promise.all( + Array.from(exeContext.subsequentPayloads).map((p) => p.promise), + ).then(() => undefined), + waitForIncrementalPayloadBatchTimeout(), + ]); +} + +function waitForIncrementalPayloadBatchTimeout(): Promise { + return new Promise((resolve) => + setTimeout(resolve, INCREMENTAL_PAYLOAD_BATCH_TIMEOUT_MS), + ); } function yieldSubsequentPayloads( From 9f5bbe820fd75eba1a8c519454948acf0d44820d Mon Sep 17 00:00:00 2001 From: Pavel Glac Date: Wed, 1 Jul 2026 15:14:45 +0200 Subject: [PATCH 10/18] changes --- .../supermassive/src/executeWithoutSchema.ts | 98 ++++++------------- 1 file changed, 30 insertions(+), 68 deletions(-) diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index ac6f80a34..4d2423878 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -82,7 +82,7 @@ const collectSubfields = memoize3( ) => _collectSubfields(exeContext, returnTypeName.name, fieldGroup), ); -const INCREMENTAL_PAYLOAD_BATCH_TIMEOUT_MS = 10; +const INCREMENTAL_PAYLOAD_BATCH_TIMEOUT_MS = 20; /** * Terminology @@ -402,7 +402,7 @@ function executeOperation( function buildResponse( exeContext: ExecutionContext, data: PromiseOrValue | null>, - didWaitForDeferredMerge?: boolean, + deferredMergeAwaited = false, ): PromiseOrValue { if (isPromise(data)) { return data.then( @@ -416,15 +416,14 @@ function buildResponse( const hooks = exeContext.fieldExecutionHooks; try { - if (exeContext.enableDeferredMerge) { - if ( - shouldWaitForDeferredMerge(exeContext, data, didWaitForDeferredMerge) - ) { - return waitForDeferredMerge(exeContext).then(() => + if (exeContext.enableDeferredMerge && data !== null) { + const pendingDeferredFragments = + includeCompletedDeferredFragmentsInResult(exeContext, data); + if (!deferredMergeAwaited && pendingDeferredFragments.length > 0) { + return raceIncrementalPayloadBatch(pendingDeferredFragments).then(() => buildResponse(exeContext, data, true), ); } - includeCompletedDeferredFragmentsInResult(exeContext, data); } const initialResult = @@ -461,45 +460,6 @@ function buildResponse( } } -function shouldWaitForDeferredMerge( - exeContext: ExecutionContext, - result: ObjMap | null, - didWaitForDeferredMerge: boolean | undefined, -): boolean { - if (didWaitForDeferredMerge || result == null) { - return false; - } - - let hasDeferredFragment = false; - for (const incrementalDataRecord of exeContext.subsequentPayloads) { - if (!isDeferredFragmentRecord(incrementalDataRecord)) { - continue; - } - - if (!incrementalDataRecord.isCompleted) { - hasDeferredFragment = true; - break; - } - } - - return hasDeferredFragment; -} - -function waitForDeferredMerge(exeContext: ExecutionContext): Promise { - const pendingDeferredFragments = Array.from(exeContext.subsequentPayloads) - .filter( - (incrementalDataRecord) => - isDeferredFragmentRecord(incrementalDataRecord) && - !incrementalDataRecord.isCompleted, - ) - .map((incrementalDataRecord) => incrementalDataRecord.promise); - - return Promise.race([ - Promise.all(pendingDeferredFragments).then(() => undefined), - waitForIncrementalPayloadBatchTimeout(), - ]); -} - /** * Implements the "Executing selection sets" section of the spec * for fields that must be executed serially. @@ -648,17 +608,17 @@ function startExecutingPatches( function includeCompletedDeferredFragmentsInResult( exeContext: ExecutionContext, - result: ObjMap | null, -): void { - if (result == null) { - return; - } + result: ObjMap, +): Array> { + const pendingDeferredFragments: Array> = []; for (const incrementalDataRecord of exeContext.subsequentPayloads) { - if ( - !isDeferredFragmentRecord(incrementalDataRecord) || - !incrementalDataRecord.isCompleted - ) { + if (!isDeferredFragmentRecord(incrementalDataRecord)) { + continue; + } + + if (!incrementalDataRecord.isCompleted) { + pendingDeferredFragments.push(incrementalDataRecord.promise); continue; } @@ -677,6 +637,8 @@ function includeCompletedDeferredFragmentsInResult( exeContext.subsequentPayloads.delete(incrementalDataRecord); } + + return pendingDeferredFragments; } /** @@ -2955,14 +2917,6 @@ function getCompletedIncrementalResults( return incrementalResults; } -function waitForCompletedSubsequentPayload( - exeContext: ExecutionContext, -): Promise { - return Promise.race( - Array.from(exeContext.subsequentPayloads).map((p) => p.promise), - ); -} - async function getCompletedIncrementalResultsWithBatching( exeContext: ExecutionContext, ): Promise> { @@ -2978,11 +2932,17 @@ async function getCompletedIncrementalResultsWithBatching( function waitForSubsequentPayloadBatch( exeContext: ExecutionContext, +): Promise { + return raceIncrementalPayloadBatch( + Array.from(exeContext.subsequentPayloads).map((p) => p.promise), + ); +} + +function raceIncrementalPayloadBatch( + promises: Array>, ): Promise { return Promise.race([ - Promise.all( - Array.from(exeContext.subsequentPayloads).map((p) => p.promise), - ).then(() => undefined), + Promise.all(promises).then(() => undefined), waitForIncrementalPayloadBatchTimeout(), ]); } @@ -3005,7 +2965,9 @@ function yieldSubsequentPayloads( return { value: undefined, done: true }; } - await waitForCompletedSubsequentPayload(exeContext); + await Promise.race( + Array.from(exeContext.subsequentPayloads).map((p) => p.promise), + ); if (isDone) { // a different call to next has exhausted all payloads From 20ebd1b1c7a911f73042f2d353db69abcabb250a Mon Sep 17 00:00:00 2001 From: Pavel Glac Date: Wed, 1 Jul 2026 17:42:14 +0200 Subject: [PATCH 11/18] cleanup --- packages/supermassive/src/executeWithoutSchema.ts | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index 4d2423878..4505ab80b 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -2923,21 +2923,15 @@ async function getCompletedIncrementalResultsWithBatching( const incremental = getCompletedIncrementalResults(exeContext); if (incremental.length && exeContext.subsequentPayloads.size > 0) { - await waitForSubsequentPayloadBatch(exeContext); + await raceIncrementalPayloadBatch( + Array.from(exeContext.subsequentPayloads).map((p) => p.promise), + ); incremental.push(...getCompletedIncrementalResults(exeContext)); } return incremental; } -function waitForSubsequentPayloadBatch( - exeContext: ExecutionContext, -): Promise { - return raceIncrementalPayloadBatch( - Array.from(exeContext.subsequentPayloads).map((p) => p.promise), - ); -} - function raceIncrementalPayloadBatch( promises: Array>, ): Promise { From 7f57ea74bbdbe3ab440807e9f96bc279ab17a1d8 Mon Sep 17 00:00:00 2001 From: pavelglac Date: Thu, 2 Jul 2026 15:52:03 +0200 Subject: [PATCH 12/18] Configure incremental batching timeout Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/__tests__/executeWithSchema.test.ts | 21 ++++++--- .../supermassive/src/executeWithoutSchema.ts | 44 ++++++++++++------- packages/supermassive/src/types.ts | 2 +- 3 files changed, 43 insertions(+), 24 deletions(-) diff --git a/packages/supermassive/src/__tests__/executeWithSchema.test.ts b/packages/supermassive/src/__tests__/executeWithSchema.test.ts index 3fa209e12..7b2e32827 100644 --- a/packages/supermassive/src/__tests__/executeWithSchema.test.ts +++ b/packages/supermassive/src/__tests__/executeWithSchema.test.ts @@ -72,13 +72,17 @@ describe("executeWithSchema - @defer behavior", () => { type Resolver = () => string | Promise; - function executeTestQuery(Obj: { critical: Resolver; deferred: Resolver }) { + function executeTestQuery( + Obj: { critical: Resolver; deferred: Resolver }, + enableIncrementalPayloadBatching?: number, + ) { return Promise.resolve( executeWithSchema({ document, definitions, enableEarlyExecution: true, enableDeferredMerge: true, + enableIncrementalPayloadBatching, resolvers: { Query: { obj: () => ({}), @@ -179,11 +183,14 @@ describe("executeWithSchema - @defer behavior", () => { }); test("includes deferred fields in the initial response when they complete within the merge timeout", async () => { - const result = await executeTestQuery({ - critical: () => "critical", - deferred: () => - new Promise((resolve) => setTimeout(resolve, 0, "deferred")), - }); + const result = await executeTestQuery( + { + critical: () => "critical", + deferred: () => + new Promise((resolve) => setTimeout(resolve, 0, "deferred")), + }, + 10, + ); expect(result).toEqual({ data: { @@ -769,7 +776,7 @@ describe("executeWithSchema - @defer behavior", () => { definitions: messageListDefinitions, enableEarlyExecution: true, enableDeferredMerge: true, - enableIncrementalPayloadBatching: true, + enableIncrementalPayloadBatching: 10, resolvers: { Query: { messages: () => [ diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index 4505ab80b..1b34e3711 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -82,8 +82,6 @@ const collectSubfields = memoize3( ) => _collectSubfields(exeContext, returnTypeName.name, fieldGroup), ); -const INCREMENTAL_PAYLOAD_BATCH_TIMEOUT_MS = 20; - /** * Terminology * @@ -128,7 +126,7 @@ export interface ExecutionContext { enablePerEventContext: boolean; enableEarlyExecution: boolean; enableDeferredMerge: boolean; - enableIncrementalPayloadBatching: boolean; + enableIncrementalPayloadBatching?: number; } /** @@ -274,7 +272,7 @@ function buildExecutionContext( enablePerEventContext: enablePerEventContext ?? true, enableEarlyExecution: enableEarlyExecution ?? false, enableDeferredMerge: enableDeferredMerge ?? false, - enableIncrementalPayloadBatching: enableIncrementalPayloadBatching ?? false, + enableIncrementalPayloadBatching, }; } @@ -419,10 +417,16 @@ function buildResponse( if (exeContext.enableDeferredMerge && data !== null) { const pendingDeferredFragments = includeCompletedDeferredFragmentsInResult(exeContext, data); - if (!deferredMergeAwaited && pendingDeferredFragments.length > 0) { - return raceIncrementalPayloadBatch(pendingDeferredFragments).then(() => - buildResponse(exeContext, data, true), - ); + const batchTimeout = exeContext.enableIncrementalPayloadBatching; + if ( + batchTimeout !== undefined && + !deferredMergeAwaited && + pendingDeferredFragments.length > 0 + ) { + return raceIncrementalPayloadBatch( + pendingDeferredFragments, + batchTimeout, + ).then(() => buildResponse(exeContext, data, true)); } } @@ -2919,12 +2923,14 @@ function getCompletedIncrementalResults( async function getCompletedIncrementalResultsWithBatching( exeContext: ExecutionContext, + batchTimeout: number, ): Promise> { const incremental = getCompletedIncrementalResults(exeContext); if (incremental.length && exeContext.subsequentPayloads.size > 0) { await raceIncrementalPayloadBatch( Array.from(exeContext.subsequentPayloads).map((p) => p.promise), + batchTimeout, ); incremental.push(...getCompletedIncrementalResults(exeContext)); } @@ -2934,17 +2940,18 @@ async function getCompletedIncrementalResultsWithBatching( function raceIncrementalPayloadBatch( promises: Array>, + batchTimeout: number, ): Promise { return Promise.race([ Promise.all(promises).then(() => undefined), - waitForIncrementalPayloadBatchTimeout(), + waitForIncrementalPayloadBatchTimeout(batchTimeout), ]); } -function waitForIncrementalPayloadBatchTimeout(): Promise { - return new Promise((resolve) => - setTimeout(resolve, INCREMENTAL_PAYLOAD_BATCH_TIMEOUT_MS), - ); +function waitForIncrementalPayloadBatchTimeout( + batchTimeout: number, +): Promise { + return new Promise((resolve) => setTimeout(resolve, batchTimeout)); } function yieldSubsequentPayloads( @@ -2968,9 +2975,14 @@ function yieldSubsequentPayloads( return { value: undefined, done: true }; } - const incremental = exeContext.enableIncrementalPayloadBatching - ? await getCompletedIncrementalResultsWithBatching(exeContext) - : getCompletedIncrementalResults(exeContext); + const batchTimeout = exeContext.enableIncrementalPayloadBatching; + const incremental = + batchTimeout !== undefined + ? await getCompletedIncrementalResultsWithBatching( + exeContext, + batchTimeout, + ) + : getCompletedIncrementalResults(exeContext); const hasNext = exeContext.subsequentPayloads.size > 0; if (!incremental.length && hasNext) { diff --git a/packages/supermassive/src/types.ts b/packages/supermassive/src/types.ts index 2857c54a7..e4a439640 100644 --- a/packages/supermassive/src/types.ts +++ b/packages/supermassive/src/types.ts @@ -259,7 +259,7 @@ export interface CommonExecutionArgs { enablePerEventContext?: boolean; enableEarlyExecution?: boolean; enableDeferredMerge?: boolean; - enableIncrementalPayloadBatching?: boolean; + enableIncrementalPayloadBatching?: number; } export type ExecutionWithoutSchemaArgs = CommonExecutionArgs & { schemaFragment: SchemaFragment; From 00c82548f69e11f23f746dbf89372dc2199720d4 Mon Sep 17 00:00:00 2001 From: pavelglac Date: Fri, 3 Jul 2026 07:00:19 +0200 Subject: [PATCH 13/18] Trim incremental batching tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/__tests__/executeWithSchema.test.ts | 59 ++++--------------- 1 file changed, 11 insertions(+), 48 deletions(-) diff --git a/packages/supermassive/src/__tests__/executeWithSchema.test.ts b/packages/supermassive/src/__tests__/executeWithSchema.test.ts index 7b2e32827..a724de63d 100644 --- a/packages/supermassive/src/__tests__/executeWithSchema.test.ts +++ b/packages/supermassive/src/__tests__/executeWithSchema.test.ts @@ -136,52 +136,6 @@ describe("executeWithSchema - @defer behavior", () => { ); } - test("returns the initial response without waiting for deferred fields beyond the merge timeout", async () => { - const critical = createDeferred(); - const deferred = createDeferred(); - - const resultPromise = executeTestQuery({ - critical: () => critical.promise, - deferred: () => deferred.promise, - }); - - critical.resolve("critical"); - const result = await resultPromise; - - expect(result).toMatchObject({ - initialResult: { - data: { - obj: { - critical: "critical", - }, - }, - hasNext: true, - }, - }); - - if (!("initialResult" in result)) { - throw new Error("Expected an incremental result"); - } - - const subsequentResultPromise = result.subsequentResults.next(); - deferred.resolve("deferred"); - - await expect(subsequentResultPromise).resolves.toMatchObject({ - value: { - incremental: [ - { - data: { - deferred: "deferred", - }, - path: ["obj"], - }, - ], - hasNext: false, - }, - done: false, - }); - }); - test("includes deferred fields in the initial response when they complete within the merge timeout", async () => { const result = await executeTestQuery( { @@ -835,12 +789,12 @@ describe("executeWithSchema - @defer behavior", () => { }); }); - test("emits deferred patches individually by default", async () => { + test("emits deferred patches individually without additional waiting by default", async () => { const firstName = createDeferred(); const secondName = createDeferred(); const thirdName = createDeferred(); - const result = await Promise.resolve( + const resultPromise = Promise.resolve( executeWithSchema({ document: messageListDocument, definitions: messageListDefinitions, @@ -858,6 +812,15 @@ describe("executeWithSchema - @defer behavior", () => { }), ); + const result = await Promise.race([ + resultPromise, + new Promise<"blocked">((resolve) => setTimeout(resolve, 0, "blocked")), + ]); + + if (result === "blocked") { + throw new Error("Initial response waited without batching enabled"); + } + if (!("initialResult" in result)) { throw new Error("Expected an incremental result"); } From 248dbb5690d7830878bb9b2dc0d20715ff3e5967 Mon Sep 17 00:00:00 2001 From: pavelglac Date: Fri, 3 Jul 2026 09:48:09 +0200 Subject: [PATCH 14/18] Remove redundant defer test Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/__tests__/executeWithSchema.test.ts | 48 ------------------- 1 file changed, 48 deletions(-) diff --git a/packages/supermassive/src/__tests__/executeWithSchema.test.ts b/packages/supermassive/src/__tests__/executeWithSchema.test.ts index a724de63d..89bbee922 100644 --- a/packages/supermassive/src/__tests__/executeWithSchema.test.ts +++ b/packages/supermassive/src/__tests__/executeWithSchema.test.ts @@ -286,54 +286,6 @@ describe("executeWithSchema - @defer behavior", () => { }); }); - test("keeps legacy defer behavior when early execution is disabled", async () => { - const result = await executeWithSchema({ - document, - definitions, - enableEarlyExecution: false, - enableDeferredMerge: false, - resolvers: { - Query: { - obj: () => ({}), - }, - Obj: { - critical: () => "critical", - deferred: () => "deferred", - }, - }, - }); - - expect(result).toMatchObject({ - initialResult: { - data: { - obj: { - critical: "critical", - }, - }, - hasNext: true, - }, - }); - - if (!("initialResult" in result)) { - throw new Error("Expected an incremental result"); - } - - await expect(result.subsequentResults.next()).resolves.toMatchObject({ - value: { - incremental: [ - { - data: { - deferred: "deferred", - }, - path: ["obj"], - }, - ], - hasNext: false, - }, - done: false, - }); - }); - test("includes nested deferred field in the initial response when it settles before critical", async () => { const critical = createDeferred(); From 4758c77943afcd50e21e8e643fbe571468b7b62f Mon Sep 17 00:00:00 2001 From: pavelglac Date: Fri, 3 Jul 2026 10:05:04 +0200 Subject: [PATCH 15/18] Keep only essential defer flag tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../src/__tests__/executeWithSchema.test.ts | 405 ------------------ 1 file changed, 405 deletions(-) diff --git a/packages/supermassive/src/__tests__/executeWithSchema.test.ts b/packages/supermassive/src/__tests__/executeWithSchema.test.ts index 89bbee922..24d27a604 100644 --- a/packages/supermassive/src/__tests__/executeWithSchema.test.ts +++ b/packages/supermassive/src/__tests__/executeWithSchema.test.ts @@ -29,47 +29,6 @@ describe("executeWithSchema - @defer behavior", () => { } `); - const nestedDefinitions = parse(` - type Query { obj: Obj } - type Obj { - critical: String - nested: Nested - } - type Nested { - deferred: String - } - `); - - const nestedDocument = parse(` - { - obj { - critical - ... on Obj @defer { - nested { - deferred - } - } - } - } - `); - - const mutationDefinitions = parse(` - type Query { noop: String } - type Mutation { - critical: String - deferred: String - } - `); - - const mutationDocument = parse(` - mutation { - critical - ... on Mutation @defer { - deferred - } - } - `); - type Resolver = () => string | Promise; function executeTestQuery( @@ -93,49 +52,6 @@ describe("executeWithSchema - @defer behavior", () => { ); } - function executeNestedTestQuery(Obj: { - critical: Resolver; - deferred: Resolver; - }) { - return Promise.resolve( - executeWithSchema({ - document: nestedDocument, - definitions: nestedDefinitions, - enableEarlyExecution: true, - enableDeferredMerge: true, - resolvers: { - Query: { - obj: () => ({}), - }, - Obj: { - critical: Obj.critical, - nested: () => ({}), - }, - Nested: { - deferred: Obj.deferred, - }, - }, - }), - ); - } - - function executeTestMutation(Mutation: { - critical: Resolver; - deferred: Resolver; - }) { - return Promise.resolve( - executeWithSchema({ - document: mutationDocument, - definitions: mutationDefinitions, - enableEarlyExecution: true, - enableDeferredMerge: true, - resolvers: { - Mutation, - }, - }), - ); - } - test("includes deferred fields in the initial response when they complete within the merge timeout", async () => { const result = await executeTestQuery( { @@ -157,58 +73,6 @@ describe("executeWithSchema - @defer behavior", () => { expect("initialResult" in (result as object)).toBe(false); }); - test("includes deferred fields in the initial response when they complete before the critical fields", async () => { - const critical = createDeferred(); - - const resultPromise = executeTestQuery({ - deferred: () => Promise.resolve("deferred"), - critical: () => critical.promise, - }); - - setTimeout(() => critical.resolve("critical"), 0); - - await expect(resultPromise).resolves.toEqual({ - data: { - obj: { - critical: "critical", - deferred: "deferred", - }, - }, - }); - }); - - test("surfaces deferred errors at the top level when piggybacked onto the initial response", async () => { - const critical = createDeferred(); - - const resultPromise = executeTestQuery({ - deferred: () => { - throw new Error("Deferred boom"); - }, - critical: () => critical.promise, - }); - - setTimeout(() => critical.resolve("critical"), 0); - - const result = await resultPromise; - - expect(result).toMatchObject({ - data: { - obj: { - critical: "critical", - deferred: null, - }, - }, - }); - expect("initialResult" in (result as object)).toBe(false); - expect( - (result as { errors?: ReadonlyArray<{ message: string }> }).errors, - ).toHaveLength(1); - expect( - (result as { errors: ReadonlyArray<{ message: string }> }).errors[0] - .message, - ).toBe("Deferred boom"); - }); - test("folds a synchronous deferred field into the initial response instead of streaming it", async () => { const result = await executeTestQuery({ critical: () => "critical", @@ -230,146 +94,6 @@ describe("executeWithSchema - @defer behavior", () => { expect("initialResult" in (result as object)).toBe(false); }); - test("delivers a synchronously-completed deferred field incrementally when merge is disabled", async () => { - // Early execution starts the deferred fragment eagerly, but with merge - // disabled it is never folded into the initial response (mirroring - // graphql-js's `enableEarlyExecution`): it is always delivered as a - // separate incremental payload. - const result = await executeWithSchema({ - document, - definitions, - enableEarlyExecution: true, - enableDeferredMerge: false, - resolvers: { - Query: { - obj: () => ({}), - }, - Obj: { - critical: () => "critical", - deferred: () => "deferred", - }, - }, - }); - - expect(result).toMatchObject({ - initialResult: { - data: { - obj: { - critical: "critical", - }, - }, - hasNext: true, - }, - }); - - if (!("initialResult" in result)) { - throw new Error("Expected an incremental result"); - } - - expect(result.initialResult.data).toEqual({ - obj: { critical: "critical" }, - }); - - await expect(result.subsequentResults.next()).resolves.toMatchObject({ - value: { - incremental: [ - { - data: { - deferred: "deferred", - }, - path: ["obj"], - }, - ], - hasNext: false, - }, - done: false, - }); - }); - - test("includes nested deferred field in the initial response when it settles before critical", async () => { - const critical = createDeferred(); - - const resultPromise = executeNestedTestQuery({ - deferred: () => Promise.resolve("deferred"), - critical: () => critical.promise, - }); - - setTimeout(() => critical.resolve("critical"), 0); - - await expect(resultPromise).resolves.toEqual({ - data: { - obj: { - critical: "critical", - nested: { - deferred: "deferred", - }, - }, - }, - }); - }); - - test("streams nested deferred field when critical fields are ready first", async () => { - const deferred = createDeferred(); - - const result = await executeNestedTestQuery({ - critical: () => "critical", - deferred: () => deferred.promise, - }); - - expect(result).toMatchObject({ - initialResult: { - data: { - obj: { - critical: "critical", - }, - }, - hasNext: true, - }, - }); - - if (!("initialResult" in result)) { - throw new Error("Expected an incremental result"); - } - - const subsequentResultPromise = result.subsequentResults.next(); - deferred.resolve("deferred"); - - await expect(subsequentResultPromise).resolves.toMatchObject({ - value: { - incremental: [ - { - data: { - nested: { - deferred: "deferred", - }, - }, - path: ["obj"], - }, - ], - hasNext: false, - }, - done: false, - }); - }); - - test("includes deferred mutation fields in the initial response when they settle before critical", async () => { - const critical = createDeferred(); - - const resultPromise = executeTestMutation({ - critical: () => critical.promise, - deferred: () => Promise.resolve("deferred"), - }); - - setTimeout(() => critical.resolve("critical"), 0); - - await expect(resultPromise).resolves.toEqual({ - data: { - critical: "critical", - deferred: "deferred", - }, - }); - }); - const twoDeferDefinitions = parse(` type Query { obj: Obj } type Obj { @@ -471,135 +195,6 @@ describe("executeWithSchema - @defer behavior", () => { }); }); - const ifDocument = parse(` - query ($shouldDefer: Boolean!) { - obj { - critical - ... on Obj @defer(if: $shouldDefer) { - deferred - } - } - } - `); - - function executeIfTestQuery( - Obj: { critical: Resolver; deferred: Resolver }, - shouldDefer: boolean, - ) { - return Promise.resolve( - executeWithSchema({ - document: ifDocument, - definitions, - enableEarlyExecution: true, - enableDeferredMerge: true, - variableValues: { shouldDefer }, - resolvers: { - Query: { - obj: () => ({}), - }, - Obj, - }, - }), - ); - } - - test("streams the deferred field when @defer(if: true) and critical is ready first", async () => { - const deferred = createDeferred(); - - const result = await executeIfTestQuery( - { - critical: () => "critical", - deferred: () => deferred.promise, - }, - true, - ); - - expect(result).toMatchObject({ - initialResult: { - data: { - obj: { - critical: "critical", - }, - }, - hasNext: true, - }, - }); - - if (!("initialResult" in result)) { - throw new Error("Expected an incremental result"); - } - - const subsequentResultPromise = result.subsequentResults.next(); - deferred.resolve("deferred"); - - await expect(subsequentResultPromise).resolves.toMatchObject({ - value: { - incremental: [ - { - data: { - deferred: "deferred", - }, - path: ["obj"], - }, - ], - hasNext: false, - }, - done: false, - }); - }); - - test("does not defer the field when @defer(if: false), even if it is slow", async () => { - const deferred = createDeferred(); - - const resultPromise = executeIfTestQuery( - { - critical: () => "critical", - deferred: () => deferred.promise, - }, - false, - ); - - // With `if: false` the fragment is not deferred, so the field is critical - // and the response must wait for it instead of streaming it. - const raced = await Promise.race([ - resultPromise, - new Promise<"pending">((resolve) => setTimeout(resolve, 0, "pending")), - ]); - - expect(raced).toBe("pending"); - - deferred.resolve("deferred"); - - await expect(resultPromise).resolves.toEqual({ - data: { - obj: { - critical: "critical", - deferred: "deferred", - }, - }, - }); - }); - - test("returns a single non-incremental result when @defer(if: false)", async () => { - const result = await executeIfTestQuery( - { - critical: () => "critical", - deferred: () => "deferred", - }, - false, - ); - - expect(result).toEqual({ - data: { - obj: { - critical: "critical", - deferred: "deferred", - }, - }, - }); - expect("initialResult" in (result as object)).toBe(false); - }); - // Schema where the critical field is non-null: when it errors, the null // bubbles up and destroys the whole `obj` branch the deferred fragment is // anchored to. From ad6db4cba307496b21c8672f2ab0150130eb9af2 Mon Sep 17 00:00:00 2001 From: Pavel Glac Date: Fri, 3 Jul 2026 13:18:48 +0200 Subject: [PATCH 16/18] revert pipeline change --- .azure-devops/graphitation-release.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.azure-devops/graphitation-release.yml b/.azure-devops/graphitation-release.yml index 367a146c9..8d9676411 100644 --- a/.azure-devops/graphitation-release.yml +++ b/.azure-devops/graphitation-release.yml @@ -55,6 +55,7 @@ extends: - script: | git config user.email "gql-svc@microsoft.com" git config user.name "Graphitation Service Account" + git fetch --depth=2 displayName: Configure git for release - script: | releaseBranch="origin/${BUILD_SOURCEBRANCH#refs/heads/}" From 7b9ed62a24cb5e63a8d7c5113d13f9a4d3e3a3ff Mon Sep 17 00:00:00 2001 From: Pavel Glac Date: Fri, 3 Jul 2026 13:19:40 +0200 Subject: [PATCH 17/18] Change files --- ...tion-supermassive-8876a266-3c69-4f89-819b-4fc57b5b4470.json} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename change/{@graphitation-supermassive-b119265a-cda6-437b-ad83-e893facef588.json => @graphitation-supermassive-8876a266-3c69-4f89-819b-4fc57b5b4470.json} (79%) diff --git a/change/@graphitation-supermassive-b119265a-cda6-437b-ad83-e893facef588.json b/change/@graphitation-supermassive-8876a266-3c69-4f89-819b-4fc57b5b4470.json similarity index 79% rename from change/@graphitation-supermassive-b119265a-cda6-437b-ad83-e893facef588.json rename to change/@graphitation-supermassive-8876a266-3c69-4f89-819b-4fc57b5b4470.json index 7aa033ac9..625ed330f 100644 --- a/change/@graphitation-supermassive-b119265a-cda6-437b-ad83-e893facef588.json +++ b/change/@graphitation-supermassive-8876a266-3c69-4f89-819b-4fc57b5b4470.json @@ -2,6 +2,6 @@ "type": "minor", "comment": "early execution and merge", "packageName": "@graphitation/supermassive", - "email": "pavelglac@gmail.com", + "email": "pavelglac@microsoft.com", "dependentChangeType": "patch" } From 50a55c2ce62c565d5eae0b1025153bbe1b01633e Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 3 Jul 2026 11:27:52 +0000 Subject: [PATCH 18/18] fix: require numeric batch timeout guard --- packages/supermassive/src/executeWithoutSchema.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/supermassive/src/executeWithoutSchema.ts b/packages/supermassive/src/executeWithoutSchema.ts index 1b34e3711..b4e947a89 100644 --- a/packages/supermassive/src/executeWithoutSchema.ts +++ b/packages/supermassive/src/executeWithoutSchema.ts @@ -419,7 +419,7 @@ function buildResponse( includeCompletedDeferredFragmentsInResult(exeContext, data); const batchTimeout = exeContext.enableIncrementalPayloadBatching; if ( - batchTimeout !== undefined && + typeof batchTimeout === "number" && !deferredMergeAwaited && pendingDeferredFragments.length > 0 ) { @@ -2977,7 +2977,7 @@ function yieldSubsequentPayloads( const batchTimeout = exeContext.enableIncrementalPayloadBatching; const incremental = - batchTimeout !== undefined + typeof batchTimeout === "number" ? await getCompletedIncrementalResultsWithBatching( exeContext, batchTimeout,