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 <keithwillcode@gmail.com> * fix: resolve TypeScript errors in worker implementation Co-Authored-By: keith@cal.com <keithwillcode@gmail.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 <keithwillcode@gmail.com> Co-authored-by: Morgan <33722304+ThyMinimalDev@users.noreply.github.com> Co-authored-by: cal.com <morgan@cal.com>
2025-06-12 11:56:13 +00:00
parent 34beadeef6
commit 2aaf672b10
7 changed files with 257 additions and 14 deletions
@@ -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<boolean>("e2e")
+        ? await getAvailableSlots({
+            input: {
+              ...query,
+              isTeamEvent,
+            },
+            ctx: {
+              req,
+            },
+          })
+        : await this.slotsWorkerService.getAvailableSlotsInWorker({
+            input: {
+              ...query,
+              isTeamEvent,
+            },
+            ctx: {
+              req,
+            },
+          });

      const { slots } = await this.slotsOutputService.getOutputSlots(
        availableSlots,
@@ -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<string, string>;
+      headers?: Record<string, string | string[]>;
+    };
+  };
+}
+
+/**
+ * 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<boolean>("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<string, string>) || {},
+                  headers: (task.options.ctx.req.headers as Record<string, string>) || {},
+                }
+              : 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<TimeSlots> {
+    return new Promise<TimeSlots>((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.");
+  }
+}
@@ -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],
 })
@@ -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"),
+    });
+  }
+});