Why is subscribe() failing to complete?

[debbyglance]

This is a problem we are seeing on our production deployment only. We have not yet been able to reproduce in a staging environment.

We are using nats.py 2.6.0 in a quart application. Each quart worker process has a single nats client connection that is used for all subscriptions.

What we observe is:

1. For whatever reason, the nats connection reconnects. Logs show that the "disconnected" and "reconnected" callbacks are called. (As an aside, this tends to happen for only 1 of 2 worker processes in a container. Not sure why one would reconnect and not the other)
2. At some point after that (we don't know yet if it is immediately or some time later), calls to await nc.subscribe() never return. subscribe() seems to be blocked somehow

When we try to recreate this in our staging environment, the nats client connection works just fine after a reconnect.

Any ideas what might cause subscribe() to block indefinitely?

[caspervonb]

Will look into this.

[debbyglance]

Thank you.

Also I see there is a DisconnectedErrCb documented here: https://docs.nats.io/using-nats/developer/connecting/reconnect
But I don't see that as a parameter to connect() in nats.py https://github.com/nats-io/nats.py/blob/main/nats/aio/client.py. Is that not yet supported in nats.py ?

[debbyglance]

Is this possible?

1. NATS connection drops and reconnects, but something is wrong with the new connection
2. Now, calls to subscribe() result in a subscribe command getting queued
3. Eventually the _flush_queue fills up and now subscribe() hangs in _flush_queue.put() ("Put an item into the queue. If the queue is full, wait until a free slot is available before adding item")

In other words, it is behaving as if nothing is processing the commands in the flush queue.

If self._pending_data_size > self._max_pending_size, force_flush is True and this would ultimately trigger a FlushTimeoutError to be reported in _flush_pending.

    async def _send_command(self, cmd: bytes, priority: bool = False) -> None:
        if priority:
            self._pending.insert(0, cmd)
        else:
            self._pending.append(cmd)
        self._pending_data_size += len(cmd)
        if self._max_pending_size > 0 and self._pending_data_size > self._max_pending_size:
            # Only flush force timeout on publish
            await self._flush_pending(force_flush=True)

But based on these defaults:
DEFAULT_PENDING_SIZE = 2 * 1024 * 1024
DEFAULT_MAX_FLUSHER_QUEUE_SIZE = 1024
With only small subscribe() commands getting queued, the service is going to hit the queue size max before the pending size max, and "subscribe()" will just never complete

[debbyglance]

@caspervonb I believe the _flusher_task is stopping, and that is why the _flush_queue is filling up.

I added some logging to the service and can see that ._flusher_task.done() returns True once the service is in this state.

The service reports unhealthy and when it is terminated the logs show:

Task exception was never retrieved
future: <Task finished name='Task-11344' coro=<Client._flusher() done, defined at /usr/local/lib/python3.10/site-packages/nats/aio/client.py:2029> exception=InvalidStateError('invalid state')>
Traceback (most recent call last):
[  File "/usr/local/lib/python3.10/site-packages/nats/aio/client.py", line 2056, in _flusher
    future.set_result(None)

InvalidStateError would be raised if the future were already done, but I don't see how that could be the case.

Using nats-py 2.6.0

[debbyglance]

Fixed: #636