diff --git a/apps/api/src/common/tracing/index.ts b/apps/api/src/common/tracing/index.ts new file mode 100644 index 00000000..fe2d0a58 --- /dev/null +++ b/apps/api/src/common/tracing/index.ts @@ -0,0 +1,3 @@ +export * from "./trace-context"; +export * from "./trace-carrier"; +export * from "./trace-propagation.service"; diff --git a/apps/api/src/common/tracing/trace-carrier.ts b/apps/api/src/common/tracing/trace-carrier.ts new file mode 100644 index 00000000..7af65977 --- /dev/null +++ b/apps/api/src/common/tracing/trace-carrier.ts @@ -0,0 +1,164 @@ +import { + createChildTraceContext, + createRootTraceContext, + formatBaggage, + formatTraceparent, + generateSpanId, + parseBaggage, + parseTraceparent, + TraceContext, +} from "./trace-context"; + +export interface HttpHeadersCarrier { + [header: string]: string | string[] | undefined; +} + +export interface JobEnvelopeWithTrace { + payload?: any; + _traceContext?: { + traceparent: string; + tracestate?: string; + baggage?: string; + tenantId?: string; + startedAt?: number; + }; + [key: string]: any; +} + +/** + * Extracts a TraceContext from incoming HTTP request headers. + * Supports W3C traceparent, baggage, and fallback headers (x-correlation-id, x-request-id, x-tenant-id). + */ +export function extractFromHttpHeaders( + headers: HttpHeadersCarrier, +): TraceContext { + const getHeader = (name: string): string | undefined => { + const val = headers[name.toLowerCase()] ?? headers[name]; + if (Array.isArray(val)) { + return val[0]; + } + return val; + }; + + const traceparentHeader = getHeader("traceparent"); + const baggageHeader = getHeader("baggage"); + const tenantIdHeader = getHeader("x-tenant-id"); + + const parsed = parseTraceparent(traceparentHeader); + const baggage = parseBaggage(baggageHeader); + + const tenantId = tenantIdHeader || baggage.tenantId || undefined; + if (tenantId) { + baggage.tenantId = tenantId; + } + + if (parsed) { + return { + traceId: parsed.traceId, + spanId: generateSpanId(), + parentId: parsed.parentId, + sampled: parsed.sampled, + tenantId, + baggage, + startedAt: Date.now(), + }; + } + + // Fallback to x-correlation-id or x-request-id if valid 32-char hex, or create new root trace + return createRootTraceContext({ + tenantId, + baggage, + }); +} + +/** + * Injects TraceContext into outgoing HTTP headers. + */ +export function injectIntoHttpHeaders( + context: TraceContext, + headers: Record = {}, +): Record { + headers["traceparent"] = formatTraceparent( + context.traceId, + context.spanId, + context.sampled, + ); + + const baggageStr = formatBaggage(context.baggage); + if (baggageStr) { + headers["baggage"] = baggageStr; + } + + if (context.tenantId) { + headers["x-tenant-id"] = context.tenantId; + } + + headers["x-trace-id"] = context.traceId; + headers["x-span-id"] = context.spanId; + + return headers; +} + +/** + * Injects trace context into a queue job payload/envelope. + */ +export function injectIntoJobEnvelope>( + jobData: T, + context: TraceContext, +): T & JobEnvelopeWithTrace { + const traceparent = formatTraceparent( + context.traceId, + context.spanId, + context.sampled, + ); + const baggage = formatBaggage(context.baggage); + + return { + ...jobData, + _traceContext: { + traceparent, + ...(baggage ? { baggage } : {}), + ...(context.tenantId ? { tenantId: context.tenantId } : {}), + startedAt: context.startedAt, + }, + }; +} + +/** + * Extracts and restores TraceContext from a job payload/envelope. + * Returns child trace context linked to the job's span. + */ +export function extractFromJobEnvelope(jobData: any): TraceContext { + if (!jobData || typeof jobData !== "object") { + return createRootTraceContext(); + } + + const envelope = jobData._traceContext; + if (!envelope || !envelope.traceparent) { + // Check if directly in payload or fallback + const tenantId = jobData.tenantId || jobData.payload?.tenantId; + return createRootTraceContext({ tenantId }); + } + + const parsed = parseTraceparent(envelope.traceparent); + if (!parsed) { + return createRootTraceContext({ tenantId: envelope.tenantId }); + } + + const baggage = parseBaggage(envelope.baggage); + const tenantId = envelope.tenantId || baggage.tenantId; + if (tenantId) { + baggage.tenantId = tenantId; + } + + const parentContext: TraceContext = { + traceId: parsed.traceId, + spanId: parsed.parentId, + sampled: parsed.sampled, + tenantId, + baggage, + startedAt: envelope.startedAt ?? Date.now(), + }; + + return createChildTraceContext(parentContext); +} diff --git a/apps/api/src/common/tracing/trace-context.spec.ts b/apps/api/src/common/tracing/trace-context.spec.ts new file mode 100644 index 00000000..de1f79b0 --- /dev/null +++ b/apps/api/src/common/tracing/trace-context.spec.ts @@ -0,0 +1,184 @@ +import { + parseTraceparent, + formatTraceparent, + parseBaggage, + formatBaggage, + generateTraceId, + generateSpanId, + createRootTraceContext, + createChildTraceContext, +} from "./trace-context"; +import { + extractFromHttpHeaders, + injectIntoHttpHeaders, + extractFromJobEnvelope, + injectIntoJobEnvelope, +} from "./trace-carrier"; +import { TracePropagationService } from "./trace-propagation.service"; + +describe("W3C Trace Context and Carrier (#1263)", () => { + describe("parseTraceparent and formatTraceparent", () => { + it("parses valid standard W3C traceparent headers", () => { + const header = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + const parsed = parseTraceparent(header); + expect(parsed).not.toBeNull(); + expect(parsed!.traceId).toBe("4bf92f3577b34da6a3ce929d0e0e4736"); + expect(parsed!.parentId).toBe("00f067aa0ba902b7"); + expect(parsed!.sampled).toBe(true); + }); + + it("parses unsampled traceparent flag", () => { + const header = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00"; + const parsed = parseTraceparent(header); + expect(parsed!.sampled).toBe(false); + }); + + it("rejects invalid formats, wrong lengths, or non-hex characters", () => { + expect(parseTraceparent("not-a-traceparent")).toBeNull(); + expect(parseTraceparent("00-shorttrace-00f067aa0ba902b7-01")).toBeNull(); + expect( + parseTraceparent("00-4bf92f3577b34da6a3ce929d0e0e4736-shortspan-01"), + ).toBeNull(); + expect( + parseTraceparent( + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-xyz", + ), + ).toBeNull(); + expect( + parseTraceparent( + "ff-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + ), + ).toBeNull(); + expect(parseTraceparent("")).toBeNull(); + expect(parseTraceparent(null)).toBeNull(); + }); + + it("rejects all-zeros traceId or parentId per W3C specification", () => { + expect( + parseTraceparent( + "00-00000000000000000000000000000000-00f067aa0ba902b7-01", + ), + ).toBeNull(); + expect( + parseTraceparent( + "00-4bf92f3577b34da6a3ce929d0e0e4736-0000000000000000-01", + ), + ).toBeNull(); + }); + + it("formats valid traceparent header string", () => { + const traceId = "4bf92f3577b34da6a3ce929d0e0e4736"; + const spanId = "00f067aa0ba902b7"; + const formatted = formatTraceparent(traceId, spanId, true); + expect(formatted).toBe( + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + ); + }); + + it("throws when formatting with invalid traceId or spanId", () => { + expect(() => formatTraceparent("invalid", "00f067aa0ba902b7")).toThrow( + TypeError, + ); + expect(() => + formatTraceparent("4bf92f3577b34da6a3ce929d0e0e4736", "short"), + ).toThrow(TypeError); + }); + }); + + describe("parseBaggage and formatBaggage", () => { + it("parses comma-separated key-value pairs with URL decoding", () => { + const raw = "userId=alice,role=admin,tenantId=acme-corp%20ltd"; + const baggage = parseBaggage(raw); + expect(baggage.userId).toBe("alice"); + expect(baggage.role).toBe("admin"); + expect(baggage.tenantId).toBe("acme-corp ltd"); + }); + + it("formats baggage object into standard string", () => { + const formatted = formatBaggage({ userId: "alice", tenantId: "acme" }); + expect(formatted).toContain("userId=alice"); + expect(formatted).toContain("tenantId=acme"); + }); + + it("handles empty or malformed baggage headers safely", () => { + expect(parseBaggage("")).toEqual({}); + expect(parseBaggage("malformed-no-equal")).toEqual({}); + expect(parseBaggage(undefined)).toEqual({}); + }); + }); + + describe("ID Generators and Context Creation", () => { + it("generates 32-hex character traceId and 16-hex character spanId", () => { + const traceId = generateTraceId(); + const spanId = generateSpanId(); + expect(traceId).toHaveLength(32); + expect(/^[0-9a-f]{32}$/.test(traceId)).toBe(true); + expect(spanId).toHaveLength(16); + expect(/^[0-9a-f]{16}$/.test(spanId)).toBe(true); + }); + + it("creates root context and propagates traceId to child context", () => { + const root = createRootTraceContext({ + tenantId: "tenant-99", + baggage: { env: "production" }, + }); + + expect(root.traceId).toBeDefined(); + expect(root.parentId).toBeUndefined(); + expect(root.tenantId).toBe("tenant-99"); + expect(root.baggage.env).toBe("production"); + + const child = createChildTraceContext(root, { + extraBaggage: { step: "parsing" }, + }); + + expect(child.traceId).toBe(root.traceId); + expect(child.parentId).toBe(root.spanId); + expect(child.spanId).not.toBe(root.spanId); + expect(child.tenantId).toBe("tenant-99"); + expect(child.baggage.step).toBe("parsing"); + expect(child.baggage.env).toBe("production"); + }); + }); + + describe("Trace Carriers (HTTP and Job Envelopes)", () => { + it("extracts from HTTP headers and injects into outgoing headers", () => { + const incomingHeaders = { + traceparent: "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + baggage: "tenantId=tenant-corp,region=us-east", + }; + + const extracted = extractFromHttpHeaders(incomingHeaders); + expect(extracted.traceId).toBe("4bf92f3577b34da6a3ce929d0e0e4736"); + expect(extracted.parentId).toBe("00f067aa0ba902b7"); + expect(extracted.tenantId).toBe("tenant-corp"); + + const outgoing: Record = {}; + injectIntoHttpHeaders(extracted, outgoing); + expect(outgoing.traceparent).toMatch( + /^00-4bf92f3577b34da6a3ce929d0e0e4736-[0-9a-f]{16}-01$/, + ); + expect(outgoing["x-tenant-id"]).toBe("tenant-corp"); + expect(outgoing["x-trace-id"]).toBe("4bf92f3577b34da6a3ce929d0e0e4736"); + }); + + it("injects trace context into job envelope and extracts it in worker", () => { + const rootCtx = createRootTraceContext({ + tenantId: "tenant-worker-test", + }); + const jobData = { + scanId: "scan-job-42", + repositoryUrl: "https://github.com/MDTechLabs/GasGuard", + }; + + const enqueuedData = injectIntoJobEnvelope(jobData, rootCtx); + expect(enqueuedData._traceContext).toBeDefined(); + expect(enqueuedData._traceContext?.tenantId).toBe("tenant-worker-test"); + + const workerCtx = extractFromJobEnvelope(enqueuedData); + expect(workerCtx.traceId).toBe(rootCtx.traceId); + expect(workerCtx.parentId).toBe(rootCtx.spanId); + expect(workerCtx.tenantId).toBe("tenant-worker-test"); + }); + }); +}); diff --git a/apps/api/src/common/tracing/trace-context.ts b/apps/api/src/common/tracing/trace-context.ts new file mode 100644 index 00000000..1973dcb2 --- /dev/null +++ b/apps/api/src/common/tracing/trace-context.ts @@ -0,0 +1,216 @@ +import * as crypto from "crypto"; + +/** + * W3C Trace Context Specification (#1263). + * + * Implements W3C Trace Context Level 1: + * traceparent: {version}-{trace-id}-{parent-id}-{trace-flags} + * - version: 2 hex characters ('00') + * - trace-id: 32 hex characters (cannot be all '0's) + * - parent-id: 16 hex characters (cannot be all '0's) + * - trace-flags: 2 hex characters (bit 1 is sampled: 01 vs 00) + */ + +export interface TraceContext { + traceId: string; + spanId: string; + parentId?: string; + sampled: boolean; + tenantId?: string; + baggage: Record; + startedAt: number; +} + +const TRACEPARENT_REGEX = + /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/i; +const ALL_ZEROS_32 = "00000000000000000000000000000000"; +const ALL_ZEROS_16 = "0000000000000000"; + +/** + * Validates and parses a W3C traceparent header string. + * Returns null if the header is malformed, invalid version, or all zeros. + */ +export function parseTraceparent( + header: string | undefined | null, +): { traceId: string; parentId: string; sampled: boolean } | null { + if (!header || typeof header !== "string") { + return null; + } + + const trimmed = header.trim(); + const match = TRACEPARENT_REGEX.exec(trimmed); + if (!match) { + return null; + } + + const [, version, traceId, parentId, flags] = match; + + // Version '00' is standard W3C. Future versions must not be ff. + if (version === "ff") { + return null; + } + + // Validate non-zero IDs per W3C specification + if ( + traceId.toLowerCase() === ALL_ZEROS_32 || + parentId.toLowerCase() === ALL_ZEROS_16 + ) { + return null; + } + + const flagNum = parseInt(flags, 16); + const sampled = (flagNum & 0x01) === 0x01; + + return { + traceId: traceId.toLowerCase(), + parentId: parentId.toLowerCase(), + sampled, + }; +} + +/** + * Formats a W3C traceparent header. + */ +export function formatTraceparent( + traceId: string, + spanId: string, + sampled: boolean = true, +): string { + if ( + !/^[0-9a-f]{32}$/i.test(traceId) || + traceId.toLowerCase() === ALL_ZEROS_32 + ) { + throw new TypeError( + "traceId must be a 32-character non-zero hexadecimal string", + ); + } + if ( + !/^[0-9a-f]{16}$/i.test(spanId) || + spanId.toLowerCase() === ALL_ZEROS_16 + ) { + throw new TypeError( + "spanId must be a 16-character non-zero hexadecimal string", + ); + } + + const flags = sampled ? "01" : "00"; + return `00-${traceId.toLowerCase()}-${spanId.toLowerCase()}-${flags}`; +} + +/** + * Parses W3C baggage header or comma-delimited key=value strings. + */ +export function parseBaggage( + header: string | undefined | null, +): Record { + if (!header || typeof header !== "string") { + return {}; + } + + const baggage: Record = {}; + const pairs = header.split(","); + + for (const pair of pairs) { + const eqIdx = pair.indexOf("="); + if (eqIdx <= 0) continue; + const key = decodeURIComponent(pair.slice(0, eqIdx).trim()); + const val = decodeURIComponent(pair.slice(eqIdx + 1).trim()); + if (key) { + baggage[key] = val; + } + } + + return baggage; +} + +/** + * Formats a key-value record into a W3C baggage header string. + */ +export function formatBaggage(baggage: Record): string { + return Object.entries(baggage) + .filter(([k, v]) => k && v !== undefined && v !== null) + .map(([k, v]) => `${encodeURIComponent(k)}=${encodeURIComponent(v)}`) + .join(","); +} + +/** + * Generates a valid 32-character hex trace ID. + */ +export function generateTraceId(): string { + let id = crypto.randomBytes(16).toString("hex"); + while (id === ALL_ZEROS_32) { + id = crypto.randomBytes(16).toString("hex"); + } + return id; +} + +/** + * Generates a valid 16-character hex span ID. + */ +export function generateSpanId(): string { + let id = crypto.randomBytes(8).toString("hex"); + while (id === ALL_ZEROS_16) { + id = crypto.randomBytes(8).toString("hex"); + } + return id; +} + +/** + * Initializes a new root trace context. + */ +export function createRootTraceContext(options?: { + tenantId?: string; + sampled?: boolean; + baggage?: Record; +}): TraceContext { + const baggage = { ...(options?.baggage ?? {}) }; + if (options?.tenantId) { + baggage.tenantId = options.tenantId; + } + + return { + traceId: generateTraceId(), + spanId: generateSpanId(), + sampled: options?.sampled ?? true, + tenantId: options?.tenantId, + baggage, + startedAt: Date.now(), + }; +} + +/** + * Creates a child span trace context linked to parent. + * Retains traceId, propagates tenantId and baggage. + */ +export function createChildTraceContext( + parent: TraceContext, + options?: { + extraBaggage?: Record; + overrideTenantId?: string; + }, +): TraceContext { + if (!parent || !parent.traceId || !parent.spanId) { + throw new TypeError("parent context must have valid traceId and spanId"); + } + + const baggage = { + ...parent.baggage, + ...(options?.extraBaggage ?? {}), + }; + + const tenantId = + options?.overrideTenantId ?? parent.tenantId ?? baggage.tenantId; + if (tenantId) { + baggage.tenantId = tenantId; + } + + return { + traceId: parent.traceId, + spanId: generateSpanId(), + parentId: parent.spanId, + sampled: parent.sampled, + tenantId, + baggage, + startedAt: Date.now(), + }; +} diff --git a/apps/api/src/common/tracing/trace-propagation.integration.spec.ts b/apps/api/src/common/tracing/trace-propagation.integration.spec.ts new file mode 100644 index 00000000..067a52d0 --- /dev/null +++ b/apps/api/src/common/tracing/trace-propagation.integration.spec.ts @@ -0,0 +1,105 @@ +import { TracePropagationService } from "./trace-propagation.service"; +import { createRootTraceContext } from "./trace-context"; +import { injectIntoJobEnvelope, extractFromJobEnvelope } from "./trace-carrier"; + +describe("Trace Propagation End-to-End Across API and Workers (#1263)", () => { + it("propagates traceId from API request into BullMQ-style job envelope and worker span execution", async () => { + const traceService = new TracePropagationService(); + + // 1. In API gateway: incoming request initializes or extracts trace context + const apiContext = createRootTraceContext({ + tenantId: "tenant-enterprise-1", + baggage: { user: "security-auditor", route: "POST /v1/scans" }, + }); + + // 2. API enqueues job with trace context injected + const apiJobPayload = { + scanId: "scan-e2e-101", + repo: "https://github.com/MDTechLabs/GasGuard", + commit: "a1b2c3d4", + }; + const queuedEnvelope = injectIntoJobEnvelope(apiJobPayload, apiContext); + + // 3. Worker dequeues job and extracts context + const workerContext = extractFromJobEnvelope(queuedEnvelope); + expect(workerContext.traceId).toBe(apiContext.traceId); + expect(workerContext.parentId).toBe(apiContext.spanId); + expect(workerContext.tenantId).toBe("tenant-enterprise-1"); + + // 4. Worker executes scan pipeline inside AsyncLocalStorage context + await traceService.runWithContext(workerContext, async (ctx) => { + expect(traceService.getActiveContext()?.traceId).toBe(apiContext.traceId); + + // Start child span for AST parsing + const { span: parseSpan } = traceService.startSpan("ast_parsing", { + stage: "parsing", + language: "solidity", + }); + expect(parseSpan.traceId).toBe(apiContext.traceId); + expect(parseSpan.parentId).toBe(workerContext.spanId); + + // Simulate work + await new Promise((resolve) => setTimeout(resolve, 15)); + traceService.endSpan(parseSpan, "ok"); + + // Start child span for Rule Evaluation + const { span: evalSpan } = traceService.startSpan("rules_evaluation", { + stage: "analysis", + }); + expect(evalSpan.traceId).toBe(apiContext.traceId); + expect(evalSpan.parentId).toBe(workerContext.spanId); + + await new Promise((resolve) => setTimeout(resolve, 10)); + traceService.endSpan(evalSpan, "ok"); + }); + + const completed = traceService.getCompletedSpans(); + expect(completed).toHaveLength(2); + expect(completed[0].operationName).toBe("ast_parsing"); + expect(completed[0].status).toBe("ok"); + expect(completed[0].durationMs).toBeGreaterThanOrEqual(10); + expect(completed[1].operationName).toBe("rules_evaluation"); + expect(completed[1].status).toBe("ok"); + }); + + it("handles high concurrency of distinct jobs across multiple tenants without context bleeding", async () => { + const traceService = new TracePropagationService(); + const concurrentJobsCount = 20; + + const runJob = async (id: number) => { + const tenantId = `tenant-${id % 3}`; + const rootCtx = createRootTraceContext({ + tenantId, + baggage: { jobId: `job-${id}` }, + }); + + const envelope = injectIntoJobEnvelope({ jobId: `job-${id}` }, rootCtx); + const workerCtx = extractFromJobEnvelope(envelope); + + return traceService.runWithContext(workerCtx, async () => { + // Interleaved delays + await new Promise((resolve) => setTimeout(resolve, Math.random() * 20)); + + const active = traceService.getActiveContext(); + expect(active?.traceId).toBe(rootCtx.traceId); + expect(active?.tenantId).toBe(tenantId); + expect(active?.baggage.jobId).toBe(`job-${id}`); + + const { span } = traceService.startSpan(`scan-task-${id}`); + expect(span.traceId).toBe(rootCtx.traceId); + expect(span.tenantId).toBe(tenantId); + traceService.endSpan(span, "ok"); + + return span.traceId; + }); + }; + + const traceIds = await Promise.all( + Array.from({ length: concurrentJobsCount }, (_, i) => runJob(i)), + ); + + // Verify all trace IDs are distinct + const uniqueIds = new Set(traceIds); + expect(uniqueIds.size).toBe(concurrentJobsCount); + }); +}); diff --git a/apps/api/src/common/tracing/trace-propagation.service.ts b/apps/api/src/common/tracing/trace-propagation.service.ts new file mode 100644 index 00000000..7c4f20b9 --- /dev/null +++ b/apps/api/src/common/tracing/trace-propagation.service.ts @@ -0,0 +1,118 @@ +import { AsyncLocalStorage } from "async_hooks"; +import { + createChildTraceContext, + createRootTraceContext, + generateSpanId, + TraceContext, +} from "./trace-context"; + +export interface ActiveSpan { + traceId: string; + spanId: string; + parentId?: string; + operationName: string; + startTime: number; + endTime?: number; + durationMs?: number; + status: "ok" | "error"; + tags: Record; + tenantId?: string; + errorMessage?: string; +} + +export class TracePropagationService { + private readonly asyncLocalStorage = new AsyncLocalStorage(); + private readonly completedSpans: ActiveSpan[] = []; + private readonly maxCompletedSpans: number; + + constructor(maxCompletedSpans: number = 1000) { + this.maxCompletedSpans = maxCompletedSpans; + } + + /** + * Returns current active trace context if present. + */ + getActiveContext(): TraceContext | undefined { + return this.asyncLocalStorage.getStore(); + } + + /** + * Runs an asynchronous or synchronous function within the scope of a TraceContext. + */ + async runWithContext( + context: TraceContext, + fn: (ctx: TraceContext) => Promise | T, + ): Promise { + return this.asyncLocalStorage.run(context, () => fn(context)); + } + + /** + * Starts a span. If an active trace context exists, links as a child span. + * If not, initializes a new root trace context. + */ + startSpan( + operationName: string, + tags: Record = {}, + parentContext?: TraceContext, + ): { span: ActiveSpan; context: TraceContext } { + const activeCtx = parentContext ?? this.getActiveContext(); + let newContext: TraceContext; + + if (activeCtx) { + newContext = createChildTraceContext(activeCtx, { + extraBaggage: tags, + }); + } else { + newContext = createRootTraceContext({ + baggage: tags, + tenantId: tags.tenantId, + }); + } + + const span: ActiveSpan = { + traceId: newContext.traceId, + spanId: newContext.spanId, + parentId: newContext.parentId, + operationName, + startTime: Date.now(), + status: "ok", + tags: { ...newContext.baggage, ...tags }, + tenantId: newContext.tenantId, + }; + + return { span, context: newContext }; + } + + /** + * Ends a span and records duration and completion status. + */ + endSpan( + span: ActiveSpan, + status: "ok" | "error" = "ok", + errorMessage?: string, + ): ActiveSpan { + span.endTime = Date.now(); + span.durationMs = span.endTime - span.startTime; + span.status = status; + if (errorMessage) { + span.errorMessage = errorMessage; + } + + this.completedSpans.push(span); + if (this.completedSpans.length > this.maxCompletedSpans) { + this.completedSpans.shift(); + } + + return span; + } + + getCompletedSpans(): ActiveSpan[] { + return [...this.completedSpans]; + } + + clearSpans(): void { + this.completedSpans.length = 0; + } +} + +export const defaultTraceService = new TracePropagationService(); diff --git a/apps/api/src/exports/incident-timeline/incident-timeline-export.service.ts b/apps/api/src/exports/incident-timeline/incident-timeline-export.service.ts new file mode 100644 index 00000000..2408683e --- /dev/null +++ b/apps/api/src/exports/incident-timeline/incident-timeline-export.service.ts @@ -0,0 +1,261 @@ +import { + ExportFormat, + IncidentExportResult, + ScanIncidentRecord, + TimelineEvent, +} from "./incident-timeline-types"; +import { IncidentTimelineStore } from "./incident-timeline-store"; + +export class IncidentTimelineExportService { + constructor( + private readonly store: IncidentTimelineStore = new IncidentTimelineStore(), + ) {} + + getStore(): IncidentTimelineStore { + return this.store; + } + + recordEvent(event: TimelineEvent): TimelineEvent { + return this.store.recordEvent(event); + } + + finalizeIncident(incident: ScanIncidentRecord): ScanIncidentRecord { + return this.store.finalizeIncident(incident); + } + + getIncident( + tenantId: string, + scanId: string, + ): ScanIncidentRecord | undefined { + return this.store.getIncident(tenantId, scanId); + } + + listIncidents( + tenantId: string, + options?: { limit?: number; offset?: number }, + ): ScanIncidentRecord[] { + return this.store.listIncidents(tenantId, options); + } + + exportIncident( + tenantId: string, + scanId: string, + format: ExportFormat, + ): IncidentExportResult { + if (!tenantId || typeof tenantId !== "string" || tenantId.trim() === "") { + throw new TypeError("tenantId must be a non-empty string"); + } + if (!scanId || typeof scanId !== "string" || scanId.trim() === "") { + throw new TypeError("scanId must be a non-empty string"); + } + + const incident = this.store.getIncident(tenantId, scanId); + if (!incident) { + throw new Error( + `Incident for scan ${scanId} not found under tenant ${tenantId}`, + ); + } + + const nowIso = new Date().toISOString(); + let data: string; + let contentType: string; + let extension: string; + + switch (format) { + case "json": + data = this.formatJson(incident); + contentType = "application/json"; + extension = "json"; + break; + case "markdown": + data = this.formatMarkdown(incident); + contentType = "text/markdown"; + extension = "md"; + break; + case "csv": + data = this.formatCsv(incident); + contentType = "text/csv"; + extension = "csv"; + break; + default: + throw new TypeError(`Unsupported export format: ${format}`); + } + + const filename = `incident-${incident.scanId}-timeline.${extension}`; + + return { + format, + contentType, + filename, + data, + metadata: { + scanId: incident.scanId, + incidentId: incident.incidentId, + tenantId: incident.tenantId, + eventCount: incident.events.length, + generatedAt: nowIso, + }, + }; + } + + formatJson(incident: ScanIncidentRecord): string { + return JSON.stringify( + { + incidentId: incident.incidentId, + scanId: incident.scanId, + tenantId: incident.tenantId, + status: incident.status, + failureCategory: incident.failureCategory, + summary: incident.summary, + repositoryUrl: incident.repositoryUrl, + commitHash: incident.commitHash, + startedAt: new Date(incident.startedAt).toISOString(), + failedAt: new Date(incident.failedAt).toISOString(), + durationMs: incident.durationMs, + workerLifecycle: incident.workerLifecycle, + diagnosticsCount: incident.diagnosticsCount, + timelineEvents: incident.events.map((e) => ({ + eventId: e.eventId, + timestamp: new Date(e.timestamp).toISOString(), + stage: e.stage, + severity: e.severity, + type: e.type, + actor: e.actor, + message: e.message, + metadata: e.metadata, + errorDetails: e.errorDetails, + })), + }, + null, + 2, + ); + } + + formatMarkdown(incident: ScanIncidentRecord): string { + const lines: string[] = []; + + lines.push(`# Incident Report: Scan ${incident.scanId}`); + lines.push(""); + lines.push(`- **Incident ID:** \`${incident.incidentId}\``); + lines.push(`- **Tenant ID:** \`${incident.tenantId}\``); + lines.push(`- **Status:** \`${incident.status.toUpperCase()}\``); + lines.push(`- **Failure Category:** \`${incident.failureCategory}\``); + lines.push(`- **Repository:** ${incident.repositoryUrl}`); + lines.push(`- **Commit:** \`${incident.commitHash}\``); + lines.push(`- **Duration:** ${incident.durationMs}ms`); + lines.push( + `- **Failure Time:** ${new Date(incident.failedAt).toISOString()}`, + ); + + if (incident.workerLifecycle) { + lines.push(""); + lines.push("## Worker Lifecycle"); + lines.push(`- **Worker ID:** \`${incident.workerLifecycle.workerId}\``); + if (incident.workerLifecycle.leaseFencingToken !== undefined) { + lines.push( + `- **Lease Fencing Token:** \`${incident.workerLifecycle.leaseFencingToken}\``, + ); + } + if (incident.workerLifecycle.leaseAcquiredAt) { + lines.push( + `- **Lease Acquired At:** ${new Date(incident.workerLifecycle.leaseAcquiredAt).toISOString()}`, + ); + } + } + + if (incident.diagnosticsCount) { + lines.push(""); + lines.push("## Diagnostics Summary"); + lines.push(`- **Total Diagnostics:** ${incident.diagnosticsCount.total}`); + lines.push( + `- **Recovered Warnings:** ${incident.diagnosticsCount.recovered}`, + ); + lines.push( + `- **Unrecovered Errors:** ${incident.diagnosticsCount.unrecovered}`, + ); + } + + lines.push(""); + lines.push("## Summary & Root Cause"); + lines.push(incident.summary); + + lines.push(""); + lines.push("## Chronological Timeline"); + lines.push( + "| Timestamp (UTC) | Stage | Severity | Event Type | Actor | Message |", + ); + lines.push("|---|---|---|---|---|---|"); + + for (const event of incident.events) { + const timeStr = new Date(event.timestamp).toISOString(); + const cleanMsg = event.message.replace(/\|/g, "\\|"); + lines.push( + `| ${timeStr} | ${event.stage} | ${event.severity.toUpperCase()} | ${event.type} | ${event.actor} | ${cleanMsg} |`, + ); + } + + const errorsWithDetails = incident.events.filter((e) => e.errorDetails); + if (errorsWithDetails.length > 0) { + lines.push(""); + lines.push("## Error Details"); + for (const errEvent of errorsWithDetails) { + lines.push( + `### [${errEvent.errorDetails!.code}] ${errEvent.errorDetails!.message}`, + ); + lines.push(`- **Category:** ${errEvent.errorDetails!.category}`); + lines.push( + `- **Retryable:** ${errEvent.errorDetails!.retryable ? "Yes" : "No"}`, + ); + if (errEvent.errorDetails!.stackExcerpt) { + lines.push("```"); + lines.push(errEvent.errorDetails!.stackExcerpt); + lines.push("```"); + } + } + } + + return lines.join("\n"); + } + + formatCsv(incident: ScanIncidentRecord): string { + const headers = [ + "event_id", + "timestamp_iso", + "stage", + "severity", + "type", + "actor", + "message", + "error_code", + "retryable", + ]; + + const rows = incident.events.map((e) => [ + escapeCsv(e.eventId), + escapeCsv(new Date(e.timestamp).toISOString()), + escapeCsv(e.stage), + escapeCsv(e.severity), + escapeCsv(e.type), + escapeCsv(e.actor), + escapeCsv(e.message), + escapeCsv(e.errorDetails?.code ?? ""), + escapeCsv( + e.errorDetails ? (e.errorDetails.retryable ? "true" : "false") : "", + ), + ]); + + return [headers.join(","), ...rows.map((r) => r.join(","))].join("\r\n"); + } +} + +function escapeCsv(value: string): string { + if ( + value.includes(",") || + value.includes('"') || + value.includes("\n") || + value.includes("\r") + ) { + return `"${value.replace(/"/g, '""')}"`; + } + return value; +} diff --git a/apps/api/src/exports/incident-timeline/incident-timeline-store.ts b/apps/api/src/exports/incident-timeline/incident-timeline-store.ts new file mode 100644 index 00000000..e3ff56b4 --- /dev/null +++ b/apps/api/src/exports/incident-timeline/incident-timeline-store.ts @@ -0,0 +1,243 @@ +import { + DEFAULT_INCIDENT_STORE_CONFIG, + FailureCategory, + IncidentTimelineStoreConfig, + ScanIncidentRecord, + TimelineEvent, +} from "./incident-timeline-types"; + +/** + * Sanitizes arbitrary text to strip secrets, API keys, private keys, and passwords. + */ +export function sanitizeIncidentText(input: string): string { + if (!input || typeof input !== "string") { + return ""; + } + return ( + input + // Hex private keys (0x followed by 64 hex characters) + .replace(/0x[a-fA-F0-9]{64}/g, "0x[REDACTED_PRIVATE_KEY]") + // Bearer / JWT tokens + .replace( + /Bearer\s+[A-Za-z0-9-_=]+\.[A-Za-z0-9-_=]+\.?[A-Za-z0-9-_.+/=]*/g, + "Bearer [REDACTED_TOKEN]", + ) + // Generic tokens/passwords/secrets + .replace( + /(password|secret|token|api_?key|auth)["']?\s*[:=]\s*["']?([^"',;\s]+)/gi, + "$1=[REDACTED]", + ) + // URL embedded credentials + .replace(/:\/\/([^:]+):([^@]+)@/g, "://$1:[REDACTED]@") + ); +} + +export class IncidentTimelineStore { + private readonly config: IncidentTimelineStoreConfig; + // Map> + private readonly tenantIncidents: Map< + string, + Map + > = new Map(); + // Buffer for events arriving before an incident record is finalized: Map> + private readonly pendingEvents: Map> = + new Map(); + + constructor(config?: Partial) { + this.config = { ...DEFAULT_INCIDENT_STORE_CONFIG, ...config }; + if (this.config.maxIncidentsPerTenant <= 0) { + throw new RangeError("maxIncidentsPerTenant must be a positive integer"); + } + if (this.config.maxEventsPerIncident <= 0) { + throw new RangeError("maxEventsPerIncident must be a positive integer"); + } + } + + /** + * Records a timeline event for a scan. Preserves strict tenant isolation. + */ + recordEvent(event: TimelineEvent): TimelineEvent { + this.validateTenantAndScan(event.tenantId, event.scanId); + + const sanitizedMessage = sanitizeIncidentText(event.message); + const sanitizedError = event.errorDetails + ? { + ...event.errorDetails, + message: sanitizeIncidentText(event.errorDetails.message), + stackExcerpt: event.errorDetails.stackExcerpt + ? sanitizeIncidentText(event.errorDetails.stackExcerpt) + : undefined, + } + : undefined; + + const sanitizedEvent: TimelineEvent = { + ...event, + message: sanitizedMessage, + errorDetails: sanitizedError, + }; + + const tenantMap = this.getOrCreateTenantIncidents(event.tenantId); + const existingIncident = tenantMap.get(event.scanId); + + if (existingIncident) { + this.addEventToIncident(existingIncident, sanitizedEvent); + } else { + // Buffer event in pending events map until incident is finalized + let tenantPending = this.pendingEvents.get(event.tenantId); + if (!tenantPending) { + tenantPending = new Map(); + this.pendingEvents.set(event.tenantId, tenantPending); + } + let scanEvents = tenantPending.get(event.scanId); + if (!scanEvents) { + scanEvents = []; + tenantPending.set(event.scanId, scanEvents); + } + if (scanEvents.length < this.config.maxEventsPerIncident) { + scanEvents.push(sanitizedEvent); + scanEvents.sort((a, b) => a.timestamp - b.timestamp); + } + } + + return sanitizedEvent; + } + + /** + * Finalizes an incident record when a scan fails or aborts. + * Merges any buffered events for this scan. + */ + finalizeIncident(incident: ScanIncidentRecord): ScanIncidentRecord { + this.validateTenantAndScan(incident.tenantId, incident.scanId); + + const tenantMap = this.getOrCreateTenantIncidents(incident.tenantId); + + // Bounded eviction per tenant + if ( + !tenantMap.has(incident.scanId) && + tenantMap.size >= this.config.maxIncidentsPerTenant + ) { + const oldestScanId = tenantMap.keys().next().value; + if (oldestScanId) { + tenantMap.delete(oldestScanId); + } + } + + const tenantPending = this.pendingEvents.get(incident.tenantId); + const pending = tenantPending?.get(incident.scanId) ?? []; + if (tenantPending) { + tenantPending.delete(incident.scanId); + } + + const existingEvents = tenantMap.get(incident.scanId)?.events ?? []; + const allEvents = [...existingEvents, ...pending, ...incident.events]; + // Deduplicate and sort chronologically + const uniqueEvents = Array.from( + new Map(allEvents.map((e) => [e.eventId, e])).values(), + ).sort((a, b) => a.timestamp - b.timestamp); + + const boundedEvents = uniqueEvents.slice( + 0, + this.config.maxEventsPerIncident, + ); + + const sanitizedIncident: ScanIncidentRecord = { + ...incident, + repositoryUrl: sanitizeIncidentText(incident.repositoryUrl), + summary: sanitizeIncidentText(incident.summary), + events: boundedEvents, + }; + + tenantMap.set(incident.scanId, sanitizedIncident); + return sanitizedIncident; + } + + /** + * Retrieves an incident by scan ID, enforcing tenant isolation. + */ + getIncident( + tenantId: string, + scanId: string, + ): ScanIncidentRecord | undefined { + this.validateTenantAndScan(tenantId, scanId); + const tenantMap = this.tenantIncidents.get(tenantId); + return tenantMap?.get(scanId); + } + + /** + * Lists all incidents for a specific tenant with optional pagination. + */ + listIncidents( + tenantId: string, + options?: { limit?: number; offset?: number }, + ): ScanIncidentRecord[] { + if (!tenantId || typeof tenantId !== "string" || tenantId.trim() === "") { + throw new TypeError("tenantId must be a non-empty string"); + } + + const tenantMap = this.tenantIncidents.get(tenantId); + if (!tenantMap) { + return []; + } + + const list = Array.from(tenantMap.values()).reverse(); // most recent first + const offset = Math.max(0, options?.offset ?? 0); + const limit = Math.max(1, options?.limit ?? 50); + + return list.slice(offset, offset + limit); + } + + /** + * Deletes an incident record for a tenant. + */ + deleteIncident(tenantId: string, scanId: string): boolean { + this.validateTenantAndScan(tenantId, scanId); + const tenantMap = this.tenantIncidents.get(tenantId); + return tenantMap?.delete(scanId) ?? false; + } + + clearTenant(tenantId: string): void { + if (!tenantId || typeof tenantId !== "string") { + throw new TypeError("tenantId must be a non-empty string"); + } + this.tenantIncidents.delete(tenantId); + this.pendingEvents.delete(tenantId); + } + + clearAll(): void { + this.tenantIncidents.clear(); + this.pendingEvents.clear(); + } + + private addEventToIncident( + incident: ScanIncidentRecord, + event: TimelineEvent, + ): void { + if (incident.events.some((e) => e.eventId === event.eventId)) { + return; + } + if (incident.events.length < this.config.maxEventsPerIncident) { + incident.events.push(event); + incident.events.sort((a, b) => a.timestamp - b.timestamp); + } + } + + private getOrCreateTenantIncidents( + tenantId: string, + ): Map { + let tenantMap = this.tenantIncidents.get(tenantId); + if (!tenantMap) { + tenantMap = new Map(); + this.tenantIncidents.set(tenantId, tenantMap); + } + return tenantMap; + } + + private validateTenantAndScan(tenantId: string, scanId: string): void { + if (!tenantId || typeof tenantId !== "string" || tenantId.trim() === "") { + throw new TypeError("tenantId must be a non-empty string"); + } + if (!scanId || typeof scanId !== "string" || scanId.trim() === "") { + throw new TypeError("scanId must be a non-empty string"); + } + } +} diff --git a/apps/api/src/exports/incident-timeline/incident-timeline-types.ts b/apps/api/src/exports/incident-timeline/incident-timeline-types.ts new file mode 100644 index 00000000..677fc479 --- /dev/null +++ b/apps/api/src/exports/incident-timeline/incident-timeline-types.ts @@ -0,0 +1,118 @@ +/** + * Incident Timeline Export Types (#1265). + * + * Captures chronological event histories, worker lifecycles, and failure diagnostics + * for failed and aborted repository scans under strict tenant isolation. + */ + +export type TimelineEventType = + | "scan_created" + | "worker_assigned" + | "lease_acquired" + | "stage_started" + | "checkpoint_recorded" + | "file_parsed" + | "parser_warning" + | "parser_error" + | "analyzer_invoked" + | "analyzer_failed" + | "resource_limit_exceeded" + | "network_timeout" + | "worker_heartbeat_missed" + | "worker_crashed" + | "lease_expired" + | "scan_aborted" + | "scan_failed" + | "scan_recovered"; + +export type TimelineEventSeverity = "info" | "warn" | "error" | "critical"; + +export type FailureCategory = + | "PARSER_FAILURE" + | "RESOURCE_EXHAUSTION" + | "TIMEOUT" + | "LEASE_EXPIRED" + | "UNSUPPORTED_LANGUAGE" + | "INTERNAL_ERROR" + | "SECURITY_VIOLATION"; + +export interface TimelineErrorDetails { + code: string; + category: string; + message: string; + retryable: boolean; + stackExcerpt?: string; +} + +export interface TimelineEvent { + eventId: string; + scanId: string; + tenantId: string; + timestamp: number; + type: TimelineEventType; + severity: TimelineEventSeverity; + stage: string; + actor: string; + message: string; + metadata?: Record; + errorDetails?: TimelineErrorDetails; +} + +export interface DiagnosticsCount { + recovered: number; + unrecovered: number; + total: number; +} + +export interface WorkerLifecycleSummary { + workerId: string; + leaseAcquiredAt?: number; + lastHeartbeatAt?: number; + leaseFencingToken?: number; + releasedAt?: number; +} + +export interface ScanIncidentRecord { + incidentId: string; + scanId: string; + tenantId: string; + repositoryUrl: string; + commitHash: string; + failureCategory: FailureCategory; + summary: string; + startedAt: number; + failedAt: number; + durationMs: number; + workerLifecycle?: WorkerLifecycleSummary; + diagnosticsCount?: DiagnosticsCount; + status: "failed" | "aborted" | "recovered"; + events: TimelineEvent[]; +} + +export type ExportFormat = "json" | "markdown" | "csv"; + +export interface IncidentExportResult { + format: ExportFormat; + contentType: string; + filename: string; + data: string; + metadata: { + scanId: string; + incidentId: string; + tenantId: string; + eventCount: number; + generatedAt: string; + }; +} + +export interface IncidentTimelineStoreConfig { + maxIncidentsPerTenant: number; + maxEventsPerIncident: number; + ttlMs: number; +} + +export const DEFAULT_INCIDENT_STORE_CONFIG: IncidentTimelineStoreConfig = { + maxIncidentsPerTenant: 50, + maxEventsPerIncident: 200, + ttlMs: 7 * 24 * 60 * 60 * 1000, // 7 days +}; diff --git a/apps/api/src/exports/incident-timeline/incident-timeline.integration.spec.ts b/apps/api/src/exports/incident-timeline/incident-timeline.integration.spec.ts new file mode 100644 index 00000000..da26d099 --- /dev/null +++ b/apps/api/src/exports/incident-timeline/incident-timeline.integration.spec.ts @@ -0,0 +1,134 @@ +import { IncidentTimelineStore } from "./incident-timeline-store"; +import { IncidentTimelineExportService } from "./incident-timeline-export.service"; +import { ScanIncidentRecord, TimelineEvent } from "./incident-timeline-types"; + +describe("Incident Timeline Concurrency & Fault Integration (#1265)", () => { + it("handles high concurrency from multiple workers emitting events for multiple scans", async () => { + const service = new IncidentTimelineExportService(); + const store = service.getStore(); + const tenantCount = 3; + const scansPerTenant = 5; + const eventsPerScan = 20; + + const eventPromises: Promise[] = []; + + for (let t = 0; t < tenantCount; t++) { + const tenantId = `tenant-${t}`; + for (let s = 0; s < scansPerTenant; s++) { + const scanId = `scan-${t}-${s}`; + for (let e = 0; e < eventsPerScan; e++) { + const promise = (async () => { + return store.recordEvent({ + eventId: `evt-${t}-${s}-${e}`, + scanId, + tenantId, + timestamp: Date.now() + e * 10, + type: e === eventsPerScan - 1 ? "scan_failed" : "stage_started", + severity: e === eventsPerScan - 1 ? "critical" : "info", + stage: e < 10 ? "parsing" : "analyzing", + actor: `worker-${(e % 4) + 1}`, + message: `Progress step ${e} executed`, + }); + })(); + eventPromises.push(promise); + } + } + } + + await Promise.all(eventPromises); + + // Now finalize all scans concurrently + const finalizePromises: Promise[] = []; + for (let t = 0; t < tenantCount; t++) { + const tenantId = `tenant-${t}`; + for (let s = 0; s < scansPerTenant; s++) { + const scanId = `scan-${t}-${s}`; + const p = (async () => { + return service.finalizeIncident({ + incidentId: `inc-${t}-${s}`, + scanId, + tenantId, + repositoryUrl: `https://github.com/org/repo-${s}`, + commitHash: `commit-${s}`, + failureCategory: "TIMEOUT", + summary: "Concurrent scan test failed due to simulated timeout", + startedAt: Date.now() - 5000, + failedAt: Date.now(), + durationMs: 5000, + status: "failed", + events: [], + }); + })(); + finalizePromises.push(p); + } + } + + await Promise.all(finalizePromises); + + // Verify each tenant has exactly scansPerTenant incidents + for (let t = 0; t < tenantCount; t++) { + const tenantId = `tenant-${t}`; + const list = service.listIncidents(tenantId); + expect(list).toHaveLength(scansPerTenant); + + for (let s = 0; s < scansPerTenant; s++) { + const scanId = `scan-${t}-${s}`; + const incident = service.getIncident(tenantId, scanId); + expect(incident).toBeDefined(); + expect(incident!.events).toHaveLength(eventsPerScan); + // Verify chronological monotonicity + for (let i = 1; i < incident!.events.length; i++) { + expect(incident!.events[i].timestamp).toBeGreaterThanOrEqual( + incident!.events[i - 1].timestamp, + ); + } + + // Test export succeeds + const jsonExport = service.exportIncident(tenantId, scanId, "json"); + expect(jsonExport.metadata.eventCount).toBe(eventsPerScan); + } + } + }); + + it("recovers after interruption and duplicate finalize calls (idempotency)", () => { + const service = new IncidentTimelineExportService(); + const tenantId = "tenant-recovery"; + const scanId = "scan-recovered"; + + service.recordEvent({ + eventId: "e-1", + scanId, + tenantId, + timestamp: 100, + stage: "discovery", + severity: "info", + type: "scan_created", + actor: "api", + message: "Started", + }); + + const incident: ScanIncidentRecord = { + incidentId: "inc-rec-1", + scanId, + tenantId, + repositoryUrl: "https://github.com/test/repo", + commitHash: "hash123", + failureCategory: "LEASE_EXPIRED", + summary: "Worker crashed and lease expired", + startedAt: 100, + failedAt: 500, + durationMs: 400, + status: "failed", + events: [], + }; + + // First worker finalizes + service.finalizeIncident(incident); + // Duplicate call from retrying supervisor/worker + service.finalizeIncident(incident); + + const saved = service.getIncident(tenantId, scanId); + expect(saved).toBeDefined(); + expect(saved!.events).toHaveLength(1); + }); +}); diff --git a/apps/api/src/exports/incident-timeline/incident-timeline.spec.ts b/apps/api/src/exports/incident-timeline/incident-timeline.spec.ts new file mode 100644 index 00000000..db658986 --- /dev/null +++ b/apps/api/src/exports/incident-timeline/incident-timeline.spec.ts @@ -0,0 +1,319 @@ +import { + IncidentTimelineStore, + sanitizeIncidentText, +} from "./incident-timeline-store"; +import { IncidentTimelineExportService } from "./incident-timeline-export.service"; +import { ScanIncidentRecord, TimelineEvent } from "./incident-timeline-types"; + +describe("Incident Timeline Export (#1265)", () => { + describe("sanitizeIncidentText", () => { + it("redacts private keys from incident descriptions and error messages", () => { + const input = + "Deployer failed using key 0x4f3edf983ac636a65a842ce7c78d9aa706d3b113bce9c46f30d7d21715b23b1d for gas settlement"; + const sanitized = sanitizeIncidentText(input); + expect(sanitized).toBe( + "Deployer failed using key 0x[REDACTED_PRIVATE_KEY] for gas settlement", + ); + expect(sanitized).not.toContain("4f3edf983ac"); + }); + + it("redacts bearer tokens and secrets", () => { + const input = + "Authorization failed: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.e30.t-IDN context token=secret_token_123"; + const sanitized = sanitizeIncidentText(input); + expect(sanitized).toContain("Bearer [REDACTED_TOKEN]"); + expect(sanitized).toContain("token=[REDACTED]"); + expect(sanitized).not.toContain("secret_token_123"); + }); + }); + + describe("IncidentTimelineStore - Tenant Isolation & Bounded Memory", () => { + it("strictly isolates incidents between tenants", () => { + const store = new IncidentTimelineStore(); + const incidentTenantA: ScanIncidentRecord = { + incidentId: "inc-1", + scanId: "scan-100", + tenantId: "tenant-alpha", + repositoryUrl: "https://github.com/org/repo-a", + commitHash: "abc1234", + failureCategory: "PARSER_FAILURE", + summary: "Syntax error on contract parse", + startedAt: 1000, + failedAt: 2000, + durationMs: 1000, + status: "failed", + events: [], + }; + + store.finalizeIncident(incidentTenantA); + + // Tenant Alpha can see scan-100 + expect(store.getIncident("tenant-alpha", "scan-100")).toBeDefined(); + + // Tenant Beta cannot see scan-100 + expect(store.getIncident("tenant-beta", "scan-100")).toBeUndefined(); + expect(store.listIncidents("tenant-beta")).toHaveLength(0); + }); + + it("buffers events recorded before incident finalization and merges them", () => { + const store = new IncidentTimelineStore(); + + // Record events before incident is finalized + store.recordEvent({ + eventId: "evt-1", + scanId: "scan-200", + tenantId: "tenant-alpha", + timestamp: 1000, + stage: "discovery", + severity: "info", + type: "scan_created", + actor: "api", + message: "Scan created", + }); + + store.recordEvent({ + eventId: "evt-2", + scanId: "scan-200", + tenantId: "tenant-alpha", + timestamp: 1200, + stage: "parsing", + severity: "error", + type: "parser_error", + actor: "ast-worker", + message: + "Failed parsing Solidity AST with key 0x1111111111111111111111111111111111111111111111111111111111111111", + errorDetails: { + code: "SYNTAX_ERR", + category: "fatal", + message: "Unexpected identifier token=foo_secret", + retryable: false, + }, + }); + + // Finalize + const finalized = store.finalizeIncident({ + incidentId: "inc-200", + scanId: "scan-200", + tenantId: "tenant-alpha", + repositoryUrl: "https://github.com/org/repo-b", + commitHash: "def5678", + failureCategory: "PARSER_FAILURE", + summary: "AST parser failure", + startedAt: 900, + failedAt: 1300, + durationMs: 400, + status: "failed", + events: [], + }); + + expect(finalized.events).toHaveLength(2); + expect(finalized.events[0].eventId).toBe("evt-1"); + expect(finalized.events[1].eventId).toBe("evt-2"); + expect(finalized.events[1].message).toContain("0x[REDACTED_PRIVATE_KEY]"); + expect(finalized.events[1].errorDetails?.message).toContain( + "token=[REDACTED]", + ); + }); + + it("bounds maximum incidents per tenant by evicting oldest", () => { + const store = new IncidentTimelineStore({ maxIncidentsPerTenant: 2 }); + + store.finalizeIncident({ + incidentId: "inc-1", + scanId: "scan-1", + tenantId: "tenant-a", + repositoryUrl: "repo", + commitHash: "c1", + failureCategory: "TIMEOUT", + summary: "Timeout 1", + startedAt: 1, + failedAt: 2, + durationMs: 1, + status: "failed", + events: [], + }); + + store.finalizeIncident({ + incidentId: "inc-2", + scanId: "scan-2", + tenantId: "tenant-a", + repositoryUrl: "repo", + commitHash: "c2", + failureCategory: "TIMEOUT", + summary: "Timeout 2", + startedAt: 3, + failedAt: 4, + durationMs: 1, + status: "failed", + events: [], + }); + + expect(store.listIncidents("tenant-a")).toHaveLength(2); + + // Add a third -> oldest (scan-1) should be evicted + store.finalizeIncident({ + incidentId: "inc-3", + scanId: "scan-3", + tenantId: "tenant-a", + repositoryUrl: "repo", + commitHash: "c3", + failureCategory: "TIMEOUT", + summary: "Timeout 3", + startedAt: 5, + failedAt: 6, + durationMs: 1, + status: "failed", + events: [], + }); + + const list = store.listIncidents("tenant-a"); + expect(list).toHaveLength(2); + expect(store.getIncident("tenant-a", "scan-1")).toBeUndefined(); + expect(store.getIncident("tenant-a", "scan-2")).toBeDefined(); + expect(store.getIncident("tenant-a", "scan-3")).toBeDefined(); + }); + + it("validates input arguments", () => { + const store = new IncidentTimelineStore(); + expect(() => store.getIncident("", "scan-1")).toThrow(TypeError); + expect(() => store.getIncident("tenant-a", "")).toThrow(TypeError); + expect(() => store.listIncidents("")).toThrow(TypeError); + }); + }); + + describe("IncidentTimelineExportService - Formats", () => { + let service: IncidentTimelineExportService; + let sampleIncident: ScanIncidentRecord; + + beforeEach(() => { + service = new IncidentTimelineExportService(); + sampleIncident = { + incidentId: "inc-999", + scanId: "scan-999", + tenantId: "tenant-prod", + repositoryUrl: "https://github.com/MDTechLabs/GasGuard", + commitHash: "abc987654321", + failureCategory: "RESOURCE_EXHAUSTION", + summary: "Worker exceeded memory limit during AST generation", + startedAt: 1697620000000, + failedAt: 1697620045000, + durationMs: 45000, + status: "failed", + workerLifecycle: { + workerId: "worker-pool-node-3", + leaseFencingToken: 42, + leaseAcquiredAt: 1697620001000, + }, + diagnosticsCount: { + recovered: 4, + unrecovered: 1, + total: 5, + }, + events: [ + { + eventId: "evt-01", + scanId: "scan-999", + tenantId: "tenant-prod", + timestamp: 1697620001000, + stage: "discovery", + severity: "info", + type: "scan_created", + actor: "api", + message: "Scan job initialized", + }, + { + eventId: "evt-02", + scanId: "scan-999", + tenantId: "tenant-prod", + timestamp: 1697620010000, + stage: "parsing", + severity: "warn", + type: "parser_warning", + actor: "ast-worker", + message: "Deep AST recursion detected", + }, + { + eventId: "evt-03", + scanId: "scan-999", + tenantId: "tenant-prod", + timestamp: 1697620045000, + stage: "parsing", + severity: "critical", + type: "resource_limit_exceeded", + actor: "worker-pool-node-3", + message: "OOM killer invoked on child process", + errorDetails: { + code: "ERR_WORKER_OOM", + category: "resource_limit", + message: "Heap allocation out of memory", + retryable: false, + stackExcerpt: "Error: Heap limit exceeded at Parser.parse", + }, + }, + ], + }; + service.finalizeIncident(sampleIncident); + }); + + it("exports incident as JSON format", () => { + const result = service.exportIncident("tenant-prod", "scan-999", "json"); + expect(result.format).toBe("json"); + expect(result.contentType).toBe("application/json"); + expect(result.filename).toBe("incident-scan-999-timeline.json"); + + const parsed = JSON.parse(result.data); + expect(parsed.incidentId).toBe("inc-999"); + expect(parsed.scanId).toBe("scan-999"); + expect(parsed.failureCategory).toBe("RESOURCE_EXHAUSTION"); + expect(parsed.workerLifecycle.workerId).toBe("worker-pool-node-3"); + expect(parsed.timelineEvents).toHaveLength(3); + expect(parsed.timelineEvents[2].errorDetails.code).toBe("ERR_WORKER_OOM"); + }); + + it("exports incident as Markdown format with full post-mortem sections", () => { + const result = service.exportIncident( + "tenant-prod", + "scan-999", + "markdown", + ); + expect(result.format).toBe("markdown"); + expect(result.contentType).toBe("text/markdown"); + expect(result.filename).toBe("incident-scan-999-timeline.md"); + + expect(result.data).toContain("# Incident Report: Scan scan-999"); + expect(result.data).toContain("RESOURCE_EXHAUSTION"); + expect(result.data).toContain("worker-pool-node-3"); + expect(result.data).toContain("Lease Fencing Token"); + expect(result.data).toContain("Diagnostics Summary"); + expect(result.data).toContain( + "| Timestamp (UTC) | Stage | Severity | Event Type | Actor | Message |", + ); + expect(result.data).toContain("OOM killer invoked on child process"); + expect(result.data).toContain("ERR_WORKER_OOM"); + }); + + it("exports incident as RFC 4180 CSV format", () => { + const result = service.exportIncident("tenant-prod", "scan-999", "csv"); + expect(result.format).toBe("csv"); + expect(result.contentType).toBe("text/csv"); + expect(result.filename).toBe("incident-scan-999-timeline.csv"); + + const lines = result.data.split("\r\n"); + expect(lines[0]).toBe( + "event_id,timestamp_iso,stage,severity,type,actor,message,error_code,retryable", + ); + expect(lines).toHaveLength(4); // header + 3 events + expect(lines[1]).toContain("evt-01"); + expect(lines[3]).toContain("ERR_WORKER_OOM"); + }); + + it("throws when exporting non-existent incident or wrong tenant", () => { + expect(() => + service.exportIncident("other-tenant", "scan-999", "json"), + ).toThrow(/not found under tenant other-tenant/); + expect(() => + service.exportIncident("tenant-prod", "non-existent-scan", "json"), + ).toThrow(/not found under tenant tenant-prod/); + }); + }); +}); diff --git a/apps/api/src/exports/incident-timeline/index.ts b/apps/api/src/exports/incident-timeline/index.ts new file mode 100644 index 00000000..6bed7cf9 --- /dev/null +++ b/apps/api/src/exports/incident-timeline/index.ts @@ -0,0 +1,3 @@ +export * from "./incident-timeline-types"; +export * from "./incident-timeline-store"; +export * from "./incident-timeline-export.service"; diff --git a/apps/api/src/scanner/parser-alerting/__fixtures__/parser-alert-regression-snapshot.json b/apps/api/src/scanner/parser-alerting/__fixtures__/parser-alert-regression-snapshot.json new file mode 100644 index 00000000..59a949a4 --- /dev/null +++ b/apps/api/src/scanner/parser-alerting/__fixtures__/parser-alert-regression-snapshot.json @@ -0,0 +1,14 @@ +{ + "description": "Deterministic regression snapshot for repeated parser failure alert evaluation", + "expectedAlertsTriggered": 1, + "expectedAlertsResolved": 1, + "expectedTriggerScanId": "scan-commit-4", + "expectedConsecutiveFailuresAtTrigger": 3, + "expectedSeverityAtTrigger": "warning", + "expectedResolvingScanId": "scan-commit-6", + "expectedDiagnosticCodes": { + "unexpected_token": 2, + "invalid_pragma": 1, + "recursion_limit_exceeded": 1 + } +} diff --git a/apps/api/src/scanner/parser-alerting/__fixtures__/repeated-parser-failures-repo.json b/apps/api/src/scanner/parser-alerting/__fixtures__/repeated-parser-failures-repo.json new file mode 100644 index 00000000..6810db5f --- /dev/null +++ b/apps/api/src/scanner/parser-alerting/__fixtures__/repeated-parser-failures-repo.json @@ -0,0 +1,63 @@ +{ + "repository": { + "url": "https://github.com/MDTechLabs/defi-vault-protocol", + "defaultBranch": "main", + "languages": ["solidity", "vyper"] + }, + "scans": [ + { + "scanId": "scan-commit-1", + "commitHash": "b1a2c3d4e5f60001", + "language": "solidity", + "parserName": "solidity-ast-parser-v2", + "hasFailed": false, + "filesScanned": 12 + }, + { + "scanId": "scan-commit-2", + "commitHash": "b1a2c3d4e5f60002", + "language": "solidity", + "parserName": "solidity-ast-parser-v2", + "hasFailed": true, + "failureCategory": "syntax_error", + "diagnosticCodes": ["unexpected_token", "invalid_pragma"], + "errorSummary": "Solidity syntax error on custom transient storage opcode" + }, + { + "scanId": "scan-commit-3", + "commitHash": "b1a2c3d4e5f60003", + "language": "solidity", + "parserName": "solidity-ast-parser-v2", + "hasFailed": true, + "failureCategory": "syntax_error", + "diagnosticCodes": ["unexpected_token"], + "errorSummary": "Syntax error persisting in VaultStorage.sol" + }, + { + "scanId": "scan-commit-4", + "commitHash": "b1a2c3d4e5f60004", + "language": "solidity", + "parserName": "solidity-ast-parser-v2", + "hasFailed": true, + "failureCategory": "recursion_limit", + "diagnosticCodes": ["recursion_limit_exceeded"], + "errorSummary": "Parser worker reached maximum recursion depth" + }, + { + "scanId": "scan-commit-5", + "commitHash": "b1a2c3d4e5f60005", + "language": "solidity", + "parserName": "solidity-ast-parser-v2", + "hasFailed": false, + "filesScanned": 14 + }, + { + "scanId": "scan-commit-6", + "commitHash": "b1a2c3d4e5f60006", + "language": "solidity", + "parserName": "solidity-ast-parser-v2", + "hasFailed": false, + "filesScanned": 14 + } + ] +} diff --git a/apps/api/src/scanner/parser-alerting/index.ts b/apps/api/src/scanner/parser-alerting/index.ts new file mode 100644 index 00000000..37f85084 --- /dev/null +++ b/apps/api/src/scanner/parser-alerting/index.ts @@ -0,0 +1,3 @@ +export * from "./parser-failure-alert-types"; +export * from "./parser-failure-alert-tracker"; +export * from "./parser-failure-alert.service"; diff --git a/apps/api/src/scanner/parser-alerting/parser-failure-alert-tracker.ts b/apps/api/src/scanner/parser-alerting/parser-failure-alert-tracker.ts new file mode 100644 index 00000000..57680cce --- /dev/null +++ b/apps/api/src/scanner/parser-alerting/parser-failure-alert-tracker.ts @@ -0,0 +1,289 @@ +import { + DEFAULT_PARSER_ALERT_CONFIG, + ParserAlert, + ParserAlertConfig, + ParserLanguage, + ParserScanRecord, +} from "./parser-failure-alert-types"; + +export function resolveParserAlertConfig( + overrides?: Partial, +): ParserAlertConfig { + if (!overrides) { + return { ...DEFAULT_PARSER_ALERT_CONFIG }; + } + + const resolved = { ...DEFAULT_PARSER_ALERT_CONFIG, ...overrides }; + + const positiveIntegers: Array = [ + "consecutiveFailuresThreshold", + "windowSize", + "minScansForRatio", + "cooldownPeriodMs", + "autoResolveConsecutiveSuccesses", + ]; + + for (const field of positiveIntegers) { + const val = resolved[field]; + if (typeof val !== "number" || !Number.isInteger(val) || val <= 0) { + throw new RangeError(`${field} must be a positive integer`); + } + } + + if ( + typeof resolved.failureRatioThreshold !== "number" || + Number.isNaN(resolved.failureRatioThreshold) || + resolved.failureRatioThreshold <= 0 || + resolved.failureRatioThreshold > 1 + ) { + throw new RangeError( + "failureRatioThreshold must be a number between 0 and 1", + ); + } + + if (resolved.minScansForRatio > resolved.windowSize) { + throw new RangeError("minScansForRatio cannot exceed windowSize"); + } + + return resolved; +} + +interface TrackedCohortState { + tenantId: string; + repositoryUrl: string; + language: ParserLanguage; + consecutiveFailures: number; + consecutiveSuccesses: number; + history: ParserScanRecord[]; + activeAlert?: ParserAlert; + diagnosticCodes: Record; +} + +export class ParserFailureAlertTracker { + private readonly config: ParserAlertConfig; + private readonly clock: () => number; + // Key: `${tenantId}::${repositoryUrl}::${language}` + private readonly cohorts: Map = new Map(); + private readonly alertHistory: ParserAlert[] = []; + + constructor( + config?: Partial, + clock: () => number = Date.now, + ) { + this.config = resolveParserAlertConfig(config); + this.clock = clock; + } + + getConfig(): ParserAlertConfig { + return { ...this.config }; + } + + /** + * Records a parser scan execution outcome and evaluates alerting rules deterministically. + */ + recordScan(record: ParserScanRecord): { + alertTriggered?: ParserAlert; + alertResolved?: ParserAlert; + isSuppressed: boolean; + } { + this.validateRecord(record); + + const key = this.buildKey( + record.tenantId, + record.repositoryUrl, + record.language, + ); + let cohort = this.cohorts.get(key); + if (!cohort) { + cohort = { + tenantId: record.tenantId, + repositoryUrl: record.repositoryUrl, + language: record.language, + consecutiveFailures: 0, + consecutiveSuccesses: 0, + history: [], + diagnosticCodes: {}, + }; + this.cohorts.set(key, cohort); + } + + const now = this.clock(); + + // Append to sliding window + cohort.history.push(record); + if (cohort.history.length > this.config.windowSize) { + cohort.history.shift(); + } + + // Aggregate diagnostic codes + if (record.diagnosticCodes) { + for (const code of record.diagnosticCodes) { + if (code && typeof code === "string") { + cohort.diagnosticCodes[code] = + (cohort.diagnosticCodes[code] ?? 0) + 1; + } + } + } + + if (!record.hasFailed) { + // Successful parse + cohort.consecutiveSuccesses += 1; + cohort.consecutiveFailures = 0; + + // Check auto-resolution of active alert + if ( + cohort.activeAlert && + cohort.consecutiveSuccesses >= + this.config.autoResolveConsecutiveSuccesses + ) { + const resolvedAlert: ParserAlert = { + ...cohort.activeAlert, + status: "resolved", + resolvedAt: now, + }; + const idx = this.alertHistory.findIndex( + (a) => a.alertId === cohort.activeAlert?.alertId, + ); + if (idx >= 0) { + this.alertHistory[idx] = resolvedAlert; + } else { + this.alertHistory.push(resolvedAlert); + } + cohort.activeAlert = undefined; + return { alertResolved: resolvedAlert, isSuppressed: false }; + } + + return { isSuppressed: false }; + } + + // Failed parse + cohort.consecutiveFailures += 1; + cohort.consecutiveSuccesses = 0; + + const failedCount = cohort.history.filter((s) => s.hasFailed).length; + const failureRatio = failedCount / cohort.history.length; + + let triggerReason: string | undefined; + if ( + cohort.consecutiveFailures >= this.config.consecutiveFailuresThreshold + ) { + triggerReason = `Consecutive parser failures (${cohort.consecutiveFailures}) reached threshold (${this.config.consecutiveFailuresThreshold})`; + } else if ( + cohort.history.length >= this.config.minScansForRatio && + failureRatio >= this.config.failureRatioThreshold + ) { + triggerReason = `Parser failure ratio (${Math.round(failureRatio * 100)}%) breached threshold (${Math.round(this.config.failureRatioThreshold * 100)}%) across ${cohort.history.length} scans`; + } + + if (!triggerReason) { + return { isSuppressed: false }; + } + + const severity = + cohort.consecutiveFailures >= this.config.consecutiveFailuresThreshold * 2 + ? "critical" + : "warning"; + + // Check if active alert already exists (cooldown suppression check) + if (cohort.activeAlert) { + const elapsedSinceTrigger = now - cohort.activeAlert.lastTriggeredAt; + if (elapsedSinceTrigger < this.config.cooldownPeriodMs) { + // Cooldown active: suppress notification flood + cohort.activeAlert.notificationCount += 1; + cohort.activeAlert.consecutiveFailures = cohort.consecutiveFailures; + cohort.activeAlert.failureRatio = Math.round(failureRatio * 100) / 100; + return { isSuppressed: true }; + } + + // Cooldown expired: re-dispatch alert with updated stats + cohort.activeAlert.lastTriggeredAt = now; + cohort.activeAlert.consecutiveFailures = cohort.consecutiveFailures; + cohort.activeAlert.failureRatio = Math.round(failureRatio * 100) / 100; + cohort.activeAlert.notificationCount += 1; + cohort.activeAlert.severity = severity; + cohort.activeAlert.triggerReason = triggerReason; + + return { alertTriggered: cohort.activeAlert, isSuppressed: false }; + } + + // Brand new alert + const alertId = `alert-${record.tenantId}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 6)}`; + const newAlert: ParserAlert = { + alertId, + tenantId: record.tenantId, + repositoryUrl: record.repositoryUrl, + language: record.language, + parserName: record.parserName, + severity, + status: "active", + triggerReason, + consecutiveFailures: cohort.consecutiveFailures, + failureRatio: Math.round(failureRatio * 100) / 100, + recentScansCount: cohort.history.length, + diagnosticCodesSummary: { ...cohort.diagnosticCodes }, + firstTriggeredAt: now, + lastTriggeredAt: now, + notificationCount: 1, + }; + + cohort.activeAlert = newAlert; + this.alertHistory.push(newAlert); + + return { alertTriggered: newAlert, isSuppressed: false }; + } + + getActiveAlerts(tenantId?: string): ParserAlert[] { + const alerts: ParserAlert[] = []; + for (const cohort of this.cohorts.values()) { + if (cohort.activeAlert) { + if (!tenantId || cohort.tenantId === tenantId) { + alerts.push(cohort.activeAlert); + } + } + } + return alerts; + } + + getAlertHistory(tenantId: string): ParserAlert[] { + if (!tenantId || typeof tenantId !== "string") { + throw new TypeError("tenantId must be a non-empty string"); + } + return this.alertHistory.filter((a) => a.tenantId === tenantId); + } + + clear(): void { + this.cohorts.clear(); + this.alertHistory.length = 0; + } + + private buildKey( + tenantId: string, + repositoryUrl: string, + language: string, + ): string { + return `${tenantId.trim()}::${repositoryUrl.trim()}::${language.trim()}`; + } + + private validateRecord(record: ParserScanRecord): void { + if (!record || typeof record !== "object") { + throw new TypeError("record must be a valid object"); + } + if ( + !record.tenantId || + typeof record.tenantId !== "string" || + record.tenantId.trim() === "" + ) { + throw new TypeError("record.tenantId must be a non-empty string"); + } + if ( + !record.repositoryUrl || + typeof record.repositoryUrl !== "string" || + record.repositoryUrl.trim() === "" + ) { + throw new TypeError("record.repositoryUrl must be a non-empty string"); + } + if (typeof record.hasFailed !== "boolean") { + throw new TypeError("record.hasFailed must be a boolean"); + } + } +} diff --git a/apps/api/src/scanner/parser-alerting/parser-failure-alert-types.ts b/apps/api/src/scanner/parser-alerting/parser-failure-alert-types.ts new file mode 100644 index 00000000..fe08166f --- /dev/null +++ b/apps/api/src/scanner/parser-alerting/parser-failure-alert-types.ts @@ -0,0 +1,93 @@ +/** + * Parser Failure Alerting Types (#1262). + * + * Defines configuration, tracking, and alerting models for repeated AST parser + * failures across repositories under strict tenant isolation. + */ + +export type ParserLanguage = + | "solidity" + | "vyper" + | "rust" + | "soroban" + | "unsupported"; + +export type ParserFailureCategory = + | "syntax_error" + | "recursion_limit" + | "unsupported_syntax" + | "worker_crash" + | "timeout"; + +export type AlertSeverity = "warning" | "critical"; + +export type AlertStatus = "active" | "resolved" | "suppressed"; + +export interface ParserScanRecord { + scanId: string; + tenantId: string; + repositoryUrl: string; + commitHash: string; + language: ParserLanguage; + parserName: string; + timestamp: number; + hasFailed: boolean; + failureCategory?: ParserFailureCategory; + diagnosticCodes?: string[]; + errorSummary?: string; +} + +export interface ParserAlertConfig { + /** Consecutive parser failures on repository before raising alert (e.g. 3). */ + consecutiveFailuresThreshold: number; + /** Sliding window failure ratio [0, 1] that triggers alert (e.g. 0.4 = 40%). */ + failureRatioThreshold: number; + /** Number of recent scans kept in sliding window for ratio evaluation (e.g. 10). */ + windowSize: number; + /** Minimum scans in window before evaluating ratio threshold (e.g. 4). */ + minScansForRatio: number; + /** Quiet period in ms between alert dispatches for same repository (e.g. 300,000ms / 5 min). */ + cooldownPeriodMs: number; + /** Consecutive successful parses required to automatically resolve active alert (e.g. 2). */ + autoResolveConsecutiveSuccesses: number; +} + +export const DEFAULT_PARSER_ALERT_CONFIG: ParserAlertConfig = { + consecutiveFailuresThreshold: 3, + failureRatioThreshold: 0.4, + windowSize: 10, + minScansForRatio: 4, + cooldownPeriodMs: 5 * 60 * 1000, // 5 minutes + autoResolveConsecutiveSuccesses: 2, +}; + +export interface ParserAlert { + alertId: string; + tenantId: string; + repositoryUrl: string; + language: ParserLanguage; + parserName: string; + severity: AlertSeverity; + status: AlertStatus; + triggerReason: string; + consecutiveFailures: number; + failureRatio: number; + recentScansCount: number; + diagnosticCodesSummary: Record; + firstTriggeredAt: number; + lastTriggeredAt: number; + resolvedAt?: number; + notificationCount: number; +} + +export interface ParserAlertMetrics { + totalScansEvaluated: number; + totalFailuresRecorded: number; + activeAlertsCount: number; + totalAlertsDispatched: number; + totalAlertsResolved: number; +} + +export interface AlertNotificationHandler { + onAlert(alert: ParserAlert): Promise | void; +} diff --git a/apps/api/src/scanner/parser-alerting/parser-failure-alert.integration.spec.ts b/apps/api/src/scanner/parser-alerting/parser-failure-alert.integration.spec.ts new file mode 100644 index 00000000..cdd06f4c --- /dev/null +++ b/apps/api/src/scanner/parser-alerting/parser-failure-alert.integration.spec.ts @@ -0,0 +1,150 @@ +import * as fs from "fs"; +import * as path from "path"; +import { ParserFailureAlertService } from "./parser-failure-alert.service"; +import { ParserAlert } from "./parser-failure-alert-types"; + +describe("Parser Failure Alerting Fixtures & Regression Integration (#1262)", () => { + it("processes realistic repository scan fixture and matches regression snapshot deterministically", async () => { + const fixturePath = path.join( + __dirname, + "__fixtures__", + "repeated-parser-failures-repo.json", + ); + const regressionPath = path.join( + __dirname, + "__fixtures__", + "parser-alert-regression-snapshot.json", + ); + + const fixtureData = JSON.parse(fs.readFileSync(fixturePath, "utf-8")); + const regressionSnapshot = JSON.parse( + fs.readFileSync(regressionPath, "utf-8"), + ); + + const triggeredAlerts: ParserAlert[] = []; + const resolvedAlerts: ParserAlert[] = []; + + let mockClock = 1700000000000; + const service = new ParserFailureAlertService( + { + consecutiveFailuresThreshold: 3, + autoResolveConsecutiveSuccesses: 2, + cooldownPeriodMs: 60000, + }, + () => mockClock, + ); + + service.registerHandler({ + onAlert: (alert) => { + if (alert.status === "active") { + triggeredAlerts.push(alert); + } else if (alert.status === "resolved") { + resolvedAlerts.push(alert); + } + }, + }); + + const tenantId = "tenant-defi-vault"; + + for (const scan of fixtureData.scans) { + mockClock += 10000; + await service.recordScan({ + scanId: scan.scanId, + tenantId, + repositoryUrl: fixtureData.repository.url, + commitHash: scan.commitHash, + language: scan.language, + parserName: scan.parserName, + timestamp: mockClock, + hasFailed: scan.hasFailed, + failureCategory: scan.failureCategory, + diagnosticCodes: scan.diagnosticCodes, + errorSummary: scan.errorSummary, + }); + } + + // Assert regression snapshot properties + expect(triggeredAlerts).toHaveLength( + regressionSnapshot.expectedAlertsTriggered, + ); + expect(resolvedAlerts).toHaveLength( + regressionSnapshot.expectedAlertsResolved, + ); + + const triggered = triggeredAlerts[0]; + expect(triggered.consecutiveFailures).toBe( + regressionSnapshot.expectedConsecutiveFailuresAtTrigger, + ); + expect(triggered.severity).toBe( + regressionSnapshot.expectedSeverityAtTrigger, + ); + expect(triggered.diagnosticCodesSummary).toEqual( + regressionSnapshot.expectedDiagnosticCodes, + ); + + const resolved = resolvedAlerts[0]; + expect(resolved.status).toBe("resolved"); + expect(resolved.resolvedAt).toBeDefined(); + + const metrics = service.getMetrics(); + expect(metrics.totalScansEvaluated).toBe(fixtureData.scans.length); + expect(metrics.activeAlertsCount).toBe(0); + expect(metrics.totalAlertsDispatched).toBe(1); + expect(metrics.totalAlertsResolved).toBe(1); + }); + + it("maintains determinism across repeated executions with identical inputs", async () => { + const runSimulation = async () => { + let time = 1000; + const service = new ParserFailureAlertService( + { + consecutiveFailuresThreshold: 2, + autoResolveConsecutiveSuccesses: 2, + }, + () => time, + ); + + const records = [ + { scanId: "s1", fail: true, codes: ["err1"] }, + { scanId: "s2", fail: true, codes: ["err2"] }, + { scanId: "s3", fail: false, codes: [] }, + { scanId: "s4", fail: false, codes: [] }, + ]; + + const events: string[] = []; + service.registerHandler({ + onAlert: (a) => { + events.push( + `${a.status}:${a.consecutiveFailures}:${JSON.stringify(a.diagnosticCodesSummary)}`, + ); + }, + }); + + for (const r of records) { + time += 100; + await service.recordScan({ + scanId: r.scanId, + tenantId: "tenant-det", + repositoryUrl: "https://repo", + commitHash: "hash", + language: "solidity", + parserName: "p1", + timestamp: time, + hasFailed: r.fail, + diagnosticCodes: r.codes, + }); + } + + return events; + }; + + const run1 = await runSimulation(); + const run2 = await runSimulation(); + + expect(run1).toEqual(run2); + expect(run1).toEqual([ + 'active:2:{"err1":1,"err2":1}', + 'resolved:2:{"err1":1,"err2":1}', + ]); + }); +}); diff --git a/apps/api/src/scanner/parser-alerting/parser-failure-alert.service.ts b/apps/api/src/scanner/parser-alerting/parser-failure-alert.service.ts new file mode 100644 index 00000000..7d2b7e68 --- /dev/null +++ b/apps/api/src/scanner/parser-alerting/parser-failure-alert.service.ts @@ -0,0 +1,97 @@ +import { + AlertNotificationHandler, + ParserAlert, + ParserAlertConfig, + ParserAlertMetrics, + ParserScanRecord, +} from "./parser-failure-alert-types"; +import { ParserFailureAlertTracker } from "./parser-failure-alert-tracker"; + +export class ParserFailureAlertService { + private readonly tracker: ParserFailureAlertTracker; + private readonly handlers: AlertNotificationHandler[] = []; + private totalScansEvaluated = 0; + private totalFailuresRecorded = 0; + private totalAlertsDispatched = 0; + private totalAlertsResolved = 0; + + constructor( + config?: Partial, + clock: () => number = Date.now, + ) { + this.tracker = new ParserFailureAlertTracker(config, clock); + } + + getTracker(): ParserFailureAlertTracker { + return this.tracker; + } + + registerHandler(handler: AlertNotificationHandler): void { + if (!handler || typeof handler.onAlert !== "function") { + throw new TypeError("handler must implement onAlert function"); + } + this.handlers.push(handler); + } + + async recordScan(record: ParserScanRecord): Promise<{ + alertTriggered?: ParserAlert; + alertResolved?: ParserAlert; + isSuppressed: boolean; + }> { + this.totalScansEvaluated += 1; + if (record.hasFailed) { + this.totalFailuresRecorded += 1; + } + + const outcome = this.tracker.recordScan(record); + + if (outcome.alertTriggered && !outcome.isSuppressed) { + this.totalAlertsDispatched += 1; + await this.dispatchToHandlers(outcome.alertTriggered); + } + + if (outcome.alertResolved) { + this.totalAlertsResolved += 1; + await this.dispatchToHandlers(outcome.alertResolved); + } + + return outcome; + } + + getActiveAlerts(tenantId?: string): ParserAlert[] { + return this.tracker.getActiveAlerts(tenantId); + } + + getAlertHistory(tenantId: string): ParserAlert[] { + return this.tracker.getAlertHistory(tenantId); + } + + getMetrics(): ParserAlertMetrics { + return { + totalScansEvaluated: this.totalScansEvaluated, + totalFailuresRecorded: this.totalFailuresRecorded, + activeAlertsCount: this.tracker.getActiveAlerts().length, + totalAlertsDispatched: this.totalAlertsDispatched, + totalAlertsResolved: this.totalAlertsResolved, + }; + } + + clear(): void { + this.tracker.clear(); + this.totalScansEvaluated = 0; + this.totalFailuresRecorded = 0; + this.totalAlertsDispatched = 0; + this.totalAlertsResolved = 0; + } + + private async dispatchToHandlers(alert: ParserAlert): Promise { + for (const handler of this.handlers) { + try { + await handler.onAlert(alert); + } catch (err) { + // Defensive: sink failure must never throw to caller or stop other sinks + console.error("[ParserAlertService] Handler dispatch error:", err); + } + } + } +} diff --git a/apps/api/src/scanner/parser-alerting/parser-failure-alert.spec.ts b/apps/api/src/scanner/parser-alerting/parser-failure-alert.spec.ts new file mode 100644 index 00000000..ed28818e --- /dev/null +++ b/apps/api/src/scanner/parser-alerting/parser-failure-alert.spec.ts @@ -0,0 +1,322 @@ +import { + ParserFailureAlertTracker, + resolveParserAlertConfig, +} from "./parser-failure-alert-tracker"; +import { ParserFailureAlertService } from "./parser-failure-alert.service"; +import { ParserAlert, ParserScanRecord } from "./parser-failure-alert-types"; + +describe("Repeated Parser Failure Alerting (#1262)", () => { + describe("resolveParserAlertConfig", () => { + it("returns defaults when no overrides provided", () => { + const config = resolveParserAlertConfig(); + expect(config.consecutiveFailuresThreshold).toBe(3); + expect(config.failureRatioThreshold).toBe(0.4); + expect(config.windowSize).toBe(10); + expect(config.cooldownPeriodMs).toBe(300_000); + expect(config.autoResolveConsecutiveSuccesses).toBe(2); + }); + + it("rejects non-positive integers", () => { + expect(() => + resolveParserAlertConfig({ consecutiveFailuresThreshold: 0 }), + ).toThrow(RangeError); + expect(() => + resolveParserAlertConfig({ consecutiveFailuresThreshold: -2 }), + ).toThrow(RangeError); + expect(() => resolveParserAlertConfig({ windowSize: 0 })).toThrow( + RangeError, + ); + }); + + it("rejects invalid failure ratio thresholds", () => { + expect(() => + resolveParserAlertConfig({ failureRatioThreshold: 0 }), + ).toThrow(RangeError); + expect(() => + resolveParserAlertConfig({ failureRatioThreshold: 1.5 }), + ).toThrow(RangeError); + expect(() => + resolveParserAlertConfig({ failureRatioThreshold: -0.1 }), + ).toThrow(RangeError); + }); + + it("rejects minScansForRatio greater than windowSize", () => { + expect(() => + resolveParserAlertConfig({ windowSize: 5, minScansForRatio: 10 }), + ).toThrow(RangeError); + }); + }); + + describe("ParserFailureAlertTracker - Consecutive Failures", () => { + let tracker: ParserFailureAlertTracker; + let mockTime: number; + + beforeEach(() => { + mockTime = 100000; + tracker = new ParserFailureAlertTracker( + { + consecutiveFailuresThreshold: 3, + cooldownPeriodMs: 60000, + autoResolveConsecutiveSuccesses: 2, + }, + () => mockTime, + ); + }); + + it("does not trigger alert before reaching consecutive failures threshold", () => { + const record1: ParserScanRecord = { + scanId: "s-1", + tenantId: "tenant-a", + repositoryUrl: "https://github.com/org/repo", + commitHash: "c1", + language: "solidity", + parserName: "solidity-ast", + timestamp: mockTime, + hasFailed: true, + failureCategory: "syntax_error", + diagnosticCodes: ["unexpected_token"], + }; + + const res1 = tracker.recordScan(record1); + expect(res1.alertTriggered).toBeUndefined(); + expect(tracker.getActiveAlerts()).toHaveLength(0); + + const record2 = { ...record1, scanId: "s-2", commitHash: "c2" }; + const res2 = tracker.recordScan(record2); + expect(res2.alertTriggered).toBeUndefined(); + }); + + it("triggers alert on third consecutive failure with diagnostic code summary", () => { + const base: ParserScanRecord = { + scanId: "s-1", + tenantId: "tenant-a", + repositoryUrl: "https://github.com/org/repo", + commitHash: "c1", + language: "solidity", + parserName: "solidity-ast", + timestamp: mockTime, + hasFailed: true, + failureCategory: "syntax_error", + diagnosticCodes: ["unexpected_token"], + }; + + tracker.recordScan({ ...base, scanId: "s-1" }); + tracker.recordScan({ + ...base, + scanId: "s-2", + diagnosticCodes: ["recursion_limit"], + }); + + const res3 = tracker.recordScan({ ...base, scanId: "s-3" }); + expect(res3.alertTriggered).toBeDefined(); + expect(res3.isSuppressed).toBe(false); + + const alert = res3.alertTriggered!; + expect(alert.status).toBe("active"); + expect(alert.severity).toBe("warning"); + expect(alert.consecutiveFailures).toBe(3); + expect(alert.diagnosticCodesSummary).toEqual({ + unexpected_token: 2, + recursion_limit: 1, + }); + + expect(tracker.getActiveAlerts("tenant-a")).toHaveLength(1); + }); + + it("suppresses repeated alert notifications while inside cooldown period", () => { + const base: ParserScanRecord = { + scanId: "s-1", + tenantId: "tenant-a", + repositoryUrl: "https://github.com/org/repo", + commitHash: "c1", + language: "solidity", + parserName: "solidity-ast", + timestamp: mockTime, + hasFailed: true, + }; + + tracker.recordScan({ ...base, scanId: "s-1" }); + tracker.recordScan({ ...base, scanId: "s-2" }); + const res3 = tracker.recordScan({ ...base, scanId: "s-3" }); + expect(res3.alertTriggered).toBeDefined(); + + // Next failure arrives 10 seconds later (cooldown is 60s) + mockTime += 10000; + const res4 = tracker.recordScan({ ...base, scanId: "s-4" }); + expect(res4.isSuppressed).toBe(true); + expect(res4.alertTriggered).toBeUndefined(); + + // Active alert has updated notification count and consecutive failure count + const active = tracker.getActiveAlerts("tenant-a")[0]; + expect(active.notificationCount).toBe(2); + expect(active.consecutiveFailures).toBe(4); + }); + + it("re-dispatches alert after cooldown period expires and escalates severity", () => { + const base: ParserScanRecord = { + scanId: "s-1", + tenantId: "tenant-a", + repositoryUrl: "https://github.com/org/repo", + commitHash: "c1", + language: "solidity", + parserName: "solidity-ast", + timestamp: mockTime, + hasFailed: true, + }; + + tracker.recordScan({ ...base, scanId: "s-1" }); + tracker.recordScan({ ...base, scanId: "s-2" }); + tracker.recordScan({ ...base, scanId: "s-3" }); + + // Jump 70 seconds ahead (cooldown is 60s), failure 4 + mockTime += 70000; + tracker.recordScan({ ...base, scanId: "s-4" }); + tracker.recordScan({ ...base, scanId: "s-5" }); + + // Jump another 70 seconds ahead, failure 6 (>= consecutiveFailuresThreshold * 2) + mockTime += 70000; + const res6 = tracker.recordScan({ ...base, scanId: "s-6" }); + expect(res6.isSuppressed).toBe(false); + expect(res6.alertTriggered).toBeDefined(); + expect(res6.alertTriggered!.severity).toBe("critical"); + }); + + it("automatically resolves alert after configured consecutive successes", () => { + const base: ParserScanRecord = { + scanId: "s-1", + tenantId: "tenant-a", + repositoryUrl: "https://github.com/org/repo", + commitHash: "c1", + language: "solidity", + parserName: "solidity-ast", + timestamp: mockTime, + hasFailed: true, + }; + + tracker.recordScan({ ...base, scanId: "s-1" }); + tracker.recordScan({ ...base, scanId: "s-2" }); + tracker.recordScan({ ...base, scanId: "s-3" }); + expect(tracker.getActiveAlerts("tenant-a")).toHaveLength(1); + + // 1st success -> not yet resolved (autoResolveConsecutiveSuccesses = 2) + const resSuccess1 = tracker.recordScan({ + ...base, + scanId: "s-4", + hasFailed: false, + }); + expect(resSuccess1.alertResolved).toBeUndefined(); + expect(tracker.getActiveAlerts("tenant-a")).toHaveLength(1); + + // 2nd success -> auto-resolves! + const resSuccess2 = tracker.recordScan({ + ...base, + scanId: "s-5", + hasFailed: false, + }); + expect(resSuccess2.alertResolved).toBeDefined(); + expect(resSuccess2.alertResolved!.status).toBe("resolved"); + expect(resSuccess2.alertResolved!.resolvedAt).toBe(mockTime); + + expect(tracker.getActiveAlerts("tenant-a")).toHaveLength(0); + expect(tracker.getAlertHistory("tenant-a")).toHaveLength(1); + }); + + it("enforces strict tenant isolation across repositories and alerts", () => { + const baseA: ParserScanRecord = { + scanId: "s-a", + tenantId: "tenant-a", + repositoryUrl: "https://github.com/org/repo", + commitHash: "c1", + language: "solidity", + parserName: "solidity-ast", + timestamp: mockTime, + hasFailed: true, + }; + + const baseB: ParserScanRecord = { + ...baseA, + scanId: "s-b", + tenantId: "tenant-b", + }; + + // Tenant A fails 3 times -> alert + tracker.recordScan({ ...baseA, scanId: "s-a1" }); + tracker.recordScan({ ...baseA, scanId: "s-a2" }); + tracker.recordScan({ ...baseA, scanId: "s-a3" }); + + expect(tracker.getActiveAlerts("tenant-a")).toHaveLength(1); + expect(tracker.getActiveAlerts("tenant-b")).toHaveLength(0); + + // Tenant B fails once -> no alert + tracker.recordScan({ ...baseB, scanId: "s-b1" }); + expect(tracker.getActiveAlerts("tenant-b")).toHaveLength(0); + }); + }); + + describe("ParserFailureAlertService - Dispatching and Metrics", () => { + it("dispatches notifications to registered handlers and computes metrics", async () => { + const dispatchedAlerts: ParserAlert[] = []; + const service = new ParserFailureAlertService({ + consecutiveFailuresThreshold: 2, + autoResolveConsecutiveSuccesses: 1, + }); + + service.registerHandler({ + onAlert: (alert) => { + dispatchedAlerts.push(alert); + }, + }); + + // Record 1 failure + await service.recordScan({ + scanId: "sc-1", + tenantId: "tenant-service-test", + repositoryUrl: "https://github.com/org/defi", + commitHash: "hash-1", + language: "rust", + parserName: "soroban-ast", + timestamp: 100, + hasFailed: true, + }); + + expect(dispatchedAlerts).toHaveLength(0); + + // Record 2nd failure -> alert dispatched + await service.recordScan({ + scanId: "sc-2", + tenantId: "tenant-service-test", + repositoryUrl: "https://github.com/org/defi", + commitHash: "hash-2", + language: "rust", + parserName: "soroban-ast", + timestamp: 200, + hasFailed: true, + }); + + expect(dispatchedAlerts).toHaveLength(1); + expect(dispatchedAlerts[0].status).toBe("active"); + + // Record success -> resolved alert dispatched + await service.recordScan({ + scanId: "sc-3", + tenantId: "tenant-service-test", + repositoryUrl: "https://github.com/org/defi", + commitHash: "hash-3", + language: "rust", + parserName: "soroban-ast", + timestamp: 300, + hasFailed: false, + }); + + expect(dispatchedAlerts).toHaveLength(2); + expect(dispatchedAlerts[1].status).toBe("resolved"); + + const metrics = service.getMetrics(); + expect(metrics.totalScansEvaluated).toBe(3); + expect(metrics.totalFailuresRecorded).toBe(2); + expect(metrics.totalAlertsDispatched).toBe(1); + expect(metrics.totalAlertsResolved).toBe(1); + expect(metrics.activeAlertsCount).toBe(0); + }); + }); +}); diff --git a/docs/HEALTH_PROBE_THRESHOLDS.md b/docs/HEALTH_PROBE_THRESHOLDS.md new file mode 100644 index 00000000..a7507275 --- /dev/null +++ b/docs/HEALTH_PROBE_THRESHOLDS.md @@ -0,0 +1,87 @@ +# Configurable Health Probe Thresholds + +## Overview + +Issue #1266 introduces configurable health probe evaluation across GasGuard's subsystems (RPC nodes, Redis connection pools, background worker pools, and AST analyzer workers). + +Instead of treating endpoints with binary alive/dead checks or fixed timeouts, the health probe system distinguishes between **Healthy**, **Degraded**, and **Unhealthy** states based on tunable latency thresholds, consecutive failure/success counts, and sliding window failure rates. + +## Configuration Surface + +The configuration interface `HealthProbeThresholds` supports: + +| Property | Type | Default | Description | +|---|---|---|---| +| `healthyLatencyMsThreshold` | `number` | `500` | Latency (ms) below which response is healthy | +| `degradedLatencyMsThreshold` | `number` | `1500` | Latency (ms) above which status becomes degraded | +| `unhealthyLatencyMsThreshold` | `number` | `5000` | Latency (ms) above which status becomes unhealthy | +| `consecutiveFailuresThreshold` | `number` | `3` | Consecutive failures required before declaring unhealthy | +| `consecutiveSuccessesThreshold` | `number` | `2` | Consecutive successes required to recover to healthy | +| `failureRateDegradedThreshold` | `number` | `0.2` (20%) | Failure rate in window triggering degraded status | +| `failureRateUnhealthyThreshold` | `number` | `0.5` (50%) | Failure rate in window triggering unhealthy status | +| `timeoutMs` | `number` | `5000` | Probe timeout in milliseconds | +| `windowSize` | `number` | `20` | Bounded sliding window size for historical metrics | +| `minSamplesForFailureRate` | `number` | `5` | Minimum samples before evaluating sliding failure rate | + +## State Transition Semantics + +``` + +--------------+ 1 failure or high latency +--------------+ + | | ---------------------------> | | + | HEALTHY | | DEGRADED | + | | <--------------------------- | | + +--------------+ consecutive successes >= N +--------------+ + | | + | | consecutive failures >= M + | | or failure rate >= 50% + | v + | +--------------+ + | | | + +-------------------------------------> | UNHEALTHY | + severe latency or critical errors | | + +--------------+ +``` + +1. **Degraded State**: + - Triggered when latency exceeds `degradedLatencyMsThreshold`. + - Triggered when a single failure occurs (`consecutiveFailures > 0` but `< consecutiveFailuresThreshold`). + - Triggered when sliding window failure rate exceeds `failureRateDegradedThreshold`. + - Transient state during recovery when successes occur but haven't reached `consecutiveSuccessesThreshold`. + +2. **Unhealthy State**: + - Triggered when consecutive failures reach `consecutiveFailuresThreshold`. + - Triggered when sliding window failure rate reaches `failureRateUnhealthyThreshold`. + - Triggered when latency exceeds `unhealthyLatencyMsThreshold`. + +3. **Recovery to Healthy**: + - Requires `consecutiveSuccessesThreshold` consecutive successful samples with latency within healthy bounds. + +## Privacy & Security + +Target endpoints, URLs, error messages, and logs are automatically sanitized via `sanitizeProbeTarget`: +- Authentication tokens (`api_key`, `token`, `secret`, `password`) are redacted. +- Basic authentication credentials in URLs (`https://user:password@node/rpc`) are scrubbed before storage and display. + +## Example Usage + +```typescript +import { HealthProbeRegistry, HealthCheckService } from "@monitoring"; + +const registry = new HealthProbeRegistry(); +const rpcProbe = registry.registerProbe("alchemy-polygon", { + healthyLatencyMsThreshold: 200, + degradedLatencyMsThreshold: 800, + unhealthyLatencyMsThreshold: 2500, + consecutiveFailuresThreshold: 3, + consecutiveSuccessesThreshold: 2, +}); + +// Record probe execution +await rpcProbe.executeProbe(async () => { + const res = await callNodeHealth(); + return { latencyMs: res.elapsedMs }; +}); + +const evalReport = rpcProbe.evaluate(); +console.log(evalReport.status); // "healthy" | "degraded" | "unhealthy" +``` diff --git a/docs/INCIDENT_TIMELINE_EXPORT.md b/docs/INCIDENT_TIMELINE_EXPORT.md new file mode 100644 index 00000000..edf07d4d --- /dev/null +++ b/docs/INCIDENT_TIMELINE_EXPORT.md @@ -0,0 +1,88 @@ +# Incident Timeline Export for Failed Scans + +## Overview + +Issue #1265 adds structured incident timeline tracking and multi-format exports for failed and aborted scans across GasGuard's repository analysis platform. + +When a repository scan fails (due to AST parser crash, out-of-memory error, execution timeout, worker lease expiration, or security violation), platform operators and security engineers require a deterministic post-mortem timeline showing exactly when each phase started, which worker handled the job, lease fencing tokens, and sanitized diagnostic errors. + +## Tenant Isolation & Concurrency Safety + +- **Tenant Boundary**: Every incident event and exported record is strictly partitioned by `tenantId`. A tenant cannot discover, read, or export another tenant's incident timeline. +- **Concurrent Worker Ingestion**: Events emitted by different workers before or after incident finalization are safely merged and deterministically ordered by timestamp. +- **Idempotency**: Retried or duplicated finalization calls from supervisors preserve existing chronological event history without duplicating records. +- **Bounded Resource Consumption**: In-memory stores enforce `maxIncidentsPerTenant` (LRU eviction of oldest incidents) and `maxEventsPerIncident` (sliding truncation of excess events). + +## Export Formats + +The `IncidentTimelineExportService` supports three export formats: + +### 1. JSON (`exportIncident(tenantId, scanId, "json")`) +Machine-readable canonical export conforming to platform observability schema. Includes worker lifecycle metadata, fencing tokens, diagnostics counts, and full error details. + +### 2. Markdown (`exportIncident(tenantId, scanId, "markdown")`) +A comprehensive incident post-mortem markdown report complete with: +- Executive summary & metadata +- Worker lifecycle & lease fencing token +- AST parser diagnostics summary (recovered vs unrecovered) +- Chronological timeline markdown table +- Error details block with code, category, retryability, and stack excerpt + +### 3. CSV (`exportIncident(tenantId, scanId, "csv")`) +RFC 4180 compliant CSV table suitable for spreadsheet and SIEM ingestion with headers: +`event_id,timestamp_iso,stage,severity,type,actor,message,error_code,retryable` + +## Data Sanitization + +All text fields, error messages, and repository URLs are sanitized using `sanitizeIncidentText` to strip: +- Ethereum/EVM 32-byte hexadecimal private keys (`0x[a-fA-F0-9]{64}`) +- Bearer / JWT authorization headers +- API keys, credentials, and tokens in URLs and message bodies + +## Usage Example + +```typescript +import { IncidentTimelineExportService, IncidentTimelineStore } from "./exports/incident-timeline"; + +const store = new IncidentTimelineStore(); +const exportService = new IncidentTimelineExportService(store); + +// Worker emits lifecycle events +store.recordEvent({ + eventId: "evt-01", + scanId: "scan-1234", + tenantId: "tenant-acme", + timestamp: Date.now(), + stage: "parsing", + severity: "error", + type: "parser_error", + actor: "ast-worker-1", + message: "AST syntax error encountered in Vault.sol", + errorDetails: { + code: "SYNTAX_ERROR", + category: "fatal", + message: "Unexpected token at line 42", + retryable: false, + }, +}); + +// Finalize incident upon scan failure +exportService.finalizeIncident({ + incidentId: "inc-1234", + scanId: "scan-1234", + tenantId: "tenant-acme", + repositoryUrl: "https://github.com/acme/defi-vaults", + commitHash: "e5f2a1b", + failureCategory: "PARSER_FAILURE", + summary: "Solidity parser encountered unrecoverable syntax error in Vault.sol", + startedAt: Date.now() - 3000, + failedAt: Date.now(), + durationMs: 3000, + status: "failed", + events: [], +}); + +// Export incident report as Markdown +const mdReport = exportService.exportIncident("tenant-acme", "scan-1234", "markdown"); +console.log(mdReport.data); +``` diff --git a/docs/PARSER_FAILURE_ALERTING.md b/docs/PARSER_FAILURE_ALERTING.md new file mode 100644 index 00000000..c158e0dc --- /dev/null +++ b/docs/PARSER_FAILURE_ALERTING.md @@ -0,0 +1,73 @@ +# Alerting for Repeated Parser Failures + +## Overview + +Issue #1262 introduces automated alerting and failure tracking for repeated Solidity, Vyper, and Soroban AST parser failures across analyzed repositories. + +When repository commits introduce unsupported syntax, compiler version mismatches, or parser worker panics, alerting platform maintainers early prevents silently corrupted scan batches and unanalyzed contracts. + +## Key Capabilities + +- **Deterministic Evaluation**: Evaluates repository scan records monotonically and deterministically, guaranteed to produce identical alert states for identical inputs and clock sequences. +- **Tenant Isolation**: Failure counts, sliding windows, and active alerts are strictly scoped to `(tenantId, repositoryUrl, language)`. Cross-tenant state leakage is impossible. +- **Configurable Alert Thresholds**: + - `consecutiveFailuresThreshold`: Triggers an alert when consecutive scans fail parsing (default `3`). + - `failureRatioThreshold`: Triggers an alert when the ratio of failures in the sliding window exceeds the threshold (default `0.4` or 40%). + - `windowSize`: Number of historical scans tracked in memory per repository (default `10`). + - `cooldownPeriodMs`: Alert deduplication window to suppress notification floods (default `300,000ms` / 5 minutes). + - `autoResolveConsecutiveSuccesses`: Automatically transitions alerts from `active` to `resolved` once consecutive parses succeed (default `2`). +- **Zero Code Leakage**: Aggregates structured diagnostic codes (e.g. `unexpected_token`, `recursion_limit_exceeded`) without storing or logging contract source code or private keys. +- **Severity Escalation**: Distinguishes between `warning` and `critical` severity based on consecutive failure depth. + +## Alert Lifecycle + +``` + Scans fail repeatedly + -----------------------> [ ACTIVE ALERT ] (warning) + | + | Further failures beyond cooldown + v + [ ACTIVE ALERT ] (critical) + | + | Consecutive successes >= autoResolveConsecutiveSuccesses + v + [ RESOLVED ] +``` + +## Example Usage + +```typescript +import { ParserFailureAlertService } from "./scanner/parser-alerting"; + +const alertService = new ParserFailureAlertService({ + consecutiveFailuresThreshold: 3, + autoResolveConsecutiveSuccesses: 2, + cooldownPeriodMs: 5 * 60 * 1000, +}); + +// Register notification sink (e.g. Slack webhook, PagerDuty, Prometheus) +alertService.registerHandler({ + onAlert: async (alert) => { + if (alert.status === "active") { + console.warn(`[ALERT] Parser failure on ${alert.repositoryUrl}: ${alert.triggerReason}`); + } else if (alert.status === "resolved") { + console.log(`[RESOLVED] Parser health restored on ${alert.repositoryUrl}`); + } + }, +}); + +// Record scan outcomes during background worker execution +await alertService.recordScan({ + scanId: "scan-1234", + tenantId: "tenant-acme", + repositoryUrl: "https://github.com/acme/defi-contracts", + commitHash: "e4f1a2b", + language: "solidity", + parserName: "solidity-ast-parser-v2", + timestamp: Date.now(), + hasFailed: true, + failureCategory: "syntax_error", + diagnosticCodes: ["unexpected_token", "invalid_pragma"], + errorSummary: "Unsupported transient storage opcode", +}); +``` diff --git a/docs/TRACE_PROPAGATION.md b/docs/TRACE_PROPAGATION.md new file mode 100644 index 00000000..b7da59b5 --- /dev/null +++ b/docs/TRACE_PROPAGATION.md @@ -0,0 +1,68 @@ +# Distributed Trace Propagation Across API and Workers + +## Overview + +Issue #1263 implements distributed trace propagation compliant with the W3C Trace Context Level 1 standard across GasGuard's API gateway, message queues, and background analyzer workers. + +By injecting and extracting trace identifiers (`traceparent`, `tracestate`, and `baggage`), distributed requests can be traced from incoming user/webhook requests through BullMQ queues into AST parsing workers and rule analyzers without losing causal linkages or tenant boundaries. + +## W3C Trace Context Standard + +Trace propagation strictly adheres to the W3C format: +``` +traceparent: {version}-{trace-id}-{parent-id}-{trace-flags} +``` +- `version`: `00` +- `trace-id`: 32-character hexadecimal string (16 bytes, non-zero) +- `parent-id`: 16-character hexadecimal string (8 bytes, non-zero) +- `trace-flags`: 2-character hexadecimal bitmask (`01` indicates recorded/sampled) + +### Baggage & Tenant Isolation + +- `baggage` preserves contextual metadata across asynchronous job transitions (such as `tenantId`, `userId`, `scanPriority`). +- Strict tenant isolation is maintained by carrying `tenantId` in the trace context and validating that background workers cannot leak or cross tenant execution contexts. + +## Carrier Layers + +1. **HTTP Headers (`extractFromHttpHeaders` / `injectIntoHttpHeaders`)**: + - Parses `traceparent`, `baggage`, and `x-tenant-id`. + - Injects correlated response headers `x-trace-id`, `x-span-id`, `x-tenant-id`. + +2. **Job Envelopes (`injectIntoJobEnvelope` / `extractFromJobEnvelope`)**: + - Wraps asynchronous queue jobs with `_traceContext` containing serialized `traceparent` and `baggage`. + - When workers pick up jobs, a child context is spawned with the same `traceId` and `parentId` pointing to the job enqueue span. + +3. **Active Context Execution (`runWithContext`)**: + - Backed by Node.js `AsyncLocalStorage`. + - Ensures any synchronous or asynchronous operations inside worker handlers have access to the active trace context without passing explicit parameters through every function. + +## Usage Example + +```typescript +import { + extractFromHttpHeaders, + injectIntoJobEnvelope, + extractFromJobEnvelope, + defaultTraceService, +} from "./common/tracing"; + +// 1. API Controller: extract from request +const traceCtx = extractFromHttpHeaders(req.headers); + +// 2. Queue Job: inject into job payload +const jobPayload = injectIntoJobEnvelope({ repoUrl, commitHash }, traceCtx); +await scanQueue.add("scan", jobPayload); + +// 3. Worker: extract and execute within context +const workerTraceCtx = extractFromJobEnvelope(job.data); +await defaultTraceService.runWithContext(workerTraceCtx, async () => { + const { span } = defaultTraceService.startSpan("ast_parsing"); + try { + await parseAstFiles(); + defaultTraceService.endSpan(span, "ok"); + } catch (err) { + defaultTraceService.endSpan(span, "error", err.message); + throw err; + } +}); +``` diff --git a/libs/monitoring/health-check.ts b/libs/monitoring/health-check.ts index 433a3cde..e3022438 100644 --- a/libs/monitoring/health-check.ts +++ b/libs/monitoring/health-check.ts @@ -1,15 +1,44 @@ import axios from "axios"; +import { + HealthProbeRegistry, + HealthProbeThresholds, + ProbeEvaluation, + sanitizeProbeTarget, +} from "./health-probe"; export interface EndpointHealth { url: string; isAlive: boolean; latency: number; lastChecked: Date; + status?: string; + evaluation?: ProbeEvaluation; } export class HealthCheckService { - async checkEndpoint(url: string): Promise { + private readonly registry: HealthProbeRegistry; + private readonly defaultThresholds?: Partial; + + constructor(defaultThresholds?: Partial) { + this.registry = new HealthProbeRegistry(); + this.defaultThresholds = defaultThresholds; + } + + getRegistry(): HealthProbeRegistry { + return this.registry; + } + + async checkEndpoint( + url: string, + overrides?: Partial, + ): Promise { const start = Date.now(); + const probe = this.registry.getOrCreateProbe( + url, + overrides || this.defaultThresholds, + ); + const timeout = probe.getThresholds().timeoutMs; + try { // Basic health check: try to get a simple response or just a TCP connection check // For RPCs, we might want to call a simple method like eth_blockNumber @@ -21,22 +50,55 @@ export class HealthCheckService { params: [], id: 1, }, - { timeout: 5000 }, + { timeout }, ); + const latency = Date.now() - start; + const evaluation = probe.recordSample({ + isAlive: true, + latencyMs: latency, + timestamp: Date.now(), + }); + return { - url, + url: sanitizeProbeTarget(url), isAlive: true, - latency: Date.now() - start, + latency, lastChecked: new Date(), + status: evaluation.status, + evaluation, }; } catch (error) { + const latency = Date.now() - start; + const errMsg = error instanceof Error ? error.message : "check failed"; + const evaluation = probe.recordSample({ + isAlive: false, + latencyMs: latency, + errorMessage: errMsg, + timestamp: Date.now(), + }); + return { - url, + url: sanitizeProbeTarget(url), isAlive: false, latency: -1, lastChecked: new Date(), + status: evaluation.status, + evaluation, }; } } + + async evaluateEndpoint( + url: string, + overrides?: Partial, + ): Promise { + const res = await this.checkEndpoint(url, overrides); + return ( + res.evaluation ?? + this.registry + .getOrCreateProbe(url, overrides || this.defaultThresholds) + .evaluate() + ); + } } diff --git a/libs/monitoring/health-probe.spec.ts b/libs/monitoring/health-probe.spec.ts new file mode 100644 index 00000000..c114a149 --- /dev/null +++ b/libs/monitoring/health-probe.spec.ts @@ -0,0 +1,335 @@ +import { + HealthProbe, + HealthProbeRegistry, + resolveHealthProbeThresholds, + sanitizeProbeTarget, + DEFAULT_HEALTH_PROBE_THRESHOLDS, +} from "./health-probe"; +import { HealthCheckService } from "./health-check"; + +describe("Health Probe Configurable Thresholds (#1266)", () => { + describe("resolveHealthProbeThresholds", () => { + it("returns default thresholds when no overrides given", () => { + const config = resolveHealthProbeThresholds(); + expect(config).toEqual(DEFAULT_HEALTH_PROBE_THRESHOLDS); + }); + + it("accepts valid partial overrides", () => { + const config = resolveHealthProbeThresholds({ + healthyLatencyMsThreshold: 200, + degradedLatencyMsThreshold: 800, + unhealthyLatencyMsThreshold: 2000, + consecutiveFailuresThreshold: 5, + consecutiveSuccessesThreshold: 3, + }); + + expect(config.healthyLatencyMsThreshold).toBe(200); + expect(config.degradedLatencyMsThreshold).toBe(800); + expect(config.unhealthyLatencyMsThreshold).toBe(2000); + expect(config.consecutiveFailuresThreshold).toBe(5); + expect(config.consecutiveSuccessesThreshold).toBe(3); + expect(config.timeoutMs).toBe(DEFAULT_HEALTH_PROBE_THRESHOLDS.timeoutMs); + }); + + it("rejects non-positive or non-integer thresholds", () => { + expect(() => resolveHealthProbeThresholds({ timeoutMs: 0 })).toThrow(RangeError); + expect(() => resolveHealthProbeThresholds({ timeoutMs: -100 })).toThrow(RangeError); + expect(() => resolveHealthProbeThresholds({ timeoutMs: 15.5 })).toThrow(RangeError); + expect(() => resolveHealthProbeThresholds({ consecutiveFailuresThreshold: -1 })).toThrow(RangeError); + expect(() => resolveHealthProbeThresholds({ windowSize: 0 })).toThrow(RangeError); + }); + + it("rejects non-monotonic latency thresholds", () => { + expect(() => + resolveHealthProbeThresholds({ + healthyLatencyMsThreshold: 1000, + degradedLatencyMsThreshold: 500, + }), + ).toThrow(RangeError); + + expect(() => + resolveHealthProbeThresholds({ + degradedLatencyMsThreshold: 3000, + unhealthyLatencyMsThreshold: 2000, + }), + ).toThrow(RangeError); + }); + + it("rejects invalid failure rate thresholds", () => { + expect(() => resolveHealthProbeThresholds({ failureRateDegradedThreshold: -0.1 })).toThrow(RangeError); + expect(() => resolveHealthProbeThresholds({ failureRateDegradedThreshold: 1.5 })).toThrow(RangeError); + expect(() => + resolveHealthProbeThresholds({ + failureRateDegradedThreshold: 0.6, + failureRateUnhealthyThreshold: 0.4, + }), + ).toThrow(RangeError); + }); + }); + + describe("sanitizeProbeTarget", () => { + it("redacts query params containing secret keys or tokens", () => { + const url = "https://eth-node.example.com?api_key=secret123&foo=bar"; + const sanitized = sanitizeProbeTarget(url); + expect(sanitized).toBe("https://eth-node.example.com?api_key=[REDACTED]&foo=bar"); + expect(sanitized).not.toContain("secret123"); + }); + + it("redacts basic auth credentials from URLs", () => { + const url = "https://user:pass1234@stellar-node.example.com/rpc"; + const sanitized = sanitizeProbeTarget(url); + expect(sanitized).toBe("https://user:[REDACTED]@stellar-node.example.com/rpc"); + expect(sanitized).not.toContain("pass1234"); + }); + + it("handles empty or non-string inputs safely", () => { + expect(sanitizeProbeTarget("")).toBe(""); + expect(sanitizeProbeTarget(null as unknown as string)).toBe(""); + }); + }); + + describe("HealthProbe state transitions", () => { + it("starts in unknown state before any samples", () => { + const probe = new HealthProbe("test-rpc"); + const evalReport = probe.evaluate(); + expect(evalReport.status).toBe("unknown"); + expect(evalReport.isHealthy).toBe(false); + expect(evalReport.samplesCount).toBe(0); + }); + + it("becomes healthy when samples are within healthy latency and alive", () => { + const probe = new HealthProbe("test-rpc", { + healthyLatencyMsThreshold: 300, + degradedLatencyMsThreshold: 1000, + unhealthyLatencyMsThreshold: 3000, + }); + + const res = probe.recordSample({ isAlive: true, latencyMs: 150 }); + expect(res.status).toBe("healthy"); + expect(res.isHealthy).toBe(true); + expect(res.consecutiveSuccesses).toBe(1); + expect(res.consecutiveFailures).toBe(0); + expect(res.failureRate).toBe(0); + }); + + it("becomes degraded when latency exceeds degraded threshold", () => { + const probe = new HealthProbe("test-rpc", { + healthyLatencyMsThreshold: 300, + degradedLatencyMsThreshold: 800, + unhealthyLatencyMsThreshold: 2000, + }); + + probe.recordSample({ isAlive: true, latencyMs: 100 }); + const res = probe.recordSample({ isAlive: true, latencyMs: 900 }); + expect(res.status).toBe("degraded"); + expect(res.isHealthy).toBe(false); + expect(res.reasons.some((r) => r.includes("breached degraded latency threshold"))).toBe(true); + }); + + it("becomes unhealthy when latency exceeds unhealthy threshold", () => { + const probe = new HealthProbe("test-rpc", { + healthyLatencyMsThreshold: 300, + degradedLatencyMsThreshold: 800, + unhealthyLatencyMsThreshold: 2000, + }); + + const res = probe.recordSample({ isAlive: true, latencyMs: 2500 }); + expect(res.status).toBe("unhealthy"); + expect(res.isHealthy).toBe(false); + expect(res.reasons.some((r) => r.includes("breached unhealthy latency threshold"))).toBe(true); + }); + + it("transitions from degraded to unhealthy upon reaching consecutive failures threshold", () => { + const probe = new HealthProbe("test-rpc", { + consecutiveFailuresThreshold: 3, + healthyLatencyMsThreshold: 500, + }); + + // 1st failure -> degraded + let evalReport = probe.recordSample({ isAlive: false, latencyMs: 100 }); + expect(evalReport.status).toBe("degraded"); + + // 2nd failure -> still degraded + evalReport = probe.recordSample({ isAlive: false, latencyMs: 100 }); + expect(evalReport.status).toBe("degraded"); + + // 3rd failure -> trips to unhealthy! + evalReport = probe.recordSample({ isAlive: false, latencyMs: 100 }); + expect(evalReport.status).toBe("unhealthy"); + expect(evalReport.consecutiveFailures).toBe(3); + }); + + it("enforces consecutiveSuccessesThreshold before recovering to healthy", () => { + const probe = new HealthProbe("test-rpc", { + consecutiveFailuresThreshold: 2, + consecutiveSuccessesThreshold: 2, + }); + + // Trip to unhealthy + probe.recordSample({ isAlive: false, latencyMs: 50 }); + probe.recordSample({ isAlive: false, latencyMs: 50 }); + expect(probe.getStatus()).toBe("unhealthy"); + + // 1st success -> moves to degraded (recovering), not fully healthy yet + let evalReport = probe.recordSample({ isAlive: true, latencyMs: 50 }); + expect(evalReport.status).toBe("degraded"); + expect(evalReport.consecutiveSuccesses).toBe(1); + + // 2nd success -> recovers to healthy! + evalReport = probe.recordSample({ isAlive: true, latencyMs: 50 }); + expect(evalReport.status).toBe("healthy"); + expect(evalReport.isHealthy).toBe(true); + }); + + it("trips to unhealthy when sliding window failure rate exceeds threshold", () => { + const probe = new HealthProbe("test-rpc", { + windowSize: 4, + failureRateDegradedThreshold: 0.25, + failureRateUnhealthyThreshold: 0.5, + consecutiveFailuresThreshold: 10, // high so consecutive won't trigger + minSamplesForFailureRate: 3, + }); + + probe.recordSample({ isAlive: true, latencyMs: 50 }); + probe.recordSample({ isAlive: false, latencyMs: 50 }); + probe.recordSample({ isAlive: true, latencyMs: 50 }); + // 1 failure out of 3 = 0.33 -> degraded + expect(probe.getStatus()).toBe("degraded"); + + // Another failure: 2 failures out of 4 = 0.5 -> unhealthy! + const res = probe.recordSample({ isAlive: false, latencyMs: 50 }); + expect(res.status).toBe("unhealthy"); + expect(res.failureRate).toBe(0.5); + }); + + it("bounds window size by evicting oldest samples", () => { + const probe = new HealthProbe("test-rpc", { windowSize: 3 }); + probe.recordSample({ isAlive: true, latencyMs: 10 }); + probe.recordSample({ isAlive: true, latencyMs: 20 }); + probe.recordSample({ isAlive: true, latencyMs: 30 }); + probe.recordSample({ isAlive: true, latencyMs: 40 }); + + const evaluation = probe.evaluate(); + expect(evaluation.samplesCount).toBe(3); + // Average of 20, 30, 40 = 30 + expect(evaluation.averageLatencyMs).toBe(30); + }); + + it("resets state cleanly", () => { + const probe = new HealthProbe("test-rpc"); + probe.recordSample({ isAlive: true, latencyMs: 50 }); + expect(probe.evaluate().samplesCount).toBe(1); + + probe.reset(); + expect(probe.evaluate().samplesCount).toBe(0); + expect(probe.getStatus()).toBe("unknown"); + }); + + it("validates invalid constructor parameters", () => { + expect(() => new HealthProbe("")).toThrow(TypeError); + expect(() => new HealthProbe(" ")).toThrow(TypeError); + }); + + it("validates invalid recordSample parameters", () => { + const probe = new HealthProbe("test"); + expect(() => probe.recordSample({ isAlive: "yes" as any, latencyMs: 10 })).toThrow(TypeError); + expect(() => probe.recordSample({ isAlive: true, latencyMs: -10 })).toThrow(RangeError); + expect(() => probe.recordSample({ isAlive: true, latencyMs: NaN })).toThrow(RangeError); + }); + }); + + describe("executeProbe async with timeout", () => { + it("records success when probe function resolves in time", async () => { + const probe = new HealthProbe("test-async", { timeoutMs: 1000 }); + const res = await probe.executeProbe(async () => { + return { latencyMs: 42, metadata: { status: 200 } }; + }); + + expect(res.status).toBe("healthy"); + expect(res.isHealthy).toBe(true); + expect(res.averageLatencyMs).toBe(42); + }); + + it("records failure when probe function throws", async () => { + const probe = new HealthProbe("test-async"); + const res = await probe.executeProbe(async () => { + throw new Error("Connection refused"); + }); + + expect(res.isHealthy).toBe(false); + expect(res.consecutiveFailures).toBe(1); + }); + + it("records failure when probe function times out", async () => { + const probe = new HealthProbe("test-timeout", { timeoutMs: 30 }); + const res = await probe.executeProbe(async () => { + await new Promise((resolve) => setTimeout(resolve, 100)); + }); + + expect(res.isHealthy).toBe(false); + expect(res.consecutiveFailures).toBe(1); + }); + }); + + describe("HealthProbeRegistry", () => { + it("registers and retrieves probes by name", () => { + const registry = new HealthProbeRegistry(); + const probe = registry.registerProbe("redis", { timeoutMs: 2000 }); + expect(registry.getProbe("redis")).toBe(probe); + expect(registry.hasProbe("redis")).toBe(true); + expect(registry.getOrCreateProbe("redis")).toBe(probe); + }); + + it("aggregates status across multiple probes accurately", () => { + const registry = new HealthProbeRegistry(); + const rpc = registry.registerProbe("rpc"); + const redis = registry.registerProbe("redis"); + const worker = registry.registerProbe("worker"); + + // rpc healthy + rpc.recordSample({ isAlive: true, latencyMs: 50 }); + // redis healthy + redis.recordSample({ isAlive: true, latencyMs: 10 }); + // worker degraded + worker.recordSample({ isAlive: true, latencyMs: 1800 }); + + const report = registry.aggregateHealth(); + expect(report.healthyCount).toBe(2); + expect(report.degradedCount).toBe(1); + expect(report.unhealthyCount).toBe(0); + expect(report.overallStatus).toBe("degraded"); + + // worker trips to unhealthy + worker.recordSample({ isAlive: false, latencyMs: 50 }); + worker.recordSample({ isAlive: false, latencyMs: 50 }); + worker.recordSample({ isAlive: false, latencyMs: 50 }); + + const report2 = registry.aggregateHealth(); + expect(report2.unhealthyCount).toBe(1); + expect(report2.overallStatus).toBe("unhealthy"); + }); + + it("removes probes and handles clear", () => { + const registry = new HealthProbeRegistry(); + registry.registerProbe("rpc"); + expect(registry.removeProbe("rpc")).toBe(true); + expect(registry.hasProbe("rpc")).toBe(false); + + registry.registerProbe("a"); + registry.registerProbe("b"); + registry.clear(); + expect(registry.aggregateHealth().overallStatus).toBe("unknown"); + }); + }); + + describe("HealthCheckService integration", () => { + it("instantiates with custom thresholds and checks endpoints", async () => { + const svc = new HealthCheckService({ + healthyLatencyMsThreshold: 250, + degradedLatencyMsThreshold: 1000, + }); + + expect(svc.getRegistry()).toBeInstanceOf(HealthProbeRegistry); + }); + }); +}); diff --git a/libs/monitoring/health-probe.ts b/libs/monitoring/health-probe.ts new file mode 100644 index 00000000..6d42bab2 --- /dev/null +++ b/libs/monitoring/health-probe.ts @@ -0,0 +1,561 @@ +/** + * Configurable Health Probe Thresholds (#1266). + * + * Provides configurable health evaluation for platform subsystems (RPC nodes, + * Redis connection pools, background worker pools, analyzer modules). + * + * Distinguishes between Healthy, Degraded, and Unhealthy based on configurable + * latency thresholds, consecutive failure/success counts, and sliding window + * failure rates. Logs and metrics exclude credentials and private parameters. + */ + +export type HealthProbeStatus = + | "healthy" + | "degraded" + | "unhealthy" + | "unknown"; + +export interface HealthProbeThresholds { + /** Maximum latency in ms considered fully healthy (e.g. 500ms). */ + healthyLatencyMsThreshold: number; + /** Latency threshold in ms above which probe status becomes degraded (e.g. 1500ms). */ + degradedLatencyMsThreshold: number; + /** Latency threshold in ms above which probe status becomes unhealthy (e.g. 5000ms). */ + unhealthyLatencyMsThreshold: number; + /** Consecutive errors needed to mark probe as unhealthy. */ + consecutiveFailuresThreshold: number; + /** Consecutive successes needed to recover from degraded/unhealthy to healthy. */ + consecutiveSuccessesThreshold: number; + /** Sliding window failure rate [0, 1] that trips status to degraded. */ + failureRateDegradedThreshold: number; + /** Sliding window failure rate [0, 1] that trips status to unhealthy. */ + failureRateUnhealthyThreshold: number; + /** Probe execution timeout in milliseconds. */ + timeoutMs: number; + /** Number of historical probe results retained in the sliding window. */ + windowSize: number; + /** Minimum samples required before evaluating sliding window failure rate. */ + minSamplesForFailureRate: number; +} + +export const DEFAULT_HEALTH_PROBE_THRESHOLDS: HealthProbeThresholds = { + healthyLatencyMsThreshold: 500, + degradedLatencyMsThreshold: 1500, + unhealthyLatencyMsThreshold: 5000, + consecutiveFailuresThreshold: 3, + consecutiveSuccessesThreshold: 2, + failureRateDegradedThreshold: 0.2, // 20% + failureRateUnhealthyThreshold: 0.5, // 50% + timeoutMs: 5000, + windowSize: 20, + minSamplesForFailureRate: 5, +}; + +export interface ProbeResult { + probeName: string; + timestamp: number; + isAlive: boolean; + latencyMs: number; + errorMessage?: string; + metadata?: Record; +} + +export interface ProbeEvaluation { + probeName: string; + status: HealthProbeStatus; + isHealthy: boolean; + consecutiveFailures: number; + consecutiveSuccesses: number; + samplesCount: number; + averageLatencyMs: number; + p95LatencyMs: number; + failureRate: number; + reasons: string[]; + lastCheckedAt?: number; +} + +export interface AggregatedHealthReport { + overallStatus: HealthProbeStatus; + timestamp: number; + healthyCount: number; + degradedCount: number; + unhealthyCount: number; + unknownCount: number; + probes: Record; +} + +/** + * Validates and resolves HealthProbeThresholds from optional overrides. + */ +export function resolveHealthProbeThresholds( + overrides?: Partial, +): HealthProbeThresholds { + if (!overrides) { + return { ...DEFAULT_HEALTH_PROBE_THRESHOLDS }; + } + + const resolved = { ...DEFAULT_HEALTH_PROBE_THRESHOLDS, ...overrides }; + + // Validate positive integer thresholds + const positiveIntegers: Array = [ + "healthyLatencyMsThreshold", + "degradedLatencyMsThreshold", + "unhealthyLatencyMsThreshold", + "consecutiveFailuresThreshold", + "consecutiveSuccessesThreshold", + "timeoutMs", + "windowSize", + "minSamplesForFailureRate", + ]; + + for (const field of positiveIntegers) { + const val = resolved[field]; + if (typeof val !== "number" || !Number.isInteger(val) || val <= 0) { + throw new RangeError(`${field} must be a positive integer`); + } + } + + // Validate monotonic latency thresholds + if ( + resolved.healthyLatencyMsThreshold >= resolved.degradedLatencyMsThreshold + ) { + throw new RangeError( + "healthyLatencyMsThreshold must be less than degradedLatencyMsThreshold", + ); + } + if ( + resolved.degradedLatencyMsThreshold >= resolved.unhealthyLatencyMsThreshold + ) { + throw new RangeError( + "degradedLatencyMsThreshold must be less than unhealthyLatencyMsThreshold", + ); + } + + // Validate failure rates in [0, 1] + if ( + typeof resolved.failureRateDegradedThreshold !== "number" || + Number.isNaN(resolved.failureRateDegradedThreshold) || + resolved.failureRateDegradedThreshold < 0 || + resolved.failureRateDegradedThreshold > 1 + ) { + throw new RangeError( + "failureRateDegradedThreshold must be a number between 0 and 1", + ); + } + + if ( + typeof resolved.failureRateUnhealthyThreshold !== "number" || + Number.isNaN(resolved.failureRateUnhealthyThreshold) || + resolved.failureRateUnhealthyThreshold < 0 || + resolved.failureRateUnhealthyThreshold > 1 + ) { + throw new RangeError( + "failureRateUnhealthyThreshold must be a number between 0 and 1", + ); + } + + if ( + resolved.failureRateDegradedThreshold >= + resolved.failureRateUnhealthyThreshold + ) { + throw new RangeError( + "failureRateDegradedThreshold must be less than failureRateUnhealthyThreshold", + ); + } + + return resolved; +} + +/** + * Sanitizes URLs, keys, and tokens to prevent leaking credentials in health logs/evaluations. + */ +export function sanitizeProbeTarget(rawTarget: string): string { + if (!rawTarget || typeof rawTarget !== "string") { + return ""; + } + return rawTarget + .replace( + /(api[_-]?key|token|auth|secret|password)=([^&]+)/gi, + "$1=[REDACTED]", + ) + .replace(/:\/\/([^:]+):([^@]+)@/g, "://$1:[REDACTED]@"); +} + +export class HealthProbe { + readonly probeName: string; + private readonly thresholds: HealthProbeThresholds; + private readonly results: ProbeResult[] = []; + private consecutiveFailures = 0; + private consecutiveSuccesses = 0; + private currentStatus: HealthProbeStatus = "unknown"; + + constructor(probeName: string, thresholds?: Partial) { + if ( + !probeName || + typeof probeName !== "string" || + probeName.trim() === "" + ) { + throw new TypeError("probeName must be a non-empty string"); + } + this.probeName = probeName.trim(); + this.thresholds = resolveHealthProbeThresholds(thresholds); + } + + getThresholds(): HealthProbeThresholds { + return { ...this.thresholds }; + } + + getStatus(): HealthProbeStatus { + return this.currentStatus; + } + + /** + * Records a probe execution sample and updates internal status. + */ + recordSample(sample: { + isAlive: boolean; + latencyMs: number; + errorMessage?: string; + metadata?: Record; + timestamp?: number; + }): ProbeEvaluation { + if (typeof sample.isAlive !== "boolean") { + throw new TypeError("sample.isAlive must be a boolean"); + } + if ( + typeof sample.latencyMs !== "number" || + Number.isNaN(sample.latencyMs) || + sample.latencyMs < 0 + ) { + throw new RangeError("sample.latencyMs must be a non-negative number"); + } + + const timestamp = sample.timestamp ?? Date.now(); + const sanitizedError = sample.errorMessage + ? sanitizeProbeTarget(sample.errorMessage) + : undefined; + + const probeResult: ProbeResult = { + probeName: this.probeName, + timestamp, + isAlive: sample.isAlive, + latencyMs: sample.latencyMs, + ...(sanitizedError ? { errorMessage: sanitizedError } : {}), + ...(sample.metadata ? { metadata: sample.metadata } : {}), + }; + + this.results.push(probeResult); + if (this.results.length > this.thresholds.windowSize) { + this.results.shift(); + } + + if (!sample.isAlive) { + this.consecutiveFailures += 1; + this.consecutiveSuccesses = 0; + } else { + this.consecutiveSuccesses += 1; + this.consecutiveFailures = 0; + } + + return this.evaluate(timestamp); + } + + /** + * Executes an asynchronous probe function with timeout enforcement. + */ + async executeProbe( + probeFn: () => Promise; + }>, + ): Promise { + const start = Date.now(); + let isAlive = false; + let latencyMs = 0; + let errorMessage: string | undefined; + let metadata: Record | undefined; + + try { + const timeoutPromise = new Promise((_, reject) => { + const timer = setTimeout(() => { + reject( + new Error(`Probe timed out after ${this.thresholds.timeoutMs}ms`), + ); + }, this.thresholds.timeoutMs); + if (typeof timer.unref === "function") { + timer.unref(); + } + }); + + const res = await Promise.race([probeFn(), timeoutPromise]); + isAlive = true; + latencyMs = + res?.latencyMs !== undefined ? res.latencyMs : Date.now() - start; + metadata = res?.metadata; + } catch (err: unknown) { + isAlive = false; + latencyMs = Date.now() - start; + errorMessage = err instanceof Error ? err.message : String(err); + } + + return this.recordSample({ + isAlive, + latencyMs, + errorMessage, + metadata, + timestamp: Date.now(), + }); + } + + /** + * Evaluates current health status based on configured thresholds. + */ + evaluate(now: number = Date.now()): ProbeEvaluation { + if (this.results.length === 0) { + return { + probeName: this.probeName, + status: "unknown", + isHealthy: false, + consecutiveFailures: 0, + consecutiveSuccesses: 0, + samplesCount: 0, + averageLatencyMs: 0, + p95LatencyMs: 0, + failureRate: 0, + reasons: ["No samples recorded"], + lastCheckedAt: undefined, + }; + } + + const latencies = this.results + .map((r) => r.latencyMs) + .sort((a, b) => a - b); + const sumLatency = latencies.reduce((acc, curr) => acc + curr, 0); + const avgLatency = sumLatency / latencies.length; + const p95Latency = calculatePercentile(latencies, 0.95); + + const failedCount = this.results.filter((r) => !r.isAlive).length; + const failureRate = failedCount / this.results.length; + + const reasons: string[] = []; + let isUnhealthy = false; + let isDegraded = false; + + const hasEnoughSamples = + this.results.length >= this.thresholds.minSamplesForFailureRate; + + // Check Unhealthy criteria + if ( + this.consecutiveFailures >= this.thresholds.consecutiveFailuresThreshold + ) { + isUnhealthy = true; + reasons.push( + `Consecutive failures (${this.consecutiveFailures}) reached threshold (${this.thresholds.consecutiveFailuresThreshold})`, + ); + } + if ( + hasEnoughSamples && + failureRate >= this.thresholds.failureRateUnhealthyThreshold + ) { + isUnhealthy = true; + reasons.push( + `Failure rate (${(failureRate * 100).toFixed(1)}%) reached unhealthy threshold (${(this.thresholds.failureRateUnhealthyThreshold * 100).toFixed(1)}%)`, + ); + } + const latestSample = this.results[this.results.length - 1]; + if ( + latestSample && + latestSample.latencyMs >= this.thresholds.unhealthyLatencyMsThreshold + ) { + isUnhealthy = true; + reasons.push( + `Latency (${latestSample.latencyMs}ms) breached unhealthy latency threshold (${this.thresholds.unhealthyLatencyMsThreshold}ms)`, + ); + } + + // Check Degraded criteria if not unhealthy + if (!isUnhealthy) { + if (this.consecutiveFailures > 0) { + isDegraded = true; + reasons.push( + `Latest probe failed (${this.consecutiveFailures} consecutive failure)`, + ); + } + if ( + hasEnoughSamples && + failureRate >= this.thresholds.failureRateDegradedThreshold + ) { + isDegraded = true; + reasons.push( + `Failure rate (${(failureRate * 100).toFixed(1)}%) reached degraded threshold (${(this.thresholds.failureRateDegradedThreshold * 100).toFixed(1)}%)`, + ); + } + if ( + latestSample && + latestSample.latencyMs >= this.thresholds.degradedLatencyMsThreshold + ) { + isDegraded = true; + reasons.push( + `Latency (${latestSample.latencyMs}ms) breached degraded latency threshold (${this.thresholds.degradedLatencyMsThreshold}ms)`, + ); + } + } + + // Recovery logic: if currently degraded or unhealthy, require consecutiveSuccessesThreshold + let determinedStatus: HealthProbeStatus = "healthy"; + if (isUnhealthy) { + determinedStatus = "unhealthy"; + } else if (isDegraded) { + determinedStatus = "degraded"; + } else { + // It meets healthy conditions now, check if recovering + if ( + (this.currentStatus === "unhealthy" || + this.currentStatus === "degraded") && + this.consecutiveSuccesses < + this.thresholds.consecutiveSuccessesThreshold + ) { + determinedStatus = "degraded"; + reasons.push( + `Recovering: consecutive successes (${this.consecutiveSuccesses}) below recovery threshold (${this.thresholds.consecutiveSuccessesThreshold})`, + ); + } else { + determinedStatus = "healthy"; + reasons.push("All thresholds within healthy bounds"); + } + } + + this.currentStatus = determinedStatus; + + return { + probeName: this.probeName, + status: this.currentStatus, + isHealthy: this.currentStatus === "healthy", + consecutiveFailures: this.consecutiveFailures, + consecutiveSuccesses: this.consecutiveSuccesses, + samplesCount: this.results.length, + averageLatencyMs: Math.round(avgLatency), + p95LatencyMs: Math.round(p95Latency), + failureRate: Math.round(failureRate * 1000) / 1000, + reasons, + lastCheckedAt: latestSample?.timestamp, + }; + } + + reset(): void { + this.results.length = 0; + this.consecutiveFailures = 0; + this.consecutiveSuccesses = 0; + this.currentStatus = "unknown"; + } +} + +/** + * Registry to register, manage, and aggregate multiple named HealthProbes. + */ +export class HealthProbeRegistry { + private readonly probes: Map = new Map(); + + registerProbe( + probeName: string, + thresholds?: Partial, + ): HealthProbe { + if ( + !probeName || + typeof probeName !== "string" || + probeName.trim() === "" + ) { + throw new TypeError("probeName must be a non-empty string"); + } + const key = probeName.trim(); + const probe = new HealthProbe(key, thresholds); + this.probes.set(key, probe); + return probe; + } + + getProbe(probeName: string): HealthProbe | undefined { + return this.probes.get(probeName.trim()); + } + + getOrCreateProbe( + probeName: string, + thresholds?: Partial, + ): HealthProbe { + const existing = this.getProbe(probeName); + if (existing) { + return existing; + } + return this.registerProbe(probeName, thresholds); + } + + hasProbe(probeName: string): boolean { + return this.probes.has(probeName.trim()); + } + + removeProbe(probeName: string): boolean { + return this.probes.delete(probeName.trim()); + } + + aggregateHealth(now: number = Date.now()): AggregatedHealthReport { + let healthyCount = 0; + let degradedCount = 0; + let unhealthyCount = 0; + let unknownCount = 0; + const probeReports: Record = {}; + + for (const [name, probe] of this.probes.entries()) { + const evaluation = probe.evaluate(now); + probeReports[name] = evaluation; + + switch (evaluation.status) { + case "healthy": + healthyCount += 1; + break; + case "degraded": + degradedCount += 1; + break; + case "unhealthy": + unhealthyCount += 1; + break; + case "unknown": + unknownCount += 1; + break; + } + } + + let overallStatus: HealthProbeStatus = "healthy"; + if (unhealthyCount > 0) { + overallStatus = "unhealthy"; + } else if (degradedCount > 0 || (unknownCount > 0 && healthyCount === 0)) { + overallStatus = "degraded"; + } else if (this.probes.size === 0) { + overallStatus = "unknown"; + } + + return { + overallStatus, + timestamp: now, + healthyCount, + degradedCount, + unhealthyCount, + unknownCount, + probes: probeReports, + }; + } + + clear(): void { + this.probes.clear(); + } +} + +function calculatePercentile( + sorted: readonly number[], + quantile: number, +): number { + if (sorted.length === 0) return 0; + const index = Math.min( + sorted.length - 1, + Math.ceil(quantile * sorted.length) - 1, + ); + return sorted[Math.max(0, index)]; +} diff --git a/libs/monitoring/index.ts b/libs/monitoring/index.ts index 2519363f..61069c93 100644 --- a/libs/monitoring/index.ts +++ b/libs/monitoring/index.ts @@ -1,4 +1,5 @@ export * from "./health-check"; +export * from "./health-probe"; // Redis connection health metrics (#1094) export * from "./redis-health-metrics";