Merge branch '3.17' into backport-14911-to-3.17

brettlangdon · web-flow · commit a441f93c614d · 2025-10-28T11:39:12.000-04:00
diff --git a/ddtrace/contrib/internal/langchain/patch.py b/ddtrace/contrib/internal/langchain/patch.py
@@ -320,10 +320,13 @@ def _on_span_started(span: Span):
         integration.record_instance(instance, span)
 
     def _on_span_finished(span: Span, streamed_chunks):
-        joined_chunks = streamed_chunks[0]
-        for chunk in streamed_chunks[1:]:
-            joined_chunks += chunk  # base message types support __add__ for concatenation
         kwargs["_dd.identifying_params"] = instance._identifying_params
+        if len(streamed_chunks):
+            joined_chunks = streamed_chunks[0]
+            for chunk in streamed_chunks[1:]:
+                joined_chunks += chunk  # base message types support __add__ for concatenation
+        else:
+            joined_chunks = []
         integration.llmobs_set_tags(span, args=args, kwargs=kwargs, response=joined_chunks, operation="chat")
 
     return shared_stream(
diff --git a/ddtrace/internal/endpoints.py b/ddtrace/internal/endpoints.py
@@ -106,15 +106,16 @@ def flush(self, max_length: int) -> dict:
         """
         Flush the endpoints to a payload, returning the first `max` endpoints.
         """
+        endpoints_snapshot = tuple(self.endpoints)
         if max_length >= len(self.endpoints):
             res = {
                 "is_first": self.is_first,
-                "endpoints": [dataclasses.asdict(ep, dict_factory=_dict_factory) for ep in self.endpoints],
+                "endpoints": [dataclasses.asdict(ep, dict_factory=_dict_factory) for ep in endpoints_snapshot],
             }
             self.reset()
             return res
         else:
-            batch = [self.endpoints.pop() for _ in range(max_length)]
+            batch = tuple(self.endpoints.pop() for _ in range(max_length))
             res = {
                 "is_first": self.is_first,
                 "endpoints": [dataclasses.asdict(ep, dict_factory=_dict_factory) for ep in batch],
diff --git a/releasenotes/notes/langchain-early-stream-end-fix-af9ca6475b3aaf00.yaml b/releasenotes/notes/langchain-early-stream-end-fix-af9ca6475b3aaf00.yaml
@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    langchain: Fixes an issue where streamed responses that end before the first chunk is received would result in an ``IndexError``.
diff --git a/releasenotes/notes/potential-race-endpoints-b8592670cdcbf023.yaml b/releasenotes/notes/potential-race-endpoints-b8592670cdcbf023.yaml
@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Fix a potential race condition in the tracer.
diff --git a/tests/contrib/langchain/test_langchain.py b/tests/contrib/langchain/test_langchain.py
@@ -539,3 +539,17 @@ def circumference_tool(radius: float) -> float:
     )
 
     calculator.invoke("2", config={"unserializable": object()})
+
+
+@pytest.mark.snapshot(ignores=["meta.error.stack", "meta.error.message"])
+def test_streamed_chat_model_with_no_output(langchain_openai, openai_url):
+    from openai import APITimeoutError
+
+    chat_model = langchain_openai.ChatOpenAI(base_url=openai_url, timeout=0.0001)
+
+    result = chat_model.stream("Hello, my name is")
+    try:
+        next(result)
+    except Exception as e:
+        if not isinstance(e, APITimeoutError):
+            assert False, f"Expected APITimeoutError, got {e}"
diff --git a/tests/internal/test_endpoints.py b/tests/internal/test_endpoints.py
@@ -0,0 +1,179 @@
+import threading
+from time import sleep
+
+import pytest
+
+from ddtrace.internal.endpoints import HttpEndPointsCollection
+
+
+@pytest.fixture
+def collection():
+    coll = HttpEndPointsCollection()
+    coll.reset()
+    yield coll
+    coll.reset()
+
+
+def test_flush_uses_tuple_snapshot(collection):
+    """Test that flush() operates on a tuple snapshot, not the original set."""
+    collection.add_endpoint("GET", "/api/users")
+    collection.add_endpoint("POST", "/api/users")
+    collection.add_endpoint("DELETE", "/api/users/123")
+
+    assert len(collection.endpoints) == 3
+    result = collection.flush(max_length=10)
+    assert result["is_first"] is True
+    assert len(result["endpoints"]) == 3
+    assert len(collection.endpoints) == 0
+
+
+def test_flush_snapshot_prevents_modification_during_iteration(collection):
+    """Test that modifying self.endpoints during flush iteration doesn't cause RuntimeError."""
+    collection.add_endpoint("GET", "/api/v1")
+    collection.add_endpoint("POST", "/api/v2")
+    collection.add_endpoint("PUT", "/api/v3")
+
+    initial_count = len(collection.endpoints)
+    assert initial_count == 3
+    result = collection.flush(max_length=10)
+
+    assert len(result["endpoints"]) == initial_count
+    assert result["is_first"] is True
+
+
+def test_concurrent_add_during_flush_does_not_break_iteration(collection):
+    """Test that adding endpoints from another thread during flush doesn't cause RuntimeError."""
+    for i in range(5):
+        collection.add_endpoint("GET", f"/api/endpoint{i}")
+
+    assert len(collection.endpoints) == 5
+
+    flush_completed = threading.Event()
+    flush_result = {}
+    exception_caught = []
+
+    def flush_thread():
+        try:
+            result = collection.flush(max_length=10)
+            flush_result["data"] = result
+            flush_completed.set()
+        except Exception as e:
+            exception_caught.append(e)
+            flush_completed.set()
+
+    def add_thread():
+        sleep(0.001)
+
+        # Try to modify the set while flush might be iterating
+        for i in range(5, 10):
+            collection.add_endpoint("POST", f"/api/new{i}")
+            sleep(0.001)
+
+    t1 = threading.Thread(target=flush_thread)
+    t2 = threading.Thread(target=add_thread)
+
+    t1.start()
+    t2.start()
+
+    t1.join(timeout=2.0)
+    t2.join(timeout=2.0)
+
+    assert flush_completed.is_set(), "Flush did not complete"
+    assert len(exception_caught) == 0, f"Exception occurred during flush: {exception_caught}"
+    assert "data" in flush_result, "Flush did not return a result"
+
+    result = flush_result["data"]
+    assert "endpoints" in result
+    assert "is_first" in result
+
+
+def test_flush_with_partial_batch(collection):
+    """Test that flush creates a tuple snapshot even when using pop() for partial batches."""
+    for i in range(10):
+        collection.add_endpoint("GET", f"/api/endpoint{i}")
+
+    assert len(collection.endpoints) == 10
+
+    result = collection.flush(max_length=5)
+
+    assert len(result["endpoints"]) == 5
+    assert result["is_first"] is True
+    assert len(collection.endpoints) == 5
+
+    result2 = collection.flush(max_length=10)
+    assert len(result2["endpoints"]) == 5
+    assert result2["is_first"] is False  # Not first anymore
+
+    assert len(collection.endpoints) == 0
+
+
+def test_partial_flush_with_concurrent_modification(collection):
+    """Test that partial flush (max_length < size) is safe from race conditions."""
+    for i in range(10):
+        collection.add_endpoint("GET", f"/api/endpoint{i}")
+
+    assert len(collection.endpoints) == 10
+
+    flush_completed = threading.Event()
+    flush_result = {}
+    exception_caught = []
+
+    def flush_thread():
+        try:
+            # Partial flush - this should trigger the else branch at line 118
+            result = collection.flush(max_length=5)
+            flush_result["data"] = result
+            flush_completed.set()
+        except Exception as e:
+            exception_caught.append(e)
+            flush_completed.set()
+
+    def add_thread():
+        sleep(0.001)
+        # Try to modify the set while flush might be iterating
+        for i in range(10, 15):
+            collection.add_endpoint("POST", f"/api/new{i}")
+            sleep(0.001)
+
+    t1 = threading.Thread(target=flush_thread)
+    t2 = threading.Thread(target=add_thread)
+
+    t1.start()
+    t2.start()
+
+    t1.join(timeout=2.0)
+    t2.join(timeout=2.0)
+
+    assert flush_completed.is_set(), "Flush did not complete"
+    assert len(exception_caught) == 0, f"Exception occurred during flush: {exception_caught}"
+    assert "data" in flush_result, "Flush did not return a result"
+
+    result = flush_result["data"]
+    assert len(result["endpoints"]) == 5
+    assert "is_first" in result
+
+
+def test_http_endpoint_hash_consistency(collection):
+    """Test that HttpEndPoint hashing works correctly for set operations."""
+    collection.add_endpoint("GET", "/api/test")
+    collection.add_endpoint("GET", "/api/test")
+    assert len(collection.endpoints) == 1
+
+    collection.add_endpoint("POST", "/api/test")
+    collection.add_endpoint("GET", "/api/other")
+    assert len(collection.endpoints) == 3
+
+
+def test_snapshot_is_tuple_type(collection):
+    """Verify that the snapshot created in flush is actually a tuple."""
+    collection.add_endpoint("GET", "/test")
+    collection.add_endpoint("POST", "/test")
+    assert isinstance(collection.endpoints, set)
+
+    result = collection.flush(max_length=10)
+    assert len(result["endpoints"]) == 2
+
+    for ep in result["endpoints"]:
+        assert isinstance(ep, dict)
+        assert "method" in ep
+        assert "path" in ep
diff --git a/tests/llmobs/llmobs_cassettes/openai/openai_chat_completions_post_0fdef9ab.yaml b/tests/llmobs/llmobs_cassettes/openai/openai_chat_completions_post_0fdef9ab.yaml
@@ -0,0 +1,154 @@
+interactions:
+- request:
+    body: '{"messages":[{"content":"Hello, my name is","role":"user"}],"model":"gpt-3.5-turbo","n":1,"stream":true,"temperature":0.7}'
+    headers:
+      ? !!python/object/apply:multidict._multidict.istr
+      - Accept
+      : - application/json
+      ? !!python/object/apply:multidict._multidict.istr
+      - Accept-Encoding
+      : - gzip, deflate
+      ? !!python/object/apply:multidict._multidict.istr
+      - Connection
+      : - keep-alive
+      Content-Length:
+      - '122'
+      ? !!python/object/apply:multidict._multidict.istr
+      - Content-Type
+      : - application/json
+      ? !!python/object/apply:multidict._multidict.istr
+      - User-Agent
+      : - OpenAI/Python 1.109.1
+      ? !!python/object/apply:multidict._multidict.istr
+      - X-Stainless-Arch
+      : - arm64
+      ? !!python/object/apply:multidict._multidict.istr
+      - X-Stainless-Async
+      : - 'false'
+      ? !!python/object/apply:multidict._multidict.istr
+      - X-Stainless-Lang
+      : - python
+      ? !!python/object/apply:multidict._multidict.istr
+      - X-Stainless-OS
+      : - MacOS
+      ? !!python/object/apply:multidict._multidict.istr
+      - X-Stainless-Package-Version
+      : - 1.109.1
+      ? !!python/object/apply:multidict._multidict.istr
+      - X-Stainless-Runtime
+      : - CPython
+      ? !!python/object/apply:multidict._multidict.istr
+      - X-Stainless-Runtime-Version
+      : - 3.10.13
+      ? !!python/object/apply:multidict._multidict.istr
+      - x-stainless-read-timeout
+      : - '0.0001'
+      ? !!python/object/apply:multidict._multidict.istr
+      - x-stainless-retry-count
+      : - '0'
+    method: POST
+    uri: https://api.openai.com/v1/chat/completions
+  response:
+    body:
+      string: 'data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"role":"assistant","content":"","refusal":null},"logprobs":null,"finish_reason":null}],"obfuscation":"dDINu"}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"content":"Assistant"},"logprobs":null,"finish_reason":null}],"obfuscation":"1yL6DImc2cky67"}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"content":"."},"logprobs":null,"finish_reason":null}],"obfuscation":"kyn0dy"}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"content":"
+        How"},"logprobs":null,"finish_reason":null}],"obfuscation":"zF8"}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"content":"
+        can"},"logprobs":null,"finish_reason":null}],"obfuscation":"ZOA"}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"content":"
+        I"},"logprobs":null,"finish_reason":null}],"obfuscation":"yPnQN"}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"content":"
+        assist"},"logprobs":null,"finish_reason":null}],"obfuscation":""}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"content":"
+        you"},"logprobs":null,"finish_reason":null}],"obfuscation":"nG8"}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"content":"
+        today"},"logprobs":null,"finish_reason":null}],"obfuscation":"a"}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{"content":"?"},"logprobs":null,"finish_reason":null}],"obfuscation":"KxSued"}
+
+
+        data: {"id":"chatcmpl-CTDgKS9E25YxCQOhLhNmhfOPiznNk","object":"chat.completion.chunk","created":1761080140,"model":"gpt-3.5-turbo-0125","service_tier":"default","system_fingerprint":null,"choices":[{"index":0,"delta":{},"logprobs":null,"finish_reason":"stop"}],"obfuscation":"Q"}
+
+
+        data: [DONE]
+
+
+        '
+    headers:
+      CF-RAY:
+      - 9923a83a5cd9081d-IAD
+      Connection:
+      - keep-alive
+      Content-Type:
+      - text/event-stream; charset=utf-8
+      Date:
+      - Tue, 21 Oct 2025 20:55:40 GMT
+      Server:
+      - cloudflare
+      Set-Cookie:
+      - __cf_bm=o7TNYeatXbBOlFxeMlhl.8fe7kXVRNTm_dL98zIje8M-1761080140-1.0.1.1-FRpu._KCnEk.aGZG5YQ75Od_Ucq8okx9WLNY3JjdbbK3P7mwxS21FIvTtyY6GlllpujKEYLWkFHG6VIIw4zZmwH0yDL04t_gvEzyjWwc1qc;
+        path=/; expires=Tue, 21-Oct-25 21:25:40 GMT; domain=.api.openai.com; HttpOnly;
+        Secure; SameSite=None
+      - _cfuvid=bZiyEgtQ9lNlCOG2DaVgkJLkXs33574MrsoVFl0iIf4-1761080140438-0.0.1.1-604800000;
+        path=/; domain=.api.openai.com; HttpOnly; Secure; SameSite=None
+      Strict-Transport-Security:
+      - max-age=31536000; includeSubDomains; preload
+      Transfer-Encoding:
+      - chunked
+      X-Content-Type-Options:
+      - nosniff
+      access-control-expose-headers:
+      - X-Request-ID
+      alt-svc:
+      - h3=":443"; ma=86400
+      cf-cache-status:
+      - DYNAMIC
+      openai-organization:
+      - datadog-staging
+      openai-processing-ms:
+      - '151'
+      openai-project:
+      - proj_gt6TQZPRbZfoY2J9AQlEJMpd
+      openai-version:
+      - '2020-10-01'
+      x-envoy-upstream-service-time:
+      - '174'
+      x-openai-proxy-wasm:
+      - v0.1
+      x-ratelimit-limit-requests:
+      - '10000'
+      x-ratelimit-limit-tokens:
+      - '50000000'
+      x-ratelimit-remaining-requests:
+      - '9999'
+      x-ratelimit-remaining-tokens:
+      - '49999993'
+      x-ratelimit-reset-requests:
+      - 6ms
+      x-ratelimit-reset-tokens:
+      - 0s
+      x-request-id:
+      - req_c9c685c9da2e4684b4613cefd4af98e8
+    status:
+      code: 200
+      message: OK
+version: 1
diff --git a/tests/llmobs/llmobs_cassettes/openai/openai_chat_completions_post_eb4ba851.yaml b/tests/llmobs/llmobs_cassettes/openai/openai_chat_completions_post_eb4ba851.yaml
diff --git a/tests/snapshots/tests.contrib.langchain.test_langchain.test_streamed_chat_model_with_no_output.json b/tests/snapshots/tests.contrib.langchain.test_langchain.test_streamed_chat_model_with_no_output.json

-Original file line number
+Diff line change
@@ @@ -0,0 +1,4 @@ @@
 +---
 +fixes:
 +  - |
 +    langchain: Fixes an issue where streamed responses that end before the first chunk is received would result in an ``IndexError``.