FastAPI: fail to reset trace_id with large (multi-chunk) request payload

Describe your environment

OS: Ubuntu 22
Python version: Python 3.10.12
Package version: 0.57b0
opentelemetry-instrumentation==0.57b0
opentelemetry-instrumentation-fastapi==0.57b0
opentelemetry-sdk==1.36.0
fastapi==0.116.1
starlette==0.47.2

What happened?

When enable tracing on FastApi request with large incoming payload (which gets streamed as multiple chunks), the trace_id does not get reset for the next Request coming over the same http connection (keep-alive).

It looks like trace context gets leaked and doesn't get reset/close properly and gets attached to the http connection, as long as the connection is kept alive, any subsequent request would get the same trace_id as first request.

Steps to Reproduce

Below is a minimal runnable example with two python files: app.py and test_app.py

app.py:

import asyncio
import logging
from fastapi import FastAPI, Request
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor

# Set up logging to see the trace IDs
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- OpenTelemetry Setup ---
# A simple tracer provider to print traces to the console
provider = TracerProvider()
processor = SimpleSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

app = FastAPI()
# Instrument the app. This middleware sets up tracing for each request.
FastAPIInstrumentor.instrument_app(app)

@app.post("/predict")
async def predict(request: Request):
    """
    This endpoint manually reads the request body, which triggers the bug
    when the payload is sent in multiple chunks.
    """
    logger.info("Handling /predict request...")
    
    # Manually reading the body to reproduce the issue
    request_bytes = await request.body()
    
    logger.info("Received payload of %d bytes", len(request_bytes))
    
    # Simulate some processing
    await asyncio.sleep(0.5)
    
    # Get the current trace ID and print it
    current_span = trace.get_current_span()
    trace_id = format(current_span.get_span_context().trace_id, 'x')
    logger.info("Trace ID for /predict: %s", trace_id)
    
    return {"status": "ok", "message": "Payload received", "trace_id": trace_id}

@app.get("/test")
async def test_trace_id():
    """
    This endpoint is used to check if the trace ID has been reset.
    """
    logger.info("Handling /test request...")
    
    # Get the current trace ID and print it
    current_span = trace.get_current_span()
    trace_id = format(current_span.get_span_context().trace_id, 'x')
    logger.info("Trace ID for /test: %s", trace_id)
    
    return {"status": "ok", "message": "Trace ID checked", "trace_id": trace_id}

test_app.py:

import asyncio
import httpx
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def run_test():
    # A large payload (10MB) to ensure it's sent in multiple chunks
    large_payload = b'A' * 1024 * 1024 * 10
    
    # Use httpx.AsyncClient as a context manager to maintain a keep-alive connection
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        logger.info("--- Sending first request to /predict with large payload ---")
        try:
            response1 = await client.post(
                "/predict",
                content=large_payload,
                headers={"Content-Type": "application/octet-stream"}
            )
            response1.raise_for_status()
            logger.info("First request successful: %s", response1.json())
        except httpx.HTTPStatusError as e:
            logger.error("Request 1 failed: %s", e.response.text)
        
        # Add a short delay to ensure the server is ready for the next request
        await asyncio.sleep(1)

        logger.info("\n--- Sending second request to /test on the same connection ---")
        try:
            response2 = await client.get("/test")
            response2.raise_for_status()
            logger.info("Second request successful: %s", response2.json())
        except httpx.HTTPStatusError as e:
            logger.error("Request 2 failed: %s", e.response.text)

if __name__ == "__main__":
    asyncio.run(run_test())

Run above two scripts in separate process:

uvicorn app:app &
python test_app.py

Expected Result

second request GET /test gets a new trace id (different than the first request POST /predict)

Actual Result

second request gets the same trace id as first request

INFO:__main__:--- Sending first request to /predict with large payload ---
INFO:httpx:HTTP Request: POST http://localhost:8000/predict "HTTP/1.1 200 OK"
INFO:__main__:First request successful: {'status': 'ok', 'message': 'Payload received', 'trace_id': '5c2d0e099c31993a72d823959aaf52d2'}
INFO:__main__:
--- Sending second request to /test on the same connection ---
INFO:httpx:HTTP Request: GET http://localhost:8000/test "HTTP/1.1 200 OK"
INFO:__main__:Second request successful: {'status': 'ok', 'message': 'Trace ID checked', 'trace_id': '5c2d0e099c31993a72d823959aaf52d2'}

Additional context

If we make request payload small enough (to be sent in 1 chunk), the second request would get a new trace id as expected

Would you like to implement a fix?

None

Run the app.py and test_app.py reproduction with the versions listed, focusing on FastAPIInstrumentor.instrument_app and the large multi-chunk request. Trace the request context across the keep-alive connection and compare the two returned trace IDs. Done means the subsequent GET /test receives a different trace ID from the preceding large POST /predict request.