fix(resource_manager): reinitialize consumer threads after os.fork()

[pyg410]

What does this PR do?

When using Gunicorn with --preload, os.fork() copies memory but not threads
(POSIX.1: https://pubs.opengroup.org/onlinepubs/9699919799/functions/fork.html).
LangfuseResourceManager starts media upload and score ingestion consumer threads
on initialization, but these threads are not inherited by forked worker processes.
As a result, all media upload and score ingestion events are silently lost, and
calling flush() blocks forever on queue.join() → Gunicorn worker timeout (SIGABRT).

Note: span export is already fork-safe via LangfuseSpanProcessor (BatchSpanProcessor
inheritance). This fix covers the remaining background threads managed by
LangfuseResourceManager.

Related PR

• fix(task_manager): reinitialize consumer threads after os.fork() #1654 (initial fork-safety fix in v2-stable)

Changes:

• Extract consumer thread initialization into _init_consumer_threads() for reuse
• Store score_ingestion_client as an instance variable for access in _at_fork_reinit()
• Add _at_fork_reinit() to reinitialize queues and consumer threads after fork
• Register os.register_at_fork(after_in_child=...) using weakref.WeakMethod to
  avoid permanent strong references that would prevent garbage collection
• Add _shutdown flag to skip reinitialization on already-stopped instances

Type of change

• Bug fix

Verification

List the main commands you ran:

 uv run pytest tests/unit/test_resource_manager.py -v

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.

Greptile Summary

This PR fixes silent data loss and flush() deadlocks in Gunicorn --preload deployments by registering an os.register_at_fork(after_in_child=...) handler that reinitializes the media upload and score ingestion consumer threads in each forked worker.

• Consumer thread startup is extracted into _init_consumer_threads() so it can be called both during initial setup and after fork; queues are intentionally recreated in the child to prevent duplicating pre-fork events across workers.
• A WeakMethod-based fork handler prevents the os.register_at_fork registry from holding a strong reference that would block garbage collection, and a _shutdown flag guards against reinitializing already-torn-down instances.
• The fix deliberately omits recreating httpx.Client / LangfuseClient instances after fork (acknowledged in comments), leaving inherited connection-pool FDs that could cause initial connection errors in workers; the after_in_child callback also has no exception handling, so a thread-creation failure propagates out of os.fork() and crashes the worker rather than degrading gracefully.

Confidence Score: 3/5

The core fix is sound and addresses a real production issue, but the after_in_child callback has no exception handling, meaning a thread-creation failure would crash the Gunicorn worker rather than falling back gracefully.

The fork-reinitialization logic is correct for the happy path: queues are replaced, old dead thread references are discarded, new consumer threads are started, and the WeakMethod approach is appropriate. The main gap is that _init_consumer_threads() is called directly inside the after_in_child callback without a try/except — if the OS refuses to create a thread at worker startup time, the exception surfaces from os.fork() and kills the worker instead of allowing it to continue with degraded telemetry. The inherited httpx connection pool FDs are an acknowledged limitation that could surface as transient connection errors in workers.

langfuse/_client/resource_manager.py warrants a second look around the _at_fork_reinit method — specifically exception handling and whether the HTTP clients need to be recreated for full fork safety.

Sequence Diagram

sequenceDiagram
    participant Master as Master Process (Gunicorn)
    participant Fork as os.fork()
    participant Parent as Parent Process
    participant Child as Child Worker Process

    Master->>Master: Langfuse() → LangfuseResourceManager.__new__()
    Master->>Master: _initialize_instance()
    Master->>Master: _init_consumer_threads() → start media + score threads
    Master->>Master: "os.register_at_fork(after_in_child=weak_reinit_lambda)"
    Master->>Fork: os.fork()
    Fork-->>Parent: returns child PID (old threads alive)
    Fork-->>Child: returns 0 (threads NOT inherited)
    Child->>Child: after_in_child: weak_reinit() → _at_fork_reinit()
    Child->>Child: _shutdown? → False → proceed
    Child->>Child: _init_consumer_threads()
    Child->>Child: new _media_upload_queue (fresh Queue)
    Child->>Child: new _score_ingestion_queue (fresh Queue)
    Child->>Child: start fresh MediaUploadConsumer threads
    Child->>Child: start fresh ScoreIngestionConsumer thread
    Child->>Child: child ready to handle requests
    Parent->>Parent: continues with original threads unchanged

Loading

Prompt To Fix All With AI

Fix the following 3 code review issues. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 3
langfuse/_client/resource_manager.py:344-378
**Unhandled exception in `after_in_child` callback crashes Gunicorn worker**

`_at_fork_reinit` calls `_init_consumer_threads`, which calls `Thread.start()`. If the OS refuses to create a thread — e.g., due to resource exhaustion (`OSError: can't start new thread`) — the exception propagates through the `after_in_child` callback chain and surfaces as an exception from `os.fork()` in the child process. Gunicorn would see the worker fail immediately at startup rather than the telemetry path degrading gracefully. Wrapping `_init_consumer_threads()` in a `try/except Exception` and logging the error would allow the child to continue (without consumer threads) instead of crashing.

### Issue 2 of 3
langfuse/_client/resource_manager.py:366-374
**Inherited `httpx.Client` connection pool is not fork-safe**

`self.httpx_client` and `self._score_ingestion_client` (which holds a reference to `self.httpx_client` as its `session`) are not recreated in `_at_fork_reinit`. After `os.fork()`, the child inherits duplicated file descriptors from the parent's connection pool. The parent and child share open TCP connections at the OS level; if the parent closes or recycles a connection, the child's pool may try to reuse a dead FD, causing connection errors on the first outbound requests from each worker. The existing comment acknowledges this but marks it as a future concern — since this PR is specifically targeting fork-safety, recreating the clients in the child would complete the fix.

### Issue 3 of 3
langfuse/_client/resource_manager.py:80-82
**Class-level `_lock` (RLock) not guarded around fork**

`LangfuseResourceManager._lock` is a `threading.RLock`. If `os.fork()` occurs while another thread holds `_lock` (i.e., during a concurrent `__new__` call), the child inherits the locked mutex with no thread to release it. Any subsequent `LangfuseResourceManager.__new__` call in the child — including indirect calls — would deadlock. The standard remedy is to register a `before` handler (`os.register_at_fork(before=cls._lock.acquire)`) and release it in both `after_in_parent` and `after_in_child`. In a typical Gunicorn `--preload` deployment this lock is already released before workers fork, so this is a latent risk rather than an immediate failure for the primary use case.

_{Reviews (1): Last reviewed commit: "fix(resource_manager): reinitialize cons..." | Re-trigger Greptile}

Greptile also left 2 inline comments on this PR.

[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.

[greptile-apps]

Unhandled exception in after_in_child callback crashes Gunicorn worker

_at_fork_reinit calls _init_consumer_threads, which calls Thread.start(). If the OS refuses to create a thread — e.g., due to resource exhaustion (OSError: can't start new thread) — the exception propagates through the after_in_child callback chain and surfaces as an exception from os.fork() in the child process. Gunicorn would see the worker fail immediately at startup rather than the telemetry path degrading gracefully. Wrapping _init_consumer_threads() in a try/except Exception and logging the error would allow the child to continue (without consumer threads) instead of crashing.

Prompt To Fix With AI

This is a comment left during a code review.
Path: langfuse/_client/resource_manager.py
Line: 344-378

Comment:
**Unhandled exception in `after_in_child` callback crashes Gunicorn worker**

`_at_fork_reinit` calls `_init_consumer_threads`, which calls `Thread.start()`. If the OS refuses to create a thread — e.g., due to resource exhaustion (`OSError: can't start new thread`) — the exception propagates through the `after_in_child` callback chain and surfaces as an exception from `os.fork()` in the child process. Gunicorn would see the worker fail immediately at startup rather than the telemetry path degrading gracefully. Wrapping `_init_consumer_threads()` in a `try/except Exception` and logging the error would allow the child to continue (without consumer threads) instead of crashing.

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

Inherited httpx.Client connection pool is not fork-safe

self.httpx_client and self._score_ingestion_client (which holds a reference to self.httpx_client as its session) are not recreated in _at_fork_reinit. After os.fork(), the child inherits duplicated file descriptors from the parent's connection pool. The parent and child share open TCP connections at the OS level; if the parent closes or recycles a connection, the child's pool may try to reuse a dead FD, causing connection errors on the first outbound requests from each worker. The existing comment acknowledges this but marks it as a future concern — since this PR is specifically targeting fork-safety, recreating the clients in the child would complete the fix.

Prompt To Fix With AI

This is a comment left during a code review.
Path: langfuse/_client/resource_manager.py
Line: 366-374

Comment:
**Inherited `httpx.Client` connection pool is not fork-safe**

`self.httpx_client` and `self._score_ingestion_client` (which holds a reference to `self.httpx_client` as its `session`) are not recreated in `_at_fork_reinit`. After `os.fork()`, the child inherits duplicated file descriptors from the parent's connection pool. The parent and child share open TCP connections at the OS level; if the parent closes or recycles a connection, the child's pool may try to reuse a dead FD, causing connection errors on the first outbound requests from each worker. The existing comment acknowledges this but marks it as a future concern — since this PR is specifically targeting fork-safety, recreating the clients in the child would complete the fix.

How can I resolve this? If you propose a fix, please make it concise.