(2.14) [IMPROVED] Durable source/mirror consumer soft reset

[MauriceVanVeen]

This PR contains two improvements to how durable consumers are used for mirroring/sourcing.

On consumer reset, pending is only recalculated when the starting sequence actually changes or there were pending messages. Repeated resets back to the ack floor (or any other sequence) with no delivery in between are therefore idempotent and recalculate pending only once.

For durable mirror/source consumers the delivery subscription (and its processing goroutine) is created up front and kept alive across resets. An idle heartbeat can now trigger a reset, so after a disruption (e.g. a server restart) recovery is near-immediate instead of waiting on the retry loop's backoff.

[MauriceVanVeen]

@claude, review.

[synadia-claude-reviewer]

Claude finished @MauriceVanVeen's task in 7m 36s —— View job

---

PR Review

• Gather context
• Review server/stream.go changes
• Review server/consumer.go changes
• Review server/jetstream_cluster.go changes
• Review test changes
• Post findings

Overall: Sound design, good test coverage. One behavior change to verify (consumer.go store-nil path), a dseq=0 window note, a transient wg count note, a failSetup teardown note, and a "for for" typo in two comment strings. No blocking issues.

See inline comments and the review summary comment for details.

[synadia-claude-reviewer]

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

---

PR Review: (2.14) [IMPROVED] Durable source/mirror consumer soft reset

Tasks:

• Load tools and post initial comment
• Read CLAUDE.md for repo guidelines
• Read and analyze changed files
• Provide comprehensive code review

---

Summary

This PR delivers two well-scoped improvements: a smarter resetLocalStartingSeq (avoids redundant streamNumPending recalculation), and a soft-reset model for durable mirror/source consumers that keeps the delivery pipeline alive across resets and enables near-immediate heartbeat-triggered recovery. The design is generally sound and the tests are meaningful. Below are the findings, in rough order of significance.

---

Findings

1. consumer.go — Behavioral change when o.store == nil

Before:

if o.store != nil {
    o.store.Reset(seq - 1)
    if o.retention == InterestPolicy { ... }  // inside store guard
    o.streamNumPending()
    o.signalNewMessages()
    return seq, true, nil
}
return seq, false, nil  // no-op if no store

After:

if o.store != nil {
    o.store.Reset(seq - 1)
}
if o.retention == InterestPolicy { ... }  // now outside store guard
if recalcPending {
    o.streamNumPending()
}
o.signalNewMessages()
return seq, true, nil  // always returns true now

Two subtle changes:

• o.signalNewMessages() now executes even when o.store == nil. Previously these consumers returned false (no reset performed). This is likely a bug fix, but it's worth confirming this is intentional for store-less consumers.
• The InterestPolicy check (which calls mset.state() and checkStateForInterestStream) now also runs without a store guard. Both of those functions work without a store, but it's a reachability change worth documenting.

2. stream.go — New sip guard in trySetupSourceConsumer

si := mset.sources[iname]
if si == nil || si.sip {
    return
}

This is new behavior (previously only si == nil was guarded). For mirrors the equivalent guard is in setupMirrorConsumer — the source path now adds it in trySetupSourceConsumer instead. This is correct (prevents concurrent setups) but slightly asymmetric from the mirror path. A short comment to explain why would help: e.g., // If setup already in progress, skip — the goroutine's defer will reschedule on failure.

When trySetupSourceConsumer returns early due to sip, the in-progress goroutine's defer still reschedules on failure. So no setup can get permanently lost. ✓

3. stream.go — Stale messages in the durable queue after soft reset

For durable mirrors/sources the IPQueue is kept alive across resets. After mirror.dseq = 0 is set and the reset request is sent, messages already enqueued from the old delivery epoch may still be in mirror.msgs. The sequence check handles most of these:

if sseq <= mset.mirror.sseq {
    // Ignore older messages. If dseq matches, update accounting.
    if dseq == mset.mirror.dseq+1 {
        mset.mirror.dseq++
    }
    ...
    return true
}

Edge case: if a stale in-queue message happens to have dseq == 1 and sseq <= mirror.sseq, mirror.dseq gets incremented to 1. The first fresh message from the reset (also dseq == 1) then fails the dseq == mset.mirror.dseq+1 check and triggers a retry. In practice this window is narrow (the queue drains fast), and this is a pre-existing characteristic of the durable consumer model, not introduced by this PR. Worth keeping in mind for future debugging.

4. stream.go — Heartbeat suppression while reset is in progress

