From fc82a975989c18ba0aec930e60257ee0aeb2b344 Mon Sep 17 00:00:00 2001 From: Andrii Kohut Date: Sat, 18 Jul 2026 13:10:02 +0200 Subject: [PATCH] feat: add worker domain events and processing metrics --- apps/worker/package.json | 6 +- apps/worker/src/__tests__/events.test.ts | 66 +++++++++++++++++++++ apps/worker/src/__tests__/processor.test.ts | 58 ++++++++++++++++++ apps/worker/src/events.ts | 60 +++++++++++++++++++ apps/worker/src/index.ts | 16 +++++ apps/worker/src/processor.ts | 29 +++++++++ apps/worker/src/runner.ts | 17 +++++- apps/worker/src/telemetry.ts | 63 ++++++++++++++++++++ docs/configuration.md | 9 +++ packages/db/src/queue.ts | 4 +- pnpm-lock.yaml | 12 ++++ 11 files changed, 337 insertions(+), 3 deletions(-) create mode 100644 apps/worker/src/__tests__/events.test.ts create mode 100644 apps/worker/src/events.ts create mode 100644 apps/worker/src/telemetry.ts diff --git a/apps/worker/package.json b/apps/worker/package.json index 0294eaf..4d669ae 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -23,6 +23,10 @@ "dependencies": { "@flakemetry/contracts": "workspace:*", "@flakemetry/core": "workspace:*", - "@flakemetry/db": "workspace:*" + "@flakemetry/db": "workspace:*", + "@opentelemetry/api": "^1.9.0", + "@opentelemetry/exporter-metrics-otlp-http": "^0.220.0", + "@opentelemetry/resources": "^2.0.0", + "@opentelemetry/sdk-metrics": "^2.0.0" } } diff --git a/apps/worker/src/__tests__/events.test.ts b/apps/worker/src/__tests__/events.test.ts new file mode 100644 index 0000000..a4ba3ae --- /dev/null +++ b/apps/worker/src/__tests__/events.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it, vi } from 'vitest' + +import { createEventBus } from '../events' + +describe('createEventBus', () => { + it('delivers a payload to every subscriber of that event', () => { + const bus = createEventBus() + const first = vi.fn() + const second = vi.fn() + bus.on('run.processed', first) + bus.on('run.processed', second) + + bus.emit('run.processed', { + runId: 'r1', + projectId: 'p1', + executions: 2, + newIdentities: 1, + movedIdentities: 0, + }) + + expect(first).toHaveBeenCalledTimes(1) + expect(second).toHaveBeenCalledTimes(1) + expect(first.mock.calls[0]?.[0]).toMatchObject({ runId: 'r1', executions: 2 }) + }) + + it('does not deliver to subscribers of other events', () => { + const bus = createEventBus() + const handler = vi.fn() + bus.on('identity.created', handler) + + bus.emit('score.updated', { + testIdentityId: 't1', + projectId: 'p1', + score: 0.4, + quarantineCandidate: false, + }) + + expect(handler).not.toHaveBeenCalled() + }) + + it('stops delivering after unsubscribe', () => { + const bus = createEventBus() + const handler = vi.fn() + const off = bus.on('identity.moved', handler) + off() + + bus.emit('identity.moved', { testIdentityId: 't1', projectId: 'p1', alias: 'a' }) + + expect(handler).not.toHaveBeenCalled() + }) + + it('isolates a throwing subscriber from the rest', () => { + const onError = vi.fn() + const bus = createEventBus(onError) + const healthy = vi.fn() + bus.on('identity.created', () => { + throw new Error('boom') + }) + bus.on('identity.created', healthy) + + bus.emit('identity.created', { testIdentityId: 't1', projectId: 'p1', fingerprint: 'fp' }) + + expect(healthy).toHaveBeenCalledTimes(1) + expect(onError).toHaveBeenCalledTimes(1) + }) +}) diff --git a/apps/worker/src/__tests__/processor.test.ts b/apps/worker/src/__tests__/processor.test.ts index ccfb262..5aed70e 100644 --- a/apps/worker/src/__tests__/processor.test.ts +++ b/apps/worker/src/__tests__/processor.test.ts @@ -2,6 +2,7 @@ import type { IngestRunBatch } from '@flakemetry/contracts' import { PrismaClient } from '@flakemetry/db' import { afterAll, beforeEach, describe, expect, it } from 'vitest' +import { createEventBus, type DomainEventMap } from '../events' import { processJob } from '../processor' const hasDb = Boolean(process.env.DATABASE_URL) @@ -96,6 +97,63 @@ describe.skipIf(!hasDb)('processJob', () => { expect(await prisma.testIdentity.count()).toBe(1) }) + it('emits domain events for identities, scores and the processed run', async () => { + const events = createEventBus() + const created: DomainEventMap['identity.created'][] = [] + const scored: DomainEventMap['score.updated'][] = [] + const processed: DomainEventMap['run.processed'][] = [] + events.on('identity.created', (payload) => created.push(payload)) + events.on('score.updated', (payload) => scored.push(payload)) + events.on('run.processed', (payload) => processed.push(payload)) + + const ctx = { ...(await seedProject()), now: NOW, events } + const result = await processJob(prisma, batch(), ctx) + + expect(created).toHaveLength(1) + expect(created[0]?.fingerprint).toBeTruthy() + expect(scored).toHaveLength(1) + expect(scored[0]?.testIdentityId).toBe(created[0]?.testIdentityId) + expect(processed).toEqual([ + { + runId: result.runId, + projectId: ctx.projectId, + executions: 2, + newIdentities: 1, + movedIdentities: 0, + }, + ]) + }) + + it('emits identity.moved when a test file moves', async () => { + const events = createEventBus() + const moved: DomainEventMap['identity.moved'][] = [] + events.on('identity.moved', (payload) => moved.push(payload)) + + const ctx = { ...(await seedProject()), now: NOW, events } + await processJob(prisma, batch(), ctx) + await processJob( + prisma, + batch({ + idempotencyKey: 'run-000002', + executions: [ + { + filePath: 'e2e/auth/login.spec.ts', + suite: 'auth', + title: 'logs in', + status: 'pass', + attempt: 1, + startedAt: new Date('2026-07-16T11:00:00Z'), + durationMs: 1600, + }, + ], + }), + ctx, + ) + + expect(moved).toHaveLength(1) + expect(moved[0]?.alias).toBeTruthy() + }) + it('stitches history across a file move via L2 identity resolution', async () => { const ctx = { ...(await seedProject()), now: NOW } await processJob(prisma, batch(), ctx) diff --git a/apps/worker/src/events.ts b/apps/worker/src/events.ts new file mode 100644 index 0000000..0b37662 --- /dev/null +++ b/apps/worker/src/events.ts @@ -0,0 +1,60 @@ +export interface DomainEventMap { + 'run.processed': { + runId: string + projectId: string + executions: number + newIdentities: number + movedIdentities: number + } + 'identity.created': { + testIdentityId: string + projectId: string + fingerprint: string + } + 'identity.moved': { + testIdentityId: string + projectId: string + alias: string + } + 'score.updated': { + testIdentityId: string + projectId: string + score: number + quarantineCandidate: boolean + } +} + +export type DomainEventName = keyof DomainEventMap + +export type DomainEventHandler = (payload: DomainEventMap[K]) => void + +export interface EventBus { + emit: (name: K, payload: DomainEventMap[K]) => void + on: (name: K, handler: DomainEventHandler) => () => void +} + +export const createEventBus = ( + onHandlerError: (error: unknown) => void = () => undefined, +): EventBus => { + const handlers = new Map>>() + + return { + emit(name, payload) { + for (const handler of handlers.get(name) ?? []) { + try { + ;(handler as DomainEventHandler)(payload) + } catch (error) { + onHandlerError(error) + } + } + }, + on(name, handler) { + const set = handlers.get(name) ?? new Set() + set.add(handler as DomainEventHandler) + handlers.set(name, set) + return () => { + set.delete(handler as DomainEventHandler) + } + }, + } +} diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index f60f63a..b7b36ff 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -1,16 +1,32 @@ import { getPrismaClient, IngestionQueue } from '@flakemetry/db' +import { createEventBus } from './events' import { createWorker } from './runner' +import { initSelfTelemetry, observeQueueDepth } from './telemetry' const prisma = getPrismaClient() const queue = new IngestionQueue(prisma) + +const selfOtelEndpoint = process.env.FLAKEMETRY_SELF_OTEL_ENDPOINT +const shutdownTelemetry = selfOtelEndpoint + ? initSelfTelemetry({ endpoint: selfOtelEndpoint }) + : undefined + +observeQueueDepth(() => queue.depth()) + +const events = createEventBus((error) => { + process.stderr.write(`worker: event handler failed ${String(error)}\n`) +}) + const worker = createWorker(prisma, queue, { pollIntervalMs: Number(process.env.POLL_INTERVAL_MS ?? 1_000), + events, }) const shutdown = () => { process.stdout.write('worker: shutting down\n') worker.stop() + void shutdownTelemetry?.() } process.on('SIGINT', shutdown) process.on('SIGTERM', shutdown) diff --git a/apps/worker/src/processor.ts b/apps/worker/src/processor.ts index 82538a5..789d100 100644 --- a/apps/worker/src/processor.ts +++ b/apps/worker/src/processor.ts @@ -9,12 +9,15 @@ import { } from '@flakemetry/core' import type { Prisma, PrismaClient } from '@flakemetry/db' +import type { EventBus } from './events' + export interface ProcessContext { orgId: string projectId: string now: Date threshold?: number minSamples?: number + events?: EventBus } export interface ProcessResult { @@ -38,6 +41,8 @@ export const processJob = async ( const finishedAt = batch.run.finishedAt ?? null const affected = new Set() + const createdEvents: { testIdentityId: string; projectId: string; fingerprint: string }[] = [] + const movedEvents: { testIdentityId: string; projectId: string; alias: string }[] = [] let newIdentities = 0 let movedIdentities = 0 @@ -120,6 +125,11 @@ export const processJob = async ( }) const entry = existing.find((item) => item.id === identityId) if (entry) entry.aliases = [...entry.aliases, resolution.addAlias] + movedEvents.push({ + testIdentityId: identityId, + projectId: ctx.projectId, + alias: resolution.addAlias, + }) } else { newIdentities += 1 const created = await tx.testIdentity.upsert({ @@ -146,6 +156,7 @@ export const processJob = async ( paramsHash, aliases: [], }) + createdEvents.push({ testIdentityId: identityId, projectId: ctx.projectId, fingerprint }) } affected.add(identityId) @@ -175,10 +186,21 @@ export const processJob = async ( return run.id }) + for (const event of createdEvents) ctx.events?.emit('identity.created', event) + for (const event of movedEvents) ctx.events?.emit('identity.moved', event) + for (const identityId of affected) { await scoreIdentity(prisma, identityId, ctx) } + ctx.events?.emit('run.processed', { + runId, + projectId: ctx.projectId, + executions: batch.executions.length, + newIdentities, + movedIdentities, + }) + return { runId, executions: batch.executions.length, @@ -260,4 +282,11 @@ const scoreIdentity = async ( create: { testIdentityId: identityId, ...data }, update: data, }) + + ctx.events?.emit('score.updated', { + testIdentityId: identityId, + projectId: ctx.projectId, + score: result.score, + quarantineCandidate: result.quarantineCandidate, + }) } diff --git a/apps/worker/src/runner.ts b/apps/worker/src/runner.ts index d572226..d59bf7a 100644 --- a/apps/worker/src/runner.ts +++ b/apps/worker/src/runner.ts @@ -1,12 +1,15 @@ import { ingestRunBatchSchema } from '@flakemetry/contracts' import type { IngestionQueue, PrismaClient } from '@flakemetry/db' +import type { EventBus } from './events' import { processJob } from './processor' +import { workerMetrics } from './telemetry' export interface WorkerOptions { pollIntervalMs?: number batchSize?: number now?: () => Date + events?: EventBus } export interface Worker { @@ -30,16 +33,28 @@ export const createWorker = ( const tick = async (): Promise => { const jobs = await queue.dequeue(batchSize) for (const job of jobs) { + const pickedUpAt = Date.now() + workerMetrics.processingLag.record(Math.max(0, pickedUpAt - job.createdAt.getTime())) try { const batch = ingestRunBatchSchema.parse(job.payload) - await processJob(prisma, batch, { orgId: job.orgId, projectId: job.projectId, now: now() }) + await processJob(prisma, batch, { + orgId: job.orgId, + projectId: job.projectId, + now: now(), + events: options.events, + }) await queue.complete(job.id) + workerMetrics.jobsProcessed.add(1) } catch (error) { const outcome = await queue.fail( job.id, error instanceof Error ? error.message : String(error), ) + workerMetrics.jobsFailed.add(1, { outcome }) + if (outcome === 'dead') workerMetrics.jobsDeadLettered.add(1) process.stderr.write(`worker: job ${job.id} failed (${outcome}): ${String(error)}\n`) + } finally { + workerMetrics.processingDuration.record(Date.now() - pickedUpAt) } } return jobs.length diff --git a/apps/worker/src/telemetry.ts b/apps/worker/src/telemetry.ts new file mode 100644 index 0000000..01bb79d --- /dev/null +++ b/apps/worker/src/telemetry.ts @@ -0,0 +1,63 @@ +import { metrics } from '@opentelemetry/api' +import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http' +import { resourceFromAttributes } from '@opentelemetry/resources' +import { MeterProvider, PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics' + +const meter = metrics.getMeter('flakemetry-worker') + +export const workerMetrics = { + jobsProcessed: meter.createCounter('flakemetry.worker.jobs_processed', { + description: 'ingestion jobs processed successfully', + }), + jobsFailed: meter.createCounter('flakemetry.worker.jobs_failed', { + description: 'ingestion jobs that failed processing', + }), + jobsDeadLettered: meter.createCounter('flakemetry.worker.jobs_dead_lettered', { + description: 'ingestion jobs moved to the dead letter state', + }), + processingLag: meter.createHistogram('flakemetry.worker.processing_lag', { + description: 'time between a job being enqueued and picked up', + unit: 'ms', + }), + processingDuration: meter.createHistogram('flakemetry.worker.processing_duration', { + description: 'time spent processing a job', + unit: 'ms', + }), +} + +export const observeQueueDepth = (getDepth: () => Promise): void => { + meter + .createObservableGauge('flakemetry.worker.queue_depth', { + description: 'pending ingestion jobs', + }) + .addCallback(async (result) => { + result.observe(await getDepth()) + }) +} + +export interface SelfTelemetryOptions { + endpoint: string + headers?: Record + exportIntervalMs?: number + serviceName?: string +} + +export const initSelfTelemetry = (options: SelfTelemetryOptions): (() => Promise) => { + const exporter = new OTLPMetricExporter({ + url: `${options.endpoint.replace(/\/+$/, '')}/v1/metrics`, + headers: options.headers, + }) + const provider = new MeterProvider({ + resource: resourceFromAttributes({ + 'service.name': options.serviceName ?? 'flakemetry-worker', + }), + readers: [ + new PeriodicExportingMetricReader({ + exporter, + exportIntervalMillis: options.exportIntervalMs ?? 30_000, + }), + ], + }) + metrics.setGlobalMeterProvider(provider) + return () => provider.shutdown() +} diff --git a/docs/configuration.md b/docs/configuration.md index f381b8b..ec4b128 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -85,6 +85,15 @@ Unknown keys are rejected with an error naming the offending path — typos fail The API also rate-limits per project token (fixed window) and returns `429` with `Retry-After` when exceeded. +### Processing worker + +| Variable | Effect | +|---|---| +| `POLL_INTERVAL_MS` | Idle poll interval between dequeue attempts | +| `FLAKEMETRY_SELF_OTEL_ENDPOINT` | OTLP endpoint for the worker's own metrics (processing lag, throughput, error rate, queue depth) | + +The worker emits domain events (`run.processed`, `identity.created`, `identity.moved`, `score.updated`) after each job commits — the seam downstream stages such as signature clustering and AI RCA subscribe to. + ## Inspecting the resolved configuration ```bash diff --git a/packages/db/src/queue.ts b/packages/db/src/queue.ts index bff713d..87e0760 100644 --- a/packages/db/src/queue.ts +++ b/packages/db/src/queue.ts @@ -19,6 +19,7 @@ export interface QueuedJob { idempotencyKey: string payload: unknown attempts: number + createdAt: Date } export interface IngestionQueueOptions { @@ -84,7 +85,8 @@ export class IngestionQueue { FOR UPDATE SKIP LOCKED ) RETURNING id, org_id AS "orgId", project_id AS "projectId", - idempotency_key AS "idempotencyKey", payload, attempts + idempotency_key AS "idempotencyKey", payload, attempts, + created_at AS "createdAt" ` } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 1f5f746..4598d25 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -100,6 +100,18 @@ importers: '@flakemetry/db': specifier: workspace:* version: link:../../packages/db + '@opentelemetry/api': + specifier: ^1.9.0 + version: 1.9.1 + '@opentelemetry/exporter-metrics-otlp-http': + specifier: ^0.220.0 + version: 0.220.0(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': + specifier: ^2.0.0 + version: 2.9.0(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-metrics': + specifier: ^2.0.0 + version: 2.9.0(@opentelemetry/api@1.9.1) devDependencies: '@flakemetry/eslint-config': specifier: workspace:*