Skip to content

Navigation Menu

Sign in
Sign up

fix(client): resolve project id once per resource manager, never raise from get_trace_url #1813

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hchittanuru3 wants to merge 3 commits into langfuse:main
base: main
Choose a base branch
Loading
from hchittanuru3:fix/project-id-resolve-once
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 15 additions & 9 deletions langfuse/_client/client.py
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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.
Expand Down
64 changes: 64 additions & 0 deletions langfuse/_client/resource_manager.py
View file Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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.

Expand All @@ -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

Expand Down
187 changes: 187 additions & 0 deletions tests/unit/test_trace_url.py
View file Open in desktop
Original file line number Diff line number Diff line change
@@ -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()

AltStyle によって変換されたページ (->オリジナル) /