// 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 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 { if (this.process) { return new Promise((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 { if (!this.ready || !this.#_process) return Promise.reject(new Error(`[shard-${this.id}] Cannot send message to dead shard.`)); return new Promise((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((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)); } } // module.exports = Shard; export default Shard;