Chapter 20: Scaling Node.js: The Cluster Module, Worker Threads, and Multi-Core Architecture

┌─────────────────────────────────────────────────────────────────────────────┐
│                 THE CLUSTER MODULE (Multi-Process Scaling)                  │
├─────────────────────────────────────────────────────────────────────────────┤
│ Primary Process (PID: 1000)                                                 │
│   ├── Spawns Worker Process 1 (PID: 1001) ──► Isolated Heap (1.4GB) + V8   │
│   ├── Spawns Worker Process 2 (PID: 1002) ──► Isolated Heap (1.4GB) + V8   │
│   ├── Spawns Worker Process 3 (PID: 1003) ──► Isolated Heap (1.4GB) + V8   │
│   └── Spawns Worker Process 4 (PID: 1004) ──► Isolated Heap (1.4GB) + V8   │
│                                                                             │
│ 📌 Communication: Inter-Process Communication (IPC serialization via pipes)  │
│ 📌 Memory: Completely isolated memory space (No shared RAM)                 │
│ 📌 Best For: Scaling HTTP / Network I/O throughput across all CPU cores     │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│                 WORKER THREADS (Multi-Threaded CPU Offloading)              │
├─────────────────────────────────────────────────────────────────────────────┤
│ Single Node.js Process (PID: 2000)                                          │
│   ├── Main Thread (Event Loop) ──► Handles all HTTP requests, timers, I/O   │
│   ├── Worker Thread 1 (Isolate) ──► Heavy CPU calculation (e.g. Scrypt/AES) │
│   └── Worker Thread 2 (Isolate) ──► Heavy CPU calculation (e.g. Image Sharp)│
│                                                                             │
│ 📌 Communication: MessagePort (`postMessage`) or SharedArrayBuffer (Zero-Copy)│
│ 📌 Memory: Shares OS process memory; separate V8 heaps                     │
│ 📌 Best For: Offloading CPU-bound tasks without blocking the main Event Loop│
└─────────────────────────────────────────────────────────────────────────────┘

Incoming TCP Connection (SYN packet on Port 3000)
                        │
                        ▼
             Primary / Master Process
          (Binds to socket on Port 3000)
                        │
         Round-Robin Connection Scheduling (SCHED_RR)
        ┌───────────────┼───────────────┐
        ▼               ▼               ▼
  Worker Process 1  Worker Process 2  Worker Process 3
    (Core 0)          (Core 1)          (Core 2)

// ❌ CRITICAL CONTAINER BUG:
const NUM_CPUS = os.cpus().length;

// ✅ CONTAINER-AWARE BEST PRACTICE:
import os from 'node:os';

export const getConcurrencyLimit = (): number => {
  // Allow explicit override via environment variable (recommended in K8s)
  if (process.env.WEB_CONCURRENCY) {
    return parseInt(process.env.WEB_CONCURRENCY, 10);
  }
  
  if (typeof os.availableParallelism === 'function') {
    return os.availableParallelism();
  }

  return os.cpus().length;
};

// src/cluster.ts
import cluster from 'node:cluster';
import os from 'node:os';
import { logger } from './common/logger/logger';

// 1. Determine CPU concurrency safely
const NUM_WORKERS =
  Number(process.env.WEB_CONCURRENCY) ||
  (typeof os.availableParallelism === 'function'
    ? os.availableParallelism()
    : os.cpus().length);

if (cluster.isPrimary) {
  logger.info(
    { masterPid: process.pid, workersToSpawn: NUM_WORKERS },
    'Primary cluster process booting'
  );

  // Track global metrics aggregated across workers via IPC
  let totalRequestsServed = 0;

  // Fork a worker for each available compute unit
  for (let i = 0; i < NUM_WORKERS; i++) {
    cluster.fork();
  }

  // 2. Track online workers and attach IPC listeners
  cluster.on('online', (worker) => {
    logger.info({ workerPid: worker.process.pid, workerId: worker.id }, 'Worker online and ready');

    // IPC: Listen to messages from worker processes
    worker.on('message', (msg: { type: string; count?: number }) => {
      if (msg.type === 'REQUEST_COUNT') {
        totalRequestsServed += msg.count || 1;
      }
    });
  });

  // 3. Auto-Healing: Replace dead workers automatically with bounded backoff
  cluster.on('exit', (worker, code, signal) => {
    logger.warn(
      { workerPid: worker.process.pid, code, signal },
      'Worker process terminated. Spawning replacement...'
    );

    // Prevent fast respawn loops if crashes are instantaneous
    setTimeout(() => {
      cluster.fork();
    }, 1000);
  });

  // 4. Zero-Downtime Rolling Restarts: Listen for reload signal (e.g. from CI/CD)
  process.on('SIGUSR2', async () => {
    logger.info('Received SIGUSR2. Performing zero-downtime rolling restart...');

    const workers = Object.values(cluster.workers || {});
    for (const worker of workers) {
      if (!worker) continue;

      logger.info({ workerId: worker.id }, 'Rolling restart: Spawning replacement worker...');
      
      // Spawn new worker first
      const newWorker = cluster.fork();

      await new Promise<void>((resolve) => {
        newWorker.once('listening', () => {
          // Gracefully disconnect and terminate old worker only after new one is ready
          worker.disconnect();
          setTimeout(() => worker.kill(), 5000);
          resolve();
        });
      });
    }

    logger.info('Rolling restart completed successfully');
  });

} else {
  // -------------------------------------------------------------
  // WORKER PROCESS EXECUTION (Runs your Express Application)
  // -------------------------------------------------------------
  import('./server').then(({ startServer }) => {
    startServer();
  });
}

