diff --git a/apps/server/src/collaboration/collaboration.gateway.ts b/apps/server/src/collaboration/collaboration.gateway.ts index ced36083..75d05c22 100644 --- a/apps/server/src/collaboration/collaboration.gateway.ts +++ b/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(); + + 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, diff --git a/apps/server/src/collaboration/extensions/authentication.extension.ts b/apps/server/src/collaboration/extensions/authentication.extension.ts index addce7af..79b4246c 100644 --- a/apps/server/src/collaboration/extensions/authentication.extension.ts +++ b/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); diff --git a/apps/server/src/collaboration/extensions/persistence.extension.ts b/apps/server/src/collaboration/extensions/persistence.extension.ts index 35310cf4..10afc82e 100644 --- a/apps/server/src/collaboration/extensions/persistence.extension.ts +++ b/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 { 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; } /** diff --git a/apps/server/src/integrations/metrics/metrics.constants.ts b/apps/server/src/integrations/metrics/metrics.constants.ts index b802e2f0..f829d58a 100644 --- a/apps/server/src/integrations/metrics/metrics.constants.ts +++ b/apps/server/src/integrations/metrics/metrics.constants.ts @@ -2,6 +2,11 @@ * Perf-metrics contract (#355). These names/labels are FIXED by the already * deployed scrape+dashboard infra (VictoriaMetrics scraping docmost:9464, * Grafana dashboards, alerts). Do NOT rename them. + * + * #402 extends the #355 table with the collab-lifecycle + MCP-tool families + * (grouped below). These are the fixed contract from #402; same "do not rename" + * rule applies. Server-side families land in Pass 1; the MCP-tool histogram is + * fed by the MCP callback in a later pass but its NAME is fixed here now. */ export const METRIC_HTTP_REQUEST_DURATION = 'http_request_duration_seconds'; export const METRIC_DB_QUERY_DURATION = 'db_query_duration_seconds'; @@ -9,6 +14,17 @@ export const METRIC_BULLMQ_QUEUE_DEPTH = 'bullmq_queue_depth'; export const METRIC_BULLMQ_JOB_DURATION = 'bullmq_job_duration_seconds'; export const METRIC_COLLAB_STORE_DURATION = 'collab_store_duration_seconds'; +// #402 additions — collaboration lifecycle + MCP tool timing. +export const METRIC_COLLAB_LOAD_DURATION = 'collab_doc_load_duration_seconds'; +export const METRIC_COLLAB_DOCS_OPEN = 'collab_docs_open'; +export const METRIC_COLLAB_DOC_LOADS_TOTAL = 'collab_doc_loads_total'; +export const METRIC_COLLAB_DOC_UNLOADS_TOTAL = 'collab_doc_unloads_total'; +export const METRIC_COLLAB_CONNECT_DURATION = 'collab_connect_duration_seconds'; +export const METRIC_COLLAB_CONNECT_TIMEOUTS_TOTAL = + 'collab_connect_timeouts_total'; +export const METRIC_COLLAB_AUTH_DURATION = 'collab_auth_duration_seconds'; +export const METRIC_MCP_TOOL_DURATION = 'mcp_tool_duration_seconds'; + // Histogram buckets (seconds). Chosen to give useful p50/p95/p99 resolution // for typical web/DB latencies without exploding series cardinality. export const HTTP_BUCKETS = [ @@ -23,6 +39,12 @@ export const COLLAB_BUCKETS = [ export const JOB_BUCKETS = [ 0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, ]; +// #402 — MCP tool-call latency. Same shape as COLLAB_BUCKETS but stretched to +// 10s at the top: an MCP tool round-trip (LLM-driven doc ops) can be slower +// than a single collab store, so keep resolution out to 10s. +export const MCP_TOOL_BUCKETS = [ + 0.005, 0.025, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, +]; /** * Extract the first SQL token (select/insert/update/delete/...) from a query, @@ -58,6 +80,30 @@ export function firstSqlToken(sql: string | undefined): string { return KNOWN_SQL_TOKENS.has(token) ? token : 'other'; } +/** + * #402 — bucket a document byte size into ONE of four fixed labels for the + * collab load/store histograms' `size_bucket` label. Using the raw byte count + * would be a continuous, unbounded label; four coarse buckets keep series + * cardinality bounded (each of collab_doc_load / collab_store gets ×4 series). + * + * The SAME function is shared by both the load and store paths so their buckets + * can never drift apart. Non-finite / negative sizes collapse to the smallest + * bucket ('lt64k') as a safe default (they can't be legitimately huge and we + * must never throw here — this runs on every store/load when metrics are on). + */ +// Module-const thresholds, built once (not per observe). +const SIZE_THRESHOLDS = { lt64k: 65536, lt256k: 262144, lt1m: 1048576 } as const; + +export function sizeBucket( + bytes: number | undefined | null, +): 'lt64k' | 'lt256k' | 'lt1m' | 'ge1m' { + if (bytes == null || !Number.isFinite(bytes) || bytes < 0) return 'lt64k'; + if (bytes < SIZE_THRESHOLDS.lt64k) return 'lt64k'; + if (bytes < SIZE_THRESHOLDS.lt256k) return 'lt256k'; + if (bytes < SIZE_THRESHOLDS.lt1m) return 'lt1m'; + return 'ge1m'; +} + /** * Whether an HTTP response must be EXCLUDED from http_request_duration_seconds. * diff --git a/apps/server/src/integrations/metrics/metrics.registry.ts b/apps/server/src/integrations/metrics/metrics.registry.ts index d62be165..37991254 100644 --- a/apps/server/src/integrations/metrics/metrics.registry.ts +++ b/apps/server/src/integrations/metrics/metrics.registry.ts @@ -1,5 +1,6 @@ import { collectDefaultMetrics, + Counter, Histogram, Gauge, Registry, @@ -9,11 +10,21 @@ import { DB_BUCKETS, HTTP_BUCKETS, JOB_BUCKETS, + MCP_TOOL_BUCKETS, METRIC_BULLMQ_JOB_DURATION, METRIC_BULLMQ_QUEUE_DEPTH, + METRIC_COLLAB_AUTH_DURATION, + METRIC_COLLAB_CONNECT_DURATION, + METRIC_COLLAB_CONNECT_TIMEOUTS_TOTAL, + METRIC_COLLAB_DOC_LOADS_TOTAL, + METRIC_COLLAB_DOC_UNLOADS_TOTAL, + METRIC_COLLAB_DOCS_OPEN, + METRIC_COLLAB_LOAD_DURATION, METRIC_COLLAB_STORE_DURATION, METRIC_DB_QUERY_DURATION, METRIC_HTTP_REQUEST_DURATION, + METRIC_MCP_TOOL_DURATION, + sizeBucket, } from './metrics.constants'; /** @@ -39,7 +50,24 @@ let httpHist: Histogram<'method' | 'route' | 'status'> | null = null; let dbHist: Histogram<'op'> | null = null; let queueDepthGauge: Gauge<'queue'> | null = null; let jobHist: Histogram<'queue'> | null = null; -let collabHist: Histogram | null = null; +// #402 — collab store now carries a size_bucket label (see sizeBucket()). +let collabHist: Histogram<'size_bucket'> | null = null; +// #402 collab-lifecycle + MCP instruments. +let collabLoadHist: Histogram<'size_bucket'> | null = null; +let docsOpenGauge: Gauge | null = null; +let docLoadsCounter: Counter | null = null; +let docUnloadsCounter: Counter | null = null; +let connectTimeoutsCounter: Counter | null = null; +let collabConnectHist: Histogram | null = null; +let collabAuthHist: Histogram | null = null; +let mcpToolHist: Histogram<'tool'> | null = null; + +// #402 — read-on-scrape source for collab_docs_open. The gauge is NEVER +// inc/dec'd (that drifts under crashes/handoffs); instead its collect() callback +// pulls the authoritative live count from here on every scrape. Registered once, +// gated, by the collaboration gateway via registerDocsOpenSource(). Null until +// then → the collect() is a no-op. +let docsOpenSource: (() => number) | null = null; function init(): void { if (registry || !enabled) return; @@ -82,10 +110,71 @@ function init(): void { collabHist = new Histogram({ name: METRIC_COLLAB_STORE_DURATION, - help: 'Collaboration onStoreDocument duration in seconds', + help: 'Collaboration onStoreDocument duration in seconds, by document size bucket', + labelNames: ['size_bucket'], buckets: COLLAB_BUCKETS, registers: [registry], }); + + collabLoadHist = new Histogram({ + name: METRIC_COLLAB_LOAD_DURATION, + help: 'Collaboration onLoadDocument DB-load duration in seconds, by document size bucket', + labelNames: ['size_bucket'], + buckets: COLLAB_BUCKETS, + registers: [registry], + }); + + docsOpenGauge = new Gauge({ + name: METRIC_COLLAB_DOCS_OPEN, + help: 'Number of collaboration documents currently open in memory', + registers: [registry], + // Read-on-scrape: pull the live count from the registered source (the + // hocuspocus instance) so the value can never drift. No-op until a source + // is registered. + collect() { + if (docsOpenSource) this.set(docsOpenSource()); + }, + }); + + docLoadsCounter = new Counter({ + name: METRIC_COLLAB_DOC_LOADS_TOTAL, + help: 'Total collaboration documents loaded into memory', + registers: [registry], + }); + + docUnloadsCounter = new Counter({ + name: METRIC_COLLAB_DOC_UNLOADS_TOTAL, + help: 'Total collaboration documents unloaded from memory', + registers: [registry], + }); + + connectTimeoutsCounter = new Counter({ + name: METRIC_COLLAB_CONNECT_TIMEOUTS_TOTAL, + help: 'Total collaboration connection setup timeouts', + registers: [registry], + }); + + collabConnectHist = new Histogram({ + name: METRIC_COLLAB_CONNECT_DURATION, + help: 'Collaboration connection acceptance duration in seconds (onConnect→connected)', + buckets: COLLAB_BUCKETS, + registers: [registry], + }); + + collabAuthHist = new Histogram({ + name: METRIC_COLLAB_AUTH_DURATION, + help: 'Collaboration onAuthenticate duration in seconds', + buckets: COLLAB_BUCKETS, + registers: [registry], + }); + + mcpToolHist = new Histogram({ + name: METRIC_MCP_TOOL_DURATION, + help: 'MCP tool-call duration in seconds, by tool name', + labelNames: ['tool'], + buckets: MCP_TOOL_BUCKETS, + registers: [registry], + }); } // Runs once when this module is first imported. Safe to call again (idempotent). @@ -121,6 +210,46 @@ export function observeJobDuration(queue: string, seconds: number): void { jobHist?.observe({ queue }, seconds); } -export function observeCollabStore(seconds: number): void { - collabHist?.observe(seconds); +export function observeCollabStore(bytes: number, seconds: number): void { + collabHist?.observe({ size_bucket: sizeBucket(bytes) }, seconds); +} + +export function observeCollabLoad(bytes: number, seconds: number): void { + collabLoadHist?.observe({ size_bucket: sizeBucket(bytes) }, seconds); +} + +/** + * Register the live open-document count source for the collab_docs_open gauge. + * Called ONCE, gated by isMetricsEnabled(), by the collaboration gateway. The + * gauge reads this on every scrape (collect()); nothing inc/dec's it. + */ +export function registerDocsOpenSource(fn: () => number): void { + docsOpenSource = fn; +} + +export function incDocLoad(): void { + docLoadsCounter?.inc(); +} + +export function incDocUnload(): void { + docUnloadsCounter?.inc(); +} + +export function incConnectTimeout(): void { + connectTimeoutsCounter?.inc(); +} + +export function observeCollabConnect(seconds: number): void { + collabConnectHist?.observe(seconds); +} + +export function observeCollabAuth(seconds: number): void { + collabAuthHist?.observe(seconds); +} + +export function observeMcpTool(tool: string, seconds: number): void { + // `tool` MUST be a bounded, registration-derived MCP tool name (the caller + // guarantees it comes from the registered-tool set) — never free-form input — + // so this label stays low-cardinality with no 'other' bucketing needed. + mcpToolHist?.observe({ tool }, seconds); } diff --git a/apps/server/src/integrations/metrics/metrics.spec.ts b/apps/server/src/integrations/metrics/metrics.spec.ts index 311ad43d..cd88134d 100644 --- a/apps/server/src/integrations/metrics/metrics.spec.ts +++ b/apps/server/src/integrations/metrics/metrics.spec.ts @@ -1,6 +1,23 @@ import { FastifyRequest } from 'fastify'; import { resolveRouteLabel } from './http-metrics.hook'; -import { firstSqlToken, isStreamingResponse } from './metrics.constants'; +import { + firstSqlToken, + isStreamingResponse, + sizeBucket, +} from './metrics.constants'; +import { + getMetricsRegistry, + incConnectTimeout, + incDocLoad, + incDocUnload, + isMetricsEnabled, + observeCollabAuth, + observeCollabConnect, + observeCollabLoad, + observeCollabStore, + observeMcpTool, + registerDocsOpenSource, +} from './metrics.registry'; describe('resolveRouteLabel (histogram route label)', () => { it('uses the ROUTE TEMPLATE, never the raw URL', () => { @@ -123,3 +140,67 @@ describe('firstSqlToken (bounded db label)', () => { expect(firstSqlToken('vacuum analyze')).toBe('other'); }); }); + +describe('sizeBucket (#402 bounded size label)', () => { + it('maps sizes to the four fixed buckets at their boundaries', () => { + // Boundaries are exclusive-upper: <65536 → lt64k, etc. + expect(sizeBucket(0)).toBe('lt64k'); + expect(sizeBucket(65535)).toBe('lt64k'); + expect(sizeBucket(65536)).toBe('lt256k'); + expect(sizeBucket(262143)).toBe('lt256k'); + expect(sizeBucket(262144)).toBe('lt1m'); + expect(sizeBucket(1048575)).toBe('lt1m'); + expect(sizeBucket(1048576)).toBe('ge1m'); + expect(sizeBucket(5_000_000)).toBe('ge1m'); + }); + + it('falls back to the smallest bucket for invalid sizes', () => { + // Non-finite / negative / nullish collapse to the smallest, safe default. + expect(sizeBucket(-1)).toBe('lt64k'); + expect(sizeBucket(NaN)).toBe('lt64k'); + expect(sizeBucket(Infinity)).toBe('lt64k'); + expect(sizeBucket(undefined)).toBe('lt64k'); + expect(sizeBucket(null)).toBe('lt64k'); + }); + + it('only ever returns one of the four fixed labels (bounded cardinality)', () => { + const labels = new Set( + [0, 65536, 262144, 1048576, -5, NaN].map((b) => sizeBucket(b)), + ); + for (const l of labels) { + expect(['lt64k', 'lt256k', 'lt1m', 'ge1m']).toContain(l); + } + }); +}); + +describe('metrics helpers are safe no-ops when METRICS_PORT is unset', () => { + // These specs run without METRICS_PORT, so the registry is never created and + // every observe/inc/set helper must be a cheap `?.` no-op that never throws. + beforeAll(() => { + // Guard the contract this suite depends on: if a CI env set METRICS_PORT, + // the assertions below would be meaningless, so fail loudly instead. + expect(process.env.METRICS_PORT).toBeUndefined(); + }); + + it('reports metrics disabled and a null registry', () => { + expect(isMetricsEnabled()).toBe(false); + expect(getMetricsRegistry()).toBeNull(); + }); + + it('does not throw from any #402 collab/MCP helper', () => { + expect(() => { + observeCollabLoad(123456, 0.01); + observeCollabStore(123456, 0.02); + observeCollabConnect(0.03); + observeCollabAuth(0.04); + observeMcpTool('some-tool', 0.05); + incDocLoad(); + incDocUnload(); + incConnectTimeout(); + // Registering a source must not create the gauge or invoke the fn. + registerDocsOpenSource(() => { + throw new Error('docsOpenSource must NOT be called when disabled'); + }); + }).not.toThrow(); + }); +});