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

provide worker files as part of the worker creation on the server

Browse Source
This commit is contained in:
Eric Allam
2024-08-11 20:25:36 +01:00
committed by Eric Allam
parent 2b85d53028
commit 6dea6e1c7d
6 changed files with 213 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/v3/services/createBackgroundWorker.server.ts
+71 -3
View File
@@ -1,4 +1,8 @@
import { CreateBackgroundWorkerRequestBody, TaskResource } from "@trigger.dev/core/v3";
import {
BackgroundWorkerFileMetadata,
CreateBackgroundWorkerRequestBody,
TaskResource,
} from "@trigger.dev/core/v3";
import type { BackgroundWorker } from "@trigger.dev/database";
import { Prisma, PrismaClientOrTransaction } from "~/db.server";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
@@ -70,7 +74,19 @@ export class CreateBackgroundWorkerService extends BaseService {
},
});
await createBackgroundTasks(body.metadata.tasks, backgroundWorker, environment, this._prisma);
const tasksToBackgroundFiles = await createBackgroundFiles(
body.metadata.fileContents,
backgroundWorker,
environment,
this._prisma
);
await createBackgroundTasks(
body.metadata.tasks,
backgroundWorker,
environment,
this._prisma,
tasksToBackgroundFiles
);
await syncDeclarativeSchedules(
body.metadata.tasks,
backgroundWorker,
@@ -121,7 +137,8 @@ export async function createBackgroundTasks(
tasks: TaskResource[],
worker: BackgroundWorker,
environment: AuthenticatedEnvironment,
prisma: PrismaClientOrTransaction
prisma: PrismaClientOrTransaction,
tasksToBackgroundFiles?: Map<string, string>
) {
for (const task of tasks) {
try {
@@ -138,6 +155,7 @@ export async function createBackgroundTasks(
queueConfig: task.queue,
machineConfig: task.machine,
triggerSource: task.triggerSource === "schedule" ? "SCHEDULED" : "STANDARD",
fileId: tasksToBackgroundFiles?.get(task.id) ?? null,
},
});
@@ -381,3 +399,53 @@ export async function syncDeclarativeSchedules(
}
}
}
export async function createBackgroundFiles(
files: Array<BackgroundWorkerFileMetadata> | undefined,
worker: BackgroundWorker,
environment: AuthenticatedEnvironment,
prisma: PrismaClientOrTransaction
) {
// Maps from each taskId to the backgroundWorkerFileId
const results = new Map<string, string>();
if (!files) {
return results;
}
for (const file of files) {
const backgroundWorkerFile = await prisma.backgroundWorkerFile.upsert({
where: {
projectId_contentHash: {
projectId: environment.projectId,
contentHash: file.contentHash,
},
},
create: {
friendlyId: generateFriendlyId("file"),
projectId: environment.projectId,
contentHash: file.contentHash,
filePath: file.filePath,
contents: Buffer.from(file.contents),
backgroundWorkers: {
connect: {
id: worker.id,
},
},
},
update: {
backgroundWorkers: {
connect: {
id: worker.id,
},
},
},
});
for (const taskId of file.taskIds) {
results.set(taskId, backgroundWorkerFile.id);
}
}
return results;
}
packages/cli-v3/src/dev/workerRuntime.ts
+44
View File
@@ -26,6 +26,9 @@ import { logger } from "../utilities/logger.js";
import { VERSION } from "../version.js";
import { BackgroundWorker, BackgroundWorkerCoordinator } from "./backgroundWorker.js";
import { getInstrumentedPackageNames } from "../build/instrumentation.js";
import { readFile } from "node:fs/promises";
import { join } from "node:path";
import { createHash } from "node:crypto";
export interface WorkerRuntime {
shutdown(): Promise<void>;
@@ -183,6 +186,11 @@ class DevWorkerRuntime implements WorkerRuntime {
return;
}
const fileContents = await this.#fetchTaskFiles(
backgroundWorker.manifest.tasks,
this.options.config.workingDir
);
const backgroundWorkerBody: CreateBackgroundWorkerRequestBody = {
localOnly: true,
metadata: {
@@ -190,6 +198,7 @@ class DevWorkerRuntime implements WorkerRuntime {
cliPackageVersion: VERSION,
tasks: backgroundWorker.manifest.tasks,
contentHash: manifest.contentHash,
fileContents,
},
supportsLazyAttempts: true,
};
@@ -210,6 +219,41 @@ class DevWorkerRuntime implements WorkerRuntime {
eventBus.emit("backgroundWorkerInitialized", backgroundWorker);
}
async #fetchTaskFiles(tasks: TaskManifest[], workingDir: string) {
const tasksGroupedByFile: Record<string, TaskManifest[]> = {};
for (const task of tasks) {
if (!tasksGroupedByFile[task.filePath]) {
tasksGroupedByFile[task.filePath] = [];
}
tasksGroupedByFile[task.filePath]!.push(task);
}
const taskFiles: Array<{
taskIds: string[];
contents: string;
contentHash: string;
filePath: string;
}> = [];
for (const [filePath, tasks] of Object.entries(tasksGroupedByFile)) {
const contents = await readFile(join(workingDir, filePath), "utf-8");
const taskIds = tasks.map((task) => task.id);
const hasher = createHash("md5");
hasher.update(contents);
taskFiles.push({
filePath,
taskIds,
contents,
contentHash: hasher.digest("hex"),
});
}
return taskFiles;
}
async #getEnvVars(): Promise<Record<string, string>> {
const environmentVariablesResponse = await this.options.client.getEnvironmentVariables(
this.options.config.project
packages/core/src/v3/schemas/resources.ts
+10
View File
@@ -15,11 +15,21 @@ export const TaskResource = z.object({
export type TaskResource = z.infer<typeof TaskResource>;
export const BackgroundWorkerFileMetadata = z.object({
filePath: z.string(),
contents: z.string(),
contentHash: z.string(),
taskIds: z.array(z.string()),
});
export type BackgroundWorkerFileMetadata = z.infer<typeof BackgroundWorkerFileMetadata>;
export const BackgroundWorkerMetadata = z.object({
packageVersion: z.string(),
contentHash: z.string(),
cliPackageVersion: z.string().optional(),
tasks: z.array(TaskResource),
fileContents: z.array(BackgroundWorkerFileMetadata).optional(),
});
export type BackgroundWorkerMetadata = z.infer<typeof BackgroundWorkerMetadata>;
packages/database/prisma/migrations/20240810090402_add_background_worker_file_model/migration.sql
+24
View File
@@ -0,0 +1,24 @@
-- AlterTable
ALTER TABLE "BackgroundWorkerTask" ADD COLUMN "fileId" TEXT;
-- CreateTable
CREATE TABLE "BackgroundWorkerFile" (
"id" TEXT NOT NULL,
"friendlyId" TEXT NOT NULL,
"filePath" TEXT NOT NULL,
"contentHash" TEXT NOT NULL,
"contents" BYTEA NOT NULL,
"backgroundWorkerId" TEXT NOT NULL,
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "BackgroundWorkerFile_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE UNIQUE INDEX "BackgroundWorkerFile_friendlyId_key" ON "BackgroundWorkerFile"("friendlyId");
-- AddForeignKey
ALTER TABLE "BackgroundWorkerFile" ADD CONSTRAINT "BackgroundWorkerFile_backgroundWorkerId_fkey" FOREIGN KEY ("backgroundWorkerId") REFERENCES "BackgroundWorker"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "BackgroundWorkerTask" ADD CONSTRAINT "BackgroundWorkerTask_fileId_fkey" FOREIGN KEY ("fileId") REFERENCES "BackgroundWorkerFile"("id") ON DELETE CASCADE ON UPDATE CASCADE;
packages/database/prisma/migrations/20240811185335_improve_background_worker_file_model/migration.sql
+38
View File
@@ -0,0 +1,38 @@
/*
Warnings:
- You are about to drop the column `backgroundWorkerId` on the `BackgroundWorkerFile` table. All the data in the column will be lost.
- A unique constraint covering the columns `[projectId,contentHash]` on the table `BackgroundWorkerFile` will be added. If there are existing duplicate values, this will fail.
- Added the required column `projectId` to the `BackgroundWorkerFile` table without a default value. This is not possible if the table is not empty.
*/
-- DropForeignKey
ALTER TABLE "BackgroundWorkerFile" DROP CONSTRAINT "BackgroundWorkerFile_backgroundWorkerId_fkey";
-- AlterTable
ALTER TABLE "BackgroundWorkerFile" DROP COLUMN "backgroundWorkerId",
ADD COLUMN "projectId" TEXT NOT NULL;
-- CreateTable
CREATE TABLE "_BackgroundWorkerToBackgroundWorkerFile" (
"A" TEXT NOT NULL,
"B" TEXT NOT NULL
);
-- CreateIndex
CREATE UNIQUE INDEX "_BackgroundWorkerToBackgroundWorkerFile_AB_unique" ON "_BackgroundWorkerToBackgroundWorkerFile"("A", "B");
-- CreateIndex
CREATE INDEX "_BackgroundWorkerToBackgroundWorkerFile_B_index" ON "_BackgroundWorkerToBackgroundWorkerFile"("B");
-- CreateIndex
CREATE UNIQUE INDEX "BackgroundWorkerFile_projectId_contentHash_key" ON "BackgroundWorkerFile"("projectId", "contentHash");
-- AddForeignKey
ALTER TABLE "BackgroundWorkerFile" ADD CONSTRAINT "BackgroundWorkerFile_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "Project"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "_BackgroundWorkerToBackgroundWorkerFile" ADD CONSTRAINT "_BackgroundWorkerToBackgroundWorkerFile_A_fkey" FOREIGN KEY ("A") REFERENCES "BackgroundWorker"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "_BackgroundWorkerToBackgroundWorkerFile" ADD CONSTRAINT "_BackgroundWorkerToBackgroundWorkerFile_B_fkey" FOREIGN KEY ("B") REFERENCES "BackgroundWorkerFile"("id") ON DELETE CASCADE ON UPDATE CASCADE;
packages/database/prisma/schema.prisma
+26
View File
@@ -464,6 +464,7 @@ model Project {
alerts ProjectAlert[]
alertStorages ProjectAlertStorage[]
bulkActionGroups BulkActionGroup[]
BackgroundWorkerFile BackgroundWorkerFile[]
}
enum ProjectVersion {
@@ -1566,6 +1567,7 @@ model BackgroundWorker {
tasks BackgroundWorkerTask[]
attempts TaskRunAttempt[]
lockedRuns TaskRun[]
files BackgroundWorkerFile[]
deployment WorkerDeployment?
@@ -1574,6 +1576,27 @@ model BackgroundWorker {
@@unique([projectId, runtimeEnvironmentId, version])
}
model BackgroundWorkerFile {
id String @id @default(cuid())
friendlyId String @unique
filePath String
contentHash String
contents Bytes
project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade)
projectId String
backgroundWorkers BackgroundWorker[]
tasks BackgroundWorkerTask[]
createdAt DateTime @default(now())
@@unique([projectId, contentHash])
}
model BackgroundWorkerTask {
id String @id @default(cuid())
slug String
@@ -1589,6 +1612,9 @@ model BackgroundWorkerTask {
project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade)
projectId String
file BackgroundWorkerFile? @relation(fields: [fileId], references: [id], onDelete: Cascade, onUpdate: Cascade)
fileId String?
runtimeEnvironment RuntimeEnvironment @relation(fields: [runtimeEnvironmentId], references: [id], onDelete: Cascade, onUpdate: Cascade)
runtimeEnvironmentId String