diff --git a/litellm/integrations/langfuse/langfuse_sdk.py b/litellm/integrations/langfuse/langfuse_sdk.py index 34d143cc114..e443b508dd8 100644 --- a/litellm/integrations/langfuse/langfuse_sdk.py +++ b/litellm/integrations/langfuse/langfuse_sdk.py @@ -191,7 +191,8 @@ def build_isolated_tracer_provider(*, environment: str | None, release: str | No resource=Resource.create(dict(attributes)), sampler=TraceIdRatioBased(sample_rate) if sample_rate < 1 else None, ) - _litellm_built_providers.add(provider) + with _LIVE_CLIENTS_LOCK: + _litellm_built_providers.add(provider) return provider @@ -242,23 +243,25 @@ def _retire_orphaned_providers() -> None: prompt-management LRU drops clients, never reaches ``shutdown_langfuse_client``, and the provider's own atexit hook would keep its export thread alive for the rest of the process. - Providers are snapshotted before the registry: a provider is only ever built and its - bundle registered inside one registry-locked block, so any provider seen in the first - snapshot is fully registered by the time the second one is taken. Runs outside both - locks because provider shutdown flushes and joins the export thread. + Holders are snapshotted last: a client is registered in the same registry-locked block + that builds its provider, so once the registry snapshot's lock has been acquired, the + client of any provider from the first snapshot is visible to the final one even when a + concurrent rotation already evicted its bundle again. Runs outside both locks because + provider shutdown flushes and joins the export thread. """ with _LIVE_CLIENTS_LOCK: candidates: Final = tuple(_litellm_built_providers) - held: Final = tuple( - getattr(resources, "tracer_provider", None) - for resources, holders in _live_clients.items() - if len(holders) > 0 - ) with LangfuseResourceManager._lock: # pyright: ignore[reportPrivateUsage] # registry has no public accessor registered: Final = tuple( getattr(resources, "tracer_provider", None) for resources in LangfuseResourceManager._instances.values() # pyright: ignore[reportPrivateUsage] # registry has no public accessor ) + with _LIVE_CLIENTS_LOCK: + held: Final = tuple( + getattr(resources, "tracer_provider", None) + for resources, holders in _live_clients.items() + if len(holders) > 0 + ) orphaned: Final = tuple(provider for provider in candidates if provider not in registered and provider not in held) for provider in orphaned: _litellm_built_providers.discard(provider) diff --git a/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py b/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py index 73e80d09c10..0693fb6ad9f 100644 --- a/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py +++ b/tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py @@ -649,6 +649,40 @@ def test_the_registrys_current_bundle_is_not_reaped_when_its_clients_die(): assert _exports(successor, exporter, "after-collection") +def test_a_sweep_overlapping_registration_and_rotation_keeps_the_live_provider(): + """A sweep can snapshot providers before a client registers, then wait on the registry + lock while that client registers and a rotation evicts its fresh bundle. Holders are + re-read after the registry snapshot, so the stale first look must not win.""" + import threading + + from litellm.integrations.langfuse.langfuse_sdk import _retire_orphaned_providers + + exporter = InMemorySpanExporter() + provider = build_isolated_tracer_provider(environment=None, release=None) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + client = Langfuse( + public_key=PUBLIC_KEY, + secret_key="sk-original", + host="http://127.0.0.1:1", + tracer_provider=provider, + span_exporter=exporter, + ) + registry_lock = LangfuseResourceManager._lock + registry_lock.acquire() + try: + sweeper = threading.Thread(target=_retire_orphaned_providers) + sweeper.start() + sweeper.join(timeout=0.5) # parks on the registry lock once its provider snapshot is taken + register_langfuse_client(client) + LangfuseResourceManager._instances.pop(PUBLIC_KEY, None) # the rotation that evicts the fresh bundle + finally: + registry_lock.release() + sweeper.join(timeout=5) + assert not sweeper.is_alive() + + assert _exports(client, exporter, "after-racing-sweep") + + def test_ssl_exporter_is_only_built_with_custom_tls_material(monkeypatch, tmp_path): """v4 exports over its own OTLP channel, so litellm's CA bundle must be rebuilt onto it.""" import litellm