Defer span processing to background worker via object storage

The OTEL ingest endpoint previously parsed, transformed, and wrote spans to ClickHouse synchronously within the HTTP request. As the span processing logic grows more complex, this coupling increases response latency and risks timeouts from OTEL exporters.

This change splits ingestion into two phases: the endpoint now validates the payload, buffers it to object storage, and enqueues a BullMQ job — returning immediately. A new background worker picks up the job, runs the transform, persists to ClickHouse, and cleans up the buffered payload. This keeps the ingest endpoint lightweight and allows processing to retry independently on failure.

The OTLP parsing and transform logic was relocated from `apps/ingest` to `@domain/spans` so both the ingest app and workers app can share it without cross-app imports.

Author: csansoon

apps/ingest/package.json

@@ -23,8 +23,10 @@
     "@platform/db-clickhouse": "workspace:*",
     "@platform/db-postgres": "workspace:*",
     "@platform/env": "workspace:*",
+    "@platform/storage-object": "workspace:*",
     "@repo/observability": "workspace:*",
     "@repo/utils": "workspace:*",
+    "bullmq": "catalog:",
     "dotenv": "catalog:",
     "effect": "catalog:",
     "hono": "catalog:",

apps/ingest/src/clients.ts

@@ -1,15 +1,22 @@
 import type { ClickHouseClient } from "@clickhouse/client"
-import type { RedisClient } from "@platform/cache-redis"
+import { SPAN_INGESTION_QUEUE } from "@domain/shared"
+import type { RedisClient, RedisConnection } from "@platform/cache-redis"
 import { createRedisClient, createRedisConnection } from "@platform/cache-redis"
 import { createClickhouseClient } from "@platform/db-clickhouse"
 import { createPostgresClient, type PostgresClient } from "@platform/db-postgres"
 import { parseEnv } from "@platform/env"
+import type { StorageDisk } from "@platform/storage-object"
+import { createStorageDisk } from "@platform/storage-object"
+import { Queue } from "bullmq"
 import { Effect } from "effect"
 
 let postgresClientInstance: PostgresClient | undefined
 let adminPostgresClientInstance: PostgresClient | undefined
 let clickhouseInstance: ClickHouseClient | undefined
 let redisInstance: RedisClient | undefined
