[FIXED] Mirror last seq mismatch from skipMsgs failure & retry

[MauriceVanVeen]

• Make skipMsgs return errors from the underlying store / Raft Propose / ProposeMulti calls instead of swallowing them, and propagate failures.
• Fix a potential nil deref in processInboundMirrorMsg if mset.mirror was cleared while the lock was released.
• Always immediately retry the mirror consumer on a last seq mismatch so a mismatch can't leave the mirror stalled for longer than necessary.

[synadia-claude-reviewer]

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

---

PR Review

• Gather context (CLAUDE.md, PR diff, relevant files)
• Analyze changes in server/stream.go
• Analyze test changes in server/jetstream_cluster_3_test.go and server/jetstream_test.go
• Review for correctness, concurrency safety, and clustering concerns
• Post final review

---

Summary

The three fixes are well-motivated and correctly implemented. Analysis below.

---

Fix 1: skipMsgs error propagation (server/stream.go)

The change to return errors from skipMsgs is correct. All three code paths are covered:

• Non-clustered: store.SkipMsgs error propagated ✓
• Clustered (delete-range feature flag): node.Propose error propagated ✓
• Clustered (legacy ProposeMulti batching): node.ProposeMulti error propagated in both the mid-loop flush and the final send ✓

One note on the legacy batching path: if ProposeMulti fails mid-loop (after earlier batches of >10,000 entries already succeeded), earlier entries were already enqueued for proposal. If the node has lost leadership, those earlier batches won't commit, and the callers roll back sseq/dseq and retry from the current store state — which is the correct recovery path. The raft layer handles the in-flight proposals on leadership change.

The two call sites in processInboundMirrorMsg and setupMirrorConsumer both correctly roll back state on error before triggering a retry.

---

Fix 2: Nil deref in processInboundMirrorMsg (server/stream.go)

The original code re-acquired mset.mu and accessed mset.mirror.* fields without checking if mset.mirror was still non-nil. The lock had been released to call processJetStreamMsg, creating a window for mset.mirror to be cleared by a concurrent cancellation.

The new code adds if mset.mirror != nil before accessing the fields and calls retryMirrorConsumer() unconditionally. If mset.mirror is nil, retryMirrorConsumer → cancelMirrorConsumer is a no-op and setupMirrorConsumer returns early on mset.cfg.Mirror == nil. Correct.

---

Fix 3: Always retry on errLastSeqMismatch (server/stream.go)

Old behavior: when sseq <= lseq the code returned false without calling retryMirrorConsumer(), which could leave the mirror stalled until the sourceHealthCheckInterval timer (10s) fired.

New behavior: both branches now call retryMirrorConsumer(). The sseq <= lseq branch sets mirror.sseq = lseq before retrying; setupMirrorConsumer overwrites this with mset.lseq anyway (line 3448), and uses state.LastSeq + 1 for the consumer's OptStartSeq, so the new consumer starts from the right place.

The olag restoration is now uniform across both branches (previously only restored in the sseq <= lseq branch).

---

Issue: IPQueue not unregistered in failSetup closure

See inline comment on stream.go:3741. The closure sets mirror.msgs = nil which causes the deferred cancelSourceInfo to skip the drain()/unregister() call. The queue stays registered in s.ipQueues until the next setupMirrorConsumer creates a new one with the same name (overwriting the stale entry). Low severity — self-healing and no goroutine is reading the queue — but it's inconsistent with the existing error path (subscription creation failure) which correctly leaves mirror.msgs set for the deferred cleanup.

---

Tests

• TestJetStreamClusterMirrorSkipMsgsPropagatesProposeFailure: correctly steps the raft node down to force errNotLeader from skipMsgs and verifies the error is returned.
• TestJetStreamMirrorProcessInboundNilDeref: race between setting/nil-ing mset.mirror and calling processInboundMirrorMsg — directly reproduces the nil deref scenario.
• TestJetStreamMirrorRetriesOnSeqAlreadySeenMismatch: injects a sseq <= lseq mismatch and asserts that retryMirrorConsumer fires by watching the qch channel close. Clean and targeted.

[MauriceVanVeen]

@claude, re-review.

[synadia-claude-reviewer]

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

---

Re-review

• Read PR diff
• Analyze server/stream.go changes
• Analyze test changes
• Post final review

---

Fix 1 — skipMsgs error propagation (server/stream.go:3338)

All three code paths now correctly surface errors:

