@betsys-nestjs/cluster-worker

Cluster - worker library

This library is quite helpful to manage multi-process applications with slave master architecture. Take this readme as a manual, how to write such an application.

Environment variables

There are no variables necessary for this library.

Dependencies

Package	Version
on-headers	^1.0.2
@nestjs/common	^10.0.0
@nestjs/core	^10.0.0
@nestjs/event-emitter	^2.0.0
@nestjs/schedule	^3.0.0
child-process	^1.0.0
reflect-metadata	^0.1.13
rxjs	^7.1.0

Usage

In order to use this module, you need to create 2 modules - one with cluster logic (master) and one with worker logic (slave) and then incorporate the cluster logic module into the app module.

Cluster module

import { ClusterModule } from '@betsys/cluster-worker';
import { Module } from '@nestjs/common';
import { CallbackProvider } from './listener/callback.provider';
import { ReplicationWorkerProvider } from './provider/replication-worker.provider';

@Module({
    imports: [
        ClusterModule.forRoot(),
    ],
    providers: [
        CallbackProvider,
        ReplicationWorkerProvider,
    ],
    exports: [
        ReplicationWorkerProvider,
    ],
})
export class ClusterLogicModule {
}

ClusterModule is the module imported from the library. CallbackProvider is a class defining, how the cluster should react to messages from worker.

import { Injectable } from '@nestjs/common';

enum MyMessage {
    DoSomething = 'DoSomething',
}

@Injectable()
export class CallbackProvider implements CallbackProviderInterface<MyMessage> {
    private callbacks: { message: MyMessage; callable: ClusterCallbackType }[] = [];

    constructor() {
        this.addCallback(MyMessage.DoSomething, this.doSomething);
    }
    
    doSomething(): void {
        ...
    }
}

When you already have defined behavior of master for given message, you need to implement a worker provider:

import {
    SingleWorkerProvider,
    WorkerProviderService,
    WorkerStatusWaitingService,
    InjectLogger,
    ClusterWorkerLoggerInterface,
} from '@betsys/cluster-worker';
import { forwardRef, Inject, Injectable } from '@nestjs/common';
import { CallbackProvider } from '../listener/callback.provider';

enum MyWorkerProviderEnum {
    SillyWorker = 'SillyWorker',
}

enum MyMessage {
    DoSomething = 'DoSomething',
}

@Injectable()
export class MyWorkerProviderService extends SingleWorkerProvider<never, MyMessage, MyWorkerProviderEnum> {
    constructor(
        @InjectLogger()
        protected readonly logger: ClusterWorkerLoggerInterface,
        protected readonly workerProvider: WorkerProviderService,
        @Inject(forwardRef(() => CallbackProvider))
        protected readonly callbackProvider: CallbackProvider,
        protected readonly waitingService: WorkerStatusWaitingService,
    ) {
        super(
            logger,
            workerProvider,
            callbackProvider,
            MyWorkerProviderEnum.SillyWorker,
            'spawn message that appears in the log',
            waitingService,
        );
    }
}

Such a worker provider can extend SingleWorkerProvider (when you need just a single worker for given task) or MultiWorkerProvider (when you need a pool of workers for the same task).

A slave worker process is represented as a WorkerMeta object in the cluster:

import cluster from 'cluster';
import { WorkerStatus } from './worker-status.enum';

export interface WorkerMeta<TData = any, TWorkerStatus = any> {
    data?: TData;
    process: cluster.Worker;
    status: WorkerStatus | TWorkerStatus;
    pingPending: boolean;
}

With the type TData you can define additional data for the worker (i.e. if it has to consume some queue from RabbitMQ, you can define the queue name over here) and TWorkerStatus defines the status the worker is in (it is your custom worker status, i.e. Replicating). WorkerStatus has three values: Booting, Ready and Error.

MultiWorkerProvider offers following methods:

// Respawns all workers, which WorkerStatus is Error
respawnWorkersWithError(): Promise<WorkerMeta<TWorkerData>[]>;
// Spawns workers with data you want to pass to it
spawnWorkersWithData(
    workerData: TWorkerData[],
    mapper: (data: TWorkerData) => WorkerMeta<TWorkerData>,
): Promise<WorkerMeta<TWorkerData>[]>;
// Spawn workers without data
spawnWorkers(count: number): Promise<WorkerMeta<TWorkerData>[]>;
// Sends a message to all workers
sendToAllWorkers(message: string, sendHandle?: string, callback?: 
    (error: Error | null) => void): boolean;
