Principle:Langfuse Langfuse OTel S3 Upload and Queue Dispatch

Knowledge Sources	Langfuse
Domains	Ingestion, Queue Architecture, Blob Storage
Last Updated	2026-02-14 00:00 GMT

Overview

OTel S3 Upload and Queue Dispatch is the principle of durably persisting raw telemetry data to blob storage before enqueuing a lightweight job reference for asynchronous processing.

Description

In high-throughput ingestion systems, directly processing every incoming request synchronously creates backpressure that degrades API latency and risks data loss during worker failures. The S3 Upload and Queue Dispatch principle solves this by introducing a two-phase commit pattern:

1. Durable persistence: The raw parsed data (ResourceSpan[]) is serialized as JSON and uploaded to S3-compatible blob storage (MinIO locally, S3 in production). The S3 key follows a time-partitioned path format: otel/{projectId}/{yyyy/mm/dd/hh/mm}/{uuid}.json. This partitioning enables efficient lifecycle management, debugging, and potential replay of ingestion data.

1. Lightweight job enqueue: A BullMQ job is added to the OtelIngestionQueue containing only a reference to the S3 file key along with authentication context (projectId, publicKey, orgId). The job payload is deliberately minimal -- the actual data lives in S3, keeping queue memory usage low even under heavy load.

This decoupling yields several benefits:

• Durability: Once data reaches S3, it survives web server crashes, queue failures, and worker restarts. The S3 object serves as the source of truth.
• Backpressure management: The web server only performs a fast S3 upload and queue enqueue, returning to the client in milliseconds. Heavy processing (parsing, deduplication, ClickHouse writes) happens asynchronously in the worker.
• Horizontal scalability: Multiple worker instances can consume from the shared BullMQ queue, processing S3 files in parallel.
• Replayability: Failed jobs can be retried by re-downloading the original S3 file, without requiring the client to re-send data.

Usage

Use this pattern when:

• Ingestion volume is high and processing latency must not affect API response time.
• Raw payloads need to be preserved for auditing, debugging, or replay.
• Processing involves multiple downstream systems (ClickHouse, Redis, PostgreSQL) where partial failures need graceful recovery.
• The ingestion path must tolerate temporary unavailability of processing infrastructure.

Theoretical Basis

The mechanism follows the Store-and-Forward pattern common in message-oriented middleware:

RECEIVE parsed ResourceSpan[] from HTTP endpoint
    |
    v
GENERATE S3 KEY
    Key = "{prefix}otel/{projectId}/{yyyy/mm/dd/hh/mm}/{uuid}.json"
    - prefix: configurable via LANGFUSE_S3_EVENT_UPLOAD_PREFIX
    - projectId: scopes the file to a project
    - time path: current UTC time for partitioning
    - uuid: ensures uniqueness within the same minute
    |
    v
UPLOAD TO S3
    - Serialize ResourceSpan[] as JSON
    - Upload to configured S3 bucket (LANGFUSE_S3_EVENT_UPLOAD_BUCKET)
    |
    v
ENQUEUE BULLMQ JOB
    Job payload = {
        id: randomUUID(),
        timestamp: current time,
        name: "OtelIngestionJob",
        payload: {
            data: { fileKey, publicKey },
            authCheck: {
                validKey: true,
                scope: { projectId, accessLevel: "project", orgId }
            },
            propagatedHeaders: (optional, for ingestion masking)
        }
    }
    |
    v
RETURN Promise (resolves when job is enqueued)

Failure modes and handling:

• S3 upload failure: The method throws, causing the HTTP handler to return an error to the client. The client SDK retries the request.
• Queue unavailable: If OtelIngestionQueue.getInstance({}) returns null, the method rejects with "Failed to instantiate otel ingestion queue", signaling infrastructure misconfiguration.
• Worker processing failure: BullMQ handles retries with configurable backoff. The S3 file persists regardless of job outcome.

S3 key design rationale:

The time-partitioned key structure enables S3 lifecycle policies to automatically expire old ingestion files, reducing storage costs without manual cleanup. The per-minute granularity balances between excessive directory depth and useful temporal grouping for debugging.

Related Pages

Implemented By

• Implementation:Langfuse_Langfuse_PublishToOtelIngestionQueue