Eric Allam
f9d57d3bd5
## Summary Adds a second backend for the realtime runs feed (`useRealtimeRun`, `subscribeToRunsWithTag`, `subscribeToBatch`), built to stay healthy when a single busy environment has many subscribers watching many runs at once. It is gated behind a feature flag with the existing backend as the default, so nothing changes for users until it is enabled per environment. ## Design A run change is published once, as a small self-describing record, to a single per-environment channel. Every feed is then a predicate over that one stream rather than owning a channel: - A per-instance router indexes the currently-held feeds by run, tag, and batch. When a run changes it hydrates the affected rows once and serializes them once, then fans the result to every matching feed. One hot shared tag watched by many subscribers costs a single database query and serialize, not one per subscriber. - Feeds that don't match a change are never woken, wake delivery per environment is coalesced on a leading edge (250ms default) so a burst of changes costs one wake, and cold reads coalesce onto a single short-TTL-cached resolve. - An admission gate bounds how many cold ClickHouse resolves run concurrently, so a mass reconnect across many distinct filters queues instead of stampeding the database. - Changes that land while a client is between long-polls are delivered on its next poll instead of waiting for the periodic backstop: each environment buffers its recent change records, subscriptions linger briefly after the last feed closes, and a newly-armed poll replays exactly the connection's gap. - The per-connection replay cursors behind that are shared across instances via Redis (a single timestamp each), so a poll landing on a different instance behind the load balancer still reads the connection's true gap instead of falling back to a cold resolve. Cursor reads have a bounded deadline and degrade to the cold-read path on any Redis trouble. - Tag subscriptions with multiple tags match runs carrying all of the tags, mirroring the existing backend's filter semantics, and live long-polls hold for about 20 seconds to match its cadence. - The per-environment channel supports Redis Cluster sharded pub/sub, so the wake path scales horizontally across shards by environment. - The backend reports its health through OpenTelemetry metrics (delivery lag, poll resolution paths, backstop outcomes, replay and cursor-store activity), with a provisioned Grafana dashboard for local development. Everything is behind the feature flag and tunable via env vars; the existing backend remains the default.
111 lines
3.8 KiB
TypeScript
111 lines
3.8 KiB
TypeScript
import { CURRENT_API_VERSION } from "~/api/versions";
|
|
import {
|
|
NativeRealtimeClient,
|
|
type RealtimeListEnvironment,
|
|
} from "~/services/realtime/nativeRealtimeClient.server";
|
|
import { type RealtimeRunRow } from "~/services/realtime/electricStreamProtocol.server";
|
|
import { EnvChangeRouter } from "~/services/realtime/envChangeRouter.server";
|
|
import { describe, expect, it } from "vitest";
|
|
|
|
function sampleRow(): RealtimeRunRow {
|
|
return {
|
|
id: "run_1",
|
|
taskIdentifier: "t",
|
|
createdAt: new Date("2026-06-07T10:00:00.000Z"),
|
|
updatedAt: new Date("2026-06-07T10:00:01.000Z"),
|
|
startedAt: null,
|
|
delayUntil: null,
|
|
queuedAt: null,
|
|
expiredAt: null,
|
|
completedAt: null,
|
|
friendlyId: "run_friendly_1",
|
|
number: 1,
|
|
isTest: false,
|
|
status: "EXECUTING",
|
|
usageDurationMs: 0,
|
|
costInCents: 0,
|
|
baseCostInCents: 0,
|
|
ttl: null,
|
|
payload: "{}",
|
|
payloadType: "application/json",
|
|
metadata: null,
|
|
metadataType: "application/json",
|
|
output: null,
|
|
outputType: "application/json",
|
|
runTags: [],
|
|
error: null,
|
|
realtimeStreams: [],
|
|
};
|
|
}
|
|
|
|
// Only the initial-snapshot path is exercised here, which touches the shared
|
|
// #buildResponse — enough to lock the response-header contract.
|
|
function makeClient(row: RealtimeRunRow | null) {
|
|
return new NativeRealtimeClient({
|
|
runReader: {
|
|
getRunById: async () => row,
|
|
hydrateByIds: async () => (row ? [row] : []),
|
|
} as any,
|
|
runListResolver: { resolveMatchingRunIds: async () => [] } as any,
|
|
// Snapshot path only; the router (over a no-op source) is never invoked here.
|
|
router: new EnvChangeRouter({
|
|
source: { subscribeToEnv: () => () => {} },
|
|
hydrator: { hydrateByIds: async () => (row ? [row] : []) },
|
|
replayWindowMs: 0,
|
|
unsubscribeLingerMs: 0,
|
|
}),
|
|
limiter: { incrementAndCheck: async () => true, decrement: async () => {} } as any,
|
|
cachedLimitProvider: { getCachedLimit: async () => 100 },
|
|
maximumCreatedAtFilterAgeMs: 24 * 60 * 60 * 1000,
|
|
});
|
|
}
|
|
|
|
const ENV: RealtimeListEnvironment = {
|
|
id: "env_1",
|
|
organizationId: "org_1",
|
|
projectId: "proj_1",
|
|
};
|
|
|
|
describe("NativeRealtimeClient response headers", () => {
|
|
it("exposes electric headers cross-origin so browser hooks can read them", async () => {
|
|
const client = makeClient(sampleRow());
|
|
const res = await client.streamRun(
|
|
"http://localhost:3030/realtime/v1/runs/run_1?offset=-1",
|
|
ENV,
|
|
"run_1",
|
|
CURRENT_API_VERSION,
|
|
undefined,
|
|
"1.0.0-beta.1" // modern client => lowercase electric-* headers
|
|
);
|
|
|
|
// Without these the deployed @electric-sql/client throws MissingHeadersError
|
|
// (it can't read the electric-* headers across origins). This regressed once.
|
|
expect(res.headers.get("access-control-allow-origin")).toBe("*");
|
|
expect(res.headers.get("access-control-expose-headers")).toBe("*");
|
|
|
|
// Initial (non-live) snapshot requires offset + handle + schema.
|
|
expect(res.headers.get("electric-offset")).toBeTruthy();
|
|
expect(res.headers.get("electric-handle")).toBeTruthy();
|
|
expect(res.headers.get("electric-schema")).toBeTruthy();
|
|
expect(res.headers.get("content-type")).toBe("application/json");
|
|
});
|
|
|
|
it("renames headers for legacy (0.4.0) clients", async () => {
|
|
const client = makeClient(sampleRow());
|
|
const res = await client.streamRun(
|
|
"http://localhost:3030/realtime/v1/runs/run_1?offset=-1",
|
|
ENV,
|
|
"run_1",
|
|
CURRENT_API_VERSION,
|
|
undefined,
|
|
undefined // no client version => legacy header names
|
|
);
|
|
|
|
expect(res.headers.get("electric-chunk-last-offset")).toBeTruthy();
|
|
expect(res.headers.get("electric-shape-id")).toBeTruthy();
|
|
expect(res.headers.get("electric-offset")).toBeNull();
|
|
expect(res.headers.get("electric-handle")).toBeNull();
|
|
expect(res.headers.get("access-control-expose-headers")).toBe("*");
|
|
});
|
|
});
|