[FIXED] Preserve max delivered messages with Interest retention

[MauriceVanVeen]

Resolves #6538

If a consumer reached max deliveries for a message, it should preserve the redelivered state and allow inspecting its content. However, if a new consumer would be created and consume this message as well, it would still be removed under Interest retention.

This PR fixes that by using the redelivered state to keep marking there's interest.

Only downside is that the redelivered state gets cleaned up after a restart (this PR does not change/fix that). So if the consumer that had a max delivery message keeps acknowledging messages and its acknowledgement floor moves up, it would clean up the redelivered state below this ack floor.

Honestly I feel like keeping messages around if max delivery is reached makes the code very complex. It would be a lot cleaner if we'd only have the acknowledgement floor, starting sequence, and pending messages in-between, not also redelivered state that can be below ack floor. It's not something we can change now I suppose, but I'd be in favor of having messages automatically be removed once max delivery is reached and all consumers have consumed the message. DLQ-style behavior would then be more explicitly (and reliably) handled by the client, for example by publishing into another stream and then TERM the message, instead of relying on advisories that could be missed.

Signed-off-by: Maurice van Veen github@mauricevanveen.com

[derekcollison]

LGTM

[derekcollison]

Do you have a test that shows it gets reset on restart?

[MauriceVanVeen]

The reset on restart happens due to func (o *consumer) checkRedelivered(slseq uint64), specifically due to the first part in this condition: if sseq <= o.asflr || (lseq > 0 && sseq > lseq).
Have not added a test for this, as it already existed prior to this PR.

[derekcollison]

If you have a branch with the test I can take a look.

[MauriceVanVeen]

Don't think there's a test currently specific for checkRedelivered (or at least not immediately obvious which test would cover it).

Actually there are multiple issues with checkRedelivered removing redelivery data:

• the issue described in this issue, where if you restart (or a leader changes) redelivery data gets removed below the ack floor
• if we become consumer leader, but are behind on stream data, slseq (based on stream's last seq) would likely also revert state back
• redelivered state only gets cleaned up on an ACK or for a call to RemoveMsg, but if the underlying store gets purged it may also skip cleaning things up. And it could lead to memory leaks if this isn't cleaned up.

Will write tests for these issues soon to concretely reproduce, and share when available.