Files
Daniel Sutton ae96b6c175 fix: read-your-writes + global-scope idempotency correctness under the run-ops split (#4284)
## What & why

Two related correctness fixes for the run-ops DB split. Under the split,
run-store reads can route to a **lagging read replica**; a just-written
run/waitpoint/batch can then be missed, causing a wrong decision.

**1. Read-your-writes → owning primary.** Surfaced first as an
intermittent `wait.until({ idempotencyKey })` re-wait on retry. Auditing
the run-store read surface found the same class at sibling sites (some
gating mutations or returning spurious 404s, others
tolerable/self-healing). Reads that must observe their own writes now
route to the owning **primary**
(`findRun`/`findWaitpoint`/`findBatchTaskRunByFriendlyId` →
`*OnPrimary`, a primary re-read on a miss, or a retryable 404 where the
SDK polls). Read-view reads stay on the replica. All additive — the
happy path is unchanged.

**2. Global-scope idempotency across the split.** A `global`-scope key
carries no per-run salt, so the same `(env, task, key)` triggered
concurrently from parents resident on **different** run-ops DBs could
dedup-miss on each DB and create a duplicate (the per-DB unique index
can't enforce cross-DB uniqueness). Such triggers (global scope, or
scope-absent, while split is active) are serialized through the existing
Redis idempotency claim, the loser resolves the winner by id across both
DBs, and the claim is reacquired on the expired/failed
clear-and-recreate path. `run`/`attempt` scope embed the run id and
never contend.

## Stacked for review

This is the **base** of a 2-PR stack, split so review is easier:
- **This PR** — production code only (34 files).
- **Stacked tests PR →
https://github.com/triggerdotdev/trigger.dev/pull/4285** — the
caller-driven guards (55 test files) on top of this branch.

## Validation

Local run-ops split, **both 2-DB and 3-DB**, fresh boot on this branch:
SDK canary 64/71 (only the known concurrency/input-streams/s3 failures),
quarantine sweep **0 unexpected** (340 pass / 16 known / 4 local) in
each topology, dashboard e2e 0 failed. No product regressions.
2026-07-19 17:57:41 +01:00

471 lines
13 KiB
TypeScript

import type { AuthenticatedEnvironment } from "@internal/run-engine";
import type { Prisma, PrismaClientOrTransaction, RuntimeEnvironment } from "@trigger.dev/database";
import { $replica, prisma } from "~/db.server";
import { runStore } from "~/v3/runStore.server";
import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
import { logger } from "~/services/logger.server";
import { getUsername } from "~/utils/username";
import { isDefaultDevBranch, sanitizeBranchName } from "@trigger.dev/core/v3/utils/gitBranch";
export type { RuntimeEnvironment };
// Prisma include shape that maps cleanly to the slim AuthenticatedEnvironment.
// Use this everywhere we fetch an env that flows to handlers — keeps the
// returned shape consistent (and the Decimal coercion in toAuthenticated()
// strips Prisma's Decimal class from the public surface).
export const authIncludeBase = {
project: true,
organization: true,
orgMember: {
select: {
userId: true,
user: { select: { id: true, displayName: true, name: true } },
},
},
} satisfies Prisma.RuntimeEnvironmentInclude;
export const authIncludeWithParent = {
...authIncludeBase,
parentEnvironment: { select: { id: true, apiKey: true } },
} satisfies Prisma.RuntimeEnvironmentInclude;
type PrismaEnvWithAuth = Prisma.RuntimeEnvironmentGetPayload<{ include: typeof authIncludeBase }>;
type PrismaEnvWithAuthAndParent = Prisma.RuntimeEnvironmentGetPayload<{
include: typeof authIncludeWithParent;
}>;
// Coerce a Prisma RuntimeEnvironment payload to the slim
// AuthenticatedEnvironment shape. Drops the columns handlers don't read
// and converts `concurrencyLimitBurstFactor` from Prisma's Decimal to a
// plain number (lossless at this scale). The optional union accepts both
// query shapes — with parentEnvironment loaded, or without it.
export function toAuthenticated(
env: PrismaEnvWithAuth | PrismaEnvWithAuthAndParent
): AuthenticatedEnvironment {
return {
id: env.id,
slug: env.slug,
type: env.type,
apiKey: env.apiKey,
organizationId: env.organizationId,
projectId: env.projectId,
orgMemberId: env.orgMemberId,
parentEnvironmentId: env.parentEnvironmentId,
branchName: env.branchName,
archivedAt: env.archivedAt,
paused: env.paused,
shortcode: env.shortcode,
maximumConcurrencyLimit: env.maximumConcurrencyLimit,
// Coerce Prisma's Decimal to a plain number — the slim type accepts
// both, but downstream consumers shouldn't have to narrow before
// doing arithmetic. Lossless at this scale (Decimal(4,2)).
concurrencyLimitBurstFactor: env.concurrencyLimitBurstFactor.toNumber(),
builtInEnvironmentVariableOverrides: env.builtInEnvironmentVariableOverrides,
createdAt: env.createdAt,
updatedAt: env.updatedAt,
project: {
id: env.project.id,
slug: env.project.slug,
name: env.project.name,
externalRef: env.project.externalRef,
engine: env.project.engine,
deletedAt: env.project.deletedAt,
defaultWorkerGroupId: env.project.defaultWorkerGroupId,
organizationId: env.project.organizationId,
builderProjectId: env.project.builderProjectId,
},
organization: {
id: env.organization.id,
slug: env.organization.slug,
title: env.organization.title,
streamBasinName: env.organization.streamBasinName,
maximumConcurrencyLimit: env.organization.maximumConcurrencyLimit,
runsEnabled: env.organization.runsEnabled,
maximumDevQueueSize: env.organization.maximumDevQueueSize,
maximumDeployedQueueSize: env.organization.maximumDeployedQueueSize,
featureFlags: env.organization.featureFlags,
apiRateLimiterConfig: env.organization.apiRateLimiterConfig,
batchRateLimitConfig: env.organization.batchRateLimitConfig,
batchQueueConcurrencyConfig: env.organization.batchQueueConcurrencyConfig,
},
orgMember: env.orgMember,
parentEnvironment: "parentEnvironment" in env ? env.parentEnvironment : null,
};
}
export async function findEnvironmentByApiKey(
apiKey: string,
branchName: string | undefined,
tx: PrismaClientOrTransaction = $replica
): Promise<AuthenticatedEnvironment | null> {
const branch = sanitizeBranchName(branchName) ?? undefined;
const include = {
...authIncludeBase,
childEnvironments: branch
? {
where: {
branchName: branch,
archivedAt: null,
},
}
: undefined,
} satisfies Prisma.RuntimeEnvironmentInclude;
let environment = await tx.runtimeEnvironment.findFirst({
where: {
apiKey,
},
include,
});
// Fall back to keys that were revoked within the grace window
if (!environment) {
const revokedApiKey = await tx.revokedApiKey.findFirst({
where: {
apiKey,
expiresAt: { gt: new Date() },
},
include: {
runtimeEnvironment: { include },
},
});
environment = revokedApiKey?.runtimeEnvironment ?? null;
}
if (!environment) {
return null;
}
//don't return deleted projects
if (environment.project.deletedAt !== null) {
return null;
}
if (environment.type === "PREVIEW") {
if (!branch) {
logger.warn("findEnvironmentByApiKey(): Preview env with no branch name provided", {
environmentId: environment.id,
});
return null;
}
const childEnvironment = environment.childEnvironments.at(0);
if (childEnvironment) {
return toAuthenticated({
...childEnvironment,
apiKey: environment.apiKey,
orgMember: environment.orgMember,
organization: environment.organization,
project: environment.project,
});
}
//A branch was specified but no child environment was found
return null;
}
// If there is a named DEV branch (other than default), return it
if (environment.type === "DEVELOPMENT" && branch !== undefined && !isDefaultDevBranch(branch)) {
const childEnvironment = environment.childEnvironments.at(0);
if (childEnvironment) {
return toAuthenticated({
...childEnvironment,
apiKey: environment.apiKey,
orgMember: environment.orgMember,
organization: environment.organization,
project: environment.project,
});
}
//A branch was specified but no child environment was found
return null;
}
return toAuthenticated(environment);
}
/**
* @deprecated We don't use public API keys (`pk_*` tokens) anymore — public
* access goes through public JWTs (see `isPublicJWT` / `validatePublicJwtKey`).
*
* Still exported because a handful of pre-RBAC routes that haven't been
* migrated to the apiBuilder still wire this lookup into their
* `authenticateApiKey` / `authenticateApiKeyWithFailure` flow. The new RBAC
* fallback (`internal-packages/rbac/src/fallback.ts`) intentionally does NOT
* call this — any pk_*-authenticated request that hits an apiBuilder route
* returns 401. That's a deliberate cutover, not an oversight.
*/
export async function findEnvironmentByPublicApiKey(
apiKey: string,
branchName: string | undefined
): Promise<AuthenticatedEnvironment | null> {
const environment = await $replica.runtimeEnvironment.findFirst({
where: {
pkApiKey: apiKey,
},
include: authIncludeBase,
});
if (!environment || environment.project.deletedAt !== null) {
return null;
}
return toAuthenticated(environment);
}
export async function findEnvironmentById(id: string): Promise<AuthenticatedEnvironment | null> {
const environment = await $replica.runtimeEnvironment.findFirst({
where: {
id,
},
include: authIncludeWithParent,
});
if (!environment || environment.project.deletedAt !== null) {
return null;
}
return toAuthenticated(environment);
}
export async function findEnvironmentBySlug(
projectId: string,
envSlug: string,
userId: string
): Promise<AuthenticatedEnvironment | null> {
const environment = await $replica.runtimeEnvironment.findFirst({
where: {
projectId: projectId,
slug: envSlug,
OR: [
{
type: {
in: ["PREVIEW", "STAGING", "PRODUCTION"],
},
},
{
type: "DEVELOPMENT",
orgMember: {
userId,
},
},
],
},
include: authIncludeBase,
});
return environment ? toAuthenticated(environment) : null;
}
// The authenticated environment plus the run scalars the realtime publish needs.
// Both come from one taskRun read — see findEnvironmentFromRun.
export type EnvironmentFromRun = {
environment: AuthenticatedEnvironment;
runTags: string[];
batchId: string | null;
};
export async function findEnvironmentFromRun(
runId: string,
tx?: PrismaClientOrTransaction
): Promise<EnvironmentFromRun | null> {
// Run-ops scalars (runTags/batchId/runtimeEnvironmentId) from the run store; the env half is
// resolved via the control-plane resolver so the run-ops DB can split without a cross-DB join.
const select = {
runTags: true,
batchId: true,
runtimeEnvironmentId: true,
} as const;
let taskRun = await runStore.findRun({ id: runId }, { select }, tx ?? $replica);
if (!taskRun) {
// Read-your-writes: a just-created run may not have replicated. Re-read the owning primary before
// treating it as absent, so runMetadataUpdated doesn't drop a live run's final metadata + publish.
taskRun = await runStore.findRun({ id: runId }, { select }, prisma);
}
if (!taskRun) {
return null;
}
const environment = await controlPlaneResolver.resolveAuthenticatedEnv(
taskRun.runtimeEnvironmentId
);
if (!environment) {
return null;
}
return {
environment,
runTags: taskRun.runTags,
batchId: taskRun.batchId,
};
}
export async function createNewSession(
environment: Pick<RuntimeEnvironment, "id">,
ipAddress: string
) {
const session = await prisma.runtimeEnvironmentSession.create({
data: {
environmentId: environment.id,
ipAddress,
},
});
await prisma.runtimeEnvironment.update({
where: {
id: environment.id,
},
data: {
currentSessionId: session.id,
},
});
return session;
}
export async function disconnectSession(environmentId: string) {
const environment = await prisma.runtimeEnvironment.findFirst({
where: {
id: environmentId,
},
});
if (!environment || !environment.currentSessionId) {
return null;
}
const session = await prisma.runtimeEnvironmentSession.update({
where: {
id: environment.currentSessionId,
},
data: {
disconnectedAt: new Date(),
},
});
await prisma.runtimeEnvironment.update({
where: {
id: environment.id,
},
data: {
currentSessionId: null,
},
});
return session;
}
export async function findLatestSession(
environmentId: string,
client: PrismaClientOrTransaction = $replica
) {
const session = await client.runtimeEnvironmentSession.findFirst({
where: {
environmentId,
},
orderBy: {
createdAt: "desc",
},
});
return session;
}
export type DisplayableInputEnvironment = Prisma.RuntimeEnvironmentGetPayload<{
select: {
id: true;
type: true;
slug: true;
orgMember: {
select: {
user: {
select: {
id: true;
name: true;
displayName: true;
};
};
};
};
};
}>;
export function displayableEnvironment(
environment: DisplayableInputEnvironment,
userId: string | undefined
) {
let userName: string | undefined = undefined;
if (environment.type === "DEVELOPMENT") {
if (!environment.orgMember) {
userName = "Deleted";
} else if (environment.orgMember.user.id !== userId) {
userName = getUsername(environment.orgMember.user);
}
}
return {
id: environment.id,
type: environment.type,
slug: environment.slug,
userName,
};
}
export async function findDisplayableEnvironment(
environmentId: string,
userId: string | undefined
) {
const environment = await $replica.runtimeEnvironment.findFirst({
where: {
id: environmentId,
},
select: {
id: true,
type: true,
slug: true,
orgMember: {
select: {
user: {
select: {
id: true,
name: true,
displayName: true,
},
},
},
},
},
});
if (!environment) {
return;
}
return displayableEnvironment(environment, userId);
}
export async function hasAccessToEnvironment({
environmentId,
projectId,
organizationId,
userId,
}: {
environmentId: string;
projectId: string;
organizationId: string;
userId: string;
}): Promise<boolean> {
const environment = await $replica.runtimeEnvironment.findFirst({
where: {
id: environmentId,
projectId: projectId,
organizationId: organizationId,
organization: { members: { some: { userId } } },
},
});
return environment !== null;
}