feat(metrics): наблюдаемость collab-цикла и MCP (#402, follow-up #355) #403

Merged
vvzvlad merged 3 commits from feat/402-collab-mcp-metrics into develop 2026-07-07 02:41:11 +03:00
12 changed files with 634 additions and 13 deletions
@@ -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,
@@ -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);
@@ -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;
}
/**
@@ -149,6 +149,15 @@ export type DocmostMcpConfig = (
has?: (uri: string) => boolean;
evict?: (uri: string) => void;
};
// Dependency-neutral metrics sink injected by McpService (mirror of the
// package's onMetric). The package emits generic (name, value, labels)
// samples; McpService maps them onto the prom-client registry. Undefined
// when metrics are disabled → the package no-ops.
onMetric?: (
name: string,
value: number,
labels?: Record<string, string>,
) => void;
};
export interface ResolvedMcpAuth {
@@ -30,6 +30,11 @@ import {
ResolvedMcpAuth,
} from './mcp-auth.helpers';
import { SandboxStore } from '../sandbox/sandbox.store';
import {
isMetricsEnabled,
observeMcpTool,
incConnectTimeout,
} from '../metrics/metrics.registry';
// Minimal shape of the embedded MCP HTTP handler exported by @docmost/mcp/http.
interface McpHttpHandler {
@@ -331,7 +336,31 @@ export class McpService implements OnModuleDestroy {
// can store blobs in the shared in-RAM store regardless of which
// credential variant resolved. The sink (put/has/evict + uri↔id
// mapping) is owned by SandboxStore.asSink().
return { ...resolved.config, sandbox: this.sandboxStore.asSink() };
// Route the package's dependency-neutral metric samples onto the
// prom-client registry. When metrics are disabled, onMetric is
// undefined → the package's tool-timer/timeout hooks are a
// negligible-overhead no-op: the registerTool wrapper still runs a
// performance.now() + async try/finally per tool call, but the
// `onMetric?.()` short-circuits so no label/object is built. (Cost
// is immaterial at LLM tool-call rate.) labels?.tool is guarded
// defensively (the tool wrapper always sets it).
return {
...resolved.config,
sandbox: this.sandboxStore.asSink(),
onMetric: isMetricsEnabled()
? (
name: string,
value: number,
labels?: Record<string, string>,
) => {
if (name === 'mcp_tool_duration_seconds') {
observeMcpTool(labels?.tool ?? 'other', value);
} else if (name === 'collab_connect_timeouts_total') {
incConnectTimeout();
}
}
: undefined,
};
},
{
identify: (req: IncomingMessage) => {
@@ -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.
*
@@ -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);
}
@@ -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();
});
});
+20
View File
@@ -137,6 +137,15 @@ export type DocmostMcpConfig = { apiUrl: string } & (
has?: (uri: string) => boolean;
evict?: (uri: string) => void;
};
// Dependency-neutral metrics sink. When present, the client emits generic
// (name, value, labels) samples; the HOST maps those names onto its own
// metrics registry (the package never depends on prom-client or the server).
// Absent in standalone/stdio mode → the client is a complete no-op here.
onMetric?: (
name: string,
value: number,
labels?: Record<string, string>,
) => void;
};
// Canonical UUID shape (versions 1–8, matching the `uuid` package's `validate`
@@ -173,6 +182,11 @@ export class DocmostClient {
// op's image blobs if the final doc put throws. Null when the sink omits them.
private sandboxHas: ((uri: string) => boolean) | null = null;
private sandboxEvict: ((uri: string) => void) | null = null;
// Optional dependency-neutral metrics sink (see DocmostMcpConfig.onMetric).
// Null on the legacy positional form and whenever the host omits it → no-op.
private onMetricFn:
| ((name: string, value: number, labels?: Record<string, string>) => void)
| null = null;
// In-flight login dedup: when the token expires, the 401 interceptor,
// ensureAuthenticated, getCollabTokenWithReauth and the two multipart retries
// can all call login() at once. Memoizing a single promise collapses that
@@ -222,6 +236,8 @@ export class DocmostClient {
this.sandboxHas = config.sandbox.has ?? null;
this.sandboxEvict = config.sandbox.evict ?? null;
}
// Legacy positional form carries no onMetric → null (complete no-op).
this.onMetricFn = config.onMetric ?? null;
this.client = axios.create({
baseURL: this.apiUrl,
// Default request timeout so a hung connection cannot wedge a per-page
@@ -447,6 +463,10 @@ export class DocmostClient {
};
connectTimer = setTimeout(() => {
// Only the actual 25s collab connect timeout fires here — the agent's
// collab connection to the server never became ready. This is the
// connect-vs-unload signal; the other finish() paths must NOT emit it.
this.onMetricFn?.("collab_connect_timeouts_total", 1);
finish(new Error("Connection timeout to collaboration server"));
}, CONNECT_TIMEOUT_MS);
+47
View File
@@ -64,6 +64,33 @@ const jsonContent = (data: any) => ({
* REST + the collaboration WebSocket using the provided service-account
* credentials and auto-re-authenticates.
*/
/**
* Wrap a tool handler so its wall-clock duration is reported through the host's
* dependency-neutral sink as `mcp_tool_duration_seconds` (labelled by tool
* name). Pure and side-effect-free apart from the optional `onMetric` call:
* - preserves the handler's exact return value (awaited);
* - observes in a `finally`, so it records on BOTH success and throw, then
* rethrows the original error unchanged (never swallowed);
* - with no `onMetric` (standalone/stdio) it is a transparent pass-through.
* Exported so the timing contract can be unit-tested without a live transport.
*/
export function timeToolHandler(
name: string,
handler: (...args: any[]) => any,
onMetric?: (name: string, value: number, labels?: Record<string, string>) => void,
): (...args: any[]) => Promise<any> {
return async (...handlerArgs: any[]) => {
const start = performance.now();
try {
return await handler(...handlerArgs);
} finally {
onMetric?.("mcp_tool_duration_seconds", (performance.now() - start) / 1000, {
tool: name,
});
}
};
}
export function createDocmostMcpServer(config: DocmostMcpConfig): McpServer {
// Pass the whole config union through: the client branches internally on
// credentials vs. getToken, so both the external /mcp (creds) and the
@@ -78,6 +105,26 @@ export function createDocmostMcpServer(config: DocmostMcpConfig): McpServer {
{ instructions: SERVER_INSTRUCTIONS },
);
// Single choke point for MCP tool timing. Both `registerShared` (below) and
// the inline `server.registerTool(...)` calls funnel through this one method,
// so monkeypatching it HERE — before any tool is registered and before
// `registerShared` captures a reference to it — times every tool with no
// per-tool boilerplate. The wrapped handler records wall-clock duration and,
// in a `finally`, feeds the host's dependency-neutral sink
// `config.onMetric("mcp_tool_duration_seconds", seconds, { tool })`. The tool
// name is the registration name (bounded cardinality). When no onMetric is
// provided (standalone/stdio) the wrapper is a pure pass-through: it still
// returns the original result and rethrows the original error unchanged.
const originalRegisterTool = server.registerTool.bind(server) as (
...args: any[]
) => any;
(server as any).registerTool = (...args: any[]) => {
const name = args[0] as string;
const handler = args[args.length - 1];
const timedHandler = timeToolHandler(name, handler, config.onMetric);
return originalRegisterTool(...args.slice(0, -1), timedHandler);
};
// Register a tool from the shared, zod-agnostic spec registry. The spec owns
// the canonical name + model-facing description + (optional) schema builder;
// only the execute body is supplied per call. buildShape is invoked with THIS
@@ -0,0 +1,87 @@
// #402 — INTEGRATION test that locks the registerTool monkeypatch installed by
// createDocmostMcpServer (src/index.ts). The sibling unit test
// (test/unit/tool-timing.test.mjs) only exercises the timeToolHandler helper in
// ISOLATION; it never constructs the server, so nothing there proves the factory
// actually (a) wraps every registered tool through that helper and (b) labels
// each sample with the tool's REGISTRATION name.
//
// Here we stand up a real McpServer via the factory, connect a real MCP Client
// over the SDK's in-memory transport, invoke one registered tool, and assert the
// host's onMetric sink received `("mcp_tool_duration_seconds", <number>, { tool
// })` with tool === the exact registration name. This locks both the "monkeypatch
// wraps tools" and "label = registration name" halves of the contract, so a
// mutation to args.slice(0, -1) / the handler-arg detection / the capture order
// is caught.
import { test } from "node:test";
import assert from "node:assert/strict";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
import { createDocmostMcpServer } from "../../build/index.js";
// The tool we drive. get_workspace has NO input schema, so protocol-level input
// validation cannot short-circuit before the handler runs — the wrapped handler
// is guaranteed to execute (and then fail on the unreachable backend, which is
// exactly what we want: the wrapper times in a finally on throw too).
const TOOL_NAME = "get_workspace";
test("the factory's registerTool monkeypatch times a live tool call and labels it with the registration name", async () => {
const calls = [];
const onMetric = (name, value, labels) => calls.push({ name, value, labels });
// Minimal valid credentials config. apiUrl points at a port that refuses
// connections immediately so the tool's backend call fails FAST (ECONNREFUSED)
// rather than hanging — the wrapper still emits the metric from its finally.
const server = createDocmostMcpServer({
apiUrl: "http://127.0.0.1:1",
email: "x@example.com",
password: "pw",
onMetric,
});
const [clientTransport, serverTransport] =
InMemoryTransport.createLinkedPair();
const client = new Client(
{ name: "test-client", version: "0.0.0" },
{ capabilities: {} },
);
await Promise.all([
server.connect(serverTransport),
client.connect(clientTransport),
]);
try {
// Invoke the tool. The backend is unreachable, so this either resolves with
// an error result (isError) or rejects — both are fine. What matters is that
// the handler ran through the timing wrapper, which fires onMetric either way.
try {
await client.callTool({ name: TOOL_NAME, arguments: {} });
} catch {
// Tolerate the expected backend failure surfacing as a thrown protocol error.
}
// The wrapper must have fed exactly the timing sample for THIS tool.
const timing = calls.filter(
(c) => c.name === "mcp_tool_duration_seconds",
);
assert.ok(
timing.length >= 1,
"onMetric must receive a mcp_tool_duration_seconds sample from the wrapped handler",
);
const sample = timing.find((c) => c.labels && c.labels.tool === TOOL_NAME);
assert.ok(
sample,
`a timing sample must be labelled with the registration name "${TOOL_NAME}"; ` +
`got labels: ${JSON.stringify(timing.map((c) => c.labels))}`,
);
assert.equal(typeof sample.value, "number");
assert.ok(sample.value >= 0, "duration must be non-negative seconds");
} finally {
await client.close();
await server.close();
}
});
@@ -0,0 +1,67 @@
import { test } from "node:test";
import assert from "node:assert/strict";
import { timeToolHandler } from "../../build/index.js";
// #402 — the registerTool wrapper times every tool through a single choke point
// and feeds the host's dependency-neutral onMetric sink. These assert the
// timing contract without a live transport (the factory monkeypatches
// server.registerTool with exactly this helper).
test("times a tool and preserves the handler's return value", async () => {
const calls = [];
const onMetric = (name, value, labels) => calls.push({ name, value, labels });
const handler = async (arg) => ({ ok: true, echo: arg });
const wrapped = timeToolHandler("get_page", handler, onMetric);
const result = await wrapped("hello");
// Return value passes through untouched.
assert.deepEqual(result, { ok: true, echo: "hello" });
// Exactly one sample, correct name/labels, numeric non-negative duration.
assert.equal(calls.length, 1);
assert.equal(calls[0].name, "mcp_tool_duration_seconds");
assert.deepEqual(calls[0].labels, { tool: "get_page" });
assert.equal(typeof calls[0].value, "number");
assert.ok(calls[0].value >= 0, "duration must be non-negative seconds");
});
test("standalone (no onMetric) is a transparent pass-through", async () => {
const handler = async (a, b) => a + b;
const wrapped = timeToolHandler("noop_tool", handler, undefined);
// No throw, result unchanged, and nothing to observe.
const result = await wrapped(2, 3);
assert.equal(result, 5);
});
test("still observes on throw, then rethrows the original error", async () => {
const calls = [];
const onMetric = (name, value, labels) => calls.push({ name, value, labels });
const boom = new Error("handler failed");
const handler = async () => {
throw boom;
};
const wrapped = timeToolHandler("boom_tool", handler, onMetric);
await assert.rejects(() => wrapped(), (err) => err === boom);
// Observed exactly once despite the throw.
assert.equal(calls.length, 1);
assert.equal(calls[0].name, "mcp_tool_duration_seconds");
assert.deepEqual(calls[0].labels, { tool: "boom_tool" });
});
test("standalone still rethrows the original error (no swallow, no observe)", async () => {
const boom = new Error("standalone failure");
const wrapped = timeToolHandler(
"boom_standalone",
async () => {
throw boom;
},
undefined,
);
await assert.rejects(() => wrapped(), (err) => err === boom);
});