feat(metrics): #402 pass 1 — collab-cycle observability (load/lifecycle/connect/auth)

Extends the #355 perf-metrics registry (all behind the METRICS_PORT hard
gate — nullable instruments, no-op helpers when unset). New families:

- collab_doc_load_duration_seconds{size_bucket} — onLoadDocument timed on
  the real DB-load paths only (the already-loaded early-return is skipped).
- size_bucket added to collab_store_duration_seconds; storeDocument returns
  the ydoc byte length (reusing its single Y.encodeStateAsUpdate, no second
  encode) so onStoreDocument observes size without extra cost.
- collab_docs_open (gauge, read-on-scrape via collect() from
  hocuspocus.getDocumentsCount() — never inc/dec'd, so it can't drift) +
  collab_doc_loads_total / collab_doc_unloads_total (afterLoad/afterUnload).
- collab_connect_duration_seconds — onConnect->connected, correlated by a
  WeakMap keyed on the shared request (leak-free; the current client uses
  one socket per document).
- collab_auth_duration_seconds — wraps onAuthenticate (success and failure).
- sizeBucket() shared helper (lt64k|lt256k|lt1m|ge1m, 4 bounded values).

The mcp_tool_duration_seconds histogram + collab_connect_timeouts_total
counter helpers are registered here but wired in pass 2 (MCP callback).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/server/src/collaboration/collaboration.gateway.ts

@@ -1,4 +1,9 @@
-import { Hocuspocus } from '@hocuspocus/server';
+import {
+  connectedPayload,
+  Extension,
+  Hocuspocus,
+  onConnectPayload,
+} from '@hocuspocus/server';
 import { IncomingMessage } from 'http';
 import WebSocket from 'ws';
 import { AuthenticationExtension } from './extensions/authentication.extension';
@@ -25,6 +30,56 @@ import {
   CollaborationHandler,
   CollabEventHandlers,
 } from './collaboration.handler';
+import {
+  incDocLoad,
+  incDocUnload,
+  isMetricsEnabled,
+  observeCollabConnect,
+  registerDocsOpenSource,
+} from '../integrations/metrics/metrics.registry';
+
+/**
+ * #402 — collab lifecycle metrics as a lightweight hocuspocus extension.
+ *
+ * - afterLoadDocument / afterUnloadDocument (fire once PER DOCUMENT) drive the
+ *   doc load/unload counters.
+ * - collab_connect_duration_seconds: I time the onConnect→connected hook pair,
+ *   i.e. connection ACCEPTANCE (which includes the auth handshake). This is the
+ *   cleanest per-connection correlation hocuspocus exposes: both payloads carry
+ *   the SAME `request` IncomingMessage object, so a WeakMap keyed on it gives a
+ *   per-connection start with NO leak (the entry is GC'd with the request if a
+ *   connection is rejected in onAuthenticate and `connected` never fires).
+ *   I deliberately do NOT observe at afterLoadDocument: that hook fires per
+ *   DOCUMENT, not per connection, so a second client joining an already-open
+ *   doc would be missed. auth/load latencies are their own separate metrics.
+ *
+ * All helpers are no-ops when METRICS_PORT is unset; these hooks are per
+ * connect/load/unload (never per message), so there is no hot-path cost.
+ */
+class CollabMetricsExtension implements Extension {
+  // Keyed by the per-connection request object → connect start time (ms).
+  private readonly connectStarts = new WeakMap<object, number>();
+
+  async onConnect(data: onConnectPayload) {
+    this.connectStarts.set(data.request, performance.now());
+  }
+
+  async connected(data: connectedPayload) {
+    const start = this.connectStarts.get(data.request);
+    if (start !== undefined) {
+      observeCollabConnect((performance.now() - start) / 1000);
+      this.connectStarts.delete(data.request);
+    }
+  }
+
+  async afterLoadDocument() {
+    incDocLoad();
+  }
+
+  async afterUnloadDocument() {
+    incDocUnload();
+  }
+}
 
 @Injectable()
 export class CollaborationGateway {
@@ -58,9 +113,18 @@ export class CollaborationGateway {
         this.authenticationExtension,
         this.persistenceExtension,
         this.loggerExtension,
+        // #402 collab lifecycle + connect-duration metrics (no-op when off).
+        new CollabMetricsExtension(),
       ],
     });
 
