fix: don't reconnect a websocket after close()

.github/workflows/release.yml

@@ -16,10 +16,7 @@ jobs:
       # Setup .npmrc file to publish to npm
       - uses: actions/setup-node@v5
         with:
-          node-version: '16.x'
+          node-version: '24.x'
           registry-url: 'https://registry.npmjs.org'
-      # Ensure npm 11.5.1 or later required by trusted publishing
-      - name: Update npm
-        run: npm install -g npm@11.5.1
       - run: npm ci
       - run: npm publish

.github/workflows/setlatest.yml

@@ -18,9 +18,6 @@ jobs:
     steps:
       - uses: actions/setup-node@v5
         with:
-          node-version: '16.x'
+          node-version: '24.x'
           registry-url: 'https://registry.npmjs.org'
-      # Ensure npm 11.5.1 or later required by trusted publishing
-      - name: Update npm
-        run: npm install -g npm@11.5.1
       - run: npm dist-tag add @hyperledger/firefly-sdk@${{ inputs.version }} latest

.github/workflows/test.yml

@@ -0,0 +1,22 @@
+name: Tests
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v5
+      - uses: actions/setup-node@v5
+        with:
+          node-version: '24.x'
+          cache: npm
+      # npm ci runs prepare, which builds the package, so a broken build fails here
+      - run: npm ci
+      - run: npm test

lib/websocket.ts

