report capacity

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

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",

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);

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];
   }
 }

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)

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');