backup: Phase 0b M5-2 - SQS side-record derivation (vis/byage/dedup) (#892)

## Summary

Phase 0b M5-2 implementation per the merged design doc
[`docs/design/2026_05_30_proposed_sqs_side_record_derivation.md`](https://github.com/bootjp/elastickv/blob/main/docs/design/2026_05_30_proposed_sqs_side_record_derivation.md)
(#885).

Resolves the SQS side-record decision gate the M5-1 encoder (#846)
explicitly deferred. Without this PR, restoring a backup silently
breaks: FIFO dedup gating (the next send after restore replays an
already-acked message), the by-age reaper (retention is silently
ignored, queues grow unboundedly), and visibility (ReceiveMessage finds
nothing).

## What lands

The encoder now emits three derived side-record families inline during
M5-1's per-message walk:

| Family | Always? | Value | Why required |
|---|---|---|---|
| `!sqs\|msg\|vis\|...` | yes | `[]byte(messageID)` | ReceiveMessage
scans this prefix |
| `!sqs\|msg\|byage\|...` | yes | `[]byte(messageID)` | reaper honors
`MessageRetentionPeriod` |
| `!sqs\|msg\|dedup\|...` | FIFO + non-empty `MessageDedupID` |
`sqsFifoDedupRecord` JSON (4 fields) | gates FIFO sends within 5-minute
window |

`!sqs|msg|group|` rows are **never** emitted: the live
`loadFifoGroupLock` treats key presence alone as "lock held" with no
graceful empty-owner decode, so any value (even zeroed) would
permanently block every group post-restore (gemini critical #885). Test
`TestSQSEncodeSideRecordsNoGroupRows` pins this.

Scope is classic queues only (`partition_count = 1`). M5-1's existing
`ErrSQSEncodeUnsupportedPartitioned` gate at `encodeQueue` means the
side-record walk is never reached for partitioned queues.
Partitioned-FIFO support requires coordinated decoder + encoder lift
(deferred to M5-3 per the design doc's "Deferred (partitioned-FIFO)"
section).

## Test plan

`go test -race -count=1 ./internal/backup/` — all green, including:

- `TestSQSEncodeSideRecordsCrossCheckClassic` — byte-identical key +
value cross-check against the live adapter constructors (`sqsMsgVisKey`,
`sqsMsgByAgeKey`, `sqsMsgDedupKey`, `sqsFifoDedupRecord`) for a
2-message FIFO fixture. The gold-standard pattern the parent encoder
design (§"Encoder cross-check") mandates.
- `TestSQSEncodeSideRecordsStandardQueueOmitsFIFOFamilies` — non-FIFO
queue: vis + byage emitted, dedup NOT.
- `TestSQSEncodeSideRecordsEmptyDedupOmitsDedupRow` — FIFO + empty
`message_deduplication_id`: dedup row NOT emitted.
- `TestSQSEncodeSideRecordsNoGroupRows` — FIFO queue across 3 distinct
`message_group_id` values: 0 `!sqs|msg|group|` rows emitted (regression
pin for gemini critical #885).

The two restore round-trip tests listed in the design doc test plan
(`TestSQSEncodeFifoRestoreRoundTrip`,
`TestSQSEncodeReaperFindsRestoredMessage`) are deferred to the M6 CLI /
Jepsen integration layer rather than this package — they require
single-node cluster boot that is heavier infra than this package's
existing pattern. The cross-check + three negative tests above pin the
contract M5-2 is responsible for at the encoder layer; cluster-level
restore validation belongs at the integration level.

`make lint` clean (verified via golangci-lint --fix on the touched
files).

## Self-review (5 passes)

1. **Data loss** — Full reconstruction eliminates the silent-redeliver /
invisible-message / retention-leak classes M5-1-only restores would
produce. `addSideRecords` surfaces every `b.Add` error and is wired to
the existing per-message walk's error path; no swallowed errors.
2. **Concurrency / distributed failures** — Pure offline derivation. No
Raft, no HLC, no locks; restored cluster's first sends/receives go
through the normal OCC path against the restored state.
3. **Performance** — One `b.Add` per emitted side row, no extra disk
reads (records already loaded by M5-1). Snapshot grows ~2-3× message
count, all small fixed-width keys. `addMessage`'s existing loop body
inflates by ~5 lines, no extra iterations.
4. **Data consistency** — Group-row prohibition pinned by
`TestSQSEncodeSideRecordsNoGroupRows`. Dedup gating on `(FIFO &&
non-empty MessageDedupID)` pinned by two tests. `visibleAt =
available_at_millis` matches the parent design's "SQS: vis zeroed by
default" contract. Cross-check test validates byte-identical output
against the live writer for a shared fixture.
5. **Test coverage** — One cross-check (classic; partitioned variant
deferred to M5-3) plus three negative tests cover the three
conditional-emission rules the design enumerates. Round-trip tests
deferred to M6/Jepsen as discussed above.

## Risk

Low. The encoder is offline; restoration is non-destructive (new
keyspace on a fresh cluster). The change is additive to M5-1's existing
per-message walk and gated on conditions verified by the negative tests;
existing M5-1 round-trip tests continue to pass (verified locally). The
single semantic-changing behavior is "M5-1-only restores silently break
FIFO/reaper/receive" → "restore now works"; no caller of M5-1 depends on
the prior broken state.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **New Features**
* SQS message backups now include preservation of visibility metadata,
age-based attributes, and FIFO deduplication records for complete
message restoration.

* **Tests**
* Added comprehensive test coverage validating SQS side record encoding
across standard and FIFO queue scenarios.

<!-- review_stack_entry_start -->

[![Review Change
Stack](https://storage.googleapis.com/coderabbit_public_assets/review-stack-in-coderabbit-ui.svg)](https://app.coderabbit.ai/change-stack/bootjp/elastickv/pull/892?utm_source=github_walkthrough&utm_medium=github&utm_campaign=change_stack)

<!-- review_stack_entry_end -->

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bootjp

internal/backup/encode_sqs.go

@@ -211,6 +211,9 @@ func (e *SQSRecordEncoder) encodeQueueMessages(b *snapshotBuilder, root *os.Root
 		if err != nil {
 			return err
 		}
+		if err := e.addSideRecords(b, meta.Name, &meta, &records[i]); err != nil {
+			return err
+		}
 		if seq > maxSeq {
 			maxSeq = seq
 		}

internal/backup/encode_sqs_side.go

@@ -0,0 +1,165 @@
+package backup
+
+import (
+	"crypto/sha256"
+	"encoding/binary"
+	"encoding/hex"
+	"encoding/json"
+
+	"github.com/cockroachdb/errors"
+)
+
+// sqsFifoDedupWindowMillis mirrors adapter/sqs_fifo.go (5 minutes). Restored
+// dedup rows that fall outside this window from the backup's send timestamp
+// will be expired the next time the live FIFO send path inspects them.
+const sqsFifoDedupWindowMillis int64 = 5 * 60 * 1000
+
+// sqsSideKeyAllocBytes mirrors the live adapter's sqsKeyCapLarge tuning
+// (adapter/sqs_messages.go:68): a 64-byte tail after the prefix is large
+// enough to hold the BE-u64 generation + visibleAt + base64url(messageID)
+// for typical queue / message IDs without forcing a re-allocation.
+const sqsSideKeyAllocBytes = 64
+
+// sqsFifoDedupRecord mirrors the live struct at adapter/sqs_fifo.go:25.
+// Duplicated here (rather than imported) so the encoder package can run
+// without a circular dependency on adapter, matching the pattern M3b-3
+// used for DynamoDB GSI helpers.
+type sqsFifoDedupRecord struct {
+	MessageID        string `json:"message_id"`
+	SendTimestampMs  int64  `json:"send_timestamp_ms"`
+	ExpiresAtMillis  int64  `json:"expires_at_millis"`
+	OriginalSequence uint64 `json:"original_sequence,omitempty"`
+}
+
+// sqsMsgVisKeyBytes reproduces adapter/sqs_messages.go sqsMsgVisKey:
+// prefix + base64url(queue) + BE-u64(gen) + BE-u64(visibleAt) +
+// base64url(messageID). Negative visibleAt clamps to zero, matching the
+// live uint64MaxZero helper.
+func sqsMsgVisKeyBytes(queueName string, gen uint64, visibleAtMillis int64, messageID string) []byte {
+	out := make([]byte, 0, len(SQSMsgVisPrefix)+sqsSideKeyAllocBytes)
+	out = append(out, SQSMsgVisPrefix...)
+	out = append(out, encodeSQSSegment(queueName)...)
+	out = binary.BigEndian.AppendUint64(out, gen)
+	out = binary.BigEndian.AppendUint64(out, sqsClampNonNegativeMillis(visibleAtMillis))
+	return append(out, encodeSQSSegment(messageID)...)
+}
+
+// sqsMsgByAgeKeyBytes reproduces adapter/sqs_keys.go sqsMsgByAgeKey:
+// prefix + base64url(queue) + BE-u64(gen) + BE-u64(sendTs) +
+// base64url(messageID). Negative sendTs clamps to zero.
+func sqsMsgByAgeKeyBytes(queueName string, gen uint64, sendTimestampMs int64, messageID string) []byte {
+	out := make([]byte, 0, len(SQSMsgByAgePrefix)+sqsSideKeyAllocBytes)
+	out = append(out, SQSMsgByAgePrefix...)
+	out = append(out, encodeSQSSegment(queueName)...)
+	out = binary.BigEndian.AppendUint64(out, gen)
+	out = binary.BigEndian.AppendUint64(out, sqsClampNonNegativeMillis(sendTimestampMs))
+	return append(out, encodeSQSSegment(messageID)...)
+}
+
+// sqsMsgDedupKeyBytes reproduces adapter/sqs_keys.go sqsMsgDedupKey:
+// prefix + base64url(queue) + BE-u64(sqsRestoreGeneration) +
+// base64url(dedupID). The live adapter signature accepts a variable gen,
+// but every M5-2 call site uses sqsRestoreGeneration (a fresh restore
+// has exactly one live generation, with no superseded counters to
+// reference), so the parameter is hardcoded here to satisfy unparam.
+func sqsMsgDedupKeyBytes(queueName, dedupID string) []byte {
+	out := make([]byte, 0, len(SQSMsgDedupPrefix)+sqsSideKeyAllocBytes)
+	out = append(out, SQSMsgDedupPrefix...)
+	out = append(out, encodeSQSSegment(queueName)...)
+	out = binary.BigEndian.AppendUint64(out, sqsRestoreGeneration)
+	return append(out, encodeSQSSegment(dedupID)...)
+}
+
+// sqsClampNonNegativeMillis mirrors adapter/sqs_messages.go uint64MaxZero:
+// wall-clock millis should never be negative, but a negative int64 would
+// silently overflow under a direct uint64() cast and produce a far-future
+// key, so clamp to zero defensively.
+func sqsClampNonNegativeMillis(v int64) uint64 {
+	if v < 0 {
+		return 0
+	}
+	return uint64(v)
+}
+
+// encodeFifoDedupRecordBytes mirrors adapter/sqs_fifo.go encodeFifoDedupRecord:
+// straight json.Marshal of the four-field struct. Wrapped with WithStack so
+// callers get a uniform stack-trace at the error site.
+func encodeFifoDedupRecordBytes(r *sqsFifoDedupRecord) ([]byte, error) {
+	b, err := json.Marshal(r)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return b, nil
+}
+
+// resolveDedupID mirrors adapter/sqs_fifo.go resolveFifoDedupID exactly: an
+// explicit MessageDeduplicationId wins; otherwise on ContentBasedDeduplication
+// queues the dedup-id is sha256(body) hex-encoded; otherwise empty (= no row).
+//
+// This is the critical CBD-correctness path: the live adapter writes only the
+// USER-SUPPLIED MessageDeduplicationId into the stored sqsStoredMessage
+// (sqs_messages.go:680), NOT the resolved one. So a CBD-FIFO message round-
+// trips through the dump as message_deduplication_id="", and a naive
+// non-empty-only gate would silently lose dedup protection on restore for
+// every CBD queue. We have to redo the live derivation at restore time.
+func resolveDedupID(rec *sqsMessageRecord, meta *sqsQueueMetaPublic) string {
+	if rec.MessageDedupID != "" {
+		return rec.MessageDedupID
+	}
+	if meta.ContentBasedDeduplication {
+		sum := sha256.Sum256(rec.Body)
+		return hex.EncodeToString(sum[:])
+	}
+	return ""
+}
+
+// addSideRecords emits the vis + byage + (conditional) dedup rows that the
+// live adapter would have written alongside the !sqs|msg|data| record M5-1
+// already stages. No !sqs|msg|group| rows are emitted at any time — see
+// docs/design/2026_05_30_proposed_sqs_side_record_derivation.md "Families"
+// table for why (loadFifoGroupLock treats key presence alone as "lock held";
+// any value would permanently block every group post-restore).
+//
+// Emission rules:
+//   - vis:   always emitted, visibleAt = rec.AvailableAtMillis. For delayed
+//     messages captured before their delay expired this is in the future;
+//     the live ReceiveMessage path honors this exactly as it would for a
+//     fresh send (the message is invisible until the scheduled time).
+//   - byage: always emitted, sendTs = rec.SendTimestampMillis (required by
+//     the reaper to honor MessageRetentionPeriod after restore).
+//   - dedup: FIFO + resolveDedupID(rec, meta) non-empty. ExpiresAtMillis =
+//     SendTimestampMs + sqsFifoDedupWindowMillis. CBD queues get a SHA-256
+//     derived dedup-id (matches adapter/sqs_fifo.go resolveFifoDedupID).
+func (e *SQSRecordEncoder) addSideRecords(b *snapshotBuilder, queueName string, meta *sqsQueueMetaPublic, rec *sqsMessageRecord) error {
+	msgIDBytes := []byte(rec.MessageID)
+
+	visKey := sqsMsgVisKeyBytes(queueName, sqsRestoreGeneration, rec.AvailableAtMillis, rec.MessageID)
+	if err := b.Add(visKey, msgIDBytes, 0); err != nil {
+		return err
+	}
+
+	byAgeKey := sqsMsgByAgeKeyBytes(queueName, sqsRestoreGeneration, rec.SendTimestampMillis, rec.MessageID)
+	if err := b.Add(byAgeKey, msgIDBytes, 0); err != nil {
+		return err
+	}
+
+	if !meta.FIFO {
+		return nil
+	}
+	dedupID := resolveDedupID(rec, meta)
+	if dedupID == "" {
+		return nil
+	}
+	dedupRec := &sqsFifoDedupRecord{
+		MessageID:        rec.MessageID,
+		SendTimestampMs:  rec.SendTimestampMillis,
+		ExpiresAtMillis:  rec.SendTimestampMillis + sqsFifoDedupWindowMillis,
+		OriginalSequence: rec.SequenceNumber,
+	}
+	val, err := encodeFifoDedupRecordBytes(dedupRec)
+	if err != nil {
+		return err
+	}
+	dedupKey := sqsMsgDedupKeyBytes(queueName, dedupID)
+	return b.Add(dedupKey, val, 0)
+}