Asyncio TypeError: int() argument must be a string, a bytes-like object or a number, not 'NoneType'

[DanielK990]

Version:

redis 5.0.7
pip list:
channels 4.0.0
channels-redis 4.0.0
redis 4.5.1

Platform: Windows 10 WSL2, Python 3.8.10, redis server runs on a Raspberry Pi

Description:

I am using Django channels with Redis.

I am using a channel to send a message from

View ---> AsyncConsumer (background worker)

def myview(request):
async_to_sync(channel_layer.send)( # <--- here the error occurs
"start",
{
"type": "start",
},
)

Another message is send from this AsyncConsumer --> AsyncWebsocketConsumer

class MyAsyncConsumer(AsyncConsumer):
async def start(self, event):
logger = logging.getLogger('django')
logger.info(event)
await self.channel_layer.send(
event['channel'],
{
"type": "finish",
},
)

Finally in the WebsocketConsumer, sends a Websocket message to the Websocket Client

class MyWebsocketConsumer(AsyncWebsocketConsumer):

async def finish(self, data):
    logger = logging.getLogger('django')
    logger.info("Reply from Worker")
    message = {
        "type" : "finish",
    }
    await self.send(json.JSONEncoder().encode(message))

When sending the message from View ---> AsyncConsumer, there is the following error:

