fix: retire disposed live ownership

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.

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)

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<TData, TVariables extends GraphqlVariables>(
 		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<TData, TVariables extends GraphqlVariables>(
 				},
 				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<TData, TVariables extends GraphqlVariables>(
 		}
 		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<TData, TVariables extends GraphqlVariables>(
 		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,13 +471,31 @@ 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();
 	host.queryState(key).live = 'off';
 	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<unknown, GraphqlVariables>,

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(

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<string, IndexProtocolClock>;
 	indexRevision?: string;
+	/** Local boundary after which a disposed live owner may be handed off. */
+	retiredAtRevision?: string;
 	indexKeys: Set<string>;
 	pathRecords: Map<string, string>;
 	cursors: readonly DistributedLiveCursor[];