• Non-clustered (node == nil): store.SkipMsgs error propagated ✓
• Clustered with delete-range flag: node.Propose error propagated ✓
• Clustered legacy batching (ProposeMulti): error propagated from both the mid-loop flush and the final send ✓

One edge case in the legacy batching path: if the first batch of >10 000 entries was already pushed to n.prop (and potentially to the WAL) before a mid-loop ProposeMulti fails, skipMsgs returns an error but those earlier entries may have already been enqueued. Looking at ProposeMulti (raft.go:942), all failure returns (errNotLeader, werr, leader-overrun step-down) happen before entries are pushed, so in practice each individual batch is all-or-nothing at the push level. The real risk is a batch that was pushed to the prop queue and processed by the WAL writer before step-down. This edge case pre-exists this PR (old code swallowed the error, leaving the mirror in a worse undefined state); the retry the new code triggers is an improvement.

Both call sites correctly roll back sseq/dseq on error before retrying.

---

Fix 2 — Nil deref in processInboundMirrorMsg (server/stream.go:3292)

Correct. The lock is dropped to call processJetStreamMsg, creating a window for a concurrent cancelMirrorConsumer to nil mset.mirror. The new if mset.mirror != nil guard before accessing mirror.lag/dseq/sseq closes that window. retryMirrorConsumer is safe to call unconditionally — it checks mset.mirror == nil internally via cancelMirrorConsumer → cancelSourceInfo.

---

Fix 3 — Always retry on errLastSeqMismatch (server/stream.go:3287)

Old behavior had two separate branches:

• sseq <= lseq: restore lag, sseq = lseq, dseq, no retry → mirror could stall until the health-check timer fired (10s)
• sseq > lseq: restore dseq, sseq, no lag restore, retry

New behavior:

• Always restores lag, dseq, sseq (+ sets sseq = lseq in the <= sub-case)
• Always calls retryMirrorConsumer()

This is correct on both counts: it eliminates the stall and also fixes the missing olag restoration in the old sseq > lseq branch. Retry storms are not a concern — setupMirrorConsumer is already rate-limited by sourceConsumerRetryThreshold (stream.go:3460).

---

On the previous inline comment (IPQueue cleanup in failSetup)

The previous review flagged that failSetup sets mirror.msgs = nil without draining first. Looking at the current code, this description does not match the implementation. The failSetup closure (stream.go:3738) calls mset.cancelSourceInfo(mirror), which does call si.msgs.drain() + si.msgs.unregister() before returning. It does not set si.msgs = nil.

What actually happens:

1. failSetup → cancelSourceInfo(mirror): drains and unregisters the queue, nils si.sub, but leaves si.msgs non-nil (pointing to a drained/unregistered queue).
2. The goroutine's deferred cleanup sees retry && mirror.sub == nil → calls cancelSourceInfo(mset.mirror) again.
3. The second call does a redundant drain() + unregister().

Both drain() (ipqueue.go:253) and unregister() (ipqueue.go:281) are idempotent — drain just clears internal slices and unregister calls sync.Map.Delete which is a no-op on a missing key. So the double-drain is safe. There is no orphaned queue in s.ipQueues after failSetup.

The mild inconsistency is that cancelSourceInfo doesn't nil si.msgs after cleanup, leading to redundant re-drain on the next call. This is a pre-existing pattern and not introduced by this PR, so it's out of scope here.

---

Tests

• TestJetStreamClusterMirrorSkipMsgsPropagatesProposeFailure: Steps the mirror stream's Raft node down, then calls skipMsgs directly under the lock. Verifies errNotLeader is returned. Correctly targets the clustered Propose/ProposeMulti path.

• TestJetStreamMirrorProcessInboundNilDeref: Two goroutines race — one sets mset.mirror and calls processInboundMirrorMsg, the other sets mset.mirror = nil. The crafted ACK reply (sseq=5, mirror.sseq=4) forces the sequential path into processJetStreamMsg which returns errLastSeqMismatch, then re-acquires the lock to access mset.mirror.* — exactly the nil-deref window. Runs for 3 seconds; under -race would also catch data races. ✓

• TestJetStreamMirrorRetriesOnSeqAlreadySeenMismatch: Sets mirror.qch to a sentinel channel and constructs the same sseq<=lseq mismatch scenario. Asserts retryMirrorConsumer closes qch within 2 seconds. Clean and targeted. ✓

---

Verdict

All three fixes are correct. The previous inline comment described behavior that doesn't match the current implementation — the IPQueue is properly cleaned up via cancelSourceInfo. The PR is in good shape.

[neilalexander]

LGTM