fix(sync): terminate sync websockets that stop answering pings

README.md

@@ -22,13 +22,15 @@ Or use the [`Dockerfile`](./Dockerfile) to build a Docker image.
 
 Server configuration is done using environment variables. The following environment variables are available:
 
-| Environment Variable  | Required | Description                                                          | Default Value    |
-| --------------------- | -------- | -------------------------------------------------------------------- | ---------------- |
-| `SERVER_BEARER_TOKEN` | Yes      | Token for authenticating API requests. Should be large random string |                  |
-| `PORT`                | No       | Port on which the server runs                                        | `8080`           |
-| `SERVER_NAME`         | No       | Friendly server name, seen by users when adding server               | `CoMapeo Server` |
-| `ALLOWED_PROJECTS`    | No       | Number of projects allowed to register with the server               | `1`              |
-| `STORAGE_DIR`         | No       | Path for storing app & project data                                  | `$CWD/data`      |
+| Environment Variable         | Required | Description                                                                                    | Default Value    |
+| ---------------------------- | -------- | ---------------------------------------------------------------------------------------------- | ---------------- |
+| `SERVER_BEARER_TOKEN`        | Yes      | Token for authenticating API requests. Should be large random string                           |                  |
+| `PORT`                       | No       | Port on which the server runs                                                                  | `8080`           |
+| `SERVER_NAME`                | No       | Friendly server name, seen by users when adding server                                         | `CoMapeo Server` |
+| `ALLOWED_PROJECTS`           | No       | Number of projects allowed to register with the server                                         | `1`              |
+| `STORAGE_DIR`                | No       | Path for storing app & project data                                                            | `$CWD/data`      |
+| `SYNC_HEARTBEAT_INTERVAL_MS` | No       | How often the server pings each sync websocket client                                          | `15000`          |
+| `SYNC_HEARTBEAT_TIMEOUT_MS`  | No       | Close a sync websocket that has sent no pong or data for this long (checked once per interval) | `30000`          |
 
 If you are using Nginx to act as a reverse proxy for your CoMapeo Cloud server, ensure your proxy headers are configured to support WebSockets.
 

src/app.js

@@ -29,6 +29,8 @@ async function comapeoServer(
     serverName,
     allowedHosts,
     allowedProjects,
+    syncHeartbeatIntervalMs,
+    syncHeartbeatTimeoutMs,
     ...comapeoPluginOpts
   },
 ) {
@@ -41,6 +43,8 @@ async function comapeoServer(
     serverBearerToken,
     serverName,
     allowedProjects,
+    syncHeartbeatIntervalMs,
+    syncHeartbeatTimeoutMs,
   })
 }
 

src/routes.js

