[hassiebp]

What does this PR do?

Since 5.0.3, flush() and shutdown() crash under gevent monkey-patching:

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'

#2025 added _join_score_queue, which reads the standard library queue's internals (all_tasks_done, unfinished_tasks). It needed them for one reason: flush() puts a wake signal on the queue, and if the consumer stops before taking it, that signal stays counted and a plain Queue.join() blocks forever.

This PR stops counting the wake signal instead:

• wake() puts the signal and immediately calls task_done() for it, but only if the put() succeeded. When the queue is full nothing is put, and nothing is acknowledged.
• The consumer no longer calls task_done() when it takes a wake signal.
• flush() is back to 5.0.2's plain self._score_queue.join(), preceded by the wake. _join_score_queue is deleted.

An orphaned wake signal is therefore never an unfinished task and cannot block join(). Only the public queue API is used (put, get, task_done, join), so this works with the standard library, gevent and eventlet queues.

Credit to @DavidTraina: #2032 found and diagnosed this bug, and its gevent check script and scenarios were used to verify this PR. The pre-acknowledged signal follows the pattern in his #2033 for media shutdown signals.

Behavior notes:

• The consumer stops with real scores still queued: flush() blocks, exactly as in 5.0.2. fix(scores): send pending scores immediately on flush() #2025 had made it return in that case; that is dropped here to avoid queue internals.
• flush() from gevent's native hub threadpool: raises gevent's LoopExit, exactly as in 5.0.2 (verified against the 5.0.2 code).
• Score created after shutdown(): dropped with a warning instead of enqueued. No consumer is left to send it, so on 5.0.2 it was never sent either, and a later flush() would block on it.
• Full queue: when wake() finds the queue full it skips the signal, so that flush() may wait up to flush_interval (default 1s), as in 5.0.3.

Not in this PR: the batch-fragmentation follow-up from #2025.

Type of change

• Bug fix

Verification

uv run --frozen pytest -n auto --dist worksteal tests/unit   # 958 passed, 2 skipped
uv run --frozen pytest tests/unit/test_score_consumer.py     # 38 passed (20 of 20 runs)
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              # clean

Tests added in tests/unit/test_score_consumer.py:

• test_flush_and_shutdown_work_under_gevent_monkey_patching runs a real gevent monkey.patch_all() subprocess: flush(), then shutdown(), then flush(). It fails on 5.0.3 with the AttributeError above.

• test_wake_on_full_queue_does_not_acknowledge_a_score covers wake() on a full queue, where it must not call task_done(). It fails if the task_done() also runs on Full.

• test_consumer_never_acknowledges_wake_signals puts a wake signal after each real score and drives every consumer path:

  • a normal batch;
  • a score carried over by the byte limit;
  • a score dropped for size;
  • a score dropped by a serialization error;
  • the pause()/shutdown drain.

  It counts every task_done(): exactly one per real score plus one per successful wake. It fails with ValueError: task_done() called too many times if the consumer acknowledges a signal.

• test_concurrent_flush_returns_only_after_own_score_is_sent[roomy|often-full] has 8 threads, each calling create_score() then flush() 25 times (5 times with a 1-slot queue). After each flush() it checks that the thread's own score has been POSTed, if it was enqueued. Any exception in the consumer thread fails the test. The 1-slot variant fails when wake() acknowledges on Full.

• test_score_after_shutdown_is_dropped_and_flush_returns: shutdown(), create_score(), then flush() returns promptly, nothing is put on the queue and nothing is sent. Without the guard flush() hangs.

• test_flush_racing_shutdown_does_not_hang: 4 flush() threads and a shutdown() thread on one client.

• test_flush_does_not_wait_for_dropped_scores and test_exit_without_shutdown_sends_pending_scores. The second covers the atexit flush, which the earlier review threads asked for.

• The existing test_flush_after_shutdown_returns and test_flush_returns_when_consumer_stops_after_liveness_check still pass. The latter now also checks that its orphaned signal stays queued without being an unfinished task.

Manual checks:

• fix(scores): make flush() and shutdown() work under gevent and eventlet #2032's gevent script on gevent 25.9.1 and 26.9.0:
  • unreachable host: flush() and shutdown() both return;
  • one score: about 3ms;
  • idle flush(): returns immediately;
  • shutdown: returns;
  • consumer stops mid-wait with a real score left queued: blocks, as in 5.0.2 (see behavior notes);
  • hub threadpool: LoopExit, the same as 5.0.2.
• eventlet monkey_patch(): flush() sends the score in about 13ms.

flush() latency on the standard library queue. Medians of 3 runs, mock transport, Python 3.12 on Linux:

	main (5.0.3)	#2032	this PR
empty flush()	17.8–18.1 µs	17.6–17.8 µs	3.1 µs
flush() with one score queued	478–492 µs	472–473 µs	470–516 µs

Checklist

• I self-reviewed the diff using code_review.md.
• I added or updated tests for behavior changes.
• I did not hand-edit generated files.
• I did not commit secrets or credentials.

  

Confidence Score: 4/5

The PR still needs the automatic-shutdown fix before merging.

Summary

Replaces reads of private queue fields with Queue.join() in a daemon helper thread.

• Score flushing works with gevent’s monkey-patched queues.

_{Reviews (2) · Last reviewed commit: "fix(scores): use only the public Queue A..." · Reviewed by Greptile}

[github-actions]

@claude review

[claude]

Claude Code Review

No review was started: this request came from a bot account. Manual reviews can only be requested by someone with write access to this repository. Ask a maintainer to comment @claude review, or have your automation post the comment from a user account with write access.

_{Tip: disable this comment in your organization's Code Review settings.}

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-10-21T07:35:40.413272Z	c47bea0	New commits

ℹ️ 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" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[claude]

Code review found no new issues

No new issues were found in this update; 1 finding from earlier reviews is still open above.

Still open from earlier reviews (1):

• 🔴 langfuse/_client/resource_manager.py:625 — On Python builds that refuse new threads during interpreter shutdown, the automatic exit flush can now raise and drop q…

If you have decided not to act on one of these findings, resolve its thread (a reply alone leaves it open) and the next review stops counting it. To review this commit again now, use Re-run on its "Claude Code Review" check.

[hassiebp]

@claude review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: c47bea06f8

ℹ️ 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".