refactor(langfuse): fold 413 halving into bounded rounds instead of recursion
Some checks are pending
LiteLLM Rust / rust-lint (push) Waiting to run
LiteLLM Rust / rust-test (push) Waiting to run
LiteLLM Rust / rust-wheel (push) Waiting to run
Terraform Provider / gofmt, vet, build, test (push) Waiting to run
Terraform Provider / Provider endpoints vs proxy OpenAPI schema (push) Waiting to run

The code-quality recursive-function gate flagged LangfuseSpanExporter.export. A batch of n spans settles within n.bit_length() halving rounds, so the split is a reduce over a frozen round state with the same posts, logs and results. The TTL gate test now asserts the gate returns without raising

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-22 19:21:21 +00:00
parent d6086f0917
commit f0c38e1de2
2 changed files with 46 additions and 14 deletions

View file

@ -9,6 +9,7 @@ from collections.abc import Iterable, Mapping, Sequence
from contextvars import ContextVar
from dataclasses import dataclass, replace
from datetime import datetime
from functools import reduce
from hashlib import sha256
from importlib.metadata import version
from itertools import chain
@ -565,6 +566,19 @@ class DiscardingSpanExporter(SpanExporter):
_ExportOutcome = Literal["delivered", "retry", "rejected", "too_large"]
_Batch = tuple[ReadableSpan, ...]
@dataclass(frozen=True, slots=True)
class _Halving:
"""One round of a 413 split: the batches still to send and the results of the ones already settled."""
pending: tuple[_Batch, ...]
settled: tuple[SpanExportResult, ...] = ()
def _halves(batch: _Batch) -> tuple[_Batch, _Batch]:
return batch[: len(batch) // 2], batch[len(batch) // 2 :]
def enable_langfuse_debug_logging() -> None:
@ -596,29 +610,47 @@ class LangfuseSpanExporter(SpanExporter):
delays: Sequence[float] = (1.0, 2.0, 4.0)
def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
body: Final = _encode(spans)
"""Halving a batch of n spans settles every span within ``n.bit_length()`` rounds, so the rounds are a fixed
fold rather than a recursion."""
rounds: Final = range(len(spans).bit_length() + 1)
final: Final = reduce(lambda halving, _: self._round(halving), rounds, _Halving(pending=(tuple(spans),)))
return (
SpanExportResult.SUCCESS
if all(result is SpanExportResult.SUCCESS for result in final.settled)
else SpanExportResult.FAILURE
)
def _round(self, halving: _Halving) -> _Halving:
sent: Final = tuple((batch, self._send_batch(batch)) for batch in halving.pending)
return _Halving(
pending=tuple(half for batch, outcome in sent if outcome == "too_large" for half in _halves(batch)),
settled=halving.settled
+ tuple(
SpanExportResult.SUCCESS if outcome == "delivered" else SpanExportResult.FAILURE
for _, outcome in sent
if outcome != "too_large"
),
)
def _send_batch(self, batch: _Batch) -> _ExportOutcome:
"""A 413 on more than one span asks for halves; on a single span the span is dropped and reported."""
body: Final = _encode(batch)
if body is None:
return SpanExportResult.FAILURE
return "rejected"
outcome: Final = self._send(body)
if outcome != "too_large":
return SpanExportResult.SUCCESS if outcome == "delivered" else SpanExportResult.FAILURE
if len(spans) == 1:
return outcome
if len(batch) == 1:
verbose_logger.error(
"Langfuse rejected a single %d byte span export to %s as too large, dropping it",
len(body),
self.endpoint,
)
return SpanExportResult.FAILURE
return "rejected"
verbose_logger.warning(
"Langfuse rejected a %d byte export of %d spans as too large, resending in halves", len(body), len(spans)
)
half: Final = len(spans) // 2
results: Final = (self.export(spans[:half]), self.export(spans[half:]))
return (
SpanExportResult.SUCCESS
if all(r is SpanExportResult.SUCCESS for r in results)
else SpanExportResult.FAILURE
"Langfuse rejected a %d byte export of %d spans as too large, resending in halves", len(body), len(batch)
)
return "too_large"
def _send(self, body: bytes) -> _ExportOutcome:
for delay in self.delays:

View file

@ -2371,7 +2371,7 @@ def test_prompt_cache_ttl_typo_is_named_before_the_sdk_is_imported(monkeypatch,
@pytest.mark.parametrize("raw", ["5", " -3 ", "+0"], ids=["whole", "negative", "signed-zero"])
def test_whole_second_prompt_cache_ttl_passes_the_gate(monkeypatch, raw):
monkeypatch.setenv("LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS", raw)
langfuse_module.raise_if_unusable_prompt_cache_ttl()
assert langfuse_module.raise_if_unusable_prompt_cache_ttl() is None
def test_stopped_logger_hands_its_export_channel_back(monkeypatch):