Repository navigation
fix(pubsub/v2): acquire bounded limits in flow controller - #12590
Conversation
| f.recordOutstandingBytes(ctx, outstandingBytes) | ||
| f.semSize.Release(f.bound(size)) | ||
| if f.limitBehavior != FlowControlIgnore { | ||
| f.semSize.Release(f.bound(size)) |
There was a problem hiding this comment.
A semaphore requires that we release the exact same number of permits that we acquired.
The acquire function has different logic for different LimitExceededBehavior values:
- In FlowControlBlock mode, it acquires f.bound(size) from the semaphore.
- In FlowControlSignalError mode, it attempts to acquire the raw int64(size).
The release function always releases f.bound(size). It incorrectly assumes that the amount acquired was also f.bound(size). Could this manifest into a bug?
There was a problem hiding this comment.
Good catch, this has actually been out for a while. Most likely, it's because the bounding behavior doesn't trigger often (unless users set flow control bytes to very low).
There was a problem hiding this comment.
So actually I remember why this was intentional.
In FlowControlSignalError mode, the acquire never goes through since the flow controller throws an error if there is not enough space for the message. In such a case, we bubble up the error ErrFlowControllerMaxOutstandingBytes. We shouldn't bound the message size here, since we consider "too large messages" an error.
This reverts commit e816d1e5e9aucd9b07665ce5fe80ecf7a8c3b950.
This change makes flow controller continue to track and record outstanding messages/bytes even when flow control mechanisms are disabled.
Fixes #12447