async for message in ws exits for unknown reasons?

[emanuele3d]

Describe the bug

The "async for message in ws" loop ends inexplicably.

To Reproduce

I'm using this code to connect with a wss:// address and process the messages it returns:

async with aiohttp.ClientSession() as session:
    async with session.ws_connect(url=self._config['wssAddress'],
                                  autoping=True, autoclose=False,
                                  ssl=True) as self._client:

        await self.subscribe_to_streams(self._config['streams'])

        async for message in self._client:
            self.message_handler(message)

If self._config['streams] contains a list of streams known to the server, everything works flawlessly.

The problem arises when I intentionally try to subscribe to a stream that the server doesn't know about. In this case I receive the following message from the server:

WSMessage(type=<WSMsgType.TEXT: 1 >, data='{"error":{"code":2,"msg":"Invalid request: invalid stream"},"id":2}', extra = '')

The message itself is what I expect and doesn't seem to be special in any way. Mysteriously however, the flow of the program exits the loop and the connection is closed.

(This happens irrespective of autoclose being set to True or False, but I don't know if this is relevant)

Expected behavior

If the server closed the connection, i.e. because its logic accepts absolutely zero errors from me, I would expect aiohttp to tell me the closure was caused by the server or see one last message from the server warning me about the closure.

And there have been other instances of the loop exiting and the connection closing seemingly without reasons, at random times over multi-hours runs when everything seems to work. Attempting to subscribe to a stream that doesn't exist is just a way to trigger the issue reliably.

If something happened client side, I would also expect aiohttp to tell me, i.e. via an exception, but I get absolutely nothing. The loop just ends and the connection is closed.

Basically I'm left wondering: is the loop exiting ONLY if the websocket connection is closed? If it is, how do I know if it's the client or the server that triggered the closure? If it isn't what are the other causes?

Logs/tracebacks

Nothing interesting at this point: one moment I get the expected error message quoted above and the next I get the logging expected after the flow of the program exits the loop, nothing in-between hinting at the cause. =(

Python Version

$ python --version
Python 3.9.9

aiohttp Version

$ python -m pip show aiohttp
Name: aiohttp
Version: 3.8.1
Summary: Async http client/server framework (asyncio)
Home-page: https://github.com/aio-libs/aiohttp
Author:
Author-email:
License: Apache 2
Location: (...)
Requires: aiosignal, async-timeout, attrs, charset-normalizer, frozenlist, multidict, yarl
Required-by:

multidict Version

$ python -m pip show multidict
Name: multidict
Version: 6.0.2
Summary: multidict implementation
Home-page: https://github.com/aio-libs/multidict
Author: Andrew Svetlov
Author-email: [email protected]
License: Apache 2
Location: (...)
Requires:
Required-by: aiohttp, yarl

yarl Version

$ python -m pip show yarl
Name: yarl
Version: 1.7.2
Summary: Yet another URL library
Home-page: https://github.com/aio-libs/yarl/
Author: Andrew Svetlov
Author-email: [email protected]
License: Apache 2
Location: (...)
Requires: idna, multidict
Required-by: aiohttp

OS

Windows 10

Related component

Client

Additional context

No response

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[Dreamsorcerer]

The code is very simple and clear:

aiohttp/aiohttp/client_ws.py

Line 321 in 65bf205

if msg.type in (WSMsgType.CLOSE, WSMsgType.CLOSING, WSMsgType.CLOSED):

It exits because the server tells it the connection is being closed. This is the expected behaviour, not an error state that should raise an exception.

[emanuele3d]

Thank you for pointing me to the relevant code @Dreamsorcerer.

From what I can see the __anext__ method returns the message. So, if the server sends a message of type WSMsgType.CLOSE, WSMsgType.CLOSING, WSMsgType.CLOSED, shouldn't I be able to receive it and process it within the async for message in ws loop just before the loop is exited?

Or does the raise StopAsyncIteration interrupts the loop first and somehow I need to capture the returned message outside of the loop?

[emanuele3d]

Meanwhile I found the close_code attribute on the ClientWebSocketResponse object, after the connection is closed by the server, is set to 1008 (policy violation). I imagine this is actually part of the Close message I am missing and comes from the server?

[Dreamsorcerer]

Or does the raise StopAsyncIteration interrupts the loop first and somehow I need to capture the returned message outside of the loop?

Yes, raising an exception is the end of a function call, no value will be returned. StopIteration will stop the loop.

I imagine this is actually part of the Close message I am missing and comes from the server?

Looking at that same code, it appears that .receive() sets the close_code to various constants or the data from the server's message:

aiohttp/aiohttp/client_ws.py

Line 282 in 65bf205

self._close_code = msg.data

[emanuele3d]

Hmmm. Thank you for pointing that out.

Looking deeper I see that:

• A message of type WSMsgType.CLOSED doesn't affect the close_code property, only stops the loop.
• A message of type WSMsgType.CLOSING only sets the self._closing field and stops the loop.
• A message of type WSMsgType.CLOSE seems to be the only one that sets the close_code property, which means msg.data was simply the code '1008' without any kind of json structure around it.

Peculiar.

Well. Please @Dreamsorcerer, have a look at the Question #6710, would that be a good approach?

[emanuele3d]

Unless WebSocketError.code can be set to 1008? That's the only other thing that could store a 1008 code in close_code, but I don't know in what cases that exception gets thrown. Would that be a server-side or client-side thing?

[Dreamsorcerer]

* A message of type `WSMsgType.CLOSED` doesn't affect the `close_code` property, only stops the loop.

As far as I can see this in an internal message indicating that the connection is already closed, not a message from the server.

* A message of type WSMsgType.CLOSING only sets the `self._closing` field and stops the loop.

Again, this appears to be an internal message from calling .close(), not from the server.

* A message of type `WSMsgType.CLOSE` seems to be the only one that sets the `close_code` property, which means msg.data was simply the code '1008' without any kind of json structure around it.

The code for .json() is just loads(self.data), so you won't get any more information from that. There appears to be an extra field which is only used for CLOSE messages: https://docs.aiohttp.org/en/stable/websocket_utilities.html#aiohttp.WSMessage.extra

So, maybe we should save this as an attribute in addition to close_code. As a temporary workaround (and to check it contains the information you're looking for), try writing the loop manually with something like:

while msg.type not in (CLOSED, CLOSING, CLOSE):
    msg = await ws.receive()

[emanuele3d]

So, sounds like the connection is actually closed by the client? How do I find out why? I only see the loop exited.

[Dreamsorcerer]

Based on what? What was the last msg you see?