Repository: mercurius-js/mercurius (js)

Issue:
Compatibility with `graphql-ws` `keepAlive`
I'm experimenting with switching from Apollo Express to Fastify and Mercurius.

I discovered something a bit weird. If I enable `keepAlive` in `graphql-ws`, it will send messages that look like `{"type": "ping"}` but it seems that something on the Mercurius side is not processing them correctly.

<img width="300" alt="Screenshot 2022-08-27 at 17 50 32" src="https://user-images.githubusercontent.com/697707/187035444-473f9634-fdc4-4376-b2cd-c332b751f4f8.png">

On the other hand, If I enable `keepAlive` in Mercurius, it sends messages like `{type: "ka"}`:
<img width="131" alt="Screenshot 2022-08-27 at 17 52 03" src="https://user-images.githubusercontent.com/697707/187035502-0d39e54a-0a4a-449d-aa03-666fcb7102f1.png">
 
`graphql-ws` completely crashes on these messages and breaks the subscription:
<img width="638" alt="Screenshot 2022-08-27 at 17 52 38" src="https://user-images.githubusercontent.com/697707/187035518-f65d114f-bb6c-48aa-9268-a038fa6dda93.png">
 
My understanding is that Mercurius uses `graphql-ws`, so what's the deal here? 🙂 

Pull request description:
feat: implement graphql-ws ping/pong
Fixes #851 

Here's a screenshot to confirm:

<img width="447" alt="Screenshot 2022-08-27 at 20 12 54" src="https://user-images.githubusercontent.com/697707/187040839-9181eaf4-3111-4c6a-a95e-768a06997b33.png">



Patch:
diff --git a/lib/subscription-client.js b/lib/subscription-client.js
index 294046c..bc14bbb 100644
--- a/lib/subscription-client.js
+++ b/lib/subscription-client.js
@@ -220,7 +220,11 @@ class SubscriptionClient {
         if (operation) {
           operation.handler(null)
           this.operations.delete(operationId)
-          this.sendMessage(operationId, this.protocolMessageTypes.GQL_ERROR, data.payload)
+          this.sendMessage(
+            operationId,
+            this.protocolMessageTypes.GQL_ERROR,
+            data.payload
+          )
         }
         break
       case this.protocolMessageTypes.GQL_COMPLETE:
@@ -238,11 +242,27 @@ class SubscriptionClient {
         }
         break
       case this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE:
+        if (this.socket) {
+          this.sendMessage(
+            operationId,
+            this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK
+          )
+        }
         break
-      /* istanbul ignore next */
       default:
+        // GQL_CONNECTION_KEEP_ALIVE_ACK is only defined in the graphql-ws protocol
+        /* istanbul ignore next */
+        if (
+          this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK &&
+          data.type === this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK
+        ) {
+          break
+        }
+
         /* istanbul ignore next */
-        throw new MER_ERR_GQL_SUBSCRIPTION_MESSAGE_INVALID(`Invalid message type "${data.type}"`)
+        throw new MER_ERR_GQL_SUBSCRIPTION_MESSAGE_INVALID(
+          `Invalid message type "${data.type}"`
+        )
     }
   }
 
