From d7c9a59cfcfc7ab5aff73c99606c3535148199c5 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Sun, 26 Jul 2026 09:07:30 +0100 Subject: [PATCH 01/13] Fix memory leak in subscriptions --- grafast/grafast/src/prepare.ts | 57 ++++++++++---------- grafast/grafast/src/utils.ts | 32 +++++++++++ postgraphile/postgraphile/graphile.config.ts | 26 +++++++++ 3 files changed, 85 insertions(+), 30 deletions(-) diff --git a/grafast/grafast/src/prepare.ts b/grafast/grafast/src/prepare.ts index ef571ed146..aa6fb6a13f 100644 --- a/grafast/grafast/src/prepare.ts +++ b/grafast/grafast/src/prepare.ts @@ -49,9 +49,9 @@ import type { StreamMaybeMoreableArray, StreamMoreableArray, } from "./interfaces.ts"; -import { promiseWithResolve } from "./promiseWithResolve.ts"; import { timeSource } from "./timeSource.ts"; import { + abortable, arrayOfLength, asyncIteratorWithCleanup, isPromiseLike, @@ -325,7 +325,7 @@ function executePreemptive( onError: ErrorBehavior, outputDataAsString: boolean, executionTimeout: number | null, - abortSignal: AbortSignal, + requestAbortSignal: AbortSignal, ): PromiseOrDirect< ExecutionResult | AsyncGenerator > { @@ -368,7 +368,7 @@ function executePreemptive( stopTime, // toSerialize: [], eventEmitter: args[$$eventEmitter], - abortSignal, + abortSignal: requestAbortSignal, }; const bucketPromise = executeBucket(rootBucket, requestContext); @@ -480,13 +480,15 @@ function executePreemptive( // `releaseUnusedIterators(rootBucket, rootBucketIndex, null)` here. const arr = bucketRootValue as StreamMoreableArray; const stream = arr[$$streamMore]; - // Do the async iterable - let stopped = false; - const { promise: abortPromise, resolve: resolveAbort } = - promiseWithResolve(); + const iteratorAbortController = new AbortController(); + requestAbortSignal.addEventListener( + "abort", + () => iteratorAbortController.abort(), + { once: true }, + ); + const iteratorAbortSignal = iteratorAbortController.signal; const iterator = newIterator((e) => { - stopped = true; - resolveAbort(); + iteratorAbortController.abort(); if (e != null) { try { const result = stream.throw?.(e); @@ -511,33 +513,28 @@ function executePreemptive( let i = 0; // eslint-disable-next-line no-constant-condition while (true) { - const next = await Promise.race([abortPromise, stream.next()]); - if (stopped || !next) { - break; - } - if (!next) { - iterator.throw(new Error("Invalid iteration")).then(null, noop); + const next = await abortable(iteratorAbortSignal, stream.next()); + if (next === undefined || next.done) { break; } - const { done, value } = next; - if (done) { - break; - } - const payload = await Promise.race([ - abortPromise, - executeStreamPayload(value, i), - ]); + const payload = await abortable( + iteratorAbortSignal, + executeStreamPayload(next.value, i), + ); if (payload === undefined) { break; } if (isAsyncIterable(payload)) { - // FIXME: do we need to avoid 'for await' because it can cause the - // stream to exit late if we're waiting on a promise and the stream - // exits in the interrim? We're assuming that no promises will be - // sufficiently long-lived for this to be an issue right now. - // TODO: should probably tie all this into an AbortController/signal too - for await (const entry of payload) { - iterator.push(entry); + const payloadIterator = payload[Symbol.asyncIterator](); + while (true) { + const next = await abortable( + iteratorAbortSignal, + payloadIterator.next(), + ); + if (next === undefined || next.done) { + break; + } + iterator.push(next.value); } } else { iterator.push(payload); diff --git a/grafast/grafast/src/utils.ts b/grafast/grafast/src/utils.ts index 834ef60237..5ea13e15c8 100644 --- a/grafast/grafast/src/utils.ts +++ b/grafast/grafast/src/utils.ts @@ -1549,3 +1549,35 @@ export function markSyncAndSafe< } return fn; } + +/** + * If `signal` is aborted, returns undefined. Otherwise returns equivalent to + * `promiseOrValue` except if the signal is aborted first it will resolve with + * `undefined` immediately. + */ +export function abortable( + signal: AbortSignal, + promiseOrValue: PromiseLike | T, +): T | undefined | Promise { + if (signal.aborted) { + return undefined; + } + if (!isPromiseLike(promiseOrValue)) { + return promiseOrValue; + } + const promise = promiseOrValue; + return new Promise((resolve, reject) => { + const resolveWithoutArgs = () => resolve(undefined); + signal.addEventListener("abort", resolveWithoutArgs, { once: true }); + return promise.then( + (val) => { + signal.removeEventListener("abort", resolveWithoutArgs); + resolve(val); + }, + (e) => { + signal.removeEventListener("abort", resolveWithoutArgs); + reject(e); + }, + ); + }); +} diff --git a/postgraphile/postgraphile/graphile.config.ts b/postgraphile/postgraphile/graphile.config.ts index 1ec76f7d7c..04e7edf1fd 100644 --- a/postgraphile/postgraphile/graphile.config.ts +++ b/postgraphile/postgraphile/graphile.config.ts @@ -434,6 +434,11 @@ const preset: GraphileConfig.Preset = { sub(topic: String!): Int gql(max: Int! = 10): Int slow: String + fast: FastSubscriptionPayload + } + type FastSubscriptionPayload { + datetime: String + count: Int } `, objects: { @@ -499,6 +504,27 @@ const preset: GraphileConfig.Preset = { [sleep], ), }, + fast: { + resolve: EXPORTABLE( + () => + function resolve(e) { + return e; + }, + [], + ), + subscribe: EXPORTABLE( + (sleep) => + async function* subscribe() { + let count = 0; + while (true) { + ++count; + yield { datetime: new Date().toISOString(), count }; + await sleep(5); + } + }, + [sleep], + ), + }, }, }, }, From 859dadb11a65b3d4e2c6ec6960baa997c6d8a571 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Sun, 26 Jul 2026 09:20:00 +0100 Subject: [PATCH 02/13] Abortable takes valueForAbort and is now exported --- grafast/grafast/src/index.ts | 3 +++ grafast/grafast/src/prepare.ts | 16 ++++++++++------ grafast/grafast/src/utils.ts | 22 ++++++++++------------ 3 files changed, 23 insertions(+), 18 deletions(-) diff --git a/grafast/grafast/src/index.ts b/grafast/grafast/src/index.ts index 11aac9113c..f3cf52a159 100644 --- a/grafast/grafast/src/index.ts +++ b/grafast/grafast/src/index.ts @@ -268,6 +268,7 @@ import type { ObjectTypeSpec, } from "./utils.ts"; import { + abortable, arrayOfLength, arraysMatch, asyncIteratorWithCleanup, @@ -414,6 +415,7 @@ export { $$idempotent, $$inhibit, $$verbatim, + abortable, access, AccessStep, applyInput, @@ -586,6 +588,7 @@ exportAsMany("grafast", { __TrackedValueStep, __ValueStep, inspect, + abortable, access, get, AccessStep, diff --git a/grafast/grafast/src/prepare.ts b/grafast/grafast/src/prepare.ts index aa6fb6a13f..19f96cf2a8 100644 --- a/grafast/grafast/src/prepare.ts +++ b/grafast/grafast/src/prepare.ts @@ -513,14 +513,17 @@ function executePreemptive( let i = 0; // eslint-disable-next-line no-constant-condition while (true) { - const next = await abortable(iteratorAbortSignal, stream.next()); - if (next === undefined || next.done) { + const rawNext = stream.next(); + const next = isPromiseLike(rawNext) + ? await abortable(iteratorAbortSignal, undefined, rawNext) + : rawNext; + if (next === undefined || next.done || iteratorAbortSignal.aborted) { break; } - const payload = await abortable( - iteratorAbortSignal, - executeStreamPayload(next.value, i), - ); + const rawPayload = executeStreamPayload(next.value, i); + const payload = isPromiseLike(rawPayload) + ? await abortable(iteratorAbortSignal, undefined, rawPayload) + : rawPayload; if (payload === undefined) { break; } @@ -529,6 +532,7 @@ function executePreemptive( while (true) { const next = await abortable( iteratorAbortSignal, + undefined, payloadIterator.next(), ); if (next === undefined || next.done) { diff --git a/grafast/grafast/src/utils.ts b/grafast/grafast/src/utils.ts index 5ea13e15c8..62cbc0b9da 100644 --- a/grafast/grafast/src/utils.ts +++ b/grafast/grafast/src/utils.ts @@ -1551,23 +1551,21 @@ export function markSyncAndSafe< } /** - * If `signal` is aborted, returns undefined. Otherwise returns equivalent to - * `promiseOrValue` except if the signal is aborted first it will resolve with - * `undefined` immediately. + * If `signal` is aborted, returns `valueForAbort`. Otherwise returns + * equivalent to `promiseOrValue` except if the signal is aborted first it will + * resolve with `valueForAbort` immediately. */ -export function abortable( +export function abortable( signal: AbortSignal, - promiseOrValue: PromiseLike | T, -): T | undefined | Promise { + valueForAbort: F, + promiseOrValue: PromiseLike, +): Promise { if (signal.aborted) { - return undefined; - } - if (!isPromiseLike(promiseOrValue)) { - return promiseOrValue; + return Promise.resolve(valueForAbort); } const promise = promiseOrValue; - return new Promise((resolve, reject) => { - const resolveWithoutArgs = () => resolve(undefined); + return new Promise((resolve, reject) => { + const resolveWithoutArgs = () => resolve(valueForAbort); signal.addEventListener("abort", resolveWithoutArgs, { once: true }); return promise.then( (val) => { From b2b624c3a5dc6d8c7ee800d67a6e57ee1ab830a7 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Sun, 26 Jul 2026 09:21:26 +0100 Subject: [PATCH 03/13] Fix potential memory leak in introspection plugin --- .../src/plugins/PgIntrospectionPlugin.ts | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts b/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts index ab7f92feb0..361b3ceed9 100644 --- a/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts +++ b/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts @@ -5,7 +5,7 @@ import { withSuperuserPgClientFromPgService, } from "@dataplan/pg"; import type { PromiseOrDirect, Step } from "grafast"; -import { constant, context, noop, object, promiseWithResolve } from "grafast"; +import { abortable,constant, context, noop, object, promiseWithResolve } from "grafast"; import type { GatherPluginContext } from "graphile-build"; import { EXPORTABLE, gatherConfig } from "graphile-build"; import type { @@ -35,7 +35,8 @@ import { import { version } from "../version.ts"; import { watchFixtures } from "../watchFixtures.ts"; -const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); +const sleep = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); /** Someone else created */ const CLASH_CODES = ["23505", "42P06", "42P07", "42710"]; @@ -685,9 +686,11 @@ export const PgIntrospectionPlugin: GraphileConfig.Plugin = { let eventStream = await pgService.pgSubscriber.subscribe("postgraphile_watch"); const $$stop = Symbol("stop"); - const { resolve, promise: abort } = - promiseWithResolve(); - unlistens.push(() => resolve($$stop)); + + const controller = new AbortController(); + const signal = controller.signal; + unlistens.push(() => void controller.abort()); + const regather = () => { // Delete the introspection results info.cache.introspectionResultsPromise = null; @@ -697,7 +700,7 @@ export const PgIntrospectionPlugin: GraphileConfig.Plugin = { callback(); }; const waitNext = () => { - const next = Promise.race([abort, eventStream.next()]); + const next = abortable(signal, $$stop, eventStream.next()); next.then( (event) => { if (event === $$stop) { @@ -739,7 +742,7 @@ export const PgIntrospectionPlugin: GraphileConfig.Plugin = { console.error( `postgraphile_watch subscription failed (${e}); waiting ${delay.toFixed(0)}ms then re-establishing`, ); - const result = await Promise.race([sleep(delay), abort]); + const result = await abortable(signal, $$stop, sleep(delay)); if (result === $$stop) { return; } From eb574c3c7d5ec919f546352409505b7967c243d8 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Sun, 26 Jul 2026 10:23:26 +0100 Subject: [PATCH 04/13] Safety --- grafast/grafast/src/engine/distributor.ts | 10 ++++++---- grafast/grafast/src/utils.ts | 10 ++++++---- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/grafast/grafast/src/engine/distributor.ts b/grafast/grafast/src/engine/distributor.ts index 6448ae540a..8ca625e129 100644 --- a/grafast/grafast/src/engine/distributor.ts +++ b/grafast/grafast/src/engine/distributor.ts @@ -27,10 +27,12 @@ export function isDistributor( } // Save on garbage collection by just using this promise for everything -const DONE_PROMISE: Promise> = Promise.resolve({ - done: true, - value: undefined, -}); +const DONE_PROMISE: Promise> = Promise.resolve( + Object.freeze({ + done: true, + value: undefined, + }), +); /** * Creates a "distributor" for the sourceIterable such that the dependent steps diff --git a/grafast/grafast/src/utils.ts b/grafast/grafast/src/utils.ts index 62cbc0b9da..df5625e63b 100644 --- a/grafast/grafast/src/utils.ts +++ b/grafast/grafast/src/utils.ts @@ -1458,10 +1458,12 @@ export const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); // Save on garbage collection by just using this promise for everything -const DONE_PROMISE: Promise> = Promise.resolve({ - done: true, - value: undefined, -}); +const DONE_PROMISE: Promise> = Promise.resolve( + Object.freeze({ + done: true, + value: undefined, + }), +); /** * Returns a new version of `iterable` that calls `callback()` on termination, From fc364efd49062fafb073377583e89fe9953f311b Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Sun, 26 Jul 2026 10:45:01 +0100 Subject: [PATCH 05/13] More memory efficient distributor pauses --- grafast/grafast/src/engine/distributor.ts | 42 ++++++++++++----------- 1 file changed, 22 insertions(+), 20 deletions(-) diff --git a/grafast/grafast/src/engine/distributor.ts b/grafast/grafast/src/engine/distributor.ts index 8ca625e129..93d82f8336 100644 --- a/grafast/grafast/src/engine/distributor.ts +++ b/grafast/grafast/src/engine/distributor.ts @@ -1,12 +1,8 @@ import * as assert from "../assert.ts"; import { isDev, noop } from "../dev.ts"; import type { Maybe } from "../interfaces.ts"; -import { - type PromiseWithResolve, - promiseWithResolve, -} from "../promiseWithResolve.ts"; import type { Step } from "../step.ts"; -import { arrayOfLength, isPromiseLike, sleep } from "../utils.ts"; +import { arrayOfLength, isPromiseLike } from "../utils.ts"; const DEFAULT_DISTRIBUTOR_BUFFER_SIZE = 1001; const DEFAULT_DISTRIBUTOR_BUFFER_SIZE_INCREMENT = 1001; @@ -101,13 +97,19 @@ export function distributor( */ const buffer: Array>> = []; - // Easy way to resolve a promise for slowing down the fastest consumer - let wmi: PromiseWithResolve | null = null; - function lowWaterMarkIncreased(): PromiseLike { - if (wmi === null) { - wmi = promiseWithResolve(); - } - return wmi.promise; + // Consumers waiting for the low-water mark to advance. + let lowWaterMarkWaiters: Set<() => void> | null = null; + function waitForLowWaterMarkOrPause(): Promise { + return new Promise((resolve) => { + const waiters = (lowWaterMarkWaiters ??= new Set()); + const done = () => { + clearTimeout(timeout); + waiters.delete(done); + resolve(); + }; + const timeout = setTimeout(done, pauseDuration); + waiters.add(done); + }); } /** @@ -187,11 +189,14 @@ export function distributor( } // Announce that the lowWaterMark advanced - if (advanced && wmi !== null) { + if (advanced && lowWaterMarkWaiters !== null) { // Avoid race condition - const deferred = wmi; - wmi = null; - deferred.resolve(); + const waiters = lowWaterMarkWaiters; + lowWaterMarkWaiters = null; + + for (const resolve of waiters) { + resolve(); + } } } } @@ -288,10 +293,7 @@ export function distributor( // Whoa there! Getting a little ahead of ourselves! Wait for the slowest // consumer to advance (or for it to time out), before resolving. // const oldLowWaterMark = lowWaterMark; - const next = Promise.race([ - lowWaterMarkIncreased(), - sleep(pauseDuration), - ]).then( + const next = waitForLowWaterMarkOrPause().then( // const advanced = lowWaterMark > oldLowWaterMark; // TODO: should we wait a little longer if we did advance so we're // not creating a new timer for each and every low watermark From e5af19e2790dfce11e89d93a64f074f453c06cfb Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Sun, 26 Jul 2026 11:21:13 +0100 Subject: [PATCH 06/13] More efficient schema waiting --- grafast/grafserv/src/core/base.ts | 58 +++++++++++++++++++++++-------- 1 file changed, 44 insertions(+), 14 deletions(-) diff --git a/grafast/grafserv/src/core/base.ts b/grafast/grafserv/src/core/base.ts index 1be1e77832..189ef3aae7 100644 --- a/grafast/grafserv/src/core/base.ts +++ b/grafast/grafserv/src/core/base.ts @@ -572,7 +572,7 @@ function defaultMakeGetExecutionConfig(): ( let latestSchema: GraphQLSchema; let latestSchemaOrPromise: PromiseOrDirect; let latestParseAndValidate: ReturnType; - let schemaPrepare: Promise | null = null; + let schemaPrepareWaiters: Set> | null = null; return function getExecutionConfig(this: GrafservBase) { // Get up to date schema, in case we're in watch mode @@ -581,16 +581,40 @@ function defaultMakeGetExecutionConfig(): ( if (schemaOrPromise !== latestSchemaOrPromise) { latestSchemaOrPromise = schemaOrPromise; if ("then" in schemaOrPromise) { - schemaPrepare = (async () => { - latestSchema = await schemaOrPromise; - latestSchemaOrPromise = schemaOrPromise; - latestParseAndValidate = makeParseAndValidateFunction( - latestSchema, - resolvedPreset, - dynamicOptions, - ); - schemaPrepare = null; - return true; + const prepareWaiters = new Set>(); + schemaPrepareWaiters = prepareWaiters; + const releaseWaiters = (error?: Error) => { + // Make sure new waiters have their own batch + if (schemaPrepareWaiters === prepareWaiters) { + schemaPrepareWaiters = null; + } + if (error) { + for (const waiter of prepareWaiters) { + waiter.reject(error); + } + } else { + for (const waiter of prepareWaiters) { + waiter.resolve(true); + } + } + prepareWaiters.clear(); + }; + + // Kick off an async task that waits for the schema to be ready, + // completes setup, then informs all waiters of the result. + (async () => { + try { + latestSchema = await schemaOrPromise; + latestSchemaOrPromise = schemaOrPromise; + latestParseAndValidate = makeParseAndValidateFunction( + latestSchema, + resolvedPreset, + dynamicOptions, + ); + releaseWaiters(); + } catch (error) { + releaseWaiters(error); + } })(); } else { if (latestSchema === schemaOrPromise) { @@ -605,10 +629,16 @@ function defaultMakeGetExecutionConfig(): ( } } } - if (schemaPrepare !== null) { + if (schemaPrepareWaiters !== null) { + const prepareWaiters = schemaPrepareWaiters; const sleeper = sleep(dynamicOptions.schemaWaitTime); - const schemaReadyPromise = Promise.race([schemaPrepare, sleeper.promise]); - return schemaReadyPromise.then((schemaReady) => { + const waiter = Promise.withResolvers(); + prepareWaiters.add(waiter); + sleeper.promise.then(() => { + prepareWaiters.delete(waiter); + waiter.resolve(false); + }, waiter.reject); + return waiter.promise.then((schemaReady) => { sleeper.release(); if (schemaReady !== true) { // Handle missing schema From ea46d6582d3b14808e449c543e89a8f07283e546 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Sun, 26 Jul 2026 11:33:41 +0100 Subject: [PATCH 07/13] docs(changeset): Address a slow memory leak in GraphQL subscriptions that can cause OOM errors when a single subscription hits many hundreds of thousands of events. The cause was racing an abortPromise with each event, causing the abort promise to build up a large list of callbacks in its internal Promise mechanics when the event always won the race. The fix was to move from using a promise for abort to an AbortController, where the abort task can be released after each event successfully resolves, avoiding memory buildup. Also applied this fix to similar patterns elsewhere in the codebase, albeit places much less sensitive to this issue. --- .changeset/mighty-buckets-wish.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 .changeset/mighty-buckets-wish.md diff --git a/.changeset/mighty-buckets-wish.md b/.changeset/mighty-buckets-wish.md new file mode 100644 index 0000000000..3664b50da6 --- /dev/null +++ b/.changeset/mighty-buckets-wish.md @@ -0,0 +1,16 @@ +--- +"graphile-build-pg": patch +"postgraphile": patch +"grafserv": patch +"grafast": patch +--- + +Address a slow memory leak in GraphQL subscriptions that can cause OOM errors +when a single subscription hits many hundreds of thousands of events. The cause +was racing an abortPromise with each event, causing the abort promise to build +up a large list of callbacks in its internal Promise mechanics when the event +always won the race. The fix was to move from using a promise for abort to an +AbortController, where the abort task can be released after each event +successfully resolves, avoiding memory buildup. Also applied this fix to similar +patterns elsewhere in the codebase, albeit places much less sensitive to this +issue. From 874af57bcc2a3c2c80f0f7b5560d85b188d9b676 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Wed, 29 Jul 2026 07:13:41 +0100 Subject: [PATCH 08/13] Lint --- grafast/grafserv/src/core/base.ts | 2 +- .../graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/grafast/grafserv/src/core/base.ts b/grafast/grafserv/src/core/base.ts index 189ef3aae7..3db65f9ead 100644 --- a/grafast/grafserv/src/core/base.ts +++ b/grafast/grafserv/src/core/base.ts @@ -602,7 +602,7 @@ function defaultMakeGetExecutionConfig(): ( // Kick off an async task that waits for the schema to be ready, // completes setup, then informs all waiters of the result. - (async () => { + void (async () => { try { latestSchema = await schemaOrPromise; latestSchemaOrPromise = schemaOrPromise; diff --git a/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts b/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts index 361b3ceed9..7192b9b009 100644 --- a/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts +++ b/graphile-build/graphile-build-pg/src/plugins/PgIntrospectionPlugin.ts @@ -5,7 +5,7 @@ import { withSuperuserPgClientFromPgService, } from "@dataplan/pg"; import type { PromiseOrDirect, Step } from "grafast"; -import { abortable,constant, context, noop, object, promiseWithResolve } from "grafast"; +import { abortable, constant, context, noop, object } from "grafast"; import type { GatherPluginContext } from "graphile-build"; import { EXPORTABLE, gatherConfig } from "graphile-build"; import type { From f54eb4878c743b833b0ec617f7974d76dc77e7a0 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Wed, 29 Jul 2026 07:31:21 +0100 Subject: [PATCH 09/13] Make sure iterators are closed --- grafast/grafast/src/prepare.ts | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/grafast/grafast/src/prepare.ts b/grafast/grafast/src/prepare.ts index 19f96cf2a8..c003c8d70d 100644 --- a/grafast/grafast/src/prepare.ts +++ b/grafast/grafast/src/prepare.ts @@ -517,7 +517,12 @@ function executePreemptive( const next = isPromiseLike(rawNext) ? await abortable(iteratorAbortSignal, undefined, rawNext) : rawNext; - if (next === undefined || next.done || iteratorAbortSignal.aborted) { + if (next?.done) { + // Stream already exited + break; + } + if (next === undefined || iteratorAbortSignal.aborted) { + stream.return?.(); break; } const rawPayload = executeStreamPayload(next.value, i); @@ -535,7 +540,12 @@ function executePreemptive( undefined, payloadIterator.next(), ); - if (next === undefined || next.done) { + if (next?.done) { + // Iterator already exited + break; + } + if (next === undefined || iteratorAbortSignal.aborted) { + payloadIterator.return?.(undefined); break; } iterator.push(next.value); From 946adf5d9d84441631241ea2f43a939320c83c09 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Wed, 29 Jul 2026 07:34:21 +0100 Subject: [PATCH 10/13] Make sure iterator is released if an error is thrown --- grafast/grafast/src/prepare.ts | 57 ++++++++++++++++++---------------- 1 file changed, 31 insertions(+), 26 deletions(-) diff --git a/grafast/grafast/src/prepare.ts b/grafast/grafast/src/prepare.ts index c003c8d70d..aef8480775 100644 --- a/grafast/grafast/src/prepare.ts +++ b/grafast/grafast/src/prepare.ts @@ -525,35 +525,40 @@ function executePreemptive( stream.return?.(); break; } - const rawPayload = executeStreamPayload(next.value, i); - const payload = isPromiseLike(rawPayload) - ? await abortable(iteratorAbortSignal, undefined, rawPayload) - : rawPayload; - if (payload === undefined) { - break; - } - if (isAsyncIterable(payload)) { - const payloadIterator = payload[Symbol.asyncIterator](); - while (true) { - const next = await abortable( - iteratorAbortSignal, - undefined, - payloadIterator.next(), - ); - if (next?.done) { - // Iterator already exited - break; - } - if (next === undefined || iteratorAbortSignal.aborted) { - payloadIterator.return?.(undefined); - break; + try { + const rawPayload = executeStreamPayload(next.value, i); + const payload = isPromiseLike(rawPayload) + ? await abortable(iteratorAbortSignal, undefined, rawPayload) + : rawPayload; + if (payload === undefined) { + break; + } + if (isAsyncIterable(payload)) { + const payloadIterator = payload[Symbol.asyncIterator](); + while (true) { + const next = await abortable( + iteratorAbortSignal, + undefined, + payloadIterator.next(), + ); + if (next?.done) { + // Iterator already exited + break; + } + if (next === undefined || iteratorAbortSignal.aborted) { + payloadIterator.return?.(undefined); + break; + } + iterator.push(next.value); } - iterator.push(next.value); + } else { + iterator.push(payload); } - } else { - iterator.push(payload); + i++; + } catch (error) { + iterator.return?.(); + throw error; } - i++; } })() .then( From bbfda775cc384a338c71ef3620638e5b0752f264 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Mon, 3 Aug 2026 20:54:03 +0100 Subject: [PATCH 11/13] Unregister listener when iterator terminates --- grafast/grafast/src/prepare.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/grafast/grafast/src/prepare.ts b/grafast/grafast/src/prepare.ts index aef8480775..105976b5d3 100644 --- a/grafast/grafast/src/prepare.ts +++ b/grafast/grafast/src/prepare.ts @@ -481,14 +481,20 @@ function executePreemptive( const arr = bucketRootValue as StreamMoreableArray; const stream = arr[$$streamMore]; const iteratorAbortController = new AbortController(); + const abortIteratorWhenRequestAborts = () => + iteratorAbortController.abort(); requestAbortSignal.addEventListener( "abort", - () => iteratorAbortController.abort(), + abortIteratorWhenRequestAborts, { once: true }, ); const iteratorAbortSignal = iteratorAbortController.signal; const iterator = newIterator((e) => { iteratorAbortController.abort(); + requestAbortSignal.removeEventListener( + "abort", + abortIteratorWhenRequestAborts, + ); if (e != null) { try { const result = stream.throw?.(e); From 1318f132b8e69a62f06ca0ef6f48245c3b82ff31 Mon Sep 17 00:00:00 2001 From: Benjie Gillam Date: Mon, 3 Aug 2026 20:56:24 +0100 Subject: [PATCH 12/13] Await promises --- grafast/grafast/src/prepare.ts | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/grafast/grafast/src/prepare.ts b/grafast/grafast/src/prepare.ts index 105976b5d3..e94b91388b 100644 --- a/grafast/grafast/src/prepare.ts +++ b/grafast/grafast/src/prepare.ts @@ -528,7 +528,10 @@ function executePreemptive( break; } if (next === undefined || iteratorAbortSignal.aborted) { - stream.return?.(); + const result = stream.return?.(); + if (isPromiseLike(result)) { + result.then(null, noop); + } break; } try { @@ -552,7 +555,10 @@ function executePreemptive( break; } if (next === undefined || iteratorAbortSignal.aborted) { - payloadIterator.return?.(undefined); + const result = payloadIterator.return?.(undefined); + if (isPromiseLike(result)) { + result.then(null, noop); + } break; } iterator.push(next.value); @@ -562,7 +568,10 @@ function executePreemptive( } i++; } catch (error) { - iterator.return?.(); + const result = iterator.return?.(); + if (isPromiseLike(result)) { + result.then(null, noop); + } throw error; } } From 635b832c6d3e7e9aa3086b24e0fb5130bbd6cea7 Mon Sep 17 00:00:00 2001 From: Benjie Date: Fri, 7 Aug 2026 15:23:39 +0100 Subject: [PATCH 13/13] Apply suggestion from @benjie --- grafast/grafast/src/utils.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/grafast/grafast/src/utils.ts b/grafast/grafast/src/utils.ts index df5625e63b..790adc18e7 100644 --- a/grafast/grafast/src/utils.ts +++ b/grafast/grafast/src/utils.ts @@ -1560,12 +1560,11 @@ export function markSyncAndSafe< export function abortable( signal: AbortSignal, valueForAbort: F, - promiseOrValue: PromiseLike, + promise: PromiseLike, ): Promise { if (signal.aborted) { return Promise.resolve(valueForAbort); } - const promise = promiseOrValue; return new Promise((resolve, reject) => { const resolveWithoutArgs = () => resolve(valueForAbort); signal.addEventListener("abort", resolveWithoutArgs, { once: true });