backup(stream): PR791 r1 codex P1 x2 — pendingTTL drain + duplicate-field preservation

Two P1 findings from chatgpt-codex on PR #791:

P1a: Buffer stream TTLs that arrive before stream rows

Pebble snapshots emit records in encoded-key order
(store/snapshot_pebble.go::iter.First()+Next()), and
`!redis|ttl|` lex-sorts BEFORE `!stream|...` because `r` < `s`.
In real snapshot order the TTL arrives FIRST, kindByKey is still
redisKindUnknown when HandleTTL fires, and the original code
counted the TTL as an orphan and dropped it — every TTL'd stream
restored as permanent.

Same root cause as the set encoder's latent bug in PR #758. This
commit adds a pendingTTL infrastructure (matching the parallel fix
on PR #790) so the expiry parks during the redisKindUnknown
window and drains when streamState first registers the user key.
The set encoder gets the same retroactive drain.

P1b: Preserve duplicate stream fields instead of map-collapsing

XADD permits duplicate field names within one entry (e.g.
`XADD s * f v1 f v2`). The protobuf entry stores the interleaved
slice verbatim, but my marshalStreamJSONL collapsed pairs into
`map[string]string`, silently dropping every duplicate. A restore
of such an entry would lose the second (and later) pair.

Fix: emit `fields` as a JSON ARRAY of `{name, value}` records
(streamFieldJSON). Order is the protobuf's interleaved order so
a restore can replay the original XADD argv exactly.

The design example at
docs/design/2026_04_29_proposed_snapshot_logical_decoder.md:338
showed object form. That representation was unsafe for streams
(though fine for hashes where the wire-level encoder normalises
field names earlier). The format is owned by Phase 0 — adjusted
in this PR before the format ships any consumers.

Caller audit (per /loop standing instruction):
- HandleTTL's redisKindUnknown branch: same semantic change as
  PR #790's r1 — previously incremented orphanTTLCount on intake;
  now buffers in pendingTTL and lets Finalize count at end.
  Same audit conclusion: no external callers of orphanTTLCount;
  TestRedisDB_OrphanTTLCountedNotBuffered updated to assert the
  new intake/Finalize split.
- streamEntryJSON.Fields type change `map → slice`: only
  marshalled by encoding/json; the only reader is the test suite,
  which is updated in this commit. No on-disk format compatibility
  concerns because Phase 0 has not shipped a consumer yet.
- New caller claimPendingTTL: called by setState (retroactive) and
  streamState (new) in this PR. hashState/listState don't call it
  because their type prefixes lex-sort BEFORE `!redis|ttl|`.
  Verified via `grep -n 'claimPendingTTL' internal/backup/`.

New tests:
- TestRedisDB_StreamDuplicateFieldsPreserved — pins P1b fix.
- TestRedisDB_StreamTTLArrivesBeforeRows — pins P1a fix for streams.
- TestRedisDB_SetTTLArrivesBeforeRows — retroactive coverage for
  PR #758's set encoder (same root cause as the stream bug).
- TestRedisDB_StreamFieldsDecodedToArray (renamed from
  ToObject) — updated to match the array shape.

Self-review:
1. Data loss — the original code DROPPED real TTL'd streams on
   every backup AND dropped duplicate-field entries' later pairs.
   This fix recovers both. No new data-loss surface introduced.
2. Concurrency — pendingTTL added to RedisDB which is already
   sequential-per-scope; no new locking required.
3. Performance — pendingTTL holds (string-userKey, uint64-expireAt)
   pairs; same allocation shape as kindByKey. Fields slice
   replaces a map of the same logical size — slightly cheaper
   actually (no hash overhead).
4. Consistency — drain happens at FIRST state registration. The
   array form preserves insertion order from the protobuf so the
   restored XADD argv matches.
5. Coverage — 4 new tests + 2 updated. All 78 redis tests pass.

Author: bootjp

internal/backup/redis_set.go

@@ -114,12 +114,24 @@ func (r *RedisDB) HandleSetMetaDelta(_, _ []byte) error { return nil }
 // setState lazily creates per-key state. Mirrors the hash/list
 // kindByKey-registration pattern so HandleSetMeta, HandleSetMember,
 // and the HandleTTL back-edge all agree on the kind.
+//
+// On first registration we drain any pendingTTL for the user key.
+// `!redis|ttl|<k>` lex-sorts BEFORE `!st|...` (because `r` < `s`),
+// so in real snapshot order the TTL arrives FIRST; HandleTTL parks
+// it in pendingTTL, and this function inlines it into the set's
+// `expire_at_ms`. Without this drain step, every TTL'd set would
+// restore as permanent — a latent bug in PR #758 surfaced by codex
+// on PR #791. Phase 0a tests added in the same PR pin the ordering.
 func (r *RedisDB) setState(userKey []byte) *redisSetState {
 	uk := string(userKey)
 	if st, ok := r.sets[uk]; ok {
 		return st
 	}
 	st := &redisSetState{members: make(map[string]struct{})}
+	if expireAtMs, ok := r.claimPendingTTL(userKey); ok {
+		st.expireAtMs = expireAtMs
+		st.hasTTL = true
+	}
 	r.sets[uk] = st
 	r.kindByKey[uk] = redisKindSet
 	return st

internal/backup/redis_stream.go

@@ -160,12 +160,23 @@ func (r *RedisDB) HandleStreamEntry(key, value []byte) error {
 // hash/list/set/zset kindByKey-registration pattern so HandleStreamMeta,
 // HandleStreamEntry, and the HandleTTL back-edge all agree on the
 // kind.
+//
+// On first registration we drain any pendingTTL for the user key.
+// `!redis|ttl|<k>` lex-sorts BEFORE `!stream|...` (because `r` < `s`),
+// so in real snapshot order the TTL arrives FIRST; HandleTTL parks
+// it in pendingTTL, and this function inlines it into the stream's
+// JSONL `_meta.expire_at_ms`. Without this drain step, every TTL'd
+// stream would restore as permanent. Codex P1 finding on PR #791.
 func (r *RedisDB) streamState(userKey []byte) *redisStreamState {
 	uk := string(userKey)
 	if st, ok := r.streams[uk]; ok {
 		return st
 	}
 	st := &redisStreamState{}
+	if expireAtMs, ok := r.claimPendingTTL(userKey); ok {
+		st.expireAtMs = expireAtMs
+		st.hasTTL = true
+	}
 	r.streams[uk] = st
 	r.kindByKey[uk] = redisKindStream
 	return st
@@ -291,17 +302,30 @@ func (r *RedisDB) writeStreamJSONL(dir string, userKey []byte, st *redisStreamSt
 	return nil
 }
 
-// streamEntryJSON is the dump-format projection of one stream entry.
-// Fields is emitted as a JSON object keyed by name (matching the
-// design's `"fields": {"event":"login","user":"alice"}` example)
-// because XADD itself enforces name/value pair shape and Redis
-// stream field names are user-controlled strings rather than
-// binary-safe bytes. A future format-version bump can switch to a
-// `[{"name":..., "value":...}]` array if reviewers find names that
-// collide under JSON-object keying.
+// streamFieldJSON is the dump-format projection of one (name, value)
+// pair from a stream entry. We emit a list of name/value records
+// rather than a JSON object because XADD permits duplicate field
+// names within one entry — e.g. `XADD s * f v1 f v2` records BOTH
+// (f, v1) and (f, v2) as distinct interleaved entries in the
+// stored protobuf's Fields slice. The design example at
+// docs/design/2026_04_29_proposed_snapshot_logical_decoder.md:338
+// showed an object shape, but that representation silently drops
+// duplicates and a restore would not reproduce the original
+// stream entry. Codex P1 (PR #791) — switched to a name/value
+// record array.
+type streamFieldJSON struct {
+	Name  string `json:"name"`
+	Value string `json:"value"`
+}
+
+// streamEntryJSON is the dump-format projection of one stream
+// entry. Fields is an ARRAY so duplicate field names within a
+// single XADD round-trip correctly. The array preserves the
+// interleaved insertion order from the protobuf so consumers can
+// re-assemble the original XADD argv.
 type streamEntryJSON struct {
 	ID     string            `json:"id"`
-	Fields map[string]string `json:"fields"`
+	Fields []streamFieldJSON `json:"fields"`
 }
 
 // streamMetaJSON is the dump-format projection of the final _meta
@@ -330,13 +354,16 @@ func marshalStreamJSONL(st *redisStreamState) ([]byte, error) {
 	var buf bytes.Buffer
 	const xaddPairWidth = 2 // (name, value) — XADD enforces even arity
 	for _, e := range st.entries {
-		fieldsMap := make(map[string]string, len(e.fields)/xaddPairWidth)
+		fields := make([]streamFieldJSON, 0, len(e.fields)/xaddPairWidth)
 		for i := 0; i+1 < len(e.fields); i += xaddPairWidth {
-			fieldsMap[e.fields[i]] = e.fields[i+1]
+			fields = append(fields, streamFieldJSON{
+				Name:  e.fields[i],
+				Value: e.fields[i+1],
+			})
 		}
 		rec := streamEntryJSON{
 			ID:     formatStreamID(e.ms, e.seq),
-			Fields: fieldsMap,
+			Fields: fields,
 		}
 		line, err := json.Marshal(rec)
 		if err != nil {

internal/backup/redis_stream_test.go

@@ -169,11 +169,62 @@ func TestRedisDB_StreamRoundTripBasic(t *testing.T) {
 	assertStreamMetaTerminator(t, meta, 3, 1714400000002, 0)
 }
 
-// TestRedisDB_StreamFieldsDecodedToObject confirms that interleaved
+// streamFieldsPair is the decoded counterpart of streamFieldJSON used
+// in assertions.
+type streamFieldsPair struct{ name, value string }
+
+// extractStreamFieldsAsPairs pulls the `fields` array out of a
+// decoded JSONL entry and returns it as a slice of (name, value)
+// pairs. Centralises the type assertions so the per-test bodies
+// stay below the cyclop ceiling and forcetypeassert lints don't
+// fire at every call site.
+func extractStreamFieldsAsPairs(t *testing.T, entry map[string]any) []streamFieldsPair {
+	t.Helper()
+	raw, ok := entry["fields"].([]any)
+	if !ok {
+		t.Fatalf("entry.fields = %T(%v), want array", entry["fields"], entry["fields"])
+	}
+	out := make([]streamFieldsPair, 0, len(raw))
+	for i, r := range raw {
+		rec, ok := r.(map[string]any)
+		if !ok {
+			t.Fatalf("entry.fields[%d] = %T(%v), want object", i, r, r)
+		}
+		name, ok := rec["name"].(string)
+		if !ok {
+			t.Fatalf("entry.fields[%d].name = %T(%v), want string", i, rec["name"], rec["name"])
+		}
+		value, ok := rec["value"].(string)
+		if !ok {
+			t.Fatalf("entry.fields[%d].value = %T(%v), want string", i, rec["value"], rec["value"])
+		}
+		out = append(out, streamFieldsPair{name: name, value: value})
+	}
+	return out
+}
+
+// assertStreamFieldsEqual checks that the decoded fields array
+// matches the expected ordered list of (name, value) pairs.
+func assertStreamFieldsEqual(t *testing.T, got []streamFieldsPair, want []streamFieldsPair) {
+	t.Helper()
+	if len(got) != len(want) {
+		t.Fatalf("len(fields) = %d, want %d", len(got), len(want))
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("fields[%d] = %v, want %v", i, got[i], want[i])
+		}
+	}
+}
+
+// TestRedisDB_StreamFieldsDecodedToArray confirms that interleaved
 // `[name1, value1, name2, value2, ...]` arrays from the protobuf
-// land as `{"name1": "value1", "name2": "value2"}` JSON objects per
-// the design example (line 338).
-func TestRedisDB_StreamFieldsDecodedToObject(t *testing.T) {
+// land as a JSONL `[{"name":...,"value":...}]` array. Switched
+// from the design's original map shape (line 338) in response to
+// codex's P1 on PR #791: XADD allows duplicate field names within
+// one entry (e.g. `XADD s * f v1 f v2`) and the map representation
+// silently collapsed them.
+func TestRedisDB_StreamFieldsDecodedToArray(t *testing.T) {
 	t.Parallel()
 	db, root := newRedisDB(t)
 	if err := db.HandleStreamMeta(streamMetaKey("s"), encodeStreamMetaValue(1, 100, 5)); err != nil {
@@ -190,18 +241,92 @@ func TestRedisDB_StreamFieldsDecodedToObject(t *testing.T) {
 	if len(entries) != 1 {
 		t.Fatalf("entries = %d, want 1", len(entries))
 	}
-	fields, ok := entries[0]["fields"].(map[string]any)
-	if !ok {
-		t.Fatalf("entries[0].fields = %T(%v), want object", entries[0]["fields"], entries[0]["fields"])
+	assertStreamFieldsEqual(t, extractStreamFieldsAsPairs(t, entries[0]), []streamFieldsPair{
+		{"event", "login"},
+		{"user", "alice"},
+		{"ip", "10.0.0.1"},
+	})
+}
+
+// TestRedisDB_StreamDuplicateFieldsPreserved pins the codex P1 fix:
+// XADD permits duplicate field names within one entry (e.g.
+// `XADD s * f v1 f v2` stores both pairs verbatim in the
+// protobuf's Fields slice). The original map-based projection
+// silently collapsed duplicates; the array shape preserves them
+// and a restore can replay the original XADD argv exactly.
+func TestRedisDB_StreamDuplicateFieldsPreserved(t *testing.T) {
+	t.Parallel()
+	db, root := newRedisDB(t)
+	if err := db.HandleStreamMeta(streamMetaKey("s"), encodeStreamMetaValue(1, 1, 0)); err != nil {
+		t.Fatal(err)
+	}
+	val := encodeStreamEntryValue(t, "1-0", []string{"f", "v1", "f", "v2", "g", "v3"})
+	if err := db.HandleStreamEntry(streamEntryKey("s", 1, 0), val); err != nil {
+		t.Fatal(err)
 	}
-	want := map[string]any{"event": "login", "user": "alice", "ip": "10.0.0.1"}
-	if len(fields) != len(want) {
-		t.Fatalf("fields = %v, want %v", fields, want)
+	if err := db.Finalize(); err != nil {
+		t.Fatal(err)
 	}
-	for k, v := range want {
-		if fields[k] != v {
-			t.Fatalf("fields[%q] = %v, want %v", k, fields[k], v)
-		}
+	entries, _ := readStreamJSONL(t, filepath.Join(root, "redis", "db_0", "streams", "s.jsonl"))
+	assertStreamFieldsEqual(t, extractStreamFieldsAsPairs(t, entries[0]), []streamFieldsPair{
+		{"f", "v1"},
+		{"f", "v2"},
+		{"g", "v3"},
+	})
+}
+
+// TestRedisDB_StreamTTLArrivesBeforeRows pins the codex P1 fix:
+// `!redis|ttl|<k>` lex-sorts BEFORE `!stream|...` because `r` <
+// `s`, so in real Pebble snapshot order the TTL arrives FIRST.
+// The encoder MUST buffer the expiry in pendingTTL and drain it
+// when streamState first registers the user key, inlining the
+// value into the JSONL `_meta.expire_at_ms`. Without this drain
+// every TTL'd stream would restore as permanent.
+func TestRedisDB_StreamTTLArrivesBeforeRows(t *testing.T) {
+	t.Parallel()
+	db, root := newRedisDB(t)
+	// Snapshot order: TTL first, then meta + entry.
+	if err := db.HandleTTL([]byte("k"), encodeTTLValue(fixedExpireMs)); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.HandleStreamMeta(streamMetaKey("k"), encodeStreamMetaValue(1, 100, 0)); err != nil {
+		t.Fatal(err)
+	}
+	val := encodeStreamEntryValue(t, "100-0", []string{"event", "login"})
+	if err := db.HandleStreamEntry(streamEntryKey("k", 100, 0), val); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	_, meta := readStreamJSONL(t, filepath.Join(root, "redis", "db_0", "streams", "k.jsonl"))
+	if streamFloat(t, meta, "expire_at_ms") != float64(fixedExpireMs) {
+		t.Fatalf("meta.expire_at_ms = %v want %d — pendingTTL drain failed", meta["expire_at_ms"], fixedExpireMs)
+	}
+}
+
+// TestRedisDB_SetTTLArrivesBeforeRows pins the same ordering fix
+// for sets (`!redis|ttl|` lex-sorts before `!st|...` because
+// `r` < `s`). Retroactive coverage for PR #758, which shipped the
+// set encoder before the pendingTTL infrastructure existed.
+func TestRedisDB_SetTTLArrivesBeforeRows(t *testing.T) {
+	t.Parallel()
+	db, root := newRedisDB(t)
+	if err := db.HandleTTL([]byte("k"), encodeTTLValue(fixedExpireMs)); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.HandleSetMeta(setMetaKey("k"), encodeSetMetaValue(1)); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.HandleSetMember(setMemberKey("k", []byte("m")), nil); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	got := readSetJSON(t, filepath.Join(root, "redis", "db_0", "sets", "k.json"))
+	if setFloat(t, got, "expire_at_ms") != float64(fixedExpireMs) {
+		t.Fatalf("expire_at_ms = %v want %d — pendingTTL drain failed", got["expire_at_ms"], fixedExpireMs)
 	}
 }