Merge branch 'main' into kloet/unit-test-template

Author: dskloetc

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@chainlink/external-adapter-framework",
-  "version": "2.14.2",
+  "version": "2.14.3",
   "main": "dist/index.js",
   "license": "MIT",
   "repository": "git://github.com/smartcontractkit/ea-framework-js.git",

src/transports/websocket.ts

@@ -426,7 +426,6 @@ export class WebSocketTransport<
       timeSinceLastActivity > 0 &&
       timeSinceLastActivity > context.adapterSettings.WS_SUBSCRIPTION_UNRESPONSIVE_TTL
 
-    let connectionClosed = this.connectionClosed()
     logger.trace(`WS conn staleness info: 
       now: ${now} |
       timeSinceLastMessage: ${timeSinceLastMessage} |
@@ -460,7 +459,7 @@ export class WebSocketTransport<
     const urlChanged = this.currentUrl !== urlFromConfig
 
     // Check if we should close the current connection
-    if (!connectionClosed && (urlChanged || connectionUnresponsive)) {
+    if (!this.connectionClosed() && (urlChanged || connectionUnresponsive)) {
       if (urlChanged) {
         logger.info('Websocket URL has changed, closing connection to reconnect...')
         censorLogs(() =>
@@ -484,7 +483,7 @@ export class WebSocketTransport<
         await sleep(1000 - timeSinceConnectionOpened)
       }
       this.wsConnection?.close(1000)
-      connectionClosed = true
+      this.wsConnection = undefined
 
       if (subscriptions.desired.length) {
         // Clear subscription metrics for all active subscriptions
@@ -500,7 +499,7 @@ export class WebSocketTransport<
     }
 
     // Check if we need to open a new connection
-    if (connectionClosed && subscriptions.desired.length) {
+    if (this.connectionClosed() && subscriptions.desired.length) {
       logger.debug('No established connection and new subscriptions available, connecting to WS')
       const options =
         this.config.options && (await this.config.options(context, subscriptions.desired))
@@ -518,14 +517,13 @@ export class WebSocketTransport<
       this.wsConnection = await this.establishWsConnection(context, urlFromConfig, options)
 
       // Now that we successfully opened the connection, we can reset the variables
-      connectionClosed = false
       this.connectionOpenedAt = Date.now()
     }
 
     // Send messages only if the connection is open
     // Otherwise we could encounter the case where we just closed the connection because there's no desired ones,
     // but without this check we'd attempt to send out all the unsubscribe messages
-    if (!connectionClosed) {
+    if (!this.connectionClosed()) {
       logger.debug('Connection is open, sending subs/unsubs if there are any')
       const builders = this.config.builders
       if (builders) {

test/transports/websocket.test.ts

@@ -13,7 +13,12 @@ import {
   type WebSocketTransportConfig,
 } from '../../src/transports'
 import { SingleNumberResultResponse, sleep } from '../../src/util'
-import { mockWebSocketProvider, runAllUntilTime, TestAdapter } from '../../src/util/testing-utils'
+import {
+  mockWebSocketProvider,
+  runAllUntil,
+  runAllUntilTime,
+  TestAdapter,
+} from '../../src/util/testing-utils'
 import { InputParameters } from '../../src/validation'
 
 export const test = untypedTest as TestFn<{
@@ -487,6 +492,78 @@ test.serial('reconnects if provider stops sending expected messages', async (t)
   await t.context.clock.runToLastAsync()
 })
 
+test.serial('should not inflate failover metric while connection is closed', async (t) => {
+  const base = 'ETH'
+  const quote = 'DOGE'
+  const WS_SUBSCRIPTION_UNRESPONSIVE_TTL = 120_000
+  process.env['METRICS_ENABLED'] = 'true'
+  eaMetrics.clear()
+
+  // Mock WS
+  mockWebSocketProvider(WebSocketClassProvider)
+  const mockWsServer = new Server(ENDPOINT_URL, { mock: false })
+  let connectionCounter = 0
+
+  mockWsServer.on('connection', (socket) => {
+    let counter = 0
+    const parseMessage = () => {
+      if (counter++ === 0) {
+        socket.send(JSON.stringify({ error: '' }))
+      }
+    }
+    connectionCounter++
+    socket.on('message', parseMessage)
+  })
+
+  const adapter = createAdapter({
+    WS_SUBSCRIPTION_TTL: 30_000,
+    WS_SUBSCRIPTION_UNRESPONSIVE_TTL,
+  })
+
+  const testAdapter = await TestAdapter.startWithMockedCache(adapter, t.context)
+
+  const error = await testAdapter.request({
+    base,
+    quote,
+  })
+  t.is(error.statusCode, 504)
+
+  await runAllUntil(t.context.clock, () => connectionCounter >= 1)
+  t.is(connectionCounter, 1)
+
+  await runAllUntilTime(t.context.clock, WS_SUBSCRIPTION_UNRESPONSIVE_TTL + 1000)
+
+  // The connection was closed because it was unresponsive but it wasn't
+  // reopened because all the subscriptions expired.
+  t.is(connectionCounter, 1)
+  ;(await testAdapter.getMetrics()).assert(t, {
+    name: 'ws_connection_failover_count',
+    labels: {
+      transport_name: 'default_single_transport',
+      url: ENDPOINT_URL,
+    },
+    expectedValue: 1,
+  })
+
+  // Even after waiting a long time, the connection counter and failover metric
+  // should remain the same.
+  await runAllUntilTime(t.context.clock, 2 * WS_SUBSCRIPTION_UNRESPONSIVE_TTL)
+  t.is(connectionCounter, 1)
+  ;(await testAdapter.getMetrics()).assert(t, {
+    name: 'ws_connection_failover_count',
+    labels: {
+      transport_name: 'default_single_transport',
+      url: ENDPOINT_URL,
+    },
+    expectedValue: 1,
+  })
+
+  testAdapter.api.close()
+  mockWsServer.close()
+  await t.context.clock.runToLastAsync()
+  process.env['METRICS_ENABLED'] = 'false'
+})
+
 test.serial('resubscribes after reconnection if server closes connection', async (t) => {
   const base = 'ETH'
   const quote = 'DOGE'