nsq: fix requeue accounting #805
Conversation
|
I think this is correct! |
|
My point is correct or the original? |
|
Yes, I agree. Thoughts @jehiah |
|
👍 this means REQ counts are incremented immediately instead of when it's re-added to the queue for processing. That probably makes more intuitive sense anyway. |
| @@ -275,6 +275,7 @@ func (t *Topic) messagePump() { | |||
| chanMsg.deferred = msg.deferred | |||
| } | |||
| if chanMsg.deferred != 0 { | |||
| atomic.AddUint64(&channel.messageCount, 1) | |||
There was a problem hiding this comment.
I don't think we want this change, though, because when the deferred message is finally published this will be incremented.
There was a problem hiding this comment.
But when the defered message finally published the messageCount do not increase.
if chanMsg.deferred != 0 {
atomic.AddUint64(&channel.messageCount, 1)
channel.StartDeferredTimeout(chanMsg, chanMsg.deferred)
continue
}
StartDeferredTimeout() will push the message into channel.deferredPQ and when the timeout message poped from channel.deferredPQ it will be published through doRequeue(). The messageCount hasn't changed in the process.
I reviewed the code so many times and verified it in practice
There was a problem hiding this comment.
@mreiferson @jehiah Am i wrong？
There was a problem hiding this comment.
Ahhh, I missed that this was an increment for the channel message count. Yes, it looks like you are correct. However, it doesn't make sense that the topic is now responsible for atomically incrementing this channel's counter, we should push this responsibility into the channel logic somewhere.
|
@sdbaiguanghe given that these are regressions, would you mind adding test cases for each of these scenarios? Thanks! |
|
see #832 |
No description provided.