diff --git a/README.md b/README.md index 88f8296..383891a 100644 --- a/README.md +++ b/README.md @@ -22,15 +22,16 @@ 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` | -| `SENTRY_DSN` | No | Sentry DSN. Reports crashes and attachments that could not be served | | -| `AUDIT_MAX_ATTACHMENTS` | No | Attachments checked per project by the attachment audit, newest first | `200` | +| 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` | +| `SENTRY_DSN` | No | Sentry DSN. Reports crashes, unserved attachments and audits that found missing ones | | +| `AUDIT_MIN_INTERVAL_MS` | No | Audit a project's attachments after each sync, at most this often. `0` disables | `3600000` | +| `AUDIT_MAX_ATTACHMENTS` | No | Attachments checked per project by the attachment audit, newest first | `200` | 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 80c7bf6..1244300 100644 --- a/src/app.js +++ b/src/app.js @@ -31,6 +31,7 @@ async function comapeoServer( allowedProjects, captureMessage, auditMaxAttachments, + onProjectAdded, ...comapeoPluginOpts }, ) { @@ -45,6 +46,7 @@ async function comapeoServer( allowedProjects, captureMessage, auditMaxAttachments, + onProjectAdded, }) } diff --git a/src/attachment-audit-scheduler.js b/src/attachment-audit-scheduler.js new file mode 100644 index 0000000..e084707 --- /dev/null +++ b/src/attachment-audit-scheduler.js @@ -0,0 +1,155 @@ +/** @import { MapeoProject } from '@comapeo/core' */ +/** @import { FastifyBaseLogger } from 'fastify' */ + +/** @typedef {ReturnType} SyncState */ +/** + * The part of a project's `$sync` the scheduler uses + * @typedef {object} WatchedSync + * @prop {() => SyncState} getState + * @prop {(event: 'sync-state', listener: (state: SyncState) => void) => unknown} on + * @prop {(event: 'sync-state', listener: (state: SyncState) => void) => unknown} off + */ +/** @typedef {{ $sync: WatchedSync }} WatchedProject */ + +/** + * @typedef {object} ProjectWatch + * @prop {string} projectId + * @prop {WatchedProject} project + * @prop {SyncState} previous + * @prop {Set} dataPeers connected devices that have had data sync enabled + * @prop {(state: SyncState) => void} onSyncState + * @prop {NodeJS.Timeout} [timer] + * @prop {number} lastStartedAt + * @prop {boolean} running + * @prop {boolean} dirty a sync ended while the audit was running + */ + +/** + * Runs an audit for a project after each sync with it ends, at most once per + * `minIntervalMs`. Audits that would come too soon are delayed to the earliest + * allowed time rather than dropped. Nothing runs at start. + */ +export class AttachmentAuditScheduler { + #audit + #minIntervalMs + #debounceMs + #log + /** @type {Map} */ + #watches = new Map() + #closed = false + + /** + * @param {object} options + * @param {(projectId: string) => Promise} options.audit must handle its own errors + * @param {number} options.minIntervalMs + * @param {number} [options.debounceMs] wait after a sync ends, so one device leaving does not audit while another is still mid-sync + * @param {Pick} options.log + */ + constructor({ audit, minIntervalMs, debounceMs = 5_000, log }) { + this.#audit = audit + this.#minIntervalMs = minIntervalMs + this.#debounceMs = debounceMs + this.#log = log + } + + /** + * @param {string} projectId + * @param {WatchedProject} project + */ + watch(projectId, project) { + if (this.#closed || this.#watches.has(projectId)) return + /** @type {ProjectWatch} */ + const watch = { + projectId, + project, + previous: project.$sync.getState(), + dataPeers: new Set(), + onSyncState: (state) => { + const ended = syncEnded(watch, state) + watch.previous = state + if (ended) this.#markDirty(watch) + }, + lastStartedAt: -Infinity, + running: false, + dirty: false, + } + this.#watches.set(projectId, watch) + project.$sync.on('sync-state', watch.onSyncState) + } + + close() { + this.#closed = true + for (const watch of this.#watches.values()) { + watch.project.$sync.off('sync-state', watch.onSyncState) + clearTimeout(watch.timer) + } + this.#watches.clear() + } + + /** @param {ProjectWatch} watch */ + #markDirty(watch) { + if (watch.running) { + watch.dirty = true + return + } + const now = Date.now() + const dueAt = Math.max( + now + this.#debounceMs, + watch.lastStartedAt + this.#minIntervalMs, + ) + clearTimeout(watch.timer) + watch.timer = setTimeout(() => this.#run(watch), dueAt - now) + watch.timer.unref() + this.#log.debug( + { projectId: watch.projectId, inMs: dueAt - now }, + 'attachment audit scheduled', + ) + } + + /** @param {ProjectWatch} watch */ + #run(watch) { + watch.running = true + watch.dirty = false + watch.lastStartedAt = Date.now() + this.#audit(watch.projectId) + .catch((err) => { + this.#log.error({ err, projectId: watch.projectId }, 'attachment audit') + }) + .then(() => { + watch.running = false + if (watch.dirty && !this.#closed) this.#markDirty(watch) + }) + } +} + +/** + * A sync session has ended when a device that was syncing data drops out of + * the remote sync state (its connection closed) or when the project stops + * wanting data blocks. Devices that only did presync, including the brief + * first connection while a device is being added to the project, are ignored. + * @param {Pick} watch + * @param {SyncState} current + */ +function syncEnded({ previous, dataPeers }, current) { + let ended = false + for (const deviceId of dataPeers) { + if (Object.hasOwn(current.remoteDeviceSyncState, deviceId)) continue + dataPeers.delete(deviceId) + ended = true + } + for (const [deviceId, device] of Object.entries( + current.remoteDeviceSyncState, + )) { + if (device.data.isSyncEnabled) dataPeers.add(deviceId) + } + return ended || (dataWanted(previous) > 0 && dataWanted(current) === 0) +} + +/** @param {SyncState} state */ +function dataWanted(state) { + let sum = 0 + for (const device of Object.values(state.remoteDeviceSyncState)) { + sum += device.data.wanted + } + return sum +} diff --git a/src/routes.js b/src/routes.js index 4ced80f..69beca1 100644 --- a/src/routes.js +++ b/src/routes.js @@ -69,6 +69,7 @@ const HOP_BY_HOP_HEADERS = new Set([ * @prop {undefined | number | string[]} [allowedProjects=1] * @prop {CaptureMessage} [captureMessage] receives one event per attachment that could not be served * @prop {number} [auditMaxAttachments] cap on attachments checked per audit + * @prop {(project: MapeoProject, projectPublicId: string) => void} [onProjectAdded] called once for each project PUT /projects adds */ /** @type {FastifyPluginAsync} */ @@ -80,6 +81,7 @@ export default async function routes( allowedProjects = 1, captureMessage, auditMaxAttachments, + onProjectAdded, }, ) { /** @type {Set | number} */ @@ -275,6 +277,7 @@ export default async function routes( const project = await this.comapeo.getProject(projectPublicId) project.$sync.start() + if (!alreadyHasThisProject) onProjectAdded?.(project, projectPublicId) return { data: { diff --git a/src/server.js b/src/server.js index ca6ee78..2ba3d0d 100644 --- a/src/server.js +++ b/src/server.js @@ -7,6 +7,8 @@ import fsPromises from 'node:fs/promises' import path from 'node:path' import comapeoServer from './app.js' +import { AttachmentAuditScheduler } from './attachment-audit-scheduler.js' +import { auditAttachments } from './attachments-audit.js' const DEFAULT_STORAGE = path.join(process.cwd(), 'data') const CORE_DIR_NAME = 'core' @@ -39,6 +41,12 @@ const schema = Type.Object({ 'Sentry DSN; error and attachment reporting is off when unset', }), ), + AUDIT_MIN_INTERVAL_MS: Type.Integer({ + minimum: 0, + default: 60 * 60 * 1000, + description: + 'audit a project after each sync, at most this often; 0 disables', + }), AUDIT_MAX_ATTACHMENTS: Type.Integer({ minimum: 1, default: 200, @@ -119,6 +127,16 @@ const fastify = createFastify({ trustProxy: true, maxParamLength: 256, }) +/** @type {AttachmentAuditScheduler | undefined} */ +let auditScheduler +if (Sentry && config.AUDIT_MIN_INTERVAL_MS > 0) { + auditScheduler = new AttachmentAuditScheduler({ + minIntervalMs: config.AUDIT_MIN_INTERVAL_MS, + audit: auditProject, + log: fastify.log, + }) +} + fastify.register(comapeoServer, { serverName: config.SERVER_NAME, serverBearerToken: config.SERVER_BEARER_TOKEN, @@ -130,6 +148,8 @@ fastify.register(comapeoServer, { clientMigrationsFolder, ...(Sentry && { captureMessage: Sentry.captureMessage }), auditMaxAttachments: config.AUDIT_MAX_ATTACHMENTS, + onProjectAdded: (project, projectId) => + auditScheduler?.watch(projectId, project), }) fastify.get('/healthcheck', async () => {}) @@ -141,9 +161,38 @@ try { process.exit(1) } +if (auditScheduler) { + for (const { projectId } of await fastify.comapeo.listProjects()) { + auditScheduler.watch(projectId, await fastify.comapeo.getProject(projectId)) + } +} + +/** @param {string} projectId */ +async function auditProject(projectId) { + try { + const project = await fastify.comapeo.getProject(projectId) + const { failures, devices, ...counts } = await auditAttachments(project, { + maxAttachments: config.AUDIT_MAX_ATTACHMENTS, + }) + fastify.log.info({ projectId, ...counts }, 'attachment audit') + if (failures.length === 0) return + Sentry?.captureMessage('attachment.audit', { + level: 'warning', + tags: { project: projectId }, + contexts: { audit: counts }, + extra: { failures: failures.slice(0, 100), devices }, + fingerprint: ['attachment.audit', projectId], + }) + } catch (err) { + fastify.log.error({ err, projectId }, 'attachment audit failed') + Sentry?.captureException(err, { tags: { project: projectId } }) + } +} + /** @param {NodeJS.Signals} signal*/ async function closeGracefully(signal) { console.log(`Received signal to terminate: ${signal}`) + auditScheduler?.close() await fastify.close() await Sentry?.close(2000) console.log('Gracefully closed fastify') diff --git a/test/add-project-endpoint.js b/test/add-project-endpoint.js index 35a8ac1..18a655f 100644 --- a/test/add-project-endpoint.js +++ b/test/add-project-endpoint.js @@ -221,3 +221,22 @@ test('adding the same project twice is idempotent', async (t) => { }) assert.equal(secondResponse.statusCode, 200) }) + +test('hands a newly added project to onProjectAdded', async (t) => { + /** @type {Array<{ projectPublicId: string, hasSync: boolean }>} */ + const added = [] + const server = createTestServer(t, { + onProjectAdded(project, projectPublicId) { + added.push({ projectPublicId, hasSync: Boolean(project.$sync) }) + }, + }) + const body = randomAddProjectBody() + const projectPublicId = projectKeyToPublicId( + Buffer.from(body.projectKey, 'hex'), + ) + + await server.inject({ method: 'PUT', url: '/projects', body }) + await server.inject({ method: 'PUT', url: '/projects', body }) + + assert.deepEqual(added, [{ projectPublicId, hasSync: true }]) +}) diff --git a/test/attachment-audit-scheduler.js b/test/attachment-audit-scheduler.js new file mode 100644 index 0000000..1847de9 --- /dev/null +++ b/test/attachment-audit-scheduler.js @@ -0,0 +1,205 @@ +import assert from 'node:assert/strict' +import { EventEmitter } from 'node:events' +import test from 'node:test' + +import { AttachmentAuditScheduler } from '../src/attachment-audit-scheduler.js' + +/** @import { TestContext } from 'node:test' */ +/** @typedef {import('../src/attachment-audit-scheduler.js').SyncState} SyncState */ + +const MINUTE = 60_000 +const HOUR = 60 * MINUTE +const DEBOUNCE = 5_000 + +/** + * @param {Record} devices data blocks wanted from each device + * @returns {SyncState} + */ +function syncState(devices) { + /** @type {SyncState['remoteDeviceSyncState']} */ + const remoteDeviceSyncState = {} + for (const [deviceId, { wanted, dataSync = true }] of Object.entries( + devices, + )) { + remoteDeviceSyncState[deviceId] = { + initial: { isSyncEnabled: true, want: 0, wanted: 0 }, + data: { isSyncEnabled: dataSync, want: 0, wanted }, + } + } + return { + initial: { isSyncEnabled: true }, + data: { isSyncEnabled: true }, + remoteDeviceSyncState, + } +} + +class FakeSync extends EventEmitter { + /** @param {SyncState} state */ + constructor(state) { + super() + this.state = state + } + getState() { + return this.state + } + /** @param {SyncState} state */ + update(state) { + this.state = state + this.emit('sync-state', state) + } +} + +/** + * @param {TestContext} t + * @param {{ minIntervalMs?: number }} [options] + */ +function setup(t, { minIntervalMs = HOUR } = {}) { + t.mock.timers.enable({ apis: ['setTimeout', 'Date'], now: 0 }) + /** @type {Array<{ projectId: string, at: number, done: () => void }>} */ + const audits = [] + const scheduler = new AttachmentAuditScheduler({ + minIntervalMs, + debounceMs: DEBOUNCE, + audit(projectId) { + return new Promise((resolve) => { + audits.push({ projectId, at: Date.now(), done: resolve }) + }) + }, + log: { debug() {}, error() {} }, + }) + t.after(() => scheduler.close()) + const sync = new FakeSync(syncState({ phone: { wanted: 3 } })) + scheduler.watch('project-a', { $sync: sync }) + return { scheduler, sync, audits } +} + +/** Let the audit's promise chain settle */ +const settle = () => + new Promise((resolve) => { + setImmediate(resolve) + }) + +test('a device leaving runs the audit after the debounce', (t) => { + const { sync, audits } = setup(t) + + sync.update(syncState({})) + t.mock.timers.tick(DEBOUNCE - 1) + assert.equal(audits.length, 0, 'not before the debounce') + t.mock.timers.tick(1) + assert.deepEqual( + audits.map(({ projectId, at }) => ({ projectId, at })), + [{ projectId: 'project-a', at: DEBOUNCE }], + ) +}) + +test('wanting nothing more from the data namespaces runs the audit', (t) => { + const { sync, audits } = setup(t) + + sync.update(syncState({ phone: { wanted: 1 } })) + sync.update(syncState({ phone: { wanted: 0 } })) + t.mock.timers.tick(DEBOUNCE) + assert.equal(audits.length, 1) +}) + +test('a device joining or sending more does not run the audit', (t) => { + const { sync, audits } = setup(t) + + sync.update(syncState({ phone: { wanted: 3 }, tablet: { wanted: 2 } })) + sync.update(syncState({ phone: { wanted: 8 }, tablet: { wanted: 2 } })) + sync.update(syncState({ phone: { wanted: 0 }, tablet: { wanted: 2 } })) + t.mock.timers.tick(HOUR) + assert.equal(audits.length, 0) +}) + +test('a device that only did presync leaving does not run the audit', (t) => { + const { sync, audits } = setup(t) + + sync.update( + syncState({ phone: { wanted: 3 }, tablet: { wanted: 0, dataSync: false } }), + ) + sync.update(syncState({ phone: { wanted: 3 } })) + t.mock.timers.tick(HOUR) + assert.equal(audits.length, 0) + + sync.update(syncState({ phone: { wanted: 3, dataSync: false } })) + sync.update(syncState({})) + t.mock.timers.tick(DEBOUNCE) + assert.equal(audits.length, 1, 'but one that synced data before does') +}) + +test('sync ends close together share one audit', (t) => { + const { sync, audits } = setup(t) + + sync.update(syncState({ phone: { wanted: 3 }, tablet: { wanted: 2 } })) + sync.update(syncState({ tablet: { wanted: 2 } })) + t.mock.timers.tick(DEBOUNCE - 1000) + sync.update(syncState({})) + t.mock.timers.tick(1000) + assert.equal(audits.length, 0, 'the second departure restarts the debounce') + t.mock.timers.tick(DEBOUNCE - 1000) + assert.equal(audits.length, 1) + assert.equal(audits[0]?.at, 2 * DEBOUNCE - 1000) +}) + +test('a second audit within the minimum interval waits for it', async (t) => { + const { sync, audits } = setup(t) + + sync.update(syncState({})) + t.mock.timers.tick(DEBOUNCE) + assert.equal(audits.length, 1) + audits[0]?.done() + await settle() + + t.mock.timers.tick(10 * MINUTE) + sync.update(syncState({ phone: { wanted: 3 } })) + sync.update(syncState({})) + t.mock.timers.tick(HOUR - 10 * MINUTE - 1) + assert.equal(audits.length, 1, 'not before an hour has passed') + t.mock.timers.tick(1) + assert.equal(audits.length, 2, 'not dropped either') + assert.equal(audits[1]?.at, DEBOUNCE + HOUR) +}) + +test('a sync ending while the audit runs queues one more run', async (t) => { + const { sync, audits } = setup(t) + + sync.update(syncState({})) + t.mock.timers.tick(DEBOUNCE) + assert.equal(audits.length, 1) + + sync.update(syncState({ phone: { wanted: 3 } })) + sync.update(syncState({})) + sync.update(syncState({ tablet: { wanted: 1 } })) + sync.update(syncState({})) + t.mock.timers.tick(HOUR) + assert.equal(audits.length, 1, 'nothing starts while one is running') + + audits[0]?.done() + await settle() + t.mock.timers.tick(DEBOUNCE) + assert.equal(audits.length, 2) + assert.equal(audits[1]?.at, DEBOUNCE + HOUR + DEBOUNCE) +}) + +test('close cancels a pending audit and stops listening', (t) => { + const { scheduler, sync, audits } = setup(t) + + sync.update(syncState({})) + scheduler.close() + t.mock.timers.tick(DEBOUNCE) + sync.update(syncState({ phone: { wanted: 1 } })) + sync.update(syncState({})) + t.mock.timers.tick(HOUR) + assert.equal(audits.length, 0) + assert.equal(sync.listenerCount('sync-state'), 0) +}) + +test('watching a project twice attaches one listener', (t) => { + const { scheduler, sync, audits } = setup(t) + + scheduler.watch('project-a', { $sync: sync }) + assert.equal(sync.listenerCount('sync-state'), 1) + sync.update(syncState({})) + t.mock.timers.tick(DEBOUNCE) + assert.equal(audits.length, 1) +})