diff --git a/README.md b/README.md index 87e7904..178599f 100644 --- a/README.md +++ b/README.md @@ -22,13 +22,14 @@ 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_INACTIVITY_TIMEOUT_MS` | No | Close a sync websocket that has received nothing for this long (ms) | `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..9ac130a 100644 --- a/src/app.js +++ b/src/app.js @@ -29,6 +29,7 @@ async function comapeoServer( serverName, allowedHosts, allowedProjects, + syncInactivityTimeoutMs, ...comapeoPluginOpts }, ) { @@ -41,6 +42,7 @@ async function comapeoServer( serverBearerToken, serverName, allowedProjects, + syncInactivityTimeoutMs, }) } diff --git a/src/routes.js b/src/routes.js index b04a28e..0a8e43e 100644 --- a/src/routes.js +++ b/src/routes.js @@ -45,12 +45,18 @@ const SUPPORTED_ATTACHMENT_TYPES = new Set( * @prop {string} serverBearerToken * @prop {string} serverName * @prop {undefined | number | string[]} [allowedProjects=1] + * @prop {number} [syncInactivityTimeoutMs=30000] */ /** @type {FastifyPluginAsync} */ export default async function routes( fastify, - { serverBearerToken, serverName, allowedProjects = 1 }, + { + serverBearerToken, + serverName, + allowedProjects = 1, + syncInactivityTimeoutMs = 30_000, + }, ) { /** @type {Set | number} */ const allowedProjectsSetOrNumber = Array.isArray(allowedProjects) @@ -277,6 +283,12 @@ 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) + terminateWhenSilent(socket, syncInactivityTimeoutMs, () => + req.log.warn( + 'sync websocket received nothing for %d ms, terminating', + syncInactivityTimeoutMs, + ), + ) wsCoreReplicator(socket, replicationStream) project.$sync.start() }, @@ -695,6 +707,23 @@ function setAttachmentURL(obs, params) { } } +/** + * No ping needed: hypercore sets a 5 s keepalive on every replication stream, + * so a live client is never silent this long. + * + * @param {import('ws').WebSocket} socket + * @param {number} timeoutMs + * @param {() => void} onTimeout + */ +function terminateWhenSilent(socket, timeoutMs, onTimeout) { + const timer = setTimeout(() => { + onTimeout() + socket.terminate() + }, timeoutMs).unref() + socket.on('message', () => timer.refresh()) + socket.once('close', () => clearTimeout(timer)) +} + /** * @param {undefined | string} headerValue * @param {string} expectedBearerToken diff --git a/src/server.js b/src/server.js index cae373d..9d71f9d 100644 --- a/src/server.js +++ b/src/server.js @@ -33,6 +33,13 @@ const schema = Type.Object({ description: 'number of projects allowed to join the server', }), ), + SYNC_INACTIVITY_TIMEOUT_MS: Type.Optional( + Type.Integer({ + minimum: 1, + description: + 'close a sync websocket that has received nothing for this long', + }), + ), }) /** @typedef {import('@sinclair/typebox').Static} Env */ @@ -85,6 +92,7 @@ fastify.register(comapeoServer, { serverName: config.SERVER_NAME, serverBearerToken: config.SERVER_BEARER_TOKEN, allowedProjects: config.ALLOWED_PROJECTS, + syncInactivityTimeoutMs: config.SYNC_INACTIVITY_TIMEOUT_MS, rootKey, coreStorage, dbFolder, diff --git a/test/sync-endpoint.js b/test/sync-endpoint.js index c7cbd5f..e455c8f 100644 --- a/test/sync-endpoint.js +++ b/test/sync-endpoint.js @@ -1,7 +1,10 @@ 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 { setTimeout as delay } from 'node:timers/promises' import { createTestServer, randomAddProjectBody } from './test-helpers.js' @@ -53,3 +56,33 @@ 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 goes silent', async (t) => { + const timeoutMs = 300 + const server = createTestServer(t, { syncInactivityTimeoutMs: timeoutMs }) + 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 + + const silentClient = new WebSocket(url) + const chattyClient = new WebSocket(url) + t.after(() => { + silentClient.terminate() + chattyClient.terminate() + }) + await Promise.all([once(silentClient, 'open'), once(chattyClient, 'open')]) + + // Empty frames count as inbound messages but never reach the replication + // stream, so the noise handshake does not fail on junk + const sender = setInterval(() => chattyClient.send(Buffer.alloc(0)), 50) + t.after(() => clearInterval(sender)) + + await once(silentClient, 'close', { signal: AbortSignal.timeout(3000) }) + await delay(timeoutMs * 2) + assert.equal(chattyClient.readyState, WebSocket.OPEN) +})