diff --git a/docs/live-query-delivery.md b/docs/live-query-delivery.md index ce58e823..78931e7a 100644 --- a/docs/live-query-delivery.md +++ b/docs/live-query-delivery.md @@ -30,6 +30,16 @@ 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. +When the last watch for a live operation is disposed, its local ownership is +retired at a monotonically increasing local boundary. A later live subscription +may take over an incomparable shared index only when that subscription started +after the retirement boundary. A subscription that started while the previous +stream was still active remains fenced against late frames from that stream. +Reopening the retired operation clears its retirement marker before transport +callbacks can arrive, so the reopened stream becomes an active owner again. +Retirement is local replica metadata; it does not claim that the server's +projection stopped or provide a server ordering signal. + 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/hydration.ts b/js/src/replica/distributed-replica/hydration.ts index 113cce33..6b5f8cac 100644 --- a/js/src/replica/distributed-replica/hydration.ts +++ b/js/src/replica/distributed-replica/hydration.ts @@ -64,6 +64,9 @@ export function serializeOperationProtocolState( ...(state.indexRevision === undefined ? {} : { indexRevision: state.indexRevision }), + ...(state.retiredAtRevision === undefined + ? {} + : { retiredAtRevision: state.retiredAtRevision }), indexKeys: Object.freeze([...state.indexKeys].sort()), pathRecords: Object.freeze( [...state.pathRecords] @@ -375,6 +378,15 @@ export function parseReplicaHydration( ) { hydrationInvalid('state.payload.nextIndexRevision'); } + if ( + operation?.retiredAtRevision !== undefined && + compareDistributedDecimal( + operation.retiredAtRevision as DistributedDecimalString, + nextIndexRevision + ) > 0 + ) { + hydrationInvalid('state.payload.nextIndexRevision'); + } } } return { @@ -405,6 +417,7 @@ export function parseOperationProtocolState( 'snapshotScope', 'indexClocks', 'indexRevision', + 'retiredAtRevision', 'indexKeys', 'pathRecords', 'cursors' @@ -502,6 +515,14 @@ export function parseOperationProtocolState( `${path}.indexRevision` ) }), + ...(raw.retiredAtRevision === undefined + ? {} + : { + retiredAtRevision: hydrationDecimal( + raw.retiredAtRevision, + `${path}.retiredAtRevision` + ) + }), indexKeys, pathRecords, cursors: Object.freeze(cursors) diff --git a/js/src/replica/distributed-replica/impl-fetch-live.ts b/js/src/replica/distributed-replica/impl-fetch-live.ts index 0882d5fb..5211aeee 100644 --- a/js/src/replica/distributed-replica/impl-fetch-live.ts +++ b/js/src/replica/distributed-replica/impl-fetch-live.ts @@ -65,7 +65,8 @@ export function emitWatchState(host: FetchLiveHost, key: string, allowFetch: boo export function closeActiveTransports(host: FetchLiveHost): void { for (const controller of host.inFlightAborts.values()) controller.abort(); host.inFlightAborts.clear(); - for (const entry of host.lives.values()) { + for (const [key, entry] of host.lives) { + retireLiveProtocol(host, key); entry.active = false; try { entry.unsubscribe(); @@ -232,6 +233,7 @@ export function retainLive( existing.count += 1; return; } + clearRetiredLiveProtocol(host, watch.key); const state = host.queryState(watch.key); state.live = 'connecting'; const entry: LiveEntry = { @@ -328,6 +330,7 @@ export function retainLive( }, error: (error) => { if (!entry.active || host.lives.get(watch.key) !== entry) return; + retireLiveProtocol(host, watch.key); entry.active = false; host.lives.delete(watch.key); const unsub = entry.unsubscribe; @@ -359,6 +362,7 @@ export function retainLive( } state.live = 'active'; } catch (error) { + retireLiveProtocol(host, watch.key); entry.active = false; host.lives.delete(watch.key); state.live = 'error'; @@ -376,6 +380,7 @@ export function fallbackFromLive( void fetchWatch(host, watch, true); return; } + retireLiveProtocol(host, watch.key); const protocol = host.operationProtocols.get(watch.key); if (protocol?.active === 'live') protocol.active = undefined; entry.active = false; @@ -426,6 +431,7 @@ export function restartLive(host: FetchLiveHost, key: string): void { const previous = host.lives.get(key); if (previous === undefined) return; const count = previous.count; + retireLiveProtocol(host, key); previous.active = false; host.lives.delete(key); try { @@ -465,6 +471,7 @@ export function releaseLive(host: FetchLiveHost, key: string): void { if (!entry) return; entry.count -= 1; if (entry.count > 0) return; + retireLiveProtocol(host, key); entry.active = false; host.lives.delete(key); entry.unsubscribe(); @@ -472,6 +479,23 @@ export function releaseLive(host: FetchLiveHost, key: string): void { host.emitState(key, false); } +/** + * Mark protocol state as no longer backed by a live transport. The state is + * retained for cache and hydration bookkeeping, but a later live stream may + * replace its incomparable membership only when that stream started after + * this boundary. + */ +function retireLiveProtocol(host: FetchLiveHost, key: string): void { + const state = host.operationProtocols.get(key)?.live; + if (state === undefined) return; + state.retiredAtRevision = host.allocateIndexRevision(); +} + +function clearRetiredLiveProtocol(host: FetchLiveHost, key: string): void { + const state = host.operationProtocols.get(key)?.live; + if (state !== undefined) state.retiredAtRevision = undefined; +} + function requestExtensions( host: FetchLiveHost, artifact: ReplicaOperationArtifact, diff --git a/js/src/replica/distributed-replica/impl.ts b/js/src/replica/distributed-replica/impl.ts index 4d8fbb33..a1807783 100644 --- a/js/src/replica/distributed-replica/impl.ts +++ b/js/src/replica/distributed-replica/impl.ts @@ -2344,6 +2344,21 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { } } if (!ownsIncomingIndex) continue; + if ( + !snapshot.indexesComparable && + state === group.live && + state.retiredAtRevision !== undefined && + liveStart !== undefined && + compareCanonicalDecimalStrings( + liveStart, + state.retiredAtRevision + ) > 0 + ) { + // A disposed stream is no longer an owner. Its boundary is + // still retained so a stream that started before disposal + // cannot win merely because the old transport later closed. + continue; + } latestOwnerRevision = latestOwnerRevision === undefined || compareCanonicalDecimalStrings( diff --git a/js/src/replica/distributed-replica/types.ts b/js/src/replica/distributed-replica/types.ts index 5d8761e7..4ae8faad 100644 --- a/js/src/replica/distributed-replica/types.ts +++ b/js/src/replica/distributed-replica/types.ts @@ -55,6 +55,8 @@ export type SerializedOperationProtocolState = { Readonly<{ scopeToken: string; position: string }> ])[]; readonly indexRevision?: string; + /** Local boundary after which a disposed live owner may be handed off. */ + readonly retiredAtRevision?: string; readonly indexKeys: readonly string[]; readonly pathRecords: readonly (readonly [string, string])[]; readonly cursors: readonly DistributedLiveCursor[]; @@ -140,6 +142,8 @@ export type OperationProtocolState = { snapshotScope?: DistributedOpaqueString; indexClocks: Map; indexRevision?: string; + /** Local boundary after which a disposed live owner may be handed off. */ + retiredAtRevision?: string; indexKeys: Set; pathRecords: Map; cursors: readonly DistributedLiveCursor[]; diff --git a/js/tests/replica-protocol.test.mjs b/js/tests/replica-protocol.test.mjs index 006203ef..b76de640 100644 --- a/js/tests/replica-protocol.test.mjs +++ b/js/tests/replica-protocol.test.mjs @@ -1975,6 +1975,356 @@ test('snapshot live replaces SSR membership, updates and removes rows, and fence assert.deepEqual(replica.read(Todos, {}).data.todos, []); }); +test('snapshot live takes over a nested graph from a disposed page subscription', () => { + const observers = []; + const previousPage = { + ...FeaturedGamesWithOwner, + live: { id: 'live:featured-owner', document: 'subscription FeaturedOwner { featuredGames { id owner { id name } } }' } + }; + const replica = createDistributedReplica({ transport: { + fetch() { throw new Error('complete snapshot must not force HTTP fallback'); }, + subscribe(_request, observer) { observers.push(observer); return () => {}; } + } }); + const oldFrame = (position, ownerName = 'Owner') => gamesFrame({ + artifact: previousPage, responseKey: 'featuredGames', + operation: previousPage.live.id, position, + ownerId: 'user-1', ownerName, indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + }); + replica.writeResult(previousPage, {}, gamesFrame({ + artifact: previousPage, responseKey: 'featuredGames', position: '1', + ownerId: 'user-1', ownerName: 'Owner', indexesComparable: false + }), 'ssr'); + const oldWatch = replica.watch(previousPage, {}, { live: true }); + observers[0].next(oldFrame('2')); + oldWatch.destroy(); + 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 current = replica.watch(GamesWithOwnerLiveOperation, {}, { live: true }); + observers[1].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '3', + ownerId: 'user-1', ownerName: 'Owner', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.deepEqual(current.get().errors, []); + assert.equal(current.get().complete, true); + assert.equal(current.get().data.games.length, 1, 'disposed page must not permanently own the shared relationship'); + observers[0].next(oldFrame('4', 'Late disposed callback')); + assert.equal(current.get().data.games[0].owner.name, 'Owner'); + current.destroy(); +}); + +test('snapshot live keeps a stream started before disposal behind the retired owner', () => { + const observers = []; + const previousPage = { + ...FeaturedGamesWithOwner, + live: { id: 'live:featured-owner-before-disposal', document: 'subscription FeaturedOwnerBeforeDisposal { featuredGames { id owner { id name } } }' } + }; + const replica = createDistributedReplica({ transport: { + fetch() { throw new Error('complete snapshot must not force HTTP fallback'); }, + subscribe(_request, observer) { observers.push(observer); return () => {}; } + } }); + const oldFrame = (ownerName) => gamesFrame({ + artifact: previousPage, responseKey: 'featuredGames', operation: previousPage.live.id, + position: '2', ownerId: 'user-1', ownerName, indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + }); + const oldWatch = replica.watch(previousPage, {}, { live: true }); + observers[0].next(oldFrame('active owner')); + 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 current = replica.watch(GamesWithOwnerLiveOperation, {}, { live: true }); + observers[1].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '3', + ownerId: 'user-1', ownerName: 'started before disposal', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.deepEqual(current.get().data.games, []); + oldWatch.destroy(); + observers[1].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '4', + ownerId: 'user-1', ownerName: 'still started before disposal', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.deepEqual(current.get().data.games, []); + current.destroy(); +}); + +test('reopening a retired live owner restores its ownership fence', () => { + const observers = []; + const previousPage = { + ...FeaturedGamesWithOwner, + live: { id: 'live:featured-owner-reopened', document: 'subscription FeaturedOwnerReopened { featuredGames { id owner { id name } } }' } + }; + const frame = (artifact, responseKey, operation, position, ownerName) => gamesFrame({ + artifact, responseKey, operation, position, ownerId: 'user-1', ownerName, + indexesComparable: false, live: { mode: 'snapshot', reset: true, cursors: [] } + }); + const replica = createDistributedReplica({ transport: { + fetch() { throw new Error('complete snapshot must not force HTTP fallback'); }, + subscribe(_request, observer) { + observers.push(observer); + if (observers.length === 2) { + observer.next(frame(previousPage, 'featuredGames', previousPage.live.id, '3', 'reopened owner')); + } + return () => {}; + } } }); + const first = replica.watch(previousPage, {}, { live: true }); + observers[0].next(frame(previousPage, 'featuredGames', previousPage.live.id, '2', 'first owner')); + first.destroy(); + const reopened = replica.watch(previousPage, {}, { live: true }); + 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 current = replica.watch(GamesWithOwnerLiveOperation, {}, { live: true }); + observers[2].next(frame( + GamesWithOwnerLiveOperation, + 'games', + GamesWithOwnerLiveOperation.live.id, + '4', + 'current contender' + )); + assert.deepEqual(current.get().data.games, []); + reopened.destroy(); + current.destroy(); +}); + +test('reopening a retired live owner clears the marker before its first frame', () => { + const observers = []; + const previousPage = { + ...FeaturedGamesWithOwner, + live: { id: 'live:featured-owner-reopened-empty', document: 'subscription FeaturedOwnerReopenedEmpty { featuredGames { id owner { id name } } }' } + }; + const frame = (artifact, responseKey, operation, position, ownerName) => gamesFrame({ + artifact, responseKey, operation, position, ownerId: 'user-1', ownerName, + indexesComparable: false, live: { mode: 'snapshot', reset: true, cursors: [] } + }); + const replica = createDistributedReplica({ transport: { + fetch() { throw new Error('complete snapshot must not force HTTP fallback'); }, + subscribe(_request, observer) { observers.push(observer); return () => {}; } + } }); + const first = replica.watch(previousPage, {}, { live: true }); + observers[0].next(frame(previousPage, 'featuredGames', previousPage.live.id, '2', 'first owner')); + first.destroy(); + // Reopen the operation, but do not deliver a frame yet. The replacement + // stream must still be treated as an active owner immediately. + const reopened = replica.watch(previousPage, {}, { live: true }); + 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 contender = replica.watch(GamesWithOwnerLiveOperation, {}, { live: true }); + observers[1].next(frame( + GamesWithOwnerLiveOperation, + 'games', + GamesWithOwnerLiveOperation.live.id, + '3', + 'contender before reopened frame' + )); + assert.deepEqual(contender.get().data.games, []); + reopened.destroy(); + contender.destroy(); +}); + +test('two watches retire shared live ownership only after final disposal', () => { + const observers = []; + const previousPage = { + ...FeaturedGamesWithOwner, + live: { id: 'live:featured-owner-refcount', document: 'subscription FeaturedOwnerRefcount { featuredGames { id owner { id name } } }' } + }; + const replica = createDistributedReplica({ transport: { + fetch() { throw new Error('complete snapshot must not force HTTP fallback'); }, + subscribe(_request, observer) { observers.push(observer); return () => {}; } + } }); + const oldFrame = (ownerName) => gamesFrame({ + artifact: previousPage, responseKey: 'featuredGames', operation: previousPage.live.id, + position: '2', ownerId: 'user-1', ownerName, indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + }); + const first = replica.watch(previousPage, {}, { live: true }); + const second = replica.watch(previousPage, {}, { live: true }); + assert.equal(observers.length, 1, 'same operation shares one live transport'); + observers[0].next(oldFrame('active owner')); + first.destroy(); + + 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 startedBeforeFinalRelease = replica.watch( + GamesWithOwnerLiveOperation, + {}, + { live: true } + ); + observers[1].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '3', + ownerId: 'user-1', ownerName: 'before final release', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.deepEqual(startedBeforeFinalRelease.get().data.games, []); + second.destroy(); + observers[0].next(oldFrame('late disposed callback')); + observers[1].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '4', + ownerId: 'user-1', ownerName: 'still before final release', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.deepEqual(startedBeforeFinalRelease.get().data.games, []); + startedBeforeFinalRelease.destroy(); + + const afterFinalRelease = replica.watch( + GamesWithOwnerLiveOperation, + {}, + { live: true } + ); + observers[2].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '5', + ownerId: 'user-1', ownerName: 'after final release', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.equal(afterFinalRelease.get().data.games[0].owner.name, 'after final release'); + afterFinalRelease.destroy(); +}); + +test('terminal live retirement permits a later snapshot handoff', () => { + for (const terminal of ['error', 'complete']) { + const observers = []; + const previousPage = { + ...FeaturedGamesWithOwner, + live: { id: `live:featured-owner-${terminal}`, document: `subscription FeaturedOwner${terminal} { featuredGames { id owner { id name } } }` } + }; + const replica = createDistributedReplica({ transport: { + fetch() { throw new Error('terminal stream test does not need HTTP fallback'); }, + subscribe(_request, observer) { observers.push(observer); return () => {}; } + } }); + const oldFrame = gamesFrame({ + artifact: previousPage, responseKey: 'featuredGames', operation: previousPage.live.id, + position: '2', ownerId: 'user-1', ownerName: `${terminal} owner`, indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + }); + replica.writeResult(previousPage, {}, gamesFrame({ + artifact: previousPage, responseKey: 'featuredGames', position: '1', + ownerId: 'user-1', ownerName: `${terminal} owner`, indexesComparable: false + }), 'ssr'); + const oldWatch = replica.watch(previousPage, {}, { live: true }); + observers[0].next(oldFrame); + if (terminal === 'error') observers[0].error(new Error('terminal test error')); + else observers[0].complete(); + // A terminal transport is no longer the page's live request. Dispose the + // watcher before another page seeds the same shared index; an error may + // otherwise be intentionally resumed by the replica on the next write. + oldWatch.destroy(); + + 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 current = replica.watch(GamesWithOwnerLiveOperation, {}, { live: true }); + observers[1].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '3', + ownerId: 'user-1', ownerName: `${terminal} handoff`, indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.equal(current.get().data.games.length, 1, `${terminal} handoff should replace the retired owner`); + assert.equal(current.get().data.games[0].owner.name, `${terminal} handoff`); + observers[0].next(oldFrame); + assert.equal(current.get().data.games[0].owner.name, `${terminal} handoff`); + current.destroy(); + } +}); + +test('retired live ownership survives dehydration before a later handoff', () => { + const observers = []; + const previousPage = { + ...FeaturedGamesWithOwner, + live: { id: 'live:featured-owner-hydrated', document: 'subscription FeaturedOwnerHydrated { featuredGames { id owner { id name } } }' } + }; + const transport = { + fetch() { throw new Error('complete snapshot must not force HTTP fallback'); }, + subscribe(_request, observer) { observers.push(observer); return () => {}; } + }; + const replica = createDistributedReplica({ transport }); + const oldFrame = gamesFrame({ + artifact: previousPage, responseKey: 'featuredGames', + operation: previousPage.live.id, position: '2', ownerId: 'user-1', + ownerName: 'retired owner', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + }); + replica.writeResult(previousPage, {}, gamesFrame({ + artifact: previousPage, responseKey: 'featuredGames', position: '1', + ownerId: 'user-1', ownerName: 'retired owner', indexesComparable: false + }), 'ssr'); + replica.read(previousPage, {}); + const oldWatch = replica.watch(previousPage, {}, { live: true }); + observers[0].next(oldFrame); + oldWatch.destroy(); + const dehydrated = replica.dehydrate(); + const oldProtocol = dehydrated.payload.operations.find((entry) => + entry.live?.retiredAtRevision !== undefined + ); + assert.ok(oldProtocol?.live?.retiredAtRevision !== undefined); + + const restoredObservers = []; + const restored = createDistributedReplica({ + transport: { + fetch() { throw new Error('complete snapshot must not force HTTP fallback'); }, + subscribe(_request, observer) { restoredObservers.push(observer); return () => {}; } + } + }); + assert.equal(restored.hydrate(dehydrated, dehydrated.scope), true); + const current = restored.watch(GamesWithOwnerLiveOperation, {}, { live: true }); + restoredObservers[0].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '3', + ownerId: 'user-1', ownerName: 'hydrated handoff', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + assert.equal(current.get().data.games.length, 1); + assert.equal(restored.hydrate(dehydrated, dehydrated.scope), true); + assert.equal(restoredObservers.length, 2); + restoredObservers[1].next(gamesFrame({ + artifact: GamesWithOwnerLiveOperation, responseKey: 'games', + operation: GamesWithOwnerLiveOperation.live.id, position: '4', + ownerId: 'user-1', ownerName: 'active receiver after hydrate', indexesComparable: false, + live: { mode: 'snapshot', reset: true, cursors: [] } + })); + const contender = restored.watch(previousPage, {}, { live: true }); + restoredObservers[2].next(oldFrame); + assert.equal(current.get().data.games[0].owner.name, 'active receiver after hydrate'); + contender.destroy(); + current.destroy(); +}); + test('snapshot live adds a row with an unchanged SSR-owned nested relationship', () => { let observer; const replica = createDistributedReplica({ transport: {