From 2aaf672b1015730acddb40bc8cb86cdb7e0c4f1d Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 12 Jun 2025 11:56:13 +0000 Subject: [PATCH] perf: Implement worker threads for getAvailableSlots to prevent CPU blocking (#21479) * Implement worker threads for getAvailableSlots to prevent CPU blocking Co-Authored-By: keith@cal.com * fix: resolve TypeScript errors in worker implementation Co-Authored-By: keith@cal.com * chore: fix slot-worker-service.ts * fix: do not use workers in E2E * fix: try to fix memory leak * chore: remove unecessary comments --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: keith@cal.com Co-authored-by: Morgan <33722304+ThyMinimalDev@users.noreply.github.com> Co-authored-by: cal.com --- apps/api/v2/src/config/app.ts | 2 +- apps/api/v2/src/env.ts | 2 +- .../controllers/slots.controller.ts | 36 ++- .../services/slots-worker.service.ts | 205 ++++++++++++++++++ .../slots/slots-2024-04-15/slots.module.ts | 8 +- .../slots-2024-04-15/workers/slots.worker.ts | 16 ++ apps/api/v2/test/setEnvVars.ts | 2 +- 7 files changed, 257 insertions(+), 14 deletions(-) create mode 100644 apps/api/v2/src/modules/slots/slots-2024-04-15/services/slots-worker.service.ts create mode 100644 apps/api/v2/src/modules/slots/slots-2024-04-15/workers/slots.worker.ts diff --git a/apps/api/v2/src/config/app.ts b/apps/api/v2/src/config/app.ts index 79861bbdf3..bbb2813cfb 100644 --- a/apps/api/v2/src/config/app.ts +++ b/apps/api/v2/src/config/app.ts @@ -36,7 +36,7 @@ const loadConfig = (): AppConfig => { app: { baseUrl: getEnv("WEB_APP_URL", "https://app.cal.com"), }, - e2e: getEnv("IS_E2E", false), + e2e: getEnv("IS_E2E", "false") === "true" ? true : false, }; }; diff --git a/apps/api/v2/src/env.ts b/apps/api/v2/src/env.ts index a2d289ab94..b3f359b853 100644 --- a/apps/api/v2/src/env.ts +++ b/apps/api/v2/src/env.ts @@ -17,7 +17,7 @@ export type Environment = { STRIPE_API_KEY: string; STRIPE_WEBHOOK_SECRET: string; WEB_APP_URL: string; - IS_E2E: boolean; + IS_E2E: string; CALCOM_LICENSE_KEY: string; GET_LICENSE_KEY_URL: string; API_KEY_PREFIX: string; diff --git a/apps/api/v2/src/modules/slots/slots-2024-04-15/controllers/slots.controller.ts b/apps/api/v2/src/modules/slots/slots-2024-04-15/controllers/slots.controller.ts index 18b5b0064c..cb4dc7b808 100644 --- a/apps/api/v2/src/modules/slots/slots-2024-04-15/controllers/slots.controller.ts +++ b/apps/api/v2/src/modules/slots/slots-2024-04-15/controllers/slots.controller.ts @@ -1,7 +1,9 @@ import { SlotsOutputService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots-output.service"; import type { RangeSlots, TimeSlots } from "@/modules/slots/slots-2024-04-15/services/slots-output.service"; +import { SlotsWorkerService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots-worker.service"; import { SlotsService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots.service"; import { Query, Body, Controller, Get, Delete, Post, Req, Res, BadRequestException } from "@nestjs/common"; +import { ConfigService } from "@nestjs/config"; import { ApiExcludeController as DocsExcludeController } from "@nestjs/swagger"; import { ApiTags as DocsTags, ApiCreatedResponse, ApiOkResponse, ApiOperation } from "@nestjs/swagger"; import { Response as ExpressResponse, Request as ExpressRequest } from "express"; @@ -25,7 +27,9 @@ import { ApiResponse, GetAvailableSlotsInput_2024_04_15 } from "@calcom/platform export class SlotsController_2024_04_15 { constructor( private readonly slotsService: SlotsService_2024_04_15, - private readonly slotsOutputService: SlotsOutputService_2024_04_15 + private readonly config: ConfigService, + private readonly slotsOutputService: SlotsOutputService_2024_04_15, + private readonly slotsWorkerService: SlotsWorkerService_2024_04_15 ) {} @Post("/reserve") @@ -162,15 +166,27 @@ export class SlotsController_2024_04_15 { query.isTeamEvent === undefined ? await this.slotsService.checkIfIsTeamEvent(query.eventTypeId) : query.isTeamEvent; - const availableSlots = await getAvailableSlots({ - input: { - ...query, - isTeamEvent, - }, - ctx: { - req, - }, - }); + + // Do not use workers in E2E, not supported by TS-JEST + const availableSlots = this.config.get("e2e") + ? await getAvailableSlots({ + input: { + ...query, + isTeamEvent, + }, + ctx: { + req, + }, + }) + : await this.slotsWorkerService.getAvailableSlotsInWorker({ + input: { + ...query, + isTeamEvent, + }, + ctx: { + req, + }, + }); const { slots } = await this.slotsOutputService.getOutputSlots( availableSlots, diff --git a/apps/api/v2/src/modules/slots/slots-2024-04-15/services/slots-worker.service.ts b/apps/api/v2/src/modules/slots/slots-2024-04-15/services/slots-worker.service.ts new file mode 100644 index 0000000000..488bdc0e73 --- /dev/null +++ b/apps/api/v2/src/modules/slots/slots-2024-04-15/services/slots-worker.service.ts @@ -0,0 +1,205 @@ +import { Injectable, Logger, OnModuleDestroy } from "@nestjs/common"; +import { ConfigService } from "@nestjs/config"; +import * as path from "path"; +import { Worker } from "worker_threads"; + +// Import WorkerOptions type +import type { GetScheduleOptions } from "@calcom/trpc/server/routers/viewer/slots/types"; + +import { TimeSlots } from "./slots-output.service"; + +/** + * Interface to define the structure of messages sent to the worker. + */ +interface WorkerMessage { + input: GetScheduleOptions["input"]; + ctx?: { + req?: { + cookies?: Record; + headers?: Record; + }; + }; +} + +/** + * Interface to define the structure of results received from the worker. + */ +interface WorkerResult { + success: boolean; + data?: TimeSlots; + error?: Error; +} + +@Injectable() +export class SlotsWorkerService_2024_04_15 implements OnModuleDestroy { + private readonly logger = new Logger(SlotsWorkerService_2024_04_15.name); + private readonly workerPool: Worker[] = []; + private readonly maxWorkers: number; + private readonly taskQueue: Array<{ + resolve: (value: TimeSlots) => void; + reject: (reason: Error) => void; + options: GetScheduleOptions; + }> = []; + private availableWorkers: Worker[] = []; + + constructor(private readonly config: ConfigService) { + this.maxWorkers = process.env.SLOTS_WORKER_POOL_SIZE + ? parseInt(process.env.SLOTS_WORKER_POOL_SIZE, 10) + : 4; + // Workers are not initialized in E@E + if (!this.config.get("e2e")) { + this.initializeWorkerPool(); + } + } + + /** + * Initializes the worker pool by creating a fixed number of worker threads. + * Each worker is set up with persistent event listeners for errors and exits. + */ + private initializeWorkerPool(): void { + for (let i = 0; i < this.maxWorkers; i++) { + this.createNewWorker(); + } + } + + /** + * Creates a new worker thread and configures its essential, persistent event listeners. + * Adds the new worker to the pool and available workers list. + */ + private createNewWorker(): void { + const worker = new Worker(path.join(__dirname, "../workers/slots.worker.js")); + + // These 'on' listeners are for the worker's overall lifecycle (crashes, exits). + // They are persistent and responsible for calling handleWorkerFailure. + worker.on("error", (err: Error) => { + this.logger.error(`Worker experienced a persistent error: ${err.message}`, err.stack); + this.handleWorkerFailure(worker); + }); + + worker.on("exit", (code: number) => { + if (code !== 0) { + this.logger.error(`Worker exited with code ${code}.`); + } + this.handleWorkerFailure(worker); + }); + + this.workerPool.push(worker); + this.availableWorkers.push(worker); + } + + /** + * Handles the failure of a worker by removing it from pools and creating a new one. + * This ensures the worker pool remains at the desired size and healthy. + * @param failedWorker The worker that failed or exited. + */ + private handleWorkerFailure(failedWorker: Worker): void { + // Remove the failed worker from both pools + this.workerPool.splice(this.workerPool.indexOf(failedWorker), 1); + this.availableWorkers = this.availableWorkers.filter((w) => w !== failedWorker); + + // Attempt to create a new worker to replace the failed one + try { + this.createNewWorker(); + } catch (error) { + this.logger.error( + `Failed to create replacement worker after failure: ${ + error instanceof Error ? error.message : String(error) + }`, + error instanceof Error ? error.stack : undefined + ); + } + + // After a worker fails, process the next task in case there are queued tasks + this.processNextTask(); + } + + /** + * Processes the next task in the queue if there are available workers. + * Assigns a task to an available worker and sets up 'once' listeners for its specific result. + */ + private processNextTask(): void { + if (this.taskQueue.length > 0 && this.availableWorkers.length > 0) { + const task = this.taskQueue.shift(); + const worker = this.availableWorkers.shift(); + + if (!task || !worker) { + // This should theoretically not happen if the checks above pass, but good for type narrowing. + return; + } + + // Prepare context for serialization + const serializableCtx: WorkerMessage["ctx"] = task.options.ctx + ? { + req: task.options.ctx.req + ? { + cookies: (task.options.ctx.req.cookies as Record) || {}, + headers: (task.options.ctx.req.headers as Record) || {}, + } + : undefined, + } + : undefined; + + try { + // Use 'once' listeners for task-specific responses and errors. + // 'once' listeners automatically remove themselves after being invoked, preventing leaks. + const messageListener = (result: WorkerResult) => { + this.availableWorkers.push(worker); // Return worker to the available pool + if (result.success) { + task.resolve(result.data as TimeSlots); + } else { + task.reject(result.error ?? new Error("An error occurred in the worker thread.")); + } + this.processNextTask(); // Attempt to process the next task + }; + + const errorListener = (err: Error) => { + this.availableWorkers.push(worker); // Ensure worker is returned + task.reject(new Error(`Worker thread error during task execution: ${err.message}`)); + this.processNextTask(); // Attempt to process the next task + }; + + worker.once("message", messageListener); // Use 'once' for task results + worker.once("error", errorListener); // Use 'once' for task-specific errors + + worker.postMessage({ + input: task.options.input, + ctx: serializableCtx, + } as WorkerMessage); + } catch (error) { + // If posting the message itself fails (e.g., serialization error) + this.availableWorkers.push(worker); // Ensure worker is returned to pool + task.reject( + new Error( + `Failed to dispatch task to worker: ${error instanceof Error ? error.message : String(error)}` + ) + ); + this.processNextTask(); // Try to process next task if available + } + } + } + + /** + * Public method to request available time slots, offloading the computation to a worker thread. + * Returns a Promise that resolves with the TimeSlots or rejects with an Error. + * @param options The GetScheduleOptions to pass to the worker. + * @returns A Promise resolving to TimeSlots. + */ + public async getAvailableSlotsInWorker(options: GetScheduleOptions): Promise { + return new Promise((resolve, reject) => { + this.taskQueue.push({ + resolve, + reject, + options, + }); + this.processNextTask(); // Attempt to process immediately + }); + } + + onModuleDestroy(): void { + this.logger.log("Terminating worker pool..."); + for (const worker of this.workerPool) { + worker.terminate(); + } + this.logger.log("Worker pool terminated."); + } +} diff --git a/apps/api/v2/src/modules/slots/slots-2024-04-15/slots.module.ts b/apps/api/v2/src/modules/slots/slots-2024-04-15/slots.module.ts index d2867babbf..06ccc0988a 100644 --- a/apps/api/v2/src/modules/slots/slots-2024-04-15/slots.module.ts +++ b/apps/api/v2/src/modules/slots/slots-2024-04-15/slots.module.ts @@ -2,13 +2,19 @@ import { EventTypesModule_2024_04_15 } from "@/ee/event-types/event-types_2024_0 import { PrismaModule } from "@/modules/prisma/prisma.module"; import { SlotsController_2024_04_15 } from "@/modules/slots/slots-2024-04-15/controllers/slots.controller"; import { SlotsOutputService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots-output.service"; +import { SlotsWorkerService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots-worker.service"; import { SlotsService_2024_04_15 } from "@/modules/slots/slots-2024-04-15/services/slots.service"; import { SlotsRepository_2024_04_15 } from "@/modules/slots/slots-2024-04-15/slots.repository"; import { Module } from "@nestjs/common"; @Module({ imports: [PrismaModule, EventTypesModule_2024_04_15], - providers: [SlotsRepository_2024_04_15, SlotsService_2024_04_15, SlotsOutputService_2024_04_15], + providers: [ + SlotsRepository_2024_04_15, + SlotsService_2024_04_15, + SlotsOutputService_2024_04_15, + SlotsWorkerService_2024_04_15, + ], controllers: [SlotsController_2024_04_15], exports: [SlotsService_2024_04_15], }) diff --git a/apps/api/v2/src/modules/slots/slots-2024-04-15/workers/slots.worker.ts b/apps/api/v2/src/modules/slots/slots-2024-04-15/workers/slots.worker.ts new file mode 100644 index 0000000000..ef1f746661 --- /dev/null +++ b/apps/api/v2/src/modules/slots/slots-2024-04-15/workers/slots.worker.ts @@ -0,0 +1,16 @@ +import { parentPort } from "worker_threads"; + +import { getAvailableSlots } from "@calcom/platform-libraries/slots"; + +parentPort?.on("message", async (data) => { + try { + const { input, ctx } = data; + const result = await getAvailableSlots({ input, ctx }); + parentPort?.postMessage({ success: true, data: result }); + } catch (error) { + parentPort?.postMessage({ + success: false, + error: error instanceof Error ? error : new Error("Unknown error"), + }); + } +}); diff --git a/apps/api/v2/test/setEnvVars.ts b/apps/api/v2/test/setEnvVars.ts index 64aa6e3853..2f2a53fdc4 100644 --- a/apps/api/v2/test/setEnvVars.ts +++ b/apps/api/v2/test/setEnvVars.ts @@ -13,7 +13,7 @@ const env: Partial> = { REDIS_URL: "redis://localhost:6379", STRIPE_API_KEY: "sk_test_51J4", STRIPE_WEBHOOK_SECRET: "whsec_51J4", - IS_E2E: true, + IS_E2E: "true", API_KEY_PREFIX: "cal_test_", GET_LICENSE_KEY_URL: " https://console.cal.com/api/license", CALCOM_LICENSE_KEY: "c4234812-12ab-42s6-a1e3-55bedd4a5bb7",