webserver-framework/src/controller/Shard.ts

// const ChildProcess = require('node:child_process');
// const { EventEmitter } = require('node:events');
import { ChildProcess, fork } from 'node:child_process';
import EventEmitter from 'node:events';

// const { Util } = require('../util');
import { Util } from '../util/index.js';
import Controller from './Controller.js';
import { ShardOptions } from '../../@types/Controller.js';
import { IPCMessage } from '../../@types/Other.js';
import { ServerOptions } from '../../@types/Server.js';
import path from 'node:path';

// Give subprocess 90s to shut down before being forcibly killed
const KillTO = 90 * 1000;
class Shard extends EventEmitter {

    // #_controller: Controller;
    #_id: number;
    #_filePath: string;
    #_args: string[];
    #_execArgv: string[];
    #_env: { [key: string]: string };
    #_respawn: boolean;
    #_serverOptions: ServerOptions;
    
    #_ready: boolean;
    #_process: ChildProcess | null;
    #_fatal: boolean;

    #_crashes: number[];
    #_spawnedAt: number;
    #_awaitingShutdown: (() => void) | null;
    #_awaitingResponse: Map<string, (args: IPCMessage) => void>;

    constructor (_controller: Controller, id: number, options: ShardOptions) {
        super();

        // this.#_controller = controller;
        if (typeof id !== 'number' || isNaN(id)) 
            throw new Error('Missing ID');
        this.#_id = id;
        
        if (!options?.path) 
            throw new Error('Missing path to file to fork');
        this.#_filePath = options.path;
        this.#_args = options.args || [];
        this.#_execArgv = options.execArgv || [];
        this.#_env = options.env || {};
        this.#_respawn = options.respawn || false;
        this.#_serverOptions = options.serverOptions || {} as ServerOptions;
        this.#_serverOptions.dir = path.resolve(options.path, '..');

        this.#_ready = false;
        this.#_process = null;
        this.#_fatal = false;

        // Keep track of crashes for preventing crash loops
        this.#_crashes = [];
        // Set in the spawn method
        this.#_spawnedAt = Date.now(); // Gets re-set once actually spawned

        this.#_awaitingShutdown = null;

        this.#_awaitingResponse = new Map();

    }

    get id () {
        return this.#_id;
    }

    get fatal () {
        return this.#_fatal;
    }

    get process () {
        return this.#_process;
    }

    get ready () {
        return this.#_ready;
    }

    get spawnedAt () {
        return this.#_spawnedAt;
    }

