fix(otel): keep last duplicate span in masking batches

langfuse/_client/client.py

@@ -222,7 +222,7 @@ class Langfuse:
 
             The hook receives one OpenTelemetry export batch. A batch is not guaranteed to contain a complete trace, request, or Langfuse observation tree. The hook usually runs on the OpenTelemetry batch span processor worker thread; during `flush()` and shutdown it may run on the caller thread. Keep it synchronous, deterministic, and fast.
 
-            Return `None` to leave the batch unchanged. Return `MaskOtelSpansResult` with `OtelSpanPatch` values to delete or replace attributes on selected spans. If the hook raises or returns an invalid batch result, Langfuse drops the whole export batch. If one returned span patch is invalid, Langfuse drops only that span from the Langfuse export.
+            Return `None` to leave the batch unchanged. Return `MaskOtelSpansResult` with `OtelSpanPatch` values to delete or replace attributes on selected spans. If a batch contains duplicate trace and span identifiers, Langfuse keeps only the last matching span. If the hook raises or returns an invalid batch result, Langfuse drops the whole export batch. If one returned span patch is invalid, Langfuse drops only that span from the Langfuse export.
 
             Example:
                 ```python

langfuse/_client/span_exporter.py

@@ -264,10 +264,9 @@ def _apply_mask_otel_spans(
         span_attributes: Sequence[tuple[ReadableSpan, Dict[str, AttributeValue]]],
     ) -> Optional[list[tuple[ReadableSpan, Dict[str, AttributeValue]]]]:
         mask_otel_spans = cast(MaskOtelSpansFunction, self._mask_otel_spans)
-        maskable_span_attributes: list[
-            tuple[ReadableSpan, Dict[str, AttributeValue]]
-        ] = []
-        span_data_by_identifier: Dict[OtelSpanIdentifier, OtelSpanData] = {}
+        span_attributes_by_identifier: Dict[
+            OtelSpanIdentifier, tuple[ReadableSpan, Dict[str, AttributeValue]]
+        ] = {}
 
         for span, attributes in span_attributes:
             if not _has_valid_span_context(span):
@@ -279,22 +278,23 @@ def _apply_mask_otel_spans(
 
             identifier = _create_otel_span_identifier(span)
 
-            if identifier in span_data_by_identifier:
-                langfuse_logger.error(
-                    "Masking error: mask_otel_spans received duplicate span identifiers. "
-                    "Dropping export batch. "
-                    f"trace_id='{identifier.trace_id}' span_id='{identifier.span_id}'"
-                )
-                return None
+            if identifier in span_attributes_by_identifier:
+                span_attributes_by_identifier.pop(identifier)
 
-            span_data_by_identifier[identifier] = _create_otel_span_data(
-                span=span, attributes=attributes, identifier=identifier
-            )
-            maskable_span_attributes.append((span, attributes))
+            span_attributes_by_identifier[identifier] = (span, attributes)
+
+        maskable_span_attributes = list(span_attributes_by_identifier.values())
 
         if not maskable_span_attributes:
             return []
 
+        span_data_by_identifier = {
+            identifier: _create_otel_span_data(
+                span=span, attributes=attributes, identifier=identifier
+            )
+            for identifier, (span, attributes) in span_attributes_by_identifier.items()
+        }
+
         try:
             result: Any = mask_otel_spans(
                 params=MaskOtelSpansParams(

langfuse/types.py

@@ -126,7 +126,9 @@ class MaskOtelSpansParams:
     A single call receives one OpenTelemetry export batch, not necessarily a
     complete trace, request, or Langfuse observation tree. Batch contents depend
     on OpenTelemetry span processor settings such as `flush_at`,
-    `flush_interval`, explicit `flush()`, and shutdown.
+    `flush_interval`, explicit `flush()`, and shutdown. If a batch contains
+    duplicate trace and span identifiers, Langfuse keeps only the last matching
+    span before calling the masking function.
 
     Example:
         ```python

tests/unit/test_mask_otel_spans.py

@@ -574,6 +574,83 @@ def mask_otel_spans(*, params: MaskOtelSpansParams):
     )
 
 
+def test_mask_otel_spans_keeps_last_duplicate_without_dropping_batch():
+    exporter = InMemorySpanExporter()
+    seen_params: list[MaskOtelSpansParams] = []
+
+    def mask_otel_spans(*, params: MaskOtelSpansParams):
+        seen_params.append(params)
+        duplicate_identifier = next(
+            identifier
+            for identifier, span in params.spans.items()
+            if span.name == "duplicate-last"
+        )
+
+        return MaskOtelSpansResult(
+            span_patches={
+                duplicate_identifier: OtelSpanPatch(
+                    set_attributes={"masking.applied": True}
+                )
+            }
+        )
+
+    transforming_exporter = span_exporter_module.LangfuseTransformingSpanExporter(
+        exporter=exporter,
+        media_manager=None,
+        mask_otel_spans=mask_otel_spans,
+    )
+    duplicate_context = SpanContext(
+        trace_id=1,
+        span_id=2,
+        is_remote=False,
+        trace_flags=TraceFlags(TraceFlags.SAMPLED),
+        trace_state=TraceState(),
+    )
+    unrelated_context = SpanContext(
+        trace_id=3,
+        span_id=4,
+        is_remote=False,
+        trace_flags=TraceFlags(TraceFlags.SAMPLED),
+        trace_state=TraceState(),
+    )
+    spans = [
+        ReadableSpan(
+            name="duplicate-first",
+            context=duplicate_context,
+            attributes={"attempt": 1},
+        ),
+        ReadableSpan(
+            name="unrelated",
+            context=unrelated_context,
+            attributes={"unrelated": True},
+        ),
+        ReadableSpan(
+            name="duplicate-last",
+            context=duplicate_context,
+            attributes={"attempt": 2},
+        ),
+    ]
+
+    result = transforming_exporter.export(spans)
+
+    assert result == SpanExportResult.SUCCESS
+    assert len(seen_params) == 1
+    assert [span.name for span in seen_params[0].spans.values()] == [
+        "unrelated",
+        "duplicate-last",
+    ]
+
+    exported_spans = exporter.get_finished_spans()
+    assert [span.name for span in exported_spans] == [
+        "unrelated",
+        "duplicate-last",
+    ]
+    assert exported_spans[1].attributes == {
+        "attempt": 2,
+        "masking.applied": True,
+    }
+
+
 def test_exporter_exception_does_not_stop_background_export_thread():
     exporter = FailsOnceSpanExporter()
     media_manager, _ = _media_manager()