reader: max_in_flight check in _send_rdy() doesn't take in-flight msgs into account

[alpaker]

Reader._send_rdy() contains logic to avoid setting a RDY count on a connnection that would allow exceeding the reader's max_in_flight. The code in question is:

if (self.total_rdy + value) > self.max_in_flight
    if not conn.rdy:
        # if we're going from RDY 0 to non-0 and we couldn't because
        # of the configured max in flight, try again
        rdy_retry_callback = functools.partial(self._rdy_retry, conn, value)
        conn.rdy_timeout = self.io_loop.add_timeout(time.time() + 5, rdy_retry_callback)
    return

The condition (self.total_rdy + value) > self.max_in_flight doesn't take into account messages that are currently in flight. This creates an opportunity for max_in_flight violations. The recipe in this report will produce one, usually within a minute.

I reported this offline to @jonmorehouse, who pointed out that it was surprising the issue hasn't surfaced before. I think the reason is that it can only hit when max_in_flight is less than the connection count, and this use case must be rare in practice:

• _send_rdy() is only called with a value parameter of 0, 1 or _connection_max_in_flight().
• So if len(conns) * _connection_max_in_flight() <= max_in_flight, then we'll never violate the in-flight constraint, because we're implicitly restricting ourselves to a per-connection limit that guarantees safety.
• But if len(conns) > max_in_flight and _connection_max_in_flight() == 1, then respecting the per-connection limit isn't enough. In this case deciding whether a non-zero RDY count is permissible on a connection requires taking into account the in-flight count on other connections.

[alpaker]

The same issue affects the Go client.

An appropriate fix for the Python client would be:

diff --git a/nsq/reader.py b/nsq/reader.py
index ad6447a..4ad1281 100644
--- a/nsq/reader.py
+++ b/nsq/reader.py
@@ -452,7 +452,10 @@ class Reader(Client):
         if value > conn.max_rdy_count:
             value = conn.max_rdy_count

-        if (self.total_rdy + value) > self.max_in_flight:
+        total_possible_other = sum(
+            max(c.last_rdy, c.in_flight) for c in self.conns.values() if c is not conn
+        )
+        if (total_possible_other + value) > self.max_in_flight:
             if not conn.rdy:
                 # if we're going from RDY 0 to non-0 and we couldn't because
                 # of the configured max in flight, try again

And analagously for the Go client.

[mreiferson]

@alpaker thanks for another detailed bug report 💯

I would expect total_rdy to represent exactly what your proposed diff is doing, but then I found https://github.com/nsqio/pynsq/blob/master/nsq/reader.py#L320, which appears to be something we missed when updating pynsq for nsqio/nsq#404.

I'm curious if removing that line addresses the issue and doesn't break any other tests?

[alpaker]

@mreiferson For consistency you'll also want to remove https://github.com/nsqio/pynsq/blob/master/nsq/async.py#L489.

I don't think the change you suggest is enough on its own. I was a little cryptic with the above diff and didn't describe the way this problem interacts with the RDY redistribution logic. There are two interlocking issues here. One is what you mention: the semantics of RDY count are currently different on the server and client. With the change you suggest they're brought into agreement and so no need to use last_rdy as in my diff.

The second issue is the need for redistribution of RDY count when max-in-flight is less than the connection count. With the change you suggest, this https://github.com/nsqio/pynsq/blob/master/nsq/reader.py#L664 becomes the only route outside of backoff by which total_rdy can decrease. So you're going to havemax_in_flight == total_rdy as long as connections with RDY greater than 0 are receiving messages more frequently than low_rdy_idle_timeout. This means that unless and until the currently active connections are drained and hit the timeout, both paths to redistribution are blocked:

1. This quantity https://github.com/nsqio/pynsq/blob/master/nsq/reader.py#L671 is going be 0.
2. A connection selected in this block for update https://github.com/nsqio/pynsq/blob/master/nsq/reader.py#L323 will be blocked in _send_rdy() and scheduled for retry (https://github.com/nsqio/pynsq/blob/master/nsq/reader.py#L460), but retries won't succeed until something decrements total_rdy.

If you do want to guarantee progress on all connections then you need to redistribute RDY away from active connections more aggressively, which means setting RDY == 0 on some connections even though they're still receiving messages. But in that case you need to be careful not to send RDY 1 on connections that are gaining RDY while the connections that are losing RDY still have messages in flight, or else the resulting race condition opens you up to a max-in-flight violation. Hence you still need to be looking at the sum of max(conn.rdy, conn.in_flight) over all connections, not just the sum of conn.rdy, or in some other way take into account not just the RDY count of a connection but also its in-flight count.

[mreiferson]

For consistency you'll also want to remove https://github.com/nsqio/pynsq/blob/master/nsq/async.py#L489.

Yep.

A connection selected in this block for update https://github.com/nsqio/pynsq/blob/master/nsq/reader.py#L323 will be blocked in _send_rdy() and scheduled for retry (https://github.com/nsqio/pynsq/blob/master/nsq/reader.py#L460), but retries won't succeed until something decrements total_rdy.

To your point, if we fixed the inconsistency w/r/t maintaining total_rdy, that block also needs to be updated. First, it no longer makes sense to perform that logic in that spot and should probably be relocated to the redistribute method. Second, it's probably more easily implemented as a simple clause here.

But in that case you need to be careful not to send RDY 1 on connections that are gaining RDY while the connections that are losing RDY still have messages in flight, or else the resulting race condition opens you up to a max-in-flight violation.

It isn't going to be possible to reasonably prevent this race condition entirely, nor do I think it's worth it. The changes we're talking about here should improve behavior without any tradeoffs, so I'm 👍.

[mreiferson]

Fixed in #179