feat: harden ingestion api with rate limiting, backpressure, gzip and self-metrics

apps/api/package.json

@@ -31,6 +31,10 @@
   "dependencies": {
     "@flakemetry/contracts": "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",
     "@trpc/server": "^11.0.0",
     "fastify": "^5.2.0"
   }

apps/api/src/__tests__/hardening.test.ts

@@ -0,0 +1,157 @@
+import { gzipSync } from 'node:zlib'
+
+import { RESOURCE_ATTR, SPAN_ATTR, SPAN_NAMES } from '@flakemetry/contracts'
+import { generateToken, hashToken, PrismaClient } from '@flakemetry/db'
+import { afterAll, beforeEach, describe, expect, it } from 'vitest'
+
+import { buildApp } from '../app'
+
+const hasDb = Boolean(process.env.DATABASE_URL)
+const prisma = new PrismaClient()
+
+const validBatch = (idempotencyKey: string) => ({
+  contractVersion: '0.1.0',
+  idempotencyKey,
+  resource: { ciProvider: 'github_actions', commitSha: 'abc1234', branch: 'main', trigger: 'push' },
+  run: { status: 'passed', startedAt: '2026-07-16T10:00:00Z' },
+  executions: [
+    {
+      filePath: 'e2e/login.spec.ts',
+      suite: 'auth',
+      title: 'logs in',
+      status: 'pass',
+      attempt: 1,
+      startedAt: '2026-07-16T10:00:01Z',
+      durationMs: 1200,
+    },
+  ],
+})
+
+const s = (value: string) => ({ stringValue: value })
+const i = (value: number) => ({ intValue: String(value) })
+const RUN_START = Date.parse('2026-07-16T10:00:00Z') * 1_000_000
+const nano = (ms: number) => String(RUN_START + ms * 1_000_000)
+
+const otlpRequest = (idempotencyKey: string) => ({
+  resourceSpans: [
+    {
+      resource: {
+        attributes: [
+          { key: RESOURCE_ATTR.project, value: s('web') },
+          { key: RESOURCE_ATTR.commitSha, value: s('abc1234') },
+          { key: RESOURCE_ATTR.branch, value: s('main') },
+          { key: RESOURCE_ATTR.ciProvider, value: s('github_actions') },
+          { key: RESOURCE_ATTR.trigger, value: s('push') },
+          { key: RESOURCE_ATTR.idempotencyKey, value: s(idempotencyKey) },
+        ],
+      },
+      scopeSpans: [
+        {
+          spans: [
+            {
+              traceId: 'a'.repeat(32),
+              spanId: 'b'.repeat(16),
+              name: SPAN_NAMES.run,
+              startTimeUnixNano: nano(0),
+              endTimeUnixNano: nano(5000),
+              status: { code: 1 },
+            },
+            {
+              traceId: 'a'.repeat(32),
+              spanId: 'c'.repeat(16),
+              parentSpanId: 'b'.repeat(16),
+              name: SPAN_NAMES.case,
+              startTimeUnixNano: nano(10),
+              endTimeUnixNano: nano(1810),
+              attributes: [
+                { key: SPAN_ATTR.fingerprint, value: s('fp-login') },
+                { key: SPAN_ATTR.title, value: s('logs in') },
+                { key: SPAN_ATTR.filePath, value: s('e2e/login.spec.ts') },
+                { key: SPAN_ATTR.status, value: s('pass') },
+                { key: SPAN_ATTR.attempt, value: i(1) },
+                { key: SPAN_ATTR.durationMs, value: i(1800) },
+              ],
+            },
+          ],
+        },
+      ],
+    },
+  ],
+})
+
+const seedToken = async () => {
+  const org = await prisma.org.create({ data: { name: 'Acme', slug: `acme-${Date.now()}` } })
+  const project = await prisma.project.create({ data: { orgId: org.id, name: 'Web', slug: 'web' } })
+  const raw = generateToken()
+  await prisma.ingestToken.create({
+    data: { orgId: org.id, projectId: project.id, name: 'ci', tokenHash: hashToken(raw) },
+  })
+  return { raw }
+}
+
+describe.skipIf(!hasDb)('api hardening', () => {
+  beforeEach(async () => {
+    await prisma.ingestionJob.deleteMany()
+    await prisma.ingestToken.deleteMany()
+    await prisma.project.deleteMany()
+    await prisma.org.deleteMany()
+  })
+
+  afterAll(async () => {
+    await prisma.$disconnect()
+  })
+
+  it('rate limits a token beyond the configured window', async () => {
+    const { raw } = await seedToken()
+    const app = buildApp({ prisma, rateLimit: { max: 1, windowMs: 60_000 } })
+    const send = (key: string) =>
+      app.inject({
+        method: 'POST',
+        url: '/v1/ingest',
+        headers: { authorization: `Bearer ${raw}` },
+        payload: validBatch(key),
+      })
+
+    expect((await send('run-000001')).statusCode).toBe(202)
+    const limited = await send('run-000002')
+    expect(limited.statusCode).toBe(429)
+    expect(limited.headers['retry-after']).toBeDefined()
+  })
+
+  it('sheds load with 503 when the queue depth exceeds the limit', async () => {
+    const { raw } = await seedToken()
+    const app = buildApp({ prisma, maxQueueDepth: 1 })
+    const send = (key: string) =>
+      app.inject({
+        method: 'POST',
+        url: '/v1/ingest',
+        headers: { authorization: `Bearer ${raw}` },
+        payload: validBatch(key),
+      })
+
+    expect((await send('run-000001')).statusCode).toBe(202)
+    const shed = await send('run-000002')
+    expect(shed.statusCode).toBe(503)
+    expect(shed.headers['retry-after']).toBeDefined()
+  })
+
+  it('decompresses gzip-encoded OTLP request bodies', async () => {
+    const { raw } = await seedToken()
+    const app = buildApp({ prisma })
+    const body = gzipSync(Buffer.from(JSON.stringify(otlpRequest('run-000001'))))
+
+    const res = await app.inject({
+      method: 'POST',
+      url: '/v1/traces',
+      headers: {
+        authorization: `Bearer ${raw}`,
+        'content-type': 'application/json',
+        'content-encoding': 'gzip',
+      },
+      payload: body,
+    })
+
+    expect(res.statusCode).toBe(200)
+    expect(await prisma.ingestionJob.count()).toBe(1)
+  })
+})

