isotropic-cluster

npm version License

A reusable and extendable platform to manage local Node.js process clusters with a clean API for primary/worker communication and lifecycle management.

Why Use This?

• Simple API for Cluster Management: Manage worker processes with an intuitive, event-driven API
• Reliable Worker Lifecycle: Automatic worker initialization and ready state management
• Robust Communication: Structured, promise-based message passing between primary and workers
• Load Balancing: Built-in round-robin worker selection with support for task-specific distribution
• Error Handling: Comprehensive error propagation and worker replacement on failure
• Event Lifecycle: Observable events for all worker lifecycle stages
• Type-based Routing: Route messages to specific handler methods based on message type

Installation

npm install isotropic-cluster

Basic Usage

Primary Process

import _ClusterPrimary from 'isotropic-cluster/lib/cluster-primary.js';
import _later from 'isotropic-later';

{
    // Create a cluster primary
    const primary = _ClusterPrimary();

    // Listen for messages from workers
    primary.on('workerMessage', ({
        data: {
            message,
            worker
        }
    }) => {
        console.log(`Received from worker ${worker.id}:`, message);
    });

    // Listen for worker lifecycle events
    primary.on('workerReady', ({
        data: {
            worker
        }
    }) => {
        console.log(`Worker ${worker.id} is ready to handle requests`);

        // Send a message to the worker
        primary.send({
            message: {
                name: 'User',
                type: 'greet'
            },
            to: worker
        });
    });

    // Start some workers
    primary.fork({
        workerCount: 4
    });

    // Later, shutdown the cluster
    _later(60000, () => {
        primary.shutDown();
    });
}

Worker Process

import _ClusterWorker from 'isotropic-cluster/lib/cluster-worker.js';

{
    // Create a cluster worker
    const worker = _ClusterWorker();

    // Handle messages from primary
    worker.on('primaryMessage', ({
        data: {
            message
        }
    }) => {
        if (message.type === 'greet') {
            console.log(`Hello, ${message.name}!`);

            // Send a response back to primary
            worker.send({
                message: {
                    content: `Greeted ${message.name}`,
                    type: 'response'
                }
            });
        }
    });

    // Handle primary disconnection
    worker.on('primaryDisconnect', () => {
        console.log('Primary disconnected, cleaning up...');
        // Perform cleanup operations
    });
}

Core Concepts

Cluster Architecture

Node.js clusters consist of a primary process that manages multiple worker processes. The primary process:

• Forks and monitors worker processes
• Distributes work among workers
• Handles worker failures and restarts

Workers are separate processes that:

• Handle specific tasks (like serving HTTP requests)
• Report their status to the primary
• Perform work assigned by the primary

Initialization Lifecycle

1. Creation: Primary and worker objects are created
2. Initialization: Internal setup occurs
3. Worker Readiness: Workers signal readiness to the primary
4. Operation: Normal operation with message passing
5. Shutdown: Clean termination of worker processes

Event-Based Communication

Both ClusterPrimary and ClusterWorker extend isotropic-pubsub, providing a complete event system:

• Subscribe to events with on(), before(), and after()
• Publish events with observable lifecycle phases
• Type-based message routing for organized handlers

Examples

HTTP Server with Worker Processes

// primary.js
import _ClusterPrimary from 'isotropic-cluster/lib/cluster-primary.js';
import _os from 'node:os';
import _process from 'node:process';

{
    const primary = _ClusterPrimary({
        workerScript: 'worker.js'
    });

    // Log worker activity
    primary.on({
        workerExit: ({
            data: {
                code,
                signal,
                worker
            }
        }) => {
            console.log(`Worker ${worker.id} exited with code ${code} and signal ${signal}`);
        },
        workerReady: ({
            data: {
                worker
            }
        }) => {
            console.log(`Worker ${worker.id} is ready to handle connections`);
        }
    });

    {
        // Start workers based on CPU count
        const cpuCount = _os.cpus().length;

        console.log(`Starting ${cpuCount} workers`);

        primary.fork({
            workerCount: cpuCount
        });
    }

    // Handle graceful shutdown
    _process.on('SIGTERM', () => {
        console.log('Shutting down server');

        primary.shutDown();
    });
}

// worker.js
import _ClusterWorker from 'isotropic-cluster/lib/cluster-worker.js';
import _http from 'node:http';

