github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Fix “transaction already closed” error when running tasks (#735)

Browse Source
This commit is contained in:
Eric Allam
2023-11-14 23:22:34 +00:00
committed by GitHub
parent 5dd1bcc589
commit 14429b8b66
9 changed files with 277 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/db.server.ts
+1 -1
View File
@@ -48,7 +48,7 @@ export async function $transaction<R>(
return await (prisma as PrismaClient).$transaction(fn, options);
} catch (error) {
if (isPrismaKnownError(error)) {
logger.debug("prisma.$transaction error", {
logger.error("prisma.$transaction error", {
code: error.code,
meta: error.meta,
stack: error.stack,
apps/webapp/app/models/task.server.ts
+4 -1
View File
@@ -1,7 +1,10 @@
import type { JobRun, Task, TaskAttempt } from "@trigger.dev/database";
import { CachedTask, ServerTask } from "@trigger.dev/core";
export type TaskWithAttempts = Task & { attempts: TaskAttempt[]; run: JobRun };
export type TaskWithAttempts = Task & {
attempts: TaskAttempt[];
run: { forceYieldImmediately: boolean };
};
export function taskWithAttemptsToServerTask(task: TaskWithAttempts): ServerTask {
return {
apps/webapp/app/platform/zodWorker.server.ts
+17 -4
View File
@@ -14,7 +14,7 @@ import { run as graphileRun, parseCronItems } from "graphile-worker";
import omit from "lodash.omit";
import { z } from "zod";
import { PrismaClient, PrismaClientOrTransaction } from "~/db.server";
import { workerLogger as logger } from "~/services/logger.server";
import { workerLogger as logger, trace } from "~/services/logger.server";
export interface MessageCatalogSchema {
[key: string]: z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion<any, any>;
@@ -270,7 +270,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
spec,
});
const job = await this.#addJob(
const { job, durationInMs } = await this.#addJob(
identifier as string,
payload,
spec,
@@ -282,6 +282,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
payload,
spec,
job,
durationInMs,
});
return job;
@@ -304,6 +305,8 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
spec: TaskSpec,
tx: PrismaClientOrTransaction
) {
const now = performance.now();
const results = await tx.$queryRawUnsafe(
`SELECT * FROM ${this.graphileWorkerSchema}.add_job(
identifier => $1::text,
@@ -327,6 +330,8 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
spec.jobKeyMode || null
);
const durationInMs = performance.now() - now;
const rows = AddJobResultsSchema.safeParse(results);
if (!rows.success) {
@@ -337,7 +342,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
const job = rows.data[0];
return job as GraphileJob;
return { job: job as GraphileJob, durationInMs: Math.floor(durationInMs) };
}
async #removeJob(jobKey: string, tx: PrismaClientOrTransaction) {
@@ -471,7 +476,15 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
throw new Error(`No task for message type: ${String(typeName)}`);
}
await task.handler(payload, job);
await trace(
{
worker_job: job,
worker_name: this.#name,
},
async () => {
await task.handler(payload, job);
}
);
}
async #handleRecurringTask(
apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts
+3 -189
View File
@@ -2,21 +2,16 @@ import type { ActionFunctionArgs } from "@remix-run/server-runtime";
import { json } from "@remix-run/server-runtime";
import {
API_VERSIONS,
RunTaskBodyOutput,
RunTaskBodyOutputSchema,
RunTaskResponseWithCachedTasksBody,
ServerTask,
} from "@trigger.dev/core";
import { TaskStatus } from "@trigger.dev/database";
import { z } from "zod";
import { $transaction, PrismaClient, prisma } from "~/db.server";
import { env } from "~/env.server";
import { prepareTasksForCaching, taskWithAttemptsToServerTask } from "~/models/task.server";
import { PrismaClient, prisma } from "~/db.server";
import { prepareTasksForCaching } from "~/models/task.server";
import { authenticateApiRequest } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
import { generateSecret } from "~/services/sources/utils.server";
import { ulid } from "~/services/ulid.server";
import { taskOperationWorker, workerQueue } from "~/services/worker.server";
import { RunTaskService } from "~/services/tasks/runTask.server";
const ParamsSchema = z.object({
runId: z.string(),
@@ -161,184 +156,3 @@ class ChangeRequestLazyLoadedCachedTasks {
};
}
}
export class RunTaskService {
#prismaClient: PrismaClient;
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
public async call(
runId: string,
idempotencyKey: string,
taskBody: RunTaskBodyOutput
): Promise<ServerTask | undefined> {
const task = await $transaction(this.#prismaClient, async (tx) => {
const existingTask = await tx.task.findUnique({
where: {
runId_idempotencyKey: {
runId,
idempotencyKey,
},
},
include: {
attempts: true,
run: true,
},
});
const delayUntilInFuture = taskBody.delayUntil && taskBody.delayUntil.getTime() > Date.now();
const callbackEnabled = taskBody.callback?.enabled;
if (existingTask) {
if (existingTask.status === "CANCELED") {
const existingTaskStatus =
delayUntilInFuture || callbackEnabled || taskBody.trigger
? "WAITING"
: taskBody.noop
? "COMPLETED"
: "RUNNING";
const resumedExistingTask = await tx.task.update({
where: {
id: existingTask.id,
},
data: {
status: existingTaskStatus,
startedAt: new Date(),
completedAt: existingTaskStatus === "COMPLETED" ? new Date() : undefined,
},
include: {
run: true,
attempts: true,
},
});
return resumedExistingTask;
}
return existingTask;
}
const run = await tx.jobRun.findUnique({
where: {
id: runId,
},
select: {
status: true,
},
});
if (!run) throw new Error("Run not found");
// If task.delayUntil is set and is in the future, we'll set the task's status to "WAITING", else set it to RUNNING
let status: TaskStatus;
if (run.status === "CANCELED") {
status = "CANCELED";
} else {
status =
delayUntilInFuture || callbackEnabled || taskBody.trigger
? "WAITING"
: taskBody.noop
? "COMPLETED"
: "RUNNING";
}
const taskId = ulid();
const callbackUrl = callbackEnabled
? `${env.APP_ORIGIN}/api/v1/tasks/${taskId}/callback/${generateSecret(12)}`
: undefined;
const task = await tx.task.create({
data: {
id: taskId,
idempotencyKey,
displayKey: taskBody.displayKey,
runConnection: taskBody.connectionKey
? {
connect: {
runId_key: {
runId,
key: taskBody.connectionKey,
},
},
}
: undefined,
icon: taskBody.icon,
run: {
connect: {
id: runId,
},
},
parent: taskBody.parentId ? { connect: { id: taskBody.parentId } } : undefined,
name: taskBody.name ?? "Task",
description: taskBody.description,
status,
startedAt: new Date(),
completedAt: status === "COMPLETED" || status === "CANCELED" ? new Date() : undefined,
noop: taskBody.noop,
delayUntil: taskBody.delayUntil,
params: taskBody.params ?? undefined,
properties: this.#filterProperties(taskBody.properties) ?? undefined,
redact: taskBody.redact ?? undefined,
operation: taskBody.operation,
callbackUrl,
style: taskBody.style ?? { style: "normal" },
childExecutionMode: taskBody.parallel ? "PARALLEL" : "SEQUENTIAL",
attempts: {
create: {
number: 1,
status: "PENDING",
},
},
},
include: {
run: true,
attempts: true,
},
});
if (task.status === "RUNNING" && typeof taskBody.operation === "string") {
// We need to schedule the operation
await taskOperationWorker.enqueue(
"performTaskOperation",
{
id: task.id,
},
{ tx, runAt: task.delayUntil ?? undefined, jobKey: `operation:${task.id}` }
);
} else if (task.status === "WAITING" && callbackUrl && taskBody.callback) {
if (taskBody.callback.timeoutInSeconds > 0) {
// We need to schedule the callback timeout
await workerQueue.enqueue(
"processCallbackTimeout",
{
id: task.id,
},
{
tx,
runAt: new Date(Date.now() + taskBody.callback.timeoutInSeconds * 1000),
jobKey: `process-callback:${task.id}`,
}
);
}
}
return task;
});
return task ? taskWithAttemptsToServerTask(task) : undefined;
}
#filterProperties(properties: RunTaskBodyOutput["properties"]): RunTaskBodyOutput["properties"] {
if (!properties) return;
return properties.filter((property) => {
if (!property) return false;
return typeof property.label === "string" && typeof property.text === "string";
});
}
}
apps/webapp/app/services/logger.server.ts
+17 -3
View File
@@ -1,18 +1,32 @@
import type { LogLevel } from "@trigger.dev/core";
import { Logger } from "@trigger.dev/core";
import { sensitiveDataReplacer } from "./sensitiveDataReplacer";
import { singleton } from "~/utils/singleton";
import { AsyncLocalStorage } from "async_hooks";
const currentFieldsStore = new AsyncLocalStorage<Record<string, unknown>>();
export function trace<T>(fields: Record<string, unknown>, fn: () => T): T {
return currentFieldsStore.run(fields, fn);
}
export const logger = new Logger(
"webapp",
(process.env.APP_LOG_LEVEL ?? "debug") as LogLevel,
["examples", "output", "connectionString", "payload"],
sensitiveDataReplacer
sensitiveDataReplacer,
() => {
const fields = currentFieldsStore.getStore();
return fields ? { ...fields } : {};
}
);
export const workerLogger = new Logger(
"worker",
(process.env.APP_LOG_LEVEL ?? "debug") as LogLevel,
["examples", "output", "connectionString"],
sensitiveDataReplacer
sensitiveDataReplacer,
() => {
const fields = currentFieldsStore.getStore();
return fields ? { ...fields } : {};
}
);
apps/webapp/app/services/tasks/runTask.server.ts
+220
View File
@@ -0,0 +1,220 @@
import { RunTaskBodyOutput, ServerTask } from "@trigger.dev/core";
import { TaskStatus } from "@trigger.dev/database";
import { $transaction, PrismaClient, prisma } from "~/db.server";
import { env } from "~/env.server";
import { taskWithAttemptsToServerTask } from "~/models/task.server";
import { generateSecret } from "~/services/sources/utils.server";
import { ulid } from "~/services/ulid.server";
import { taskOperationWorker, workerQueue } from "~/services/worker.server";
export class RunTaskService {
#prismaClient: PrismaClient;
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
public async call(
runId: string,
idempotencyKey: string,
taskBody: RunTaskBodyOutput
): Promise<ServerTask | undefined> {
const delayUntilInFuture = taskBody.delayUntil
? taskBody.delayUntil.getTime() > Date.now()
: false;
const callbackEnabled = taskBody.callback?.enabled ?? false;
// First
const existingTask = await this.#handleExistingTask(
runId,
idempotencyKey,
taskBody,
delayUntilInFuture,
callbackEnabled
);
if (existingTask) {
return taskWithAttemptsToServerTask(existingTask);
}
const run = await this.#prismaClient.jobRun.findUnique({
where: {
id: runId,
},
select: {
status: true,
forceYieldImmediately: true,
},
});
if (!run) throw new Error("Run not found");
const runConnection = taskBody.connectionKey
? await this.#prismaClient.runConnection.findUnique({
where: {
runId_key: {
runId,
key: taskBody.connectionKey,
},
},
select: {
id: true,
},
})
: undefined;
const results = await $transaction(this.#prismaClient, async (tx) => {
// If task.delayUntil is set and is in the future, we'll set the task's status to "WAITING", else set it to RUNNING
let status: TaskStatus;
if (run.status === "CANCELED") {
status = "CANCELED";
} else {
status =
delayUntilInFuture || callbackEnabled || taskBody.trigger
? "WAITING"
: taskBody.noop
? "COMPLETED"
: "RUNNING";
}
const taskId = ulid();
const callbackUrl = callbackEnabled
? `${env.APP_ORIGIN}/api/v1/tasks/${taskId}/callback/${generateSecret(12)}`
: undefined;
const task = await tx.task.create({
data: {
id: taskId,
idempotencyKey,
displayKey: taskBody.displayKey,
runConnectionId: runConnection ? runConnection.id : undefined,
icon: taskBody.icon,
runId,
parentId: taskBody.parentId,
name: taskBody.name ?? "Task",
description: taskBody.description,
status,
startedAt: new Date(),
completedAt: status === "COMPLETED" || status === "CANCELED" ? new Date() : undefined,
noop: taskBody.noop,
delayUntil: taskBody.delayUntil,
params: taskBody.params ?? undefined,
properties: this.#filterProperties(taskBody.properties) ?? undefined,
redact: taskBody.redact ?? undefined,
operation: taskBody.operation,
callbackUrl,
style: taskBody.style ?? { style: "normal" },
childExecutionMode: taskBody.parallel ? "PARALLEL" : "SEQUENTIAL",
},
});
const taskAttempt = await tx.taskAttempt.create({
data: {
number: 1,
taskId: task.id,
status: "PENDING",
},
});
if (task.status === "RUNNING" && typeof taskBody.operation === "string") {
// We need to schedule the operation
await taskOperationWorker.enqueue(
"performTaskOperation",
{
id: task.id,
},
{ tx, runAt: task.delayUntil ?? undefined, jobKey: `operation:${task.id}` }
);
} else if (task.status === "WAITING" && callbackUrl && taskBody.callback) {
if (taskBody.callback.timeoutInSeconds > 0) {
// We need to schedule the callback timeout
await workerQueue.enqueue(
"processCallbackTimeout",
{
id: task.id,
},
{
tx,
runAt: new Date(Date.now() + taskBody.callback.timeoutInSeconds * 1000),
jobKey: `process-callback:${task.id}`,
}
);
}
}
return { task, taskAttempt };
});
if (!results) {
return;
}
const { task, taskAttempt } = results;
return task
? taskWithAttemptsToServerTask({ ...task, attempts: [taskAttempt], run })
: undefined;
}
async #handleExistingTask(
runId: string,
idempotencyKey: string,
taskBody: RunTaskBodyOutput,
delayUntilInFuture: boolean,
callbackEnabled: boolean
) {
const existingTask = await this.#prismaClient.task.findUnique({
where: {
runId_idempotencyKey: {
runId,
idempotencyKey,
},
},
include: {
attempts: true,
run: true,
},
});
if (existingTask) {
if (existingTask.status === "CANCELED") {
const existingTaskStatus =
delayUntilInFuture || callbackEnabled || taskBody.trigger
? "WAITING"
: taskBody.noop
? "COMPLETED"
: "RUNNING";
const resumedExistingTask = await this.#prismaClient.task.update({
where: {
id: existingTask.id,
},
data: {
status: existingTaskStatus,
startedAt: new Date(),
completedAt: existingTaskStatus === "COMPLETED" ? new Date() : undefined,
},
include: {
run: true,
attempts: true,
},
});
return resumedExistingTask;
}
return existingTask;
}
}
#filterProperties(properties: RunTaskBodyOutput["properties"]): RunTaskBodyOutput["properties"] {
if (!properties) return;
return properties.filter((property) => {
if (!property) return false;
return typeof property.label === "string" && typeof property.text === "string";
});
}
}
packages/core/src/logger.ts
+5 -1
View File
@@ -17,17 +17,20 @@ export class Logger {
readonly #level: number;
#filteredKeys: string[] = [];
#jsonReplacer?: (key: string, value: unknown) => unknown;
#additionalFields: () => Record<string, unknown>;
constructor(
name: string,
level: LogLevel = "info",
filteredKeys: string[] = [],
jsonReplacer?: (key: string, value: unknown) => unknown
jsonReplacer?: (key: string, value: unknown) => unknown,
additionalFields?: () => Record<string, unknown>
) {
this.#name = name;
this.#level = logLevels.indexOf((process.env.TRIGGER_LOG_LEVEL ?? level) as LogLevel);
this.#filteredKeys = filteredKeys;
this.#jsonReplacer = createReplacer(jsonReplacer);
this.#additionalFields = additionalFields ?? (() => ({}));
}
// Return a new Logger instance with the same name and a new log level
@@ -78,6 +81,7 @@ export class Logger {
) {
const structuredLog = {
...structureArgs(safeJsonClone(args) as Record<string, unknown>[], this.#filteredKeys),
...this.#additionalFields(),
timestamp: new Date(),
name: this.#name,
message,
pnpm-lock.yaml
Generated
+2 -6
View File
@@ -18207,10 +18207,6 @@ packages:
dependencies:
safe-buffer: 5.2.1
/content-type/1.0.4:
resolution: {integrity: sha512-hIP3EEPs8tB9AT1L+NUqtwOAps4mk2Zob89MWXMHjHWg9milF/j4osnnQLXBCBFBk/tvIG/tUc9mOUJiPBhPXA==}
engines: {node: '>= 0.6'}
/content-type/1.0.5:
resolution: {integrity: sha512-nTjqfcBFEipKdXCv4YDQWCfmcLZKm81ldF0pAopTvyrFGVbcR6P/VAAd5G7N+0tTr8QqiU0tFadD6FK4NtJwOA==}
engines: {node: '>= 0.6'}
@@ -21253,7 +21249,7 @@ packages:
array-flatten: 1.1.1
body-parser: 1.20.1
content-disposition: 0.5.4
content-type: 1.0.4
content-type: 1.0.5
cookie: 0.5.0
cookie-signature: 1.0.6
debug: 2.6.9
@@ -31854,7 +31850,7 @@ packages:
dependencies:
bs-logger: 0.2.6
fast-json-stable-stringify: 2.1.0
jest: 29.6.2_@types+node@16.18.11
jest: 29.6.2_@types+node@18.15.13
jest-util: 29.6.2
json5: 2.2.3
lodash.memoize: 4.1.2
references/job-catalog/src/edge-cases.ts
+8
View File
@@ -92,6 +92,14 @@ client.defineJob({
throw new Error(`Expected string, got ${typeof result9}: ${JSON.stringify(result9)}`);
}
const result10 = await io.runTask("big-json", async (task) => {
return fetch("https://jsonplaceholder.typicode.com/photos").then((res) => res.json());
});
if (!Array.isArray(result10)) {
throw new Error(`Expected array, got ${typeof result10}: ${JSON.stringify(result10)}`);
}
await io.wait("wait-1", 1);
},
});