Node.js Clustering Explained: Scale to Every CPU Core

// cluster-internals-demo.js
// Run with: node cluster-internals-demo.js
// Shows exactly which PID handles each request

const cluster = require('node:cluster');
const http = require('node:http');
const os = require('node:os');

const TOTAL_CPU_CORES = os.cpus().length;
const PORT = 3000;

if (cluster.isPrimary) {
  console.log(`Primary process PID: ${process.pid}`);
  console.log(`Detected ${TOTAL_CPU_CORES} CPU cores — forking ${TOTAL_CPU_CORES} workers...\n`);

  // Fork one worker per logical CPU core
  for (let coreIndex = 0; coreIndex < TOTAL_CPU_CORES; coreIndex++) {
    const worker = cluster.fork();

    // Listen for messages sent from workers back to the primary
    worker.on('message', (msg) => {
      if (msg.type === 'REQUEST_HANDLED') {
        console.log(`[Primary] Worker ${worker.id} (PID ${worker.process.pid}) handled request #${msg.requestCount}`);
      }
    });
  }

  // The primary is notified when a worker dies unexpectedly
  cluster.on('exit', (deadWorker, exitCode, signal) => {
    console.warn(`\n[Primary] Worker ${deadWorker.id} died (PID: ${deadWorker.process.pid}, code: ${exitCode}, signal: ${signal})`);
    console.log('[Primary] Forking a replacement worker immediately...');
    cluster.fork(); // Zero-downtime restart — the port never goes dark
  });

} else {
  // ─── WORKER PROCESS ───────────────────────────────────────────────
  // Every worker runs this block independently in its own V8 instance

  let requestsHandledByThisWorker = 0;

  const server = http.createServer((req, res) => {
    requestsHandledByThisWorker++;

    // Simulate a lightweight CPU task (real apps might do JWT verify, template render, etc.)
    const payload = JSON.stringify({
      workerPid: process.pid,
      workerId: cluster.worker.id,
      requestsHandled: requestsHandledByThisWorker,
      message: 'Handled by worker'
    });

    // Tell the primary which worker handled this (demonstrates IPC messaging)
    process.send({ type: 'REQUEST_HANDLED', requestCount: requestsHandledByThisWorker });

    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(payload);
  });

  // Workers all listen on the same port — the primary owns the actual socket
  server.listen(PORT, () => {
    console.log(`Worker ${cluster.worker.id} (PID: ${process.pid}) is ready on port ${PORT}`);
  });
}

// production-cluster.js
// A battle-hardened cluster manager with:
//   - Automatic respawn with crash-loop protection
//   - Graceful shutdown on SIGTERM / SIGUSR2
//   - Per-worker health tracking
// Run with: node production-cluster.js

const cluster = require('node:cluster');
const http = require('node:http');
const os = require('node:os');

const WORKER_COUNT = os.cpus().length;
const PORT = 3000;
const CRASH_LOOP_THRESHOLD_MS = 1000; // If a worker restarts faster than this, we back off
const MAX_RESTART_ATTEMPTS = 5;       // Give up after this many rapid restarts

