chore: reduce indirection

.changeset/feat-live-query-async-iterable.md

@@ -0,0 +1,5 @@
+---
+'@sveltejs/kit': minor
+---
+
+feat: live query instances are now themselves async-iterable

documentation/docs/20-core-concepts/60-remote-functions.md

@@ -288,7 +288,23 @@ If the connection drops, `connected` becomes `false`. SvelteKit will attempt to
 
 Unlike `query`, live queries do not have a `refresh()` method, as they are self-updating.
 
-If you need direct, imperative access to the underlying stream of values (rather than the reactive `current` property), live queries expose a `.run()` method that returns an `AsyncGenerator<T>`. This can only be called outside render — in event handlers, `load` functions, and so on.
+If you need direct, imperative access to the underlying stream of values (rather than the reactive `current` property), live query instances are themselves [async-iterable](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/for-await...of). You can `for await` over the instance directly:
+
+```js
+// @errors: 7006
+import { getTime } from './time.remote.js';
+// ---cut---
+async function logTimes() {
+	for await (const value of getTime()) {
+		console.log(value);
+		if (someCondition) break;
+	}
+}
+```
+
+Multiple consumers of the same live query (whether reactive — via `await` or `current` — or imperative `for await` loops) share a single underlying connection. The first value yielded to a `for await` iterator is the most-recently-received value, if one is already available, mirroring the semantics of awaiting the resource directly. Subsequent yields fire whenever a new value arrives from the server. If values arrive faster than the consumer drains the iterator, only the latest pending value is kept — live streams are not event logs.
+
+On the server, `for await` likewise joins a per-request shared iteration of the underlying generator, so concurrent consumers within the same request don't run the user-defined generator multiple times.
 
 > [!NOTE] It's essential that you don't cache live query responses in a service worker, since the cloned response will continue streaming long after the page is closed. Make sure that your caching logic excludes any responses with a `Cache-Control` header that includes `no-store`.
 

packages/kit/src/exports/public.d.ts

@@ -2268,20 +2268,15 @@ export type RemoteQuery<T> = RemoteResource<T> & {
 	withOverride(update: (current: T) => T): RemoteQueryOverride;
 };
 
-export type RemoteLiveQuery<T> = RemoteResource<T> & {
-	/**
-	 * Returns an async iterator with live updates.
-	 * Unlike awaiting the resource directly, this can only be used _outside_ render
-	 * (i.e. in load functions, event handlers and so on)
-	 */
-	run(): AsyncGenerator<T>;
-	/** `true` if the live stream is currently connected. */
-	readonly connected: boolean;
-	/** `true` once the current live stream iterator is done. */
-	readonly done: boolean;
-	/** Reconnects the live stream immediately. */
-	reconnect(): Promise<void>;
-};
+export type RemoteLiveQuery<T> = RemoteResource<T> &
+	AsyncIterable<T> & {
+		/** `true` if the live stream is currently connected. */
+		readonly connected: boolean;
+		/** `true` once the current live stream iterator is done. */
+		readonly done: boolean;
+		/** Reconnects the live stream immediately. */
+		reconnect(): Promise<void>;
+	};
 
 export type RemoteQueryOverride = () => void;
 

packages/kit/src/runtime/app/server/remote/query.js

@@ -1,5 +1,5 @@
 /** @import { RemoteLiveQuery, RemoteLiveQueryFunction, RemoteQuery, RemoteQueryFunction } from '@sveltejs/kit' */
-/** @import { RemoteInternals, MaybePromise, RequestState, RemoteQueryLiveInternals, RemoteQueryBatchInternals, RemoteQueryInternals, RemoteLiveQueryUserFunctionReturnType } from 'types' */
+/** @import { RemoteInternals, MaybePromise, RequestState, RemoteQueryLiveInternals, RemoteQueryBatchInternals, RemoteQueryInternals, RemoteLiveQueryUserFunctionReturnType, SharedServerLiveIterator } from 'types' */
 /** @import { StandardSchemaV1 } from '@standard-schema/spec' */
 import { get_request_store } from '@sveltejs/kit/internal/server';
 import { create_remote_key, stringify, stringify_remote_arg } from '../../../shared.js';
