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 <keithwillcode@gmail.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 <keithwillcode@gmail.com> * Moved the cleanup method back to internal tasker --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
co-authored by
Devin AI
parent
9724bc07a6
commit
1739114490
@@ -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 });
|
||||
}
|
||||
```
|
||||
|
||||
@@ -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 });
|
||||
}
|
||||
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
const count = await Task.cleanup();
|
||||
console.info(`Cleaned up ${count} tasks`);
|
||||
}
|
||||
|
||||
async cancel(id: string): Promise<string> {
|
||||
const task = await Task.cancel(id);
|
||||
return task.id;
|
||||
|
||||
@@ -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<void> {
|
||||
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 });
|
||||
}
|
||||
}
|
||||
@@ -42,7 +42,6 @@ export type TaskerCreate = <TaskKey extends keyof TaskPayloads>(
|
||||
export interface Tasker {
|
||||
/** Create a new task with the given type and payload. */
|
||||
create: TaskerCreate;
|
||||
processQueue(): Promise<void>;
|
||||
cleanup(): Promise<void>;
|
||||
cancel(id: string): Promise<string>;
|
||||
cancelWithReference(referenceUid: string, type: TaskTypes): Promise<string | null>;
|
||||
|
||||
Reference in New Issue
Block a user