// ─── PRIMARY PROCESS ──────────────────────────────────────────────────────────
if (cluster.isPrimary) {
  console.log(`[Primary] PID ${process.pid} starting ${WORKER_COUNT} workers`);

  // Track restart metadata per worker ID to detect crash loops
  const workerRestartLog = new Map(); // workerId -> { count, lastRestartAt }

  function spawnWorker() {
    const worker = cluster.fork();
    const workerId = worker.id;

    // Initialise restart tracking for this worker slot
    if (!workerRestartLog.has(workerId)) {
      workerRestartLog.set(workerId, { count: 0, lastRestartAt: Date.now() });
    }

    worker.on('message', (msg) => {
      if (msg.type === 'READY') {
        console.log(`[Primary] Worker ${workerId} (PID ${worker.process.pid}) is healthy`);
        // Reset crash counter once we know the worker started successfully
        workerRestartLog.set(workerId, { count: 0, lastRestartAt: Date.now() });
      }
    });

    return worker;
  }

  // Fork all initial workers
  for (let i = 0; i < WORKER_COUNT; i++) spawnWorker();

  cluster.on('exit', (deadWorker, exitCode, signal) => {
    const workerId = deadWorker.id;
    const restartData = workerRestartLog.get(workerId) || { count: 0, lastRestartAt: 0 };
    const timeSinceLastRestart = Date.now() - restartData.lastRestartAt;
    const isRapidCrash = timeSinceLastRestart < CRASH_LOOP_THRESHOLD_MS;

    console.warn(`[Primary] Worker ${workerId} exited — code: ${exitCode}, signal: ${signal}`);

    if (isRapidCrash) {
      restartData.count++;
      restartData.lastRestartAt = Date.now();
      workerRestartLog.set(workerId, restartData);

      if (restartData.count >= MAX_RESTART_ATTEMPTS) {
        // Crash loop detected — stop respawning to protect the system
        console.error(`[Primary] Worker ${workerId} is in a crash loop after ${restartData.count} attempts. Not restarting. Investigate immediately.`);
        return;
      }

      // Exponential backoff before respawning
      const backoffDelay = Math.pow(2, restartData.count) * 100; // 200ms, 400ms, 800ms...
      console.warn(`[Primary] Rapid crash detected (#${restartData.count}). Backing off ${backoffDelay}ms before respawn.`);
      setTimeout(spawnWorker, backoffDelay);
    } else {
      // Normal exit or clean restart — respawn immediately
      restartData.count = 0;
      workerRestartLog.set(workerId, restartData);
      spawnWorker();
    }
  });

  // ── Graceful shutdown: send SIGTERM to each worker in sequence ──────────────
  // Trigger with: kill -SIGTERM <primaryPID>  or  npm stop
  process.on('SIGTERM', () => {
    console.log('[Primary] SIGTERM received — initiating graceful shutdown');

    const allWorkers = Object.values(cluster.workers);
    let workersRemaining = allWorkers.length;

    allWorkers.forEach((worker) => {
      // Ask the worker to finish in-flight requests then exit cleanly
      worker.send({ type: 'GRACEFUL_SHUTDOWN' });

      worker.on('exit', () => {
        workersRemaining--;
        if (workersRemaining === 0) {
          console.log('[Primary] All workers shut down cleanly. Primary exiting.');
          process.exit(0);
        }
      });
    });

    // Force-kill any worker that hasn't exited within 10 seconds
    setTimeout(() => {
      console.error('[Primary] Timeout — force-killing remaining workers');
      allWorkers.forEach((w) => w.kill('SIGKILL'));
      process.exit(1);
    }, 10_000);
  });

} else {
  // ─── WORKER PROCESS ─────────────────────────────────────────────────────────

  let activeConnections = 0;
  let isShuttingDown = false;

  const server = http.createServer((req, res) => {
    if (isShuttingDown) {
      // Reject new requests during graceful drain — return 503 Service Unavailable
      res.writeHead(503, { 'Connection': 'close' });
      res.end('Server is restarting, please retry');
      return;
    }

    activeConnections++;

    res.on('finish', () => {
      activeConnections--;
      // If we're draining and this was the last connection, exit now
      if (isShuttingDown && activeConnections === 0) shutdownNow();
    });

    // Simulate an actual request handler
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({ workerId: cluster.worker.id, pid: process.pid }));
  });

  function shutdownNow() {
    server.close(() => {
      console.log(`[Worker ${cluster.worker.id}] Closed cleanly — exiting`);
      process.exit(0);
    });
  }

  // Listen for graceful shutdown instruction from primary
  process.on('message', (msg) => {
    if (msg.type === 'GRACEFUL_SHUTDOWN') {
      console.log(`[Worker ${cluster.worker.id}] Graceful shutdown started — draining ${activeConnections} connections`);
      isShuttingDown = true;
      if (activeConnections === 0) shutdownNow(); // Already idle, exit immediately
    }
  });

  server.listen(PORT, () => {
    // Signal to primary that we're up and healthy
    process.send({ type: 'READY' });
  });
}

// cluster-redis-session.js
// Demonstrates the RIGHT way to handle sessions across clustered workers
// using Redis as the shared state store.
//
// Prerequisites: npm install ioredis express express-session connect-redis
// Assumes Redis is running on localhost:6379

const cluster = require('node:cluster');
const os = require('node:os');

