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.
This commit is contained in:
Dan Sutton
@@ -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, `<lsn>_<n>` 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 "<a>~<b>" 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,
|
||||
};
|
||||
}
|
||||
@@ -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<Response> {
|
||||
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<MergedShape> {
|
||||
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<ParsedShape> {
|
||||
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
|
||||
|
||||
@@ -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> = {}): 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"}}');
|
||||
});
|
||||
});
|
||||
@@ -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");
|
||||
}
|
||||
);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user