Daniel Sutton
e4ae8cbcd4
## Summary On the run-ops database split, a run that waits (`triggerAndWait`, `batchTriggerAndWait`, `wait.forToken`) could hang forever after its wait had already completed. The runner reads a resume from `/snapshots/since` exactly once: if that read returned the resume snapshot without its completed-waitpoints, the runner logged "executing without completed waitpoints", advanced its cursor, and never re-read it, so the awaiting run never continued. ## Root cause The resume snapshot and its completed-waitpoint rows were written as two separate commits. This regressed when the split replaced Prisma's atomic nested `connect` with an FK-free insert (in [#4163](https://github.com/triggerdotdev/trigger.dev/pull/4163)), and `/snapshots/since` is served from a read replica. A fetch landing in the sub-millisecond gap between the two commits, or a multi-reader replica serving the snapshot from a different point in time than its join rows, delivered an empty resume. Because the runner consumes each snapshot once and treats an empty resume as terminal, a single stale read was fatal and produced a permanent, nondeterministic hang. ## Fixes - Commit a snapshot and its completed-waitpoint links in one transaction, restoring the atomicity the split removed. - Repair the completed-waitpoints from the owning primary when a multi-reader replica serves the snapshot without its join rows. This covers single-waitpoint resumes, which carry no `completedWaitpointOrder` and so were missed by the count-based repair. - Read the primary in the checkpoint `WAIT_FOR_BATCH` pre-check, so a batch that already resumed is not re-suspended into a stall. - Fall back to the primary when a waitpoint token misses both read replicas, so a token completed immediately after it was minted no longer returns a spurious 404. - Route batch-item creation by `batchTaskRunId`, consistent with the batch-completion count and the row's foreign key. - Reject control-plane-only relation selects on the dedicated schema with a clear error instead of an opaque Prisma failure, and stop `createDateTimeWaitpoint` bypassing residency routing through a caller transaction. Verified against the deployed split topology: a resume snapshot and its completed-waitpoints are now always delivered together, so the runner can no longer drop a resume.
76 lines
2.7 KiB
TypeScript
76 lines
2.7 KiB
TypeScript
import type { PrismaReplicaClient } from "~/db.server";
|
|
import {
|
|
$replica as defaultLegacyReplica,
|
|
runOpsNewPrisma as defaultNewPrimary,
|
|
runOpsNewReplica as defaultNewClient,
|
|
runOpsSplitReadEnabled as defaultSplitReadEnabled,
|
|
} from "~/db.server";
|
|
import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
|
|
|
|
type ResolveWaitpointDeps = {
|
|
newClient?: PrismaReplicaClient;
|
|
legacyReplica?: PrismaReplicaClient;
|
|
newPrimary?: PrismaReplicaClient;
|
|
splitEnabled?: boolean;
|
|
isPastRetention?: (id: string) => boolean;
|
|
};
|
|
|
|
// Safe defaults matching the deps `complete`/`callback` pass, so a bare caller still fans
|
|
// out to the dedicated run-ops replica (NEW-resident waitpoints) before control-plane.
|
|
export type ResolveWaitpointReadThroughDefaults = {
|
|
newClient: PrismaReplicaClient;
|
|
legacyReplica: PrismaReplicaClient;
|
|
newPrimary: PrismaReplicaClient;
|
|
splitEnabled: boolean;
|
|
};
|
|
|
|
const productionDefaults: ResolveWaitpointReadThroughDefaults = {
|
|
newClient: defaultNewClient,
|
|
legacyReplica: defaultLegacyReplica,
|
|
newPrimary: defaultNewPrimary as unknown as PrismaReplicaClient,
|
|
splitEnabled: defaultSplitReadEnabled,
|
|
};
|
|
|
|
export async function resolveWaitpointThroughReadThrough<T>(opts: {
|
|
waitpointId: string;
|
|
environmentId: string;
|
|
read: (client: PrismaReplicaClient) => Promise<T | null>;
|
|
deps?: ResolveWaitpointDeps;
|
|
defaults?: ResolveWaitpointReadThroughDefaults;
|
|
}): Promise<T | null> {
|
|
const defaults = opts.defaults ?? productionDefaults;
|
|
|
|
const splitEnabled = opts.deps?.splitEnabled ?? defaults.splitEnabled;
|
|
|
|
const result = await readThroughRun({
|
|
runId: opts.waitpointId,
|
|
environmentId: opts.environmentId,
|
|
readNew: (client) => opts.read(client),
|
|
readLegacy: (replica) => opts.read(replica),
|
|
deps: {
|
|
splitEnabled,
|
|
newClient: opts.deps?.newClient ?? defaults.newClient,
|
|
legacyReplica: opts.deps?.legacyReplica ?? defaults.legacyReplica,
|
|
isPastRetention: opts.deps?.isPastRetention,
|
|
},
|
|
});
|
|
|
|
if (result.source === "new" || result.source === "legacy-replica") {
|
|
return result.value;
|
|
}
|
|
// past-retention is an intentional not-found: the token is gone.
|
|
if (result.source === "past-retention") {
|
|
return null;
|
|
}
|
|
|
|
// Read-your-writes fallback for a token completed immediately after mint, before it replicated:
|
|
// re-read from the run-ops PRIMARY only. We deliberately never read the control-plane/legacy
|
|
// primary here (that is the load the replica-only read-through exists to shed), so a legacy-resident
|
|
// token that misses its replica stays a miss and the caller retries, rather than adding primary load.
|
|
const fromNewPrimary = await opts.read(opts.deps?.newPrimary ?? defaults.newPrimary);
|
|
if (fromNewPrimary != null) {
|
|
return fromNewPrimary;
|
|
}
|
|
return null;
|
|
}
|