@@ -45,13 +45,26 @@ const SUPPORTED_ATTACHMENT_TYPES = new Set(
  * @prop {string} serverBearerToken
  * @prop {string} serverName
  * @prop {undefined | number | string[]} [allowedProjects=1]
+ * @prop {number} [syncHeartbeatIntervalMs=15000]
+ * @prop {number} [syncHeartbeatTimeoutMs=30000]
  */
 
 /** @type {FastifyPluginAsync<RouteOptions, RawServerDefault, TypeBoxTypeProvider>} */
 export default async function routes(
   fastify,
-  { serverBearerToken, serverName, allowedProjects = 1 },
+  {
+    serverBearerToken,
+    serverName,
+    allowedProjects = 1,
+    syncHeartbeatIntervalMs = 15_000,
+    syncHeartbeatTimeoutMs = 30_000,
+  },
 ) {
+  if (syncHeartbeatTimeoutMs <= syncHeartbeatIntervalMs) {
+    throw new Error(
+      'syncHeartbeatTimeoutMs must be greater than syncHeartbeatIntervalMs',
+    )
+  }
   /** @type {Set<string> | number} */
   const allowedProjectsSetOrNumber = Array.isArray(allowedProjects)
     ? new Set(allowedProjects)
@@ -277,6 +290,11 @@ export default async function routes(
       // The preValidation hook ensures that the project exists
       const project = await this.comapeo.getProject(req.params.projectPublicId)
       const replicationStream = replicateProject(project, false)
+      startHeartbeat(socket, {
+        intervalMs: syncHeartbeatIntervalMs,
+        timeoutMs: syncHeartbeatTimeoutMs,
+        onTimeout: () => req.log.warn('sync websocket heartbeat timed out'),
+      })
       wsCoreReplicator(socket, replicationStream)
       project.$sync.start()
     },
@@ -695,6 +713,34 @@ function setAttachmentURL(obs, params) {
   }
 }
 
+/**
+ * A phone that drops off the network sends no FIN, so without this the socket
+ * stays open until TCP retransmits give up (minutes).
+ *
+ * @param {import('ws').WebSocket} socket
+ * @param {{ intervalMs: number, timeoutMs: number, onTimeout: () => void }} options
+ */
+function startHeartbeat(socket, { intervalMs, timeoutMs, onTimeout }) {
+  let lastSeen = Date.now()
+  const markAlive = () => {
+    lastSeen = Date.now()
+  }
+  socket.on('pong', markAlive)
+  socket.on('message', markAlive)
+  const timer = setInterval(() => {
+    if (socket.readyState !== socket.OPEN) {
+      clearInterval(timer)
+    } else if (Date.now() - lastSeen > timeoutMs) {
+      clearInterval(timer)
+      onTimeout()
+      socket.terminate()
+    } else {
+      socket.ping()
+    }
+  }, intervalMs)
+  socket.once('close', () => clearInterval(timer))
+}
+
 /**
  * @param {undefined | string} headerValue
  * @param {string} expectedBearerToken

src/server.js

@@ -33,6 +33,19 @@ const schema = Type.Object({
       description: 'number of projects allowed to join the server',
     }),
   ),
+  SYNC_HEARTBEAT_INTERVAL_MS: Type.Optional(
+    Type.Integer({
+      minimum: 1,
+      description: 'how often to ping clients on the sync websocket',
+    }),
+  ),
+  SYNC_HEARTBEAT_TIMEOUT_MS: Type.Optional(
+    Type.Integer({
+      minimum: 1,
+      description:
+        'close a sync websocket that has sent no pong or data for this long',
+    }),
+  ),
 })
 
 /** @typedef {import('@sinclair/typebox').Static<typeof schema>} Env */
@@ -85,6 +98,8 @@ fastify.register(comapeoServer, {
   serverName: config.SERVER_NAME,
   serverBearerToken: config.SERVER_BEARER_TOKEN,
   allowedProjects: config.ALLOWED_PROJECTS,
+  syncHeartbeatIntervalMs: config.SYNC_HEARTBEAT_INTERVAL_MS,
+  syncHeartbeatTimeoutMs: config.SYNC_HEARTBEAT_TIMEOUT_MS,
   rootKey,
   coreStorage,
   dbFolder,

test/sync-endpoint.js

@@ -1,6 +1,8 @@
 import { keyToPublicId as projectKeyToPublicId } from '@mapeo/crypto'
+import { WebSocket } from 'ws'
 
 import assert from 'node:assert/strict'
+import { once } from 'node:events'
 import test from 'node:test'
 
 import { createTestServer, randomAddProjectBody } from './test-helpers.js'
@@ -53,3 +55,30 @@ test('sync endpoint returns error with an invalid project public ID', async (t)
   assert.equal(response.statusCode, 400)
   assert.equal(response.json().error.code, 'BAD_REQUEST')
 })
+
+test('sync endpoint closes a client that stops answering pings', async (t) => {
+  const server = createTestServer(t, {
+    syncHeartbeatIntervalMs: 100,
+    syncHeartbeatTimeoutMs: 300,
+  })
+  const addProjectBody = randomAddProjectBody()
+  const projectPublicId = projectKeyToPublicId(
+    Buffer.from(addProjectBody.projectKey, 'hex'),
+  )
+  await server.inject({ method: 'PUT', url: '/projects', body: addProjectBody })
+
+  const serverAddress = await server.listen()
+  const url = serverAddress.replace(/^http/u, 'ws') + '/sync/' + projectPublicId
+
+  // @ts-expect-error autoPong (ws >= 8.15) is missing from @types/ws
+  const deadClient = new WebSocket(url, { autoPong: false })
+  const liveClient = new WebSocket(url)
+  t.after(() => {
+    deadClient.terminate()
+    liveClient.terminate()
+  })
+  await Promise.all([once(deadClient, 'open'), once(liveClient, 'open')])
+
+  await once(deadClient, 'close', { signal: AbortSignal.timeout(3000) })
+  assert.equal(liveClient.readyState, WebSocket.OPEN, 'live client stays open')
+})