fix(sync): handle replication stream errors on the sync websocket

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()
     },
   )

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,

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)
+})