+    // #402 — read-on-scrape source for collab_docs_open. Wire ONCE, gated, so
+    // nothing runs when metrics are disabled. The gauge's collect() pulls the
+    // live count from the hocuspocus instance on each scrape (no inc/dec drift).
+    if (isMetricsEnabled()) {
+      registerDocsOpenSource(() => this.hocuspocus.getDocumentsCount());
+    }
+
     if (this.withRedis) {
       this.redisClient = new RedisClient({
         host: this.redisConfig.host,

apps/server/src/collaboration/extensions/authentication.extension.ts

@@ -16,6 +16,7 @@ import { isUserDisabled } from '../../common/helpers';
 import { getPageId } from '../collaboration.util';
 import { JwtCollabPayload, JwtType } from '../../core/auth/dto/jwt-payload';
 import { resolveProvenance } from '../../common/decorators/auth-provenance.decorator';
+import { observeCollabAuth } from '../../integrations/metrics/metrics.registry';
 
 @Injectable()
 export class AuthenticationExtension implements Extension {
@@ -30,6 +31,18 @@ export class AuthenticationExtension implements Extension {
   ) {}
 
   async onAuthenticate(data: onAuthenticatePayload) {
+    // #402 — time the whole auth (verify + user/page/permission lookups) into
+    // collab_auth_duration_seconds. finally so failed auths are timed too.
+    // No-op when METRICS_PORT is unset. Behavior unchanged.
+    const start = performance.now();
+    try {
+      return await this.doAuthenticate(data);
+    } finally {
+      observeCollabAuth((performance.now() - start) / 1000);
+    }
+  }
+
+  private async doAuthenticate(data: onAuthenticatePayload) {
     const { documentName, token } = data;
     const pageId = getPageId(documentName);
 

apps/server/src/collaboration/extensions/persistence.extension.ts

@@ -41,7 +41,10 @@ import {
   HISTORY_INTERVAL,
 } from '../constants';
 import { TransclusionService } from '../../core/page/transclusion/transclusion.service';
-import { observeCollabStore } from '../../integrations/metrics/metrics.registry';
+import {
+  observeCollabLoad,
+  observeCollabStore,
+} from '../../integrations/metrics/metrics.registry';
 
 /**
  * #251 — wire format of the client→server stateless message that signals a
@@ -150,10 +153,14 @@ export class PersistenceExtension implements Extension {
     const { documentName, document } = data;
     const pageId = getPageId(documentName);
 
+    // #402 — the early return below (live doc already non-empty) does NOT touch
+    // the DB, so it is deliberately NOT timed. We only observe the real DB-load
+    // work, and only on each real-load return, tagged by the loaded doc size.
     if (!document.isEmpty('default')) {
       return;
     }
 
+    const startedAt = performance.now();
     const page = await this.pageRepo.findById(pageId, {
       includeContent: true,
       includeYdoc: true,
@@ -171,6 +178,7 @@ export class PersistenceExtension implements Extension {
       const dbState = new Uint8Array(page.ydoc);
 
       Y.applyUpdate(doc, dbState);
+      observeCollabLoad(dbState.length, (performance.now() - startedAt) / 1000);
       return doc;
     }
 
@@ -184,26 +192,42 @@ export class PersistenceExtension implements Extension {
         tiptapExtensions,
       );
 
-      Y.encodeStateAsUpdate(ydoc);
+      // Reuse this single encode for the size label (do NOT add a second one).
+      const encoded = Y.encodeStateAsUpdate(ydoc);
+      observeCollabLoad(
+        encoded.byteLength,
+        (performance.now() - startedAt) / 1000,
+      );
       return ydoc;
     }
 
     this.logger.debug(`creating fresh ydoc: ${pageId}`);
+    observeCollabLoad(0, (performance.now() - startedAt) / 1000);
     return new Y.Doc();
   }
 
   async onStoreDocument(data: onStoreDocumentPayload) {
     // #355 — time the full store (persist + post-store side effects) into
-    // collab_store_duration_seconds. No-op when METRICS_PORT is unset.
+    // collab_store_duration_seconds. #402 — also tag by document size bucket.
+    // No-op when METRICS_PORT is unset.
     const startedAt = performance.now();
+    // Default 0 so a throw before storeDocument returns still records a
+    // (smallest-bucket) observation rather than dropping the timing entirely.
+    let bytes = 0;
     try {
-      await this.storeDocument(data);
+      bytes = await this.storeDocument(data);
     } finally {
-      observeCollabStore((performance.now() - startedAt) / 1000);
+      observeCollabStore(bytes, (performance.now() - startedAt) / 1000);
     }
   }
 
-  private async storeDocument(data: onStoreDocumentPayload) {
+  /**
+   * Persist the document. Returns the serialized ydoc byte size (used as the
+   * store histogram's size_bucket). The single Y.encodeStateAsUpdate below is
+   * the ONLY serialization — its byteLength is reused for the label (no second
+   * encode).
+   */
+  private async storeDocument(data: onStoreDocumentPayload): Promise<number> {
     const { documentName, document, context } = data;
 
     const pageId = getPageId(documentName);
@@ -455,6 +479,11 @@ export class PersistenceExtension implements Extension {
 
       await this.enqueuePageHistory(page, lastUpdatedSource);
     }
+
+    // #402 — report the serialized size for the store histogram's size_bucket.
+    // ydocState is always computed above (there is no earlier no-write return in
+    // this method), so this reflects the doc that was serialized this store.
+    return ydocState.byteLength;
   }
 
   /**