// processInboundMirrorMsg
if ldseq := parseInt64(sliceHeader(JSLastConsumerSeq, m.hdr)); ldseq > 0 && uint64(ldseq) != mset.mirror.dseq {
    needsRetry = !mset.mirror.sip

Suppressing retries when sip is set prevents cascading resets during a pending setup — correct. However, if the setup goroutine's response arrives after sip is cleared (goroutine defer), but before mirror.dseq is updated by any incoming message, the next heartbeat would trigger another retry. That's also correct (one more retry rather than a cascade), just worth being aware of.

5. stream.go — srcDurableConsumerWaitTime (5 s) may be tight under load

var srcDurableConsumerWaitTime = 5 * time.Second

Five seconds is reasonable for a reset-only request vs. 30 s for a full consumer creation. Package-level, so it can be overridden in tests. One concern: under a heavily loaded cluster the reset RPC might take longer than 5 s, causing a spurious retry and re-sending the reset request. Not a correctness issue (the retry path is safe), but it can produce unexpected churn during a cluster-wide restart.

6. createMirrorPipeline — stale comment

// Create a new queue each time
mirror.msgs = newIPQueue[*inMsg](mset.srv, qname)

For durable mirrors createMirrorPipeline is only called once (when mirror.sub == nil). The comment "Create a new queue each time" was accurate in the old ephemeral-only path but is now misleading. Minor nit.

7. Test coverage — no heartbeat-triggered recovery integration test

The four new tests cover:

• Pipeline stability across repeated soft resets (subscription/goroutine pointer identity).
• Teardown + rebuild (force-cancel then retry).
• Message flow integrity post-reset.

What's not covered: an end-to-end scenario where the source server restarts, heartbeat-detect the gap, and recovery completes faster than the retry-loop backoff. The PR description calls this out as the key user-visible improvement. A test that bounces the server and measures time-to-first-message-after-restart would give confidence in the primary motivation. Fix this →

---

Positive notes

• force parameter threading is clean and complete — every call site is correctly updated, and the semantics (lifecycle vs. recovery) are clear.
• createMirrorPipeline / createMirrorRoutine / createSourcePipeline are good extractions that eliminate code duplication between initial setup and soft-reset paths.
• ready.Wait() is correctly omitted for the durable goroutine (held lock would deadlock; comment explains why).
• mirror.cname is correctly maintained — set from mset.cfg.Mirror.Consumer.Name at the top of setupMirrorConsumer on every call, so isCurrentSub works correctly across resets.
• recalcPending optimization in resetLocalStartingSeq is correct: o.sseq != seq || len(o.pending) > 0 captures both the sequence-changed case and the "stale pending entries" case.
• jetstream_cluster.go non-leader path (!o.isLeader()) correctly skips streamNumPending regardless of recalcPending — the leader handles it.

---

Overall

Well-structured change with good test coverage for the core invariants. The consumer.go store-guard change (#1) and the stale-queue edge case (#3) are the two items I'd most want a second set of eyes on before merge. The rest are nits or follow-up considerations.

[synadia-claude-reviewer]

Review Summary

The two changes here are well-motivated and cleanly implemented.

Idempotent pending recalculation (consumer.go, jetstream_cluster.go) — The resetLocalStartingSeq optimisation is straightforward and correct. The check o.sseq != seq || len(o.pending) > 0 accurately identifies when a recalculation is genuinely needed and the signalNewMessages() call is now unconditional so delivery is always re-triggered.

One thing to verify: the refactor moved checkStateForInterestStream and signalNewMessages outside the if o.store != nil block (see inline), changing the return value from false to true for store-less consumers and calling signalNewMessages where it previously was not. In practice JetStream consumers always have a store so this is likely a dead code path, but worth a quick audit that no consumer is legitimately created without a store at this call site.

Durable sub kept alive across resets (stream.go) — The design is sound. Key correctness properties hold:

• sip=true suppresses heartbeat-triggered retries and makes setupMirrorConsumer a no-op while a reset is in flight, preventing concurrent setups.
• Pre-reset messages queued during the reset window are correctly discarded via gap detection (dseq=0 means anything arriving with dseq>1 breaks the processing loop), and the reset consumer re-delivers from dseq=1.
• cancelSourceInfo / cancelMirrorConsumer still work on the durable path when a hard cancel is needed (leadership loss, forced stall retry).
• The wg.Wait() skip for durables is safe because the old and new goroutines use different msgs queues (the old one is drained by cancelSourceInfo).

See inline comments for: a dseq=0 window note, a wg transient count note, a failSetup teardown note, and a trivial "for for" typo in two places.

No blocking issues. The test coverage (four new tests covering pre-existing sub, teardown+rebuild, and the source-side analogues) is a solid addition.

[roeschter]

Downloads.zip
Attached logs from customer POC

1. The durable consumer case is no covered. Mirror keeps subscription (recreates it on retry)
2. For the old style auto create consumer this does not work (yet). Why not use the same mechanism? Is it really required to generate a new delivery subject? What about re-using it and always listening as with the durable consumers case?

[roeschter]

Otherwise this tests fine. (for the durable case)
I think we need that included soon.

[MauriceVanVeen]

2. For the old style auto create consumer this does not work (yet). Why not use the same mechanism? Is it really required to generate a new delivery subject? What about re-using it and always listening as with the durable consumers case?

Having a stable delivery subject is one, but the fundamental issue is that the "old style" ephemeral consumers have an InactiveThreshold, which means they get cleaned up after 10 seconds. A stable delivery subject doesn't help if the consumer is cleaned up.

Only a durable consumer keeps the same delivery subject and isn't removed, so there this optimization can be used. I don't think the "old style" ephemeral case can be improved much, and even if we would some edge cases would remain, whereas the "new style" durable consumers don't have any of these issues.

[MauriceVanVeen]

I'll move this PR to ready for review, since it's confirmed to work as intended for the new durable sourcing consumers.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 11e522fdd4

ℹ️ 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".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 15c1e9cf21

ℹ️ 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 4m 30s —— View job

---

Brief Re-review: (2.14) [IMPROVED] Durable source/mirror consumer soft reset

Tasks:

• Load tools
• Review recent commits since last review (15c1e9c, 28c667f, ce0d819)
• Check Codex P2 findings (AckPolicy gate concern)
• Review heartbeat-triggered reset implementation
• Post findings

---

Since the last review, three meaningful commits landed. Here's what changed and how it looks.

Commit 15c1e9c — Durable mirror/source tear down on AckFlowControl mismatch

This directly addresses the Codex P2 concern. Both mirror and source paths now call cancelSourceInfo when the reset response comes back with AckPolicy != AckFlowControl:

• Mirror: mset.cancelSourceInfo(mirror) at line 3797
• Source: mset.cancelSourceInfo(si) at line 4283

After cancellation, sub = nil. The defer increments fails. On the next retry, si.sub == nil && si.fails > 0, so the optimistic pre-creation path is skipped — the pipeline is only rebuilt after a validated successful response. The Codex P2 concern is fully addressed.

One note on sequencing: cancelSourceInfo is called under the stream lock (the goroutine already holds it at this point), then retry = true is set, lock released, and goroutine returns. The defer then runs scheduleSetupMirrorConsumerRetry(). This is correct — the retry chain is preserved.

Commits 28c667f + ce0d819 — Heartbeat-triggered durable consumer resets

These implement the core motivation of this PR: near-immediate recovery via heartbeat. The change is clean:

// Before:
needsRetry = true
mset.retrySourceConsumerAtSeq(si.iname, si.sseq+1)

// After:
needsRetry = !si.sip
...
if needsRetry {
    mset.retrySourceConsumerAtSeq(si.iname, si.sseq+1, false)  // soft reset
}

Two things working together here:

1. !si.sip gate: if a reset is already in flight (sip=true), the heartbeat doesn't trigger another one. No cascading resets. ✓
2. force=false: the retry is a soft reset — existing subscription is kept alive. ✓

The stall detection path (processAllSourceMsgs stall ticker) still uses force=true, which is correct — a genuine stall warrants a hard retry.

Previously raised items — current status

Item	Status
consumer.go store-nil behavioral change	No change — o.store is always non-nil for JetStream consumers; safe dead code
cname maintenance for durables	✓ — mirrors: line 3537, sources: line 4058, set from config on every call
"for for" typo in comments	✓ Fixed — comments now read "for durable consumers" (single "for")
Heartbeat-triggered recovery test coverage	Still absent — but the manual test by @roeschter validates the key scenario for the durable case

Overall

The new commits are clean and address the open items from the previous review. No blocking issues. The core invariants hold: sip prevents concurrent setups, heartbeat-detected gaps trigger near-immediate soft resets, and misconfigured durable consumers (wrong AckPolicy) are now properly torn down instead of being left in a partially-active state.