+let redisConnectionInstance: RedisConnection | undefined
+let storageDiskInstance: StorageDisk | undefined
+let spanIngestionQueueInstance: Queue | undefined
 
 export const getPostgresClient = (): PostgresClient => {
   if (!postgresClientInstance) {
@@ -44,3 +51,24 @@ export const getRedisClient = (): RedisClient => {
   }
   return redisInstance
 }
+
+const getRedisConnection = (): RedisConnection => {
+  if (!redisConnectionInstance) {
+    redisConnectionInstance = createRedisConnection()
+  }
+  return redisConnectionInstance
+}
+
+export const getStorageDisk = (): StorageDisk => {
+  if (!storageDiskInstance) {
+    storageDiskInstance = createStorageDisk()
+  }
+  return storageDiskInstance
+}
+
+export const getSpanIngestionQueue = (): Queue => {
+  if (!spanIngestionQueueInstance) {
+    spanIngestionQueueInstance = new Queue(SPAN_INGESTION_QUEUE, { connection: getRedisConnection() })
+  }
+  return spanIngestionQueueInstance
+}

apps/ingest/src/otlp/proto.ts

@@ -1,5 +1,5 @@
+import type { OtlpExportTraceServiceRequest } from "@domain/spans"
 import protobuf from "protobufjs"
-import type { OtlpExportTraceServiceRequest } from "./types.ts"
 
 /**
  * Protobuf schema for OTLP ExportTraceServiceRequest.
@@ -132,15 +132,14 @@ function bytesToHex(bytes: Uint8Array | number[]): string {
 
 /**
  * Recursively normalize a decoded protobuf object to match the OTLP/JSON shape:
- * - Uint8Array bytes fields → lowercase hex strings
- * - Long objects → decimal strings
+ * - Uint8Array bytes fields -> lowercase hex strings
+ * - Long objects -> decimal strings
  */
 function normalizeValue(value: unknown): unknown {
   if (value === null || value === undefined) return value
   if (value instanceof Uint8Array) return bytesToHex(value)
   if (Array.isArray(value)) return value.map(normalizeValue)
 
-  // protobufjs Long objects have low/high/unsigned properties
   if (typeof value === "object" && "low" in value && "high" in value) {
     const long = value as { low: number; high: number; unsigned: boolean }
     const bigint = BigInt(long.high >>> 0) * BigInt(2 ** 32) + BigInt(long.low >>> 0)

apps/ingest/src/routes/traces.ts

@@ -1,14 +1,11 @@
-import { OrganizationId } from "@domain/shared"
-import { SpanRepository } from "@domain/spans"
-import { ChSqlClientLive, SpanRepositoryLive } from "@platform/db-clickhouse"
+import { OrganizationId, ProjectId, putInDisk } from "@domain/shared"
+import type { OtlpExportTraceServiceRequest } from "@domain/spans"
 import { Effect } from "effect"
 import type { Hono } from "hono"
-import { getClickhouseClient } from "../clients.ts"
+import { getSpanIngestionQueue, getStorageDisk } from "../clients.ts"
 import { authMiddleware } from "../middleware/auth.ts"
 import { projectMiddleware } from "../middleware/project.ts"
 import { decodeOtlpProtobuf } from "../otlp/proto.ts"
-import { transformOtlpToSpans } from "../otlp/transform.ts"
-import type { OtlpExportTraceServiceRequest } from "../otlp/types.ts"
 import type { IngestEnv } from "../types.ts"
 
 interface TracesRouteContext {
@@ -31,22 +28,38 @@ export const registerTracesRoute = ({ app }: TracesRouteContext) => {
       return c.json({ error: "Invalid OTLP payload" }, 400)
     }
 
-    const spans = transformOtlpToSpans(request, {
-      organizationId: c.get("organizationId"),
-      projectId: c.get("projectId"),
-      apiKeyId: c.get("apiKeyId"),
-    })
-
-    if (spans.length > 0) {
-      const orgId = OrganizationId(c.get("organizationId"))
-      await Effect.runPromise(
-        Effect.gen(function* () {
-          const repo = yield* SpanRepository
-          yield* repo.insert(spans)
-        }).pipe(Effect.provide(SpanRepositoryLive), Effect.provide(ChSqlClientLive(getClickhouseClient(), orgId))),
-      )
+    if (!request.resourceSpans?.length) {
+      return c.json({})
     }
 
+    const organizationId = c.get("organizationId")
+    const projectId = c.get("projectId")
+    const apiKeyId = c.get("apiKeyId")
+    const ingestedAt = new Date().toISOString()
+
+    const payload = JSON.stringify({ request, context: { organizationId, projectId, apiKeyId }, ingestedAt })
+
+    const disk = getStorageDisk()
+    const storageKey = await Effect.runPromise(
+      putInDisk(disk, {
+        namespace: "ingest",
+        organizationId: OrganizationId(organizationId),
+        projectId: ProjectId(projectId),
+        content: payload,
+      }),
+    )
+
+    const queue = getSpanIngestionQueue()
+    await queue.add(
+      "process-spans",
+      { storageKey },
+      {
+        jobId: storageKey,
+        attempts: 3,
+        backoff: { type: "exponential", delay: 2000 },
+      },
+    )
+
     return c.json({})
   })
 }

apps/workers/package.json

@@ -13,10 +13,15 @@
     "check:fix": "biome check --fix src"
   },
   "dependencies": {
+    "@clickhouse/client": "catalog:",
+    "@domain/shared": "workspace:*",
+    "@domain/spans": "workspace:*",
     "@platform/cache-redis": "workspace:*",
+    "@platform/db-clickhouse": "workspace:*",
     "@platform/db-postgres": "workspace:*",
     "@platform/events-outbox": "workspace:*",
     "@platform/queue-bullmq": "workspace:*",
+    "@platform/storage-object": "workspace:*",
     "@repo/observability": "workspace:*",
     "@types/pg": "catalog:",
     "bullmq": "catalog:",

apps/workers/src/clients.ts

@@ -1,10 +1,16 @@
+import type { ClickHouseClient } from "@clickhouse/client"
 import type { RedisConnection } from "@platform/cache-redis"
 import { createRedisConnection } from "@platform/cache-redis"
+import { createClickhouseClient } from "@platform/db-clickhouse"
 import { createPostgresPool } from "@platform/db-postgres"
+import type { StorageDisk } from "@platform/storage-object"
+import { createStorageDisk } from "@platform/storage-object"
 import type { Pool } from "pg"
 
 let redisConnectionInstance: RedisConnection | undefined
 let pgPoolInstance: Pool | undefined
+let clickhouseInstance: ClickHouseClient | undefined
+let storageDiskInstance: StorageDisk | undefined
 
 export const getRedisConnection = (): RedisConnection => {
   if (!redisConnectionInstance) {
@@ -19,3 +25,17 @@ export const getPostgresPool = (maxConnections?: number): Pool => {
   }
   return pgPoolInstance
 }
+
+export const getClickhouseClient = (): ClickHouseClient => {
+  if (!clickhouseInstance) {
+    clickhouseInstance = createClickhouseClient()
+  }
+  return clickhouseInstance
+}
+
+export const getStorageDisk = (): StorageDisk => {
+  if (!storageDiskInstance) {
+    storageDiskInstance = createStorageDisk()
+  }
+  return storageDiskInstance
+}

apps/workers/src/server.ts

@@ -4,8 +4,9 @@ import { createPollingOutboxConsumer } from "@platform/events-outbox"
 import { createBullmqEventsPublisher } from "@platform/queue-bullmq"
 import { createLogger } from "@repo/observability"
 import { config as loadDotenv } from "dotenv"
-import { getPostgresPool, getRedisConnection } from "./clients.ts"
+import { getClickhouseClient, getPostgresPool, getRedisConnection, getStorageDisk } from "./clients.ts"
 import { createEventsWorker } from "./workers/events.ts"
+import { createSpanIngestionWorker } from "./workers/span-ingestion.ts"
 
 const nodeEnv = process.env.NODE_ENV || "development"
 const envFilePath = fileURLToPath(new URL(`../../../.env.${nodeEnv}`, import.meta.url))
@@ -27,6 +28,12 @@ const outboxConsumer = createPollingOutboxConsumer(
   eventsPublisher,
 )
 
+const { queue: spanIngestionQueue, worker: spanIngestionWorker } = createSpanIngestionWorker({
+  redisConnection,
+  clickhouseClient: getClickhouseClient(),
+  storageDisk: getStorageDisk(),
+})
+
 const logger = createLogger("workers")
 
 eventsWorker.on("ready", () => {
@@ -35,10 +42,20 @@ eventsWorker.on("ready", () => {
   logger.info("workers ready and outbox consumer started")
 })
 
+spanIngestionWorker.on("ready", () => {
+  logger.info("span ingestion worker ready")
+})
+
+spanIngestionWorker.on("failed", (job, error) => {
+  logger.error({ jobId: job?.id, storageKey: job?.data?.storageKey, error }, "span ingestion job failed")
+})
+
 process.on("SIGINT", async () => {
   await outboxConsumer.stop()
   await pgPool.end()
   await eventsQueue.close()
   await eventsWorker.close()
+  await spanIngestionQueue.close()
+  await spanIngestionWorker.close()
   process.exit(0)
 })

apps/workers/src/workers/span-ingestion.ts

@@ -0,0 +1,62 @@
+import type { ClickHouseClient } from "@clickhouse/client"
+import { type OrganizationId, SPAN_INGESTION_QUEUE, type StorageDiskPort } from "@domain/shared"
+import type { OtlpExportTraceServiceRequest, TransformContext } from "@domain/spans"
+import { SpanRepository, transformOtlpToSpans } from "@domain/spans"
+import type { RedisConnection } from "@platform/cache-redis"
+import { ChSqlClientLive, SpanRepositoryLive } from "@platform/db-clickhouse"
+import { type Job, Queue, Worker } from "bullmq"
+import { Effect } from "effect"
+
+interface StoredPayload {
+  readonly request: OtlpExportTraceServiceRequest
+  readonly context: Omit<TransformContext, "ingestedAt">
+  readonly ingestedAt?: string
+}
+
+interface SpanIngestionWorkerConfig {
+  readonly redisConnection: RedisConnection
+  readonly clickhouseClient: ClickHouseClient
+  readonly storageDisk: StorageDiskPort
+}
+
+const createProcessor =
+  ({ clickhouseClient, storageDisk }: { clickhouseClient: ClickHouseClient; storageDisk: StorageDiskPort }) =>
+  async (job: Job<{ storageKey: string }>) => {
+    const { storageKey } = job.data
+
+    const raw = await storageDisk.get(storageKey)
+    const { request, context, ingestedAt } = JSON.parse(raw) as StoredPayload
+
+    const spans = transformOtlpToSpans(request, {
+      ...context,
+      ingestedAt: ingestedAt ? new Date(ingestedAt) : new Date(),
+    })
+
+    if (spans.length > 0) {
+      await Effect.runPromise(
+        Effect.gen(function* () {
+          const repo = yield* SpanRepository
+          yield* repo.insert(spans)
+        }).pipe(
+          Effect.provide(SpanRepositoryLive),
+          Effect.provide(ChSqlClientLive(clickhouseClient, context.organizationId as OrganizationId)),
+        ),
+      )
+    }
+
+    await storageDisk.delete(storageKey)
+  }
+
+export const createSpanIngestionWorker = ({
+  redisConnection,
+  clickhouseClient,
+  storageDisk,
+}: SpanIngestionWorkerConfig) => {
+  const queue = new Queue(SPAN_INGESTION_QUEUE, { connection: redisConnection })
+
+  const worker = new Worker(SPAN_INGESTION_QUEUE, createProcessor({ clickhouseClient, storageDisk }), {
+    connection: redisConnection,
+  })
+
+  return { queue, worker }
+}

packages/domain/shared/src/index.ts

@@ -1,6 +1,7 @@
 export * from "./ch-sql-client.ts"
 export * from "./errors.ts"
 export * from "./id.ts"
+export * from "./queues.ts"
 export * from "./seeds.ts"
 export * from "./sql-client.ts"
 export * from "./storage.ts"

packages/domain/shared/src/queues.ts

@@ -0,0 +1 @@
+export const SPAN_INGESTION_QUEUE = "span-ingestion" as const