fix(resource-manager): prevent shutdown queue deadlock - #1800
Conversation
|
@claude review |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 685d7f79b1
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| self._shutdown = True | ||
| if self._instances.get(self.public_key) is self: | ||
| self._instances.pop(self.public_key) | ||
| self._media_manager.begin_shutdown() |
There was a problem hiding this comment.
When shutdown begins with pending OpenTelemetry spans containing base64 media attributes, begin_shutdown() runs before flush(). The transforming span exporter therefore sees the media manager as shut down, returns the original data URI, and exports the span without queuing its media upload; large payloads may also cause the span export to be rejected. Media admission should remain enabled through the tracer-provider flush, then be closed before joining the media queue.
AGENTS.md reference: AGENTS.md:L134-L134
Useful? React with 👍 / 👎.
| self._api_client = api_client | ||
| self._httpx_client = httpx_client | ||
| self._queue = media_upload_queue | ||
| with self._state_lock: |
There was a problem hiding this comment.
If another thread holds _state_lock when a preloaded process forks, the child inherits the locked state without the owning thread, and its registered _at_fork_reinit() handler blocks forever here while calling reinitialize(). The resource manager already replaces its class lock for this reason; the media manager's newly introduced lock also needs to be replaced in the child before it is acquired.
AGENTS.md reference: AGENTS.md:L134-L134
Useful? React with 👍 / 👎.
| if self._shutdown: | ||
| return |
There was a problem hiding this comment.
When two wrappers sharing this manager call shutdown() concurrently, the second caller observes _shutdown immediately after the first sets it and returns even though the first may still be flushing network batches or joining workers. Code relying on shutdown() having completed can then tear down dependencies or exit while export is still underway; idempotence should make later callers wait for completion rather than return on the start-state flag.
AGENTS.md reference: AGENTS.md:L134-L134
Useful? React with 👍 / 👎.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| mask_otel_spans=mask_otel_spans, | ||
| ) | ||
| tracer_provider.add_span_processor(langfuse_processor) | ||
| self._span_processor = langfuse_processor | ||
|
|
||
| self._otel_tracer = tracer_provider.get_tracer( | ||
| LANGFUSE_TRACER_NAME, |
There was a problem hiding this comment.
🔴 This PR's new eviction logic in __new__ lets a shut-down LangfuseResourceManager for a given public_key be replaced by a fresh one, but shutdown() only stops the old LangfuseSpanProcessor — it is never detached from the shared global TracerProvider (OTel has no remove_span_processor API), so each shutdown+recreate cycle for the same key permanently appends another dead processor to the provider's processor list. This is an unbounded leak in exactly the flow this PR targets (repeated same-key client recreation in tests/app restarts), and it also means every subsequent span export/force_flush call additionally iterates over the accumulated dead processors.
Extended reasoning...
The bug: _init_tracer_provider() (bottom of resource_manager.py) only constructs a new TracerProvider when the OTel global default is still a ProxyTracerProvider. The very first Langfuse client created in a process calls otel_trace_api.set_tracer_provider(provider), which is a one-time, non-overridable action in the OTel SDK. Every subsequent call to _init_tracer_provider() — for the same public_key or a different one — hits the else branch and simply returns that one shared global provider.
In _initialize_instance() (lines 262-274), each time a LangfuseResourceManager is constructed for a key with tracing_enabled=True, a brand-new LangfuseSpanProcessor is created and appended to that shared provider via tracer_provider.add_span_processor(langfuse_processor). add_span_processor on OTel's SDK TracerProvider only ever appends to an internal list; there is no public API to remove an entry once added.
shutdown() (lines ~648-660) sets self._shutdown = True, evicts the instance from _instances, flushes, joins the consumer threads, and finally calls self._span_processor.shutdown(). That stops the BatchSpanProcessor background thread and shuts down its exporter, but it does not call anything on self.tracer_provider to detach/remove the processor — and there is no such API to call. The dead processor object stays registered on the shared provider forever.
Why this PR changes the picture: Before this PR, __new__ returned the cached instance for a public_key unconditionally (if public_key in cls._instances: return cls._instances[public_key]), even if it had already been shut down. So calling Langfuse(public_key=pk) again after shutdown() never constructed a new manager and never added a second processor — the leak path was unreachable for the same key. This PR's whole point is to change that: __new__ now pops a shut-down instance out of _instances and builds a genuinely fresh LangfuseResourceManager, which runs _initialize_instance() again and appends a brand-new LangfuseSpanProcessor to the shared provider. This is precisely the scenario exercised by the PR's own new test, test_shutdown_evicts_manager_and_rejects_stale_client_tasks (shutdown → construct a fresh client for the same key), and it's the pytest/app-restart use case the PR description says it is fixing.
Step-by-step proof:
- Process starts, first
Langfuse(public_key="pk")is created anywhere._init_tracer_provider()sees aProxyTracerProviderdefault, so it createsprovider_Aand callsset_tracer_provider(provider_A)._initialize_instance()buildsprocessor_1and callsprovider_A.add_span_processor(processor_1). client.shutdown()is called._span_processor.shutdown()stopsprocessor_1's thread/exporter, butprovider_A._active_span_processor(its internal composite) still holdsprocessor_1.Langfuse(public_key="pk")is constructed again (e.g. a new pytest test, or an app restarting the client).__new__sees the cached instance's_shutdown == True, pops it from_instances, and builds a fresh manager._init_tracer_provider()now sees the global default isprovider_A(not aProxyTracerProvideranymore), so it returnsprovider_Aunchanged._initialize_instance()buildsprocessor_2and callsprovider_A.add_span_processor(processor_2).provider_Anow holds both the deadprocessor_1and the liveprocessor_2. Repeat steps 2-3 N times (e.g. N tests in a suite that each create/shutdown a same-key client) andprovider_Aaccumulates N dead processors that are never freed — each one still holding its exporter, HTTP client references, and internal buffers.- Every span emitted afterward, and every
force_flush()/shutdown ofprovider_A, now iterates over all N+1 processors, so the overhead (not just memory) grows with the number of create/shutdown cycles.
Impact: In a long-running process (e.g. a web app that reconstructs its Langfuse client on config reload) or in a test suite that repeatedly builds/tears-down same-key clients — exactly the pattern this PR's new test and its stated goal cover — this leaks a LangfuseSpanProcessor (plus its exporter and any queued references) per cycle, unboundedly. It's not a crash or incorrect trace data (the dead processors are harmless no-ops after shutdown()), but it is a genuine, PR-introduced resource leak in the exact code path this PR adds.
Suggested fix: Track the processor per-manager and either (a) reuse/replace it in place via a wrapper that supports swapping its inner processor without needing a new add_span_processor call, or (b) give each LangfuseResourceManager its own isolated TracerProvider instead of relying on the ambient global one when one isn't explicitly passed in, so a shut-down manager's provider (and all its processors) can simply be dropped and garbage collected.
| with self._lock: | ||
| if self._shutdown: | ||
| return | ||
|
|
||
| self._shutdown = True | ||
| if self._instances.get(self.public_key) is self: | ||
| self._instances.pop(self.public_key) | ||
| self._media_manager.begin_shutdown() | ||
|
|
||
| # Unregister the atexit handler first | ||
| atexit.unregister(self.shutdown) | ||
|
|
||
| self.flush() | ||
| self._stop_and_join_consumer_threads() | ||
| try: | ||
| self.flush() | ||
| finally: | ||
| self._stop_and_join_consumer_threads() |
There was a problem hiding this comment.
🔴 shutdown() now calls self._media_manager.begin_shutdown() before self.flush(), so any span still buffered in the BatchSpanProcessor at shutdown time has its media silently dropped instead of uploaded during the force-flush that is supposed to export it. This mainly affects media embedded via third-party OTEL instrumentation (raw base64 attributes only processed at export time), which is exactly the case flush()/shutdown() exists to protect. Move begin_shutdown() to run after flush() completes.
Extended reasoning...
The bug: In LangfuseResourceManager.shutdown() (resource_manager.py:636-651), self._media_manager.begin_shutdown() is invoked inside the very first locked block, immediately setting MediaManager._shutdown = True. Only afterward does self.flush() run, which calls self.tracer_provider.force_flush() to force the BatchSpanProcessor to export any spans that are still sitting in its internal buffer (i.e. spans that have not yet hit the normal batch-size/interval trigger).
The code path that triggers it: Export goes through LangfuseTransformingSpanExporter.export() -> _process_media_attributes() -> _process_media_attribute_value() -> MediaManager._find_and_process_media() (span_exporter.py:108-249). This PR added a guard at the top of _find_and_process_media (media_manager.py:111-116) that checks self._shutdown and, if true, returns the data completely unprocessed with only a warning log — no extraction of the base64 payload, no LangfuseMedia object created, no enqueue onto the media upload queue. Because begin_shutdown() already ran before flush() triggers this export, every span force-flushed during shutdown hits this early return.
Why nothing else prevents it: The media upload consumer threads are still alive at this point — they are only paused/joined later, in _stop_and_join_consumer_threads(), which shutdown() calls in the finally block after flush(). So structurally there is no reason the media couldn't be uploaded; the only thing blocking it is that _find_and_process_media now refuses to even enqueue the job because _shutdown was flipped too early. Before this PR, MediaManager had no _shutdown flag at all, so this force_flush-triggered processing worked correctly and the subsequent self._media_upload_queue.join() (still inside flush()) would wait for the real upload to complete.
Scope: This does not affect the common case of media created via the Langfuse SDK's own span/generation API, since _process_media_and_apply_mask already converts base64 into a @@@langfuseMedia:...@@@ reference and enqueues the upload synchronously at span-creation time (span.py:603) — by the time export runs, _find_and_process_media sees the reference token and no-ops regardless of _shutdown. The affected path is media that only becomes visible at export time: raw base64 attributes set by third-party OTEL instrumentation (e.g. OpenInference/OpenLLMetry vision instrumentation) that bypass the Langfuse SDK's span-creation hooks entirely. For those spans, export is the only place media is ever extracted, and that is precisely the moment this PR now blocks.
Step-by-step proof:
- A third-party OTEL instrumentation library creates an OTEL span with an attribute containing a raw base64 image (a
data:image/png;base64,...string), without going through Langfuse's SDK-level span API. - This span has not yet been auto-flushed by the BatchSpanProcessor (still within its batch buffer) when the application calls
client.shutdown(). shutdown()acquiresself._lock, setsself._shutdown = True, and callsself._media_manager.begin_shutdown(), which setsMediaManager._shutdown = True— all before any flush has occurred.shutdown()callsself.flush(), which callsself.tracer_provider.force_flush(), forcing the BatchSpanProcessor to export the buffered span immediately.- Export reaches
LangfuseTransformingSpanExporter.export()->_process_media_attributes->MediaManager._find_and_process_media()for the base64 attribute. - The new guard at the top of
_find_and_process_mediaseesself._shutdown is Trueand returns the raw data unchanged, logging only a warning — noUploadMediaJobis ever created or enqueued. flush()then callsself._media_upload_queue.join(), which returns immediately since nothing was ever enqueued for this media.- The span is exported to Langfuse with the raw, un-uploaded base64 instead of a Langfuse media reference — the image is never uploaded to media storage, a silent data loss that the flush/shutdown mechanism exists specifically to prevent.
The fix: Move the self._media_manager.begin_shutdown() call to run after self.flush() completes (e.g. into the finally block alongside _stop_and_join_consumer_threads()), so that any force-flushed spans are still processed for media before the manager is marked as shut down.
| ) | ||
|
|
||
| if should_sample: | ||
| langfuse_logger.debug( | ||
| f"Score: Enqueuing event type={event['type']} for trace_id={event['body'].trace_id} name={event['body'].name} value={event['body'].value}" | ||
| ) | ||
| self._score_ingestion_queue.put(event, block=False) | ||
| with self._lock: | ||
| if self._shutdown: | ||
| langfuse_logger.warning( | ||
| "Score: Dropping event because the Langfuse client has already been shut down." | ||
| ) | ||
| return | ||
|
|
||
| langfuse_logger.debug( | ||
| f"Score: Enqueuing event type={event['type']} for trace_id={event['body'].trace_id} name={event['body'].name} value={event['body'].value}" | ||
| ) | ||
| self._score_ingestion_queue.put(event, block=False) | ||
|
|
||
| except Full: | ||
| langfuse_logger.warning( |
There was a problem hiding this comment.
🟡 add_score_task/add_trace_task (lines 518, 548) now do with self._lock:, but LangfuseResourceManager never defines an instance-level self._lock — only the class attribute _lock = threading.RLock() exists (shared across all clients). This makes every score/trace enqueue in the process contend on the same lock that __new__ holds for the full duration of constructing a brand-new manager for any public_key, and that shutdown()/reset() hold while evicting instances — a cross-client contention point that didn't exist before this PR. Recommend using a dedicated per-instance lock (e.g. self._state_lock) instead.
Extended reasoning...
The bug: LangfuseResourceManager only ever defines _lock as a class attribute — _lock = threading.RLock() at the class body level, later reassigned at the class level again in _at_fork_reinit (LangfuseResourceManager._lock = threading.RLock()). No _initialize_instance (or anywhere else) ever assigns an instance-level self._lock. That means when the new code in add_score_task (line 518) and add_trace_task (line 548) does with self._lock: to guard the shutdown check + queue.put, Python attribute lookup falls through to the class attribute — so self._lock resolves to the exact same RLock object used by __new__ (line ~137) to guard the entire _instances singleton registry, and used by reset() (line ~484) and shutdown() (line ~636).\n\nWhy this matters: __new__ acquires cls._lock and holds it for the entire duration of constructing a brand-new resource manager when a client for a new (or evicted) public_key is created — this includes _initialize_instance, which builds httpx/OTEL clients, sets up the span processor, and starts consumer threads (thread.start()). Before this PR, add_score_task/add_trace_task took no lock at all; each client's own Queue already provides thread-safe enqueue, so different clients' enqueues were fully independent. After this PR, every score/trace enqueue for every client in the process now contends on this single process-wide lock.\n\nConcrete walkthrough: Suppose a process has two Langfuse clients, A (public_key="a") and B (public_key="b"), both already constructed and issuing scores continuously. Now a third client C is being constructed for public_key="c" (e.g., a multi-tenant server discovering a new tenant, or a pytest suite that creates+shuts down clients between tests). Thread 1 calls LangfuseResourceManager(public_key="c", ...), enters __new__, acquires cls._lock, and begins _initialize_instance — creating httpx clients, setting up the OTEL tracer/span processor, and starting the consumer thread. This can take non-trivial wall-clock time (network-adjacent setup, thread spawn). Meanwhile, thread 2 calls client_a.create_score(...), which eventually reaches add_score_task, which does with self._lock: — but self._lock for client A is the same object as cls._lock currently held by thread 1's construction of client C. Thread 2 blocks until client C's construction finishes, even though client A's queue and consumer are completely unrelated and idle. The same happens for client B's enqueues, and for any enqueue happening while shutdown() or reset() is running for an unrelated key.\n\nWhy nothing currently prevents this: the per-instance Queue objects (_score_ingestion_queue) are already internally synchronized and were sufficient for safe concurrent enqueue before this PR; the new locking was added only to make the shutdown-check-then-enqueue sequence atomic per-instance (to fix the real bug this PR targets — enqueuing into a queue whose consumer has already stopped). That's a legitimate goal, but it should use a lock scoped to the instance, not the class-level singleton-registry lock. Since self._lock was never assigned per-instance, the code accidentally reuses the class lock.\n\nImpact: this is a contention/latency regression, not a correctness break — queue.put(..., block=False) cannot deadlock, and the critical sections in __new__/shutdown/reset are typically short-lived and infrequent relative to the hot-path enqueue calls. In single-client deployments the effect is negligible since construction precedes any enqueues. But in multi-tenant setups or test suites that repeatedly construct/shut down clients concurrently with active traffic on other keys, this reintroduces a global choke point on the ingestion hot path that the pre-PR code never had.\n\nFix: add a dedicated per-instance lock in _initialize_instance (e.g. self._state_lock = threading.Lock()) and use that in add_score_task/add_trace_task/shutdown instead of self._lock, decoupling the per-instance shutdown-guard from the class-level singleton-registry lock.
1 participant
Summary
Root cause
LangfuseResourceManagerwas cached by public key after shutdown. Existing wrappers could enqueue work after its consumers had stopped, leavingQueue.unfinished_tasksnon-zero and causing a laterflush()orshutdown()to block forever inQueue.join().User impact
Independent application or pytest instances using the same public key can now shut down without deadlocking. A client constructed after shutdown receives a fresh resource manager with live workers; stale wrappers drop asynchronous work with a warning instead of queueing work that cannot be consumed.
Validation
LANGFUSE_PUBLIC_KEY=test-public-key LANGFUSE_SECRET_KEY=test-secret-key LANGFUSE_BASE_URL=http://localhost:9 UV_CACHE_DIR=/tmp/langfuse-python-uv-cache bash scripts/codex/quick-check.sh(677 passed, 2 skipped)UV_CACHE_DIR=/tmp/langfuse-python-uv-cache uv run --frozen pytest tests/unit/test_media.py -q(19 passed)unfinished 0, creates a fresh manager with a live consumer, and returns from shutdownLinear: LFE-14771
Closes #1799
Greptile Summary
The PR prevents stale shared resource managers from accepting asynchronous work after shutdown and allows same-key clients to reconstruct fresh workers.
Confidence Score: 5/5
The PR appears safe to merge, with shutdown admission and resource reconstruction consistently synchronized across the changed paths.
Shutdown now marks and evicts the old manager before draining work, stale score, trace, and media producers reject new queue entries, and subsequent same-key clients receive newly initialized workers.
Flowchart
%%{init: {'theme': 'neutral'}}%% flowchart TD A[Shared active resource manager] --> B[Client calls shutdown] B --> C[Mark manager and media pipeline shut down] C --> D[Evict manager from singleton cache] D --> E[Flush existing queues] E --> F[Stop consumer threads and span processor] C --> G[Stale wrapper submits asynchronous work] G --> H[Drop work with warning] D --> I[New same-key client is constructed] I --> J[Create fresh manager with live workers]Reviews (1): Last reviewed commit: "fix(resource-manager): prevent shutdown ..." | Re-trigger Greptile
Context used (3)