diff --git a/packages/ws-worker/CHANGELOG.md b/packages/ws-worker/CHANGELOG.md index 7a474503e..d76d74f3e 100644 --- a/packages/ws-worker/CHANGELOG.md +++ b/packages/ws-worker/CHANGELOG.md @@ -1,5 +1,11 @@ # ws-worker +## 1.16.0 + +### Minor Changes + +- 64176c6: Report on configured capacity when connecting to the worker channel + ## 1.15.4 ### Patch Changes diff --git a/packages/ws-worker/package.json b/packages/ws-worker/package.json index b7be5260b..9d3b841ba 100644 --- a/packages/ws-worker/package.json +++ b/packages/ws-worker/package.json @@ -1,6 +1,6 @@ { "name": "@openfn/ws-worker", - "version": "1.15.4", + "version": "1.16.0", "description": "A Websocket Worker to connect Lightning to a Runtime Engine", "main": "dist/index.js", "type": "module", diff --git a/packages/ws-worker/src/channels/worker-queue.ts b/packages/ws-worker/src/channels/worker-queue.ts index a293e4f42..3a081c42a 100644 --- a/packages/ws-worker/src/channels/worker-queue.ts +++ b/packages/ws-worker/src/channels/worker-queue.ts @@ -20,6 +20,7 @@ const connectToWorkerQueue = ( options: { messageTimeout?: number; claimTimeout?: number; + capacity?: number; SocketConstructor?: any; } ) => { @@ -28,6 +29,7 @@ const connectToWorkerQueue = ( // This can be overridden by different channels (although we tend not to) messageTimeout = DEFAULT_MESSAGE_TIMEOUT_SECONDS, claimTimeout = DEFAULT_CLAIM_TIMEOUT_SECONDS, + capacity, SocketConstructor = PhxSocket, } = options; @@ -72,7 +74,8 @@ const connectToWorkerQueue = ( didOpen = true; shouldReportConnectionError = true; - const channel = socket.channel('worker:queue') as Channel; + const joinPayload = { capacity }; + const channel = socket.channel('worker:queue', joinPayload) as Channel; channel.onMessage = (ev, load) => { events.emit('message', ev, load); diff --git a/packages/ws-worker/src/mock/sockets.ts b/packages/ws-worker/src/mock/sockets.ts index 5d04864b8..9d2ac18b2 100644 --- a/packages/ws-worker/src/mock/sockets.ts +++ b/packages/ws-worker/src/mock/sockets.ts @@ -121,10 +121,12 @@ export class MockSocket { }); } - channel(topic: string) { + channel(topic: string, params?: any) { if (!this.allChannels[topic]) { this.allChannels[topic] = mockChannel(); } + // @ts-ignore + this.allChannels[topic]._joinParams = params; return this.allChannels[topic]; } } diff --git a/packages/ws-worker/src/server.ts b/packages/ws-worker/src/server.ts index 57e41ec41..a5555fac6 100644 --- a/packages/ws-worker/src/server.ts +++ b/packages/ws-worker/src/server.ts @@ -169,6 +169,7 @@ function connect(app: ServerApp, logger: Logger, options: ServerOptions = {}) { messageTimeout: options.socketTimeoutSeconds ?? options.messageTimeoutSeconds, claimTimeout: options.claimTimeoutSeconds, + capacity: options.maxWorkflows, }) .on('connect', onConnect) .on('disconnect', onDisconnect) diff --git a/packages/ws-worker/test/channels/worker-queue.test.ts b/packages/ws-worker/test/channels/worker-queue.test.ts index cbbdf2a80..f2a21cee2 100644 --- a/packages/ws-worker/test/channels/worker-queue.test.ts +++ b/packages/ws-worker/test/channels/worker-queue.test.ts @@ -106,5 +106,32 @@ test('should fail to connect with an invalid auth token', async (t) => { }); }); +test('should pass capacity in join payload when provided', (t) => { + return new Promise((done) => { + connectToWorkerQueue('www', 'a', 'secret', logger, { + capacity: 10, + SocketConstructor: MockSocket as any, + }).on('connect', ({ socket }) => { + const channel = socket.allChannels['worker:queue']; + // @ts-ignore - accessing test property + t.is(channel._joinParams.capacity, 10); + done(); + }); + }); +}); + +test('should pass capacity in join payload when not provided', (t) => { + return new Promise((done) => { + connectToWorkerQueue('www', 'a', 'secret', logger, { + SocketConstructor: MockSocket as any, + }).on('connect', ({ socket }) => { + const channel = socket.allChannels['worker:queue']; + // @ts-ignore - accessing test property + t.is(channel._joinParams.capacity, undefined); + done(); + }); + }); +}); + // TODO maybe? test.todo('should reconnect with backoff when connection is dropped');