diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 2b2889545..50d0d8e6d 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -350,7 +350,6 @@ def __init__( or os.environ.get(LANGFUSE_RELEASE, None) or get_common_release_envs() ) - self._project_id: Optional[str] = None 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,15 +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.""" - if not self._project_id: - proj = self.api.projects.get() - if not proj.data or not proj.data[0].id: - return None - - self._project_id = proj.data[0].id + """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._resources is None: + langfuse_logger.debug( + "Operation skipped: _get_project_id - Client is not initialized." + ) + return None - return self._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 new file mode 100644 index 000000000..9b27ef1a4 --- /dev/null +++ b/tests/unit/test_trace_url.py @@ -0,0 +1,187 @@ +"""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 -- 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 +from types import SimpleNamespace +from typing import List, Optional +from unittest.mock import Mock + +import pytest + +from langfuse import Langfuse +from langfuse._client.get_client import get_client +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_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") + + 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()