// This is obvious, right?
getWorkers(): WorkerMeta<TWorkerData>[];

An example of mapper to spawn workers with data would be:

(data: MyDataType): WorkerMeta<MyDataType> => ({
    data,
    pingPending: false,
    process: cluster.fork(),
    status: WorkerStatus.Booting,
});

Number of workers will be defined by number of elements in workerData parameter.

SingleWorkerProvider provides following methods:

spawnWorker(): Promise<WorkerMeta<TWorkerData>>;
getWorker(): WorkerMeta<TWorkerData> | null;
// Little bit morbid, right?
destroyWorker(): Promise<void>;

Worker module

When you have everything set up with cluster, it comes the time to tune up the worker:

import {
    ClusterModule,
    StatusService,
    WorkerListenerService,
    WorkerModule as LibWorkerModule,
    WorkerStatus,
    WorkerCallbackProvider,
    InjectLogger,
    ClusterWorkerLoggerInterface,
} from '@betsys/cluster-worker';
import { Module } from '@nestjs/common';
import cluster from 'cluster';

@Module({
    imports: [
        ClusterModule.forRoot(),
        LibWorkerModule.forRoot(),
    ],
    providers: [
        WorkerCallbackProvider,
    ],
})
export class WorkerLogicModule {
    constructor(
        @InjectLogger()
        private readonly logger: ClusterWorkerLoggerInterface,
        private readonly statusService: StatusService,
        private readonly workerCallbackProvider: WorkerCallbackProvider,
        private readonly workerListenerService: WorkerListenerService,
    ) {
        if (!cluster.isWorker) {
            throw new Error('This service must be run only from worker process');
        }
    }

    async onApplicationBootstrap(): Promise<void> {
        const callbacks = this.workerCallbackProvider.getCallbacks();
        this.workerListenerService.addCustomListeners(callbacks);
        this.statusService.changeStatus(WorkerStatus.Ready);
    }
}

In WorkerCallbackProvider you can define, how the worker will react to a message from master.

enum MyMessage {
    DoSomethingSlave = 'DoSomethingSlave',
}

@Injectable()
export class WorkerCallbackProvider
    implements WorkerCallbackProviderInterface<MyMessage> {
    private callbacks: { message: MyMessage; callable: WorkerCallbackType }[] = [];

    constructor() {
        this.doSomething = this.doSomething.bind(this);
        
        this.addCallback(MyMessage.DoSomethingSlave, this.doSomething);
    }

    doSomething(): void {
        ...
    }
}

AppModule and starting of the app

Now it comes the time to put everything together in AppModule:

import { ClusterModule } from '@betsys/cluster-worker';
import { Module } from '@nestjs/common';
import { ClusterLogicModule } from './cluster-logic/cluster-logic.module';

@Module({
    imports: [
        ClusterLogicModule,
        ClusterModule.forRoot(),
    ],
})
export class AppModule implements BeforeApplicationShutdown {
    constructor(
        // This is from cluster-logic module
        private myWorkerProvider: MyWorkerProvider,
        private readonly workerStatusWaitingService: WorkerStatusWaitingService,
    ) {
    }

    async onApplicationBootstrap(): Promise<void> {
        const worker = await this.myWorkerProvider.spawnWorker();
        await this.workerStatusWaitingService.waitUntilWorkerMetaHasStatus(
            worker,
            WorkerStatus.Ready,
        );
        worker.process.send({ type: MyMessage.DoSomething });
    }
}

Now all you have to do is to define main.ts:

import cluster from 'cluster';
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { WorkerModule } from './worker-logic/worker-logic.module';

async function bootstrapMaster(): Promise<void> {
    const app = await NestFactory.create(AppModule);
    app.enableShutdownHooks();

    const port = Number(process.env.CONFIG_LISTEN_PORT);
    await app.listen(port);

    const logger = await app.resolve(Logger);
    logger.info(`Application listens on port ${port}`);
}

async function bootstrapWorker(): Promise<void> {
    const app = await NestFactory.create(WorkerLogicModule);
    app.enableShutdownHooks();
    await app.init();
}

const bootstrap = cluster.isMaster ? bootstrapMaster : bootstrapWorker;
bootstrap()
    .catch((e) => {
        console.error(e);
        process.exit(1);
    });

So basically you start the AppModule, which is the master process (so bootstrapMaster is called). Within this process, a new slave process is spawned and the main.ts is called again. However, this time, as it is a slave process, bootstrapWorker is called. Now you can send messages from master to worker or from worker to master (you just need to define callbacks for a particular message).