backup(zset): PR790 r2 codex P1 — handle legacy !redis|zset| single-blob layout

Codex flagged that the wide-column zset encoder skips the legacy
consolidated single-key blob layout the live store still writes.
A zset stored only as `!redis|zset|<userKey>` (with the magic-
prefixed pb.RedisZSetValue body) is silently dropped from backup
output and any inline TTL becomes an orphan — user-visible
sorted-set data loss.

Live-side references (adapter, not changed by this commit):
- adapter/redis_compat_types.go:82 — redisZSetPrefix
- adapter/redis_compat_commands.go:3495-3508 — writes the blob for
  non-empty persisted zset updates
- adapter/redis_compat_helpers.go:610-631 — reads it as the
  fallback when no wide-column members exist

Fix: new public RedisDB.HandleZSetLegacyBlob method that decodes
the magic-prefixed pb.RedisZSetValue and registers the same per-
member state HandleZSetMember would. The wide-column merge case
(mid-migration snapshot containing BOTH layouts for the same user
key) works because `!redis|zset|` sorts BEFORE `!zs|...` so the
blob arrives first and wide-column rows then update / add members
via the latest-wins map.

Inline TTL: `!redis|zset|<k>` sorts BEFORE `!redis|ttl|<k>`, so
HandleTTL after this handler sees redisKindZSet already and
folds via the case-redisKindZSet branch. No pendingTTL detour
needed for this ordering.

Fail-closed contract (matches existing wide-column path):
- Missing magic prefix → ErrRedisInvalidZSetLegacyBlob
- Unmarshal error    → ErrRedisInvalidZSetLegacyBlob
- NaN score          → ErrRedisInvalidZSetLegacyBlob (Redis ZADD
                       rejects NaN at wire level)

Caller audit (per /loop standing instruction): new public method
HandleZSetLegacyBlob has no external callers. Verified via
'grep -rn HandleZSetLegacyBlob --include=*.go' — all matches inside
the test file in this PR. The cmd/elastickv-snapshot-decode
dispatcher (Phase 0a follow-up, not yet built) will route the
`!redis|zset|` prefix to this handler.

