From 17391144901194e05ff96a327d09caad0588fc12 Mon Sep 17 00:00:00 2001 From: Keith Williams Date: Mon, 1 Sep 2025 04:08:06 -0300 Subject: [PATCH] perf: separate task creation from processing to eliminate compilation overhead (#23485) * refactor: separate task creation from processing to eliminate compilation overhead - Create TaskProcessor class to handle task processing logic - Move processQueue and cleanup methods to TaskProcessor - Remove task handler imports from InternalTasker creation path - Eliminate 2.5-3s compilation overhead by only loading task handlers during processing - Maintain existing tasker.create() API for all callers - Use composition pattern for clean separation of concerns Co-Authored-By: keith@cal.com refactor: completely separate TaskProcessor from InternalTasker - Remove processQueue() and cleanup() methods from Tasker interface - Eliminate TaskProcessor dependency from InternalTasker class - Update API endpoints (cron.ts, cleanup.ts) to use TaskProcessor directly - Update README documentation to show correct TaskProcessor usage - Complete separation ensures task creation no longer imports task handlers - Eliminates 2.5-3s compilation overhead while preserving all functionality Co-Authored-By: keith@cal.com * Moved the cleanup method back to internal tasker --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- packages/features/tasker/README.md | 10 +++-- packages/features/tasker/api/cron.ts | 5 ++- packages/features/tasker/internal-tasker.ts | 33 +--------------- packages/features/tasker/task-processor.ts | 42 +++++++++++++++++++++ packages/features/tasker/tasker.ts | 1 - 5 files changed, 52 insertions(+), 39 deletions(-) create mode 100644 packages/features/tasker/task-processor.ts diff --git a/packages/features/tasker/README.md b/packages/features/tasker/README.md index 91f31d6b3c..bb063991ad 100644 --- a/packages/features/tasker/README.md +++ b/packages/features/tasker/README.md @@ -45,11 +45,12 @@ For simplicity sake will explain how the `InternalTasker` works: ```ts // /app/api/tasks/cron/route.ts - import tasker from "@calcom/features/tasker"; + import { TaskProcessor } from "@calcom/features/tasker/task-processor"; export async function GET() { // authenticate the call... - await tasker.processQueue(); + const processor = new TaskProcessor(); + await processor.processQueue(); return Response.json({ success: true }); } ``` @@ -62,11 +63,12 @@ For simplicity sake will explain how the `InternalTasker` works: ```ts // /app/api/tasks/cleanup/route.ts - import tasker from "@calcom/features/tasker"; + import { TaskProcessor } from "@calcom/features/tasker/task-processor"; export async function GET() { // authenticate the call... - await tasker.cleanup(); + const processor = new TaskProcessor(); + await processor.cleanup(); return Response.json({ success: true }); } ``` diff --git a/packages/features/tasker/api/cron.ts b/packages/features/tasker/api/cron.ts index df2717f120..79c48262fb 100644 --- a/packages/features/tasker/api/cron.ts +++ b/packages/features/tasker/api/cron.ts @@ -1,14 +1,15 @@ import type { NextRequest } from "next/server"; import { NextResponse } from "next/server"; -import tasker from ".."; +import { TaskProcessor } from "../task-processor"; async function handler(request: NextRequest) { const authHeader = request.headers.get("authorization"); if (authHeader !== `Bearer ${process.env.CRON_SECRET}`) { return new Response("Unauthorized", { status: 401 }); } - await tasker.processQueue(); + const processor = new TaskProcessor(); + await processor.processQueue(); return NextResponse.json({ success: true }); } diff --git a/packages/features/tasker/internal-tasker.ts b/packages/features/tasker/internal-tasker.ts index bf667eef04..cc357f1bad 100644 --- a/packages/features/tasker/internal-tasker.ts +++ b/packages/features/tasker/internal-tasker.ts @@ -1,7 +1,6 @@ import { Task } from "./repository"; import type { TaskTypes } from "./tasker"; import { type TaskerCreate, type Tasker } from "./tasker"; -import tasksMap, { tasksConfig } from "./tasks"; /** * This is the default internal Tasker that uses the Task repository to create tasks. @@ -14,42 +13,12 @@ export class InternalTasker implements Tasker { const payloadString = typeof payload === "string" ? payload : JSON.stringify(payload); return Task.create(type, payloadString, options); }; - async processQueue(): Promise { - const tasks = await Task.getNextBatch(); - console.info(`Processing ${tasks.length} tasks`, tasks); - const tasksPromises = tasks.map(async (task) => { - console.info( - `Processing task ${task.id}, attempt:${task.attempts} maxAttempts:${task.maxAttempts} lastFailedAttempt:${task.lastFailedAttemptAt}`, - task - ); - const taskHandlerGetter = tasksMap[task.type as keyof typeof tasksMap]; - if (!taskHandlerGetter) throw new Error(`Task handler not found for type ${task.type}`); - const taskConfig = tasksConfig[task.type as keyof typeof tasksConfig]; - const taskHandler = await taskHandlerGetter(); - return taskHandler(task.payload) - .then(async () => { - await Task.succeed(task.id); - }) - .catch(async (error) => { - console.info(`Retrying task ${task.id}: ${error}`); - await Task.retry({ - taskId: task.id, - lastError: error instanceof Error ? error.message : "Unknown error", - minRetryIntervalMins: - taskConfig && "minRetryIntervalMins" in taskConfig ? taskConfig.minRetryIntervalMins : null, - }); - }); - }); - const settled = await Promise.allSettled(tasksPromises); - const failed = settled.filter((result) => result.status === "rejected"); - const succeded = settled.filter((result) => result.status === "fulfilled"); - console.info({ failed, succeded }); - } async cleanup(): Promise { const count = await Task.cleanup(); console.info(`Cleaned up ${count} tasks`); } + async cancel(id: string): Promise { const task = await Task.cancel(id); return task.id; diff --git a/packages/features/tasker/task-processor.ts b/packages/features/tasker/task-processor.ts new file mode 100644 index 0000000000..491b696c30 --- /dev/null +++ b/packages/features/tasker/task-processor.ts @@ -0,0 +1,42 @@ +import { Task } from "./repository"; +import tasksMap, { tasksConfig } from "./tasks"; + +/** + * TaskProcessor handles the processing of tasks from the queue. + * This is separated from task creation to avoid importing all task handlers + * when only creating tasks, which eliminates compilation overhead. + */ +export class TaskProcessor { + async processQueue(): Promise { + const tasks = await Task.getNextBatch(); + console.info(`Processing ${tasks.length} tasks`, tasks); + + const tasksPromises = tasks.map(async (task) => { + console.info( + `Processing task ${task.id}, attempt:${task.attempts} maxAttempts:${task.maxAttempts} lastFailedAttempt:${task.lastFailedAttemptAt}`, + task + ); + const taskHandlerGetter = tasksMap[task.type as keyof typeof tasksMap]; + if (!taskHandlerGetter) throw new Error(`Task handler not found for type ${task.type}`); + const taskConfig = tasksConfig[task.type as keyof typeof tasksConfig]; + const taskHandler = await taskHandlerGetter(); + return taskHandler(task.payload) + .then(async () => { + await Task.succeed(task.id); + }) + .catch(async (error) => { + console.info(`Retrying task ${task.id}: ${error}`); + await Task.retry({ + taskId: task.id, + lastError: error instanceof Error ? error.message : "Unknown error", + minRetryIntervalMins: + taskConfig && "minRetryIntervalMins" in taskConfig ? taskConfig.minRetryIntervalMins : null, + }); + }); + }); + const settled = await Promise.allSettled(tasksPromises); + const failed = settled.filter((result) => result.status === "rejected"); + const succeded = settled.filter((result) => result.status === "fulfilled"); + console.info({ failed, succeded }); + } +} diff --git a/packages/features/tasker/tasker.ts b/packages/features/tasker/tasker.ts index ce603d3ec2..8d145eeacb 100644 --- a/packages/features/tasker/tasker.ts +++ b/packages/features/tasker/tasker.ts @@ -42,7 +42,6 @@ export type TaskerCreate = ( export interface Tasker { /** Create a new task with the given type and payload. */ create: TaskerCreate; - processQueue(): Promise; cleanup(): Promise; cancel(id: string): Promise; cancelWithReference(referenceUid: string, type: TaskTypes): Promise;