diff --git a/README.md b/README.md index 906cdcd10..c555924c2 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,10 @@ ## How it works +Live queries stream authorized results even when row-level permissions prevent +safe resume cursors. [Snapshot and resumable delivery](docs/live-query-delivery.md) +describe the two modes and their consistency guarantees. + Write the domain once, compose it into one `Service` or several, then generate the client. Each stage below uses real code from [`tests/e2e-ui`](tests/e2e-ui). diff --git a/docs/gateway/live-sharing.md b/docs/gateway/live-sharing.md index 1ef09998d..404f7bf58 100644 --- a/docs/gateway/live-sharing.md +++ b/docs/gateway/live-sharing.md @@ -19,12 +19,16 @@ upstream subscription and producer. HTTP control requests do not execute result Groups bind the exact document, variables, origin subject/cache scope, policy, schema and protocol. Each consumer keeps its own transport ID, expiry, queue and freshness requirements. Different resume cursors start independent replay. Handoff -requires the same operation plus an exact supported, comparable cursor vector and +requires the same operation plus an exact resumable, comparable cursor vector and matching data; the consumer's replay frames remain queued before future shared frames. Unknown cursors keep independent streams. Equal data alone never proves cursor equality. Duplicate suppression hashes the whole data and protocol envelope, so new confirmation evidence is delivered even when values are unchanged. +Cursorless `live.mode = "snapshot"` responses cannot prove shared replay handoff; +they retain independent authorized subscriptions and reconnect with fresh queries. +See [live query delivery](../live-query-delivery.md) for the wire contract. + Defaults bound a coordinator to 256 groups, 1,024 consumers per group, 16 pending frames per consumer, 1 MiB per full frame, eight retained history frames and a one-hour group lifetime. Native ingress bounds socket/request counts and wire diff --git a/docs/live-query-delivery.md b/docs/live-query-delivery.md new file mode 100644 index 000000000..970b4a4c5 --- /dev/null +++ b/docs/live-query-delivery.md @@ -0,0 +1,28 @@ +# Live queries and authorization + +`@live` keeps an authorized query current. Resuming missed projection changes is +a separate capability: a partition-wide cursor can reveal activity outside a +row-filtered result, even when the rows themselves are protected. + +Every live response declares `extensions.distributed.live.mode`: + +- `snapshot`: a fresh authorized replacement result, with `reset: true` and + `cursors: []`. It carries no comparable index vector or projection observations. + The client keeps listening and reconnects with a fresh query, not a resume token. +- `resumable`: a result with matching, nonempty index and cursor vectors. Existing + resume validation, reset, replay and causal reconciliation rules apply. + +Snapshot delivery does not relax read permissions or invent causal evidence. +Changes affecting only denied rows must not produce activity frames. A row +leaving the authorized result disappears from that operation's membership; +absence is not a globally authoritative deletion or tombstone. + +Within one current subscription, snapshot frames are ordered. The client fences +older HTTP requests when a live result takes ownership, and fences callbacks +from disposed subscriptions and previous authorization generations. Local cache +membership revisions are not server projection positions and cannot confirm +optimistic commands. + +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/README.md b/js/README.md index ff30b5e97..e0669ad52 100644 --- a/js/README.md +++ b/js/README.md @@ -51,6 +51,12 @@ to validate that committed artifacts are current without rewriting them. A co-located route document can opt into SSR and live continuation: +Row-filtered `@live` queries remain subscribed using authorized replacement +snapshots; cursor-capable queries retain resumable delivery. Neither requires +application polling. The v5 wire protocol uses `live.mode` (`snapshot` or +`resumable`), replacing `live.supported`; upgrade client and server together. +See [live query delivery](../docs/live-query-delivery.md). + ```graphql query Todos @load @live { todos(order_by: [{ status: asc }, { todo_id: asc }]) { diff --git a/js/src/protocol.ts b/js/src/protocol.ts index e99c285c8..d6b3dd457 100644 --- a/js/src/protocol.ts +++ b/js/src/protocol.ts @@ -149,10 +149,10 @@ export type DistributedQuerySnapshot = Readonly< } >; -/** Per-frame decision about live support, reset, and resumable cursors. */ +/** Per-frame delivery mode, reset decision, and resumable cursors. */ export type DistributedLiveMetadata = Readonly< Record & { - supported: boolean; + mode: 'snapshot' | 'resumable'; reset: boolean; cursors: readonly DistributedLiveCursor[]; } @@ -596,7 +596,8 @@ function parseSnapshot(value: unknown): DistributedQuerySnapshot { function parseLive(value: unknown): DistributedLiveMetadata { const path = 'extensions.distributed.live'; const live = record(value, path); - if (typeof live.supported !== 'boolean') invalid(`${path}.supported`); + if (live.mode !== 'snapshot' && live.mode !== 'resumable') invalid(`${path}.mode`); + if ('supported' in live) invalid(`${path}.supported`); if (typeof live.reset !== 'boolean') invalid(`${path}.reset`); if ( !Array.isArray(live.cursors) || @@ -613,12 +614,12 @@ function parseLive(value: unknown): DistributedLiveMetadata { cursors.map((cursor) => cursor.projection), `${path}.cursors` ); - if (!live.supported && (!live.reset || cursors.length !== 0)) { + if (live.mode === 'snapshot' && (!live.reset || cursors.length !== 0)) { invalid(path); } return Object.freeze({ ...live, - supported: live.supported, + mode: live.mode, reset: live.reset, cursors }) as DistributedLiveMetadata; @@ -630,7 +631,10 @@ function validateLiveSnapshot( ): void { if (live === undefined) return; if (snapshot === undefined) invalid('extensions.distributed.snapshot'); - if (!live.supported) return; + if (live.mode === 'snapshot') { + if (snapshot.indexesComparable) invalid('extensions.distributed.snapshot.indexesComparable'); + return; + } if (!snapshot.indexesComparable) { invalid('extensions.distributed.snapshot.indexesComparable'); } diff --git a/js/src/replica/distributed-replica/clocks.ts b/js/src/replica/distributed-replica/clocks.ts index ff20bc6db..44eafb9ab 100644 --- a/js/src/replica/distributed-replica/clocks.ts +++ b/js/src/replica/distributed-replica/clocks.ts @@ -122,7 +122,7 @@ export function latestCursors( live: DistributedProtocolEnvelope['live'] ): readonly DistributedLiveCursor[] { if (live !== undefined) { - return live.supported ? live.cursors : Object.freeze([]); + return live.mode === 'resumable' ? live.cursors : Object.freeze([]); } return Object.freeze( snapshot.indexes.flatMap((index) => diff --git a/js/src/replica/distributed-replica/impl-fetch-live.ts b/js/src/replica/distributed-replica/impl-fetch-live.ts index 650c8182e..72f55dbaf 100644 --- a/js/src/replica/distributed-replica/impl-fetch-live.ts +++ b/js/src/replica/distributed-replica/impl-fetch-live.ts @@ -1,7 +1,5 @@ -import { CacheRevisionConflictError } from '../../internal/cache-engine.js'; import type { GraphqlVariables } from '../../types.js'; import { - parseGraphqlResponseExtensions, type DistributedLiveCursor, type DistributedProtocolEnvelope } from '../../protocol.js'; @@ -309,13 +307,8 @@ export function retainLive( } return; } - let unsupportedLive = false; try { - unsupportedLive = - parseGraphqlResponseExtensions(result.extensions) - ?.distributed?.live?.supported === false; - if (unsupportedLive) state.live = 'off'; - const distributed = host.writeCanonicalResult( + host.writeCanonicalResult( watch.artifact, watch.variables, result, @@ -323,26 +316,10 @@ export function retainLive( undefined, projectionGeneration ); - if (distributed.live?.supported === false) { - fallbackFromLive(host, watch, entry); - return; - } state.live = 'active'; entry.operationGeneration = host.operationGeneration(watch.key); } catch (error) { - if ( - unsupportedLive && - error instanceof CacheRevisionConflictError - ) { - /* - * Revision zero is shared by provisional fallbacks. - * Another operation may already have filled the same - * semantic index differently; HTTP remains authoritative. - */ - fallbackFromLive(host, watch, entry); - return; - } state.live = 'error'; state.errors = stableErrors(state.errors, [graphqlError(error)]); host.emitState(watch.key, false); @@ -429,14 +406,13 @@ export function fallbackFromLive( /* * Keep the inactive entry as an authorization-generation-scoped * sentinel. Query ingestion calls resumeLiveWatches(); deleting this - * entry would otherwise reopen an unsupported or completed stream + * entry would otherwise reopen a completed stream * immediately. Authorization invalidation clears it and may retry. * - * A supported live frame advances the operation generation. Any HTTP + * An accepted live frame advances the operation generation. Any HTTP * request that was already running is therefore doomed by its response - * fence; drain it before starting the authoritative fallback. A first - * unsupported frame never advances the generation, so its overlapping - * HTTP request remains valid and can be reused directly. + * fence; drain it before starting the authoritative fallback. A stream + * that completed before its first frame can reuse the in-flight query. */ if (supersededFlight === undefined) { refresh(); diff --git a/js/src/replica/distributed-replica/impl.ts b/js/src/replica/distributed-replica/impl.ts index 83024fefa..e3eea0813 100644 --- a/js/src/replica/distributed-replica/impl.ts +++ b/js/src/replica/distributed-replica/impl.ts @@ -1184,21 +1184,9 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { } this.#validateLiveSnapshot(snapshot, live); - const unsupportedLive = - source === 'live' && live?.supported === false; + const snapshotLive = source === 'live' && live?.mode === 'snapshot'; const reset = live?.reset === true; const group = this.#operationProtocols.get(key)!; - /* - * An unsupported subscription response is an authorized fallback - * snapshot, not a live source. In particular, a row-filtered snapshot - * has no comparable index vector, so retaining live ownership here - * would reject every later query handoff. Relinquish any prior live - * ownership without advancing the generation; the forced HTTP fallback - * starts against the generation that remains after this frame. - */ - if (unsupportedLive && group.active === 'live') { - group.active = undefined; - } const previousActiveSource = group.active; const handoff = previousActiveSource !== undefined && @@ -1226,8 +1214,22 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { requestRevision, source ); + // A non-resumable stream has no causal vector to compare. An HTTP + // refresh may replace it only if the request began after its last + // accepted membership. fetchWatch also fences intervening live frames. + // Taking query ownership restarts the stream, fencing queued callbacks. + const snapshotRefresh = + source === 'network' && + previousActiveSource === 'live' && + !snapshot.indexesComparable && + activeState?.snapshotScope === undefined && + activeState?.indexRevision !== undefined && + requestRevision !== undefined && + compareCanonicalDecimalStrings(requestRevision, activeState.indexRevision) > 0; const handoffBlocked = handoff && + !snapshotLive && + !snapshotRefresh && ( !snapshot.indexesComparable || !isComparableHandoffDisposition(ownDisposition) || @@ -1251,7 +1253,6 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { : sharedDisposition.disposition ?? disposition; } const sourceSwitched = - !unsupportedLive && !handoffBlocked && isComparableHandoffDisposition(disposition) && this.#activateOperationSource( @@ -1316,24 +1317,16 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { !rejectedHandoff && disposition !== 'lower' && disposition !== 'incomparable'; - /* - * Revision zero is the cache engine's lowest legal checkpoint. It lets - * an unsupported live response fill an empty cache immediately while - * guaranteeing that any HTTP request revision can replace it. - */ + // Snapshot frames receive local membership revisions, not fabricated + // projection clocks. Live ownership fences older overlapping HTTP work. const indexRevision = - unsupportedLive - ? '0' + writeIndexes && + (sourceSwitched || sharedDisposition.disposition === 'higher') + ? this.#allocateIndexRevision() : writeIndexes && - ( - sourceSwitched || - sharedDisposition.disposition === 'higher' - ) - ? this.#allocateIndexRevision() - : writeIndexes && - sharedDisposition.disposition === 'equal' && - sharedDisposition.indexRevision !== undefined - ? sharedDisposition.indexRevision + sharedDisposition.disposition === 'equal' && + sharedDisposition.indexRevision !== undefined + ? sharedDisposition.indexRevision : snapshot.indexesComparable && disposition === 'equal' && operationState.indexRevision !== undefined @@ -1615,7 +1608,7 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { } else if (live?.reset === true || !snapshot.indexesComparable) { operationState.cursors = Object.freeze([]); } - if (source === 'live' && !unsupportedLive) { + if (source === 'live') { this.#advanceOperationGeneration(key); } if (source !== 'live' && sourceSwitched) { @@ -2838,7 +2831,7 @@ export class DistributedReplicaImpl implements DistributedReplicaApi { snapshot: DistributedQuerySnapshot, live: DistributedProtocolEnvelope['live'] ): void { - if (live === undefined || !live.supported) return; + if (live === undefined || live.mode === 'snapshot') return; if (!snapshot.indexesComparable) { protocolInvalid( 'extensions.distributed.snapshot.indexesComparable' diff --git a/js/tests/fixtures/adapter-conformance.mjs b/js/tests/fixtures/adapter-conformance.mjs index fb200614a..c39ab1305 100644 --- a/js/tests/fixtures/adapter-conformance.mjs +++ b/js/tests/fixtures/adapter-conformance.mjs @@ -302,6 +302,7 @@ export function todoFrame( authorizationGeneration = 'auth-1', position = '1', source = 'query', + mode = 'resumable', reset = false, errors } = {} @@ -343,9 +344,9 @@ export function todoFrame( snapshot: { scopeToken: `snapshot:${root.field}`, recordsComplete: true, - indexesComparable: true, + indexesComparable: mode === 'resumable', records, - indexes: [ + indexes: mode === 'snapshot' ? [] : [ { projection: 'todos-projector', scopeToken: `index:${root.field}`, @@ -358,9 +359,9 @@ export function todoFrame( ...(source === 'live' ? { live: { - supported: true, - reset, - cursors: [resume] + mode, + reset: mode === 'snapshot' || reset, + cursors: mode === 'snapshot' ? [] : [resume] } } : {}) diff --git a/js/tests/fixtures/unique-key-artifact.mjs b/js/tests/fixtures/unique-key-artifact.mjs index b4fc191d7..ba4c5edad 100644 --- a/js/tests/fixtures/unique-key-artifact.mjs +++ b/js/tests/fixtures/unique-key-artifact.mjs @@ -26,7 +26,7 @@ export function frame(position, targetId, targetTitle) { protocolVersion: artifact.protocol.version, schemaHash: artifact.protocol.schemaHash, authorizationGeneration: 'auth-1', cacheScope: 'unique-key-cache', operation: position === '1' ? artifact.id : artifact.live.id, - ...(position === '1' ? {} : { live: { supported: true, reset: false, cursors: [ + ...(position === '1' ? {} : { live: { mode: "resumable", reset: false, cursors: [ { projection: 'unique-key-projector', position, token: `resume:${position}` } ] } }), snapshot: { scopeToken: 'unique-key-snapshot', recordsComplete: true, indexesComparable: true, diff --git a/js/tests/protocol-transport.test.mjs b/js/tests/protocol-transport.test.mjs index 9d067a8be..35542556d 100644 --- a/js/tests/protocol-transport.test.mjs +++ b/js/tests/protocol-transport.test.mjs @@ -81,7 +81,7 @@ function distributedEnvelope() { observations: [] }, live: { - supported: true, + mode: "resumable", reset: false, cursors: [ { @@ -237,7 +237,7 @@ test('protocol parser requires exact live snapshot alignment', () => { parseDistributedProtocolEnvelope({ ...envelope, snapshot: undefined, - live: { supported: false, reset: true, cursors: [] } + live: { mode: "snapshot", reset: true, cursors: [] } }), (error) => error instanceof DistributedProtocolError && @@ -253,7 +253,7 @@ test('protocol parser requires exact live snapshot alignment', () => { indexes: [], observations: [] }, - live: { supported: true, reset: true, cursors: [] } + live: { mode: "resumable", reset: true, cursors: [] } }), (error) => error instanceof DistributedProtocolError && @@ -333,6 +333,28 @@ test('protocol parser requires exact live snapshot alignment', () => { } }); +test('snapshot delivery requires an explicit cursorless reset and fails closed on old modes', () => { + const base = distributedEnvelope(); + const envelope = { + ...base, + snapshot: { ...base.snapshot, indexesComparable: false, indexes: [], observations: [] }, + live: { mode: 'snapshot', reset: true, cursors: [] } + }; + assert.equal(parseDistributedProtocolEnvelope(envelope).live.mode, 'snapshot'); + for (const live of [ + { supported: false, reset: true, cursors: [] }, + { ...envelope.live, supported: false }, + { ...envelope.live, mode: 'unknown' }, + { ...envelope.live, reset: false }, + { ...envelope.live, cursors: base.live.cursors } + ]) { + assert.throws(() => parseDistributedProtocolEnvelope({ ...envelope, live }), DistributedProtocolError); + } + assert.throws(() => parseDistributedProtocolEnvelope({ + ...envelope, snapshot: base.snapshot + }), DistributedProtocolError); +}); + test('protocol parser accepts 64 resume cursors and rejects 65', () => { const cursors = Array.from({ length: 65 }, (_, index) => ({ projection: `projection-${index}`, @@ -349,7 +371,7 @@ test('protocol parser accepts 64 resume cursors and rejects 65', () => { const accepted = parseDistributedProtocolEnvelope({ ...envelope, snapshot: { ...envelope.snapshot, indexes: indexes.slice(0, 64) }, - live: { supported: true, reset: false, cursors: cursors.slice(0, 64) } + live: { mode: "resumable", reset: false, cursors: cursors.slice(0, 64) } }); assert.equal(accepted.live.cursors.length, 64); assert.equal(accepted.snapshot.indexes.length, 64); @@ -358,7 +380,7 @@ test('protocol parser accepts 64 resume cursors and rejects 65', () => { () => parseDistributedProtocolEnvelope({ ...envelope, - live: { supported: true, reset: false, cursors } + live: { mode: "resumable", reset: false, cursors } }), (error) => error instanceof DistributedProtocolError && @@ -369,7 +391,7 @@ test('protocol parser accepts 64 resume cursors and rejects 65', () => { parseDistributedProtocolEnvelope({ ...envelope, snapshot: { ...envelope.snapshot, indexes }, - live: { supported: false, reset: true, cursors: [] } + live: { mode: "snapshot", reset: true, cursors: [] } }), (error) => error instanceof DistributedProtocolError && diff --git a/js/tests/replica-protocol.test.mjs b/js/tests/replica-protocol.test.mjs index 92230bf7c..20f70258a 100644 --- a/js/tests/replica-protocol.test.mjs +++ b/js/tests/replica-protocol.test.mjs @@ -416,11 +416,11 @@ function wireFrame(options = {}) { options.live === undefined ? undefined : { - supported: options.live.supported ?? true, + mode: options.live.mode ?? 'resumable', reset: options.live.reset ?? false, cursors: options.live.cursors ?? - (options.live.supported === false ? [] : [resume]) + (options.live.mode === 'snapshot' ? [] : [resume]) }; return { data: { todos: rows }, @@ -725,7 +725,7 @@ test('comparable live snapshot cannot drop an Eventual list row after confirmati position: '2', operation: Todos.live.id, rows: [{ id: 'todo-1', title: 'first' }], - live: { supported: true } + live: { mode: "resumable" } }, 'live' ); @@ -750,7 +750,7 @@ test('comparable live snapshot cannot drop an Eventual list row after confirmati { id: 'todo-1', title: 'first' }, { id: 'todo-2', title: 'posted' } ], - live: { supported: true }, + live: { mode: "resumable" }, records: [ { path: ['todos', '0'], @@ -832,7 +832,7 @@ test('Eventual membership fences are independent per index', () => { { id: 'todo-1', title: 'first' }, { id: 'todo-2', title: 'posted' } ], - live: { supported: true } + live: { mode: "resumable" } }, 'live' ); @@ -845,7 +845,7 @@ test('Eventual membership fences are independent per index', () => { indexScope: 'index:todos-open', snapshotScope: 'snapshot:todos-open', rows: [{ id: 'todo-1', title: 'first' }], - live: { supported: true } + live: { mode: "resumable" } }, 'live', TodosOpen @@ -867,7 +867,7 @@ test('Eventual membership fences are independent per index', () => { { id: 'todo-1', title: 'first' }, { id: 'todo-2', title: 'posted' } ], - live: { supported: true } + live: { mode: "resumable" } }, 'live', TodosOpen @@ -915,7 +915,7 @@ test('overlapping Eventual commands retain every membership-fence owner', () => position: '2', operation: Todos.live.id, rows: [{ id: 'todo-1', title: 'first' }], - live: { supported: true } + live: { mode: "resumable" } }, 'live' ); @@ -1438,7 +1438,7 @@ test('a comparable shared root cannot promote an incomparable nested sibling', a ownerName: 'unfenced live frame', operation: GamesWithOwnerLiveOperation.live.id, live: { - supported: true, + mode: "resumable", reset: false, cursors: [ { @@ -1469,7 +1469,7 @@ test('an older operation reset cannot erase a shared index owned by a newer arti operation: Todos.live.id, position: '5', rows: [{ id: 'todo-live', title: 'old live owner' }], - live: { supported: true } + live: { mode: "resumable" } }, 'live' ); @@ -1490,7 +1490,7 @@ test('an older operation reset cannot erase a shared index owned by a newer arti operation: Todos.live.id, position: '8', rows: [{ id: 'todo-reset', title: 'older reset' }], - live: { supported: true, reset: true } + live: { mode: "resumable", reset: true } }, 'live' ); @@ -1508,7 +1508,7 @@ test('reset preserves an equal-vector index with another operation co-owner', () operation: Todos.live.id, position: '5', rows: [{ id: 'todo-shared', title: 'shared snapshot' }], - live: { supported: true } + live: { mode: "resumable" } }, 'live' ); @@ -1529,7 +1529,7 @@ test('reset preserves an equal-vector index with another operation co-owner', () operation: Todos.live.id, position: '4', rows: [{ id: 'todo-reset', title: 'older reset' }], - live: { supported: true, reset: true } + live: { mode: "resumable", reset: true } }, 'live' ); @@ -1785,7 +1785,7 @@ test('a live handoff fences an HTTP response launched in the prior generation', watch.destroy(); }); -test('unsupported live fallback cannot fence HTTP membership or later revalidation', async () => { +test('snapshot live stays attached and fences overlapping HTTP membership', async () => { const fetches = []; const subscriptions = []; let unsubscribeCount = 0; @@ -1819,14 +1819,14 @@ test('unsupported live fallback cannot fence HTTP membership or later revalidati rows: [{ id: 'todo-live', title: 'provisional live fallback' }], recordScope: 'record:live', indexesComparable: false, - live: { supported: false, reset: true } + live: { mode: "snapshot", reset: true } }) ); assert.deepEqual(replica.read(Todos, {}).data.todos, [ { id: 'todo-live', title: 'provisional live fallback' } ]); - assert.equal(watch.get().live, 'off'); - assert.equal(unsubscribeCount, 1); + assert.equal(watch.get().live, 'active'); + assert.equal(unsubscribeCount, 0); await Promise.resolve(); fetches[0].resolve( @@ -1839,12 +1839,12 @@ test('unsupported live fallback cannot fence HTTP membership or later revalidati ); await new Promise((resolve) => setImmediate(resolve)); assert.deepEqual(replica.read(Todos, {}).data.todos, [ - { id: 'todo-http', title: 'newer HTTP membership' } + { id: 'todo-live', title: 'provisional live fallback' } ]); assert.equal( subscriptions.length, 1, - 'query fallback must not immediately reopen an unsupported stream' + 'snapshot delivery must keep the original subscription attached' ); const revalidation = replica.revalidate({ @@ -1866,13 +1866,19 @@ test('unsupported live fallback cannot fence HTTP membership or later revalidati assert.deepEqual(replica.read(Todos, {}).data.todos, [ { id: 'todo-revalidated', title: 'revalidated membership' } ]); - assert.equal(subscriptions.length, 1); + assert.equal(subscriptions.length, 2, 'refresh restarts the snapshot stream'); + assert.equal(subscriptions[1].request.resume, undefined); + subscriptions[0].observer.next(wireFrame({ + operation: 'live:todos', rows: [], indexesComparable: false, + live: { mode: 'snapshot', reset: true } + })); + assert.equal(replica.read(Todos, {}).data.todos[0].id, 'todo-revalidated'); watch.destroy(); - assert.equal(unsubscribeCount, 1); + assert.equal(unsubscribeCount, 2); }); -test('conflicting provisional live fallbacks still close and yield to HTTP', async () => { +test('snapshot live cannot supersede an incomparable sibling index owner', async () => { const fetches = []; let liveObserver; let unsubscribeCount = 0; @@ -1902,7 +1908,7 @@ test('conflicting provisional live fallbacks still close and yield to HTTP', asy rows: [{ id: 'todo-other', title: 'other provisional membership' }], recordScope: 'record:other', indexesComparable: false, - live: { supported: false, reset: true } + live: { mode: "snapshot", reset: true } }), 'live' ); @@ -1915,30 +1921,103 @@ test('conflicting provisional live fallbacks still close and yield to HTTP', asy rows: [{ id: 'todo-live', title: 'conflicting provisional membership' }], recordScope: 'record:live', indexesComparable: false, - live: { supported: false, reset: true } + live: { mode: "snapshot", reset: true } }) ); - assert.equal(watch.get().live, 'off'); - assert.equal(unsubscribeCount, 1); + assert.equal(watch.get().live, 'active'); + assert.equal(unsubscribeCount, 0); - await Promise.resolve(); - fetches[0].resolve( - wireFrame({ - rows: [{ id: 'todo-http', title: 'authoritative HTTP membership' }], - recordScope: 'record:http', - indexesComparable: false - }) - ); - await new Promise((resolve) => setImmediate(resolve)); + assert.equal(fetches.length, 0, 'a complete live result needs no HTTP fallback'); + assert.deepEqual(replica.read(Todos, {}).data.todos, [ + { id: 'todo-other', title: 'other provisional membership' } + ]); + liveObserver.next(wireFrame({ + operation: 'live:todos', rows: [], indexesComparable: false, + live: { mode: 'snapshot', reset: true } + })); assert.deepEqual(replica.read(Todos, {}).data.todos, [ - { id: 'todo-http', title: 'authoritative HTTP membership' } + { id: 'todo-other', title: 'other provisional membership' } ]); + assert.equal(watch.get().live, 'active'); + assert.equal(unsubscribeCount, 0); watch.destroy(); assert.equal(unsubscribeCount, 1); }); -test('completed supported live streams relinquish ownership and fall back to HTTP', async () => { +test('snapshot live replaces SSR membership, updates and removes rows, and fences disposal', () => { + 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(Todos, {}, wireFrame({ + rows: [{ id: 'todo-1', title: 'SSR title' }], indexesComparable: false + }), 'ssr'); + const watch = replica.watch(Todos, {}, { live: true }); + observer.next(wireFrame({ + operation: 'live:todos', rows: [{ id: 'todo-1', title: 'updated title' }], + revision: '2', indexesComparable: false, + live: { mode: 'snapshot', reset: true } + })); + assert.equal(replica.read(Todos, {}).data.todos[0].title, 'updated title'); + observer.next(wireFrame({ + operation: 'live:todos', rows: [], indexesComparable: false, + live: { mode: 'snapshot', reset: true } + })); + assert.deepEqual(replica.read(Todos, {}).data.todos, []); + assert.equal(watch.get().live, 'active'); + watch.destroy(); + observer.next(wireFrame({ + operation: 'live:todos', rows: [{ id: 'late', title: 'disposed callback' }], + indexesComparable: false, live: { mode: 'snapshot', reset: true } + })); + assert.deepEqual(replica.read(Todos, {}).data.todos, []); +}); + +test('snapshot authorization changes discard the old stream and reconnect without cursors', async () => { + const subscriptions = []; + const requests = []; + const replica = createDistributedReplica({ transport: { + fetch(request) { + return new Promise((resolve) => requests.push({ request, resolve })); + }, + subscribe(request, observer) { + subscriptions.push({ request, observer }); + return () => {}; + } + } }); + write(replica, { rows: [{ id: 'alice', title: 'private alice' }], indexesComparable: false }, 'ssr'); + const watch = replica.watch(Todos, {}, { live: true }); + subscriptions[0].observer.next(wireFrame({ + operation: 'live:todos', rows: [{ id: 'alice', title: 'private alice' }], + indexesComparable: false, live: { mode: 'snapshot', reset: true } + })); + replica.invalidateAuthorization(); + await Promise.resolve(); + assert.equal(requests.length, 1); + subscriptions[0].observer.next(wireFrame({ + operation: 'live:todos', rows: [{ id: 'alice', title: 'late private alice' }], + indexesComparable: false, live: { mode: 'snapshot', reset: true } + })); + assert.equal(watch.get().complete, false); + requests[0].resolve(wireFrame({ + cacheScope: 'cache:b', rows: [{ id: 'bob', title: 'private bob' }], + recordScope: 'record:b', indexesComparable: false + })); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(subscriptions.length, 2); + assert.equal(subscriptions[1].request.resume, undefined); + assert.equal(watch.get().data.todos[0].id, 'bob'); + subscriptions[1].observer.next(wireFrame({ + operation: 'live:todos', cacheScope: 'cache:b', rows: [], + indexesComparable: false, live: { mode: 'snapshot', reset: true } + })); + assert.deepEqual(watch.get().data.todos, []); + watch.destroy(); +}); + +test('completed resumable live streams relinquish ownership and fall back to HTTP', async () => { const fetches = []; const subscriptions = []; let unsubscribeCount = 0; @@ -1971,7 +2050,7 @@ test('completed supported live streams relinquish ownership and fall back to HTT position: '2', rows: [{ id: 'todo-live', title: 'supported live membership' }], recordScope: 'record:live', - live: { supported: true, reset: true } + live: { mode: "resumable", reset: true } }) ); assert.deepEqual(replica.read(Todos, {}).data.todos, [ diff --git a/js/tests/replica-result-observation.test.mjs b/js/tests/replica-result-observation.test.mjs index 35e515d1a..40a8030fa 100644 --- a/js/tests/replica-result-observation.test.mjs +++ b/js/tests/replica-result-observation.test.mjs @@ -116,7 +116,7 @@ function frame(revision, value, operation = Query.id) { ...(operation === Query.live.id ? { live: { - supported: true, + mode: 'resumable', reset: false, cursors: [ { diff --git a/js/tests/sveltekit-ssr.test.mjs b/js/tests/sveltekit-ssr.test.mjs index 2bde3a62e..baeef371c 100644 --- a/js/tests/sveltekit-ssr.test.mjs +++ b/js/tests/sveltekit-ssr.test.mjs @@ -113,7 +113,7 @@ const forwardedTodosBoundary = defineDistributedBoundaryOperation( }) ); -function serverHarness() { +function serverHarness(mode = 'resumable') { const calls = []; const server = createDistributedSvelteKitServer({ boundaries: [todosBoundary], @@ -150,6 +150,7 @@ function serverHarness() { ], { cacheScope: `cache:${token}`, + mode, position } ) @@ -163,8 +164,9 @@ function serverHarness() { }; } -test('static @load SSR is request-isolated and hydration avoids a duplicate first fetch', async () => { - const harness = serverHarness(); +for (const mode of ['resumable', 'snapshot']) { +test(`static @load SSR is isolated and hydration avoids a duplicate fetch (${mode})`, async () => { + const harness = serverHarness(mode); const [alice, bob] = await Promise.all([ harness.server.load(harness.event('alice')), harness.server.load(harness.event('bob')) @@ -212,7 +214,7 @@ test('static @load SSR is request-isolated and hydration avoids a duplicate firs payload: todoFrame( TodosArtifact, [{ id: 'todo-alice', title: 'alice:live', status: 'open' }], - { cacheScope: 'cache:alice', position: '2', source: 'live' } + { cacheScope: 'cache:alice', position: '2', source: 'live', mode } ) }); await flushMicrotasks(); @@ -230,6 +232,7 @@ test('static @load SSR is request-isolated and hydration avoids a duplicate firs assert.equal(socket.closed, true); client.destroy(); }); +} test('SSR only awaits parent data for forwarded-prop bindings', async () => { const harness = serverHarness(); diff --git a/src/gateway/delivery/live.rs b/src/gateway/delivery/live.rs index 9ada4bd5a..f1ed4b196 100644 --- a/src/gateway/delivery/live.rs +++ b/src/gateway/delivery/live.rs @@ -181,7 +181,7 @@ impl LiveFrame { let canonical = canonical_json(&payload).unwrap_or(bytes); let protocol = &payload["extensions"]["distributed"]; let cursors = &protocol["live"]["cursors"]; - let cursor = if protocol["live"]["supported"] == true + let cursor = if protocol["live"]["mode"] == "resumable" && protocol["snapshot"]["indexesComparable"] == true && cursors.as_array().is_some_and(|cursors| { !cursors.is_empty() diff --git a/src/gateway/native/live.rs b/src/gateway/native/live.rs index b4e29f6e7..34ab796a7 100644 --- a/src/gateway/native/live.rs +++ b/src/gateway/native/live.rs @@ -499,7 +499,7 @@ mod tests { serde_json::json!({"data":{"rows":[{"title":"unchanged"}]},"extensions":{"distributed":{ "protocolVersion":1,"schemaHash":"schema","authorizationGeneration":"policy","cacheScope":"alice","operation":"operation", "snapshot":{"recordsComplete":true,"indexesComparable":true,"records":[],"indexes":[{"projection":"rows","scopeToken":"scope","position":position.to_string()}],"observations":[proof]}, - "live":{"supported":true,"reset":false,"cursors":[{"projection":"rows","position":position.to_string(),"token":format!("token-{position}")}]} + "live":{"mode":"resumable","reset":false,"cursors":[{"projection":"rows","position":position.to_string(),"token":format!("token-{position}")}]} }}}) } struct DropCount(Arc); diff --git a/src/graphql/protocol/mod.rs b/src/graphql/protocol/mod.rs index 21d99c9e0..03f27d159 100644 --- a/src/graphql/protocol/mod.rs +++ b/src/graphql/protocol/mod.rs @@ -25,7 +25,7 @@ pub(crate) use token::{ pub(crate) use types::{ DistributedCommandConsistency, DistributedCommandMetadata, DistributedCommandState, DistributedEnvelopeV1, DistributedIndexRevision, DistributedLiveCursor, - DistributedLiveMetadata, DistributedProjectionDisposition, DistributedProjectionExpectation, - DistributedProjectionObservation, DistributedQuerySnapshot, DistributedRecordRevision, - DistributedTrustedPreset, RequestedLiveResume, + DistributedLiveMetadata, DistributedLiveMode, DistributedProjectionDisposition, + DistributedProjectionExpectation, DistributedProjectionObservation, DistributedQuerySnapshot, + DistributedRecordRevision, DistributedTrustedPreset, RequestedLiveResume, }; diff --git a/src/graphql/protocol/tests.rs b/src/graphql/protocol/tests.rs index abef4469f..7ad481d63 100644 --- a/src/graphql/protocol/tests.rs +++ b/src/graphql/protocol/tests.rs @@ -519,7 +519,7 @@ fn stream_frames_are_immutable_fifo_and_do_not_bleed_forward() { .record_query_metadata( first.clone(), Some(DistributedLiveMetadata { - supported: true, + mode: DistributedLiveMode::Resumable, reset: true, cursors: vec![first.indexes[0].resume.clone().unwrap()], }), @@ -529,7 +529,7 @@ fn stream_frames_are_immutable_fifo_and_do_not_bleed_forward() { .record_query_metadata( second.clone(), Some(DistributedLiveMetadata { - supported: true, + mode: DistributedLiveMode::Resumable, reset: false, cursors: vec![second.indexes[0].resume.clone().unwrap()], }), diff --git a/src/graphql/protocol/types.rs b/src/graphql/protocol/types.rs index 64f4b7291..63f7c6293 100644 --- a/src/graphql/protocol/types.rs +++ b/src/graphql/protocol/types.rs @@ -170,11 +170,19 @@ impl DistributedQuerySnapshot { } } -/// Per-frame resumability decision for a live operation. +/// Authorized snapshot delivery does not imply partition-wide resumability. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum DistributedLiveMode { + Snapshot, + Resumable, +} + +/// Per-frame delivery mode for a live operation. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct DistributedLiveMetadata { - pub(crate) supported: bool, + pub(crate) mode: DistributedLiveMode, pub(crate) reset: bool, pub(crate) cursors: Vec, } diff --git a/src/graphql/query_protocol.rs b/src/graphql/query_protocol.rs index 1ca252fd3..bb31a5e46 100644 --- a/src/graphql/query_protocol.rs +++ b/src/graphql/query_protocol.rs @@ -18,9 +18,9 @@ use super::engine::GraphqlPool; #[cfg(any(feature = "sqlite", feature = "postgres"))] use super::execute; use super::protocol::{ - DistributedIndexRevision, DistributedLiveMetadata, DistributedQuerySnapshot, - DistributedRecordRevision, ProtocolResponseAccumulator, RequestedLiveResume, - MAX_LIVE_RESUME_CURSORS, + DistributedIndexRevision, DistributedLiveMetadata, DistributedLiveMode, + DistributedQuerySnapshot, DistributedRecordRevision, ProtocolResponseAccumulator, + RequestedLiveResume, MAX_LIVE_RESUME_CURSORS, }; use super::surface::{Surface, SurfaceProjectionOwner, SurfaceRowPolicy}; use crate::projection::placement::ProjectionBindingState; @@ -167,7 +167,7 @@ impl QueryProtocolRuntime { /// the plan. Join targets such as `auth_users` are often populated by /// integration handlers without a unit-resume owner; they still participate /// in dirty matching via `tables_touched`, but must not force - /// `live.supported = false` for an otherwise resumable root model (e.g. + /// snapshot-only live delivery for an otherwise resumable root model (e.g. /// `chat_messages` with a nested `author` selection). pub(crate) fn index_plan(&self, role_surface: &Surface, tables: &[String]) -> QueryIndexPlan { if tables.is_empty() { @@ -787,13 +787,23 @@ where } None => (None, Vec::new()), }; - let snapshot = wire_query_snapshot( + let mut snapshot = wire_query_snapshot( accumulator, prepared, record_metadata, partitions, live_changes, )?; + if live + .as_ref() + .is_some_and(|metadata| metadata.mode == DistributedLiveMode::Snapshot) + { + // A safe query vector can still exceed the live cursor budget. Snapshot + // delivery has one contract regardless of why resumability is unavailable. + snapshot.indexes_comparable = false; + snapshot.indexes.clear(); + snapshot.observations.clear(); + } Ok(ProtocolQueryExecution { value: executed.value, snapshot, @@ -894,7 +904,7 @@ where { return Ok(PreparedLiveMetadata { metadata: DistributedLiveMetadata { - supported: false, + mode: DistributedLiveMode::Snapshot, reset: true, cursors: Vec::new(), }, @@ -1094,7 +1104,7 @@ where Ok(PreparedLiveMetadata { metadata: DistributedLiveMetadata { - supported: true, + mode: DistributedLiveMode::Resumable, reset, cursors: current, }, diff --git a/src/graphql/subscribe.rs b/src/graphql/subscribe.rs index e87a71b20..c4a9c9f0b 100644 --- a/src/graphql/subscribe.rs +++ b/src/graphql/subscribe.rs @@ -290,7 +290,7 @@ async fn execute_list( let next_live_resume = executed .live .as_ref() - .filter(|live| live.supported) + .filter(|live| live.mode == super::protocol::DistributedLiveMode::Resumable) .map(|live| RequestedLiveResume::Cursors(live.cursors.clone())) .unwrap_or(RequestedLiveResume::Absent); Ok(ExecutedLiveQuery { diff --git a/tests/e2e-ui/crates/projections/src/chat.rs b/tests/e2e-ui/crates/projections/src/chat.rs index 97198e94c..0ea8d6582 100644 --- a/tests/e2e-ui/crates/projections/src/chat.rs +++ b/tests/e2e-ui/crates/projections/src/chat.rs @@ -1,15 +1,16 @@ //! Chat: mutation + projection. //! //! Unit partition so the lobby `@live` subscription can advertise resumable -//! index evidence (`live.supported = true`). Room isolation stays on the +//! index evidence (`live.mode = "resumable"`). Room isolation stays on the //! GraphQL document (`where: { room_id: { _eq: "lobby" } }`). Expression //! partitions are correct for multi-room worker sharding, but they make live -//! indexes incomparable and the client falls back to Idle. +//! indexes incomparable, so live delivery uses authorized snapshots without +//! partition-wide resume cursors. +use chat_domain::ChatMessagePostedDomainEvent; use distributed::mutation_file; use distributed::projection::lower::{DirectCandidate, ProjectionDescriptor}; use distributed::Mutation; -use chat_domain::ChatMessagePostedDomainEvent; use e2e_readmodels::ChatMessages; /// Mutation: complete-row upsert for chat messages. @@ -53,11 +54,18 @@ mod tests { occurrence.descriptor(), &ChatMessagePostedDomainEvent::descriptor() ); - let lowered = CHAT_MESSAGES.server_executor().unwrap().plan(occurrence).unwrap(); + let lowered = CHAT_MESSAGES + .server_executor() + .unwrap() + .plan(occurrence) + .unwrap(); let TableMutation::UpsertRow(row) = &lowered.write_plan.mutations[0] else { panic!("expected upsert"); }; - assert_eq!(row.values.get("body"), Some(&RowValue::String("hello".into()))); + assert_eq!( + row.values.get("body"), + Some(&RowValue::String("hello".into())) + ); assert_eq!( SaveChatMessage().program().operations()[0].kind(), MutationKind::Upsert diff --git a/tests/edge_query_delivery.rs b/tests/edge_query_delivery.rs index 697d3cd3a..91ea24131 100644 --- a/tests/edge_query_delivery.rs +++ b/tests/edge_query_delivery.rs @@ -388,7 +388,7 @@ fn live_scope_replay_and_proof_sensitive_frames() { ) .is_err()); let mut payload: serde_json::Value = serde_json::from_slice(&snapshot(&admitted).body).unwrap(); - payload["extensions"]["distributed"]["live"] = json!({"supported":true,"cursors":[{"projection":"todos","position":"2","token":"cursor"}]}); + payload["extensions"]["distributed"]["live"] = json!({"mode":"resumable","cursors":[{"projection":"todos","position":"2","token":"cursor"}]}); let first = LiveFrame::from_origin(&admitted, payload.clone(), None, 4096).unwrap(); assert!( first.same_frame(&LiveFrame::from_origin(&admitted, payload.clone(), None, 4096).unwrap()) @@ -406,7 +406,7 @@ fn live_scope_replay_and_proof_sensitive_frames() { !first.same_cursor(&changed), "same projector cursor does not cover external writes" ); - payload["extensions"]["distributed"]["live"]["supported"] = false.into(); + payload["extensions"]["distributed"]["live"]["mode"] = "snapshot".into(); let unsupported = LiveFrame::from_origin(&admitted, payload, None, 4096).unwrap(); assert!(!unsupported.same_cursor(&unsupported)); let mut stronger = context(); diff --git a/tests/graphql_query_protocol/main.rs b/tests/graphql_query_protocol/main.rs index 325e4a294..85378dafc 100644 --- a/tests/graphql_query_protocol/main.rs +++ b/tests/graphql_query_protocol/main.rs @@ -206,12 +206,21 @@ async fn project_item_with_id( } async fn delete_item(repository: &SqliteRepository, bus: &InMemoryBus, sequence: u64) { + delete_item_with_id(repository, bus, sequence, ROW_ID).await; +} + +async fn delete_item_with_id( + repository: &SqliteRepository, + bus: &InMemoryBus, + sequence: u64, + id: &str, +) { bus.publish_message( Message::new( FACT_NAME, MessageKind::Event, serde_json::to_vec(&json!({ - "id": ROW_ID, + "id": id, "title": "deleted row", "delete": true })) @@ -404,7 +413,7 @@ fn assert_live_frame( assert_eq!(snapshot["indexes"][0]["position"], expected_position); let live = &distributed["live"]; - assert_eq!(live["supported"], true, "{response}"); + assert_eq!(live["mode"], "resumable", "{response}"); assert_eq!(live["reset"], expected_reset, "{response}"); assert_eq!(live["cursors"].as_array().map(Vec::len), Some(1)); assert_eq!(live["cursors"][0]["projection"], PROJECTOR_NAME); @@ -957,7 +966,7 @@ async fn live_subscription_replays_delete_tombstone_and_observation() { "{deleted}" ); let distributed = distributed_envelope(&deleted); - assert_eq!(distributed["live"]["supported"], true); + assert_eq!(distributed["live"]["mode"], "resumable"); assert_eq!(distributed["live"]["reset"], false); assert_eq!(distributed["live"]["cursors"][0]["position"], "2"); assert_eq!(distributed["snapshot"]["indexes"][0]["position"], "2"); @@ -1124,7 +1133,7 @@ async fn row_filtered_surface_never_exposes_partition_wide_live_activity() { assert_eq!(records[0]["model"], "CausalQueryView"); assert_eq!(records[0]["tombstone"], false); assert_opaque_token(&records[0]["scopeToken"], "record-revision"); - assert_eq!(envelope["live"]["supported"], false, "{initial}"); + assert_eq!(envelope["live"]["mode"], "snapshot", "{initial}"); assert_eq!(envelope["live"]["reset"], true, "{initial}"); assert_eq!(envelope["live"]["cursors"], json!([]), "{initial}"); @@ -1142,6 +1151,50 @@ async fn row_filtered_surface_never_exposes_partition_wide_live_activity() { .is_err(), "a denied-row commit must not leak a cursor, causation, tombstone, or activity frame" ); + + delete_item_with_id( + &fixture.repository, + &fixture.bus, + 3, + "other-principal-private-row", + ) + .await; + assert!( + tokio::time::timeout(Duration::from_millis(350), stream.next()) + .await + .is_err(), + "a denied-row deletion must not leak activity or a tombstone" + ); + project_item(&fixture.repository, &fixture.bus, 4, "visible change").await; + let changed = next_wire_frame(&mut stream).await; + assert_eq!( + changed["data"]["causal_query_views"], + json!([{ "title": "visible change" }]) + ); + let metadata = distributed_envelope(&changed); + assert_eq!( + metadata["live"], + json!({"mode":"snapshot","reset":true,"cursors":[]}) + ); + assert_eq!(metadata["snapshot"]["indexes"], json!([])); + assert_eq!(metadata["snapshot"]["observations"], json!([])); + + drop(stream); + let mut stream = engine.execute_stream(&user_session(), Request::new(LIVE_SUBSCRIPTION)); + let reconnected = next_wire_frame(&mut stream).await; + assert_eq!(reconnected["data"], changed["data"]); + assert_eq!(distributed_envelope(&reconnected)["live"], metadata["live"]); + + delete_item(&fixture.repository, &fixture.bus, 5).await; + let deleted = next_wire_frame(&mut stream).await; + assert_eq!(deleted["data"]["causal_query_views"], json!([])); + let metadata = distributed_envelope(&deleted); + assert_eq!( + metadata["live"], + json!({"mode":"snapshot","reset":true,"cursors":[]}) + ); + assert_eq!(metadata["snapshot"]["records"], json!([])); + assert_eq!(metadata["snapshot"]["observations"], json!([])); } async fn query_over_http_and_graphql_ws( diff --git a/tests/graphql_query_protocol_postgres/main.rs b/tests/graphql_query_protocol_postgres/main.rs index 8b76bda23..da9def676 100644 --- a/tests/graphql_query_protocol_postgres/main.rs +++ b/tests/graphql_query_protocol_postgres/main.rs @@ -236,7 +236,7 @@ async fn postgres_emits_exact_revisions_and_accepts_an_exact_live_resume() { engine.execute_stream(&user_session(), Request::new(LIVE_SUBSCRIPTION)); let initial = next_wire_frame(&mut initial_stream).await; let initial_live = &distributed_envelope(&initial)["live"]; - assert_eq!(initial_live["supported"], true, "{initial}"); + assert_eq!(initial_live["mode"], "resumable", "{initial}"); assert_eq!(initial_live["reset"], false, "{initial}"); let cursors = initial_live["cursors"].clone(); assert_eq!(cursors[0]["projection"], PROJECTOR_NAME); @@ -247,7 +247,7 @@ async fn postgres_emits_exact_revisions_and_accepts_an_exact_live_resume() { let mut resumed_stream = engine.execute_stream(&user_session(), request_with_resume(cursors)); let resumed = next_wire_frame(&mut resumed_stream).await; let resumed_live = &distributed_envelope(&resumed)["live"]; - assert_eq!(resumed_live["supported"], true, "{resumed}"); + assert_eq!(resumed_live["mode"], "resumable", "{resumed}"); assert_eq!(resumed_live["reset"], false, "{resumed}"); assert_eq!(resumed_live["cursors"][0]["position"], "1"); }