(2.14) [FIXED] AckAll on WQ removed messages outside consumer filter

[MauriceVanVeen]

Normally AckAll on a WorkQueue isn't permitted, since AckExplicit or AckFlowControl is required. However, if you'd source from a WorkQueue starting from 2.14, this uses a consumer with AckFlowControl, which functions like AckAll. A filtered source consumer could then ack messages outside the consumer filter, resulting in those messages being removed. This path also is engaged for Interest streams, but isn't problematic there since messages without a consumer have no interest, so removing them is correct.

This PR fixes that by only calling mset.ackMsg on messages that were pending prior to the AckAll.

Resolves #8423

[synadia-claude-reviewer]

Claude finished @MauriceVanVeen's task in 5m 46s —— View job

---

Code Review

• Gather context and read PR diff
• Review server/consumer.go changes
• Review server/jetstream_cluster.go changes
• Review server/jetstream_cluster_3_test.go changes
• Post final review summary

---

The fix is correct. Both paths have been properly updated to restrict mset.ackMsg calls to only sequences that match the consumer's filter.

server/consumer.go (non-clustered path)

The pivot to collecting candidates via o.pending is elegant and sound. o.pending is populated exclusively with sequences actually delivered to this consumer — which by definition match its filter — so no extra subject matching is needed. The ackInPlace && seq != sseq guard correctly prevents sseq from being double-acked (it's always handled unconditionally after the switch). The removal of the now-unnecessary floor variable and the old FIXME comment about inefficiency is accurate, since the new cost is proportional to the number of actually-delivered messages rather than the full sequence gap.

One minor note: ackAllSeqs is populated from map iteration over o.pending, so its order is non-deterministic, but mset.ackMsg is order-independent (each call independently acquires mset.mu and checks noInterest), so this is fine.

server/jetstream_cluster.go (clustered path)

The store-based traversal for the filtered case is correct. The loop advances seq to the found nseq before the implicit seq++, so it steps to nseq+1 on the next iteration — no messages are re-examined or skipped. The err != nil || nseq >= sseq break is conservative: on a store error it avoids acking anything further in the range (rather than blindly removing messages), which is the safer behavior.

One invariant worth noting for future readers: the isFiltered && filters == nil condition and the subsequent o.subjf[0] access are safe together. filters is non-nil only when len(o.subjf) > 1, so filters == nil guarantees at most one element in o.subjf. Since isFiltered() returns false when o.subjf == nil, the combination isFiltered && filters == nil guarantees exactly one entry at index 0.

server/jetstream_cluster_3_test.go

Test coverage is thorough:

• TestJetStreamClusterWorkQueueSourceMustNotAckOtherSubjects covers the core bug: a sourced message at seq 5 whose AckAll must not purge x.other messages at seqs 2–4.
• TestJetStreamClusterScheduledMessagesWithWorkQueueSource covers the gap-with-purged-schedules case, which is the harder scenario since the ack range spans sequences that no longer exist in the store.

Both tests validate R1 and R3 paths, covering both the non-clustered (consumer.go) and clustered (jetstream_cluster.go) fixes.

No issues found.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: f15d723e76

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[MauriceVanVeen]

@claude, brief re-review.

[synadia-claude-reviewer]

Claude finished @MauriceVanVeen's task in 1m 38s —— View job

---

Re-review

• Read PR diff
• Review server/consumer.go changes
• Review server/jetstream_cluster.go changes (including Codex concern)
• Post findings

---

The fix looks correct. Here's a focused assessment of each path and the Codex concern.

server/consumer.go (non-clustered)

ackAllSeqs is collected from o.pending while holding the consumer lock, excluding sseq itself (the seq != sseq guard). The lock is then released, and mset.ackMsg is called for sseq followed by everything in ackAllSeqs. Since o.pending is populated only with sequences that match the consumer's filter by construction, no extra subject matching is needed. Clean.

server/jetstream_cluster.go (clustered)

The new code collects from state.Pending (store-backed, captured via BorrowState before UpdateAcks clears it), filtering for seq < sseq. Then mset.ackMsg(o, sseq) is called explicitly, followed by each collected seq. This is symmetric with the non-clustered path.

Re: Codex's filter-update concern

The concern was that the clustered path used current consumer filters rather than the pre-ack pending state. This does not apply to the current code. The PR does not use filters in the clustered path at all — it uses state.Pending, which is the set of sequences actually delivered to the consumer (and not yet acked), regardless of what the filter is now or was. This is exactly the right set. The concern would have been valid against a filter-based implementation, but this PR chose the better approach (pending-based) for both paths.

One invariant worth noting: state.Pending is captured before o.store.UpdateAcks(dseq, sseq) runs, so it reflects the correct pre-ack state. After UpdateAcks, the store clears those pending entries — hence the // Capture before updating store, which clears the pending below. comment in the diff, which is accurate.

No issues found. The two paths are now properly symmetric and correct.

[neilalexander]

Do we have a sense of how big this array can get?

[MauriceVanVeen]

At most MaxAckPending

[neilalexander]

Since the default is 1000, we could potentially preallocate on the stack, should reduce the number of reallocations and heap escapes:

var _ackAllSeqs [1000]uint64
ackAllSeqs := _ackAllSeqs[:0]

[MauriceVanVeen]

Discussed to instead use ackAllSeqs = make([]uint64, 0, min(uint64(len(o.pending)), sgap-1)). To only allocate for AckAll/AckFlowControl policy, and not for the normal paths.

[MauriceVanVeen]

@claude, brief re-review.

[synadia-claude-reviewer]

Claude finished @MauriceVanVeen's task in 1m 24s —— View job

---

Re-review

• Read PR diff
• Review server/consumer.go changes (including neilalexander/MauriceVanVeen allocation discussion)
• Review server/jetstream_cluster.go changes
• Post findings

---

The one material change since the prior re-review is the allocation strategy for ackAllSeqs, updated in response to neilalexander's comment.

server/consumer.go — allocation

var ackAllSeqs []uint64 is declared at function scope (nil slice) and then, inside the AckAll/AckFlowControl case, allocated with:

ackAllSeqs = make([]uint64, 0, min(uint64(len(o.pending)), sgap-1))

This is the right approach. The capacity is min(pending count, gap−1), which is a tight upper bound:

• You can't collect more than len(o.pending) entries from the pending map.
• You can't collect more than sgap−1 sequences (those strictly between o.asflr and sseq).

The sgap−1 underflow concern is safe: the sseq <= o.asflr early-return at line 3712 guarantees sgap ≥ 1, so sgap−1 ≥ 0. When sgap == 1, the capacity is min(len(pending), 0) = 0, which is correct — there's nothing to collect.

The allocation only happens on the AckAll/AckFlowControl path, so AckExplicit consumers pay zero cost. This is better than the fixed 1000-element stack array neilalexander suggested, which would allocate 8 KB on every processAckMsgLocked call regardless of policy.

server/jetstream_cluster.go — allocation

Same pattern:

ackAllSeqs = make([]uint64, 0, min(uint64(len(state.Pending)), sagap-1))

Here the sagap > 1 guard before the make ensures sagap−1 ≥ 1, so no underflow. The use of state.Pending (captured before UpdateAcks) is correct and matches what the non-clustered path does with o.pending.

Both paths are now symmetric in structure and allocation strategy.

No issues found.

[neilalexander]

LGTM

[vividcloudpark]

Hi, @MauriceVanVeen @neilalexander
Could you clarify whether #8431 is included in v2.14.6-RC.1 or planned to be
backported to the 2.14.x release line?

I initially assumed that #8431 was included because it is referenced in #8450.
However, #8450 appears to contain only #8430, #8427, and #8449.

Looking at the v2.14.6-RC.1 source, the AckAll handling also still appears to
acknowledge a contiguous sequence range rather than only the messages present in the
pending set. Therefore, it seems that the fix from #8431 is not included.

Could you please confirm:

1. Is (2.14) [FIXED] AckAll on WQ removed messages outside consumer filter #8431 absent from v2.14.6-RC.1?
2. If so, is a backport planned for v2.14.6 final or a later 2.14.x release?
3. Is there a recommended workaround for users who need this fix on the 2.14.x release
   line?

[neilalexander]

We will be picking in this PR into 2.14.6 now that #8528 is merged too.