From d3e947f7b189beee966622ede26f8f2994d55ab9 Mon Sep 17 00:00:00 2001 From: Gregor MacLennan Date: Thu, 5 Nov 2026 02:59:57 +0000 Subject: [PATCH 1/2] test(sync): close the sync websocket when a client stops answering pings --- test/sync-endpoint.js | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/test/sync-endpoint.js b/test/sync-endpoint.js index c7cbd5f..27cae5f 100644 --- a/test/sync-endpoint.js +++ b/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') +}) From bc979dfe5d3e1517701440cd24f5e5647afb397f Mon Sep 17 00:00:00 2001 From: Gregor MacLennan Date: Thu, 5 Nov 2026 03:01:34 +0000 Subject: [PATCH 2/2] fix(sync): terminate sync websockets that stop answering pings The server never detected a dead websocket: no ws ping/pong and no secret-stream timeout, so a phone that dropped off the network stayed connected until TCP retransmits gave up. Ping each /sync client on an interval and terminate it when no pong arrives within the timeout. Both are configurable via SYNC_HEARTBEAT_INTERVAL_MS and SYNC_HEARTBEAT_TIMEOUT_MS (defaults 15 s / 30 s). --- README.md | 16 +++++++++------- src/app.js | 4 ++++ src/routes.js | 48 +++++++++++++++++++++++++++++++++++++++++++++++- src/server.js | 15 +++++++++++++++ 4 files changed, 75 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 87e7904..cdccf0c 100644 --- a/README.md +++ b/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. diff --git a/src/app.js b/src/app.js index f3b0771..d50a849 100644 --- a/src/app.js +++ b/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, }) } diff --git a/src/routes.js b/src/routes.js index b04a28e..f91e086 100644 --- a/src/routes.js +++ b/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} */ 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 | 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 diff --git a/src/server.js b/src/server.js index cae373d..661d12e 100644 --- a/src/server.js +++ b/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} 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,