diff --git a/docs/live-query-delivery.md b/docs/live-query-delivery.md index 970b4a4c..ce58e823 100644 --- a/docs/live-query-delivery.md +++ b/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. diff --git a/js/src/replica/distributed-replica/impl-fetch-live.ts b/js/src/replica/distributed-replica/impl-fetch-live.ts index 72f55dba..0882d5fb 100644 --- a/js/src/replica/distributed-replica/impl-fetch-live.ts +++ b/js/src/replica/distributed-replica/impl-fetch-live.ts @@ -238,7 +238,8 @@ export function retainLive( 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); diff --git a/js/src/replica/distributed-replica/impl.ts b/js/src/replica/distributed-replica/impl.ts index e3eea081..4d8fbb33 100644 --- a/js/src/replica/distributed-replica/impl.ts +++ b/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; } diff --git a/js/src/replica/distributed-replica/types.ts b/js/src/replica/distributed-replica/types.ts index 8dee8016..5d8761e7 100644 --- a/js/src/replica/distributed-replica/types.ts +++ b/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; }; diff --git a/js/tests/replica-protocol.test.mjs b/js/tests/replica-protocol.test.mjs index 20f70258..006203ef 100644 --- a/js/tests/replica-protocol.test.mjs +++ b/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 = []; diff --git a/tests/e2e-ui/gateway/refresh.mjs b/tests/e2e-ui/gateway/refresh.mjs index cdab9ca8..c5895a35 100644 --- a/tests/e2e-ui/gateway/refresh.mjs +++ b/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); diff --git a/tests/e2e-ui/gateway/run.mjs b/tests/e2e-ui/gateway/run.mjs index 92b69396..1282a02d 100644 --- a/tests/e2e-ui/gateway/run.mjs +++ b/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(); diff --git a/tests/e2e-ui/ui/src/routes/+layout.svelte b/tests/e2e-ui/ui/src/routes/+layout.svelte index e8425bb0..cd8bb97f 100644 --- a/tests/e2e-ui/ui/src/routes/+layout.svelte +++ b/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).__captureReplicaDiagnostics === true + ? createReplicaDiagnostics() : undefined; + if (replicaDiagnostics) { + (globalThis as typeof globalThis & Record).__replicaDiagnosticSnapshot = () => replicaDiagnostics.snapshot(); + } const client = provideDistributed({ + ...(replicaDiagnostics ? { replica: { diagnostics: replicaDiagnostics } } : {}), boundaries: DISTRIBUTED_BOUNDARY_OPERATIONS, session: pageData.session, browser,