[DavidTraina]

What does this PR do?

Calling shutdown() more than once on a client, one call after another or from two threads at once, makes the next flush() or shutdown() block forever.

import threading

from langfuse import Langfuse

langfuse = Langfuse(
    public_key="pk-lf-test", secret_key="sk-lf-test", base_url="http://localhost:3000"
)
langfuse.shutdown()
langfuse.shutdown()

flush = threading.Thread(target=langfuse.flush, daemon=True)
flush.start()
flush.join(timeout=3)
print("flush() returned:", not flush.is_alive())  # main: False, this branch: True

You don't need to call shutdown() twice on the same object to hit this. After shutdown(), Langfuse(...) with the same public key and get_client() return a client backed by the same shut-down resource manager. So code that creates a client and calls shutdown() at the end of each request or job hangs on the third request.

Cause. shutdown() stops the media upload consumers by putting one shutdown sentinel per consumer on the media upload queue. Each sentinel counts as an unfinished task until a consumer takes it and calls task_done(). The first shutdown()'s sentinels are taken and the consumers exit. A second or concurrent shutdown() puts more sentinels on a queue that no consumer reads anymore. unfinished_tasks never returns to 0, so the _media_upload_queue.join() in flush() waits forever. The score queue is not affected, because its wait already stops once no score consumer is alive.

Fix. MediaManager.signal_shutdown() marks each sentinel done right after putting it, and the consumer no longer calls task_done() when it takes one. A sentinel is a wake-up signal, not an upload, so join() no longer waits for it. Uploads are counted exactly as before. The change uses only put() and task_done(), so the media queue behaves the same with gevent's monkey-patched Queue.

This also fixes a sentinel left behind by a single shutdown(). If an upload is enqueued between shutdown()'s own flush() and its shutdown signal, the consumer finishes that upload, sees that it has been paused, and exits without taking its sentinel. On main, every later flush() then hangs as well.

Unchanged

• The first shutdown() still delivers pending spans, media (create, upload, status update) and scores, and a consumer blocked on get() still wakes immediately.
• A later shutdown() still force-flushes spans created after the first one.
• Fork reinit, the atexit handler, get_client(), re-creating a client, and LangfuseResourceManager.reset().
• Still open, as on main: media logged after shutdown() has signalled the consumers is never uploaded, so flush() still waits for it. That needs a separate change.

Considered and not done

• Signalling only consumers that are still alive, under a new lock in shutdown(). This also fixes both reported cases. But _at_fork_reinit would have to replace the lock, because a fork while another thread holds it would deadlock the child. An existing test's fake consumers would need changes. And it misses the single-shutdown() case above.
• Making shutdown() run only once. A later shutdown() would then stop flushing spans created after the first one, and code that calls shutdown() at the end of every request relies on that.
• Making flush() stop waiting on the media queue once no media consumer is alive, as it does for scores. This would also cover media logged after shutdown, which would then be dropped instead of hanging.
  • A wait that returns as soon as the queue drains needs the stdlib queue's undocumented all_tasks_done. gevent's Queue lacks it and needs join(timeout=...) instead, so that means two code paths.
  • With only the public API, the stdlib path has to poll. A 0.1 s poll added about 80 ms (median, with a 20 ms upload) to each flush() with pending media in a local benchmark.
  • It also leaves the extra sentinels counted.

Type of change

• Bug fix
• New feature
• Breaking change
• Refactor
• Documentation update
• Tooling, CI, or repo maintenance

Verification

List the main commands you ran:

uv run --frozen ruff check .                                # All checks passed!
uv run --frozen ruff format --check langfuse/_task_manager/media_manager.py tests/unit/test_resource_manager.py  # 2 files already formatted
uv run --frozen mypy langfuse --no-error-summary            # no errors
uv run --frozen pytest -n auto --dist worksteal tests/unit  # 951 passed, 2 skipped (Python 3.14)
UV_PROJECT_ENVIRONMENT=/tmp/lf-py311 uv run --frozen --python 3.11 pytest tests/unit/test_resource_manager.py \
  tests/unit/test_score_consumer.py tests/unit/test_media_manager.py tests/unit/test_media.py  # 78 passed