{
    const worker = _ClusterWorker();

    // Create HTTP server
    const server = _http.createServer((request, response) => {
        response.writeHead(200);
        response.end(`Hello from worker ${_ClusterWorker.workerId}`);
    });

    // Start listening on port 3000
    server.listen(3000, () => {
        console.log(`Worker ${_ClusterWorker.workerId} listening on port 3000`);
    });

    // Handle shutdown
    worker.on('primaryDisconnect', () => {
        console.log('Primary asked worker to disconnect, closing server');
        server.close(() => {
            process.exit(0);
        });
    });
}

Task Distribution with Round-Robin

import _ClusterPrimary from 'isotropic-cluster/lib/cluster-primary.js';

{
    const primary = _ClusterPrimary(),

        // Queue of tasks to process
        tasks = [{
            data: '...',
            id: 1
        }, {
            data: '...',
            id: 2
        }, {
            data: '...',
            id: 3
        }, {
            // ... more tasks
        }],

        processQueue = () => {
            if (tasks.length) {
                const task = tasks.shift(), // Get the next task
                    worker = primary.roundRobin(); // Select a worker using round-robin

                // Send the task to the worker
                primary.send({
                    message: {
                        task,
                        type: 'processTask'
                    },
                    to: worker
                }).then(() => {
                    console.log(`Task ${task.id} sent to worker ${worker.id}`);
                }).catch(error => {
                    console.error(`Failed to send task ${task.id}:`, error);

                    // Put task back in queue
                    tasks.unshift(task);
                });
            }
        };

    // Listen for worker readiness
    primary.on('workerReady', () => {
        // Start processing tasks when at least one worker is ready
        processQueue();
    });

    // Handle task completion
    primary.on('workerMessage', ({
        data: {
            message
        }
    }) => {
        if (message.type === 'taskComplete') {
            console.log(`Task ${message.taskId} completed with result:`, message.result);

            // Process next task
            processQueue();
        }
    });

    // Start workers
    primary.fork({
        workerCount: 4
    });
}

Specialized Task Distribution

import _ClusterPrimary from 'isotropic-cluster/lib/cluster-primary.js';

{
    const primary = _ClusterPrimary(),

        // Different types of tasks
        tasksByType = {
            compute: [{
                data: '...',
                id: 'c1'
            }, {
                data: '...',
                id: 'c2'
            }],
            io: [{
                data: '...',
                id: 'i1'
            }, {
                data: '...',
                id: 'i2'
            }],
            lightweight: [{
                data: '...',
                id: 'l1'
            }, {
                data: '...',
                id: 'l2'
            }]
        },

        processQueue = queueType => {
            const tasks = tasksByType[queueType];

            if (tasks.length) {
                const task = tasks.shift(), // Get the next task
                    worker = primary.roundRobin({ // Select a worker using queue-specific round-robin
                        tag: queueType
                    });

                // Send the task to the worker
                primary.send({
                    message: {
                        task,
                        type: 'processTask'
                    },
                    to: worker
                }).then(() => {
                    console.log(`Task ${task.id} sent to worker ${worker.id}`);
                }).catch(error => {
                    console.error(`Failed to send task ${task.id}:`, error);

                    // Put task back in queue
                    tasks.unshift(task);
                });
            }
        };

    // Listen for worker readiness
    primary.on('workerReady', () => {
        // Start processing tasks when at least one worker is ready
        processQueue('compute');
        processQueue('io');
        processQueue('lightweight');
    });

    // Handle task completion
    primary.on('workerMessage', ({
        data: {
            message
        }
    }) => {
        if (message.type === 'taskComplete') {
            console.log(`Task ${message.taskId} completed with result:`, message.result);

            // Process the next task from each queue
            processQueue('compute');
            processQueue('io');
            processQueue('lightweight');
        }
    });

    // Start workers
    primary.fork({
        workerCount: 6
    });
}

Type-Based Message Routing

Message passing between the primary process and a worker process is very common. When messages are coming from multiple places with different payloads and different contexts, it can be difficult to process with a single message event handler function. There is a built-in shortcut for handling messages of different types. It requires the message to be an object with a string property named type. When the ClusterPrimary instance receives a message from a worker with a type property, it will execute a method with the name _eventWorkerMessage_${type}. You don't have to subscribe to the message event. The ClusterPrimary instance just needs to have the correct methods defined. This can be done easily with isotropic-make.

When a ClusterWorker instance receives a message from the primary with a type property, it will execute a method with the name _eventPrimaryMessage_${type}.