    async spawn (waitForReady = false) {

        if (this.fatal) 
            throw new Error(`[shard-${this.id}] Process died fatally and cannot be restarted. Fix the issue before trying again.`);
        if (this.process) 
            throw new Error(`[shard-${this.id}] A process for this shard already exists!`);

        this.#_process = fork(this.#_filePath, this.#_args, {
            env: {
                ...this.#_env,
                SHARD_ID: this.id.toString()
            },
            execArgv: this.#_execArgv
        })
            .on('message', this._handleMessage.bind(this))
            .on('exit', this._handleExit.bind(this))
            .on('disconnect', this._handleDisconnect.bind(this)); // Don't know if this is going to help, but monitoring whether this gets called whenever a process on its own closes the IPC channel
        
        this.#_process.once('spawn', () => {
            this.emit('spawn');
            if (!this.#_process) 
                throw new Error('Shut up TS');
            this.#_process.send({ _start: this.#_serverOptions });
            this.#_spawnedAt = Date.now();
        });
        if (!waitForReady) 
            return;
        
        return new Promise((resolve, reject) => {
            this.once('ready', resolve);
            this.once('disconnect', () => reject(new Error(`[shard-${this.id}] Shard disconnected while starting up`)));
            this.once('death', () => reject(new Error(`[shard-${this.id}] Shard died while starting`)));
            setTimeout(() => reject(new Error(`[shard-${this.id}] Shard timed out while starting`)), 30_000);
        });

    }

    async respawn (delay = 500) {
        await this.kill();
        if (delay) 
            await Util.wait(delay);
        return this.spawn();
    }

    /**
     * Sends a shutdown command to the shard, if it doesn't respond within 5 seconds it gets killed
     * TODO: Add a check to see if the process actually ends and print out a warning if it hasn't
     *
     * @return {*} 
     * @memberof Shard
     */
    kill (): Promise<void> | void {
        if (this.process) {
            return new Promise<void>((resolve) => {
                // Clear out all other exit listeners so they don't accidentally start the process up again
                if (!this.#_process) 
                    return resolve();
                this.#_process.removeAllListeners('exit');
                // Set timeout for force kill
                const to = setTimeout(() => {
                    if (!this.#_process) 
                        return resolve();
                    this.#_process.kill();
                    resolve();
                }, KillTO);
                // Gracefully handle exit
                this.#_process.once('exit', (code, signal) => {
                    clearTimeout(to);
                    this._handleExit(code, signal, false);
                    resolve();
                });
                // Clear the force kill timeout if the process responds with a shutdown echo, allowing it time to gracefully close all connections
                this.once('shutdown', () => {
                    clearTimeout(to);
                });
                
                this.#_process.send({ _shutdown: true });
            });
        }
        this._handleExit(null, null, false);
    }

    send (message: IPCMessage, expectResponse = false): Promise<IPCMessage | void> {
        
        if (!this.ready || !this.#_process) 
            return Promise.reject(new Error(`[shard-${this.id}] Cannot send message to dead shard.`));
        
        return new Promise<IPCMessage | void>((resolve, reject) => {

            if (expectResponse) {
                message._id = Util.randomUUID();
                const to = setTimeout(reject, 10_000, [ new Error('Message timeout') ]);
                this.#_awaitingResponse.set(message._id, (args: IPCMessage) => {
                    clearTimeout(to);
                    resolve(args);
                });
            }

            this.#_process?.send(message, err => {
                if (err) 
                    return reject(err);
                
                if (!expectResponse) 
                    resolve();
            });

        });
    }

    awaitShutdown () {
        this.#_respawn = false;
        return new Promise<void>((resolve) => {
            this.#_awaitingShutdown = resolve;
        });
    }

    _handleMessage (message: IPCMessage) {
        if (message) {
            if (message._ready) {
                this.#_ready = true;
                this.emit('ready');
                return;
            } else if (message._shutdown) {
                setTimeout(() => {
                    if (this.process) 
                        this.process.kill('SIGKILL');
                }, KillTO);
                this.#_ready = false;
                this.emit('shutdown');
                return;
            } else if (message._fatal) {
                this.#_process?.removeAllListeners();
                this.#_ready = false;
                this.#_fatal = true;
                this._handleExit(null, null, false);
                return this.emit('fatal', message);
            } else if (message._id) {
                const promise = this.#_awaitingResponse.get(message._id);
                if (promise) 
                    return promise(message);
            }
        }

        this.emit('message', message);

    }

    _handleDisconnect () {
        this.emit('disconnect');
    }

    _handleExit (code: number | null, _signal: string | null, respawn = this.#_respawn) {
        if (this.process) 
            this.process.removeAllListeners();
        
        this.emit('death');
        if (this.#_awaitingShutdown) 
            this.#_awaitingShutdown();

        if (code !== 0) 
            this.#_crashes.push(Date.now() - this.spawnedAt);

        this.#_ready = false;
        this.#_process = null;

        const len = this.#_crashes.length;
        if (len > 2) {
            const last3 = this.#_crashes.slice(len - 3);
            const sum = last3.reduce((s, val) => {
                s += val;
                return s;
            }, 0);
            const avg = sum / 3;
            // If average run duration is below 60 mins send a notification about detected crash loop and stop the respawn
            if (avg < 60 * 60 * 1000) {
                this.emit('warn', `Crash loop detected, average run time for the last 3 spawns: ${avg}`);
            }
            respawn = false;
        }

        if (respawn) 
            this.spawn().catch(err => this.emit('error', err));
        
    }

}

export default Shard;