• New test. tests/unit/test_resource_manager.py::test_flush_returns_after_repeated_shutdown[sequential|concurrent] runs two shutdown() calls, then flush(), each on a thread with join(timeout=...). An unhandled exception in any thread also fails it.
  • On main both cases fail (flush() is still blocked after 2 s). Teardown still finishes, because the test acknowledges the leftover sentinels.
  • On this branch both pass, 30 of 30 runs.
• Stress, stdlib threads. Each iteration runs 4 flush() and 2 shutdown() threads on one client, then one more flush(). Main hung in 5 of 5 iterations; this branch hung in 0 of 40.
• Real gevent 25.9.1 (monkey.patch_all(), Python 3.14, queue.Queue is gevent's Queue). On main and on this branch, flush() under gevent fails earlier, in the score-queue wait (all_tasks_done does not exist on gevent's Queue). This PR does not touch that wait.
  • To check the media queue anyway, the script below swaps in a gevent-safe score wait for the check only. Main prints False twice; this branch prints True twice.
  • With GEVENT_MONKEY_DISABLE_QUEUE_QUEUE=true (stdlib Queue on patched threading) and no stand-in, this branch also returns in the sequential, concurrent and 4 flush + 2 shutdown cases.

gevent check

from gevent import monkey

monkey.patch_all()

import gevent

from langfuse import Langfuse
from langfuse._client.resource_manager import LangfuseResourceManager


# Stand-in for the score-queue wait, which uses Queue.all_tasks_done and fails
# under gevent. It keeps that failure from hiding the media queue.
def join_score_queue(self):
    q = self._score_queue
    while q.unfinished_tasks and any(c.is_alive() for c in self._score_consumers):
        q.join(timeout=0.1)


LangfuseResourceManager._join_score_queue = join_score_queue


def returns(fn):
    greenlet = gevent.spawn(fn)
    greenlet.join(timeout=3)
    return greenlet.successful()


def client(public_key):
    return Langfuse(
        public_key=public_key, secret_key="sk-lf-test", base_url="http://localhost:3000"
    )


langfuse = client("pk-lf-sequential")
langfuse.shutdown()
langfuse.shutdown()
print("flush() after 2 shutdowns returned:", returns(langfuse.flush))

langfuse = client("pk-lf-concurrent")
gevent.joinall([gevent.spawn(langfuse.shutdown) for _ in range(2)])
print("flush() after 2 concurrent shutdowns returned:", returns(langfuse.flush))

• Other probes, under threads and under gevent with the stand-in:
  • The first shutdown() still sends the pending span, media (POST, PUT, PATCH) and score.
  • A later shutdown() still exports spans created after the first one.
  • The single-shutdown() case above no longer leaves unfinished_tasks at 1.
  • reset() followed by a new client works.
  • Under threads: fork children (of a live and of a shut-down parent) and an atexit-only exit work.
• Skipped.
  • ruff format --check . on the whole repo reports langfuse/_client/resource_manager.py (trailing whitespace in flush()), on main as well. This PR does not touch that file.
  • E2E and live-provider suites: the change is covered by unit tests and needs no server.

Checklist

• I self-reviewed the diff using code_review.md.
• I added or updated tests for behavior changes.
• I updated docs, examples, or .env.template if needed. (No public API change; nothing to update.)
• I did not hand-edit generated files; if generated files changed, I used the upstream regeneration path.
• I did not commit secrets or credentials.

Confidence Score: 5/5

The PR appears safe to merge; no actionable issues were found.

Summary

Moves shutdown-sentinel task counting from the media consumer to signal_shutdown(). Repeated shutdown calls no longer leave unfinished signals that block flush().

• Media flushes return after clients are shut down more than once.

_{Reviews (1) · Last reviewed commit: "test(media): fail the repeated-shutdown ..." · Reviewed by Greptile}

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.