fix: hand off older query graphs to live subscriptions

docs/live-query-delivery.md

@@ -23,6 +23,13 @@ from disposed subscriptions and previous authorization generations. Local cache
 membership revisions are not server projection positions and cannot confirm
 optimistic commands.
 
+A subscription also records its local start order. Snapshot delivery can take
+over shared relationships owned by queries that preceded that subscription,
+including SSR seeds from a layout or another island. This lets an initially
+empty result acquire rows without waiting for another page load. It does not
+override an independently active live stream or query ownership acquired after
+the subscription started; those results have no safe cross-stream ordering.
+
 This is a breaking v5 protocol change: `mode` replaces the ambiguous `supported`
 boolean. Upgrade server and generated-client runtime together. Old or unknown
 wire forms fail closed; applications do not need a polling or reload workaround.

js/src/replica/distributed-replica/impl-fetch-live.ts

@@ -238,7 +238,8 @@ export function retainLive<TData, TVariables extends GraphqlVariables>(
 		count: 1,
 		unsubscribe: () => undefined,
 		active: true,
-		protocolGeneration: host.protocolGenerationSequence()
+		protocolGeneration: host.protocolGenerationSequence(),
+		startRevision: host.allocateIndexRevision()
 	};
 	host.lives.set(watch.key, entry);
 	const resume = host.resumeCursors(watch.key);

js/src/replica/distributed-replica/impl.ts

@@ -2320,6 +2320,9 @@ export class DistributedReplicaImpl implements DistributedReplicaApi {
 		if (incomingIndexKeys.size === 0) return { compared: false };
 		const confirmedRevisions =
 			this.#confirmedIndexFences(incomingIndexKeys);
+		const liveStart = source === 'live' && !snapshot.indexesComparable
+			? this.#lives.get(currentKey)?.startRevision
+			: undefined;
 		let compared = false;
 		let lower = false;
 		let higher = false;
@@ -2370,6 +2373,13 @@ export class DistributedReplicaImpl implements DistributedReplicaApi {
 					disposition === 'fresh' ||
 					disposition === 'incomparable'
 				) {
+					// A registered stream starts after the query seeds already in
+					// this replica. It can take over those seeds, but not another
+					// live stream or query ownership acquired after it started.
+					if (
+						liveStart !== undefined && state === group.query &&
+						compareCanonicalDecimalStrings(liveStart, state.indexRevision) > 0
+					) continue;
 					incomparable = true;
 					continue;
 				}

js/src/replica/distributed-replica/types.ts

@@ -30,6 +30,8 @@ export type LiveEntry = {
 	unsubscribe: () => void;
 	active: boolean;
 	protocolGeneration: number;
+	/** Local fence for taking over query snapshots that preceded this stream. */
+	startRevision: string;
 	operationGeneration?: number;
 };
 

js/tests/replica-protocol.test.mjs

@@ -1975,6 +1975,86 @@ test('snapshot live replaces SSR membership, updates and removes rows, and fence
 	assert.deepEqual(replica.read(Todos, {}).data.todos, []);
 });
 
