Daniel Sutton
bea7e2be90
## Summary Run-graph data (runs, batches, waitpoints, and their related tables) can now live in a database separate from the control plane, with every read and write routed to the correct database by each run's residency. This makes reading and writing run data more reliable once the two are split, and is a no-op for single-database installs. ## Design - Run-graph table access goes through the run-store router, which selects the legacy or the new run-ops store per run instead of assuming one shared client. - The legacy run-ops client is now independently pointable, so legacy run data can be served from its own database (and replica) rather than the control-plane connection. - Run-graph writes go straight to the run-graph database instead of being forwarded through the control plane, and replication targets are split so runs in the new database still replicate to analytics without under-counting. - Read-through slots refuse the control-plane client, so a missing residency fails loudly instead of silently reading the wrong database. - Migration `20260710120000_drop_remaining_run_graph_seam_foreign_keys` drops the foreign keys that still crossed the run-graph / control-plane seam, which is what lets the two live in separate databases. The split stays off unless explicitly enabled and the two databases are confirmed physically distinct; startup fails closed otherwise. Verified by running the full dashboard end-to-end suite against both a single-database configuration and a three-database configuration (control plane, the new database, and a physically separate legacy database), with runs on both residencies. No misrouted reads in either configuration.
1101 lines
35 KiB
TypeScript
1101 lines
35 KiB
TypeScript
import type { CompleteBatchResult } from "@internal/run-engine";
|
|
import { SpanKind } from "@internal/tracing";
|
|
import { tryCatch } from "@trigger.dev/core/utils";
|
|
import { createJsonErrorObject, sanitizeError, TaskRunErrorCodes } from "@trigger.dev/core/v3";
|
|
import { RunId } from "@trigger.dev/core/v3/isomorphic";
|
|
import {
|
|
$replica,
|
|
prisma,
|
|
runOpsNewReplica,
|
|
runOpsLegacyReplica,
|
|
runOpsNewPrismaClient,
|
|
runOpsNewReplicaClient,
|
|
runOpsLegacyPrismaClient,
|
|
} from "~/db.server";
|
|
import { env } from "~/env.server";
|
|
import { findEnvironmentById, findEnvironmentFromRun } from "~/models/runtimeEnvironment.server";
|
|
import { TriggerFailedTaskService } from "~/runEngine/services/triggerFailedTask.server";
|
|
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
|
|
import { logger } from "~/services/logger.server";
|
|
import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
|
|
import { reportInvocationUsage } from "~/services/platform.v3.server";
|
|
import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server";
|
|
import { MetadataTooLargeError } from "~/utils/packets";
|
|
import { QueueSizeLimitExceededError } from "~/v3/services/common.server";
|
|
import { TriggerTaskService } from "~/v3/services/triggerTask.server";
|
|
import { tracer } from "~/v3/tracer.server";
|
|
import { createExceptionPropertiesFromError } from "./eventRepository/common.server";
|
|
import { getEventRepositoryForStore, recordRunDebugLog } from "./eventRepository/index.server";
|
|
import { roomFromFriendlyRunId, socketIo } from "./handleSocketIo.server";
|
|
import { engine } from "./runEngine.server";
|
|
import { runStore } from "./runStore.server";
|
|
import { mintAnchoredRunFriendlyId } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server";
|
|
import { isSplitEnabled } from "~/v3/runOpsMigration/splitMode.server";
|
|
import { PerformTaskRunAlertsService } from "./services/alerts/performTaskRunAlerts.server";
|
|
import {
|
|
handleBatchCompletion,
|
|
QUEUE_SIZE_LIMIT_EXCEEDED_ERROR_CODE,
|
|
readRunForEvent,
|
|
readRunForEventOrThrow,
|
|
type EventReadDeps,
|
|
} from "./runEngineHandlersShared.server";
|
|
|
|
export function registerRunEngineEventBusHandlers() {
|
|
// Resolve the split-mode gate ONCE at registration scope (never per-event).
|
|
const splitEnabledPromise = isSplitEnabled();
|
|
const eventReadDeps = async (): Promise<EventReadDeps> => ({
|
|
store: runStore,
|
|
newReplica: runOpsNewReplica,
|
|
legacyReplica: runOpsLegacyReplica,
|
|
splitEnabled: await splitEnabledPromise,
|
|
});
|
|
engine.eventBus.on("runSucceeded", async ({ time, run, organization, environment }) => {
|
|
const [taskRunError, taskRun] = await tryCatch(
|
|
readRunForEventOrThrow(
|
|
run.id,
|
|
environment.id,
|
|
{
|
|
id: true,
|
|
friendlyId: true,
|
|
traceId: true,
|
|
spanId: true,
|
|
parentSpanId: true,
|
|
createdAt: true,
|
|
completedAt: true,
|
|
taskIdentifier: true,
|
|
projectId: true,
|
|
runtimeEnvironmentId: true,
|
|
environmentType: true,
|
|
isTest: true,
|
|
organizationId: true,
|
|
taskEventStore: true,
|
|
// Piggyback the realtime run-changed publish on this existing read so the
|
|
// per-env channel carries the membership keys (no separate query). No-op when
|
|
// the native backend is disabled.
|
|
runTags: true,
|
|
batchId: true,
|
|
},
|
|
await eventReadDeps()
|
|
)
|
|
);
|
|
|
|
if (taskRunError) {
|
|
logger.error("[runSucceeded] Failed to find task run", {
|
|
error: taskRunError,
|
|
runId: run.id,
|
|
});
|
|
return;
|
|
}
|
|
|
|
publishChangeRecord({
|
|
runId: taskRun.id,
|
|
envId: environment.id,
|
|
tags: taskRun.runTags,
|
|
batchId: taskRun.batchId,
|
|
updatedAtMs: run.updatedAt.getTime(),
|
|
});
|
|
|
|
const eventRepository = await getEventRepositoryForStore(
|
|
run.taskEventStore,
|
|
taskRun.organizationId ?? organization.id
|
|
);
|
|
|
|
const [completeSuccessfulRunEventError] = await tryCatch(
|
|
eventRepository.completeSuccessfulRunEvent({
|
|
run: taskRun,
|
|
endTime: time,
|
|
})
|
|
);
|
|
|
|
if (completeSuccessfulRunEventError) {
|
|
logger.error("[runSucceeded] Failed to complete successful run event", {
|
|
error: completeSuccessfulRunEventError,
|
|
runId: run.id,
|
|
});
|
|
}
|
|
});
|
|
|
|
// Handle alerts
|
|
engine.eventBus.on("runFailed", async ({ time, run }) => {
|
|
try {
|
|
await PerformTaskRunAlertsService.enqueue(run.id);
|
|
} catch (error) {
|
|
logger.error("[runFailed] Failed to enqueue alerts", {
|
|
error: error instanceof Error ? error.message : error,
|
|
runId: run.id,
|
|
spanId: run.spanId,
|
|
});
|
|
}
|
|
});
|
|
|
|
// Handle events
|
|
engine.eventBus.on("runFailed", async ({ time, run, organization, environment }) => {
|
|
const sanitizedError = sanitizeError(run.error);
|
|
const exception = createExceptionPropertiesFromError(sanitizedError);
|
|
|
|
const [taskRunError, taskRun] = await tryCatch(
|
|
readRunForEventOrThrow(
|
|
run.id,
|
|
environment.id,
|
|
{
|
|
id: true,
|
|
friendlyId: true,
|
|
traceId: true,
|
|
spanId: true,
|
|
parentSpanId: true,
|
|
createdAt: true,
|
|
completedAt: true,
|
|
taskIdentifier: true,
|
|
projectId: true,
|
|
runtimeEnvironmentId: true,
|
|
environmentType: true,
|
|
isTest: true,
|
|
organizationId: true,
|
|
taskEventStore: true,
|
|
// Piggyback the realtime run-changed publish on this existing read (no-op when
|
|
// the native backend is disabled).
|
|
runTags: true,
|
|
batchId: true,
|
|
},
|
|
await eventReadDeps()
|
|
)
|
|
);
|
|
|
|
if (taskRunError) {
|
|
logger.error("[runFailed] Failed to find task run", {
|
|
error: taskRunError,
|
|
runId: run.id,
|
|
});
|
|
return;
|
|
}
|
|
|
|
publishChangeRecord({
|
|
runId: taskRun.id,
|
|
envId: environment.id,
|
|
tags: taskRun.runTags,
|
|
batchId: taskRun.batchId,
|
|
updatedAtMs: run.updatedAt.getTime(),
|
|
});
|
|
|
|
const eventRepository = await getEventRepositoryForStore(
|
|
run.taskEventStore,
|
|
taskRun.organizationId ?? organization.id
|
|
);
|
|
|
|
const [completeFailedRunEventError] = await tryCatch(
|
|
eventRepository.completeFailedRunEvent({
|
|
run: taskRun,
|
|
endTime: time,
|
|
exception,
|
|
})
|
|
);
|
|
|
|
if (completeFailedRunEventError) {
|
|
logger.error("[runFailed] Failed to complete failed run event", {
|
|
error: completeFailedRunEventError,
|
|
runId: run.id,
|
|
});
|
|
}
|
|
});
|
|
|
|
engine.eventBus.on("runAttemptFailed", async ({ time, run }) => {
|
|
const sanitizedError = sanitizeError(run.error);
|
|
const exception = createExceptionPropertiesFromError(sanitizedError);
|
|
|
|
const [taskRunError, taskRun] = await tryCatch(
|
|
readRunForEventOrThrow(
|
|
run.id,
|
|
// runAttemptFailed carries no environment param; the env is derived from
|
|
// the read row afterwards. environmentId is informational for read-through
|
|
// (residency is keyed on runId), so an empty value is safe here.
|
|
"",
|
|
{
|
|
id: true,
|
|
friendlyId: true,
|
|
traceId: true,
|
|
spanId: true,
|
|
parentSpanId: true,
|
|
createdAt: true,
|
|
completedAt: true,
|
|
taskIdentifier: true,
|
|
projectId: true,
|
|
runtimeEnvironmentId: true,
|
|
environmentType: true,
|
|
isTest: true,
|
|
organizationId: true,
|
|
taskEventStore: true,
|
|
// Piggyback the realtime run-changed publish on this existing read (no-op when
|
|
// the native backend is disabled).
|
|
runTags: true,
|
|
batchId: true,
|
|
},
|
|
await eventReadDeps()
|
|
)
|
|
);
|
|
|
|
if (taskRunError) {
|
|
logger.error("[runAttemptFailed] Failed to find task run", {
|
|
error: taskRunError,
|
|
runId: run.id,
|
|
});
|
|
return;
|
|
}
|
|
|
|
publishChangeRecord({
|
|
runId: taskRun.id,
|
|
envId: taskRun.runtimeEnvironmentId,
|
|
tags: taskRun.runTags,
|
|
batchId: taskRun.batchId,
|
|
updatedAtMs: run.updatedAt.getTime(),
|
|
});
|
|
|
|
if (!taskRun.organizationId) {
|
|
logger.error("[runAttemptFailed] Task run has no organization id", {
|
|
runId: run.id,
|
|
});
|
|
return;
|
|
}
|
|
|
|
const eventRepository = await getEventRepositoryForStore(
|
|
run.taskEventStore,
|
|
taskRun.organizationId
|
|
);
|
|
|
|
const [createAttemptFailedRunEventError] = await tryCatch(
|
|
eventRepository.createAttemptFailedRunEvent({
|
|
run: taskRun,
|
|
endTime: time,
|
|
attemptNumber: run.attemptNumber,
|
|
exception,
|
|
})
|
|
);
|
|
|
|
if (createAttemptFailedRunEventError) {
|
|
logger.error("[runAttemptFailed] Failed to create attempt failed run event", {
|
|
error: createAttemptFailedRunEventError,
|
|
runId: run.id,
|
|
});
|
|
}
|
|
});
|
|
|
|
engine.eventBus.on(
|
|
"cachedRunCompleted",
|
|
async ({ time, span, blockedRunId, hasError, cachedRunId }) => {
|
|
const [parentSpanId, spanId] = span.id.split(":");
|
|
|
|
if (!spanId || !parentSpanId) {
|
|
logger.debug("[cachedRunCompleted] Invalid span id", {
|
|
spanId,
|
|
parentSpanId,
|
|
});
|
|
return;
|
|
}
|
|
|
|
const deps = await eventReadDeps();
|
|
|
|
const [cachedRunError, cachedRun] = await tryCatch(
|
|
readRunForEventOrThrow(
|
|
cachedRunId ?? "",
|
|
"",
|
|
{
|
|
id: true,
|
|
friendlyId: true,
|
|
traceId: true,
|
|
spanId: true,
|
|
parentSpanId: true,
|
|
createdAt: true,
|
|
completedAt: true,
|
|
taskIdentifier: true,
|
|
projectId: true,
|
|
runtimeEnvironmentId: true,
|
|
environmentType: true,
|
|
isTest: true,
|
|
organizationId: true,
|
|
},
|
|
deps
|
|
)
|
|
);
|
|
|
|
if (cachedRunError) {
|
|
logger.error("[cachedRunCompleted] Failed to find cached run", {
|
|
error: cachedRunError,
|
|
cachedRunId,
|
|
});
|
|
return;
|
|
}
|
|
|
|
const [blockedRunError, blockedRun] = await tryCatch(
|
|
readRunForEvent(
|
|
blockedRunId,
|
|
"",
|
|
{
|
|
id: true,
|
|
friendlyId: true,
|
|
traceId: true,
|
|
spanId: true,
|
|
parentSpanId: true,
|
|
createdAt: true,
|
|
completedAt: true,
|
|
taskIdentifier: true,
|
|
projectId: true,
|
|
runtimeEnvironmentId: true,
|
|
environmentType: true,
|
|
isTest: true,
|
|
organizationId: true,
|
|
taskEventStore: true,
|
|
},
|
|
deps
|
|
)
|
|
);
|
|
|
|
if (blockedRunError) {
|
|
logger.error("[cachedRunCompleted] Failed to find blocked run", {
|
|
error: blockedRunError,
|
|
blockedRunId,
|
|
});
|
|
}
|
|
|
|
if (!blockedRun) {
|
|
logger.error("[cachedRunCompleted] Blocked run not found", {
|
|
blockedRunId,
|
|
});
|
|
return;
|
|
}
|
|
|
|
if (!blockedRun.organizationId) {
|
|
logger.error("[cachedRunCompleted] Blocked run has no organization id", {
|
|
blockedRunId,
|
|
});
|
|
return;
|
|
}
|
|
|
|
const eventRepository = await getEventRepositoryForStore(
|
|
blockedRun.taskEventStore,
|
|
blockedRun.organizationId
|
|
);
|
|
|
|
const [completeCachedRunEventError] = await tryCatch(
|
|
eventRepository.completeCachedRunEvent({
|
|
run: cachedRun,
|
|
blockedRun,
|
|
spanId,
|
|
parentSpanId,
|
|
spanCreatedAt: span.createdAt,
|
|
isError: hasError,
|
|
endTime: time,
|
|
})
|
|
);
|
|
|
|
if (completeCachedRunEventError) {
|
|
logger.error("[cachedRunCompleted] Failed to complete cached run event", {
|
|
error: completeCachedRunEventError,
|
|
cachedRunId,
|
|
});
|
|
}
|
|
}
|
|
);
|
|
|
|
engine.eventBus.on("runExpired", async ({ time, run, organization, environment }) => {
|
|
if (!run.ttl) {
|
|
return;
|
|
}
|
|
|
|
const [taskRunError, taskRun] = await tryCatch(
|
|
readRunForEventOrThrow(
|
|
run.id,
|
|
environment.id,
|
|
{
|
|
id: true,
|
|
friendlyId: true,
|
|
traceId: true,
|
|
spanId: true,
|
|
parentSpanId: true,
|
|
createdAt: true,
|
|
completedAt: true,
|
|
taskIdentifier: true,
|
|
projectId: true,
|
|
runtimeEnvironmentId: true,
|
|
environmentType: true,
|
|
isTest: true,
|
|
organizationId: true,
|
|
taskEventStore: true,
|
|
// Piggyback the realtime run-changed publish on this existing read (no-op when
|
|
// the native backend is disabled).
|
|
runTags: true,
|
|
batchId: true,
|
|
},
|
|
await eventReadDeps()
|
|
)
|
|
);
|
|
|
|
if (taskRunError) {
|
|
logger.error("[runExpired] Failed to find task run", {
|
|
error: taskRunError,
|
|
runId: run.id,
|
|
});
|
|
return;
|
|
}
|
|
|
|
publishChangeRecord({
|
|
runId: taskRun.id,
|
|
envId: environment.id,
|
|
tags: taskRun.runTags,
|
|
batchId: taskRun.batchId,
|
|
updatedAtMs: run.updatedAt.getTime(),
|
|
});
|
|
|
|
const eventRepository = await getEventRepositoryForStore(
|
|
taskRun.taskEventStore,
|
|
taskRun.organizationId ?? organization.id
|
|
);
|
|
|
|
const [completeExpiredRunEventError] = await tryCatch(
|
|
eventRepository.completeExpiredRunEvent({
|
|
run: taskRun,
|
|
endTime: time,
|
|
ttl: run.ttl,
|
|
})
|
|
);
|
|
|
|
if (completeExpiredRunEventError) {
|
|
logger.error("[runExpired] Failed to complete expired run event", {
|
|
error: completeExpiredRunEventError,
|
|
runId: run.id,
|
|
});
|
|
}
|
|
});
|
|
|
|
engine.eventBus.on("runCancelled", async ({ time, run, organization, environment }) => {
|
|
const [taskRunError, taskRun] = await tryCatch(
|
|
readRunForEventOrThrow(
|
|
run.id,
|
|
environment.id,
|
|
{
|
|
id: true,
|
|
friendlyId: true,
|
|
traceId: true,
|
|
spanId: true,
|
|
parentSpanId: true,
|
|
createdAt: true,
|
|
completedAt: true,
|
|
taskIdentifier: true,
|
|
projectId: true,
|
|
runtimeEnvironmentId: true,
|
|
environmentType: true,
|
|
isTest: true,
|
|
organizationId: true,
|
|
taskEventStore: true,
|
|
// Piggyback the realtime run-changed publish on this existing read (no-op when
|
|
// the native backend is disabled).
|
|
runTags: true,
|
|
batchId: true,
|
|
},
|
|
await eventReadDeps()
|
|
)
|
|
);
|
|
|
|
if (taskRunError) {
|
|
logger.error("[runCancelled] Task run not found", {
|
|
error: taskRunError,
|
|
runId: run.id,
|
|
});
|
|
return;
|
|
}
|
|
|
|
publishChangeRecord({
|
|
runId: taskRun.id,
|
|
envId: environment.id,
|
|
tags: taskRun.runTags,
|
|
batchId: taskRun.batchId,
|
|
updatedAtMs: run.updatedAt.getTime(),
|
|
});
|
|
|
|
const eventRepository = await getEventRepositoryForStore(
|
|
taskRun.taskEventStore,
|
|
taskRun.organizationId ?? organization.id
|
|
);
|
|
|
|
const error = createJsonErrorObject(run.error);
|
|
|
|
const [cancelRunEventError] = await tryCatch(
|
|
eventRepository.cancelRunEvent({
|
|
reason: error.message,
|
|
run: taskRun,
|
|
cancelledAt: time,
|
|
})
|
|
);
|
|
|
|
if (cancelRunEventError) {
|
|
logger.error("[runCancelled] Failed to cancel run event", {
|
|
error: cancelRunEventError,
|
|
runId: run.id,
|
|
});
|
|
}
|
|
});
|
|
|
|
engine.eventBus.on(
|
|
"runRetryScheduled",
|
|
async ({ time, run, environment, retryAt, organization }) => {
|
|
try {
|
|
if (retryAt && time && time >= retryAt) {
|
|
return;
|
|
}
|
|
|
|
let retryMessage = `Retry ${
|
|
typeof run.attemptNumber === "number" ? `#${run.attemptNumber - 1}` : ""
|
|
} delay`;
|
|
|
|
if (run.nextMachineAfterOOM) {
|
|
retryMessage += ` after OOM`;
|
|
}
|
|
|
|
const eventRepository = await getEventRepositoryForStore(
|
|
run.taskEventStore ?? "taskEvent",
|
|
organization.id
|
|
);
|
|
|
|
await eventRepository.recordEvent(retryMessage, {
|
|
startTime: BigInt(time.getTime() * 1000000),
|
|
taskSlug: run.taskIdentifier,
|
|
environment,
|
|
attributes: {
|
|
properties: {
|
|
retryAt: retryAt.toISOString(),
|
|
nextMachine: run.nextMachineAfterOOM,
|
|
},
|
|
runId: run.friendlyId,
|
|
style: {
|
|
icon: "schedule-attempt",
|
|
},
|
|
},
|
|
context: run.traceContext as Record<string, string | undefined>,
|
|
endTime: retryAt,
|
|
});
|
|
} catch (error) {
|
|
logger.error("[runRetryScheduled] Failed to record retry event", {
|
|
error: error instanceof Error ? error.message : error,
|
|
runId: run.id,
|
|
spanId: run.spanId,
|
|
});
|
|
}
|
|
}
|
|
);
|
|
|
|
engine.eventBus.on("runAttemptStarted", async ({ time, run, organization }) => {
|
|
try {
|
|
if (run.attemptNumber === 1 && run.baseCostInCents > 0) {
|
|
await reportInvocationUsage(organization.id, run.baseCostInCents, { runId: run.id });
|
|
}
|
|
} catch (error) {
|
|
logger.error("[runAttemptStarted] Failed to report invocation usage", {
|
|
error: error instanceof Error ? error.message : error,
|
|
runId: run.id,
|
|
orgId: organization.id,
|
|
});
|
|
}
|
|
});
|
|
|
|
engine.eventBus.on("runMetadataUpdated", async ({ time, run }) => {
|
|
const result = await findEnvironmentFromRun(run.id);
|
|
|
|
if (!result) {
|
|
logger.error("[runMetadataUpdated] Failed to find environment", { runId: run.id });
|
|
return;
|
|
}
|
|
|
|
const { environment, runTags, batchId } = result;
|
|
|
|
try {
|
|
const updateResult = await updateMetadataService.call(run.id, run.metadata, environment);
|
|
// Realtime run-changed publish, after the write so the router's hydrate sees the new
|
|
// row. A full record (env + tags + batchId + the committed updatedAt watermark), so
|
|
// feeds route by index. Nothing written here (no-op or buffered) = nothing to announce.
|
|
if (updateResult?.updatedAtMs !== undefined) {
|
|
publishChangeRecord({
|
|
runId: run.id,
|
|
envId: environment.id,
|
|
tags: runTags,
|
|
batchId,
|
|
updatedAtMs: updateResult.updatedAtMs,
|
|
});
|
|
}
|
|
} catch (e) {
|
|
if (e instanceof MetadataTooLargeError) {
|
|
logger.warn("[runMetadataUpdated] Failed to update metadata, too large", {
|
|
taskRun: run.id,
|
|
error:
|
|
e instanceof Error
|
|
? {
|
|
name: e.name,
|
|
message: e.message,
|
|
stack: e.stack,
|
|
}
|
|
: e,
|
|
});
|
|
} else {
|
|
logger.error("[runMetadataUpdated] Failed to update metadata", {
|
|
taskRun: run.id,
|
|
error:
|
|
e instanceof Error
|
|
? {
|
|
name: e.name,
|
|
message: e.message,
|
|
stack: e.stack,
|
|
}
|
|
: e,
|
|
});
|
|
}
|
|
}
|
|
});
|
|
|
|
engine.eventBus.on("executionSnapshotCreated", async ({ time, run, snapshot }) => {
|
|
const eventResult = await recordRunDebugLog(
|
|
run.id,
|
|
`[engine] ${snapshot.executionStatus} - ${snapshot.description}`,
|
|
{
|
|
attributes: {
|
|
properties: {
|
|
snapshotId: snapshot.id,
|
|
snapshotDescription: snapshot.description,
|
|
snapshotStatus: snapshot.executionStatus,
|
|
workerId: snapshot.workerId ?? undefined,
|
|
runnerId: snapshot.runnerId ?? undefined,
|
|
},
|
|
},
|
|
startTime: time,
|
|
}
|
|
);
|
|
|
|
if (!eventResult.success) {
|
|
logger.error("[executionSnapshotCreated] Failed to record event", {
|
|
runId: run.id,
|
|
snapshot,
|
|
error: eventResult.error,
|
|
});
|
|
}
|
|
});
|
|
|
|
engine.eventBus.on("workerNotification", async ({ time, run, snapshot }) => {
|
|
logger.debug("[workerNotification] Notifying worker", { time, runId: run.id, snapshot });
|
|
|
|
// Notify the worker
|
|
try {
|
|
const runFriendlyId = RunId.toFriendlyId(run.id);
|
|
const room = roomFromFriendlyRunId(runFriendlyId);
|
|
|
|
//send the notification to connected workers
|
|
socketIo.workerNamespace
|
|
.to(room)
|
|
.emit("run:notify", { version: "1", run: { friendlyId: runFriendlyId } });
|
|
|
|
//send the notification to connected dev workers
|
|
socketIo.devWorkerNamespace
|
|
.to(room)
|
|
.emit("run:notify", { version: "1", run: { friendlyId: runFriendlyId } });
|
|
|
|
if (!env.RUN_ENGINE_DEBUG_WORKER_NOTIFICATIONS) {
|
|
return;
|
|
}
|
|
|
|
// Record notification event
|
|
const eventResult = await recordRunDebugLog(
|
|
run.id,
|
|
// don't prefix this with [engine] - "run:notify" is the correct prefix
|
|
`run:notify platform -> supervisor: ${snapshot.executionStatus}`,
|
|
{
|
|
attributes: {
|
|
properties: {
|
|
snapshotId: snapshot.id,
|
|
snapshotStatus: snapshot.executionStatus,
|
|
},
|
|
},
|
|
startTime: time,
|
|
}
|
|
);
|
|
|
|
if (!eventResult.success) {
|
|
logger.error("[workerNotification] Failed to record event", {
|
|
runId: run.id,
|
|
snapshot,
|
|
error: eventResult.error,
|
|
});
|
|
}
|
|
} catch (error) {
|
|
logger.error("[workerNotification] Failed to notify worker", {
|
|
error: error instanceof Error ? error.message : error,
|
|
runId: run.id,
|
|
snapshot,
|
|
});
|
|
|
|
// Record notification event
|
|
const eventResult = await recordRunDebugLog(
|
|
run.id,
|
|
// don't prefix this with [engine] - "run:notify" is the correct prefix
|
|
`run:notify ERROR platform -> supervisor: ${snapshot.executionStatus}`,
|
|
{
|
|
attributes: {
|
|
properties: {
|
|
snapshotId: snapshot.id,
|
|
snapshotStatus: snapshot.executionStatus,
|
|
error: error instanceof Error ? error.message : String(error),
|
|
},
|
|
},
|
|
startTime: time,
|
|
}
|
|
);
|
|
|
|
if (!eventResult.success) {
|
|
logger.error("[workerNotification] Failed to record event", {
|
|
runId: run.id,
|
|
snapshot,
|
|
error: eventResult.error,
|
|
});
|
|
}
|
|
}
|
|
});
|
|
|
|
engine.eventBus.on("incomingCheckpointDiscarded", async ({ time, run, snapshot, checkpoint }) => {
|
|
const eventResult = await recordRunDebugLog(
|
|
run.id,
|
|
`[engine] Checkpoint discarded: ${checkpoint.discardReason}`,
|
|
{
|
|
attributes: {
|
|
properties: {
|
|
snapshotId: snapshot.id,
|
|
...checkpoint.metadata,
|
|
},
|
|
},
|
|
startTime: time,
|
|
}
|
|
);
|
|
|
|
if (!eventResult.success) {
|
|
logger.error("[incomingCheckpointDiscarded] Failed to record event", {
|
|
runId: run.id,
|
|
snapshot,
|
|
error: eventResult.error,
|
|
});
|
|
}
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Set up the BatchQueue processing callbacks.
|
|
* These handle creating runs from batch items and completing batches.
|
|
*
|
|
* Payload handling:
|
|
* - If payloadType is "application/store", the payload is an R2 path (already offloaded)
|
|
* - DefaultPayloadProcessor in TriggerTaskService will pass it through without re-offloading
|
|
* - The run engine will download from R2 when the task executes
|
|
*/
|
|
export function setupBatchQueueCallbacks() {
|
|
// Resolve the split-mode gate ONCE at registration scope (never per-callback).
|
|
const splitEnabledPromise = isSplitEnabled();
|
|
|
|
// Item processing callback - creates a run for each batch item
|
|
engine.setBatchProcessItemCallback(
|
|
async ({ batchId, friendlyId, itemIndex, item, meta, attempt, isFinalAttempt }) => {
|
|
return tracer.startActiveSpan(
|
|
"batch.processItem",
|
|
{
|
|
kind: SpanKind.INTERNAL,
|
|
attributes: {
|
|
"batch.id": friendlyId,
|
|
"batch.item_index": itemIndex,
|
|
"batch.task": item.task,
|
|
"batch.environment_id": meta.environmentId,
|
|
"batch.parent_run_id": meta.parentRunId ?? "",
|
|
"batch.attempt": attempt,
|
|
"batch.is_final_attempt": isFinalAttempt,
|
|
},
|
|
},
|
|
async (span) => {
|
|
// Anchor every item mint on the BATCH's friendlyId so a mid-batch mint-flag flip
|
|
// can't split an item (or pre-failed item) from its BatchTaskRun row.
|
|
const mintItemRunFriendlyId = () =>
|
|
mintAnchoredRunFriendlyId(
|
|
friendlyId,
|
|
(item.options as { region?: string } | undefined)?.region
|
|
);
|
|
|
|
const triggerFailedTaskService = new TriggerFailedTaskService({
|
|
prisma,
|
|
engine,
|
|
replicaPrisma: $replica,
|
|
});
|
|
|
|
// Check for pre-marked error items (e.g. oversized payloads)
|
|
const itemError = item.options?.__error as string | undefined;
|
|
if (itemError) {
|
|
const errorCode = (item.options?.__errorCode as string) ?? "ITEM_ERROR";
|
|
|
|
let environment: AuthenticatedEnvironment | undefined;
|
|
try {
|
|
environment = (await findEnvironmentById(meta.environmentId)) ?? undefined;
|
|
} catch {
|
|
// Best-effort environment lookup
|
|
}
|
|
|
|
if (environment) {
|
|
const failedRunId = await triggerFailedTaskService.call({
|
|
taskId: item.task,
|
|
environment,
|
|
payload: item.payload ?? "{}",
|
|
payloadType: item.payloadType as string,
|
|
errorMessage: itemError,
|
|
errorCode: errorCode as TaskRunErrorCodes,
|
|
parentRunId: meta.parentRunId,
|
|
resumeParentOnCompletion: meta.resumeParentOnCompletion,
|
|
batch: { id: batchId, index: itemIndex },
|
|
runFriendlyId: mintItemRunFriendlyId(),
|
|
traceContext: meta.traceContext as Record<string, unknown> | undefined,
|
|
spanParentAsLink: meta.spanParentAsLink,
|
|
});
|
|
|
|
if (failedRunId) {
|
|
span.setAttribute("batch.result.pre_failed", true);
|
|
span.setAttribute("batch.result.run_id", failedRunId);
|
|
span.end();
|
|
return { success: true as const, runId: failedRunId };
|
|
}
|
|
}
|
|
|
|
// Fallback if TriggerFailedTaskService or environment lookup fails
|
|
span.end();
|
|
return { success: false as const, error: itemError, errorCode };
|
|
}
|
|
|
|
let environment: AuthenticatedEnvironment | undefined;
|
|
try {
|
|
environment = (await findEnvironmentById(meta.environmentId)) ?? undefined;
|
|
|
|
if (!environment) {
|
|
span.setAttribute("batch.result.error", "Environment not found");
|
|
span.end();
|
|
|
|
return {
|
|
success: false as const,
|
|
error: "Environment not found",
|
|
errorCode: "ENVIRONMENT_NOT_FOUND",
|
|
};
|
|
}
|
|
|
|
const triggerTaskService = new TriggerTaskService();
|
|
|
|
// Normalize payload - for application/store (R2 paths), this passes through as-is
|
|
const payload = normalizePayload(item.payload, item.payloadType);
|
|
|
|
const runFriendlyId = mintItemRunFriendlyId();
|
|
|
|
const result = await triggerTaskService.call(
|
|
item.task,
|
|
environment,
|
|
{
|
|
payload,
|
|
options: {
|
|
...(item.options as Record<string, unknown>),
|
|
payloadType: item.payloadType,
|
|
parentRunId: meta.parentRunId,
|
|
resumeParentOnCompletion: meta.resumeParentOnCompletion,
|
|
parentBatch: batchId,
|
|
},
|
|
},
|
|
{
|
|
triggerVersion: meta.triggerVersion,
|
|
traceContext: meta.traceContext as Record<string, unknown> | undefined,
|
|
spanParentAsLink: meta.spanParentAsLink,
|
|
batchId,
|
|
batchIndex: itemIndex,
|
|
runFriendlyId,
|
|
realtimeStreamsVersion: meta.realtimeStreamsVersion,
|
|
planType: meta.planType,
|
|
triggerSource: meta.parentRunId ? "sdk" : (meta.triggerSource ?? "api"),
|
|
triggerAction: "trigger",
|
|
},
|
|
"V2"
|
|
);
|
|
|
|
if (result) {
|
|
span.setAttribute("batch.result.run_id", result.run.friendlyId);
|
|
span.end();
|
|
return { success: true as const, runId: result.run.friendlyId };
|
|
} else {
|
|
logger.error("[BatchQueue] TriggerTaskService returned undefined", {
|
|
batchId,
|
|
friendlyId,
|
|
itemIndex,
|
|
task: item.task,
|
|
environmentId: meta.environmentId,
|
|
attempt,
|
|
isFinalAttempt,
|
|
});
|
|
|
|
span.setAttribute("batch.result.error", "TriggerTaskService returned undefined");
|
|
|
|
// Only create a pre-failed run on the final attempt; otherwise let the retry mechanism handle it
|
|
if (isFinalAttempt) {
|
|
const failedRunId = await triggerFailedTaskService.call({
|
|
taskId: item.task,
|
|
environment,
|
|
payload: item.payload,
|
|
payloadType: item.payloadType as string,
|
|
errorMessage: "TriggerTaskService returned undefined",
|
|
parentRunId: meta.parentRunId,
|
|
resumeParentOnCompletion: meta.resumeParentOnCompletion,
|
|
batch: { id: batchId, index: itemIndex },
|
|
runFriendlyId: mintItemRunFriendlyId(),
|
|
options: item.options as Record<string, unknown>,
|
|
traceContext: meta.traceContext as Record<string, unknown> | undefined,
|
|
spanParentAsLink: meta.spanParentAsLink,
|
|
errorCode: TaskRunErrorCodes.BATCH_ITEM_COULD_NOT_TRIGGER,
|
|
});
|
|
|
|
span.end();
|
|
|
|
if (failedRunId) {
|
|
return { success: true as const, runId: failedRunId };
|
|
}
|
|
} else {
|
|
span.end();
|
|
}
|
|
|
|
return {
|
|
success: false as const,
|
|
error: "TriggerTaskService returned undefined",
|
|
errorCode: "TRIGGER_FAILED",
|
|
};
|
|
}
|
|
} catch (error) {
|
|
const errorMessage = error instanceof Error ? error.message : String(error);
|
|
|
|
// Queue-size-limit rejections are a customer-overload scenario (the
|
|
// env's queue is at its configured max). Retrying is pointless — the
|
|
// same item will fail again — and creating pre-failed TaskRuns for
|
|
// every item of every retried batch is exactly what chews through
|
|
// DB capacity when a noisy tenant fills their queue. Signal the
|
|
// BatchQueue to skip retries and skip pre-failed run creation, and
|
|
// let the completion callback collapse the per-item errors into a
|
|
// single summary row.
|
|
if (error instanceof QueueSizeLimitExceededError) {
|
|
logger.warn("[BatchQueue] Batch item rejected: queue size limit reached", {
|
|
batchId,
|
|
friendlyId,
|
|
itemIndex,
|
|
task: item.task,
|
|
environmentId: meta.environmentId,
|
|
maximumSize: error.maximumSize,
|
|
});
|
|
|
|
span.setAttribute("batch.result.error", errorMessage);
|
|
span.setAttribute("batch.result.errorCode", QUEUE_SIZE_LIMIT_EXCEEDED_ERROR_CODE);
|
|
span.setAttribute("batch.result.skipRetries", true);
|
|
span.end();
|
|
|
|
return {
|
|
success: false as const,
|
|
error: errorMessage,
|
|
errorCode: QUEUE_SIZE_LIMIT_EXCEEDED_ERROR_CODE,
|
|
skipRetries: true,
|
|
};
|
|
}
|
|
|
|
logger.error("[BatchQueue] Failed to trigger batch item", {
|
|
batchId,
|
|
friendlyId,
|
|
itemIndex,
|
|
task: item.task,
|
|
environmentId: meta.environmentId,
|
|
attempt,
|
|
isFinalAttempt,
|
|
error,
|
|
});
|
|
|
|
span.setAttribute("batch.result.error", errorMessage);
|
|
span.recordException(error instanceof Error ? error : new Error(String(error)));
|
|
|
|
// Only create a pre-failed run on the final attempt; otherwise let the retry mechanism handle it
|
|
if (isFinalAttempt && environment) {
|
|
const failedRunId = await triggerFailedTaskService.call({
|
|
taskId: item.task,
|
|
environment,
|
|
payload: item.payload,
|
|
payloadType: item.payloadType as string,
|
|
errorMessage,
|
|
parentRunId: meta.parentRunId,
|
|
resumeParentOnCompletion: meta.resumeParentOnCompletion,
|
|
batch: { id: batchId, index: itemIndex },
|
|
runFriendlyId: mintItemRunFriendlyId(),
|
|
options: item.options as Record<string, unknown>,
|
|
traceContext: meta.traceContext as Record<string, unknown> | undefined,
|
|
spanParentAsLink: meta.spanParentAsLink,
|
|
errorCode: TaskRunErrorCodes.BATCH_ITEM_COULD_NOT_TRIGGER,
|
|
});
|
|
|
|
span.end();
|
|
|
|
if (failedRunId) {
|
|
return { success: true as const, runId: failedRunId };
|
|
}
|
|
} else {
|
|
span.end();
|
|
}
|
|
|
|
return {
|
|
success: false as const,
|
|
error: errorMessage,
|
|
errorCode: "TRIGGER_ERROR",
|
|
};
|
|
}
|
|
}
|
|
);
|
|
}
|
|
);
|
|
|
|
// Batch completion callback - updates Postgres with results. The source callback
|
|
// is a thin wrapper that resolves the split-mode gate and supplies the run-ops
|
|
// handles; the body lives in handleBatchCompletion for testability.
|
|
engine.setBatchCompletionCallback(async (result: CompleteBatchResult) => {
|
|
await handleBatchCompletion(result, {
|
|
splitEnabled: await splitEnabledPromise,
|
|
newReplica: runOpsNewReplicaClient,
|
|
newWriter: runOpsNewPrismaClient,
|
|
legacyWriter: runOpsLegacyPrismaClient,
|
|
tryCompleteBatch: (batchId) => engine.tryCompleteBatch({ batchId }),
|
|
});
|
|
});
|
|
|
|
logger.info("BatchQueue callbacks configured");
|
|
}
|
|
|
|
/**
|
|
* Normalize the payload from BatchQueue.
|
|
*
|
|
* Handles different payload types:
|
|
* - "application/store": Already offloaded to R2, payload is the path - pass through as-is
|
|
* - "application/json": May be a pre-serialized JSON string - parse to avoid double-stringification
|
|
* - Other types: Pass through as-is
|
|
*
|
|
* @param payload - The raw payload from the batch item
|
|
* @param payloadType - The payload type (e.g., "application/json", "application/store")
|
|
*/
|
|
function normalizePayload(payload: unknown, payloadType?: string): unknown {
|
|
// Only process "application/json" payloads
|
|
// For all other types (including undefined), return as-is
|
|
if (payloadType !== "application/json") {
|
|
return payload;
|
|
}
|
|
|
|
// For JSON payloads, if payload is a string, try to parse it
|
|
// This handles pre-serialized JSON from the SDK
|
|
if (typeof payload === "string") {
|
|
try {
|
|
return JSON.parse(payload);
|
|
} catch {
|
|
// If it's not valid JSON, return as-is
|
|
return payload;
|
|
}
|
|
}
|
|
|
|
return payload;
|
|
}
|