In the recovery case, this will oversubscribe RDY count (it will send **at-least** RDY 1 per connection even if you set a max in_flight_count of 1). 

I'm ok deferring this specific fix because it will be close as-is and appears to be a pre-existing issue. 

The code managing that over-subscribed state with `self.total_ready` is deficient in two ways. 1) It doesn't account for multiple tasks (we've implemented in_flight_count as per-task), and 2) it is never incremented (oops!) when RDY is updated, so it can't be checked against for calculation. It appears that it only works at initial connection time now.

I'll describe these issues in a separate pull request.