mirror of
https://github.com/n8n-io/n8n.git
synced 2024-12-28 22:19:41 -08:00
fix(core): Correct invalid WS status code on removing connection (#11901)
This commit is contained in:
parent
97269a3703
commit
1d80225d26
|
@ -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;
|
||||
|
|
|
@ -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<WebSocket>();
|
||||
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');
|
||||
|
|
|
@ -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),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue