webserver-framework/src/controller/Shard.js

const ChildProcess = require('node:child_process');
const { EventEmitter } = require('node:events');

const { Util } = require('../util');

class Shard extends EventEmitter {

    constructor (controller, id, options = {}) {
        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 || {};

        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 = null;

    }

    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 = ChildProcess.fork(this.filePath, this.args, { env: { ...this.env, SHARD_ID: this.id }, 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');
            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 () {
        if (this.process) {
            return new Promise((resolve) => {
                // Clear out all other exit listeners so they don't accidentally start the process up again
                this.process.removeAllListeners('exit');
                // Set timeout for force kill
                const to = setTimeout(() => {
                    this.process.kill();
                    resolve();
                }, 5000);
                // 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) {
        return new Promise((resolve, reject) => {
            if (this.ready && this.process) {
                this.process.send(message, err => {
                    if (err) reject(err);
                    else resolve();
                });
            } else reject(new Error(`[shard-${this.id}] Cannot send message to dead shard.`));
        });
    }

    _handleMessage (message) {
        if (message) {
            if (message._ready) {
                this.ready = true;
                this.emit('ready');
                return;
            } else if (message._shutdown) {
                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);
                this.emit('fatal', message);
            }
        }

        this.emit('message', message);

    }

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

    _handleExit (code, signal, respawn = this._respawn) {
        if (this.process) this.process.removeAllListeners();
        this.emit('death');

        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));
        
    }

}

module.exports = Shard;