diff --git a/src/routes.js b/src/routes.js index b04a28e..0b8d65d 100644 --- a/src/routes.js +++ b/src/routes.js @@ -277,7 +277,10 @@ 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) - wsCoreReplicator(socket, replicationStream) + wsCoreReplicator(socket, replicationStream).catch((err) => { + req.log.warn({ err }, 'sync websocket replication failed') + socket.terminate() + }) project.$sync.start() }, ) diff --git a/src/server.js b/src/server.js index cae373d..c7dddba 100644 --- a/src/server.js +++ b/src/server.js @@ -81,6 +81,9 @@ const fastify = createFastify({ trustProxy: true, maxParamLength: 256, }) +process.on('unhandledRejection', (err) => { + fastify.log.error({ err }, 'unhandled promise rejection') +}) fastify.register(comapeoServer, { serverName: config.SERVER_NAME, serverBearerToken: config.SERVER_BEARER_TOKEN, diff --git a/test/sync-endpoint.js b/test/sync-endpoint.js index c7cbd5f..f0b59b0 100644 --- a/test/sync-endpoint.js +++ b/test/sync-endpoint.js @@ -1,7 +1,10 @@ import { keyToPublicId as projectKeyToPublicId } from '@mapeo/crypto' import assert from 'node:assert/strict' +import { randomBytes } from 'node:crypto' +import { once } from 'node:events' import test from 'node:test' +import { setImmediate as nextLoop } 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('server survives a sync client sending garbage instead of a handshake', async (t) => { + const server = createTestServer(t) + const addProjectBody = randomAddProjectBody() + const projectPublicId = projectKeyToPublicId( + Buffer.from(addProjectBody.projectKey, 'hex'), + ) + await server.inject({ method: 'PUT', url: '/projects', body: addProjectBody }) + + /** @type {unknown[]} */ + const unhandledRejections = [] + /** @param {unknown} err */ + const onUnhandledRejection = (err) => unhandledRejections.push(err) + process.on('unhandledRejection', onUnhandledRejection) + t.after(() => process.off('unhandledRejection', onUnhandledRejection)) + + const ws = await server.injectWS('/sync/' + projectPublicId) + // 3-byte frame length prefix followed by a garbage handshake message. An + // unframed payload would just be buffered while the server waits for more. + const garbage = Buffer.concat([Buffer.from([64, 0, 0]), randomBytes(64)]) + // The handler drops frames that arrive before it has attached the pipeline. + const sending = setInterval(() => ws.send(garbage), 10) + await once(ws, 'close') + clearInterval(sending) + await nextLoop() + + assert.deepEqual(unhandledRejections, []) + const response = await server.inject({ method: 'GET', url: '/info' }) + assert.equal(response.statusCode, 200) +})