Defer span processing to background worker via object storage

[csansoon]

Summary

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.

[csansoon]

@cursoragent review this PR

[cursor]

Now I have a thorough understanding of the entire PR. Let me write the review.

Here is my review of this PR:

---

PR Review: Defer span processing to background worker via object storage

Overall assessment: The architecture is sound -- decoupling ingest from processing via object storage + BullMQ is the right move for keeping the OTEL endpoint fast and making retries possible. The OTLP module relocation to @domain/spans is clean. There are several issues to address before merging, ranging from a leftover debug log to missing error handling and architectural concerns.

Must Fix

1. Leftover console.log in production code (packages/domain/spans/src/otlp/transform.ts:147)

   console.log("📥", span.name)

   This looks like a debug log that was not reverted before the PR. It will fire for every single span processed in production. Remove it.

2. protobufjs in a domain package violates architecture rules -- The AGENTS.md specifies "Domain must never import concrete DB/cache/queue/object storage clients" and more broadly, "Web Standards First" for domain packages. protobufjs is a Node.js-native library (uses Buffer, fs, path internally). The OTLP proto decoding is a platform-level concern (wire-format parsing), not business logic.

   • Consider keeping proto.ts in a platform package (e.g., @platform/otlp or keeping it in apps/ingest) and only moving transform.ts + types.ts into @domain/spans. The worker can import the decoder from the platform package.
   • Alternatively, if the team decides this is acceptable in domain, document why with a brief comment per the AGENTS.md convention.

3. Queue name "span-ingestion" is duplicated as a magic string in three places: apps/ingest/src/clients.ts:70, apps/workers/src/workers/span-ingestion.ts:45, and :47. Extract this to a shared constant (e.g., in @domain/shared or a shared constants module) to prevent silent mismatches.

4. No error handling on storageDisk.get() or JSON.parse() in the worker processor (apps/workers/src/workers/span-ingestion.ts:27-28):

   const raw = await storageDisk.get(storageKey)
   const { request, context } = JSON.parse(raw) as StoredPayload

   • storageDisk.get() can throw if the key doesn't exist (race condition, already deleted, storage failure).
   • JSON.parse(raw) will throw on corrupt data.
   • If these throw, BullMQ will retry the job (default 0 retries), but the error message won't be very useful. Wrap with typed errors or at least add structured logging so failures are diagnosable. Also consider configuring retry attempts on the worker.

5. storageDisk.delete() failure silently loses the span data (span-ingestion.ts:37). If ClickHouse insert succeeds but delete fails, the job throws and BullMQ may retry, potentially causing duplicate inserts. Consider either:

   • Making span inserts idempotent (e.g., via ReplacingMergeTree dedup, which ClickHouse likely already has), or
   • Catching the delete error separately and logging it without failing the job.

Should Fix

6. ingestAt timestamp drift -- The ingestedAt is set inside transformOtlpToSpans() at processing time (worker execution), not at actual HTTP ingest time. This could be seconds to minutes later under load. Consider capturing ingestedAt at the ingest endpoint and including it in the stored payload so the timestamp reflects when the data actually arrived.

7. getClickhouseClient is now unused in apps/ingest -- The import was removed from traces.ts, but getClickhouseClient() is still defined and exported in apps/ingest/src/clients.ts. If nothing else in the ingest app uses it, remove it (and the @clickhouse/client dependency from apps/ingest/package.json) to avoid dead code. Knip should catch this.

8. No BullMQ worker options configured -- The worker in span-ingestion.ts uses all BullMQ defaults (0 retries, no concurrency tuning, no rate limiting). For a span ingestion pipeline that will handle high throughput:

   • Set concurrency on the worker for parallel job processing.
   • Set attempts and backoff on jobs or worker defaults for transient ClickHouse/storage failures.
   • Consider removeOnComplete / removeOnFail to prevent Redis memory growth.

9. jobId: storageKey in traces.ts:52 makes jobs idempotent by storage key, which is good. However, each storage key contains a crypto.randomUUID(), so two identical payloads get two different jobs. This is probably fine (OTEL exporters retry with the same payload, but each HTTP request is a separate ingest), just flagging that the dedup is per-request, not per-content.

10. @domain/shared/storage.ts -- the StorageDiskPort interface only has put() but the worker needs get() and delete(). The actual StorageDisk (flydrive Disk) has these methods, but the domain port doesn't declare them. Either:

    • Expand StorageDiskPort to include get and delete, or
    • Acknowledge that the worker directly uses the platform StorageDisk type (which it does) and the port is only used by putInDisk.

Nits

11. The @domain/spans index exports OTLP types alongside domain entities. This mixes concerns slightly -- consumers looking at @domain/spans now see decodeOtlpProtobuf alongside SpanRepository. Consider a sub-path export (e.g., @domain/spans/otlp) if the package convention supports it.

12. The removed comments in transform.ts (e.g., // Semantic convention attribute keys..., // Split remaining attributes by value type) were actually useful for orientation in a dense transform function. The AGENTS.md says "avoid comments except for genuinely non-obvious reasoning," but attribute extraction from OTLP semantic conventions is arguably non-obvious to new readers. Consider keeping the most helpful ones.

13. Andres's review comment about using the centralized putInDisk from @domain/shared/storage.ts appears to have been addressed (the ingest namespace was added to storage.ts). Good.

---

Summary: The PR's architecture is solid. The main blockers are the leftover debug log, the duplicated queue name, and the lack of error handling in the worker processor. The protobufjs in domain is worth a discussion. The rest are improvements that would make the system more production-ready.