Thoughts about loop issue

[WilliamDEdwards]

Hi,

We recently experienced middleware being in a 'loop' on our production environment. I'm wondering if somebody has a clever idea as to what logic issue we're hitting here.

I built an event/listener system for our FastAPI application. To that end, we have the following middleware, which 'flushes' events that were 'deferred' (my own terminology) during the request:

class FlushDeferredEventsMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        try:
            await self.app(scope, receive, send)
        finally:
            event_dispatcher.flush()

Recently, we hit a bug that caused the flush call to fail. Full traceback:

Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]: ERROR:    Exception in ASGI application
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]: Traceback (most recent call last):
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/uvicorn/protocols/http/httptools_impl.py", line 416, in run_asgi
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     result = await app(  # type: ignore[func-returns-value]
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:         self.scope, self.receive, self.send
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     )
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     ^
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/uvicorn/middleware/proxy_headers.py", line 60, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     return await self.app(scope, receive, send)
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/fastapi/applications.py", line 1135, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     await super().__call__(scope, receive, send)
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/starlette/applications.py", line 107, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     await self.middleware_stack(scope, receive, send)
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/starlette/middleware/errors.py", line 186, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     raise exc
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/starlette/middleware/errors.py", line 164, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     await self.app(scope, receive, _send)
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/opentelemetry/instrumentation/asgi/__init__.py", line 795, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     await self.app(scope, otel_receive, otel_send)
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/starlette/middleware/errors.py", line 186, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     raise exc
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/starlette/middleware/errors.py", line 164, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     await self.app(scope, receive, _send)
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/opentelemetry/instrumentation/fastapi/__init__.py", line 307, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     await self.app(scope, receive, send)
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/app/api/middleware.py", line 13, in __call__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     event_dispatcher.flush()
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     ~~~~~~~~~~~~~~~~~~~~~~^^
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/app/events/dispatcher.py", line 25, in flush
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     listener_instance.handle()
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     ~~~~~~~~~~~~~~~~~~~~~~~~^^
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/app/listeners/object_log.py", line 59, in handle
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     dto = ObjectLogCreateDto(
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:         object_id=self.event.database_schema.id,
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     ...<5 lines>...
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:         type=type_,
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     )
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   File "/opt/clusterapi/lib/python3.13/site-packages/pydantic/main.py", line 250, in __init__
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]: pydantic_core._pydantic_core.ValidationError: 1 validation error for ObjectLogCreateDto
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]: object_model_name
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:   String should match pattern '^[a-zA-Z]+$' [type=string_pattern_mismatch, input_value='n8ninstance', input_type=str]
Jan 30 18:19:32 cluster-api1.cyberfusion.cloud python3[20505]:     For further information visit https://errors.pydantic.dev/2.12/v/string_pattern_mismatch

So far, so good. But this exception started occurring at least once every second (and not that many requests were coming in, so I doubt it's some ASGI race condition).

In other words, our middleware was in a loop. I have not found any logic for Starlette to 'retry' middleware. I did find https://starlette.dev/middleware/#asgi-middleware-should-be-stateless, which states that "middleware should be stateless", which is a bit of a grey area in my case...

event_dispatcher is a ContextVar instance of EventDispatcher:

event_dispatcher: EventDispatcher = ContextVar(
    "event_dispatcher", default=EventDispatcher()
).get()

... which contains a list of event listeners to handle:

class EventDispatcher:
    def __init__(self) -> None:
        self._listener_instances: list[ListenerInterface] = []

... and therefore arguably contains state outside __call__ as the documentation prescribes, but not with the ability to "lead to conflicting variable reads/writes across requests" as ContextVars are per-request. (Let me know if I'm skipping a thought here.)

I'm unable to reproduce this issue locally, causing the same ValidationError as in production:

class FlushDeferredEventsMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        try:
            await self.app(scope, receive, send)
        finally:
            class X(BaseModel):
                x: int

            X(x='a')
            #event_dispatcher.flush()

I haven't been able to wrap my head around how all this came together to form an infinite loop (I had to restart Uvicorn to 'fix' this).

Clever ideas?

[cookesan]

I would look closely at this line:

event_dispatcher: EventDispatcher = ContextVar(
    "event_dispatcher", default=EventDispatcher()
).get()

That does not leave you with a per-request ContextVar. It calls .get() at import time and assigns the returned default EventDispatcher instance to event_dispatcher. If that dispatcher holds a list of listener instances, that list is shared process-wide unless you clear or replace it yourself.

So if flush() fails before removing the bad listener/event, every later request that reaches the middleware can try to flush the same bad state again. That can look like Starlette retrying the middleware, even though it is really shared state being flushed repeatedly.

I would keep the actual ContextVar object, set a fresh dispatcher for the request, call .get() inside the request path, and clear it in a finally. Also consider catching/logging flush errors so a bad deferred event does not poison the process-global queue.

[WilliamDEdwards]

Hi @cookesan,

Thanks for your reply.

From my original message:

this exception started occurring at least once every second (and not that many requests were coming in, so I doubt it's some ASGI race condition).