@@ -14,6 +14,7 @@ import {
 import { handle_error_and_jsonify } from '../../../server/utils.js';
 import { HttpError, SvelteKitError } from '@sveltejs/kit/internal';
 import { noop } from '../../../../utils/functions.js';
+import { SharedIterator } from '../../../../utils/shared-iterator.js';
 
 /**
  * Creates a remote query. When called from the browser, the function will be invoked on the server via a `fetch` call.
@@ -153,24 +154,6 @@ function live(validate_or_fn, maybe_fn) {
 	const run = (event, state, get_input) =>
 		run_remote_generator(event, state, false, get_input, fn, __.name);
 
-	/**
-	 * @param {any} generator
-	 * @returns {Promise<any>}
-	 */
-	const first_value = async (generator) => {
-		try {
-			const { value, done } = await generator.next();
-
-			if (done) {
-				throw new Error(`query.live '${__.name}' did not yield a value`);
-			}
-
-			return value;
-		} finally {
-			await generator.return(undefined);
-		}
-	};
-
 	/** @type {RemoteQueryLiveInternals} */
 	const __ = {
 		type: 'query_live',
@@ -181,8 +164,8 @@ function live(validate_or_fn, maybe_fn) {
 		bind(payload, validated_arg) {
 			const { event, state } = get_request_store();
 
-			return create_live_query_resource(__, payload, state, () =>
-				first_value(run(event, state, () => validated_arg))
+			return create_live_query_resource(__, payload, state, event.request.signal, () =>
+				run(event, state, () => validated_arg)
 			);
 		}
 	};
@@ -198,8 +181,8 @@ function live(validate_or_fn, maybe_fn) {
 		const { event, state } = get_request_store();
 		const payload = stringify_remote_arg(arg, state.transport);
 
-		return create_live_query_resource(__, payload, state, () =>
-			first_value(run(event, state, () => validate(arg)))
+		return create_live_query_resource(__, payload, state, event.request.signal, () =>
+			run(event, state, () => validate(arg))
 		);
 	};
 
@@ -476,13 +459,21 @@ function create_query_resource(__, payload, state, fn) {
  * @param {RemoteQueryLiveInternals} __
  * @param {string} payload — the stringified raw argument (i.e. the cache key the client will use)
  * @param {RequestState} state
- * @param {() => Promise<any>} get_first_value
+ * @param {AbortSignal} signal — the request signal; aborts in-flight iteration when the client disconnects
+ * @param {() => AsyncGenerator<any, void, void>} get_generator
  * @returns {RemoteLiveQuery<any>}
  */
-function create_live_query_resource(__, payload, state, get_first_value) {
+function create_live_query_resource(__, payload, state, signal, get_generator) {
 	/** @type {Promise<any> | null} */
 	let promise = null;
 
+	const get_first_value = async () => {
+		for await (const value of get_generator()) {
+			return value;
+		}
+		throw new Error(`query.live '${__.name}' did not yield a value`);
+	};
+
 	const get_promise = () => {
 		return (promise ??= get_response(__, payload, state, get_first_value));
 	};
@@ -536,19 +527,117 @@ function create_live_query_resource(__, payload, state, get_first_value) {
 			reconnects.set(create_remote_key(__.id, payload), get_promise());
 			return Promise.resolve();
 		},
+		/** @ts-expect-error This method no longer exists */
 		run() {
-			throw new Error('Cannot call .run() on a live query on the server');
+			throw new Error(
+				'`.run()` has been removed from live queries. Use `for await (const value of liveQuery())` instead.'
+			);
 		},
 		/** @type {Promise<any>['then']} */
 		then(onfulfilled, onrejected) {
 			return get_promise().then(onfulfilled, onrejected);
 		},
+		[Symbol.asyncIterator]() {
+			const key = create_remote_key(__.id, payload);
+
+			const cache = (state.remote.live_iterators ??= {});
+			const cached = (cache[key] ??= create_shared_live_iterator(signal, get_generator));
+
+			return cached.subscribe();
+		},
 		get [Symbol.toStringTag]() {
 			return 'LiveQueryResource';
 		}
 	};
 }
 
+/**
+ * Wraps a lazily-created live-query generator so that multiple `for await`
+ * consumers within the same request share one underlying iteration. The first
+ * subscriber starts the generator; values are broadcast to all subscribers
+ * via a `SharedIterator`. When the last subscriber unsubscribes, the generator
+ * is closed via `generator.return(undefined)`.
+ *
+ * If `signal` aborts (typically because the client has disconnected), the
+ * pump is torn down and any in-flight `next()` calls on consumer iterators
+ * resolve with `{ done: true }`, so suspended `for await` loops unwind
+ * cleanly rather than leaking.
+ *
+ * @param {AbortSignal} signal
+ * @param {() => AsyncGenerator<any, void, void>} get_generator
+ * @returns {SharedServerLiveIterator}
+ */
+function create_shared_live_iterator(signal, get_generator) {
+	/** @type {AsyncGenerator<any, void, void> | null} */
+	let generator = null;
+
+	// Set to `true` when we deliberately close the generator (because every
+	// subscriber has unsubscribed, or the request was aborted). The pump's
+	// `g.next()` will reject as a result; we use this flag to swallow that
+	// abort error rather than surfacing it through `fan_out.fail()`.
+	let aborted = false;
+
+	const close_generator = () => {
+		void generator?.return().catch(noop);
+		generator = null;
+		aborted = true;
+	};
+
+	const fan_out = new SharedIterator({
+		on_first_subscribe: () => {
+			// Don't bother starting the pump if the request has already been
+			// aborted between cache creation and first subscription.
+			if (signal.aborted) {
+				fan_out.done();
+				return;
+			}
+
+			generator = get_generator();
+			const g = generator;
+
+			void (async () => {
+				try {
+					while (true) {
+						const result = await g.next();
+						if (result.done) {
+							fan_out.done();
+							return;
+						}
+						fan_out.push(result.value);
+					}
+				} catch (error) {
+					if (!aborted) fan_out.fail(error);
+				} finally {
+					close_generator();
+				}
+			})();
+		},
+		on_last_unsubscribe: () => {
+			close_generator();
+		}
+	});
+
+	// On request abort, tear down the pump and notify subscribers. `done()` is
+	// used (rather than `fail()`) because an aborted request is a normal
+	// termination — there's no error to surface to user code that's already
+	// been disconnected from the client.
+	const on_abort = () => {
+		close_generator();
+		fan_out.done();
+	};
+	if (signal.aborted) {
+		on_abort();
+	} else {
+		signal.addEventListener('abort', on_abort, { once: true });
+	}
+
+	return {
+		subscribe() {
+			return fan_out.subscribe();
+		}
+	};
+}
+
 // Add batch as a property to the query function
 Object.defineProperty(query, 'batch', { value: batch, enumerable: true });
 Object.defineProperty(query, 'live', { value: live, enumerable: true });

packages/kit/src/runtime/client/remote-functions/cache.svelte.js

@@ -1,4 +1,5 @@
 import { tick } from 'svelte';
+import { once } from '../../../utils/functions.js';
 
 /**
  * @template R
@@ -114,7 +115,7 @@ export class CacheController {
 	 */
 	manual_ref = (entry, id, payload) => {
 		entry.proxy_count++;
-		return () => this.deref(entry, id, payload);
+		return once(() => this.deref(entry, id, payload));
 	};
 
 	/**

packages/kit/src/runtime/client/remote-functions/query-live/instance.svelte.js

@@ -4,13 +4,15 @@ import * as devalue from 'devalue';
 import { HttpError, Redirect } from '@sveltejs/kit/internal';
 import { noop, once } from '../../../../utils/functions.js';
 import { with_resolvers } from '../../../../utils/promise.js';
+import { SharedIterator } from '../../../../utils/shared-iterator.js';
 import { tick } from 'svelte';
 import { unfriendly_hydratable } from '../../../shared.js';
 import { create_live_iterator } from './iterator.js';
 
 /**
  * @template T
  * @implements {Promise<T>}
+ * @implements {AsyncIterable<T>}
  */
 export class LiveQuery {
 	#id;
@@ -43,6 +45,15 @@ export class LiveQuery {
 	#interrupt = null;
 	#attempt = 0;
 
+	/**
+	 * Fan-out for `for await` consumers attached to this LiveQuery's shared
+	 * stream. New subscribers see the most-recently-emitted value (if any) as
+	 * their first yield. Subsequent yields fire whenever `set()` is called.
+	 *
+	 * @type {SharedIterator<T>}
+	 */
+	#fan_out = new SharedIterator();
+
 	/** @type {Promise<T>['then']} */
 	// @ts-expect-error TS doesn't understand that the promise returns something
 	#then = $derived.by(() => {
@@ -130,6 +141,7 @@ export class LiveQuery {
 				}
 
 				this.#done = true;
+				this.#fan_out.done();
 			} catch (error) {
 				if (controller.signal.aborted) break;
 
@@ -220,9 +232,38 @@ export class LiveQuery {
 			window.removeEventListener('pageshow', this.#on_pageshow);
 		}
 
+		this.#fan_out.done();
+
 		void this.#interrupt?.();
 	}
 
+	/**
+	 * Iterate the stream of values yielded by this live query. Multiple
+	 * iterators share the underlying connection; the most-recently emitted value
+	 * (if any) is yielded first to each new iterator, mirroring the semantics
+	 * of awaiting the query directly.
+	 *
+	 * Backpressure note: if values arrive faster than the consumer drains, only
+	 * the latest pending value is kept. This matches the reactive `.current`
+	 * semantics — live streams are not event logs.
+	 *
+	 * @returns {AsyncGenerator<T, void, void>}
+	 */
+	[Symbol.asyncIterator]() {
+		this.#start();
+
+		// Seed the new iterator with the current value (if any) so the first
+		// `.next()` resolves synchronously — mirroring `await liveQuery()`
+		// semantics. If the query has hard-failed, `#fan_out` is already closed
+		// with the terminal error and `subscribe()` returns an iterator whose
+		// first `.next()` rejects.
+		return this.#fan_out.subscribe(
+			this.#ready && this.#error === undefined
+				? { initial_value: { value: /** @type {T} */ (this.#raw) } }
+				: undefined
+		);
+	}
+
 	get then() {
 		this.#start();
 		return this.#then;
@@ -290,6 +331,10 @@ export class LiveQuery {
 		promise.catch(noop);
 		this.#done = false;
 		this.#attempt = 0;
+		// The previous fan-out may have been closed by `done()`/`fail()`. Future
+		// `for await` consumers need a fresh, open fan-out attached to the new
+		// `#main` lifetime.
+		this.#fan_out = new SharedIterator();
 		this.#main({ on_connect, on_connect_failed }).catch(noop);
 		await promise;
 	}
@@ -308,6 +353,8 @@ export class LiveQuery {
 		} else {
 			this.#promise = Promise.resolve();
 		}
+
+		this.#fan_out.push(value);
 	}
 
 	/** @param {unknown} error */
@@ -330,6 +377,8 @@ export class LiveQuery {
 			promise.catch(noop);
 			this.#promise = promise;
 		}
+
+		this.#fan_out.fail(error);
 	}
 
 	get [Symbol.toStringTag]() {