Eric Allam
d3906241a5
🦋 Changesets PR / Create Release PR (push) Has been cancelled
🚀 Publish Trigger.dev Docker / 📣 Dispatch main image (push) Has been cancelled
🚀 Publish Trigger.dev Docker / typecheck (push) Failing after 2s
🚀 Publish Trigger.dev Docker / units (push) Failing after 2s
🚀 Publish Trigger.dev Docker / publish-webapp (push) Has been skipped
🚀 Publish Trigger.dev Docker / publish-worker-v4 (push) Has been skipped
🚀 Publish Trigger.dev Docker / scan-webapp (push) Has been skipped
🚀 Publish Trigger.dev Docker / scan-supervisor (push) Has been skipped
## Summary
The realtime runs feed hydrates run rows from read replicas, which means
it needs a replica-lag gate to avoid serving a run's previous state
right after a write. Setting
`REALTIME_BACKEND_NATIVE_RUN_READS_FROM_PRIMARY=1` reads those rows from
each run store's primary instead, so there is no lag to gate against: no
probe, no wake delay, no stale-read retries. Off by default, so nothing
changes unless you set it.
## Design
The run stores already decide replica-vs-primary from the *brand* on the
read client they are handed: a branded replica keeps the read on the
owning store's replica, an unbranded writer escalates it to that store's
own primary. So this is a one-line choice at the hydrator, and it stays
correct across topologies. With the run-ops split on, each leg lands on
its own writer and the caller's client is never forwarded across
databases; with the split off, it is the single database's primary.
```ts
const runReader = new RunHydrator({
readClient: runReadsFromPrimary ? prisma : $replica,
runStore,
});
```
The same flag skips constructing the lag estimator, since probing a
replica the feed no longer reads would be measuring the wrong thing.
Independently, `AuroraReplicaLagSource` detected Aurora by letting
`aurora_replica_status()` fail, on the assumption that the app-level
catch made that free. It isn't: an unresolvable function is a query
error the driver reports to the error log on every sample, so a
non-Aurora replica produced a continuous stream of error events while
the estimator quietly fell through to its next candidate. It now
resolves the function with `to_regproc` and memoizes the answer, so the
unparseable call never reaches the wire.
81 lines
2.9 KiB
TypeScript
81 lines
2.9 KiB
TypeScript
import { describe, expect, it, vi } from "vitest";
|
|
import { PostgresRunStore } from "@internal/run-store";
|
|
import { buildHydratorSelect, RunHydrator } from "~/services/realtime/runReader.server";
|
|
|
|
describe("buildHydratorSelect", () => {
|
|
it("returns the full select when nothing is skipped", () => {
|
|
const select = buildHydratorSelect([]);
|
|
expect(select.id).toBe(true);
|
|
expect(select.payload).toBe(true);
|
|
expect(select.output).toBe(true);
|
|
expect(select.metadata).toBe(true);
|
|
expect(select.error).toBe(true);
|
|
});
|
|
|
|
it("keeps protocol-reserved columns even when asked to skip them", () => {
|
|
// Reserved columns are always emitted by the serializer, so hydration must keep
|
|
// them regardless of skipColumns or the output is null/incorrect.
|
|
const select = buildHydratorSelect([
|
|
"status",
|
|
"taskIdentifier",
|
|
"createdAt",
|
|
"friendlyId",
|
|
"payload",
|
|
]);
|
|
expect(select.status).toBe(true);
|
|
expect(select.taskIdentifier).toBe(true);
|
|
expect(select.createdAt).toBe(true);
|
|
expect(select.friendlyId).toBe(true);
|
|
// A non-reserved skipped column is still dropped.
|
|
expect(select.payload).toBeUndefined();
|
|
});
|
|
|
|
it("drops skipped columns but always keeps id + updatedAt", () => {
|
|
const select = buildHydratorSelect(["payload", "output", "metadata", "error"]);
|
|
expect(select.payload).toBeUndefined();
|
|
expect(select.output).toBeUndefined();
|
|
expect(select.metadata).toBeUndefined();
|
|
expect(select.error).toBeUndefined();
|
|
// Needed internally regardless of skipColumns (keys the row, drives the diff/offset).
|
|
expect(select.id).toBe(true);
|
|
expect(select.updatedAt).toBe(true);
|
|
// A non-skipped column survives.
|
|
expect(select.status).toBe(true);
|
|
});
|
|
});
|
|
|
|
describe("RunHydrator.hydrateByIds column projection", () => {
|
|
function makeHydrator() {
|
|
let capturedSelect: Record<string, boolean> | undefined;
|
|
const replica = {
|
|
taskRun: {
|
|
findMany: vi.fn(async ({ select }: { select: Record<string, boolean> }) => {
|
|
capturedSelect = select;
|
|
return [];
|
|
}),
|
|
},
|
|
} as any;
|
|
const runStore = new PostgresRunStore({ prisma: replica, readOnlyPrisma: replica });
|
|
return {
|
|
hydrator: new RunHydrator({ readClient: replica, runStore }),
|
|
getSelect: () => capturedSelect,
|
|
};
|
|
}
|
|
|
|
it("projects the SELECT by skipColumns", async () => {
|
|
const { hydrator, getSelect } = makeHydrator();
|
|
await hydrator.hydrateByIds("env_1", ["run_1"], ["payload", "output"]);
|
|
const select = getSelect()!;
|
|
expect(select.payload).toBeUndefined();
|
|
expect(select.output).toBeUndefined();
|
|
expect(select.id).toBe(true);
|
|
expect(select.updatedAt).toBe(true);
|
|
});
|
|
|
|
it("selects the full column set when no skipColumns are given", async () => {
|
|
const { hydrator, getSelect } = makeHydrator();
|
|
await hydrator.hydrateByIds("env_1", ["run_1"]);
|
|
expect(getSelect()!.payload).toBe(true);
|
|
});
|
|
});
|