diff --git a/lib/subscription-connection.js b/lib/subscription-connection.js
index f08b290..51ed9dc 100644
--- a/lib/subscription-connection.js
+++ b/lib/subscription-connection.js
@@ -74,11 +74,19 @@ module.exports = class SubscriptionConnection {
         break
       case this.protocolMessageTypes.GQL_START: {
         if (this.isReady) {
-          this.handleGQLStart(data).catch(e => {
-            this.sendMessage(this.protocolMessageTypes.GQL_ERROR, id, e.message)
+          this.handleGQLStart(data).catch((e) => {
+            this.sendMessage(
+              this.protocolMessageTypes.GQL_ERROR,
+              id,
+              e.message
+            )
           })
         } else {
-          this.sendMessage(this.protocolMessageTypes.GQL_CONNECTION_ERROR, undefined, { message: 'Connection has not been established yet.' })
+          this.sendMessage(
+            this.protocolMessageTypes.GQL_CONNECTION_ERROR,
+            undefined,
+            { message: 'Connection has not been established yet.' }
+          )
           return this.handleConnectionClose()
         }
         break
@@ -86,8 +94,30 @@ module.exports = class SubscriptionConnection {
       case this.protocolMessageTypes.GQL_STOP:
         await this.handleGQLStop(data)
         break
+      case this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE:
+        // GQL_CONNECTION_KEEP_ALIVE_ACK is only defined in the graphql-ws protocol
+        /* istanbul ignore next */
+        if (this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK) {
+          this.sendMessage(
+            this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK,
+            id
+          )
+        }
+        break
       default:
-        this.sendMessage(this.protocolMessageTypes.GQL_ERROR, id, 'Invalid payload type')
+        // GQL_CONNECTION_KEEP_ALIVE_ACK is only defined in the graphql-ws protocol
+        if (
+          this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK &&
+          type === this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK
+        ) {
+          break
+        }
+
+        this.sendMessage(
+          this.protocolMessageTypes.GQL_ERROR,
+          id,
+          'Invalid payload type'
+        )
     }
   }
 
diff --git a/lib/subscription-protocol.js b/lib/subscription-protocol.js
index 79654b9..94a5878 100644
--- a/lib/subscription-protocol.js
+++ b/lib/subscription-protocol.js
@@ -13,7 +13,8 @@ module.exports.getProtocolByName = function (name) {
         GQL_CONNECTION_INIT: 'connection_init', // Client -> Server
         GQL_CONNECTION_ACK: 'connection_ack', // Server -> Client
         GQL_CONNECTION_ERROR: 'connection_error', // Server -> Client
-        GQL_CONNECTION_KEEP_ALIVE: 'ka', // Server -> Client
+        GQL_CONNECTION_KEEP_ALIVE: 'ping', // Bidirectional
+        GQL_CONNECTION_KEEP_ALIVE_ACK: 'pong', // Bidirectional
         GQL_CONNECTION_TERMINATE: 'connection_terminate', // Client -> Server
         GQL_START: 'subscribe', // Client -> Server
         GQL_DATA: 'next', // Server -> Client
diff --git a/package.json b/package.json
index dbb6997..b8a9f47 100644
--- a/package.json
+++ b/package.json
@@ -32,7 +32,7 @@
     "@graphql-tools/merge": "^8.0.0",
     "@graphql-tools/schema": "^8.0.0",
     "@graphql-tools/utils": "^8.0.0",
-    "@sinonjs/fake-timers": "^9.0.0",
+    "@sinonjs/fake-timers": "^9.1.2",
     "@types/node": "^18.0.0",
     "@types/ws": "^8.2.0",
     "@types/isomorphic-form-data": "^2.0.0",
@@ -48,7 +48,7 @@
     "snazzy": "^9.0.0",
     "split2": "^4.0.0",
     "standard": "^17.0.0",
-    "tap": "^16.0.0",
+    "tap": "^16.3.0",
     "tsd": "^0.22.0",
     "typescript": "^4.3.5",
     "wait-on": "^6.0.0"


Test patch:
diff --git a/test/gateway/retry-failed-services.js b/test/gateway/retry-failed-services.js
index db168db..3eb13e8 100644
--- a/test/gateway/retry-failed-services.js
+++ b/test/gateway/retry-failed-services.js
@@ -119,11 +119,11 @@ test('gateway - retry mandatory failed services on startup', async (t) => {
     advanceTimeDelta: 100
   })
 
-  const service1 = await createTestService(5001, userService.schema, userService.resolvers)
+  const service1 = await createTestService(0, userService.schema, userService.resolvers)
 
   let service2 = null
   setTimeout(async () => {
-    service2 = await createTestService(5002, postService.schema, postService.resolvers)
+    service2 = await createTestService(5113, postService.schema, postService.resolvers)
   }, 5000)
 
   const app = Fastify()
@@ -140,12 +140,12 @@ test('gateway - retry mandatory failed services on startup', async (t) => {
       services: [
         {
           name: 'user',
-          url: 'http://localhost:5001/graphql',
+          url: `http://localhost:${service1.server.address().port}/graphql`,
           mandatory: false
         },
         {
           name: 'post',
-          url: 'http://localhost:5002/graphql',
+          url: 'http://localhost:5113/graphql',
           mandatory: true
         }
       ]
@@ -221,11 +221,11 @@ test('gateway - should not call onGatewayReplaceSchemaHandler if the hook is not
     advanceTimeDelta: 100
   })
 
-  const service1 = await createTestService(5001, userService.schema, userService.resolvers)
+  const service1 = await createTestService(0, userService.schema, userService.resolvers)
 
   let service2 = null
   setTimeout(async () => {
-    service2 = await createTestService(5002, postService.schema, postService.resolvers)
+    service2 = await createTestService(5111, postService.schema, postService.resolvers)
   }, 5000)
 
   const app = Fastify()
@@ -242,12 +242,12 @@ test('gateway - should not call onGatewayReplaceSchemaHandler if the hook is not
       services: [
         {
           name: 'user',
-          url: 'http://localhost:5001/graphql',
+          url: `http://localhost:${service1.server.address().port}/graphql`,
           mandatory: false
         },
         {
           name: 'post',
-          url: 'http://localhost:5002/graphql',
+          url: 'http://localhost:5111/graphql',
           mandatory: true
         }
       ],
@@ -319,7 +319,7 @@ test('gateway - dont retry non-mandatory failed services on startup', async (t)
     advanceTimeDelta: 100
   })
 
-  const service1 = await createTestService(5001, userService.schema, userService.resolvers)
+  const service1 = await createTestService(0, userService.schema, userService.resolvers)
 
   const app = Fastify()
   t.teardown(async () => {
@@ -334,12 +334,12 @@ test('gateway - dont retry non-mandatory failed services on startup', async (t)
       services: [
         {
           name: 'user',
-          url: 'http://localhost:5001/graphql',
+          url: `http://localhost:${service1.server.address().port}/graphql`,
           mandatory: false
         },
         {
           name: 'post',
-          url: 'http://localhost:5002/graphql',
+          url: 'http://localhost:5112/graphql',
           mandatory: false
         }
       ],
@@ -406,11 +406,11 @@ test('gateway - should log error if retry throws', async (t) => {
     advanceTimeDelta: 100
   })
 
-  const service1 = await createTestService(5001, userService.schema, userService.resolvers)
+  const service1 = await createTestService(0, userService.schema, userService.resolvers)
 
   let service2 = null
   setTimeout(async () => {
-    service2 = await createTestService(5002, postService.schema, postService.resolvers)
+    service2 = await createTestService(5113, postService.schema, postService.resolvers)
   }, 2000)
 
   const app = Fastify()
@@ -436,12 +436,12 @@ test('gateway - should log error if retry throws', async (t) => {
       services: [
         {
           name: 'user',
-          url: 'http://localhost:5001/graphql',
+          url: `http://localhost:${service1.server.address().port}/graphql`,
           mandatory: false
         },
         {
           name: 'post',
-          url: 'http://localhost:5002/graphql',
+          url: 'http://localhost:5113/graphql',
           mandatory: true
         }
       ],
@@ -456,8 +456,8 @@ test('gateway - should log error if retry throws', async (t) => {
 
   await app.ready()
 
-  for (let i = 0; i < 10; i++) {
-    await clock.tickAsync(1000)
+  for (let i = 0; i < 100; i++) {
+    await clock.tickAsync(100)
   }
 })
 
@@ -497,7 +497,7 @@ test('gateway - stop retrying after no. of retries exceeded', async (t) => {
         },
         {
           name: 'post',
-          url: 'http://localhost:5002/graphql',
+          url: 'http://localhost:5114/graphql',
           mandatory: true
         }
       ],
diff --git a/test/subscription-client.js b/test/subscription-client.js
index 8071851..0717fb9 100644
--- a/test/subscription-client.js
+++ b/test/subscription-client.js
@@ -379,7 +379,14 @@ test('subscription client sends GQL_CONNECTION_KEEP_ALIVE when the keep alive op
         ws.send(JSON.stringify({ id: '1', type: 'connection_ack' }))
       } else if (data.type === 'start') {
         ws.send(JSON.stringify({ id: '2', type: 'complete' }))
-      } else if (data.type === 'ka') {
+      } else if (data.type === 'ping') {
+        // this is a client sent ping, we reply with our pong
+        ws.send(JSON.stringify({ id: '3', type: 'pong' }))
+
+        // send a ping to the client so that it replies with its own pong
+        // pings and pongs are bidirectional
+        ws.send(JSON.stringify({ id: '4', type: 'ping' }))
+      } else if (data.type === 'pong') {
         client.close()
         server.close()
         t.end()
@@ -414,7 +421,7 @@ test('subscription client not throwing error on GQL_CONNECTION_KEEP_ALIVE type p
     ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
 
     clock.setInterval(() => {
-      ws.send(JSON.stringify({ type: 'ka' }))
+      ws.send(JSON.stringify({ type: 'ping' }))
     }, 200)
   })
 
diff --git a/test/subscription-connection.js b/test/subscription-connection.js
index 323543d..29bbd20 100644
--- a/test/subscription-connection.js
+++ b/test/subscription-connection.js
@@ -341,6 +341,69 @@ test('subscription connection handles when GQL_START is called before GQL_INIT',
   }))
 })
 
+test('subscription connection replies to GQL_CONNECTION_KEEP_ALIVE message with GQL_CONNECTION_KEEP_ALIVE_ACK', async (t) => {
+  t.plan(1)
+
+  const sc = new SubscriptionConnection(
+    {
+      on () {},
+      close () {},
+      send (message) {
+        t.equal(
+          JSON.stringify({
+            type: 'pong',
+            id: 1
+          }),
+          message
+        )
+      },
+      protocol: GRAPHQL_TRANSPORT_WS
+    },
+    {}
+  )
+
+  await sc.handleMessage(
+    JSON.stringify({
+      id: 1,
+      type: 'ping',
+      payload: {}
+    })
+  )
+})
+
+test('subscription connection does not error if client sends GQL_CONNECTION_KEEP_ALIVE_ACK', async (t) => {
+  t.plan(1)
+
+  const sc = new SubscriptionConnection(
+    {
+      on () {},
+      close () {},
+      send (message) {
+        t.fail()
+      },
+      protocol: GRAPHQL_TRANSPORT_WS
+    },
+    {}
+  )
+
+  await sc.handleMessage(
+    JSON.stringify({
+      id: 1,
+      type: 'pong',
+      payload: {}
+    })
+  )
+
+  await sc.handleMessage(
+    JSON.stringify({
+      id: 1,
+      type: 'complete'
+    })
+  )
+
+  t.equal(sc.subscriptionContexts.size, 0)
+})
+
 test('subscription connection extends context with onConnect return value', async (t) => {
   t.plan(3)
 
diff --git a/test/subscription-protocol.js b/test/subscription-protocol.js
index 4bc8c26..4a3b6a8 100644
--- a/test/subscription-protocol.js
+++ b/test/subscription-protocol.js
@@ -20,7 +20,8 @@ test('getProtocolByName returns correct protocol message types', t => {
     GQL_CONNECTION_INIT: 'connection_init',
     GQL_CONNECTION_ACK: 'connection_ack',
     GQL_CONNECTION_ERROR: 'connection_error',
-    GQL_CONNECTION_KEEP_ALIVE: 'ka',
+    GQL_CONNECTION_KEEP_ALIVE: 'ping',
+    GQL_CONNECTION_KEEP_ALIVE_ACK: 'pong',
     GQL_CONNECTION_TERMINATE: 'connection_terminate',
     GQL_START: 'subscribe',
     GQL_DATA: 'next',