apps/api/src/__tests__/rate-limit.test.ts

@@ -0,0 +1,32 @@
+import { describe, expect, it } from 'vitest'
+
+import { createRateLimiter } from '../rate-limit'
+
+describe('createRateLimiter', () => {
+  it('allows up to max requests per window then rejects', () => {
+    const limiter = createRateLimiter({ max: 2, windowMs: 1_000, now: () => 1_000 })
+
+    expect(limiter.check('a').allowed).toBe(true)
+    expect(limiter.check('a').allowed).toBe(true)
+    const blocked = limiter.check('a')
+    expect(blocked.allowed).toBe(false)
+    expect(blocked.retryAfterMs).toBeGreaterThan(0)
+  })
+
+  it('resets the counter when the window elapses', () => {
+    let clock = 1_000
+    const limiter = createRateLimiter({ max: 1, windowMs: 1_000, now: () => clock })
+
+    expect(limiter.check('a').allowed).toBe(true)
+    expect(limiter.check('a').allowed).toBe(false)
+    clock += 1_000
+    expect(limiter.check('a').allowed).toBe(true)
+  })
+
+  it('tracks keys independently', () => {
+    const limiter = createRateLimiter({ max: 1, windowMs: 1_000, now: () => 0 })
+    expect(limiter.check('a').allowed).toBe(true)
+    expect(limiter.check('b').allowed).toBe(true)
+    expect(limiter.check('a').allowed).toBe(false)
+  })
+})

apps/api/src/app.ts

@@ -1,30 +1,83 @@
+import { createGunzip } from 'node:zlib'
+
 import {
   ingestRunBatchSchema,
   otlpToIngestBatch,
   otlpTraceRequestSchema,
 } from '@flakemetry/contracts'
 import { IngestionQueue, type PrismaClient } from '@flakemetry/db'
 import { fastifyTRPCPlugin } from '@trpc/server/adapters/fastify'
-import Fastify, { type FastifyInstance } from 'fastify'
+import Fastify, {
+  type FastifyInstance,
+  type FastifyReply,
+  type FastifyServerOptions,
+} from 'fastify'
 
 import { authenticateProject } from './auth'
+import { createRateLimiter } from './rate-limit'
+import { apiMetrics, observeQueueDepth } from './telemetry'
 import { createContextFactory } from './trpc/context'
 import { appRouter } from './trpc/router'
 
 export interface AppOptions {
   prisma: PrismaClient
   queue?: IngestionQueue
   bodyLimitBytes?: number
+  logger?: FastifyServerOptions['logger']
+  maxQueueDepth?: number
+  rateLimit?: { max: number; windowMs: number }
+  rateLimitNow?: () => number
 }
 
