[DavidTraina]

What does this PR do?

Since v5.0.3 (#2025), flush() and shutdown() raise AttributeError when gevent or eventlet has monkey-patched the process:

Traceback (most recent call last):
  File "langfuse/_client/resource_manager.py", line 632, in shutdown
    self.flush()
  File "langfuse/_client/resource_manager.py", line 600, in flush
    self._join_score_queue()
  File "langfuse/_client/resource_manager.py", line 614, in _join_score_queue
    with score_queue.all_tasks_done:
AttributeError: 'gevent._gevent_cqueue.Queue' object has no attribute 'all_tasks_done'

_join_score_queue() waits on all_tasks_done, an undocumented Condition of the standard library queue.Queue. Since gevent 25.4.1, monkey.patch_all() replaces queue.Queue with gevent's own Queue, and eventlet's monkey_patch() replaces it too. Neither replacement has all_tasks_done. On an app server running gevent workers, this raised on every flush() and on every worker exit. shutdown() raised before _stop_and_join_consumer_threads(), so the score consumer was never stopped and queued scores were not sent. v5.0.2 called Queue.join(), which all three queues provide.

The fix:

• ScoreConsumer notifies a threading.Condition owned by the resource manager after task_done().
• _join_score_queue() waits on that Condition, with the same 0.1 s timeout and consumer-liveness check as before. It no longer uses all_tasks_done, so the same code runs on every queue type. It still reads unfinished_tasks, which all three queues have.
• gevent and eventlet patch threading, so the wait is cooperative whenever the consumer is a greenlet.

#2025's two guarantees hold for every queue type:

• flush() still wakes the consumer and returns as soon as the queue drains, because the consumer's notify wakes it. It does not wait for flush_interval.
• flush() still cannot hang when the consumers stop first. It rechecks every 0.1 s whether a consumer is alive, even when a consumer dies without notifying, as when its greenlet is killed.

Considered and not done:

• gevent's own Queue.join(timeout=...) when the queue has no all_tasks_done. It is the smallest change, but eventlet's join() takes no timeout, so it would raise TypeError there. With gevent 25.9.1 it also froze the whole process when flush() ran on gevent's hub threadpool with a score queued: the consumer's task_done() spun in gevent's cross-thread wakeup while holding the GIL. gevent 26.9.0 did not freeze.
• Polling unfinished_tasks with a short sleep when the queue has no all_tasks_done. It works everywhere, but adds up to one poll interval to every flush() with pending work. With a 5 ms interval, a flush() with one score over a local HTTP server took 6.0-6.4 ms under gevent, against 0.7 ms with this PR.
• Running the standard library's Queue.join() on a helper thread and waiting on that thread with a timeout. That needs only the public queue API, but it starts a thread on every flush(), which raised the median empty flush() from about 9 us to 44 us. It also leaves the helper blocked in join() when every consumer stops first.
• Building the SDK's queues from gevent's saved original queue.Queue. That only covers gevent, so eventlet stays broken, and it opts out of the queue gevent adopted to fix a urllib3 deadlock.
• A unit test under real gevent. That would add gevent, greenlet, zope.event and zope.interface to the dev group, and it would need a subprocess because monkey patching is process-wide. The unit tests use a queue double instead, and I ran the gevent script below by hand.

Workarounds on v5.0.3 until a release:

• Pin v5.0.2. Its flush() works under gevent and eventlet, but waits up to flush_interval for pending scores.
• Under gevent, set GEVENT_MONKEY_DISABLE_QUEUE_QUEUE=true. It keeps the standard library queue.Queue for the whole process, which brings back the queue gevent replaced to fix a urllib3 deadlock. gevent marks this switch as temporary.

#2031 adds the same all_tasks_done wait for the media upload queue. It would raise the same error under gevent.

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 .                          # 575 files already formatted
uv run --frozen mypy langfuse --no-error-summary               # no errors
uv run --frozen pytest tests/unit/test_score_consumer.py tests/unit/test_resource_manager.py tests/unit/test_experiment.py  # 87 passed
uv run --frozen pytest -n auto --dist worksteal tests/unit     # 954 passed, 2 skipped

The three targeted files also pass on Python 3.11.

The three flush tests from #2025 now run twice: once with the standard library Queue and once with GeventLikeQueue, a test double that wraps a standard library Queue and hides all_tasks_done. On main, the gevent cases fail with AttributeError: all_tasks_done.

The new test_flush_returns_when_consumer_stops_while_flush_waits also runs for both queue types. It blocks the consumer in an upload while flush() waits, then stops the consumer without a final notify and with an item still queued, and checks that flush() returns. It fails if the wait has no timeout or skips the liveness check.

ruff format, which pre-commit runs, also removes the trailing whitespace that #2025 left in flush().

flush() latency with flush_interval=5, on Python 3.14 and macOS arm64. Each cell is the range of the median across 2-3 runs.

Scenario	main	this PR
Standard library, one score, local HTTP server	0.68-0.70 ms	0.67-0.77 ms
Standard library, nothing queued, mock transport	13.1-13.3 us	12.8 us
gevent 25.9.1, one score, local HTTP server	AttributeError	0.67-0.78 ms
gevent 25.9.1, nothing queued, mock transport	AttributeError	11.5-11.7 us

Manual gevent check with the script below, run against gevent 25.9.1 and 26.9.0:

Scenario	main	this PR
Unreachable host: flush()	AttributeError	returns after the consumer gives up, as without gevent
Unreachable host: shutdown()	AttributeError, consumer still alive	returns, consumer stopped
Local server returning 202, flush_interval=5: flush() with one score	AttributeError	returns in about 3 ms, score posted
flush() with nothing queued	AttributeError	returns in under 0.1 ms
Consumer stops while flush() waits, leaving an item unprocessed	AttributeError	returns as soon as the consumer has stopped
flush() on gevent's hub threadpool, one score queued	AttributeError	returns, score posted

The same scenarios, apart from the threadpool one, pass under eventlet 0.41.3, where main raises the same AttributeError. I also checked that flush() returns with monkey.patch_all(time=False), with time.sleep mocked, from a native OS thread and at interpreter exit. A greenlet running alongside flush() keeps being scheduled while flush() waits on a slow POST.

Mixing threads is still limited, as on v5.0.2: after a flush() on gevent's threadpool and another on the main thread, in either order, the second raises gevent's InvalidThreadUseError from the media queue's join().

gevent check script

Run it with uv run --frozen --with gevent python gevent_check.py from the checkout.

from gevent import monkey

monkey.patch_all()

import http.server  # noqa: E402
import threading  # noqa: E402
import time  # noqa: E402

from langfuse import Langfuse  # noqa: E402
from langfuse._client.resource_manager import LangfuseResourceManager  # noqa: E402


class Handler(http.server.BaseHTTPRequestHandler):
    def do_POST(self):
        self.rfile.read(int(self.headers.get("content-length", 0)))
        body = b'{"message":"Accepted"}'
        self.send_response(202)
        self.send_header("content-type", "application/json")
        self.send_header("content-length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()


def make_client(base_url):
    with LangfuseResourceManager._lock:
        LangfuseResourceManager._instances.clear()
    return Langfuse(
        public_key="pk-g", secret_key="sk-g", base_url=base_url, flush_interval=5
    )


def timed(label, fn):
    started = time.monotonic()
    try:
        fn()
        print(f"{label}: returned in {time.monotonic() - started:.4f}s")
    except Exception as e:
        print(f"{label}: raised {type(e).__name__}: {e}")


# Unreachable host
client = make_client("http://127.0.0.1:9")
client.create_score(name="a", value=1.0, trace_id="a" * 32)
timed("unreachable flush", client.flush)
timed("unreachable shutdown", client.shutdown)
print("consumer alive:", client._resources._score_consumers[0].is_alive())

# Reachable server
client = make_client(f"http://127.0.0.1:{server.server_address[1]}")
client.create_score(name="a", value=1.0, trace_id="a" * 32)
timed("flush with one score", client.flush)
timed("flush with nothing queued", client.flush)
timed("shutdown", client.shutdown)

# Consumer stops while flush() waits, leaving an item unprocessed
client = make_client(f"http://127.0.0.1:{server.server_address[1]}")
consumer = client._resources._score_consumers[0]
wake = consumer.wake


def stop_later():
    consumer.running = False
    wake()
    client._resources._score_queue.put("orphan")


consumer.wake = stop_later
timed("flush with consumer stopping mid-wait", client.flush)

# flush() on gevent's hub threadpool, with a score queued. The queues are then
# tied to the threadpool's hub, so skip the atexit shutdown.
import os  # noqa: E402
import sys  # noqa: E402

import gevent  # noqa: E402

client = make_client(f"http://127.0.0.1:{server.server_address[1]}")
client.create_score(name="a", value=1.0, trace_id="a" * 32)
timed(
    "flush on the hub threadpool",
    lambda: gevent.get_hub().threadpool.spawn(client.flush).get(timeout=10),
)
sys.stdout.flush()
os._exit(0)

Not run: tests/e2e and tests/live_provider, which need a server and provider keys. The request bodies are unchanged.

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. (n/a, no public behavior change)
• 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: 4/5

The PR appears safe to merge, with a non-blocking gap in the test for a consumer stopping during the timed wait.

Summary

Adds a timed join() fallback for score queues without all_tasks_done, allowing flush() and shutdown() to work with gevent.

• Score flushing waits with the queue API available in each process.

_{Reviews (1) · Last reviewed commit: "fix(scores): make flush() and 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.

[hassiebp]

Closing this in favor of #2034 - thanks David!