diff --git a/packages/backend/src/common/sentry-sampler.spec.ts b/packages/backend/src/common/sentry-sampler.spec.ts new file mode 100644 index 00000000..a240fd09 --- /dev/null +++ b/packages/backend/src/common/sentry-sampler.spec.ts @@ -0,0 +1,46 @@ +import { sampleRateFor } from './sentry-sampler'; + +const rates = { base: 0.05, mcp: 0.01 }; + +describe('sampleRateFor', () => { + it('samples MCP calls at the MCP rate', () => { + expect( + sampleRateFor( + { name: 'POST', attributes: { 'http.method': 'POST', 'http.target': '/mcp/abc123?x=1' } }, + rates, + ), + ).toBe(0.01); + expect( + sampleRateFor( + { name: 'POST', normalizedRequest: { method: 'POST', url: 'https://cloud.example.com/mcp/abc' } }, + rates, + ), + ).toBe(0.01); + expect(sampleRateFor({ name: 'POST /mcp/abc' }, rates)).toBe(0.01); + }); + + it('samples other requests at the base rate', () => { + expect( + sampleRateFor( + { name: 'GET', attributes: { 'http.request.method': 'GET', 'url.path': '/api/connectors' } }, + rates, + ), + ).toBe(0.05); + // Not the MCP route, only a lookalike prefix. + expect( + sampleRateFor({ name: 'GET', attributes: { 'http.method': 'GET', 'http.target': '/mcpx' } }, rates), + ).toBe(0.05); + }); + + it('never samples health probes', () => { + expect( + sampleRateFor({ name: 'GET', attributes: { 'http.method': 'GET', 'http.target': '/health' } }, rates), + ).toBe(0); + }); + + it('never samples root spans that are not requests', () => { + expect(sampleRateFor({ name: 'prisma:client:operation' }, rates)).toBe(0); + expect(sampleRateFor({ name: 'pg-pool.connect', attributes: { 'db.system': 'postgresql' } }, rates)).toBe(0); + expect(sampleRateFor({ name: 'SELECT "public"."kg_edges"."id" FROM "public"."kg_edges"' }, rates)).toBe(0); + }); +}); diff --git a/packages/backend/src/common/sentry-sampler.ts b/packages/backend/src/common/sentry-sampler.ts new file mode 100644 index 00000000..c5d86f4f --- /dev/null +++ b/packages/backend/src/common/sentry-sampler.ts @@ -0,0 +1,74 @@ +/** + * Which traces reach Sentry. + * + * A flat 5 % rate sent about 700,000 spans a day in September 2026, far past + * what a Sentry plan includes. Most of them were not requests at all: + * + * - root spans with no request behind them: Prisma and pg-pool spans from + * the knowledge-graph cron and other background work, each one sampled as + * a transaction of its own; + * - MCP tool calls, the bulk of the traffic, each carrying dozens of database + * spans. + * + * So: only incoming HTTP requests are traced; MCP calls at a lower rate than + * the rest (they are many and alike); health probes never. The decision of + * an incoming trace is not inherited: MCP clients send their own sampled + * traceparent, and that must not decide our volume. + */ + +export interface SamplerRates { + /** Every HTTP request other than MCP and health probes. */ + base: number; + /** POST /mcp/:serverId and the other MCP transport routes. */ + mcp: number; +} + +interface SamplingInput { + name: string; + attributes?: Record; + normalizedRequest?: { url?: string; method?: string }; +} + +function str(v: unknown): string | undefined { + return typeof v === 'string' && v.length > 0 ? v : undefined; +} + +/** The request path, from whichever of the places it can be at sampling time. */ +function requestPath(ctx: SamplingInput): string | undefined { + const a = ctx.attributes ?? {}; + const raw = + str(a['url.path']) ?? + str(a['http.target']) ?? + str(a['http.route']) ?? + str(a['url.full']) ?? + str(a['http.url']) ?? + str(ctx.normalizedRequest?.url); + if (!raw) { + // "POST /mcp/abc" — the span name once the route is known. + const m = /^[A-Z]+ (\/\S*)/.exec(ctx.name); + return m?.[1]; + } + try { + return raw.startsWith('/') ? raw.split('?')[0] : new URL(raw).pathname; + } catch { + return raw.split('?')[0]; + } +} + +function isHttpRequest(ctx: SamplingInput): boolean { + const a = ctx.attributes ?? {}; + return Boolean( + str(a['http.request.method']) ?? + str(a['http.method']) ?? + str(ctx.normalizedRequest?.method) ?? + (/^(GET|POST|PUT|PATCH|DELETE|HEAD|OPTIONS) /.test(ctx.name) ? 'yes' : undefined), + ); +} + +export function sampleRateFor(ctx: SamplingInput, rates: SamplerRates): number { + if (!isHttpRequest(ctx)) return 0; + const path = requestPath(ctx) ?? ''; + if (/^\/health(\/|$)/.test(path)) return 0; + if (/^\/mcp(\/|$)/.test(path)) return rates.mcp; + return rates.base; +} diff --git a/packages/backend/src/instrument.ts b/packages/backend/src/instrument.ts index db28ce29..0e00296f 100644 --- a/packages/backend/src/instrument.ts +++ b/packages/backend/src/instrument.ts @@ -12,6 +12,7 @@ */ import * as Sentry from '@sentry/nestjs'; import { scrubBreadcrumb, scrubEvent } from './common/sentry-scrub'; +import { sampleRateFor } from './common/sentry-sampler'; const dsn = process.env.SENTRY_DSN; @@ -30,7 +31,14 @@ if (dsn) { release: process.env.SENTRY_RELEASE || process.env.npm_package_version, // Tracing is opt-in on top of error reporting because it adds overhead. - tracesSampleRate: sample(process.env.SENTRY_TRACES_SAMPLE_RATE, 0.0), + // SENTRY_TRACES_SAMPLE_RATE applies to HTTP requests; MCP calls take + // SENTRY_MCP_TRACES_SAMPLE_RATE (default: a fifth of it); background work + // and health probes are not traced. See ./common/sentry-sampler.ts. + tracesSampler: (() => { + const base = sample(process.env.SENTRY_TRACES_SAMPLE_RATE, 0.0); + const rates = { base, mcp: sample(process.env.SENTRY_MCP_TRACES_SAMPLE_RATE, base / 5) }; + return (ctx: Parameters[0]) => sampleRateFor(ctx, rates); + })(), profilesSampleRate: sample(process.env.SENTRY_PROFILES_SAMPLE_RATE, 0.0), sendDefaultPii: false, diff --git a/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts b/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts index 181745a7..1bebfeb2 100644 --- a/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts +++ b/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts @@ -225,4 +225,40 @@ describe('KgObservationalService.ingestOrganization', () => { const kinds = prisma.kgEdge.create.mock.calls.map((c: any[]) => c[0].data.kind).sort(); expect(kinds).toEqual(['produces_consumes', 'same_identity']); }); + + it('writes each edge once per pass, however many values link the same nodes', async () => { + const occ = (hash: string) => [ + { valueHash: hash, connectorId: 'c2', entity: 'order', field: 'customer_id', direction: 'output' }, + { valueHash: hash, connectorId: 'c1', entity: 'customer', field: 'id', direction: 'input' }, + ]; + const { svc, prisma } = make({ + pages: [[row('r1', 'c2', 1, 'erp_get_order')]], + payloads: { r1: { input: {}, output: { order: { customer_id: 'CUST-12345' } } } }, + valueSeen: ['h1', 'h2', 'h3', 'h4'].flatMap(occ), + }); + // produces_consumes already seen 3 times; same_identity is new. + prisma.kgEdge.findUnique.mockImplementation(async (args: any) => + args.where.organizationId_sourceNodeId_targetNodeId_kind.kind === 'produces_consumes' + ? { id: 'e1', observations: 3, isManual: false, status: 'active' } + : null, + ); + await svc.ingestOrganization(ORG); + + expect(prisma.kgEdge.findUnique).toHaveBeenCalledTimes(2); + expect(prisma.kgEdge.update).toHaveBeenCalledTimes(1); + const upd = prisma.kgEdge.update.mock.calls[0][0].data; + // Four values, four observations, as four sequential bumps would leave it. + expect(upd.observations).toBe(7); + expect(upd.confidence).toBeCloseTo(0.55 + 0.05 * 6); + expect(upd.matchKey).toBe('id'); + + expect(prisma.kgEdge.create).toHaveBeenCalledTimes(1); + const created = prisma.kgEdge.create.mock.calls[0][0].data; + expect(created.kind).toBe('same_identity'); + expect(created.observations).toBe(4); + expect(created.confidence).toBeCloseTo(0.2 + 0.05 * 3); + expect(created.status).toBe('suggested'); + + expect(prisma.kgEdge.updateMany).toHaveBeenCalledTimes(1); + }); }); diff --git a/packages/backend/src/knowledge-graph/kg-observational.service.ts b/packages/backend/src/knowledge-graph/kg-observational.service.ts index dcf020d6..1be722e8 100644 --- a/packages/backend/src/knowledge-graph/kg-observational.service.ts +++ b/packages/backend/src/knowledge-graph/kg-observational.service.ts @@ -1,5 +1,6 @@ import { randomUUID } from 'node:crypto'; import { Injectable, Logger } from '@nestjs/common'; +import * as Sentry from '@sentry/nestjs'; import { PrismaService } from '../common/prisma.service'; import { extractEntity } from './static/entity-extraction'; import { fkCandidate } from './static/fk-inference'; @@ -46,6 +47,18 @@ interface PageRow { tool: { name: string } | null; } +/** Observations of one edge within a correlate pass, written once at the end. */ +interface EdgeBump { + src: string; + tgt: string; + kind: string; + times: number; + matchKey?: string; + base: number; + cap: number; + status: string; +} + interface ValueRow { organizationId: string; connectorId: string; @@ -91,7 +104,10 @@ export class KgObservationalService { } this.inFlight.add(organizationId); try { - return await this.ingest(organizationId); + // Not traced. The ingest is scheduled from inside an MCP tool call, so + // its hundreds of queries became spans of that call's trace: most of + // the cloud's Sentry volume in September 2026. Errors are still reported. + return await Sentry.suppressTracing(() => this.ingest(organizationId)); } finally { this.inFlight.delete(organizationId); } @@ -280,7 +296,11 @@ export class KgObservationalService { const src = await this.nodeId(refCache, organizationId, b.connectorId, b.from); const tgt = await this.nodeId(refCache, organizationId, b.connectorId, b.to); if (src && tgt && src !== tgt) { - await this.bumpEdge(organizationId, src, tgt, 'references', { + await this.bumpEdge(organizationId, { + src, + tgt, + kind: 'references', + times: 1, matchKey: b.field, base: 0.6, cap: 0.9, @@ -306,7 +326,11 @@ export class KgObservationalService { const nb = await this.nodeId(coCache, organizationId, cb, eb); if (!na || !nb || na === nb) continue; const [src, tgt] = na < nb ? [na, nb] : [nb, na]; - await this.bumpEdge(organizationId, src, tgt, 'related', { + await this.bumpEdge(organizationId, { + src, + tgt, + kind: 'related', + times: 1, base: 0.3, cap: 0.7, status: 'suggested', @@ -464,6 +488,23 @@ export class KgObservationalService { } const nodeCache = new Map(); // `${connectorId}::${entity}` -> nodeId + // Many values link the same two nodes, so one pass used to write the same + // edge again for every one of them: a read and an update per value, most + // of the database work behind an MCP call. The bumps are counted here and + // each distinct edge is written once, with the same end state. + const bumps = new Map(); + const confirmedRefs = new Map(); + const bump = (src: string, tgt: string, kind: string, b: Omit) => { + const key = `${src}|${tgt}|${kind}`; + const prev = bumps.get(key); + if (prev) { + prev.times++; + // Sequential updates left the last matchKey that was set. + if (b.matchKey !== undefined) prev.matchKey = b.matchKey; + } else { + bumps.set(key, { src, tgt, kind, times: 1, ...b }); + } + }; let edgeCount = 0; for (const occ of byHash.values()) { @@ -477,17 +518,14 @@ export class KgObservationalService { const src = await this.nodeId(nodeCache, organizationId, p.connectorId, p.entity); const tgt = await this.nodeId(nodeCache, organizationId, c.connectorId, c.entity); if (!src || !tgt || src === tgt) continue; - await this.bumpEdge(organizationId, src, tgt, 'produces_consumes', { + bump(src, tgt, 'produces_consumes', { matchKey: c.field, base: 0.55, cap: 0.95, status: 'active', }); // The data confirms any static FK guess between the same nodes. - await this.prisma.kgEdge.updateMany({ - where: { organizationId, sourceNodeId: src, targetNodeId: tgt, kind: 'references', source: 'STATIC' }, - data: { source: 'OBSERVED', confidence: 0.8, lastSeenAt: new Date() }, - }); + confirmedRefs.set(`${src}|${tgt}`, [src, tgt]); edgeCount++; } } @@ -505,7 +543,7 @@ export class KgObservationalService { const nb = await this.nodeId(nodeCache, organizationId, b.connectorId, b.entity); if (!na || !nb || na === nb) continue; const [src, tgt] = na < nb ? [na, nb] : [nb, na]; - await this.bumpEdge(organizationId, src, tgt, 'same_identity', { + bump(src, tgt, 'same_identity', { matchKey: a.field === b.field ? a.field : undefined, base: 0.2, cap: 0.6, @@ -517,6 +555,16 @@ export class KgObservationalService { } } } + + for (const b of bumps.values()) { + await this.bumpEdge(organizationId, b); + } + for (const [src, tgt] of confirmedRefs.values()) { + await this.prisma.kgEdge.updateMany({ + where: { organizationId, sourceNodeId: src, targetNodeId: tgt, kind: 'references', source: 'STATIC' }, + data: { source: 'OBSERVED', confidence: 0.8, lastSeenAt: new Date() }, + }); + } return edgeCount; } @@ -546,13 +594,9 @@ export class KgObservationalService { return node.id; } - private async bumpEdge( - organizationId: string, - sourceNodeId: string, - targetNodeId: string, - kind: string, - opts: { matchKey?: string; base: number; cap: number; status: string }, - ): Promise { + /** Add `times` observations to an edge, creating it if it is new. */ + private async bumpEdge(organizationId: string, b: EdgeBump): Promise { + const { src: sourceNodeId, tgt: targetNodeId, kind, times } = b; const existing = await this.prisma.kgEdge.findUnique({ where: { organizationId_sourceNodeId_targetNodeId_kind: { @@ -564,6 +608,8 @@ export class KgObservationalService { }, select: { id: true, observations: true, isManual: true, status: true }, }); + const observations = (existing?.observations ?? 0) + times; + const confidence = Math.min(b.cap, b.base + 0.05 * (observations - 1)); if (!existing) { await this.prisma.kgEdge.create({ data: { @@ -571,28 +617,27 @@ export class KgObservationalService { sourceNodeId, targetNodeId, kind, - matchKey: opts.matchKey, + matchKey: b.matchKey, source: 'OBSERVED', - confidence: Math.min(opts.cap, opts.base), - observations: 1, - status: opts.status, + confidence, + observations, + status: b.status, }, }); return; } - const observations = existing.observations + 1; await this.prisma.kgEdge.update({ where: { id: existing.id }, data: { observations, - confidence: Math.min(opts.cap, opts.base + 0.05 * (observations - 1)), + confidence, source: 'OBSERVED', - matchKey: opts.matchKey, + matchKey: b.matchKey, lastSeenAt: new Date(), // Never silently re-open a link the user rejected, nor downgrade a confirmed one. ...(existing.isManual || existing.status === 'rejected' ? {} - : { status: opts.status }), + : { status: b.status }), }, }); }