(2.14) Stream backup/restore v2

[neilalexander]

Signed-off-by: Neil Twigg neil@nats.io

[neilalexander]

@codex review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 52e0934729

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

[neilalexander]

@codex Please re-review and amend your earlier reviews as needed.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 61fbafc9d2

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

[neilalexander]

@claude review

[synadia-claude-reviewer]

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

---

Claude's Review

Tasks:

• Gather context
• Read changed files
• Analyze code
• Post review

---

Overview

This PR introduces a v2 stream backup/restore format using S2-compressed tar archives, replacing the v1 format for new snapshots while maintaining backwards compatibility for restore. The implementation is generally clean and well-structured. Below are the findings, from most to least critical.

---

P1: Dead warning path after streamSnapshot returns

server/jetstream_api.go, post-streamSnapshot block (~line 4188)

s.streamSnapshot(acc, mset, sr, &req)
// ...
if err := <-sr.errCh; err != _EMPTY_ {
    s.Warnf("Snapshot for stream '%s > %s' failed after %v: %s", ...)
} else {
    s.Noticef("Completed snapshot ...")
}

streamSnapshot itself reads from sr.errCh in its inner select loop:

case err := <-sr.errCh:
    hdr = []byte(fmt.Sprintf("NATS/1.0 500 %s\r\n\r\n", err))
    goto done

errCh is a buffered channel of size 1. When the snapshotting goroutine writes an error and then closes the channel, streamSnapshot consumes the error value. After streamSnapshot returns, the second <-sr.errCh reads "" from the already-closed channel — so the Warnf branch never fires. The success Noticef will always be logged, even when a streaming error occurred. The intent is correct but the mechanism is broken.

Fix this →

---

P1: Unbounded allocation in parseSnapshotMessagePreamble

server/stream_backup.go:426

subj := make([]byte, slen)
if _, err := io.ReadFull(r, subj); err != nil {

slen is read from the archive preamble without any upper bound check. A malicious or corrupted archive could advertise an enormous slen (e.g. 2^30), causing a multi-GB allocation before io.ReadFull fails. A simple guard would prevent this:

if slen > maxSubjectLength { // e.g. 4096 or nats.MaxPayload
    return "", 0, fmt.Errorf("subject length %d exceeds maximum", slen)
}

Similarly, hlen should be capped at a reasonable value (e.g. max header size). This matters because RestoreStreamV2 is reachable from the network-facing restore API.

Fix this →

---

P2: bufio.NewReader allocated per message

server/stream_backup.go:367

br := bufio.NewReader(tr)
subj, hlen, err := parseSnapshotMessagePreamble(br)

A new bufio.Reader is created for each message in the restore loop. For streams with millions of messages this allocates millions of short-lived objects. Consider allocating one outside the loop and calling br.Reset(tr) before each use.

---

P2: TODO for TTL handling is incomplete

server/stream_backup.go:392

// TODO(nat): check TTL and discard new headers
ttl, err := getMessageTTL(mhdr)

The comment implies messages whose TTL has already expired should be skipped or have their TTL stripped on restore, but this isn't implemented. Restoring an archive with short-TTL messages could cause them to be immediately expired (or even show negative TTLs). This should either be addressed or filed as a follow-up issue.

---

P3: nstate.Consumers inconsistency for non-clustered, with-consumers path

server/stream_backup.go:116-127

For non-clustered snapshots with consumers, streamState.Consumers is left as state.Consumers (from FastState) while the actual entries written are from store.Consumers(). These two sources are not under the same lock, so a consumer added or removed between FastState and the iteration would cause state.json to report a consumer count that doesn't match the tar entries. On restore this would cause a mismatch between nstate.Consumers and the actual entries. This is an inherent TOCTOU race, but it could be tightened by computing the count from the iterated slice, consistent with the clustered path:

} else {
    consumers := slices.Collect(store.Consumers())
    streamState.Consumers = len(consumers)
    // then range consumers instead of store.Consumers() below
}

---

Addressed from previous reviews

• ✅ Codex P0 (subject separator in preamble): parseSnapshotMessagePreamble correctly reads the space separator explicitly, then the subject bytes, then \r\n. Tests confirm empty and non-empty subjects work.
• ✅ Codex P1 (validate hlen before slicing): bounds check at server/stream_backup.go:376 (if hlen > len(buf)).
• ✅ Codex P1 (consumer count in state.json): streamState is a copy modified before marshaling; streamState.Consumers = 0 is applied before json.Marshal(streamState).
• ✅ Codex P1 (re-review) (use stream account for clustered consumer lookups): now uses sa.Client.serviceAccount() instead of globalAccountName.
• ✅ Codex P1 (re-review) (limit enforcement on restore): checkAllLimits is called incrementally in both the consumer and message loops.
• ✅ wallyqs (cmu.RLock vs cmu.Lock): fileStore.Consumers() correctly uses cmu.RLock().
• ℹ️ Codex P2 (no rollback on failure): acknowledged as intentional by @neilalexander — partial restore is preferable to no restore for corrupted archives.

---

Minor notes

• The version detection in processStreamRestore (sniffing the first tar header name to distinguish v1 vs v2) is pragmatic and clever, though it means two seeks and two S2 decompress initialisations per restore. The performance impact is negligible for restore.
• ConsumerConfig.clone() is a useful addition with correct deep copies of all slice/map fields.
• The memStore consumer tracking change from int counter to []ConsumerStore slice is a good prerequisite for Consumers() but slightly changes RemoveConsumer from O(1) to O(n). For typical consumer counts this is fine.