If the Consumer had a name containing `%`, it could result in
reply subject failing to format with `fmt.Sprintf`, as the `%`
was not properly escaped with `%%`.
Signed-off-by: Tomasz Pietrek <tomasz@nats.io>
1. Do not process an ack if we are closed.
2. When checking for needing an ack for a given consumer, hold lock entire time.
3. During recovery and restarts we check if we need to replay acks to the parent stream.
Signed-off-by: Derek Collison <derek@nats.io>
Server did check for timeouts in `processWaiting`,
but that needs to be also checked in `nextWaiting` in case of
tight timings, as `nextWaiting` can remove Pull Request based on
timeouts too.
Signed-off-by: Tomasz Pietrek <tomasz@nats.io>
Optimized and fixed a bug in filestore filteredPending().
Optimized memstore FilteredState().
Added comprehensive tests for NumPending() and FilteredState().
Signed-off-by: Derek Collison <derek@nats.io>
We noticed this was being called alot in user environments.
When the consumer was filtered with a wilcard and the stream had a high cardinality of subjects and was falling behind this could take a substantial amount of time.
Signed-off-by: Derek Collison <derek@nats.io>
We were snappshotting more then needed, so double check that we should be doing this at the stream and consumer level.
At the raft level, we should have always been compacting the WAL to last+1, so made that consistent. Also fixed bug that would not skip last if more items behind the snapshot.
Signed-off-by: Derek Collison <derek@nats.io>
This was ill-advised by me, not understanding that the messages stream for MQTT was interested policy based.
Interest policy based streams require consumers to match the replica count.
Signed-off-by: Derek Collison <derek@nats.io>
When a stream had a large number of consumers on a server that were sparse, the signaling mechanism would do a linear scan to signal matching consumers. As usage patterns have continued to have more consumers that are filteres and sparse, meaning a message is destined for a single or small number of consumers.
This change moves selection to a sublist that tracks only active consumer leaders for selection, which optimizes selection of consumers to signal when the number of consumers is large.
Signed-off-by: Derek Collison <derek@nats.io>
A stream with a WorkQueue retention policy is supposed to allow
more than one consumer if they user filtered subjects, but those
subjects should not overlap.
There was an issue that if a new consumer had a filter subject
"wider" than an existing one, the error was not detected and
the new consumer was incorrectly accepted.
Resolves#3639
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
When we deleted a consumer from an interest policy stream we would make sure to clean up any unacked messages.
However we only based start from the ack floor for the consumer and did not take into account the first sequence of the stream.
Signed-off-by: Derek Collison <derek@nats.io>