Parallel bug class: the SAME issue exists for `!redis|hash|`,
`!redis|set|`, and `!redis|stream|` legacy blob prefixes. Those
encoders shipped in earlier PRs (#725, #758, #791). Each needs
its own legacy-blob handler in a follow-up PR; this commit fixes
only the zset case codex flagged on PR #790.

New tests:
- TestRedisDB_ZSetLegacyBlobRoundTrip — basic round-trip
- TestRedisDB_ZSetLegacyBlobThenWideColumnMerges — mid-migration
- TestRedisDB_ZSetLegacyBlobWithInlineTTL — TTL ordering
- TestRedisDB_ZSetLegacyBlobRejectsMissingMagic — fail-closed
- TestRedisDB_ZSetLegacyBlobRejectsNaNScore — fail-closed
- TestRedisDB_ZSetLegacyBlobRejectsMalformedKey — fail-closed

Self-review:
1. Data loss — exact opposite: this commit RECOVERS zsets that
   were silently dropped. New fail-closed guards prevent silently
   importing a corrupt blob.
2. Concurrency — no new shared state; per-DB sequential as before.
3. Performance — one protobuf Unmarshal per legacy zset key
   (same as the live read path). Member map shares the same
   latest-wins behavior as wide-column intake.
4. Consistency — merge order (blob first, wide-column second) is
   determined by snapshot lex order; tested explicitly.
5. Coverage — 6 new tests. All 84 redis tests pass.

Author: bootjp

internal/backup/redis_zset.go

@@ -8,7 +8,9 @@ import (
 	"path/filepath"
 	"sort"
 
+	pb "github.com/bootjp/elastickv/proto"
 	cockroachdberr "github.com/cockroachdb/errors"
+	gproto "google.golang.org/protobuf/proto"
 )
 
 // Redis zset encoder. Translates raw !zs|... snapshot records into the
@@ -36,17 +38,48 @@ const (
 	RedisZSetScorePrefix     = "!zs|scr|"
 	RedisZSetMetaDeltaPrefix = "!zs|meta|d|"
 
+	// RedisZSetLegacyBlobPrefix is the consolidated single-key
+	// layout the live store still writes for non-empty persisted
+	// zsets (`adapter/redis_compat_types.go:82` redisZSetPrefix,
+	// produced by `adapter/redis_compat_commands.go:3495-3508` and
+	// read by `adapter/redis_compat_helpers.go:610-631` as the
+	// fallback when no wide-column members exist). A backup that
+	// skipped this prefix would silently drop legacy-only zsets;
+	// HandleZSetLegacyBlob decodes the blob and registers the same
+	// per-member state HandleZSetMeta + HandleZSetMember would.
+	// Codex P1 finding on PR #790 (round 2).
+	RedisZSetLegacyBlobPrefix = "!redis|zset|"
+
 	// redisZSetScoreSize is the size of the IEEE 754 big-endian score
 	// stored in !zs|mem| values. Same constant as zsetMetaSizeBytes in
 	// store/zset_helpers.go; duplicated here to keep the backup
 	// package free of internal/storage imports.
 	redisZSetScoreSize = 8
+
+	// redisZSetLegacyProtoPrefixLen is the on-disk magic prefix size
+	// for `!redis|zset|` values
+	// (`adapter/redis_storage_codec.go:15` storedRedisZSetProtoPrefix).
+	redisZSetLegacyProtoPrefixLen = 4
 )
 
+// redisZSetLegacyProtoPrefix mirrors
+// adapter/redis_storage_codec.go:15 storedRedisZSetProtoPrefix. A
+// rename on the live side without an accompanying backup update
+// surfaces as ErrRedisInvalidZSetLegacyBlob on decode of any real
+// cluster dump.
+var redisZSetLegacyProtoPrefix = []byte{0x00, 'R', 'Z', 0x01}
+
 // ErrRedisInvalidZSetMeta is returned when an !zs|meta| value is not
 // the expected 8-byte big-endian member count.
 var ErrRedisInvalidZSetMeta = cockroachdberr.New("backup: invalid !zs|meta| value")
 
+// ErrRedisInvalidZSetLegacyBlob is returned when a `!redis|zset|`
+// value's magic prefix is missing, its protobuf body fails to
+// unmarshal, or its decoded scores include NaN. Fail-closed for
+// the same reason as ErrRedisInvalidZSetMember: silently accepting
+// a corrupt blob would lose the entire zset's contents at restore.
+var ErrRedisInvalidZSetLegacyBlob = cockroachdberr.New("backup: invalid !redis|zset| value")
+
 // ErrRedisInvalidZSetMember is returned when an !zs|mem| value is not
 // the expected 8-byte IEEE 754 score, or contains a NaN score (Redis's
 // ZADD command rejects NaN, so a NaN at backup time indicates store
@@ -153,6 +186,91 @@ func (r *RedisDB) HandleZSetScore(_, _ []byte) error { return nil }
 // source of truth at backup time.
 func (r *RedisDB) HandleZSetMetaDelta(_, _ []byte) error { return nil }
 
+// HandleZSetLegacyBlob processes one `!redis|zset|<userKey>` record.
+// This is the consolidated single-key layout the live store still
+// writes for non-empty persisted zsets (see RedisZSetLegacyBlobPrefix
+// docstring). The encoded value is a magic-prefixed
+// `pb.RedisZSetValue` carrying every (member, score) pair.
+//
+// Decoded entries land in the same per-key state HandleZSetMember
+// would have produced, so the per-zset JSON output is identical
+// regardless of which layout the live store used. NaN scores fail
+// closed at intake, matching HandleZSetMember's contract.
+//
+// The legacy prefix `!redis|zset|` lex-sorts BEFORE `!zs|...` and
+// BEFORE `!redis|ttl|`, so when a zset is stored in both formats
+// (mid-migration), this handler creates the state first and the
+// later wide-column records merge into it — duplicate
+// HandleZSetMember calls follow the same latest-wins policy.
+//
+// `!redis|zset|` ALSO sorts BEFORE `!redis|ttl|`, so an inline TTL
+// on the same user key will reach HandleTTL after this handler has
+// already registered redisKindZSet. The HandleTTL redisKindZSet
+// branch then folds the expiry into st.expireAtMs via zsetState
+// (which itself drains pendingTTL — a no-op here since the typed
+// record came first).
+func (r *RedisDB) HandleZSetLegacyBlob(key, value []byte) error {
+	userKey, ok := parseZSetLegacyBlobKey(key)
+	if !ok {
+		return cockroachdberr.Wrapf(ErrRedisInvalidZSetLegacyBlob, "key: %q", key)
+	}
+	entries, err := decodeZSetLegacyBlobValue(value)
+	if err != nil {
+		return err
+	}
+	st := r.zsetState(userKey)
+	for _, e := range entries {
+		st.members[e.member] = e.score
+	}
+	return nil
+}
+
+// zsetLegacyEntry is the per-(member, score) projection extracted
+// from a `!redis|zset|` blob's protobuf body.
+type zsetLegacyEntry struct {
+	member string
+	score  float64
+}
+
+// parseZSetLegacyBlobKey strips `!redis|zset|` and returns the
+// user-key bytes. Unlike the wide-column meta key there is no
+// userKeyLen prefix — the live store appends the user key directly
+// (`adapter/redis_compat_types.go:177` ZSetKey).
+func parseZSetLegacyBlobKey(key []byte) ([]byte, bool) {
+	rest := bytes.TrimPrefix(key, []byte(RedisZSetLegacyBlobPrefix))
+	if len(rest) == len(key) {
+		return nil, false
+	}
+	return rest, true
+}
+
+// decodeZSetLegacyBlobValue strips the magic prefix and unmarshals
+// the protobuf body into a slice of (member, score) entries.
+// Rejects NaN scores (same fail-closed contract as
+// HandleZSetMember).
+func decodeZSetLegacyBlobValue(value []byte) ([]zsetLegacyEntry, error) {
+	if len(value) < redisZSetLegacyProtoPrefixLen ||
+		!bytes.Equal(value[:redisZSetLegacyProtoPrefixLen], redisZSetLegacyProtoPrefix) {
+		return nil, cockroachdberr.Wrapf(ErrRedisInvalidZSetLegacyBlob,
+			"missing or corrupt magic prefix (len=%d)", len(value))
+	}
+	msg := &pb.RedisZSetValue{}
+	if err := gproto.Unmarshal(value[redisZSetLegacyProtoPrefixLen:], msg); err != nil {
+		return nil, cockroachdberr.Wrapf(ErrRedisInvalidZSetLegacyBlob,
+			"unmarshal: %v", err)
+	}
+	out := make([]zsetLegacyEntry, 0, len(msg.GetEntries()))
+	for _, e := range msg.GetEntries() {
+		score := e.GetScore()
+		if math.IsNaN(score) {
+			return nil, cockroachdberr.Wrapf(ErrRedisInvalidZSetLegacyBlob,
+				"NaN score for member %q", e.GetMember())
+		}
+		out = append(out, zsetLegacyEntry{member: e.GetMember(), score: score})
+	}
+	return out, nil
+}
+
 // zsetState lazily creates per-key state. Mirrors the hash/list/set
 // kindByKey-registration pattern so HandleZSetMeta, HandleZSetMember,
 // and the HandleTTL back-edge all agree on the kind.

internal/backup/redis_zset_test.go

@@ -8,9 +8,41 @@ import (
 	"path/filepath"
 	"testing"
 
+	pb "github.com/bootjp/elastickv/proto"
 	"github.com/cockroachdb/errors"
+	gproto "google.golang.org/protobuf/proto"
 )
 
+// encodeZSetLegacyBlobValue produces the magic-prefixed protobuf
+// wire format the live store writes for `!redis|zset|<userKey>`
+// values (mirror of adapter/redis_storage_codec.go::marshalZSetValue).
+func encodeZSetLegacyBlobValue(t *testing.T, entries []zsetLegacyEntry) []byte {
+	t.Helper()
+	msg := &pb.RedisZSetValue{}
+	for _, e := range entries {
+		msg.Entries = append(msg.Entries, &pb.RedisZSetEntry{
+			Member: e.member,
+			Score:  e.score,
+		})
+	}
+	body, err := gproto.Marshal(msg)
+	if err != nil {
+		t.Fatalf("marshal pb.RedisZSetValue: %v", err)
+	}
+	out := make([]byte, 0, redisZSetLegacyProtoPrefixLen+len(body))
+	out = append(out, redisZSetLegacyProtoPrefix...)
+	out = append(out, body...)
+	return out
+}
+
+// zsetLegacyBlobKey is the test-side mirror of
+// adapter/redis_compat_types.go:177 ZSetKey:
+// `!redis|zset|<userKey>` (no userKeyLen prefix).
+func zsetLegacyBlobKey(userKey string) []byte {
+	out := []byte(RedisZSetLegacyBlobPrefix)
+	return append(out, userKey...)
+}
+
 // encodeZSetMetaValue builds the 8-byte BE member-count value used by
 // the live store/zset_helpers.go (mirror of store.MarshalZSetMeta).
 func encodeZSetMetaValue(memberCount int64) []byte {
@@ -608,3 +640,138 @@ func TestRedisDB_ZSetMaxInt64DeclaredLen(t *testing.T) {
 		t.Fatalf("math.MaxInt64 boundary must be accepted, got %v", err)
 	}
 }
+
+// TestRedisDB_ZSetLegacyBlobRoundTrip pins the codex P1 fix: a
+// zset stored only via the legacy `!redis|zset|<userKey>` blob
+// must surface in the dump with all its members. Without
+// HandleZSetLegacyBlob, the encoder would skip the record and
+// produce an empty zsets/ output for that key — silent backup
+// data loss.
+func TestRedisDB_ZSetLegacyBlobRoundTrip(t *testing.T) {
+	t.Parallel()
+	db, root := newRedisDB(t)
+	value := encodeZSetLegacyBlobValue(t, []zsetLegacyEntry{
+		{member: "alice", score: 100},
+		{member: "bob", score: 50},
+		{member: "charlie", score: 30},
+	})
+	if err := db.HandleZSetLegacyBlob(zsetLegacyBlobKey("leaderboard"), value); err != nil {
+		t.Fatalf("HandleZSetLegacyBlob: %v", err)
+	}
+	if err := db.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	got := readZSetJSON(t, filepath.Join(root, "redis", "db_0", "zsets", "leaderboard.json"))
+	// Members sorted by member-name bytes (matches the wide-column
+	// encoder's output policy).
+	assertZSetMembersEqual(t, zsetMembersArray(t, got), []zsetMemberEntry{
+		{member: "alice", scoreNum: 100, scoreKind: "number"},
+		{member: "bob", scoreNum: 50, scoreKind: "number"},
+		{member: "charlie", scoreNum: 30, scoreKind: "number"},
+	})
+}
+
+// TestRedisDB_ZSetLegacyBlobThenWideColumnMerges pins the mid-
+// migration shape: when a snapshot carries BOTH the legacy
+// `!redis|zset|<k>` blob AND wide-column `!zs|mem|<k>...` rows for
+// the same user key (which the live store does during the
+// migration window), the encoder must produce a single merged
+// zset. `!redis|zset|` lex-sorts before `!zs|...` so the blob
+// arrives first; the wide-column rows then update / add members.
+func TestRedisDB_ZSetLegacyBlobThenWideColumnMerges(t *testing.T) {
+	t.Parallel()
+	db, root := newRedisDB(t)
+	legacy := encodeZSetLegacyBlobValue(t, []zsetLegacyEntry{
+		{member: "alice", score: 1},
+		{member: "bob", score: 2},
+	})
+	if err := db.HandleZSetLegacyBlob(zsetLegacyBlobKey("k"), legacy); err != nil {
+		t.Fatal(err)
+	}
+	// Wide-column rows: update alice's score, add charlie.
+	if err := db.HandleZSetMember(zsetMemberKey("k", []byte("alice")), encodeZSetScoreValue(99)); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.HandleZSetMember(zsetMemberKey("k", []byte("charlie")), encodeZSetScoreValue(3)); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	got := readZSetJSON(t, filepath.Join(root, "redis", "db_0", "zsets", "k.json"))
+	assertZSetMembersEqual(t, zsetMembersArray(t, got), []zsetMemberEntry{
+		{member: "alice", scoreNum: 99, scoreKind: "number"}, // wide-column won
+		{member: "bob", scoreNum: 2, scoreKind: "number"},    // legacy survived
+		{member: "charlie", scoreNum: 3, scoreKind: "number"},
+	})
+}
+
+// TestRedisDB_ZSetLegacyBlobWithInlineTTL pins that a TTL'd zset
+// stored only via the legacy blob round-trips its expiry. The
+// snapshot order is `!redis|zset|<k>` (sorts before `!redis|ttl|`),
+// so HandleZSetLegacyBlob runs first and registers redisKindZSet,
+// then HandleTTL routes via the redisKindZSet branch (no
+// pendingTTL detour needed for this ordering).
+func TestRedisDB_ZSetLegacyBlobWithInlineTTL(t *testing.T) {
+	t.Parallel()
+	db, root := newRedisDB(t)
+	value := encodeZSetLegacyBlobValue(t, []zsetLegacyEntry{{member: "m", score: 1}})
+	if err := db.HandleZSetLegacyBlob(zsetLegacyBlobKey("k"), value); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.HandleTTL([]byte("k"), encodeTTLValue(fixedExpireMs)); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Finalize(); err != nil {
+		t.Fatal(err)
+	}
+	got := readZSetJSON(t, filepath.Join(root, "redis", "db_0", "zsets", "k.json"))
+	if zsetFloat(t, got, "expire_at_ms") != float64(fixedExpireMs) {
+		t.Fatalf("expire_at_ms = %v want %d", got["expire_at_ms"], fixedExpireMs)
+	}
+}
+
+// TestRedisDB_ZSetLegacyBlobRejectsMissingMagic pins fail-closed
+// behaviour: a `!redis|zset|<k>` value without the magic prefix
+// fails at intake rather than silently decoding garbage protobuf.
+func TestRedisDB_ZSetLegacyBlobRejectsMissingMagic(t *testing.T) {
+	t.Parallel()
+	db, _ := newRedisDB(t)
+	body, err := gproto.Marshal(&pb.RedisZSetValue{})
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	err = db.HandleZSetLegacyBlob(zsetLegacyBlobKey("k"), body) // no magic prefix
+	if !errors.Is(err, ErrRedisInvalidZSetLegacyBlob) {
+		t.Fatalf("err=%v want ErrRedisInvalidZSetLegacyBlob", err)
+	}
+}
+
+// TestRedisDB_ZSetLegacyBlobRejectsNaNScore pins NaN-fail-closed
+// parallel to HandleZSetMember's contract. Redis ZADD rejects NaN
+// at the wire level, so a NaN in storage indicates corruption.
+func TestRedisDB_ZSetLegacyBlobRejectsNaNScore(t *testing.T) {
+	t.Parallel()
+	db, _ := newRedisDB(t)
+	value := encodeZSetLegacyBlobValue(t, []zsetLegacyEntry{{member: "m", score: math.NaN()}})
+	err := db.HandleZSetLegacyBlob(zsetLegacyBlobKey("k"), value)
+	if !errors.Is(err, ErrRedisInvalidZSetLegacyBlob) {
+		t.Fatalf("err=%v want ErrRedisInvalidZSetLegacyBlob", err)
+	}
+}
+
+// TestRedisDB_ZSetLegacyBlobRejectsMalformedKey pins that a
+// `!redis|zset|` key with no trailing user-key bytes fails parse.
+func TestRedisDB_ZSetLegacyBlobRejectsMalformedKey(t *testing.T) {
+	t.Parallel()
+	db, _ := newRedisDB(t)
+	value := encodeZSetLegacyBlobValue(t, []zsetLegacyEntry{{member: "m", score: 1}})
+	// Key has the prefix but no trailing user-key bytes — parser must
+	// still accept it (empty user key is technically valid Redis).
+	// Use a key that doesn't have the prefix to trigger the parse
+	// failure.
+	err := db.HandleZSetLegacyBlob([]byte("not-the-right-prefix|k"), value)
+	if !errors.Is(err, ErrRedisInvalidZSetLegacyBlob) {
+		t.Fatalf("err=%v want ErrRedisInvalidZSetLegacyBlob", err)
+	}
+}