expose load and receive sync message metrics

Problem: when analyzing a sync server under heavy load it is difficult
to identify what is causing the load. There are lots of questions we
might have such as "what documents are in memory?" or "which documents are
taking longest  in `receiveSyncMessage`?". 

Solution: expose a `"doc-metrics"` event on `Repo` which is fired
whenever a documnt is loaded or receives a sync message. This can be
wired up to Prometheus or other monitoring tools to allow operators to
answer these questions.

Add num ops and changes to events

packages/automerge-repo/src/Repo.ts

@@ -27,7 +27,10 @@ import { StorageAdapterInterface } from "./storage/StorageAdapterInterface.js"
 import { StorageSubsystem } from "./storage/StorageSubsystem.js"
 import { StorageId } from "./storage/types.js"
 import { CollectionSynchronizer } from "./synchronizer/CollectionSynchronizer.js"
-import { SyncStatePayload } from "./synchronizer/Synchronizer.js"
+import {
+  DocSyncMetrics,
+  SyncStatePayload,
+} from "./synchronizer/Synchronizer.js"
 import type { AnyDocumentId, DocumentId, PeerId } from "./types.js"
 
 function randomPeerId() {
@@ -104,6 +107,9 @@ export class Repo extends EventEmitter<RepoEvents> {
       networkSubsystem.send(message)
     })
 
+    // Forward metrics from doc synchronizers
+    this.synchronizer.on("metrics", event => this.emit("doc-metrics", event))
+
     if (this.#remoteHeadsGossipingEnabled) {
       this.synchronizer.on("open-doc", ({ peerId, documentId }) => {
         this.#remoteHeadsSubscriptions.subscribePeerToDoc(peerId, documentId)
@@ -113,6 +119,12 @@ export class Repo extends EventEmitter<RepoEvents> {
     // STORAGE
     // The storage subsystem has access to some form of persistence, and deals with save and loading documents.
     const storageSubsystem = storage ? new StorageSubsystem(storage) : undefined
+    if (storageSubsystem) {
+      storageSubsystem.on("document-loaded", event =>
+        this.emit("doc-metrics", { type: "doc-loaded", ...event })
+      )
+    }
+
     this.storageSubsystem = storageSubsystem
 
     // NETWORK
@@ -638,6 +650,7 @@ export interface RepoEvents {
   "delete-document": (arg: DeleteDocumentPayload) => void
   /** A document was marked as unavailable (we don't have it and none of our peers have it) */
   "unavailable-document": (arg: DeleteDocumentPayload) => void
+  "doc-metrics": (arg: DocMetrics) => void
 }
 
 export interface DocumentPayload {
@@ -647,3 +660,13 @@ export interface DocumentPayload {
 export interface DeleteDocumentPayload {
   documentId: DocumentId
 }
+
+export type DocMetrics =
+  | DocSyncMetrics
+  | {
+      type: "doc-loaded"
+      documentId: DocumentId
+      durationMillis: number
+      numOps: number
+      numChanges: number
+    }

packages/automerge-repo/src/storage/StorageSubsystem.ts

@@ -8,12 +8,22 @@ import { ChunkInfo, StorageKey, StorageId } from "./types.js"
 import { keyHash, headsHash } from "./keyHash.js"
 import { chunkTypeFromKey } from "./chunkTypeFromKey.js"
 import * as Uuid from "uuid"
+import { EventEmitter } from "eventemitter3"
+
+type StorageSubsystemEvents = {
+  "document-loaded": (arg: {
+    documentId: DocumentId
+    durationMillis: number
+    numOps: number
+    numChanges: number
+  }) => void
+}
 
 /**
  * The storage subsystem is responsible for saving and loading Automerge documents to and from
  * storage adapter. It also provides a generic key/value storage interface for other uses.
  */
-export class StorageSubsystem {
+export class StorageSubsystem extends EventEmitter<StorageSubsystemEvents> {
   /** The storage adapter to use for saving and loading documents */
   #storageAdapter: StorageAdapterInterface
 
@@ -29,6 +39,7 @@ export class StorageSubsystem {
   #log = debug(`automerge-repo:storage-subsystem`)
 
   constructor(storageAdapter: StorageAdapterInterface) {
+    super()
     this.#storageAdapter = storageAdapter
   }
 
@@ -130,7 +141,14 @@ export class StorageSubsystem {
     if (binary.length === 0) return null
 
     // Load into an Automerge document
+    const start = performance.now()
     const newDoc = A.loadIncremental(A.init(), binary) as A.Doc<T>
+    const end = performance.now()
+    this.emit("document-loaded", {
+      documentId,
+      durationMillis: end - start,
+      ...A.stats(newDoc),
+    })
 
     // Record the latest heads for the document
     this.#storedHeads.set(documentId, A.getHeads(newDoc))

packages/automerge-repo/src/synchronizer/CollectionSynchronizer.ts

@@ -58,6 +58,7 @@ export class CollectionSynchronizer extends Synchronizer {
     docSynchronizer.on("message", event => this.emit("message", event))
     docSynchronizer.on("open-doc", event => this.emit("open-doc", event))
     docSynchronizer.on("sync-state", event => this.emit("sync-state", event))
+    docSynchronizer.on("metrics", event => this.emit("metrics", event))
     return docSynchronizer
   }
 

packages/automerge-repo/src/synchronizer/DocSynchronizer.ts

@@ -351,11 +351,19 @@ export class DocSynchronizer extends Synchronizer {
 
     this.#withSyncState(message.senderId, syncState => {
       this.#handle.update(doc => {
+        const start = performance.now()
         const [newDoc, newSyncState] = A.receiveSyncMessage(
           doc,
           syncState,
           message.data
         )
+        const end = performance.now()
+        this.emit("metrics", {
+          type: "receive-sync-message",
+          documentId: this.#handle.documentId,
+          durationMillis: end - start,
+          ...A.stats(doc),
+        })
 
         this.#setSyncState(message.senderId, newSyncState)
 

packages/automerge-repo/src/synchronizer/Synchronizer.ts

@@ -15,6 +15,7 @@ export interface SynchronizerEvents {
   message: (payload: MessageContents) => void
   "sync-state": (payload: SyncStatePayload) => void
   "open-doc": (arg: OpenDocMessage) => void
+  metrics: (arg: DocSyncMetrics) => void
 }
 
 /** Notify the repo that the sync state has changed  */
@@ -23,3 +24,11 @@ export interface SyncStatePayload {
   documentId: DocumentId
   syncState: SyncState
 }
+
+export type DocSyncMetrics = {
+  type: "receive-sync-message"
+  documentId: DocumentId
+  durationMillis: number
+  numOps: number
+  numChanges: number
+}