From 0084704ed2b60898d167af35be4bc359eeae25ae Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Mon, 22 Jun 2026 16:01:37 +0100 Subject: [PATCH] fix(webapp): gate runTableV2 on native realtime instead of merging Electric shapes Electric realtime shapes are bound to a single table, so a task_run_v2 run was invisible to realtime subscriptions. The previous approach merged two Electric shapes per tag/batch feed under a composite cursor, which doubled Electric long-poll connections for those feeds. Electric is being retired in favor of the native realtime backend, which is table-agnostic and already observes both run tables, so that merge is throwaway. Drop the Electric dual-shape merge (revert realtimeClient to its single-table form, remove the merge module) and instead gate runTableV2 on the native backend: a run only routes to task_run_v2 when the deployment has native realtime enabled and the org's realtimeBackend flag is native. This keeps v2 runs realtime-observable without touching Electric, and the gate auto-satisfies once Electric is removed and native is the default. The idempotency pre-gate claim inherits the same gate. --- .../realtime/electricShapeMerge.server.ts | 167 ---------- .../app/services/realtimeClient.server.ts | 288 +----------------- apps/webapp/test/electricShapeMerge.test.ts | 201 ------------ apps/webapp/test/realtimeClient.test.ts | 7 +- 4 files changed, 5 insertions(+), 658 deletions(-) delete mode 100644 apps/webapp/app/services/realtime/electricShapeMerge.server.ts delete mode 100644 apps/webapp/test/electricShapeMerge.test.ts diff --git a/apps/webapp/app/services/realtime/electricShapeMerge.server.ts b/apps/webapp/app/services/realtime/electricShapeMerge.server.ts deleted file mode 100644 index cfe494e4a..000000000 --- a/apps/webapp/app/services/realtime/electricShapeMerge.server.ts +++ /dev/null @@ -1,167 +0,0 @@ -/** - * Pure helpers for merging TWO upstream Electric shapes (one per physical run - * table — `TaskRun` and `task_run_v2`) into a single shape the realtime client - * consumes. A tag-list or batch feed matches runs in both tables during/after a - * `runTableV2` cutover, but an Electric shape is bound to one table, so the - * proxy fans out to two shapes and presents one composite continuation - * (`handle` / `offset` / `cursor`) that the client round-trips opaquely. - * - * Kept dependency-free (no DB/Redis/fetch) so the merge logic is unit-testable. - */ - -// Separator packing the two per-table continuation values into one opaque -// token. Electric's handle/offset/cursor values are alphanumeric plus `_`/`-` -// (UUID-ish handles, `_` offsets, numeric cursors) and never contain -// `~`, so it is collision-free for this charset. -export const COMPOSITE_SEP = "~"; - -export const UP_TO_DATE_MESSAGE = { headers: { control: "up-to-date" } } as const; -export const MUST_REFETCH_MESSAGE = { headers: { control: "must-refetch" } } as const; - -/** A parsed per-table shape response: continuation headers + the change rows. */ -export type ParsedShape = { - status: number; - handle?: string; - offset?: string; - cursor?: string; - schema?: string; - /** Change messages only (control messages stripped). */ - changes: unknown[]; - upToDate: boolean; - mustRefetch: boolean; -}; - -/** The prior per-table continuation the client sent (used when a shape is left - * un-polled because the other returned first). */ -export type PriorContinuation = { - handleA?: string; - offsetA: string; - cursorA?: string; - handleB?: string; - offsetB: string; - cursorB?: string; -}; - -export type MergedShape = - | { mustRefetch: true } - | { - mustRefetch: false; - changes: unknown[]; - handle: string; - offset: string; - cursor?: string; - schema?: string; - }; - -/** - * Split a composite "~" value back into its per-table parts. A value with - * no separator (or null/empty) means the client hasn't been handed a composite - * yet (the initial request before any shape exists) -> both undefined. - */ -export function decodeCompositePart(value: string | null | undefined): { - a: string | undefined; - b: string | undefined; -} { - if (!value) return { a: undefined, b: undefined }; - const idx = value.indexOf(COMPOSITE_SEP); - if (idx === -1) return { a: undefined, b: undefined }; - return { - a: value.slice(0, idx) || undefined, - b: value.slice(idx + COMPOSITE_SEP.length) || undefined, - }; -} - -/** - * The offset is never absent — Electric uses "-1" for the initial request — so - * a bare value applies to BOTH shapes (initial), and a composite splits. - */ -export function decodeCompositeOffset(offset: string): { a: string; b: string } { - const idx = offset.indexOf(COMPOSITE_SEP); - if (idx === -1) return { a: offset, b: offset }; - return { a: offset.slice(0, idx), b: offset.slice(idx + COMPOSITE_SEP.length) }; -} - -export function encodeComposite(a: string, b: string): string { - return `${a}${COMPOSITE_SEP}${b}`; -} - -/** Parse the raw body + headers of one upstream shape response. */ -export function parseShapeMessages( - status: number, - headers: { - handle?: string; - offset?: string; - cursor?: string; - schema?: string; - }, - bodyText: string -): ParsedShape { - const base = { status, ...headers }; - if (status >= 400) { - return { ...base, changes: [], upToDate: false, mustRefetch: status === 409 }; - } - let parsed: unknown; - try { - parsed = bodyText.trim() ? JSON.parse(bodyText) : []; - } catch { - // Unparseable body — safest is to make the client refetch the shape. - return { ...base, changes: [], upToDate: false, mustRefetch: true }; - } - if (!Array.isArray(parsed)) { - return { ...base, changes: [], upToDate: false, mustRefetch: true }; - } - const messages = parsed as Array<{ headers?: { control?: string } }>; - const changes = messages.filter((m) => !m?.headers?.control); - const mustRefetch = messages.some((m) => m?.headers?.control === "must-refetch"); - const upToDate = messages.some((m) => m?.headers?.control === "up-to-date"); - return { ...base, changes, upToDate, mustRefetch }; -} - -/** - * Merge two parsed per-table shapes into one composite payload. If either shape - * needs a refetch (409 / must-refetch / unparseable), the whole composite is - * reset. Otherwise the change rows are concatenated (the client merges by key, - * so order across tables doesn't matter) and the continuation values are packed - * per table, falling back to the client's prior value for a shape that wasn't - * re-polled this round. - */ -export function mergeParsedShapes( - a: ParsedShape, - b: ParsedShape, - prior: PriorContinuation -): MergedShape { - if (a.mustRefetch || b.mustRefetch || a.status >= 400 || b.status >= 400) { - return { mustRefetch: true }; - } - const cursorA = a.cursor ?? prior.cursorA; - const cursorB = b.cursor ?? prior.cursorB; - const cursor = - cursorA !== undefined || cursorB !== undefined - ? encodeComposite(cursorA ?? "", cursorB ?? "") - : undefined; - return { - mustRefetch: false, - changes: [...a.changes, ...b.changes], - handle: encodeComposite(a.handle ?? prior.handleA ?? "", b.handle ?? prior.handleB ?? ""), - offset: encodeComposite(a.offset ?? prior.offsetA, b.offset ?? prior.offsetB), - cursor, - schema: a.schema ?? b.schema, - }; -} - -/** A synthetic "no change this round" result for a shape left un-polled because - * the other returned changes first; carries its prior continuation forward. */ -export function unpolledShape( - which: "a" | "b", - prior: PriorContinuation -): ParsedShape { - return { - status: 200, - handle: which === "a" ? prior.handleA : prior.handleB, - offset: which === "a" ? prior.offsetA : prior.offsetB, - cursor: which === "a" ? prior.cursorA : prior.cursorB, - changes: [], - upToDate: true, - mustRefetch: false, - }; -} diff --git a/apps/webapp/app/services/realtimeClient.server.ts b/apps/webapp/app/services/realtimeClient.server.ts index d2e68c64e..12b93f199 100644 --- a/apps/webapp/app/services/realtimeClient.server.ts +++ b/apps/webapp/app/services/realtimeClient.server.ts @@ -1,18 +1,6 @@ import { json } from "@remix-run/server-runtime"; import { tryCatch } from "@trigger.dev/core/utils"; -import { isKsuidId, safeParseNaturalLanguageDurationAgo } from "@trigger.dev/core/v3/isomorphic"; -import { - decodeCompositeOffset, - decodeCompositePart, - mergeParsedShapes, - MUST_REFETCH_MESSAGE, - parseShapeMessages, - unpolledShape, - UP_TO_DATE_MESSAGE, - type MergedShape, - type ParsedShape, - type PriorContinuation, -} from "./realtime/electricShapeMerge.server"; +import { safeParseNaturalLanguageDurationAgo } from "@trigger.dev/core/v3/isomorphic"; import { Callback, Result } from "ioredis"; import { randomUUID } from "node:crypto"; import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis.server"; @@ -61,11 +49,6 @@ const DEFAULT_ELECTRIC_COLUMNS = [ const RESERVED_COLUMNS = ["id", "taskIdentifier", "friendlyId", "status", "createdAt"]; const RESERVED_SEARCH_PARAMS = ["createdAt", "tags", "skipColumns"]; -// The two physical run tables a realtime shape can target. A run lives in -// exactly one, keyed by id format (ksuid -> task_run_v2, cuid -> TaskRun). -const TASK_RUN_TABLE = 'public."TaskRun"'; -const TASK_RUN_V2_TABLE = 'public."task_run_v2"'; - export type RealtimeClientOptions = { electricOrigin: string | string[]; redis: RedisWithClusterOptions; @@ -135,15 +118,10 @@ export class RealtimeClient { clientVersion?: string, signal?: AbortSignal ) { - // Route the shape to the physical table the run lives in: a v2 run's id is - // a KSUID (task_run_v2), a legacy run's a cuid (TaskRun). The run was - // already resolved by the route, so this id is authoritative. - const table = isKsuidId(runId) ? TASK_RUN_V2_TABLE : TASK_RUN_TABLE; return this.#streamRunsWhere( url, environment, `id='${runId}'`, - table, apiVersion, requestOptions, clientVersion, @@ -167,7 +145,7 @@ export class RealtimeClient { const whereClause = whereClauses.join(" AND "); - return this.#streamRunsAcrossTables( + return this.#streamRunsWhere( url, environment, whereClause, @@ -201,7 +179,7 @@ export class RealtimeClient { const whereClause = whereClauses.join(" AND "); - const response = await this.#streamRunsAcrossTables( + const response = await this.#streamRunsWhere( url, environment, whereClause, @@ -300,7 +278,6 @@ export class RealtimeClient { url: URL | string, environment: RealtimeEnvironment, whereClause: string, - table: string, apiVersion: API_VERSIONS, requestOptions?: RealtimeRequestOptions, clientVersion?: string, @@ -310,7 +287,6 @@ export class RealtimeClient { url, environment, whereClause, - table, requestOptions, clientVersion ); @@ -324,266 +300,10 @@ export class RealtimeClient { ); } - // Stream a feed that spans BOTH physical run tables (the tag-list and batch - // feeds) by running two upstream Electric shapes — public."TaskRun" and - // public."task_run_v2" — under a single composite continuation the client - // round-trips opaquely. A run lives in exactly one table, so the union of the - // two shapes is the full feed; the client merges by row key and never learns - // there are two shapes. See electricShapeMerge.server.ts for the pure logic. - // - // Cost: this opens TWO upstream Electric long-polls per tag/batch - // subscription (vs one for a single-table feed), so these feeds use ~2x - // Electric connections while an org has runs across both tables. Single-run - // subscriptions are unaffected — one shape, routed to the run's table by id - // format. - async #streamRunsAcrossTables( - url: URL | string, - environment: RealtimeEnvironment, - whereClause: string, - apiVersion: API_VERSIONS, - requestOptions?: RealtimeRequestOptions, - clientVersion?: string, - signal?: AbortSignal - ): Promise { - const $url = new URL(url.toString()); - const isLive = isLiveRequestUrl($url); - const incomingHandle = extractShapeId($url); - const incomingOffset = $url.searchParams.get("offset") ?? "-1"; - const incomingCursor = $url.searchParams.get("cursor"); - - const handles = decodeCompositePart(incomingHandle); - const offsets = decodeCompositeOffset(incomingOffset); - const cursors = decodeCompositePart(incomingCursor); - - const prior: PriorContinuation = { - handleA: handles.a, - offsetA: offsets.a, - cursorA: cursors.a, - handleB: handles.b, - offsetB: offsets.b, - cursorB: cursors.b, - }; - - const urlA = this.#constructMergeShapeUrl( - $url, - environment, - whereClause, - TASK_RUN_TABLE, - { handle: handles.a, offset: offsets.a, cursor: cursors.a }, - requestOptions, - clientVersion - ); - const urlB = this.#constructMergeShapeUrl( - $url, - environment, - whereClause, - TASK_RUN_V2_TABLE, - { handle: handles.b, offset: offsets.b, cursor: cursors.b }, - requestOptions, - clientVersion - ); - - // One concurrency slot for the composite live request: it maps to a single - // client request even though we fan out to two upstream long-polls. - let requestId: string | undefined; - if (isLive && incomingHandle) { - const concurrencyLimit = await this.cachedLimitProvider.getCachedLimit( - environment.organizationId, - 100_000 - ); - if (!concurrencyLimit) { - logger.error("Failed to get concurrency limit", { - organizationId: environment.organizationId, - }); - return json({ error: "Failed to get concurrency limit" }, { status: 500 }); - } - requestId = randomUUID(); - if (!(await this.#incrementAndCheck(environment.id, requestId, concurrencyLimit))) { - return json({ error: "Too many concurrent requests" }, { status: 429 }); - } - } - - try { - const merged = await this.#raceAndMergeShapes(urlA, urlB, isLive, prior, signal); - return this.#buildMergeResponse(merged, isLive, apiVersion, clientVersion); - } finally { - if (requestId) { - await this.#decrementConcurrency(environment.id, requestId); - } - } - } - - // Build the per-table Electric URL, replacing the composite continuation the - // client sent with this table's decoded part. - #constructMergeShapeUrl( - baseUrl: URL, - environment: RealtimeEnvironment, - whereClause: string, - table: string, - perTable: { handle?: string; offset: string; cursor?: string }, - requestOptions?: RealtimeRequestOptions, - clientVersion?: string - ): URL { - const electricUrl = this.#constructRunsElectricUrl( - baseUrl, - environment, - whereClause, - table, - requestOptions, - clientVersion - ); - // Upstream always speaks current Electric (handle, not shape_id). - electricUrl.searchParams.delete("shape_id"); - if (perTable.handle !== undefined) { - electricUrl.searchParams.set("handle", perTable.handle); - } else { - electricUrl.searchParams.delete("handle"); - } - electricUrl.searchParams.set("offset", perTable.offset); - if (perTable.cursor !== undefined) { - electricUrl.searchParams.set("cursor", perTable.cursor); - } else { - electricUrl.searchParams.delete("cursor"); - } - return electricUrl; - } - - // Fetch both shapes. For a live request, return as soon as ONE yields changes - // (or needs a refetch) and carry the other's prior continuation forward — so a - // change on either table isn't delayed by the other's idle long-poll. If the - // first to settle had nothing, wait for the other before responding. - async #raceAndMergeShapes( - urlA: URL, - urlB: URL, - isLive: boolean, - prior: PriorContinuation, - signal?: AbortSignal - ): Promise { - const ctlA = new AbortController(); - const ctlB = new AbortController(); - const link = (ctl: AbortController) => - signal ? AbortSignal.any([signal, ctl.signal]) : ctl.signal; - - let aRes: ParsedShape | undefined; - let bRes: ParsedShape | undefined; - const pA = this.#fetchShape(urlA, link(ctlA)).then((r) => { - aRes = r; - return "a" as const; - }); - const pB = this.#fetchShape(urlB, link(ctlB)).then((r) => { - bRes = r; - return "b" as const; - }); - // A shape we don't end up awaiting (the race loser we abort, or the sibling - // left pending when the catch below rethrows) must not surface as an - // unhandled rejection. Attach detached no-op catches up front; the - // race/await paths still observe the original rejections through their own - // reactions, so this only swallows an otherwise-orphaned rejection. - void pA.catch(() => {}); - void pB.catch(() => {}); - - try { - if (!isLive) { - await Promise.all([pA, pB]); - return mergeParsedShapes(aRes!, bRes!, prior); - } - - const actionable = (r: ParsedShape) => - r.mustRefetch || r.status >= 400 || r.changes.length > 0; - - const first = await Promise.race([pA, pB]); - const firstRes = first === "a" ? aRes! : bRes!; - if (actionable(firstRes)) { - // Got changes/refetch from one shape; abort the other and return - // immediately. Its rejection is already swallowed by the catch attached - // above, so the abort can't surface as an unhandled rejection. - (first === "a" ? ctlB : ctlA).abort(); - return first === "a" - ? mergeParsedShapes(aRes!, unpolledShape("b", prior), prior) - : mergeParsedShapes(unpolledShape("a", prior), bRes!, prior); - } - - // First settled empty (idle timeout) — wait for the other. - await (first === "a" ? pB : pA); - return mergeParsedShapes(aRes!, bRes!, prior); - } catch (error) { - ctlA.abort(); - ctlB.abort(); - throw error; - } - } - - async #fetchShape(electricUrl: URL, signal?: AbortSignal): Promise { - const resp = await longPollingFetch(electricUrl.toString(), { signal }); - const headers = { - handle: - resp.headers.get("electric-handle") ?? resp.headers.get("electric-shape-id") ?? undefined, - offset: - resp.headers.get("electric-offset") ?? - resp.headers.get("electric-chunk-last-offset") ?? - undefined, - cursor: resp.headers.get("electric-cursor") ?? undefined, - schema: resp.headers.get("electric-schema") ?? undefined, - }; - if (resp.status >= 400) { - try { - await resp.body?.cancel(); - } catch {} - return parseShapeMessages(resp.status, headers, ""); - } - const bodyText = await resp.text(); - return parseShapeMessages(resp.status, headers, bodyText); - } - - #buildMergeResponse( - merged: MergedShape, - isLive: boolean, - apiVersion: API_VERSIONS, - clientVersion?: string - ): Response { - const responseHeaders = new Headers(); - responseHeaders.set("content-type", "application/json"); - responseHeaders.set("cache-control", "no-store"); - // Match the native client: expose electric-* headers cross-origin or the - // deployed react-hooks fail with MissingHeadersError. - responseHeaders.set("access-control-allow-origin", "*"); - responseHeaders.set("access-control-expose-headers", "*"); - - if (merged.mustRefetch) { - // Reset the client's shape state; it refetches both tables from scratch. - return new Response(JSON.stringify([MUST_REFETCH_MESSAGE, UP_TO_DATE_MESSAGE]), { - status: 409, - headers: responseHeaders, - }); - } - - if (clientVersion) { - responseHeaders.set("electric-handle", merged.handle); - responseHeaders.set("electric-offset", merged.offset); - } else { - responseHeaders.set("electric-shape-id", merged.handle); - responseHeaders.set("electric-chunk-last-offset", merged.offset); - } - if (isLive) { - // The client requires electric-cursor on every live response (its live - // cache-buster). Fall back to the offset if neither shape provided one. - responseHeaders.set("electric-cursor", merged.cursor ?? merged.offset); - } else if (merged.schema !== undefined) { - // Non-live responses require electric-schema. - responseHeaders.set("electric-schema", merged.schema); - } - - const body = JSON.stringify([...merged.changes, UP_TO_DATE_MESSAGE]); - const finalBody = - apiVersion === CURRENT_API_VERSION ? body : this.#rewriteResponseBodyForNoneApiVersion(body); - return new Response(finalBody, { status: 200, headers: responseHeaders }); - } - #constructRunsElectricUrl( url: URL | string, environment: RealtimeEnvironment, whereClause: string, - table: string, requestOptions?: RealtimeRequestOptions, clientVersion?: string ): URL { @@ -602,7 +322,7 @@ export class RealtimeClient { }); electricUrl.searchParams.set("where", whereClause); - electricUrl.searchParams.set("table", table); + electricUrl.searchParams.set("table", 'public."TaskRun"'); if (!clientVersion) { // If the client version is not provided, that means we're using an older client diff --git a/apps/webapp/test/electricShapeMerge.test.ts b/apps/webapp/test/electricShapeMerge.test.ts deleted file mode 100644 index 7f0bf9e0b..000000000 --- a/apps/webapp/test/electricShapeMerge.test.ts +++ /dev/null @@ -1,201 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { - decodeCompositeOffset, - decodeCompositePart, - encodeComposite, - mergeParsedShapes, - parseShapeMessages, - unpolledShape, - type ParsedShape, - type PriorContinuation, -} from "~/services/realtime/electricShapeMerge.server"; - -const INSERT = { - key: '"public"."TaskRun"/"r1"', - value: { id: "r1" }, - headers: { operation: "insert" }, -}; -const UPDATE = { - key: '"public"."task_run_v2"/"r2"', - value: { id: "r2" }, - headers: { operation: "update" }, -}; - -function shape(overrides: Partial = {}): ParsedShape { - return { - status: 200, - handle: "h", - offset: "o", - cursor: "c", - schema: '{"id":{"type":"text"}}', - changes: [], - upToDate: true, - mustRefetch: false, - ...overrides, - }; -} - -const PRIOR: PriorContinuation = { - handleA: "HA", - offsetA: "OA", - cursorA: "CA", - handleB: "HB", - offsetB: "OB", - cursorB: "CB", -}; - -describe("decodeCompositePart", () => { - it("returns both undefined for null / no separator", () => { - expect(decodeCompositePart(null)).toEqual({ a: undefined, b: undefined }); - expect(decodeCompositePart(undefined)).toEqual({ a: undefined, b: undefined }); - expect(decodeCompositePart("")).toEqual({ a: undefined, b: undefined }); - // A bare value with no separator means "not a composite yet" -> initial. - expect(decodeCompositePart("solo")).toEqual({ a: undefined, b: undefined }); - }); - - it("splits a composite into its two parts", () => { - expect(decodeCompositePart("hA~hB")).toEqual({ a: "hA", b: "hB" }); - }); - - it("treats an empty side as undefined", () => { - expect(decodeCompositePart("hA~")).toEqual({ a: "hA", b: undefined }); - expect(decodeCompositePart("~hB")).toEqual({ a: undefined, b: "hB" }); - }); -}); - -describe("decodeCompositeOffset", () => { - it("applies a bare offset (e.g. the initial -1) to both shapes", () => { - expect(decodeCompositeOffset("-1")).toEqual({ a: "-1", b: "-1" }); - }); - - it("splits a composite offset", () => { - expect(decodeCompositeOffset("26800552_0~26800999_2")).toEqual({ - a: "26800552_0", - b: "26800999_2", - }); - }); - - it("round-trips through encodeComposite", () => { - expect(decodeCompositeOffset(encodeComposite("x_1", "y_2"))).toEqual({ a: "x_1", b: "y_2" }); - }); -}); - -describe("parseShapeMessages", () => { - const headers = { handle: "h", offset: "o", cursor: "c", schema: "s" }; - - it("extracts change rows and the up-to-date flag", () => { - const body = JSON.stringify([INSERT, { headers: { control: "up-to-date" } }]); - const parsed = parseShapeMessages(200, headers, body); - expect(parsed.changes).toEqual([INSERT]); - expect(parsed.upToDate).toBe(true); - expect(parsed.mustRefetch).toBe(false); - }); - - it("treats a bare up-to-date as no changes", () => { - const parsed = parseShapeMessages( - 200, - headers, - JSON.stringify([{ headers: { control: "up-to-date" } }]) - ); - expect(parsed.changes).toEqual([]); - expect(parsed.upToDate).toBe(true); - }); - - it("flags must-refetch from a 409 status", () => { - const parsed = parseShapeMessages(409, headers, ""); - expect(parsed.mustRefetch).toBe(true); - expect(parsed.changes).toEqual([]); - }); - - it("flags must-refetch from a control message", () => { - const body = JSON.stringify([ - { headers: { control: "must-refetch" } }, - { headers: { control: "up-to-date" } }, - ]); - expect(parseShapeMessages(200, headers, body).mustRefetch).toBe(true); - }); - - it("flags must-refetch for an unparseable / non-array body", () => { - expect(parseShapeMessages(200, headers, "not json").mustRefetch).toBe(true); - expect(parseShapeMessages(200, headers, "{}").mustRefetch).toBe(true); - }); - - it("treats an empty body as no changes (not up-to-date)", () => { - const parsed = parseShapeMessages(200, headers, ""); - expect(parsed.changes).toEqual([]); - expect(parsed.upToDate).toBe(false); - expect(parsed.mustRefetch).toBe(false); - }); -}); - -describe("mergeParsedShapes", () => { - it("concatenates change rows from both tables", () => { - const merged = mergeParsedShapes( - shape({ changes: [INSERT], handle: "hA", offset: "oA", cursor: "cA" }), - shape({ changes: [UPDATE], handle: "hB", offset: "oB", cursor: "cB" }), - PRIOR - ); - expect(merged.mustRefetch).toBe(false); - if (merged.mustRefetch) return; - expect(merged.changes).toEqual([INSERT, UPDATE]); - expect(merged.handle).toBe(encodeComposite("hA", "hB")); - expect(merged.offset).toBe(encodeComposite("oA", "oB")); - expect(merged.cursor).toBe(encodeComposite("cA", "cB")); - }); - - it("resets when either shape needs a refetch", () => { - expect(mergeParsedShapes(shape({ mustRefetch: true }), shape(), PRIOR)).toEqual({ - mustRefetch: true, - }); - expect(mergeParsedShapes(shape(), shape({ status: 409 }), PRIOR)).toEqual({ - mustRefetch: true, - }); - }); - - it("falls back to the prior continuation for a shape that returned nothing", () => { - // B was left un-polled (the other table returned changes first). - const merged = mergeParsedShapes( - shape({ changes: [INSERT], handle: "hA2", offset: "oA2", cursor: "cA2" }), - unpolledShape("b", PRIOR), - PRIOR - ); - expect(merged.mustRefetch).toBe(false); - if (merged.mustRefetch) return; - expect(merged.changes).toEqual([INSERT]); - expect(merged.handle).toBe(encodeComposite("hA2", "HB")); - expect(merged.offset).toBe(encodeComposite("oA2", "OB")); - expect(merged.cursor).toBe(encodeComposite("cA2", "CB")); - }); - - it("uses the prior cursor when a returned shape omits it", () => { - const merged = mergeParsedShapes( - shape({ cursor: undefined, handle: "hA", offset: "oA" }), - shape({ cursor: "cB", handle: "hB", offset: "oB" }), - PRIOR - ); - if (merged.mustRefetch) throw new Error("unexpected refetch"); - // a omitted cursor -> prior.cursorA ("CA"); b returned "cB". - expect(merged.cursor).toBe(encodeComposite("CA", "cB")); - }); - - it("omits the cursor entirely when neither shape nor prior has one (initial snapshot)", () => { - const initialPrior: PriorContinuation = { offsetA: "-1", offsetB: "-1" }; - const merged = mergeParsedShapes( - shape({ cursor: undefined, handle: "hA", offset: "oA" }), - shape({ cursor: undefined, handle: "hB", offset: "oB" }), - initialPrior - ); - if (merged.mustRefetch) throw new Error("unexpected refetch"); - expect(merged.cursor).toBeUndefined(); - }); - - it("carries schema from whichever shape supplied it", () => { - const merged = mergeParsedShapes( - shape({ schema: undefined }), - shape({ schema: '{"id":{"type":"text"}}' }), - PRIOR - ); - if (merged.mustRefetch) throw new Error("unexpected refetch"); - expect(merged.schema).toBe('{"id":{"type":"text"}}'); - }); -}); diff --git a/apps/webapp/test/realtimeClient.test.ts b/apps/webapp/test/realtimeClient.test.ts index cdff50e3d..d98213e5b 100644 --- a/apps/webapp/test/realtimeClient.test.ts +++ b/apps/webapp/test/realtimeClient.test.ts @@ -237,13 +237,8 @@ describe.skipIf(process.env.GITHUB_ACTIONS)("RealtimeClient", () => { const chunkOffset = headers["electric-offset"]; expect(response.status).toBe(200); - // The tag/list feed spans both physical run tables, so streamRuns merges - // two upstream Electric shapes (TaskRun + task_run_v2) under one composite - // cursor: handle and offset each pack the two per-table values joined by - // "~". Both shapes are at "0_0" for the initial snapshot. expect(shapeId).toBeDefined(); - expect(shapeId).toContain("~"); - expect(chunkOffset).toBe("0_0~0_0"); + expect(chunkOffset).toBe("0_0"); } );