diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index f3a29a6..9280337 100644 --- a/.github/workflows/release.yml +++ b/.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 diff --git a/.github/workflows/setlatest.yml b/.github/workflows/setlatest.yml index 90482b5..113ba1c 100644 --- a/.github/workflows/setlatest.yml +++ b/.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 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..a8bdf5d --- /dev/null +++ b/.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 diff --git a/lib/websocket.ts b/lib/websocket.ts index f3d79b0..0918f72 100644 --- a/lib/websocket.ts +++ b/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 { const closedPromise = new Promise((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 { diff --git a/test/websocket.ts b/test/websocket.ts new file mode 100644 index 0000000..15470a6 --- /dev/null +++ b/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((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((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()'); + }); +});