// packages/database/src/db/client.ts
const sql = postgres(process.env.DATABASE_URL, {
  max: 20, // 20 connections per pool
});

Server has 8 CPU Cores ──► Forks 8 Cluster Workers
Each worker allocates: max: 20 database connections
Total Open Connections to PostgreSQL = 8 × 20 = 160 connections!

FATAL: remaining connection slots are reserved for non-replicated superuser connections

Max Pool Size per Worker = (Total Allowed DB Connections - Reserve Connections) / (Number of Workers × Number of Server Instances)

Pool Size = (100 - 10) / 8 ≈ 10 connections per worker

// ❌ BLOCKS THE ENTIRE EVENT LOOP FOR 3 SECONDS:
app.post('/api/v1/compute-hash', (req, res) => {
  const result = heavyCpuCalculation(req.body.data); // Synchronous 3000ms loop!
  res.json({ result });
});

Main Thread (Event Loop)                           Worker Thread
       │                                                 │
       │  1. Receives HTTP Request                       │
       │  2. Dispatches task via `postMessage(data)`    │
       │────────────────────────────────────────────────►│
       │                                                 │ (Runs heavy CPU loop)
       │  3. Continues serving other HTTP requests!      │ (Event loop remains free!)
       │                                                 │
       │  4. Receives result via `parentPort.on('message')│
       │◄────────────────────────────────────────────────│
       │  5. Returns HTTP 200 OK to Client               │

pnpm add piscina

// src/workers/crypto.worker.ts
import crypto from 'node:crypto';

export interface HashTask {
  password: string;
  salt: string;
}

export interface HashResult {
  derivedKeyHex: string;
  durationMs: number;
}

// Piscina automatically calls this function for every queued task
export default function hashPassword(task: HashTask): HashResult {
  const start = performance.now();

  // Heavy CPU-bound computation: Scrypt with high cost parameters (N=65536)
  const key = crypto.scryptSync(task.password, task.salt, 64, {
    cost: 65536,
    blockSize: 8,
    parallelization: 1,
  });

  const durationMs = Math.round(performance.now() - start);

  return {
    derivedKeyHex: key.toString('hex'),
    durationMs,
  };
}

// src/common/services/cpu-pool.service.ts
import Piscina from 'piscina';
import path from 'node:path';
import os from 'node:os';
import { HashTask, HashResult } from '../../workers/crypto.worker';

const concurrency = typeof os.availableParallelism === 'function' 
  ? os.availableParallelism() 
  : os.cpus().length;

export class CpuPoolService {
  private static pool: Piscina;

  public static initialize(): void {
    this.pool = new Piscina({
      filename: path.resolve(__dirname, '../../workers/crypto.worker.js'),
      // Keep worker threads pre-warmed matching available cores
      minThreads: Math.max(1, Math.floor(concurrency / 2)),
      maxThreads: concurrency,
      // Terminate idle threads after 30 seconds of inactivity to save RAM
      idleTimeout: 30000,
      // Max queued tasks before rejecting with backpressure error
      maxQueue: 1000,
    });
  }

  /**
   * Dispatches task to the pre-warmed worker pool.
   * Returns a Promise that resolves when computation completes.
   */
  public static async executeHash(payload: HashTask): Promise<HashResult> {
    if (!this.pool) {
      throw new Error('CpuPoolService is not initialized.');
    }
    return this.pool.run(payload) as Promise<HashResult>;
  }
}

// src/features/crypto/crypto.controller.ts
import { Request, Response, NextFunction } from 'express';
import { CpuPoolService } from '../../common/services/cpu-pool.service';
import { ApiResponse } from '../../common/responses/api-response';

export class CryptoController {
  public hashPasswordAsync = async (
    req: Request,
    res: Response,
    next: NextFunction
  ): Promise<void> => {
    try {
      const { password, salt } = req.body;

      // Offloaded to worker thread pool: HTTP event loop stays at 0% latency!
      const result = await CpuPoolService.executeHash({ password, salt });

      ApiResponse.ok(res, 'Key derived successfully without event loop lag', result);
    } catch (error) {
      next(error);
    }
  };
}

// 1. Allocate 1024 bytes of shared memory accessible by all threads
const sharedBuffer = new SharedArrayBuffer(1024);

// 2. Wrap buffer in an Int32Array view (256 integers)
const sharedInt32 = new Int32Array(sharedBuffer);

// 3. Atomically add to index 0 (Thread-safe, avoids race conditions)
Atomics.add(sharedInt32, 0, 5);

// 4. Thread-safe read
const currentValue = Atomics.load(sharedInt32, 0);

// 5. Thread synchronization: Wait until index 0 equals expected value
// Atomics.wait(sharedInt32, 0, 0, 1000); // Block worker thread up to 1000ms

┌─────────────────────────────────────────────────────────────────────────────┐
│                          ORCHESTRATION COMPARISON                           │
├─────────────────────────────────────────────────────────────────────────────┤
│ 1. Native node:cluster:                                                     │
│    - Zero external dependencies. Complete code-level control.               │
│    - Best for: Custom infrastructure and learning internals.                │
│                                                                             │
│ 2. PM2 Cluster Mode (Process Manager):                                      │
│    - Wraps node:cluster automatically without changing application code.    │
│    - Built-in zero-downtime reload (`pm2 reload app`), log rotation, alerts.│
│    - Best for: Dedicated Virtual Machines (AWS EC2, DigitalOcean Droplets). │
│                                                                             │
│ 3. Kubernetes / Docker Replicas (Container-Native):                         │
│    - "1 Container = 1 Core" paradigm.                                       │
│    - Horizontal Pod Autoscaler (HPA) scales pods horizontally based on CPU. │
│    - Best for: Enterprise cloud-native Kubernetes environments.             │
└─────────────────────────────────────────────────────────────────────────────┘

// ecosystem.config.js
module.exports = {
  apps: [
    {
      name: 'api-server',
      script: './dist/server.js',
      instances: 'max',       // Automatically scales across all available cores
      exec_mode: 'cluster',   // Enables node:cluster mode
      max_memory_restart: '1G', // Restarts worker if memory leak exceeds 1GB
      env: {
        NODE_ENV: 'production',
        PORT: 3000,
      },
    },
  ],
};

┌────────────────────────────────────────────────────────────────────────────┐
│                    PRODUCTION MULTI-CORE SCALING CHECKLIST                 │
├────────────────────────────────────────────────────────────────────────────┤
│ [ ] Container-Aware Core Detection: Use `os.availableParallelism()` rather │
│     than `os.cpus().length` to prevent container CPU quota throttling.     │
│                                                                            │
│ [ ] Auto-Healing Implemented: `cluster.on('exit')` monitors worker health  │
│     and respawns dead workers with bounded backoff.                        │
│                                                                            │
│ [ ] Zero-Downtime Reloads: `SIGUSR2` rolling reload updates workers        │
│     sequentially without dropping active HTTP connections.                 │
│                                                                            │
│ [ ] Database Pool Scaled: `max` pool size divided by worker count to       │
│     prevent PostgreSQL connection exhaustion (`max_connections`).          │
│                                                                            │
│ [ ] State Extracted to Redis: Sessions, rate limiters, and caches backed   │
│     by Redis to prevent in-memory state partitioning across workers.        │
│                                                                            │
│ [ ] CPU-Heavy Tasks Delegated: Computationally intensive operations        │
│     offloaded to worker threads, keeping the main Event Loop free.         │
│                                                                            │
│ [ ] Pre-Warmed Thread Pool: Thread pool managed via `piscina` rather than  │
│     allocating expensive `new Worker()` instances per request.             │
│                                                                            │
│ [ ] Zero-Copy Memory for Large Buffers: `SharedArrayBuffer` + `Atomics`   │
│     utilized for massive cross-thread datasets without clone overhead.     │
└────────────────────────────────────────────────────────────────────────────┘