if (cluster.isPrimary) {
  const workerCount = os.cpus().length;
  console.log(`[Primary] Forking ${workerCount} workers with Redis-backed sessions`);

  for (let i = 0; i < workerCount; i++) cluster.fork();

  cluster.on('exit', (worker) => {
    console.warn(`[Primary] Worker ${worker.id} died — respawning`);
    cluster.fork();
  });

} else {
  // ─── WORKER: Sets up Express with Redis session store ─────────────────────
  const express = require('express');
  const session = require('express-session');
  const { createClient } = require('redis'); // redis@4
  const { RedisStore } = require('connect-redis');

  const app = express();

  // Each worker creates its own Redis client connection
  // (connection pooling is handled internally by ioredis/redis)
  const redisClient = createClient({ url: 'redis://localhost:6379' });

  redisClient.on('error', (err) => {
    console.error(`[Worker ${cluster.worker.id}] Redis connection error:`, err.message);
  });

  // Connect to Redis before starting the server
  redisClient.connect().then(() => {
    console.log(`[Worker ${cluster.worker.id}] Connected to Redis`);

    // Sessions are stored IN Redis — not in worker memory
    // So any worker can read any user's session correctly
    app.use(session({
      store: new RedisStore({ client: redisClient }),
      secret: process.env.SESSION_SECRET || 'replace-with-env-var-in-production',
      resave: false,
      saveUninitialized: false,
      cookie: {
        secure: process.env.NODE_ENV === 'production', // HTTPS only in prod
        httpOnly: true,  // Prevent XSS from reading the cookie
        maxAge: 1000 * 60 * 60 // 1 hour session lifetime
      }
    }));

    app.use(express.json());

    // Route: login — stores user info in Redis-backed session
    app.post('/login', (req, res) => {
      const { username } = req.body;
      if (!username) return res.status(400).json({ error: 'username required' });

      // req.session is now written to Redis automatically
      req.session.authenticatedUser = { username, loginTime: new Date().toISOString() };
      req.session.handledByWorkerId = cluster.worker.id; // For demonstration

      res.json({
        message: `Logged in as ${username}`,
        sessionId: req.session.id,
        workerThatCreatedSession: cluster.worker.id
      });
    });

    // Route: profile — reads session from Redis (any worker can serve this correctly)
    app.get('/profile', (req, res) => {
      if (!req.session.authenticatedUser) {
        return res.status(401).json({ error: 'Not authenticated' });
      }

      res.json({
        user: req.session.authenticatedUser,
        sessionCreatedByWorker: req.session.handledByWorkerId,
        sessionServedByWorker: cluster.worker.id, // Will often be different — that's fine!
        message: 'Session correctly shared across all workers via Redis'
      });
    });

    // Route: demonstrates broken approach (DON'T do this in a cluster)
    const BROKEN_IN_MEMORY_COUNTER = { visits: 0 }; // Each worker has its own copy!

    app.get('/bad-counter', (req, res) => {
      BROKEN_IN_MEMORY_COUNTER.visits++;
      res.json({
        warning: 'This counter is per-worker, not global!',
        workerVisitCount: BROKEN_IN_MEMORY_COUNTER.visits,
        workerId: cluster.worker.id,
        fix: 'Use Redis INCR for a globally accurate counter'
      });
    });

    // Route: correct cross-worker counter using Redis INCR
    app.get('/good-counter', async (req, res) => {
      // INCR is atomic in Redis — safe across all workers simultaneously
      const globalVisitCount = await redisClient.incr('global:visit_counter');
      res.json({
        globalVisitCount,
        servedByWorkerId: cluster.worker.id,
        message: 'This count is accurate across all workers'
      });
    });

    app.listen(3000, () => {
      console.log(`[Worker ${cluster.worker.id}] HTTP server ready on port 3000`);
    });
  });
}

// cluster-with-thread-pool.js
// Production pattern: Cluster for concurrency + Worker Threads for CPU work
//
// Each cluster worker manages its own thread pool for CPU-intensive tasks.
// This prevents a heavy computation from ever blocking the HTTP event loop.
//
// Run: node cluster-with-thread-pool.js
// npm install is not needed — uses only Node.js built-ins

const cluster = require('node:cluster');
const http = require('node:http');
const os = require('node:os');
const { Worker, isMainThread, parentPort, workerData } = require('node:worker_threads');
const { fileURLToPath } = require('node:url');
const path = require('node:path');

