From f49b4d62977d367c52bd9229de59436ff05af577 Mon Sep 17 00:00:00 2001 From: Hemanth Chittanuru Date: Thu, 13 Aug 2026 15:21:56 -0400 Subject: [PATCH 1/3] fix(client): resolve project id at most once and never raise from get_trace_url `_get_project_id` cached only a successful lookup. When the lookup failed -- an auth error, or no project for the configured keys -- `self._project_id` stayed `None`, so the next call issued the blocking request again. Callers that generate one URL per row (rendering trace links in a list view) turned that into one blocking HTTP request per row, per render. The lookup was also unguarded, so a transport or auth error propagated out of `get_trace_url` into the caller. Every other public accessor on the client returns `None` rather than raising, and URL generation is a convenience that should not be able to take down the code path that logs a link. A project id does not change for a given key pair, so the outcome is now resolved at most once per client instance, negative outcomes included, and failures are logged and swallowed. The tradeoff is that a transient failure disables URL generation for the life of the client; the warning says so. Co-Authored-By: Claude Opus 5 --- langfuse/_client/client.py | 39 +++++++++++++++--- tests/unit/test_trace_url.py | 79 ++++++++++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+), 5 deletions(-) create mode 100644 tests/unit/test_trace_url.py diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 2b2889545..0a0ad138b 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -351,6 +351,7 @@ def __init__( or get_common_release_envs() ) self._project_id: Optional[str] = None + self._project_id_resolved: bool = False if sample_rate is None: sample_rate = float(os.environ.get(LANGFUSE_SAMPLE_RATE, 1.0)) if not 0.0 <= sample_rate <= 1.0: @@ -2408,13 +2409,41 @@ def get_current_observation_id(self) -> Optional[str]: return self._get_otel_span_id(current_otel_span) if current_otel_span else None def _get_project_id(self) -> Optional[str]: - """Fetch and return the current project id. Persisted across requests. Returns None if no project id is found for api keys.""" - if not self._project_id: + """Fetch and return the current project id. Persisted across requests. Returns None if no project id is found for api keys. + + The lookup is a blocking network request, so its outcome is resolved at + most once per client instance -- including a negative outcome. A project + id never changes for a given key pair, so caching only the success meant + that a failing lookup was retried on every call: `get_trace_url` is + commonly called once per row when rendering links, which turned an auth + failure or an empty project list into one blocking request per row. + + Failures are logged and swallowed rather than raised. This is only ever + reached from URL generation, which is a convenience: it returns None when + the project id is unavailable, and must not surface an exception into the + caller's code path. + """ + if self._project_id_resolved: + return self._project_id + + self._project_id_resolved = True + + try: proj = self.api.projects.get() - if not proj.data or not proj.data[0].id: - return None + except Exception as e: + langfuse_logger.warning( + f"Could not resolve project id: {e}. URL generation is disabled for this client instance." + ) + return None + + if not proj.data or not proj.data[0].id: + langfuse_logger.warning( + "Could not resolve project id: no project found for the configured API keys. " + "URL generation is disabled for this client instance." + ) + return None - self._project_id = proj.data[0].id + self._project_id = proj.data[0].id return self._project_id diff --git a/tests/unit/test_trace_url.py b/tests/unit/test_trace_url.py new file mode 100644 index 000000000..0e4f8bf5e --- /dev/null +++ b/tests/unit/test_trace_url.py @@ -0,0 +1,79 @@ +"""Tests for project id resolution behind `get_trace_url`. + +The project id lookup is a blocking network request. `get_trace_url` is commonly +called once per row when rendering links, so the lookup must be attempted at most +once per client instance -- on failure as well as on success -- and it must never +raise into the caller. +""" + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from langfuse import Langfuse +from langfuse._client.resource_manager import LangfuseResourceManager + +TRACE_ID = "1234567890abcdef1234567890abcdef" + + +@pytest.fixture +def client(monkeypatch): + """A client with fake credentials; `api` is replaced per test.""" + with LangfuseResourceManager._lock: + LangfuseResourceManager._instances.clear() + + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-trace-url") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-trace-url") + monkeypatch.setenv("LANGFUSE_BASE_URL", "http://localhost:3000") + + langfuse = Langfuse() + + yield langfuse + + with LangfuseResourceManager._lock: + LangfuseResourceManager._instances.clear() + + +def _api_returning(project_id): + api = Mock() + api.projects.get.return_value = SimpleNamespace( + data=[SimpleNamespace(id=project_id)] if project_id else [] + ) + return api + + +def test_project_id_is_fetched_once_on_success(client): + client.api = _api_returning("project-id") + + first = client.get_trace_url(trace_id=TRACE_ID) + second = client.get_trace_url(trace_id=TRACE_ID) + + assert first == f"http://localhost:3000/project/project-id/traces/{TRACE_ID}" + assert second == first + assert client.api.projects.get.call_count == 1 + + +def test_project_id_is_not_refetched_when_no_project_is_found(client): + client.api = _api_returning(None) + + assert client.get_trace_url(trace_id=TRACE_ID) is None + assert client.get_trace_url(trace_id=TRACE_ID) is None + assert client.api.projects.get.call_count == 1 + + +def test_project_id_lookup_failure_does_not_raise_and_is_not_retried(client): + api = Mock() + api.projects.get.side_effect = RuntimeError("401 Unauthorized") + client.api = api + + assert client.get_trace_url(trace_id=TRACE_ID) is None + assert client.get_trace_url(trace_id=TRACE_ID) is None + assert api.projects.get.call_count == 1 + + +def test_no_project_id_lookup_without_a_trace_id(client): + client.api = _api_returning("project-id") + + assert client.get_trace_url() is None + assert client.api.projects.get.call_count == 0 From 6d24ddcddd6a7c7f5c9e9eb14146a6a52fb09efe Mon Sep 17 00:00:00 2001 From: Hemanth Chittanuru Date: Thu, 13 Aug 2026 15:29:07 -0400 Subject: [PATCH 2/3] fix(client): guard project id resolution with a lock Addresses review feedback on the previous commit: setting the resolved flag before the request completed let a concurrent caller observe the pre-resolution state -- it saw the flag, read a still-`None` project id, and returned no URL while a lookup that would have produced one was in flight. The resolution now happens under a lock, and the flag is set only once the outcome is known. A concurrent caller either sees a fully resolved state or waits for the in-flight lookup. The fetch moves to `_fetch_project_id` so the locked section stays one assignment. `test_concurrent_callers_all_see_the_resolved_project_id` fails against the previous commit and passes here. Co-Authored-By: Claude Opus 5 --- langfuse/_client/client.py | 21 ++++++++++++++--- tests/unit/test_trace_url.py | 45 ++++++++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 0a0ad138b..34995ef10 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -7,6 +7,7 @@ import logging import os import re +import threading import urllib.parse import uuid import warnings @@ -352,6 +353,7 @@ def __init__( ) self._project_id: Optional[str] = None self._project_id_resolved: bool = False + self._project_id_lock = threading.Lock() if sample_rate is None: sample_rate = float(os.environ.get(LANGFUSE_SAMPLE_RATE, 1.0)) if not 0.0 <= sample_rate <= 1.0: @@ -2422,12 +2424,25 @@ def _get_project_id(self) -> Optional[str]: reached from URL generation, which is a convenience: it returns None when the project id is unavailable, and must not surface an exception into the caller's code path. + + The resolution is guarded by a lock, and the flag is set only once the + outcome is known: concurrent callers either see a fully resolved state or + wait for the in-flight lookup, rather than reading a `None` project id + that a request in flight is about to fill in. """ if self._project_id_resolved: return self._project_id - self._project_id_resolved = True + with self._project_id_lock: + # Another thread may have resolved it while this one waited. + if not self._project_id_resolved: + self._project_id = self._fetch_project_id() + self._project_id_resolved = True + + return self._project_id + def _fetch_project_id(self) -> Optional[str]: + """Fetch the project id for the configured keys. Returns None if unavailable.""" try: proj = self.api.projects.get() except Exception as e: @@ -2443,9 +2458,9 @@ def _get_project_id(self) -> Optional[str]: ) return None - self._project_id = proj.data[0].id + project_id: Optional[str] = proj.data[0].id - return self._project_id + return project_id def get_trace_url(self, *, trace_id: Optional[str] = None) -> Optional[str]: """Get the URL to view a trace in the Langfuse UI. diff --git a/tests/unit/test_trace_url.py b/tests/unit/test_trace_url.py index 0e4f8bf5e..99d8e47e9 100644 --- a/tests/unit/test_trace_url.py +++ b/tests/unit/test_trace_url.py @@ -6,7 +6,9 @@ raise into the caller. """ +import threading from types import SimpleNamespace +from typing import List, Optional from unittest.mock import Mock import pytest @@ -72,6 +74,49 @@ def test_project_id_lookup_failure_does_not_raise_and_is_not_retried(client): assert api.projects.get.call_count == 1 +def test_concurrent_callers_all_see_the_resolved_project_id(client): + """A caller must not observe the pre-resolution state of an in-flight lookup. + + The lookup is slow enough here that every other thread arrives while the + first one is still waiting on the network. Each must block and then see the + resolved id -- not a `None` that a request in flight is about to fill in. + """ + started = threading.Event() + release = threading.Event() + + def slow_get(): + started.set() + release.wait(timeout=5) + return SimpleNamespace(data=[SimpleNamespace(id="project-id")]) + + api = Mock() + api.projects.get.side_effect = slow_get + client.api = api + + results: List[Optional[str]] = [] + results_lock = threading.Lock() + + def call(): + url = client.get_trace_url(trace_id=TRACE_ID) + with results_lock: + results.append(url) + + threads = [threading.Thread(target=call) for _ in range(8)] + for thread in threads: + thread.start() + + assert started.wait(timeout=5), "the first lookup never started" + release.set() + + for thread in threads: + thread.join(timeout=5) + assert not thread.is_alive() + + expected = f"http://localhost:3000/project/project-id/traces/{TRACE_ID}" + assert results == [expected] * 8 + assert api.projects.get.call_count == 1 + + def test_no_project_id_lookup_without_a_trace_id(client): client.api = _api_returning("project-id") From b629c44b7754aaa224ff4186558d1fe8158aae14 Mon Sep 17 00:00:00 2001 From: Hemanth Chittanuru Date: Thu, 13 Aug 2026 15:36:50 -0400 Subject: [PATCH 3/3] fix(client): move project id resolution to the resource manager The cache was on the wrong object. `get_client()` builds a new `Langfuse` on every call, so state memoized there does not outlive the call that created it: `get_client().get_trace_url(trace_id=...)` per row still issued one blocking `projects.get()` per row, which is the case the earlier commits set out to fix. `LangfuseResourceManager` is the singleton, keyed by public key -- the exact granularity a project id has -- and its class docstring already names it the owner of "retrieving and caching project information for URL generation". The resolution, its lock and the negative outcome move there; `Langfuse` delegates. Two things fall out of the move: - `_at_fork_reinit` now replaces the project id lock, next to the class lock it already replaces for the same reason. A parent thread holding it across the lookup at fork time would otherwise deadlock the first caller in the child. - `Langfuse._get_project_id` checks `_resources` instead of letting the blanket `except` swallow `AttributeError("Langfuse client is not initialized")`. A client built without keys returns None quietly rather than logging a warning that blames project id resolution for a missing client. Co-Authored-By: Claude Opus 5 --- langfuse/_client/client.py | 62 +++++-------------------- langfuse/_client/resource_manager.py | 64 ++++++++++++++++++++++++++ tests/unit/test_trace_url.py | 67 +++++++++++++++++++++++++++- 3 files changed, 141 insertions(+), 52 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 34995ef10..50d0d8e6d 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -7,7 +7,6 @@ import logging import os import re -import threading import urllib.parse import uuid import warnings @@ -351,9 +350,6 @@ def __init__( or os.environ.get(LANGFUSE_RELEASE, None) or get_common_release_envs() ) - self._project_id: Optional[str] = None - self._project_id_resolved: bool = False - self._project_id_lock = threading.Lock() if sample_rate is None: sample_rate = float(os.environ.get(LANGFUSE_SAMPLE_RATE, 1.0)) if not 0.0 <= sample_rate <= 1.0: @@ -2411,56 +2407,22 @@ def get_current_observation_id(self) -> Optional[str]: return self._get_otel_span_id(current_otel_span) if current_otel_span else None def _get_project_id(self) -> Optional[str]: - """Fetch and return the current project id. Persisted across requests. Returns None if no project id is found for api keys. - - The lookup is a blocking network request, so its outcome is resolved at - most once per client instance -- including a negative outcome. A project - id never changes for a given key pair, so caching only the success meant - that a failing lookup was retried on every call: `get_trace_url` is - commonly called once per row when rendering links, which turned an auth - failure or an empty project list into one blocking request per row. - - Failures are logged and swallowed rather than raised. This is only ever - reached from URL generation, which is a convenience: it returns None when - the project id is unavailable, and must not surface an exception into the - caller's code path. - - The resolution is guarded by a lock, and the flag is set only once the - outcome is known: concurrent callers either see a fully resolved state or - wait for the in-flight lookup, rather than reading a `None` project id - that a request in flight is about to fill in. + """Fetch and return the current project id. Returns None if no project id is found for api keys. + + Delegates to the resource manager, which caches the outcome. The cache + cannot live here: `get_client()` constructs a new `Langfuse` on every + call, so anything memoized on this object dies with it, and a caller + generating one URL per row would issue one blocking lookup per row. The + resource manager is a singleton keyed by public key, which is the + granularity a project id has. """ - if self._project_id_resolved: - return self._project_id - - with self._project_id_lock: - # Another thread may have resolved it while this one waited. - if not self._project_id_resolved: - self._project_id = self._fetch_project_id() - self._project_id_resolved = True - - return self._project_id - - def _fetch_project_id(self) -> Optional[str]: - """Fetch the project id for the configured keys. Returns None if unavailable.""" - try: - proj = self.api.projects.get() - except Exception as e: - langfuse_logger.warning( - f"Could not resolve project id: {e}. URL generation is disabled for this client instance." - ) - return None - - if not proj.data or not proj.data[0].id: - langfuse_logger.warning( - "Could not resolve project id: no project found for the configured API keys. " - "URL generation is disabled for this client instance." + if self._resources is None: + langfuse_logger.debug( + "Operation skipped: _get_project_id - Client is not initialized." ) return None - project_id: Optional[str] = proj.data[0].id - - return project_id + return self._resources.get_project_id() def get_trace_url(self, *, trace_id: Optional[str] = None) -> Optional[str]: """Get the URL to view a trace in the Langfuse UI. diff --git a/langfuse/_client/resource_manager.py b/langfuse/_client/resource_manager.py index 67c44920a..ee19df1d0 100644 --- a/langfuse/_client/resource_manager.py +++ b/langfuse/_client/resource_manager.py @@ -275,6 +275,12 @@ def _initialize_instance( # Prompt cache self.prompt_cache = PromptCache() + # Project id cache. Resolved lazily on first use, at most once per + # instance -- see get_project_id. + self._project_id: Optional[str] = None + self._project_id_resolved = False + self._project_id_lock = threading.Lock() + # Register shutdown handler atexit.register(self.shutdown) @@ -399,6 +405,57 @@ def _init_consumer_threads(self) -> None: ingestion_consumer.start() self._ingestion_consumers.append(ingestion_consumer) + def get_project_id(self) -> Optional[str]: + """Return the project id for this instance's credentials, or None if unavailable. + + The lookup is a blocking network request, so its outcome is resolved at + most once per instance -- including a negative outcome. A project id does + not change for a given key pair, so caching only the success meant a + failing lookup was repeated on every call: `Langfuse.get_trace_url` is + commonly called once per row when rendering links, which turned an auth + failure or an empty project list into one blocking request per row. + + This lives on the resource manager rather than on `Langfuse` because + `get_client()` constructs a new `Langfuse` on every call; state cached on + that object does not outlive the call that created it. The singleton here + is keyed by public key, which is the granularity a project id has. + + Failures are logged and swallowed rather than raised: the only caller is + URL generation, which is a convenience and must not surface an exception + into the caller's code path. + """ + if self._project_id_resolved: + return self._project_id + + with self._project_id_lock: + # Another thread may have resolved it while this one waited. + if not self._project_id_resolved: + self._project_id = self._fetch_project_id() + self._project_id_resolved = True + + return self._project_id + + def _fetch_project_id(self) -> Optional[str]: + """Fetch the project id for this instance's credentials over the API.""" + try: + projects = self.api.projects.get() + except Exception as e: + langfuse_logger.warning( + f"Could not resolve project id: {e}. URL generation is disabled until the client is recreated." + ) + return None + + if not projects.data or not projects.data[0].id: + langfuse_logger.warning( + "Could not resolve project id: no project found for the configured API keys. " + "URL generation is disabled until the client is recreated." + ) + return None + + project_id: Optional[str] = projects.data[0].id + + return project_id + def _at_fork_reinit(self) -> None: """Reinitialize consumer threads after fork in child process. @@ -421,6 +478,13 @@ def _at_fork_reinit(self) -> None: # even if this particular instance was already shut down. LangfuseResourceManager._lock = threading.RLock() + # Same hazard for the project id lock: a thread in the parent may have + # been holding it across the lookup at fork time, which would deadlock + # the first caller here. The cached value survives (a project id is not + # per-process); only the mutex is replaced, so a lookup interrupted by + # the fork is simply retried in this child. + self._project_id_lock = threading.Lock() + if self._shutdown: return diff --git a/tests/unit/test_trace_url.py b/tests/unit/test_trace_url.py index 99d8e47e9..9b27ef1a4 100644 --- a/tests/unit/test_trace_url.py +++ b/tests/unit/test_trace_url.py @@ -2,8 +2,9 @@ The project id lookup is a blocking network request. `get_trace_url` is commonly called once per row when rendering links, so the lookup must be attempted at most -once per client instance -- on failure as well as on success -- and it must never -raise into the caller. +once -- on failure as well as on success -- and it must never raise into the +caller. "At most once" is per resource manager rather than per `Langfuse`, since +`get_client()` returns a new `Langfuse` on every call. """ import threading @@ -14,6 +15,7 @@ import pytest from langfuse import Langfuse +from langfuse._client.get_client import get_client from langfuse._client.resource_manager import LangfuseResourceManager TRACE_ID = "1234567890abcdef1234567890abcdef" @@ -122,3 +124,64 @@ def test_no_project_id_lookup_without_a_trace_id(client): assert client.get_trace_url() is None assert client.api.projects.get.call_count == 0 + + +def test_project_id_is_shared_across_client_instances(client): + """`get_client()` returns a new `Langfuse` per call; the cache must outlive it. + + This is why the resolution lives on the resource manager. Were it on the + `Langfuse` object, rendering one link per row via `get_client()` would issue + one blocking lookup per row. + """ + api = _api_returning("project-id") + client.api = api + expected = f"http://localhost:3000/project/project-id/traces/{TRACE_ID}" + + assert client.get_trace_url(trace_id=TRACE_ID) == expected + assert api.projects.get.call_count == 1 + + for _ in range(3): + other = get_client() + assert other is not client + assert other.get_trace_url(trace_id=TRACE_ID) == expected + + assert api.projects.get.call_count == 1 + + +def test_a_failed_lookup_is_not_retried_by_a_later_client_instance(client): + """The negative outcome has to outlive the instance too. + + An auth failure is the case that produced a blocking request per rendered + row, so caching it only for the lifetime of one `Langfuse` object would + leave the original problem in place. + """ + api = Mock() + api.projects.get.side_effect = RuntimeError("401 Unauthorized") + client.api = api + + assert client.get_trace_url(trace_id=TRACE_ID) is None + + for _ in range(3): + assert get_client().get_trace_url(trace_id=TRACE_ID) is None + + assert api.projects.get.call_count == 1 + + +def test_fork_replaces_the_project_id_lock(client): + """A lock held at fork time would deadlock the child, so it is replaced. + + `_at_fork_reinit` resets it before the `_shutdown` early return, for the same + reason the class lock is reset there: the child needs a usable lock whether + or not this instance was torn down. Setting `_shutdown` here stops the method + before it recreates HTTP clients and consumer threads, which this test does + not exercise. + """ + resources = client._resources + held = resources._project_id_lock + held.acquire() # a thread in the parent, mid-lookup when the fork happened + resources._shutdown = True + + resources._at_fork_reinit() + + assert resources._project_id_lock is not held + assert not resources._project_id_lock.locked()