Internal Server Error: /
Traceback (most recent call last):
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/asgiref/sync.py", line 486, in thread_handler
raise exc_info[1]
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/django/core/handlers/exception.py", line 42, in inner
response = await get_response(request)
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/django/core/handlers/base.py", line 253, in _get_response_async
response = await wrapped_callback(
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/asgiref/sync.py", line 448, in call
ret = await asyncio.wait_for(future, timeout=None)
File "/usr/lib/python3.8/asyncio/tasks.py", line 455, in wait_for
return await fut
File "/usr/lib/python3.8/concurrent/futures/thread.py", line 57, in run
result = self.fn(*self.args, **self.kwargs)
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/asgiref/sync.py", line 490, in thread_handler
return func(*args, **kwargs)
File "/mnt/d/Documents/Python/myapp/myapp/myview/views.py", line 30, in myview
async_to_sync(channel_layer.send)(
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/asgiref/sync.py", line 240, in call
return call_result.result()
File "/usr/lib/python3.8/concurrent/futures/_base.py", line 437, in result
return self.__get_result()
File "/usr/lib/python3.8/concurrent/futures/_base.py", line 389, in __get_result
raise self._exception
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/asgiref/sync.py", line 306, in main_wrap
result = await self.awaitable(*args, **kwargs)
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/channels_redis/core.py", line 218, in send
await connection.zremrangebyscore(
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/redis/asyncio/client.py", line 514, in execute_command
return await conn.retry.call_with_retry(
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/redis/asyncio/retry.py", line 59, in call_with_retry
return await do()
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/redis/asyncio/client.py", line 488, in _send_command_parse_response
return await self.parse_response(conn, command_name, **options)
File "/mnt/d/Documents/python/myapp/myapp/.venv/lib/python3.8/site-packages/redis/asyncio/client.py", line 547, in parse_response
retval = self.response_callbacks[command_name](response, **options)
TypeError: int() argument must be a string, a bytes-like object or a number, not 'NoneType'

The AsyncConsumer worker never receives the message.

Channel Router Configuration:

application = ProtocolTypeRouter({
"http": django_asgi_app,
"channel": ChannelNameRouter({
"start": MyAsyncConsumer.as_asgi(),
}),
"websocket": AllowedHostsOriginValidator(
AuthMiddlewareStack(URLRouter(myapp.routing.websocket_urlpatterns))
),
})

[sileht]

We have the same issue from time to time:

    while (await self.redis_links.stream.zcard("streams")) > 0:
/private/var/folders/8d/j9lvdc21765533srqtq9309h0000gn/T/cdt.20221110-120225.57AuEoKl/venv/mergify-engine-kvCo2t7F-py3.11/lib/python3.11/site-packages/redis/asyncio/client.py:514: in execute_command
    return await conn.retry.call_with_retry(
/private/var/folders/8d/j9lvdc21765533srqtq9309h0000gn/T/cdt.20221110-120225.57AuEoKl/venv/mergify-engine-kvCo2t7F-py3.11/lib/python3.11/site-packages/redis/asyncio/retry.py:59: in call_with_retry
    return await do()
/private/var/folders/8d/j9lvdc21765533srqtq9309h0000gn/T/cdt.20221110-120225.57AuEoKl/venv/mergify-engine-kvCo2t7F-py3.11/lib/python3.11/site-packages/redis/asyncio/client.py:488: in _send_command_parse_response
    return await self.parse_response(conn, command_name, **options)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = Redis<ConnectionPool<Connection<host=localhost,port=33189,db=53,client_name=engine-<unknown>/functional-fixture/stream>>>, connection = Connection<host=localhost,port=33189,db=53,client_name=engine-<unknown>/functional-fixture/stream>, command_name = 'ZCARD', options = {}, response = None

    async def parse_response(
        self, connection: Connection, command_name: Union[str, bytes], **options
    ):
        """Parses a response from the Redis server"""
        try:
            if NEVER_DECODE in options:
                response = await connection.read_response(disable_decoding=True)
                options.pop(NEVER_DECODE)
            else:
                response = await connection.read_response()
        except ResponseError:
            if EMPTY_RESPONSE in options:
                return options[EMPTY_RESPONSE]
            raise

        if EMPTY_RESPONSE in options:
            options.pop(EMPTY_RESPONSE)

        if command_name in self.response_callbacks:
            # Mypy bug: https://github.com/python/mypy/issues/10977
            command_name = cast(str, command_name)
>           retval = self.response_callbacks[command_name](response, **options)
E           TypeError: int() argument must be a string, a bytes-like object or a real number, not 'NoneType'

[sevdog]

This may be related on the changes introduces in redis.asyncio.connection in 4.4.0:

v4.3.5...v4.4.0#diff-5924d2cbce9d62737f0024477d4a89f2314da1fd83da5b46dd02d85127d665cb

The current implementation of _readline method is very different from the previous one.

master

redis-py/redis/asyncio/connection.py

Lines 331 to 347 in 6c708c2

	async def _readline(self) -> bytes:
	"""
	read an unknown number of bytes up to the next '\r\n'
	line separator, which is discarded.
	"""
	found = self._buffer.find(b"\r\n", self._pos)
	if found >= 0:
	result = self._buffer[self._pos : found]
	else:
	tail = self._buffer[self._pos :]
	data = await self._stream.readline()
	if not data.endswith(b"\r\n"):
	raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
	result = (tail + data)[:-2]
	self._chunks.append(data)
	self._pos += len(result) + 2
	return result

PS: for some reason it uses the raw-bytes while there is the constant SYM_CRLF in the module.

4.3

redis-py/redis/asyncio/connection.py

Lines 304 to 324 in 54a1dce

	async def readline(self) -> bytes:
	buf = self._buffer
	if buf is None:
	raise RedisError("Buffer is closed.")
	buf.seek(self.bytes_read)
	data = buf.readline()
	while not data.endswith(SYM_CRLF):
	# there's more data in the socket that we need
	await self._read_from_socket()
	buf.seek(self.bytes_read)
	data = buf.readline()
	self.bytes_read += len(data)
	# purge the buffer when we've consumed it all so it doesn't
	# grow forever
	if self.bytes_read == self.bytes_written:
	self.purge()
	return data[:-2]

[sevdog]

Trying to go deeper in this issue it seems to happens also with hiredis parser. The problem is somehow related to the changes between 4.3.5 and 4.4.0.

There are some situation in which calling ZREMRANGEBYSCORE results in a null-response:

>           retval = self.response_callbacks[command_name](response, **options)
E           TypeError: int() argument must be a string, a bytes-like object or a real number, not 'NoneType'

command_name = 'ZREMRANGEBYSCORE'
connection = Connection<host=localhost,port=6379,db=0>
options    = {}
response   = None
self       = Redis<ConnectionPool<Connection<host=localhost,port=6379,db=0>>>

Inspection with a debugger this is happens when HiredisParser.read_from_socket method finds b'$-1\r\n' as buffer. Unfortunately I was not able to break-into the problem also with the PythonParser

[dvora-h]

I think #2695 should fix this bug