[anishgurjar]

What does this PR do?

PR title must follow Conventional Commits, for example feat: add dataset scoring helper or fix(openai): preserve trace context.

Relates to langfuse/langfuse#11104

Guards flush() against blocking forever on the media upload queue when no consumer thread is left to drain it.

LangfuseResourceManager.flush() ends with self._media_upload_queue.join(). Queue.join() blocks until every queued item is marked done, but media upload consumers exit on pause() during shutdown() without draining the queue. Any interleaving that leaves items pending with no live consumer - concurrent shutdown()/flush() calls, a media-bearing span ending between the join and the consumer stop, or a later shutdown() after such an interleaving - makes the next flush() block forever.

This mirrors the score-queue guard added in #2025 for the same class, one call site up: wait on the queue's all_tasks_done condition and return early when no consumer thread is alive to drain it. It is a minimal alternative to the stale #1473, touching only resource_manager.py.

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 pytest tests/unit/test_resource_manager.py -q
uv run pytest tests/unit -q
uv run ruff check langfuse/_client/resource_manager.py tests/unit/test_resource_manager.py
uv run ruff format --check langfuse/_client/resource_manager.py tests/unit/test_resource_manager.py
uv run mypy langfuse/_client/resource_manager.py

The new regression test (consumers stopped with an item pending, flush() on a watchdog thread) fails on main - flush() never returns - and passes with the guard.

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.
• 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 test coverage gap for consumers stopping during flush().

Summary

Replaces the media queue's unconditional wait with _join_media_upload_queue().

• Media queue flushes stop waiting when no upload worker remains.

No runtime tests were run during this review.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart TD
  A["flush()"] --> B{"Unfinished media work?"}
  B -->|No| E["Return"]
  B -->|Yes| C{"Any consumer alive?"}
  C -->|No| E
  C -->|Yes| D["Wait for completion or 0.1 seconds"]
  D --> B

Loading

_{Reviews (1) · Last reviewed commit: "test: cover flush() with stopped media c..." · 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.

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[anishgurjar]

cc @hassiebp

[greptile-apps]

Shutdown race lacks a test

The new test stops every media consumer before calling flush(), so it never reaches the timed wait in _join_media_upload_queue(). It would still pass if that wait became untimed, bringing back the hang when the last consumer exits without another queue notification.

Add a test that keeps a consumer alive until the wait starts, then stops it with an unfinished item. This is a non-blocking coverage improvement for the concurrent-shutdown case.

Prompt To Fix With AI

This is a comment left during a code review.
Path: tests/unit/test_resource_manager.py
Line: 499-503

Comment:
**Shutdown race lacks a test**

The new test stops every media consumer before calling `flush()`, so it never reaches the timed wait in `_join_media_upload_queue()`. It would still pass if that wait became untimed, bringing back the hang when the last consumer exits without another queue notification.

Add a test that keeps a consumer alive until the wait starts, then stops it with an unfinished item. This is a non-blocking coverage improvement for the concurrent-shutdown case.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!