diff --git a/packages/cli/src/constants.ts b/packages/cli/src/constants.ts index df52a36b96..78d7671e1b 100644 --- a/packages/cli/src/constants.ts +++ b/packages/cli/src/constants.ts @@ -164,3 +164,13 @@ export const LOWEST_SHUTDOWN_PRIORITY = 0; export const DEFAULT_SHUTDOWN_PRIORITY = 100; /** Highest priority, meaning shut down happens before all other groups */ export const HIGHEST_SHUTDOWN_PRIORITY = 200; + +export const WsStatusCodes = { + CloseNormal: 1000, + CloseGoingAway: 1001, + CloseProtocolError: 1002, + CloseUnsupportedData: 1003, + CloseNoStatus: 1005, + CloseAbnormal: 1006, + CloseInvalidData: 1007, +} as const; diff --git a/packages/cli/src/runners/__tests__/task-runner-ws-server.test.ts b/packages/cli/src/runners/__tests__/task-runner-ws-server.test.ts index 003e2c5f4e..b092e08fed 100644 --- a/packages/cli/src/runners/__tests__/task-runner-ws-server.test.ts +++ b/packages/cli/src/runners/__tests__/task-runner-ws-server.test.ts @@ -1,10 +1,23 @@ import type { TaskRunnersConfig } from '@n8n/config'; import { mock } from 'jest-mock-extended'; +import type WebSocket from 'ws'; -import { Time } from '@/constants'; +import { Time, WsStatusCodes } from '@/constants'; import { TaskRunnerWsServer } from '@/runners/runner-ws-server'; describe('TaskRunnerWsServer', () => { + describe('removeConnection', () => { + it('should close with 1000 status code by default', async () => { + const server = new TaskRunnerWsServer(mock(), mock(), mock(), mock(), mock()); + const ws = mock(); + server.runnerConnections.set('test-runner', ws); + + await server.removeConnection('test-runner'); + + expect(ws.close).toHaveBeenCalledWith(WsStatusCodes.CloseNormal); + }); + }); + describe('heartbeat timer', () => { it('should set up heartbeat timer on server start', async () => { const setIntervalSpy = jest.spyOn(global, 'setInterval'); diff --git a/packages/cli/src/runners/runner-ws-server.ts b/packages/cli/src/runners/runner-ws-server.ts index 8de395d971..3a5fa53029 100644 --- a/packages/cli/src/runners/runner-ws-server.ts +++ b/packages/cli/src/runners/runner-ws-server.ts @@ -4,7 +4,7 @@ import { ApplicationError } from 'n8n-workflow'; import { Service } from 'typedi'; import type WebSocket from 'ws'; -import { Time } from '@/constants'; +import { Time, WsStatusCodes } from '@/constants'; import { Logger } from '@/logging/logger.service'; import { DefaultTaskRunnerDisconnectAnalyzer } from './default-task-runner-disconnect-analyzer'; @@ -21,15 +21,7 @@ function heartbeat(this: WebSocket) { this.isAlive = true; } -const enum WsStatusCode { - CloseNormal = 1000, - CloseGoingAway = 1001, - CloseProtocolError = 1002, - CloseUnsupportedData = 1003, - CloseNoStatus = 1005, - CloseAbnormal = 1006, - CloseInvalidData = 1007, -} +type WsStatusCode = (typeof WsStatusCodes)[keyof typeof WsStatusCodes]; @Service() export class TaskRunnerWsServer { @@ -62,7 +54,7 @@ export class TaskRunnerWsServer { void this.removeConnection( runnerId, 'failed-heartbeat-check', - WsStatusCode.CloseNoStatus, + WsStatusCodes.CloseNoStatus, ); this.runnerLifecycleEvents.emit('runner:failed-heartbeat-check'); return; @@ -156,7 +148,7 @@ export class TaskRunnerWsServer { async removeConnection( id: TaskRunner['id'], reason: DisconnectReason = 'unknown', - code?: WsStatusCode, + code: WsStatusCode = WsStatusCodes.CloseNormal, ) { const connection = this.runnerConnections.get(id); if (connection) { @@ -181,7 +173,8 @@ export class TaskRunnerWsServer { // shutting them down await Promise.all( Array.from(this.runnerConnections.keys()).map( - async (id) => await this.removeConnection(id, 'shutting-down', WsStatusCode.CloseGoingAway), + async (id) => + await this.removeConnection(id, 'shutting-down', WsStatusCodes.CloseGoingAway), ), ); }