+type Admission = { ok: true } | { ok: false; status: number; reason: string; retryAfterMs: number }
+
 export const buildApp = (options: AppOptions): FastifyInstance => {
   const { prisma } = options
   const queue = options.queue ?? new IngestionQueue(prisma)
   const app = Fastify({
-    logger: false,
+    logger: options.logger ?? false,
     bodyLimit: options.bodyLimitBytes ?? 8 * 1024 * 1024,
   })
 
+  const limiter = createRateLimiter({
+    max: options.rateLimit?.max ?? 600,
+    windowMs: options.rateLimit?.windowMs ?? 60_000,
+    now: options.rateLimitNow,
+  })
+
+  observeQueueDepth(() => queue.depth())
+
+  app.addHook('preParsing', async (request, _reply, payload) => {
+    if (request.headers['content-encoding'] !== 'gzip') return payload
+    delete request.headers['content-encoding']
+    delete request.headers['content-length']
+    return payload.pipe(createGunzip())
+  })
+
+  app.addHook('onResponse', async (request, reply) => {
+    apiMetrics.requestDuration.record(reply.elapsedTime, {
+      route: request.routeOptions.url ?? request.url,
+      status: reply.statusCode,
+    })
+  })
+
+  const admit = async (projectId: string): Promise<Admission> => {
+    const decision = limiter.check(projectId)
+    if (!decision.allowed) {
+      apiMetrics.rateLimited.add(1)
+      return { ok: false, status: 429, reason: 'rate_limited', retryAfterMs: decision.retryAfterMs }
+    }
+    if (options.maxQueueDepth != null && (await queue.depth()) >= options.maxQueueDepth) {
+      apiMetrics.backpressured.add(1)
+      return { ok: false, status: 503, reason: 'backpressure', retryAfterMs: 1_000 }
+    }
+    return { ok: true }
+  }
+
+  const setRetryAfter = (reply: FastifyReply, ms: number): void => {
+    reply.header('retry-after', Math.max(1, Math.ceil(ms / 1_000)))
+  }
+
   app.get('/health', async () => ({ status: 'ok', service: 'api' }))
 
   void app.register(fastifyTRPCPlugin, {
@@ -38,6 +91,12 @@ export const buildApp = (options: AppOptions): FastifyInstance => {
       return reply.code(401).send({ error: 'unauthorized', message: 'missing or invalid token' })
     }
 
+    const admission = await admit(project.projectId)
+    if (!admission.ok) {
+      setRetryAfter(reply, admission.retryAfterMs)
+      return reply.code(admission.status).send({ error: admission.reason })
+    }
+
     const parsed = ingestRunBatchSchema.safeParse(request.body)
     if (!parsed.success) {
       return reply.code(400).send({
@@ -57,6 +116,9 @@ export const buildApp = (options: AppOptions): FastifyInstance => {
       payload: JSON.parse(JSON.stringify(batch)),
     })
 
+    apiMetrics.runsAccepted.add(1)
+    apiMetrics.executionsAccepted.add(batch.executions.length)
+
     return reply.code(202).send({
       receiptId: jobId,
       acceptedExecutions: batch.executions.length,
@@ -70,6 +132,14 @@ export const buildApp = (options: AppOptions): FastifyInstance => {
       return reply.code(401).send({ error: 'unauthorized', message: 'missing or invalid token' })
     }
 
+    const admission = await admit(project.projectId)
+    if (!admission.ok) {
+      setRetryAfter(reply, admission.retryAfterMs)
+      return reply
+        .code(admission.status)
+        .send({ partialSuccess: { rejectedSpans: '0', errorMessage: admission.reason } })
+    }
+
     const parsedRequest = otlpTraceRequestSchema.safeParse(request.body)
     if (!parsedRequest.success) {
       return reply.code(400).send({
@@ -96,6 +166,9 @@ export const buildApp = (options: AppOptions): FastifyInstance => {
       payload: JSON.parse(JSON.stringify(batch)),
     })
 
+    apiMetrics.runsAccepted.add(1)
+    apiMetrics.executionsAccepted.add(batch.executions.length)
+
     return reply.code(200).send({})
   })
 

apps/api/src/index.ts

@@ -1,13 +1,35 @@
 import { getPrismaClient } from '@flakemetry/db'
 
 import { buildApp } from './app'
+import { initSelfTelemetry } from './telemetry'
 
 export type { AppRouter } from './app'
 
 const port = Number(process.env.PORT ?? 4000)
 const host = process.env.HOST ?? '0.0.0.0'
 
-const app = buildApp({ prisma: getPrismaClient() })
+const selfOtelEndpoint = process.env.FLAKEMETRY_SELF_OTEL_ENDPOINT
+if (selfOtelEndpoint) {
+  const shutdown = initSelfTelemetry({ endpoint: selfOtelEndpoint })
+  for (const signal of ['SIGINT', 'SIGTERM'] as const) {
+    process.on(signal, () => {
+      void shutdown()
+    })
+  }
+}
+
+const maxQueueDepth = process.env.FLAKEMETRY_MAX_QUEUE_DEPTH
+  ? Number(process.env.FLAKEMETRY_MAX_QUEUE_DEPTH)
+  : undefined
+
+const app = buildApp({
+  prisma: getPrismaClient(),
+  logger: {
+    level: process.env.LOG_LEVEL ?? 'info',
+    redact: ['req.headers.authorization'],
+  },
+  maxQueueDepth,
+})
 
 app
   .listen({ port, host })