feat(attachments): audit attachments after each sync and report missing ones to Sentry

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.
 

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

src/attachment-audit-scheduler.js

@@ -0,0 +1,155 @@
+/** @import { MapeoProject } from '@comapeo/core' */
+/** @import { FastifyBaseLogger } from 'fastify' */
+
+/** @typedef {ReturnType<MapeoProject['$sync']['getState']>} 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<string>} 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<string, ProjectWatch>} */
+  #watches = new Map()
+  #closed = false
+
+  /**
+   * @param {object} options
+   * @param {(projectId: string) => Promise<void>} 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<FastifyBaseLogger, 'debug' | 'error'>} 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<ProjectWatch, 'previous' | 'dataPeers'>} 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
+}

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<RouteOptions, RawServerDefault, TypeBoxTypeProvider>} */
@@ -80,6 +81,7 @@ export default async function routes(
     allowedProjects = 1,
     captureMessage,
     auditMaxAttachments,
+    onProjectAdded,
   },
 ) {
   /** @type {Set<string> | 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: {

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')

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