From 70a3856076e55e7456c18c4a2ded4a3dfa27995c Mon Sep 17 00:00:00 2001 From: Nic Polumeyv <162764842+Nic-Polumeyv@users.noreply.github.com> Date: Thu, 13 Aug 2026 15:33:50 -0400 Subject: [PATCH 1/5] fix: stop touching the query.live stream controller after teardown --- .changeset/pre/live-query-stream-teardown.md | 5 + .../src/runtime/server/remote-functions.js | 102 ++++++++++++------ 2 files changed, 73 insertions(+), 34 deletions(-) create mode 100644 .changeset/pre/live-query-stream-teardown.md diff --git a/.changeset/pre/live-query-stream-teardown.md b/.changeset/pre/live-query-stream-teardown.md new file mode 100644 index 000000000000..7e6667abb7a6 --- /dev/null +++ b/.changeset/pre/live-query-stream-teardown.md @@ -0,0 +1,5 @@ +--- +'@sveltejs/kit': patch +--- + +fix: don't touch the `query.live` stream controller after teardown, and make response cancellation observable via the generator's `request.signal` diff --git a/packages/kit/src/runtime/server/remote-functions.js b/packages/kit/src/runtime/server/remote-functions.js index 43674bc0abf9..c5d09f617a4a 100644 --- a/packages/kit/src/runtime/server/remote-functions.js +++ b/packages/kit/src/runtime/server/remote-functions.js @@ -88,61 +88,93 @@ async function handle_remote_call_internal(event, state, options, manifest, id) new URL(event.request.url).searchParams.get('payload') ); - const generator = internals.run(event, state, parse_remote_arg(payload)); + // aborted by `teardown()`, so unlike the request signal it also fires on + // response teardown, which the generator could otherwise never observe + const cancellation = new AbortController(); - const encoder = new TextEncoder(); + const live_event = { + ...event, + request: new Request(event.request, { signal: cancellation.signal }) + }; + + const generator = internals.run(live_event, state, parse_remote_arg(payload)); - let closed = false; + const encoder = new TextEncoder(); /** @type {ReturnType | undefined} */ let keep_alive; + // the controller is only reachable through this sink, which goes inert on + // close — either side can tear the stream down while `pull` is suspended + /** @type {ReturnType} */ + let sink; + + /** @param {ReadableStreamDefaultController} controller */ + function create_sink(controller) { + let open = true; + + return { + /** @param {string} data */ + write(data) { + if (!open) return; + controller.enqueue(encoder.encode(data)); + schedule_keep_alive(); + }, + close() { + if (!open) return; + open = false; + controller.close(); + }, + /** the platform already closed the controller (stream cancellation) */ + abandon() { + open = false; + } + }; + } + /** * (Re)schedule the keep-alive comment. Called whenever a message is sent, so * that a keep-alive is only emitted once `KEEP_ALIVE_INTERVAL` has elapsed * without any other activity. - * @param {ReadableStreamDefaultController} controller */ - function schedule_keep_alive(controller) { + function schedule_keep_alive() { clearTimeout(keep_alive); keep_alive = setTimeout(() => { - if (closed || event.request.signal.aborted) return; // SSE comments (lines starting with `:`) are ignored by the client - controller.enqueue(encoder.encode(': keep-alive\n\n')); - schedule_keep_alive(controller); + sink.write(': keep-alive\n\n'); }, KEEP_ALIVE_INTERVAL); } - /** - * @param {ReadableStreamDefaultController} controller - * @param {any} payload - */ - function send(controller, payload) { - controller.enqueue(encoder.encode('data: ' + JSON.stringify(payload) + '\n\n')); - schedule_keep_alive(controller); + /** @param {any} payload */ + function send(payload) { + sink.write('data: ' + JSON.stringify(payload) + '\n\n'); } /** @type {string | undefined} */ let result = undefined; - async function cancel() { - if (closed) return; - closed = true; + let torn_down = false; + + async function teardown() { + if (torn_down) return; + torn_down = true; clearTimeout(keep_alive); + cancellation.abort(); + sink.close(); await generator.return(undefined); } - event.request.signal.addEventListener('abort', cancel, { once: true }); + event.request.signal.addEventListener('abort', teardown, { once: true }); return new Response( new ReadableStream({ start(controller) { - schedule_keep_alive(controller); + sink = create_sink(controller); + schedule_keep_alive(); }, - async pull(controller) { - if (event.request.signal.aborted) { - await cancel(); - controller.close(); + async pull() { + if (torn_down || event.request.signal.aborted) { + await teardown(); return; } @@ -150,15 +182,15 @@ async function handle_remote_call_internal(event, state, options, manifest, id) while (true) { const { value, done } = await generator.next(); - if (done) { - await cancel(); - controller.close(); + // teardown may have started while we were suspended + if (done || torn_down) { + await teardown(); return; } // only send changed data if (result !== (result = stringify(value))) { - send(controller, { + send({ type: 'result', result }); @@ -167,9 +199,9 @@ async function handle_remote_call_internal(event, state, options, manifest, id) } } } catch (error) { - if (!event.request.signal.aborted) { + if (!torn_down) { if (error instanceof Redirect) { - send(controller, { + send({ type: 'redirect', location: error.location }); @@ -181,18 +213,20 @@ async function handle_remote_call_internal(event, state, options, manifest, id) error ); - send(controller, { + send({ type: 'error', error: transformed }); } } - await cancel(); - controller.close(); + await teardown(); } }, - cancel + async cancel() { + sink.abandon(); + await teardown(); + } }), { headers: { From bfe314c87e8b288674db8bd5376d2c730c08ebc4 Mon Sep 17 00:00:00 2001 From: Nic Polumeyv <162764842+Nic-Polumeyv@users.noreply.github.com> Date: Thu, 13 Aug 2026 16:08:58 -0400 Subject: [PATCH 2/5] use shared text_encoder --- packages/kit/src/runtime/server/remote-functions.js | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/packages/kit/src/runtime/server/remote-functions.js b/packages/kit/src/runtime/server/remote-functions.js index c5d09f617a4a..74f380887f84 100644 --- a/packages/kit/src/runtime/server/remote-functions.js +++ b/packages/kit/src/runtime/server/remote-functions.js @@ -15,6 +15,7 @@ import { normalize_error } from '../../utils/error.js'; import { check_incorrect_fail_use, get_action_location } from './page/actions.js'; import { DEV } from 'esm-env'; import { deserialize_binary_form } from '../form-utils.js'; +import { text_encoder } from '../utils.js'; import { with_version_header } from './utils.js'; /** @@ -99,8 +100,6 @@ async function handle_remote_call_internal(event, state, options, manifest, id) const generator = internals.run(live_event, state, parse_remote_arg(payload)); - const encoder = new TextEncoder(); - /** @type {ReturnType | undefined} */ let keep_alive; @@ -117,7 +116,7 @@ async function handle_remote_call_internal(event, state, options, manifest, id) /** @param {string} data */ write(data) { if (!open) return; - controller.enqueue(encoder.encode(data)); + controller.enqueue(text_encoder.encode(data)); schedule_keep_alive(); }, close() { From dd2749148f5760439d519a0cb7e558ac7d874dd2 Mon Sep 17 00:00:00 2001 From: Nic Polumeyv <162764842+Nic-Polumeyv@users.noreply.github.com> Date: Thu, 13 Aug 2026 17:43:57 -0400 Subject: [PATCH 3/5] restructure query.live streaming as a generator pipeline --- .../src/runtime/server/remote-functions.js | 190 +++++++----------- packages/kit/src/runtime/server/utils.js | 34 ++++ packages/kit/src/runtime/server/utils.spec.js | 86 ++++++++ 3 files changed, 193 insertions(+), 117 deletions(-) create mode 100644 packages/kit/src/runtime/server/utils.spec.js diff --git a/packages/kit/src/runtime/server/remote-functions.js b/packages/kit/src/runtime/server/remote-functions.js index 74f380887f84..094c91426158 100644 --- a/packages/kit/src/runtime/server/remote-functions.js +++ b/packages/kit/src/runtime/server/remote-functions.js @@ -15,8 +15,7 @@ import { normalize_error } from '../../utils/error.js'; import { check_incorrect_fail_use, get_action_location } from './page/actions.js'; import { DEV } from 'esm-env'; import { deserialize_binary_form } from '../form-utils.js'; -import { text_encoder } from '../utils.js'; -import { with_version_header } from './utils.js'; +import { stream_from_iterator, with_version_header } from './utils.js'; /** * How long (in milliseconds) to wait after the last message was sent before @@ -25,6 +24,8 @@ import { with_version_header } from './utils.js'; */ const KEEP_ALIVE_INTERVAL = 30_000; +const KEEP_ALIVE = Symbol('keep-alive'); + /** @type {typeof handle_remote_call_internal} */ export async function handle_remote_call(event, state, options, manifest, id) { return record_span({ @@ -89,10 +90,19 @@ async function handle_remote_call_internal(event, state, options, manifest, id) new URL(event.request.url).searchParams.get('payload') ); - // aborted by `teardown()`, so unlike the request signal it also fires on - // response teardown, which the generator could otherwise never observe + // aborted whenever the stream is torn down, so unlike the request signal it + // also fires on response teardown, which the generator could otherwise + // never observe const cancellation = new AbortController(); + if (event.request.signal.aborted) { + cancellation.abort(); + } else { + event.request.signal.addEventListener('abort', () => cancellation.abort(), { + once: true + }); + } + const live_event = { ...event, request: new Request(event.request, { signal: cancellation.signal }) @@ -100,133 +110,79 @@ async function handle_remote_call_internal(event, state, options, manifest, id) const generator = internals.run(live_event, state, parse_remote_arg(payload)); - /** @type {ReturnType | undefined} */ - let keep_alive; - - // the controller is only reachable through this sink, which goes inert on - // close — either side can tear the stream down while `pull` is suspended - /** @type {ReturnType} */ - let sink; - - /** @param {ReadableStreamDefaultController} controller */ - function create_sink(controller) { - let open = true; - - return { - /** @param {string} data */ - write(data) { - if (!open) return; - controller.enqueue(text_encoder.encode(data)); - schedule_keep_alive(); - }, - close() { - if (!open) return; - open = false; - controller.close(); - }, - /** the platform already closed the controller (stream cancellation) */ - abandon() { - open = false; - } - }; - } + /** @param {any} payload */ + const frame = (payload) => 'data: ' + JSON.stringify(payload) + '\n\n'; /** - * (Re)schedule the keep-alive comment. Called whenever a message is sent, so - * that a keep-alive is only emitted once `KEEP_ALIVE_INTERVAL` has elapsed - * without any other activity. + * Resolves with the next iterator result, or with `KEEP_ALIVE` once + * `KEEP_ALIVE_INTERVAL` has elapsed without one. + * @param {Promise>} pending + * @returns {Promise | typeof KEEP_ALIVE>} */ - function schedule_keep_alive() { - clearTimeout(keep_alive); - keep_alive = setTimeout(() => { - // SSE comments (lines starting with `:`) are ignored by the client - sink.write(': keep-alive\n\n'); - }, KEEP_ALIVE_INTERVAL); - } - - /** @param {any} payload */ - function send(payload) { - sink.write('data: ' + JSON.stringify(payload) + '\n\n'); + function next_or_keep_alive(pending) { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => resolve(KEEP_ALIVE), KEEP_ALIVE_INTERVAL); + pending.then( + (result) => { + clearTimeout(timer); + resolve(result); + }, + (error) => { + clearTimeout(timer); + reject(error); + } + ); + }); } /** @type {string | undefined} */ let result = undefined; - let torn_down = false; + // everything the stream sends, as a generator of SSE strings — it holds no + // reference to the stream controller, so it cannot touch a dead one + async function* frames() { + /** @type {Promise> | null} */ + let pending = null; + + try { + while (true) { + pending ??= generator.next(); + const winner = await next_or_keep_alive(pending); + + if (winner === KEEP_ALIVE) { + // SSE comments (lines starting with `:`) are ignored by the client + yield ': keep-alive\n\n'; + continue; + } - async function teardown() { - if (torn_down) return; - torn_down = true; - clearTimeout(keep_alive); - cancellation.abort(); - sink.close(); - await generator.return(undefined); - } + pending = null; - event.request.signal.addEventListener('abort', teardown, { once: true }); + if (winner.done) return; - return new Response( - new ReadableStream({ - start(controller) { - sink = create_sink(controller); - schedule_keep_alive(); - }, - async pull() { - if (torn_down || event.request.signal.aborted) { - await teardown(); - return; + // only send changed data + if (result !== (result = stringify(winner.value))) { + yield frame({ type: 'result', result }); } - - try { - while (true) { - const { value, done } = await generator.next(); - - // teardown may have started while we were suspended - if (done || torn_down) { - await teardown(); - return; - } - - // only send changed data - if (result !== (result = stringify(value))) { - send({ - type: 'result', - result - }); - - return; - } - } - } catch (error) { - if (!torn_down) { - if (error instanceof Redirect) { - send({ - type: 'redirect', - location: error.location - }); - } else { - const transformed = await handle_error_and_jsonify( - event, - state, - options, - error - ); - - send({ - type: 'error', - error: transformed - }); - } - } - - await teardown(); + } + } catch (error) { + if (!cancellation.signal.aborted) { + if (error instanceof Redirect) { + yield frame({ type: 'redirect', location: error.location }); + } else { + yield frame({ + type: 'error', + error: await handle_error_and_jsonify(event, state, options, error) + }); } - }, - async cancel() { - sink.abandon(); - await teardown(); } - }), + } finally { + cancellation.abort(); + await generator.return(undefined); + } + } + + return new Response( + stream_from_iterator(frames(), () => cancellation.abort()), { headers: { 'cache-control': 'private, no-store', diff --git a/packages/kit/src/runtime/server/utils.js b/packages/kit/src/runtime/server/utils.js index 990decb06821..67105a351655 100644 --- a/packages/kit/src/runtime/server/utils.js +++ b/packages/kit/src/runtime/server/utils.js @@ -1,5 +1,39 @@ import { text } from '@sveltejs/kit'; import { ENDPOINT_METHODS } from '../../constants.js'; +import { text_encoder } from '../utils.js'; + +/** + * Builds a text stream from an iterator of string chunks. The controller is + * confined here so that a chunk arriving after the stream was torn down is + * dropped instead of hitting a closed controller — either side can tear the + * stream down while `pull` is suspended on `iterator.next()`. + * @param {AsyncIterator} iterator + * @param {() => void} [oncancel] called when the consumer cancels the stream + * @returns {ReadableStream} + */ +export function stream_from_iterator(iterator, oncancel) { + let open = true; + + return new ReadableStream({ + async pull(controller) { + const { value, done } = await iterator.next(); + + if (!open) return; + + if (done) { + open = false; + controller.close(); + } else { + controller.enqueue(text_encoder.encode(value)); + } + }, + async cancel() { + open = false; + oncancel?.(); + await iterator.return?.(undefined); + } + }); +} /** * @param {Partial>} mod diff --git a/packages/kit/src/runtime/server/utils.spec.js b/packages/kit/src/runtime/server/utils.spec.js new file mode 100644 index 000000000000..e8218a07e399 --- /dev/null +++ b/packages/kit/src/runtime/server/utils.spec.js @@ -0,0 +1,86 @@ +import { expect, test, vi } from 'vitest'; +import { stream_from_iterator } from './utils.js'; + +const decoder = new TextDecoder(); + +/** @param {string[]} chunks */ +async function* from(chunks) { + for (const chunk of chunks) { + await Promise.resolve(); + yield chunk; + } +} + +test('streams encoded chunks and closes when the iterator is done', async () => { + const reader = stream_from_iterator(from(['one', 'two'])).getReader(); + + const first = await reader.read(); + expect(decoder.decode(first.value)).toBe('one'); + + const second = await reader.read(); + expect(decoder.decode(second.value)).toBe('two'); + + await expect(reader.read()).resolves.toEqual({ value: undefined, done: true }); +}); + +test('cancellation notifies the producer and returns the iterator', async () => { + let finished = false; + const oncancel = vi.fn(); + + async function* source() { + try { + await Promise.resolve(); + yield 'one'; + yield 'two'; + } finally { + finished = true; + } + } + + const reader = stream_from_iterator(source(), oncancel).getReader(); + await reader.read(); + + await reader.cancel(); + + expect(oncancel).toHaveBeenCalledOnce(); + expect(finished).toBe(true); +}); + +// https://github.com/sveltejs/kit/issues/16778 +test('a value that arrives after cancellation is dropped', async () => { + /** @type {(result: IteratorResult) => void} */ + let resolve_next = () => {}; + let returned = false; + + /** @type {AsyncIterator} */ + const iterator = { + next: () => new Promise((resolve) => (resolve_next = resolve)), + return: () => { + returned = true; + return Promise.resolve({ value: undefined, done: true }); + } + }; + + const reader = stream_from_iterator(iterator).getReader(); + const read = reader.read(); // pull is now suspended on `iterator.next()` + await Promise.resolve(); + + await reader.cancel(); + expect(returned).toBe(true); + + // the parked value lands after teardown + resolve_next({ value: 'late', done: false }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + await expect(read).resolves.toEqual({ value: undefined, done: true }); +}); + +test('cancellation tolerates iterators without a return method', async () => { + /** @type {AsyncIterator} */ + const iterator = { next: () => new Promise(() => {}) }; + + const reader = stream_from_iterator(iterator).getReader(); + await Promise.resolve(); + + await expect(reader.cancel()).resolves.toBeUndefined(); +}); From 953e5423aa28ab75e7d280299a0de44f68a9d897 Mon Sep 17 00:00:00 2001 From: Nic Polumeyv <162764842+Nic-Polumeyv@users.noreply.github.com> Date: Thu, 13 Aug 2026 17:47:59 -0400 Subject: [PATCH 4/5] trim adapter tests to what actually pins behavior --- packages/kit/src/runtime/server/utils.spec.js | 18 ++++-------------- 1 file changed, 4 insertions(+), 14 deletions(-) diff --git a/packages/kit/src/runtime/server/utils.spec.js b/packages/kit/src/runtime/server/utils.spec.js index e8218a07e399..7f12a67fa81a 100644 --- a/packages/kit/src/runtime/server/utils.spec.js +++ b/packages/kit/src/runtime/server/utils.spec.js @@ -47,7 +47,7 @@ test('cancellation notifies the producer and returns the iterator', async () => }); // https://github.com/sveltejs/kit/issues/16778 -test('a value that arrives after cancellation is dropped', async () => { +test('cancellation settles while the producer is parked', async () => { /** @type {(result: IteratorResult) => void} */ let resolve_next = () => {}; let returned = false; @@ -65,22 +65,12 @@ test('a value that arrives after cancellation is dropped', async () => { const read = reader.read(); // pull is now suspended on `iterator.next()` await Promise.resolve(); + // neither cancel() nor the pending read may wait for the parked next() await reader.cancel(); expect(returned).toBe(true); + await expect(read).resolves.toEqual({ value: undefined, done: true }); - // the parked value lands after teardown + // the parked value landing afterwards is a no-op resolve_next({ value: 'late', done: false }); await new Promise((resolve) => setTimeout(resolve, 0)); - - await expect(read).resolves.toEqual({ value: undefined, done: true }); -}); - -test('cancellation tolerates iterators without a return method', async () => { - /** @type {AsyncIterator} */ - const iterator = { next: () => new Promise(() => {}) }; - - const reader = stream_from_iterator(iterator).getReader(); - await Promise.resolve(); - - await expect(reader.cancel()).resolves.toBeUndefined(); }); From e3b23e4a5546951b85bbd306b8345f13518646da Mon Sep 17 00:00:00 2001 From: Nic Polumeyv <162764842+Nic-Polumeyv@users.noreply.github.com> Date: Thu, 13 Aug 2026 18:06:38 -0400 Subject: [PATCH 5/5] split handle_remote_call_internal into resolve, dispatch and response helpers --- .../src/runtime/server/remote-functions.js | 286 +++++++++--------- 1 file changed, 149 insertions(+), 137 deletions(-) diff --git a/packages/kit/src/runtime/server/remote-functions.js b/packages/kit/src/runtime/server/remote-functions.js index 094c91426158..0b76701e2394 100644 --- a/packages/kit/src/runtime/server/remote-functions.js +++ b/packages/kit/src/runtime/server/remote-functions.js @@ -1,7 +1,7 @@ /** @import { RequestEvent, SSRManifest } from '@sveltejs/kit' */ /** @import { RemoteForm } from '$app/server' */ /** @import { ActionResult } from '$app/forms' */ -/** @import { RemoteFormInternals, RemoteFunctionData, RemoteFunctionResponse, RemoteInternals, RequestState, SSROptions } from 'types' */ +/** @import { RemoteFormInternals, RemoteFunctionData, RemoteFunctionResponse, RemoteInternals, RemoteQueryLiveInternals, RequestState, SSROptions } from 'types' */ import { json, error } from '@sveltejs/kit'; import { Redirect, SvelteKitError } from '@sveltejs/kit/internal'; @@ -44,13 +44,11 @@ export async function handle_remote_call(event, state, options, manifest, id) { } /** - * @param {RequestEvent} event - * @param {RequestState} state - * @param {SSROptions} options + * Looks a remote function up in the manifest by its request id. * @param {SSRManifest} manifest * @param {string} id */ -async function handle_remote_call_internal(event, state, options, manifest, id) { +async function resolve_remote_function(manifest, id) { const [hash, name, additional_args] = id.split('/'); const remotes = manifest._.remotes; @@ -61,136 +59,168 @@ async function handle_remote_call_internal(event, state, options, manifest, id) if (!fn) error(404); - /** @type {RemoteInternals} */ - const internals = fn.__; + return { fn, internals: /** @type {RemoteInternals} */ (fn.__), additional_args }; +} - event.tracing.current.setAttributes({ - 'sveltekit.remote.call.type': internals.type, - 'sveltekit.remote.call.name': internals.name - }); +/** + * @param {RemoteFunctionData} data + * @param {HeadersInit | undefined} headers + */ +function result_response(data, headers) { + return json( + /** @type {RemoteFunctionResponse} */ ({ + type: 'result', + data: stringify(data) + }), + { headers } + ); +} - /** @type {HeadersInit | undefined} */ - const headers = state.prerendering ? undefined : { 'cache-control': 'private, no-store' }; +/** + * Handles a `query.live` call: runs the generator and streams its values as + * server-sent events. + * @param {RequestEvent} event + * @param {RequestState} state + * @param {SSROptions} options + * @param {RemoteQueryLiveInternals} internals + */ +function handle_live_query(event, state, options, internals) { + if (event.request.method !== 'GET') { + throw new SvelteKitError( + 405, + 'Method Not Allowed', + `\`query.live\` functions must be invoked via GET request, not ${event.request.method}` + ); + } - try { - /** @type {RemoteFunctionData} */ - const data = {}; + const payload = /** @type {string} */ (new URL(event.request.url).searchParams.get('payload')); - switch (internals.type) { - case 'query_live': { - if (event.request.method !== 'GET') { - throw new SvelteKitError( - 405, - 'Method Not Allowed', - `\`query.live\` functions must be invoked via GET request, not ${event.request.method}` - ); - } + // aborted whenever the stream is torn down, so unlike the request signal it + // also fires on response teardown, which the generator could otherwise + // never observe + const cancellation = new AbortController(); - const payload = /** @type {string} */ ( - new URL(event.request.url).searchParams.get('payload') - ); + if (event.request.signal.aborted) { + cancellation.abort(); + } else { + event.request.signal.addEventListener('abort', () => cancellation.abort(), { + once: true + }); + } - // aborted whenever the stream is torn down, so unlike the request signal it - // also fires on response teardown, which the generator could otherwise - // never observe - const cancellation = new AbortController(); + const live_event = { + ...event, + request: new Request(event.request, { signal: cancellation.signal }) + }; - if (event.request.signal.aborted) { - cancellation.abort(); - } else { - event.request.signal.addEventListener('abort', () => cancellation.abort(), { - once: true - }); - } + const generator = internals.run(live_event, state, parse_remote_arg(payload)); - const live_event = { - ...event, - request: new Request(event.request, { signal: cancellation.signal }) - }; - - const generator = internals.run(live_event, state, parse_remote_arg(payload)); - - /** @param {any} payload */ - const frame = (payload) => 'data: ' + JSON.stringify(payload) + '\n\n'; - - /** - * Resolves with the next iterator result, or with `KEEP_ALIVE` once - * `KEEP_ALIVE_INTERVAL` has elapsed without one. - * @param {Promise>} pending - * @returns {Promise | typeof KEEP_ALIVE>} - */ - function next_or_keep_alive(pending) { - return new Promise((resolve, reject) => { - const timer = setTimeout(() => resolve(KEEP_ALIVE), KEEP_ALIVE_INTERVAL); - pending.then( - (result) => { - clearTimeout(timer); - resolve(result); - }, - (error) => { - clearTimeout(timer); - reject(error); - } - ); - }); + /** @param {any} payload */ + const frame = (payload) => 'data: ' + JSON.stringify(payload) + '\n\n'; + + /** + * Resolves with the next iterator result, or with `KEEP_ALIVE` once + * `KEEP_ALIVE_INTERVAL` has elapsed without one. + * @param {Promise>} pending + * @returns {Promise | typeof KEEP_ALIVE>} + */ + function next_or_keep_alive(pending) { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => resolve(KEEP_ALIVE), KEEP_ALIVE_INTERVAL); + pending.then( + (result) => { + clearTimeout(timer); + resolve(result); + }, + (error) => { + clearTimeout(timer); + reject(error); } + ); + }); + } - /** @type {string | undefined} */ - let result = undefined; + /** @type {string | undefined} */ + let result = undefined; - // everything the stream sends, as a generator of SSE strings — it holds no - // reference to the stream controller, so it cannot touch a dead one - async function* frames() { - /** @type {Promise> | null} */ - let pending = null; + // everything the stream sends, as a generator of SSE strings — it holds no + // reference to the stream controller, so it cannot touch a dead one + async function* frames() { + /** @type {Promise> | null} */ + let pending = null; - try { - while (true) { - pending ??= generator.next(); - const winner = await next_or_keep_alive(pending); + try { + while (true) { + pending ??= generator.next(); + const winner = await next_or_keep_alive(pending); - if (winner === KEEP_ALIVE) { - // SSE comments (lines starting with `:`) are ignored by the client - yield ': keep-alive\n\n'; - continue; - } + if (winner === KEEP_ALIVE) { + // SSE comments (lines starting with `:`) are ignored by the client + yield ': keep-alive\n\n'; + continue; + } - pending = null; + pending = null; - if (winner.done) return; + if (winner.done) return; - // only send changed data - if (result !== (result = stringify(winner.value))) { - yield frame({ type: 'result', result }); - } - } - } catch (error) { - if (!cancellation.signal.aborted) { - if (error instanceof Redirect) { - yield frame({ type: 'redirect', location: error.location }); - } else { - yield frame({ - type: 'error', - error: await handle_error_and_jsonify(event, state, options, error) - }); - } - } - } finally { - cancellation.abort(); - await generator.return(undefined); - } + // only send changed data + if (result !== (result = stringify(winner.value))) { + yield frame({ type: 'result', result }); + } + } + } catch (error) { + if (!cancellation.signal.aborted) { + if (error instanceof Redirect) { + yield frame({ type: 'redirect', location: error.location }); + } else { + yield frame({ + type: 'error', + error: await handle_error_and_jsonify(event, state, options, error) + }); } + } + } finally { + cancellation.abort(); + await generator.return(undefined); + } + } - return new Response( - stream_from_iterator(frames(), () => cancellation.abort()), - { - headers: { - 'cache-control': 'private, no-store', - 'content-type': 'text/event-stream' - } - } - ); + return new Response( + stream_from_iterator(frames(), () => cancellation.abort()), + { + headers: { + 'cache-control': 'private, no-store', + 'content-type': 'text/event-stream' } + } + ); +} +/** + * @param {RequestEvent} event + * @param {RequestState} state + * @param {SSROptions} options + * @param {SSRManifest} manifest + * @param {string} id + */ +async function handle_remote_call_internal(event, state, options, manifest, id) { + const { fn, internals, additional_args } = await resolve_remote_function(manifest, id); + + event.tracing.current.setAttributes({ + 'sveltekit.remote.call.type': internals.type, + 'sveltekit.remote.call.name': internals.name + }); + + /** @type {HeadersInit | undefined} */ + const headers = state.prerendering ? undefined : { 'cache-control': 'private, no-store' }; + + try { + /** @type {RemoteFunctionData} */ + const data = {}; + + switch (internals.type) { + case 'query_live': + return handle_live_query(event, state, options, internals); case 'query_batch': { if (event.request.method !== 'POST') { @@ -251,13 +281,7 @@ async function handle_remote_call_internal(event, state, options, manifest, id) if (data._.issues) { // special case — don't serialize refreshes/reconnects - return json( - /** @type {RemoteFunctionResponse} */ ({ - type: 'result', - data: stringify(data) - }), - { headers } - ); + return result_response(data, headers); } break; @@ -299,24 +323,12 @@ async function handle_remote_call_internal(event, state, options, manifest, id) await collect_remote_data(data, event, state, options); - return json( - /** @type {RemoteFunctionResponse} */ ({ - type: 'result', - data: stringify(data) - }), - { headers } - ); + return result_response(data, headers); } catch (error) { if (error instanceof Redirect) { const data = await collect_remote_data({ redirect: error.location }, event, state, options); - return json( - /** @type {RemoteFunctionResponse} */ ({ - type: 'result', - data: stringify(data) - }), - { headers } - ); + return result_response(data, headers); } const transformed = await handle_error_and_jsonify(event, state, options, error);