@@ -30,11 +30,12 @@ export class FireFlyWebSocket {
   private readonly logger = new Logger(FireFlyWebSocket.name);
 
   private socket?: WebSocket;
-  private closed? = () => {};
+  private notifyClosed? = () => {};
   private pingTimer?: NodeJS.Timeout;
   private disconnectTimer?: NodeJS.Timeout;
   private reconnectTimer?: NodeJS.Timeout;
   private disconnectDetected = false;
+  private closing = false;
 
   constructor(
     private options: FireFlyWebSocketOptions,
@@ -44,12 +45,10 @@ export class FireFlyWebSocket {
   }
 
   private connect() {
-    // Ensure we've cleaned up any old socket
+    // Ensure we've cleaned up any old socket. close() clears the reconnect timer and marks
+    // us as closing, so clear that flag again now we are re-opening.
     this.close();
-    if (this.reconnectTimer) {
-      clearTimeout(this.reconnectTimer);
-      delete this.reconnectTimer;
-    }
+    this.closing = false;
 
     let url = `${this.options.host}/ws`;
     if (this.options.ephemeral !== undefined) {
@@ -65,7 +64,7 @@ export class FireFlyWebSocket {
       auth,
       handshakeTimeout: this.options.heartbeatInterval,
     }));
-    this.closed = undefined;
+    this.notifyClosed = undefined;
 
     socket
       .on('open', () => {
@@ -95,9 +94,9 @@ export class FireFlyWebSocket {
         this.logger.error('Error', err.stack);
       })
       .on('close', () => {
-        if (this.closed) {
+        if (this.notifyClosed) {
           this.logger.log('Closed');
-          this.closed(); // do this after all logging
+          this.notifyClosed(); // do this after all logging
         } else {
           this.disconnectDetected = true;
           this.reconnect('Closed by peer');
@@ -153,6 +152,12 @@ export class FireFlyWebSocket {
   }
 
   private reconnect(msg: string) {
+    // A handler that was already in flight when close() was called must not be able to
+    // bring the socket back - unexpected-response in particular reconnects from a deferred
+    // stream flush, which can land after the close has completed.
+    if (this.closing) {
+      return;
+    }
     if (!this.reconnectTimer) {
       this.close();
       this.logger.error(`Websocket closed: ${msg}`);
@@ -184,8 +189,16 @@ export class FireFlyWebSocket {
 
   async close(wait?: boolean): Promise<void> {
     const closedPromise = new Promise<void>((resolve) => {
-      this.closed = resolve;
+      this.notifyClosed = resolve;
     });
+    // Closing is final until connect() is called again. Without this, a reconnect that was
+    // already scheduled still fires, and a caller that has dropped its reference to this
+    // socket has no way left to stop it retrying.
+    this.closing = true;
+    if (this.reconnectTimer) {
+      clearTimeout(this.reconnectTimer);
+      delete this.reconnectTimer;
+    }
     this.clearPingTimers();
     if (this.socket) {
       try {

test/websocket.ts

@@ -0,0 +1,137 @@
+import * as assert from 'assert';
+import { AddressInfo } from 'net';
+import { WebSocketServer } from 'ws';
+import { FireFlyWebSocket } from '../lib/websocket';
+
+// These are private, but there is no other way to simulate a handler that was still in flight
+// when the socket was closed, or to guarantee a socket is down at the end of a test.
+type SocketInternals = {
+  options: { reconnectDelay: number };
+  reconnectTimer?: NodeJS.Timeout;
+  reconnect(msg: string): void;
+};
+
+const internals = (socket: FireFlyWebSocket) => socket as unknown as SocketInternals;
+
+const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+describe('websocket', () => {
+  const reconnectDelay = 100;
+
+  let server: WebSocketServer;
+  let sockets: FireFlyWebSocket[];
+  let upgradeAttempts: number;
+  let acceptConnections: boolean;
+  let previousLogLevel: string | undefined;
+
+  beforeEach(async () => {
+    // Logger reads the level when it is constructed, so this has to happen before any socket
+    previousLogLevel = process.env.FF_SDK_LOG_LEVEL;
+    process.env.FF_SDK_LOG_LEVEL = 'NONE';
+
+    sockets = [];
+    upgradeAttempts = 0;
+    acceptConnections = false;
+    server = new WebSocketServer({
+      port: 0,
+      verifyClient: (info, cb) => {
+        upgradeAttempts++;
+        // Rejecting sends a 401, so the attempt reaches the unexpected-response handler
+        // that the SDK reconnects from
+        acceptConnections ? cb(true) : cb(false, 401, 'Unauthorized', {});
+      },
+    });
+    await new Promise<void>((resolve) => server.on('listening', resolve));
+  });
+
+  afterEach(async () => {
+    // close() should stop a socket reconnecting on its own. Disabling reconnection here
+    // too means that if it ever stops doing so, the test fails its assertion instead of
+    // leaving a socket retrying against a closed port, which hangs the whole run.
+    for (const socket of sockets) {
+      const socketInternals = internals(socket);
+      socketInternals.options.reconnectDelay = -1;
+      if (socketInternals.reconnectTimer) {
+        clearTimeout(socketInternals.reconnectTimer);
+        delete socketInternals.reconnectTimer;
+      }
+      await socket.close();
+    }
+    process.env.FF_SDK_LOG_LEVEL = previousLogLevel;
+    await new Promise<void>((resolve, reject) =>
+      server.close((err) => (err ? reject(err) : resolve())),
+    );
+  });
+
+  function newSocket() {
+    const { port } = server.address() as AddressInfo;
+    const socket = new FireFlyWebSocket(
+      {
+        host: `ws://127.0.0.1:${port}`,
+        namespace: 'ns',
+        subscriptions: [],
+        reconnectDelay,
+        heartbeatInterval: 30000,
+        ephemeral: { namespace: 'ns', filter: { events: 'x' } },
+      },
+      () => {},
+    );
+    sockets.push(socket);
+    return socket;
+  }
+
+  async function waitForAttempts(count: number) {
+    while (upgradeAttempts < count) {
+      await delay(5);
+    }
+  }
+
+  it('reconnects when the peer drops the connection', async () => {
+    acceptConnections = true;
+    let dropped = false;
+    server.on('connection', (peer) => {
+      if (!dropped) {
+        dropped = true;
+        peer.terminate();
+      }
+    });
+
+    newSocket();
+
+    // A second attempt is the socket coming back after the drop
+    await waitForAttempts(2);
+  });
+
+  it('does not reconnect after close() when a reconnect is already pending', async () => {
+    const socket = newSocket();
+
+    // A second attempt means the first rejection scheduled a reconnect. Closing half a
+    // backoff later puts us in the state an application is in when it closes a socket that
+    // has been failing - the connection is down and the SDK is waiting to retry it.
+    await waitForAttempts(2);
+    await delay(reconnectDelay / 2);
+    await socket.close();
+
+    // The socket is closed, so the server should see no further connection attempts
+    const attemptsAtClose = upgradeAttempts;
+    await delay(reconnectDelay * 5);
+    assert.strictEqual(
+      upgradeAttempts - attemptsAtClose,
+      0,
+      'socket kept reconnecting after close()',
+    );
+  });
+
+  it('does not reconnect when an in-flight handler reconnects after close()', async () => {
+    const socket = newSocket();
+    await waitForAttempts(1);
+    await socket.close();
+
+    // unexpected-response reconnects from a deferred stream flush, so a rejection that was
+    // still being drained when we closed lands after the close has completed
+    const attemptsAtClose = upgradeAttempts;
+    internals(socket).reconnect('FireFly connect error [401]');
+    await delay(reconnectDelay * 5);
+    assert.strictEqual(upgradeAttempts - attemptsAtClose, 0, 'socket reconnected after close()');
+  });
+});