import _ClusterPrimary from 'isotropic-cluster/lib/cluster-primary.js';
import _make from 'isotropic-make';

// Extend ClusterPrimary with custom message handler methods
const _AdvancedPrimary = _make('AdvancedPrimary', _ClusterPrimary, {
    _eventWorkerMessage_log ({
        data: {
            message,
            worker
        }
    }) {
        // Handler for messages with type: 'log'
        console.log(`[Worker ${worker.id} LOG] ${message.level}: ${message.text}`);
    },
    _eventWorkerMessage_metric({
        data: {
            message
    } }) {
        // Handler for messages with type: 'metric'
        console.log(`METRIC: ${message.name} = ${message.value}`);

        // Store in database, etc.
        this._storeMetric({
            name: message.name,
            value: message.value
        });
    },
    _eventWorkerMessage_requestHelp({
        data: {
            message,
            worker
        }
    }) {
        // Handler for messages with type: 'requestHelp'
        console.log(`Worker ${worker.id} needs help with: ${message.problem}`);

        // Send help back to the worker
        this.send({
            message: {
                solution: this._getSolution(message.problem),
                type: 'helpResponse'
            },
            to: worker
        });
    },
    // Supporting methods
    _getSolution (problem) {
        // Return solution based on problem
        return `Solution for ${problem}`;
    },
    _storeMetric ({
        name,
        value
    }) {
        // Store metric in database
    }
});

{
    const primary = _AdvancedPrimary();

    primary.fork({
        workerCount: 2
    });
}

API Reference

ClusterPrimary

Constructor

const primary = _ClusterPrimary(options);

Options:

• restartBackoff: Controls how dead workers are replaced. By default, replacements are paced by an isotropic-backoff instance with default levels. Pass a backoff configuration object to customize the levels, pass an existing backoff instance to share one, or pass false to disable the backoff and replace dead workers immediately and indefinitely.
• workerArgs: Array of arguments to pass to worker processes
• workerScript: Path to worker script (defaults to current script)
• workerSilent: Whether to suppress worker stdout/stderr
• workerStdio: Configuration for worker stdio
• initialize: Whether to automatically initialize (defaults to true)

Methods

• destroy(): Clean up and destroy the primary instance
• fork({ workerCount }): Start new worker processes. workerCount is optional and defaults to 1.
• roundRobin({ tag }): Select a worker using round-robin distribution. The tag is optional. It returns the worker object that has been least recently selected for the given tag, or undefined if there are no workers.
• send({ message, to }): Send a message to one or more workers. to can be a worker object or a worker id. It can also be an array of either. Returns a promise.
• shutDown(): Gracefully shut down all workers

Properties

• restartBackoff: The backoff instance pacing worker replacements, or undefined when disabled
• workerById: Object mapping worker ids to worker objects
• workers: Array of all active worker objects

Events

• addWorker: When a ready worker is added to the pool
• destroy: When destroy begins
• destroyComplete When destroy completes
• fork: When fork() is called to start one or more workers
• initialize: When initialization begins
• initializeComplete: When initialization completes successfully
• initializeError: When initialization fails
• removeWorker: When a worker is removed from the pool
• restartGiveUp: When the restart backoff is exhausted and no workers remain; the cluster shuts down
• shutDown: When the cluster is shutting down
• shutDownComplete: When all workers have been shut down
• workerDisconnect: When a worker disconnects
• workerError: When a worker encounters an error
• workerExit: When a worker exits
• workerFork: When a new worker is forked
• workerListening: When a worker sets up a server and is listening
• workerMessage: When a message is received from a worker
• workerOnline: When a worker is online
• workerReady: When a worker is ready to receive work

ClusterWorker

Constructor

const worker = _ClusterWorker(options);

Options:

• initialize: Whether to automatically initialize (defaults to true)

Methods

• destroy({ timeout }): Clean up and destroy the worker instance. The worker disconnects its IPC channel and, if it has not exited on its own within timeout milliseconds (default 6765), is force-killed.
• send({ message }): Send a message to the primary process

Static Properties

• worker: Reference to the Node.js cluster worker object
• workerId: Id of the current worker

Events

• destroy: When destroy begins
• destroyComplete When destroy completes
• initialize: When initialization begins
• initializeComplete: When initialization completes successfully
• initializeError: When initialization fails
• primaryDisconnect: When the primary disconnects
• primaryMessage: When a message is received from the primary

Advanced Usage

Custom Worker Initialization

