From f5255d6f765aee79e00a4eac10ade8fd998163db Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Fri, 26 Jan 2024 13:02:26 +0100 Subject: [PATCH] Fix message seq reuse on forwarding (#579) When messages were forwarded they were sent using the low level connector .send method which accepts a message and sends it over the link. This method sends the message as is with no modification which means that at no point during the forwarding logic was the message sequence number that originated from Link A renumbered to the correct sequence number to send out on Link B which it was forwarded on. This in turn caused the logic which drops acked messages from the send buffer to break which in turn meant the resume could cause duplication of message and/or messages to be lost. With enough duplicated messages the resume of one connection could cause other connections to drop and also initiate resume. Fix by renumbering the sequence on forwarded messages. In the future the sequences sent and received should also be checked to make sure this kind of problem does not happen again, this is not done here because partially updating a running cluster with sequence problems to one that checks and enforces sequences numbers will be problematic. --- CHANGELOG.md | 7 +++++ packages/controller/src/ControllerRouter.ts | 6 ++-- packages/host/src/Host.ts | 10 +++--- packages/lib/src/link/connectors.ts | 34 +++++++++++++++++++-- packages/lib/src/link/link.ts | 5 +-- test/mock.js | 4 +++ 6 files changed, 54 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ece0937..7e03d59a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## Version 2.0.0-alpha.16 + +## Changes + +- Fixed message sequences not being renumbered when forwarded between links causing reconnects to send duplicated messages. + + ## Version 2.0.0-alpha.15 ### Features diff --git a/packages/controller/src/ControllerRouter.ts b/packages/controller/src/ControllerRouter.ts index 4679312d..4072c941 100644 --- a/packages/controller/src/ControllerRouter.ts +++ b/packages/controller/src/ControllerRouter.ts @@ -84,7 +84,7 @@ export default class ControllerRouter { if (message.type === "request") { nextHop.forwardRequest(message as lib.MessageRequest, origin); } else { - nextHop.connector.send(message); + nextHop.connector.forward(message); } } else { this.warnUnrouted(message, msg); @@ -107,13 +107,13 @@ export default class ControllerRouter { if (dst.id === lib.Address.host || dst.id === lib.Address.instance) { for (let hostConnection of this.controller.wsServer.hostConnections.values()) { if (hostConnection !== origin && (!plugin || hostConnection.plugins.has(plugin))) { - hostConnection.connector.send(message); + hostConnection.connector.forward(message); } } } else if (dst.id === lib.Address.control) { for (let controlConnection of this.controller.wsServer.controlConnections.values()) { if (controlConnection !== origin) { - controlConnection.connector.send(message); + controlConnection.connector.forward(message); } } } else { diff --git a/packages/host/src/Host.ts b/packages/host/src/Host.ts index b1b9dd41..c2fc5d67 100644 --- a/packages/host/src/Host.ts +++ b/packages/host/src/Host.ts @@ -173,16 +173,16 @@ export class HostRouter { if (dst.id === lib.Address.instance) { for (let instanceConnection of this.host.instanceConnections.values()) { if (instanceConnection !== origin && (!plugin || instanceConnection.plugins.has(plugin))) { - instanceConnection.connector.send(message); + instanceConnection.connector.forward(message); } } if (this.host !== origin && (!plugin || this.host.serverPlugins.has(plugin))) { - this.host.connector.send(message); + this.host.connector.forward(message); } } else if (dst.id === lib.Address.control) { if (this.host !== origin) { if (!plugin || this.host.serverPlugins.has(plugin)) { - this.host.connector.send(message); + this.host.connector.forward(message); } } else { logger.warn(`Received control broadcast of ${(message as lib.MessageEvent).name} from controller.`); @@ -208,7 +208,7 @@ export class HostRouter { } this.host._connectInstance(dst.id).then(instanceConnection => { - instanceConnection.connector.send(message); + instanceConnection.connector.forward(message); }).catch(err => { logger.error(`Error starting instance:\n${err.stack}`, this.host.instanceLogMeta(dst.id)); origin.connector.sendResponseError( @@ -222,7 +222,7 @@ export class HostRouter { if (message.type === "request") { nextHop.forwardRequest(message as lib.MessageRequest, origin); } else { - nextHop.connector.send(message); + nextHop.connector.forward(message); } } catch (err: any) { if (message.type === "request") { diff --git a/packages/lib/src/link/connectors.ts b/packages/lib/src/link/connectors.ts index 2bc15f19..999800ba 100644 --- a/packages/lib/src/link/connectors.ts +++ b/packages/lib/src/link/connectors.ts @@ -33,7 +33,30 @@ export abstract class BaseConnector extends events.EventEmitter { this._seq = 1; } - abstract send(message: libData.Message): void; + protected abstract send(message: libData.MessageRoutable): void; + + forward(message: libData.MessageRoutable) { + const seq = this._seq; + this._seq += 1; + // The message type currently can't be inferred like a tagged union + // TODO fix typing of message so it can be deduced from .type + if (message.type === "request") { + const request = message as libData.MessageRequest; + this.send(new libData.MessageRequest(seq, request.src, request.dst, request.name, request.data)); + } else if (message.type === "response") { + const response = message as libData.MessageResponse; + this.send(new libData.MessageResponse(seq, response.src, response.dst, response.data)); + } else if (message.type === "responseError") { + const error = message as libData.MessageResponseError; + this.send(new libData.MessageResponseError(seq, error.src, error.dst, error.data)); + } else if (message.type === "event") { + const event = message as libData.MessageEvent; + this.send(new libData.MessageEvent(seq, event.src, event.dst, event.name, event.data)); + } else { + throw new Error(`Cannot forward message type ${(message as any).type}`); + } + return seq; + } sendRequest( request: Request, @@ -242,7 +265,7 @@ export abstract class WebSocketBaseConnector extends BaseConnector { * * @param message - Message to send. */ - send(message: libData.MessageRoutable) { + protected send(message: libData.MessageRoutable) { if (!["connected", "resuming"].includes(this._state)) { throw new libErrors.SessionLost("No session"); } @@ -295,6 +318,13 @@ export abstract class WebSocketBaseConnector extends BaseConnector { await this.close(1000, "Disconnect"); } + /** + * Notify other end that the link is ready to disconnect + */ + sendDisconnectReady() { + this._sendInternal(new libData.MessageDisconnect("ready")); + } + /** * True if the connection is established and active */ diff --git a/packages/lib/src/link/link.ts b/packages/lib/src/link/link.ts index 520e5fe4..e3a746c0 100644 --- a/packages/lib/src/link/link.ts +++ b/packages/lib/src/link/link.ts @@ -132,7 +132,8 @@ export class Link { } ).finally( () => { - this.connector.send(new libData.MessageDisconnect("ready")); + // Only WebSocket connectors issue disconnection events. + (this.connector as WebSocketBaseConnector).sendDisconnectReady(); }, ); }); @@ -561,7 +562,7 @@ export class Link { dst: message.dst, }; this._forwardedRequests.set(message.src.index(), pending); - this.connector.send(message); + this.connector.forward(message); } sendEvent(event: Event, dst: libData.Address) { diff --git a/test/mock.js b/test/mock.js index a00729c9..171c5923 100644 --- a/test/mock.js +++ b/test/mock.js @@ -80,6 +80,10 @@ class MockConnector extends lib.BaseConnector { this.sentMessages.push(message); setImmediate(() => this.emit("send", message)); } + + sendDisconnectReady() { + this.send(new lib.MessageDisconnect("ready")); + } } class MockServer extends events.EventEmitter {