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.
70 lines
2.6 KiB
TypeScript
70 lines
2.6 KiB
TypeScript
import { getMeter } from "@internal/tracing";
|
|
import { $replica } from "~/db.server";
|
|
import { runStore } from "~/v3/runStore.server";
|
|
import { env } from "~/env.server";
|
|
import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
|
|
import { singleton } from "~/utils/singleton";
|
|
import { realtimeClient } from "../realtimeClientGlobal.server";
|
|
import { ClickHouseRunListResolver } from "./clickHouseRunListResolver.server";
|
|
import { RunHydrator } from "./runReader.server";
|
|
import { RealtimeShadowComparator } from "./shadowCompare.server";
|
|
import { ShadowRealtimeClient } from "./shadowRealtimeClient.server";
|
|
|
|
/**
|
|
* Process-singleton wiring for the shadow-compare client. Only constructed
|
|
* when an org's `realtimeBackend` flag is set to "shadow".
|
|
*/
|
|
function initializeShadowRealtimeClient(): ShadowRealtimeClient {
|
|
const compares = getMeter("realtime-shadow").createCounter("realtime_shadow.compares", {
|
|
description:
|
|
"Dual-run shadow-compare outcomes (Electric vs native). kind=serialization|membership, result=match|diverge|skew.",
|
|
});
|
|
|
|
const comparator = new RealtimeShadowComparator({
|
|
runReader: new RunHydrator({ readClient: $replica, runStore }),
|
|
runListResolver: new ClickHouseRunListResolver({
|
|
getClickhouse: (organizationId) =>
|
|
clickhouseFactory.getClickhouseForOrganization(organizationId, "realtime"),
|
|
prisma: $replica,
|
|
}),
|
|
});
|
|
|
|
return new ShadowRealtimeClient({
|
|
electric: realtimeClient,
|
|
comparator,
|
|
maximumCreatedAtFilterAgeMs: env.REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS,
|
|
maxListResults: env.REALTIME_BACKEND_NATIVE_MAX_LIST_RESULTS,
|
|
onOutcome: (outcome) => {
|
|
const { feed } = outcome;
|
|
if (outcome.serializationMatched) {
|
|
compares.add(outcome.serializationMatched, {
|
|
feed,
|
|
kind: "serialization",
|
|
result: "match",
|
|
});
|
|
}
|
|
if (outcome.serializationDiverged) {
|
|
compares.add(outcome.serializationDiverged, {
|
|
feed,
|
|
kind: "serialization",
|
|
result: "diverge",
|
|
});
|
|
}
|
|
if (outcome.serializationSkew) {
|
|
compares.add(outcome.serializationSkew, { feed, kind: "serialization", result: "skew" });
|
|
}
|
|
if (outcome.membershipMatch !== undefined) {
|
|
compares.add(1, {
|
|
feed,
|
|
kind: "membership",
|
|
result: outcome.membershipMatch ? "match" : "diverge",
|
|
});
|
|
}
|
|
},
|
|
});
|
|
}
|
|
|
|
export function getShadowRealtimeClient(): ShadowRealtimeClient {
|
|
return singleton("shadowRealtimeClient", initializeShadowRealtimeClient);
|
|
}
|