+test('snapshot live adds a row with an unchanged SSR-owned nested relationship', () => {
+	let observer;
+	const replica = createDistributedReplica({ transport: {
+		fetch() { throw new Error('complete snapshot must not force HTTP fallback'); },
+		subscribe(_request, next) { observer = next; return () => {}; }
+	} });
+	replica.writeResult(FeaturedGamesWithOwner, {}, gamesFrame({
+		artifact: FeaturedGamesWithOwner, responseKey: 'featuredGames',
+		position: '1', ownerId: 'user-1', ownerName: 'Owner', indexesComparable: false
+	}), 'ssr');
+	const empty = gamesFrame({
+		artifact: GamesWithOwnerLiveOperation, responseKey: 'games',
+		position: '1', ownerId: 'user-1', ownerName: 'Owner', indexesComparable: false
+	});
+	empty.data.games = [];
+	empty.extensions.distributed.snapshot.records = [];
+	replica.writeResult(GamesWithOwnerLiveOperation, {}, empty, 'ssr');
+	const watch = replica.watch(GamesWithOwnerLiveOperation, {}, { live: true });
+	observer.next(gamesFrame({
+		artifact: GamesWithOwnerLiveOperation, responseKey: 'games',
+		operation: GamesWithOwnerLiveOperation.live.id,
+		position: '2', ownerId: 'user-1', ownerName: 'Owner', indexesComparable: false,
+		live: { mode: 'snapshot', reset: true, cursors: [] }
+	}));
+	assert.deepEqual(watch.get().errors, []);
+	assert.equal(watch.get().complete, true);
+	assert.equal(watch.get().data.games.length, 1, 'unchanged sibling relationship must not freeze an empty root');
+	assert.equal(watch.get().data.games[0].owner.name, 'Owner');
+	assert.equal(replica.read(FeaturedGamesWithOwner, {}).data.featuredGames[0].owner.name, 'Owner');
+	observer.next(gamesFrame({
+		artifact: GamesWithOwnerLiveOperation, responseKey: 'games',
+		operation: GamesWithOwnerLiveOperation.live.id,
+		position: '3', ownerId: 'user-2', ownerName: 'Changed', indexesComparable: false,
+		live: { mode: 'snapshot', reset: true, cursors: [] }
+	}));
+	assert.equal(watch.get().data.games[0].owner.name, 'Changed');
+	const removed = gamesFrame({
+		artifact: GamesWithOwnerLiveOperation, responseKey: 'games',
+		operation: GamesWithOwnerLiveOperation.live.id,
+		position: '4', ownerId: 'user-2', ownerName: 'Changed', indexesComparable: false,
+		live: { mode: 'snapshot', reset: true, cursors: [] }
+	});
+	removed.data.games = [];
+	removed.extensions.distributed.snapshot.records = [];
+	observer.next(removed);
+	assert.deepEqual(watch.get().data.games, []);
+	assert.equal(watch.get().live, 'active');
+	watch.destroy();
+});
+
+test('snapshot live cannot take over query ownership acquired after stream start', () => {
+	let observer;
+	const replica = createDistributedReplica({ transport: {
+		fetch() { throw new Error('complete snapshot must not force HTTP fallback'); },
+		subscribe(_request, next) { observer = next; return () => {}; }
+	} });
+	const empty = gamesFrame({
+		artifact: GamesWithOwnerLiveOperation, responseKey: 'games',
+		position: '1', ownerId: 'user-1', ownerName: 'Owner', indexesComparable: false
+	});
+	empty.data.games = [];
+	empty.extensions.distributed.snapshot.records = [];
+	replica.writeResult(GamesWithOwnerLiveOperation, {}, empty, 'ssr');
+	const watch = replica.watch(GamesWithOwnerLiveOperation, {}, { live: true });
+	replica.writeResult(FeaturedGamesWithOwner, {}, gamesFrame({
+		artifact: FeaturedGamesWithOwner, responseKey: 'featuredGames',
+		position: '2', ownerId: 'user-2', ownerName: 'New query', indexesComparable: false
+	}), 'network');
+	observer.next(gamesFrame({
+		artifact: GamesWithOwnerLiveOperation, responseKey: 'games',
+		operation: GamesWithOwnerLiveOperation.live.id,
+		position: '1', ownerId: 'user-1', ownerName: 'Old stream', indexesComparable: false,
+		live: { mode: 'snapshot', reset: true, cursors: [] }
+	}));
+	assert.deepEqual(watch.get().errors, []);
+	assert.deepEqual(watch.get().data.games, []);
+	assert.equal(replica.read(FeaturedGamesWithOwner, {}).data.featuredGames[0].owner.name, 'New query');
+	watch.destroy();
+});
+
 test('snapshot authorization changes discard the old stream and reconnect without cursors', async () => {
 	const subscriptions = [];
 	const requests = [];

tests/e2e-ui/gateway/refresh.mjs

@@ -56,7 +56,11 @@ export async function verifySessionRefreshContinuity(page, origin) {
         await page.evaluate(()=>new Promise(resolve=>requestAnimationFrame(()=>requestAnimationFrame(resolve))));
       }
       assert.deepEqual(navigations,[],'session refresh must not navigate the document');
-      assert.equal(await page.evaluate(()=>globalThis.__refreshContinuity.lost),false,route+' rows were removed during token refresh: '+JSON.stringify(await page.evaluate(()=>globalThis.__refreshContinuity.events)));
+      const lost = await page.evaluate(()=>globalThis.__refreshContinuity.lost);
+      if (lost) {
+        console.error('Redacted refresh replica diagnostics:', JSON.stringify(await page.evaluate(()=>globalThis.__replicaDiagnosticSnapshot?.())));
+      }
+      assert.equal(lost,false,route+' rows were removed during token refresh: '+JSON.stringify(await page.evaluate(()=>globalThis.__refreshContinuity.events)));
     } finally {
       releaseRefresh();
       await page.unroute('**/api/auth/refresh',refreshRoute);

tests/e2e-ui/gateway/run.mjs

@@ -55,7 +55,9 @@ try{
       const ack=await fetch(publicOrigin+'/__distributed/lifecycle',{method:'POST',headers:{origin:publicOrigin,'content-type':'application/json'},body:JSON.stringify({participantId:participant,transitionId:'gateway_ci_no_transition',ok:true})});
       assert.equal(ack.status,409,'same-origin acknowledgement reaches lifecycle state validation');
     }
-    browser=await chromium.launch();const context=await browser.newContext();const page=await context.newPage();
+    browser=await chromium.launch();const context=await browser.newContext();
+    await context.addInitScript(()=>{globalThis.__captureReplicaDiagnostics=true;});
+    const page=await context.newPage();
     const errors=[];page.on('pageerror',error=>errors.push(error.message));
     await page.goto(publicOrigin);await page.getByRole('link',{name:/log in|sign in/i}).first().click();
     await page.getByRole('button',{name:'Continue as Alice'}).click();

tests/e2e-ui/ui/src/routes/+layout.svelte

@@ -2,6 +2,7 @@
 	import '../app.css';
 	import '$lib/styles/chrome.css';
 	import { browser } from '$app/environment';
+	import { createReplicaDiagnostics } from '@hops-ops/distributed/replica';
 	import { page } from '$app/state';
 	import { onDestroy, untrack } from 'svelte';
 	import type { Snippet } from 'svelte';
@@ -35,7 +36,13 @@
 		diagnostics.__distributedReloadState = lifecycleDemoState;
 	}
 
+	const replicaDiagnostics = browser && (globalThis as typeof globalThis & Record<string, unknown>).__captureReplicaDiagnostics === true
+		? createReplicaDiagnostics() : undefined;
+	if (replicaDiagnostics) {
+		(globalThis as typeof globalThis & Record<string, unknown>).__replicaDiagnosticSnapshot = () => replicaDiagnostics.snapshot();
+	}
 	const client = provideDistributed({
+		...(replicaDiagnostics ? { replica: { diagnostics: replicaDiagnostics } } : {}),
 		boundaries: DISTRIBUTED_BOUNDARY_OPERATIONS,
 		session: pageData.session,
 		browser,