import _ClusterWorker from 'isotropic-cluster/lib/cluster-worker.js';
import _make from 'isotropic-make';
import _mongoose from 'mongoose';

// Custom worker with database connection
const _DbWorker = _make('DbWorker', _ClusterWorker, {
    get db () {
        return this._db;
    },
    // Custom initialization
    async _initialize() {
        // Connect to database
        console.log(`Worker ${_ClusterWorker.workerId} connecting to database...`);

        this._db = await _mongoose.connect('mongodb://localhost/myapp');

        // Perform other initialization
        this._setupModels();
        await this._loadInitialData();
    },
    _initializeComplete () {
        console.log(`Worker ${_ClusterWorker.workerId} database initialized`);
    },
    // Clean up resources on destroy
    _destroy (...args) {
        // Close database connection
        if (this._db) {
            console.log(`Worker ${_ClusterWorker.workerId} closing database connection`);
            this._db.disconnect();
            this._db = void null;
        }

        // Call parent destroy
        return Reflect.apply(_ClusterWorker.prototype._destroy, this, args);
    },
    // Helper methods
    async _loadInitialData () {
        // Load any required initial data
    },
    _setupModels () {
        // Define mongoose models
    }
});

{
    // Create the worker
    const worker = _DbWorker();
}

Auto-Restart on Worker Failure

The ClusterPrimary automatically restarts workers that die unexpectedly, ensuring application resilience. Replacements are paced by a restart backoff: a worker that crashes repeatedly (for example, one that fails on startup) is replaced immediately once, then with escalating delays. Each worker that reaches the ready state resets the backoff. When the backoff is exhausted and other workers remain, the primary logs an error and continues with the remaining workers; when none remain, it publishes a restartGiveUp event and shuts down rather than looping forever.

import _ClusterPrimary from 'isotropic-cluster/lib/cluster-primary.js';

{
    // Customize the restart pacing: three immediate replacements, then five
    // escalating delays from 500ms, then give up.
    const primary = _ClusterPrimary({
        restartBackoff: {
            levels: [{
                count: 3
            }, {
                count: 5,
                delay: 500,
                factor: 3,
                maximumDelay: 30000
            }]
        }
    });

    primary.on('restartGiveUp', () => {
        // The cluster could not keep any worker alive and is shutting down.
    });

    primary.fork({
        workerCount: 4
    });
}

Health Monitoring

import _ClusterPrimary from 'isotropic-cluster/lib/cluster-primary.js';
import _later from 'isotropic-later';

{
    const primary = _ClusterPrimary(),
        workerHealth = {},

        sendHealthChecks = () => {
            // Send regular health checks to workers
            for (const worker of primary.workers) {
                primary.send({
                    message: {
                        timestamp: Date.now(),
                        type: 'healthCheck'
                    },
                    to: worker
                }).catch(error => {
                    console.error(`Failed to send health check to worker ${worker.id}:`, error);
                });
            }

            _later(30000, sendHealthChecks);
        };

    // Process health responses
    primary.on('workerMessage', ({
        data: {
            message,
            worker
        }
    }) => {
        if (message.type === 'healthStatus') {
            workerHealth[worker.id] = {
                activeRequests: message.activeRequests,
                lastResponse: Date.now(),
                memoryUsage: message.memoryUsage,
                uptime: message.uptime
            };

            // Check for memory leaks or other issues
            if (message.memoryUsage.heapUsed > 1.5 * 1024 * 1024 * 1024) { // 1.5GB
                console.warn(`Worker ${worker.id} using excessive memory, scheduling restart`);

                // Gracefully restart the worker
                primary.send({
                    message: {
                        type: 'prepareForRestart'
                    },
                    to: worker
                }).then(() => {
                    // Worker will finish current tasks and exit
                });
            }
        }
    });

    // Start workers
    primary.fork({
        workerCount: 4
    });

    sendHealthChecks();
}

Integration with Other isotropic Modules

isotropic-cluster works seamlessly with other modules in the isotropic ecosystem:

• isotropic-backoff: Escalating delays for worker replacement
• isotropic-error: Nested error reporting
• isotropic-initializable: Parent-to-child initialization sequence
• isotropic-logger: Structured logging
• isotropic-make: Create constructor functions with inheritance and mixins
• isotropic-pubsub: Event system for the lifecycle events

Contributing

Please refer to CONTRIBUTING.md for contribution guidelines.

Issues

If you encounter any issues, please file them at https://github.com/ibi-group/isotropic-cluster/issues