// ─── THREAD WORKER CODE (runs when this file is loaded as a worker_thread) ───
// When Node loads this as a worker thread (not the main cluster worker),
// isMainThread is false, so only the CPU task runs — no HTTP server.
if (!isMainThread) {
  // This block executes ONLY inside a worker thread, not in any cluster process
  const { taskType, inputValue } = workerData;

  if (taskType === 'FIBONACCI') {
    // Deliberately naive recursive fib — CPU-bound, great for demonstration
    function fibonacci(n) {
      if (n <= 1) return n;
      return fibonacci(n - 1) + fibonacci(n - 2);
    }

    const result = fibonacci(inputValue);
    // Send result back to the cluster worker that spawned this thread
    parentPort.postMessage({ success: true, result, computedFor: inputValue });
  }

  // Process exits naturally when postMessage is done
  return; // Stop the rest of the file from running in thread context
}

// ─── CLUSTER PRIMARY ──────────────────────────────────────────────────────────
if (cluster.isPrimary) {
  const cpuCount = os.cpus().length;
  console.log(`[Primary] Forking ${cpuCount} cluster workers, each with their own thread pool`);

  for (let i = 0; i < cpuCount; i++) cluster.fork();

  cluster.on('exit', (worker) => {
    console.warn(`[Primary] Worker ${worker.id} exited — respawning`);
    cluster.fork();
  });

} else {
  // ─── CLUSTER WORKER: HTTP server + thread pool ──────────────────────────────

  const THREAD_POOL_SIZE = 2; // Each cluster worker maintains 2 threads for CPU tasks
  const pendingThreadTasks = new Map(); // taskId -> { resolve, reject }
  let nextTaskId = 0;

  // Simple thread pool implementation
  // Each cluster worker keeps THREAD_POOL_SIZE threads warm and ready
  const threadPool = Array.from({ length: THREAD_POOL_SIZE }, () => {
    const thread = new Worker(__filename); // Load THIS same file as a thread

    // When a thread sends back a result, resolve the matching promise
    thread.on('message', (result) => {
      const taskId = result.taskId;
      const pending = pendingThreadTasks.get(taskId);
      if (pending) {
        pendingThreadTasks.delete(taskId);
        pending.resolve(result);
      }
    });

    thread.on('error', (err) => console.error(`[Worker ${cluster.worker.id}] Thread error:`, err));
    return thread;
  });

  let threadRoundRobinIndex = 0;

  // Dispatch a CPU task to the next available thread in the pool
  function runInThread(taskType, inputValue) {
    return new Promise((resolve, reject) => {
      const taskId = nextTaskId++;
      pendingThreadTasks.set(taskId, { resolve, reject });

      // Pick next thread in round-robin fashion
      const targetThread = threadPool[threadRoundRobinIndex % THREAD_POOL_SIZE];
      threadRoundRobinIndex++;

      // postMessage is how cluster workers send work to their threads
      targetThread.postMessage({ taskType, inputValue, taskId });
    });
  }

  // Fix: threads need to echo taskId back so we can resolve the right promise
  // (Modify the thread block above to include taskId in the postMessage response)

  const server = http.createServer(async (req, res) => {
    const url = new URL(req.url, `http://localhost`);

    if (url.pathname === '/fibonacci') {
      const n = parseInt(url.searchParams.get('n') || '35', 10);

      if (n > 45) {
        res.writeHead(400);
        res.end(JSON.stringify({ error: 'n must be <= 45 for this demo' }));
        return;
      }

      const startTime = Date.now();

      // This runs in a THREAD — the event loop stays free for other requests!
      const threadResult = await runInThread('FIBONACCI', n);

      res.writeHead(200, { 'Content-Type': 'application/json' });
      res.end(JSON.stringify({
        fibonacci: threadResult.result,
        n,
        computedInMs: Date.now() - startTime,
        computedByWorker: cluster.worker.id,
        computedByThread: 'thread pool — event loop was never blocked'
      }));
    } else {
      // Non-CPU route responds instantly — proving event loop stayed unblocked
      res.writeHead(200);
      res.end(JSON.stringify({ status: 'healthy', workerId: cluster.worker.id }));
    }
  });

  server.listen(3000, () => {
    console.log(`[Worker ${cluster.worker.id}] Ready. Thread pool size: ${THREAD_POOL_SIZE}`);
  });
}