[P2] Keep FastStream observability alive through complete shutdown #434

Priority: P2. Recommended before 2.0. Audit rank: 3.

Audited commit: 55903b3ccc19c77cb57b74332ee67bc2e61fedcd (2026-11-04). Reproductions ran on Python 3.14.7. Relevant installed versions: FastAPI 0.162.0, Starlette 1.10.0, FastStream 0.7.8, FastMCP 4.1.5, Sentry 2.73.0, structlog 26.1.0, prometheus-client 0.26.0 and OTel SDK 1.45.2. Floor validation remains part of implementation.

Problem and evidence

FastStreamBootstrapper attaches
teardown through on_shutdown. Installed FastStream 0.7.8 runs these hooks before broker.stop()
(faststream/_internal/application.py:284) and exits the user lifespan after shutdown
(faststream/asgi/app.py:249). An actual ASGI lifespan with only broker network start/stop mocked
exports a during-service span, but drops the user-shutdown span with the SDK message
Processor is already shutdown, ignoring call.

A preceding user shutdown callback that raises can also prevent the teardown hook from running;
this follows the sequential hook loop and is the same lifecycle ownership problem.

Bracket the complete lifecycle, with teardown in finally after broker/user cleanup. Merely
moving to after_shutdown still runs before user lifespan exit. Validate supported FastStream
floor/current versions, startup failure, shutdown failure and argument preservation before
choosing the hook. Record the ordering change in the migration guide under repository rules.

Reproduction

Save each block to a temporary .py file and run it in a fresh process from the repository using uv run --no-sync python /path/to/repro.py with the project development environment installed. Telemetry uses synthetic values and in-memory/mocked transports.

FastStream shutdown span

import contextlib
from unittest.mock import AsyncMock, patch
from faststream.asgi import AsgiFastStream
from faststream.redis import RedisBroker
from starlette.testclient import TestClient
from lite_bootstrap import FastStreamBootstrapper, FastStreamConfig
from opentelemetry.trace import get_tracer_provider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter


@contextlib.asynccontextmanager
async def lifespan():
    yield
    with get_tracer_provider().get_tracer("audit").start_as_current_span("user-shutdown"):
        pass


broker = RedisBroker()
app = AsgiFastStream(broker, lifespan=lifespan)
bootstrapper = FastStreamBootstrapper(
    FastStreamConfig(
        application=app,
        logging_enabled=False,
        opentelemetry_log_traces=True,
    )
)
bootstrapper.bootstrap()
exporter = InMemorySpanExporter()
get_tracer_provider().add_span_processor(SimpleSpanProcessor(exporter))
with patch.object(broker, "start", new=AsyncMock()), patch.object(broker, "stop", new=AsyncMock()), TestClient(app):
    with get_tracer_provider().get_tracer("audit").start_as_current_span("during-service"):
        pass
print("exported", [span.name for span in exporter.get_finished_spans()])
print("bootstrapper still active", bootstrapper.is_bootstrapped)

Acceptance criteria

• Start with a failing lifespan test proving a user-shutdown span is exported.
• Keep telemetry active through broker drain and user lifespan shutdown, then teardown exactly once.
• Cleanup still runs when startup or a shutdown callback fails; preserve the original error and lifespan arguments.
• Validate the lifecycle hook against supported FastStream floor/current versions and update the migration guide.