Langfuse.shutdown() hangs forever on _score_ingestion_queue.join() — cached LangfuseResourceManager is reused after shutdown

Summary

Langfuse.shutdown() (and flush()) can block forever on self._score_ingestion_queue.join() in LangfuseResourceManager.flush().

Root cause: LangfuseResourceManager instances are cached process-wide by public_key (LangfuseResourceManager._instances), but shutdown() does not remove the instance from the cache. Any Langfuse client constructed afterwards with the same public_key silently receives the already-shut-down manager: its ingestion consumer threads are dead, so any event enqueued after that point (e.g. create_score(), which checks neither _shutdown nor consumer liveness) is never consumed, Queue.unfinished_tasks never reaches 0, and the next flush()/shutdown() blocks forever on Queue.join().

Minimal reproduction

import faulthandler
import threading

from langfuse import Langfuse

faulthandler.dump_traceback_later(10, exit=True)

COMMON_KWARGS = dict(
    public_key='pk-lf-repro',
    secret_key='sk-lf-repro',
    host='http://localhost:9',  # unreachable on purpose; nothing is sent in this repro
)

client_a = Langfuse(**COMMON_KWARGS)
client_b = Langfuse(**COMMON_KWARGS)

assert client_a._resources is client_b._resources  # same cached resource manager

client_a.shutdown()  # stops the shared ingestion consumer threads for good

client_b.create_score(name='quality', value=1.0)  # enqueued, but no consumer is alive anymore

print('threads alive:', [t.name for t in threading.enumerate()], flush=True)
print('score queue unfinished tasks:', client_b._resources._score_ingestion_queue.unfinished_tasks, flush=True)

client_b.shutdown()  # hangs forever on _score_ingestion_queue.join()

print('UNREACHABLE: shutdown returned', flush=True)

Output (the watchdog kills the process after 10 s; without it the hang is indefinite):

threads alive: ['MainThread', 'OtelBatchSpanRecordProcessor', 'Thread-3']
score queue unfinished tasks: 1
Timeout (0:00:10)!
...
Thread 0x00000001f38bde80 (most recent call first):
  File ".../python3.12/threading.py", line 355 in wait
  File ".../python3.12/queue.py", line 90 in join
  File ".../langfuse/_client/resource_manager.py", line 608 in flush
  File ".../langfuse/_client/resource_manager.py", line 620 in shutdown
  File ".../langfuse/_client/client.py", line 2340 in shutdown
  File "langfuse_deadlock_repro.py", line 34 in <module>

Note that no IngestionConsumer threads are alive when client_b enqueues the score — they were joined by client_a.shutdown().

How we hit this in practice

In a pytest suite, several independently built application instances each create their own Langfuse client with the same public_key and shut it down in their teardown. Because all of them share the single cached LangfuseResourceManager, the first teardown kills the consumer threads for everyone; a later teardown then hangs the whole test process forever on _score_ingestion_queue.join(). The same pattern can hang a production graceful shutdown if two clients with one key exist in a process (e.g. app + sidecar worker, or re-initialization after a config reload).

Expected behavior

Either of:

• shutdown() removes the instance from LangfuseResourceManager._instances, so the next Langfuse(...) with that key builds a fresh, working manager; and/or
• add_score_task() / add_trace_task() refuse to enqueue (or __new__ refuses to return a manager) once _shutdown is set; and/or
• flush() uses a bounded wait instead of a bare Queue.join(), so a lost consumer cannot block shutdown indefinitely.

Environment

• langfuse 4.14.1
• Python 3.12.x (CPython), macOS 15 (arm64) — also reproduced on Linux in CI
• No Langfuse server involved; the hang is purely client-side