fix: Live query reconnect issues

.changeset/fix-live-query-fanout-orphan.md

@@ -0,0 +1,5 @@
+---
+'@sveltejs/kit': patch
+---
+
+fix: preserve active `for await` consumers across `query.live` reconnects

.changeset/fix-live-query-handshake-deadlock.md

@@ -0,0 +1,5 @@
+---
+'@sveltejs/kit': patch
+---
+
+fix: settle `query.live` reconnect promise on all exit paths, preventing `invalidateAll()` from deadlocking when a live query is offline or interrupted

.changeset/fix-live-query-undefined-clobber.md

@@ -0,0 +1,5 @@
+---
+'@sveltejs/kit': patch
+---
+
+fix: preserve last value when a `query.live` stream completes without yielding on reconnect

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

@@ -124,6 +124,7 @@ export class LiveQuery {
 
 		/** @type {PromiseWithResolvers<void>} */
 		const { promise: stopped, resolve: on_stop } = with_resolvers();
+		let connected = false;
 
 		while (!this.#done) {
 			const controller = new AbortController();
@@ -136,16 +137,22 @@ export class LiveQuery {
 			const generator = create_live_iterator(this.#id, this.#payload, controller, () => {
 				this.#connected = true;
 				this.#attempt = 0;
+				connected = true;
 				on_connect();
 			});
 
 			try {
 				const { done, value } = await generator.next();
 
-				// TODO how much special handling does this need?
-				// should we even try to reconnect if this is the case?
-				if (done && !this.#ready) {
-					throw new Error('Live query completed before yielding a value');
+				if (done) {
+					if (!this.#ready) {
+						throw new Error('Live query completed before yielding a value');
+					}
+					// stream completed without yielding (e.g. the generator returned
+					// immediately on reconnect) — keep the last good value
+					this.#done = true;
+					this.#fan_out.done();
+					break;
 				}
 
 				this.set(value);
@@ -202,6 +209,12 @@ export class LiveQuery {
 		}
 
 		this.#interrupt = null;
+		// If the loop exited without ever successfully connecting, settle the
+		// reconnect handshake so callers (e.g. `invalidateAll()`) never await
+		// a forever-pending promise.
+		if (!connected) {
+			on_connect_failed(this.#error ?? new Error('Live query connection was interrupted'));
+		}
 		on_stop();
 	}
 
@@ -345,10 +358,13 @@ 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();
+		// Keep the existing fan-out open so active `for await` consumers
+		// continue receiving values from the new connection without interruption.
+		// Only replace it if it was already closed by a prior `done()`/`fail()`
+		// (e.g. reconnecting after a finite or hard-failed stream)
+		if (this.#fan_out.closed) {
+			this.#fan_out = new SharedIterator();
+		}
 		this.#main({ on_connect, on_connect_failed }).catch(noop);
 		await promise;
 	}

packages/kit/src/utils/shared-iterator.js

@@ -52,6 +52,11 @@ export class SharedIterator {
 	/** @type {unknown} */
 	#terminal_error = undefined;
 
+	/** Whether `done()` or `fail()` has been broadcast. */
+	get closed() {
+		return this.#closed;
+	}
+
 	/**
 	 * @param {(instance: SharedIterator<T>) => (() => void)} [start]
 	 */

packages/kit/test/apps/async/src/routes/remote/live/+page.svelte

@@ -1,4 +1,5 @@
 <script lang="ts">
+	import { invalidateAll } from '$app/navigation';
 	import LiveView from './LiveView.svelte';
 	import {
 		increment,
@@ -44,6 +45,39 @@
 			for_await_values = `error: ${(error as Error).message}`;
 		}
 	}
+
+	let stream_log = $state('');
+	let stream_iteration: AbortController | null = null;
+
+	async function start_stream_log() {
+		stream_iteration?.abort();
+		stream_iteration = new AbortController();
+		const signal = stream_iteration.signal;
+		stream_log = '';
+		const collected: number[] = [];
+
+		try {
+			for await (const value of get_count()) {
+				if (signal.aborted) break;
+				collected.push(value);
+				stream_log = collected.join(',');
+			}
+		} catch (error) {
+			stream_log = `error: ${(error as Error).message}`;
+		}
+	}
+
+	let invalidate_state = $state('idle');
+
+	async function run_invalidate_all() {
+		invalidate_state = 'pending';
+		try {
+			await invalidateAll();
+			invalidate_state = 'resolved';
+		} catch {
+			invalidate_state = 'rejected';
+		}
+	}
 </script>
 
 <button id="increment" onclick={() => increment()}>increment</button>
@@ -74,3 +108,8 @@
 <button id="start-for-await" onclick={start_for_await}>start for-await</button>
 <p id="for-await-count">{for_await_count}</p>
 <p id="for-await-values">{for_await_values}</p>
+
+<button id="start-stream-log" onclick={start_stream_log}>start stream log</button>
+<p id="stream-log">{stream_log}</p>
+<button id="run-invalidate-all" onclick={run_invalidate_all}>invalidate all</button>
+<p id="invalidate-state">{invalidate_state}</p>

packages/kit/test/apps/async/test/client.test.js

@@ -773,6 +773,47 @@ test.describe('remote function mutations', () => {
 		await page.click('#reset');
 	});
 
+	test('for await consumers continue receiving values across invalidateAll-triggered reconnects', async ({
+		page
+	}) => {
+		await page.goto('/remote/live');
+		await page.click('#reset');
+		await expect(page.locator('#count')).toHaveText('0');
+
+		await page.click('#start-stream-log');
+		// the first value should be the current value (0)
+		await expect(page.locator('#stream-log')).toHaveText(/0/);
+
+		// invalidateAll() calls reconnect(), which keeps the existing fan-out
+		// open so active `for await` consumers continue receiving values from
+		// the new connection without interruption.
+		await page.click('#run-invalidate-all');
+		await expect(page.locator('#invalidate-state')).toHaveText('resolved');
+
+		// Trigger a new value — the still-attached consumer must see it.
+		// Before the fix the fan-out was replaced, orphaning the subscriber so
+		// this value never arrived and the loop hung.
+		await page.click('#increment');
+		await expect(page.locator('#stream-log')).toContainText('1');
+
+		await page.click('#reset');
+	});
+
+	test('invalidateAll resolves while a live query is offline', async ({ page, context }) => {
+		await page.goto('/remote/live');
+		await page.click('#reset');
+		await expect(page.locator('#connected')).toHaveText('true');
+
+		await context.setOffline(true);
+
+		// reconnect()'s handshake must settle on every #main exit path (here:
+		// offline) so that awaiting invalidateAll() doesn't deadlock.
+		await page.click('#run-invalidate-all');
+		await expect(page.locator('#invalidate-state')).toHaveText(/resolved|rejected/);
+
+		await context.setOffline(false);
+	});
+
 	test('query.live cleans up server iterator on reload', async ({ page }) => {
 		await page.goto('/remote/live');
 		await page.click('#stats');