From 3aeede927a5462c09bb08c279329ab7cf3397e4a Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Fri, 16 Oct 2026 15:58:21 +0900 Subject: [PATCH 1/6] docs: record the decoder's gaps on files a running node persists The decoder as first shipped read only the raw store stream. A fsm-snap/.fsm carries the FSM snapshot header in front and the engine's CRC32C footer behind; the store streams every MVCC version, so routing each one wrote stale values and resurrected deleted keys; the SST ingest format and the store's stored-key budget were refused; and the engine's hex file names were parsed as decimal. Add a dated addendum to the decoder design doc recording each gap and the fix the following commits implement, and point the Background section's framing description at it. Found in review on #1275. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01S5GoEpeK4aYNkUxWxutJTz --- ...29_implemented_snapshot_logical_decoder.md | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) diff --git a/docs/design/2026_04_29_implemented_snapshot_logical_decoder.md b/docs/design/2026_04_29_implemented_snapshot_logical_decoder.md index 5e2f4aec4..1d3570a98 100644 --- a/docs/design/2026_04_29_implemented_snapshot_logical_decoder.md +++ b/docs/design/2026_04_29_implemented_snapshot_logical_decoder.md @@ -15,6 +15,12 @@ Date: 2026-04-29 > `docs/operations/snapshot_restore.md`. This doc > remains the format owner; the encoder doc owns the reverse-direction > wire-format reconstruction. +> +> **Addendum (2026-10-16):** the decoder as first shipped read only the +> raw store stream, so a file a running node persists did not decode, +> and a store stream with MVCC history produced a wrong dump. §"Reading +> files a running node persists" records the gaps and the fix, and +> corrects the framing description in §"Background". ## Background @@ -727,6 +733,65 @@ bespoke parser, the format has failed its goal. | `FuzzFilenameEncoding` | Encoder/decoder never panics and is bijective on arbitrary byte input | | `BenchmarkSnapshotDecodeThroughput` | Establishes baseline throughput per adapter so Phase 1 has a regression target | +## Reading files a running node persists (addendum, 2026-10-16) + +Found in review of the positioning roadmap (#1275), after #1267 taught +the reader to skip the store's reserved metadata keys: the decoder still +could not read the `fsm-snap/.fsm` a node writes, and on a raw +store stream with history it produced a wrong dump. Each gap and the +decoder's answer: + +- **Framing.** A persisted file is not the raw store stream §"Background" + describes. `kvFSMSnapshot.WriteTo` (`kv/snapshot.go`) writes an FSM + snapshot header in front of the store payload (`EKVTHLC1`: magic and + ceiling ms; `EKVTHLC2`: magic, a BE uint16 length, ceiling ms, cutover + index, and optional extension bytes), and `writeFSMSnapshotPayload` + (`internal/raftengine/etcd`) appends a 4-byte BE CRC32C (Castagnoli) + footer over everything before it. `backup.ReadSnapshotFile` accepts + that file, a raw store payload (encoder output), and a raw payload + followed by the footer (what a header-less node persisted). With a + header the footer is required and verified; without one it is + recognised when the last four bytes equal the CRC32C of the rest. The + verified value goes into MANIFEST `source.fsm_crc32c`, a field this doc + defined and the CLI never filled. +- **MVCC history.** The store streams every physical version of a key + (`userKey || BE(^commitTS)` in bytewise order, so newest first), not + the one version per key an encoder-produced snapshot holds. Routing + each version wrote older values over newer ones and resurrected a + deleted key from the version under its tombstone. The decoder now + keeps each key's newest version and drops the key when that version is + a tombstone. Versions of one key need not be adjacent, since a key that + extends another can sort between them, so the filter tracks the keys + that are prefixes of the current encoded key, and it fails closed on + out-of-order input. `DecodeCounters.Superseded` counts the hidden + versions. +- **SST ingest format.** With `ELASTICKV_PEBBLE_SST_INGEST_SNAPSHOT` on, + the store payload is the `EKVSSTI1` bundle: a manifest with its + SHA-256, then the SST files. The decoder verifies the manifest and each + file's digest, checks that the files' key ranges are ordered and + disjoint, and reads each SST with Pebble's sstable reader, yielding the + same entries as the legacy stream. This is the one place the package + links a storage library, because the files are in Pebble's format. +- **Key budget.** The reader capped keys at the 1 MiB logical size, while + the store streams stored keys up to `MaxSnapshotStoredKeySize`, 1 MiB + plus a 32-byte internal envelope such as migration staging. The reader + now accepts the stored budget; the encoder keeps the logical cap for + the user keys it writes. +- **File names.** The engine names a file `%016x.fsm`, in hex, and + `SnapshotIndexFromPath` parsed the stem as decimal, so + `0000000000000040.fsm` read back as index 40 instead of 64. A 16-digit + hex stem now parses as hex; any other stem keeps the decimal parse. + +Still out of scope: encrypted entries (`ErrSnapshotEncryptedEntry`; the +decoder links no keyring). Expiry handling is unchanged. + +Tests: the CLI decodes a `.fsm` persisted through the engine's writer +(`PreparePhysicalSnapshotRestore`) and records its footer in the +manifest; a live store with an overwrite and a delete dumps the newest +value and no deleted key, through the legacy stream and through the SST +bundle; an interleaving case and a rapid property test pin the filter; +framing, header, footer, and SST corruption cases fail with typed errors. + ## References - `2026_04_14_implemented_etcd_snapshot_disk_offload.md` — the From 32198b50cd55481c710cf090584b19f31ec01ece Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Fri, 16 Oct 2026 16:00:19 +0900 Subject: [PATCH 2/6] backup: add failing tests for decoding files a running node persists Each test fails on main: - TestRun_DecodesOnDiskFSMFile persists a real KV FSM snapshot through the Raft engine's file writer (FSM snapshot header, store stream, CRC32C footer) and decodes it with the CLI; main rejects the file on its EKVTHLC1 magic. - TestDecodeSnapshot_LiveStoreKeepsNewestVersion decodes a live store with an overwrite and a delete; main writes the stale "old" value and resurrects the deleted key. - TestDecodeSnapshot_SSTIngestSnapshot decodes the EKVSSTI1 bundle a store emits with SST ingest snapshots on; main rejects its magic. - TestMaxSnapshotEncodedKeySize_MatchesStore and TestReadSnapshot_AcceptsStoredKeyBudget pin the reader's key budget to the store's stored-key budget; main refuses a staged maximum-size key. - TestSnapshotIndexFromPath and the two CLI golden tests use the engine's %016x file names; main parses them as decimal (index 64 is 0000000000000040.fsm and read back as 40). Found in review on #1275. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01S5GoEpeK4aYNkUxWxutJTz --- .../fsm_file_test.go | 97 +++++++++ cmd/elastickv-snapshot-decode/main_test.go | 4 +- internal/backup/snapshot_live_store_test.go | 188 ++++++++++++++++++ internal/backup/source_test.go | 6 +- 4 files changed, 292 insertions(+), 3 deletions(-) create mode 100644 cmd/elastickv-snapshot-decode/fsm_file_test.go create mode 100644 internal/backup/snapshot_live_store_test.go diff --git a/cmd/elastickv-snapshot-decode/fsm_file_test.go b/cmd/elastickv-snapshot-decode/fsm_file_test.go new file mode 100644 index 000000000..29c8737f6 --- /dev/null +++ b/cmd/elastickv-snapshot-decode/fsm_file_test.go @@ -0,0 +1,97 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/bootjp/elastickv/internal/backup" + etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd" + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" +) + +// onDiskFSMFile writes one Redis string through a real Pebble store, takes +// the KV FSM's snapshot (the FSM snapshot header in front of the store +// stream), and persists it with the Raft engine's snapshot file writer, +// which appends the CRC32C footer. The result is the +// `fsm-snap/.fsm` file an operator finds in a node's data +// directory. It returns the file path and the footer's CRC32C. +func onDiskFSMFile(t *testing.T, userKey, body string) (string, uint32) { + t.Helper() + dir := t.TempDir() + st, err := store.NewPebbleStore(filepath.Join(dir, "fsm.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + value := append([]byte{0xFF, 0x01, 0x00}, body...) + muts := []*store.KVPairMutation{{Op: store.OpTypePut, Key: []byte(backup.RedisStringPrefix + userKey), Value: value}} + if err := st.ApplyMutationsRaftAt(context.Background(), muts, nil, 1, 1, 64); err != nil { + t.Fatalf("apply: %v", err) + } + + snap, err := kv.NewKvFSMWithHLC(st, kv.NewHLC()).Snapshot() + if err != nil { + t.Fatalf("fsm snapshot: %v", err) + } + defer snap.Close() + var payload bytes.Buffer + if _, err := snap.WriteTo(&payload); err != nil { + t.Fatalf("write fsm snapshot: %v", err) + } + payloadPath := filepath.Join(dir, "payload") + mustWriteFSM(t, payloadPath, payload.Bytes()) + + res, err := etcdraftengine.PreparePhysicalSnapshotRestore(etcdraftengine.PhysicalSnapshotRestoreOptions{ + InputFSMPath: payloadPath, + DataDir: filepath.Join(dir, "node"), + Index: 64, + Term: 1, + Peers: []etcdraftengine.Peer{{NodeID: 1, ID: "n1", Address: "127.0.0.1:50051"}}, + }) + if err != nil { + t.Fatalf("persist .fsm: %v", err) + } + return res.FSMPath, res.CRC32C +} + +// TestRun_DecodesOnDiskFSMFile drives the CLI over the file a running node +// actually persists: the FSM snapshot header, the store stream, and the +// engine's CRC32C footer. The decoder used to expect the raw store stream +// at byte 0, so every on-disk `.fsm` failed on its magic. +func TestRun_DecodesOnDiskFSMFile(t *testing.T) { + t.Parallel() + in, crc := onDiskFSMFile(t, "greeting", "hello") + out := filepath.Join(t.TempDir(), "out") + + var stderr bytes.Buffer + if err := run([]string{"-input", in, "-output", out, "-cluster-id", "ek-disk"}, &stderr); err != nil { + t.Fatalf("run: %v\nstderr=%s", err, stderr.String()) + } + assertBlobMatches(t, filepath.Join(out, "redis", "db_0", "strings", "greeting.bin"), "hello") + assertManifest(t, filepath.Join(out, "MANIFEST.json"), manifestShape{ + Phase: "phase0-snapshot-decode", + SnapshotIndex: 64, + ClusterID: "ek-disk", + }) + body, err := os.ReadFile(filepath.Join(out, "MANIFEST.json")) //nolint:gosec // test path + if err != nil { + t.Fatalf("read manifest: %v", err) + } + var m struct { + Source struct { + FSMCRC32C string `json:"fsm_crc32c"` + } `json:"source"` + } + if err := json.Unmarshal(body, &m); err != nil { + t.Fatalf("unmarshal manifest: %v", err) + } + if want := fmt.Sprintf("%08x", crc); m.Source.FSMCRC32C != want { + t.Fatalf("source.fsm_crc32c = %q, want the verified footer %q", m.Source.FSMCRC32C, want) + } +} diff --git a/cmd/elastickv-snapshot-decode/main_test.go b/cmd/elastickv-snapshot-decode/main_test.go index a63c7f34b..8237f8206 100644 --- a/cmd/elastickv-snapshot-decode/main_test.go +++ b/cmd/elastickv-snapshot-decode/main_test.go @@ -33,7 +33,7 @@ const ( func TestRun_DecodesSimpleRedisStringSnapshot(t *testing.T) { t.Parallel() dir := t.TempDir() - in := filepath.Join(dir, "0000000000000064.fsm") + in := filepath.Join(dir, "0000000000000040.fsm") out := filepath.Join(dir, "out") mustWriteFSM(t, in, redisStringSnapshot(t, "greeting", "hello")) @@ -68,7 +68,7 @@ func TestRun_DecodesSimpleRedisStringSnapshot(t *testing.T) { func TestRun_DecodesLivePebbleSnapshot(t *testing.T) { t.Parallel() dir := t.TempDir() - in := filepath.Join(dir, "0000000000000064.fsm") + in := filepath.Join(dir, "0000000000000040.fsm") out := filepath.Join(dir, "out") mustWriteFSM(t, in, liveRedisStringSnapshot(t, "greeting", "hello")) diff --git a/internal/backup/snapshot_live_store_test.go b/internal/backup/snapshot_live_store_test.go new file mode 100644 index 000000000..98d1ede4b --- /dev/null +++ b/internal/backup/snapshot_live_store_test.go @@ -0,0 +1,188 @@ +package backup + +import ( + "bytes" + "context" + "encoding/binary" + "os" + "path/filepath" + "testing" + + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" +) + +// sstIngestSnapshotEnv mirrors store's sstIngestSnapshotEnv: the switch +// that makes a live Pebble store emit the EKVSSTI1 bundle instead of the +// EKVPBBL1 stream. +const sstIngestSnapshotEnv = "ELASTICKV_PEBBLE_SST_INGEST_SNAPSHOT" + +// liveRedisStringValue wraps body in the live Redis string-value envelope +// (magic 0xFF, version 0x01, flags 0x00 = no TTL) the adapter writes. +func liveRedisStringValue(body string) []byte { + return append([]byte{0xFF, 0x01, 0x00}, body...) +} + +// liveHistoryWrite is one replicated write applied to the test store. +type liveHistoryWrite struct { + key string + body string // ignored when del is set + del bool + commitTS uint64 +} + +// liveHistory gives three Redis strings MVCC history, the way a running +// node accumulates versions between compactions: "updated" is overwritten, +// "deleted" is deleted after its first write, and "kept" is written once. +var liveHistory = []liveHistoryWrite{ + {key: "updated", body: "old", commitTS: 10}, + {key: "deleted", body: "gone", commitTS: 11}, + {key: "kept", body: "same", commitTS: 12}, + {key: "updated", body: "new", commitTS: 20}, + {key: "deleted", del: true, commitTS: 21}, +} + +// liveStoreSnapshotWithHistory applies liveHistory to a real Pebble store, +// one Raft entry per write, and returns the store's native snapshot stream. +func liveStoreSnapshotWithHistory(t *testing.T) []byte { + t.Helper() + st, err := store.NewPebbleStore(filepath.Join(t.TempDir(), "fsm.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + var appliedIndex uint64 + for _, w := range liveHistory { + mut := &store.KVPairMutation{Op: store.OpTypePut, Key: []byte(RedisStringPrefix + w.key), Value: liveRedisStringValue(w.body)} + if w.del { + mut = &store.KVPairMutation{Op: store.OpTypeDelete, Key: []byte(RedisStringPrefix + w.key)} + } + appliedIndex++ + if err := st.ApplyMutationsRaftAt(context.Background(), []*store.KVPairMutation{mut}, nil, w.commitTS, w.commitTS, appliedIndex); err != nil { + t.Fatalf("apply %s@%d: %v", w.key, w.commitTS, err) + } + } + snap, err := st.Snapshot() + if err != nil { + t.Fatalf("snapshot: %v", err) + } + defer snap.Close() + var buf bytes.Buffer + if _, err := snap.WriteTo(&buf); err != nil { + t.Fatalf("write snapshot: %v", err) + } + return buf.Bytes() +} + +// assertLiveHistoryDump checks the Redis dump DecodeSnapshot produced from +// liveHistory: the newest version of each key, and no file for the key +// whose newest version is a delete. +func assertLiveHistoryDump(t *testing.T, out string) { + t.Helper() + dir := filepath.Join(out, "redis", "db_0", "strings") + for key, want := range map[string]string{"updated": "new", "kept": "same"} { + got, err := os.ReadFile(filepath.Join(dir, key+".bin")) //nolint:gosec // test path + if err != nil { + t.Fatalf("read %s: %v", key, err) + } + if string(got) != want { + t.Fatalf("%s = %q, want the newest version %q", key, got, want) + } + } + if _, err := os.Stat(filepath.Join(dir, "deleted.bin")); !os.IsNotExist(err) { + t.Fatalf("deleted key was resurrected from an older version (stat err=%v)", err) + } +} + +func decodeLiveHistory(t *testing.T, stream []byte) string { + t.Helper() + out := filepath.Join(t.TempDir(), "out") + if _, err := DecodeSnapshot(bytes.NewReader(stream), DecodeOptions{OutRoot: out, Adapters: AdapterSet{Redis: true}}); err != nil { + t.Fatalf("DecodeSnapshot: %v", err) + } + return out +} + +// TestDecodeSnapshot_LiveStoreKeepsNewestVersion pins that a dump of a live +// store with MVCC history carries each key's newest version only: the +// store streams every physical version, so routing each one wrote stale +// values over newer ones and resurrected deleted keys from the version +// below their tombstone. +func TestDecodeSnapshot_LiveStoreKeepsNewestVersion(t *testing.T) { + t.Parallel() + stream := liveStoreSnapshotWithHistory(t) + if !bytes.HasPrefix(stream, []byte(PebbleSnapshotMagic)) { + t.Fatalf("expected the legacy %s stream, got %q", PebbleSnapshotMagic, stream[:8]) + } + assertLiveHistoryDump(t, decodeLiveHistory(t, stream)) +} + +// TestDecodeSnapshot_SSTIngestSnapshot pins that the decoder reads the +// EKVSSTI1 bundle a live store emits when SST ingest snapshots are on, +// with the same newest-version semantics as the legacy stream. +func TestDecodeSnapshot_SSTIngestSnapshot(t *testing.T) { + t.Setenv(sstIngestSnapshotEnv, "true") + stream := liveStoreSnapshotWithHistory(t) + if !bytes.HasPrefix(stream, []byte("EKVSSTI1")) { + t.Fatalf("expected an EKVSSTI1 bundle with %s=true, got %q", sstIngestSnapshotEnv, stream[:8]) + } + assertLiveHistoryDump(t, decodeLiveHistory(t, stream)) +} + +// TestMaxSnapshotEncodedKeySize_MatchesStore pins the reader's key budget +// to the live store's physical stored-key budget plus the timestamp +// suffix, so a snapshot carrying a migration-staged maximum-size key is +// not rejected offline. +func TestMaxSnapshotEncodedKeySize_MatchesStore(t *testing.T) { + t.Parallel() + if want := store.MaxSnapshotStoredKeySize + snapshotTSSize; MaxSnapshotEncodedKeySize != want { + t.Fatalf("MaxSnapshotEncodedKeySize = %d, want store.MaxSnapshotStoredKeySize + %d = %d", + MaxSnapshotEncodedKeySize, snapshotTSSize, want) + } +} + +// oneEntryStream builds an EKVPBBL1 stream holding one cleartext entry +// whose encoded key is exactly encodedKeyLen bytes. +func oneEntryStream(t *testing.T, encodedKeyLen int) []byte { + t.Helper() + var buf bytes.Buffer + buf.WriteString(PebbleSnapshotMagic) + mustBinaryWrite(t, &buf, uint64(1)) + key := bytes.Repeat([]byte{'k'}, encodedKeyLen) + copy(key, "!txn|lock|") + binary.BigEndian.PutUint64(key[len(key)-snapshotTSSize:], ^uint64(1)) + mustBinaryWrite(t, &buf, uint64(len(key))) + buf.Write(key) + value := make([]byte, snapshotValueHeaderSize) + mustBinaryWrite(t, &buf, uint64(len(value))) + buf.Write(value) + return buf.Bytes() +} + +func mustBinaryWrite(t *testing.T, buf *bytes.Buffer, v uint64) { + t.Helper() + if err := binary.Write(buf, binary.LittleEndian, v); err != nil { + t.Fatalf("binary.Write: %v", err) + } +} + +// TestReadSnapshot_AcceptsStoredKeyBudget pins that a key at the store's +// physical budget decodes and one byte more is still refused. +func TestReadSnapshot_AcceptsStoredKeyBudget(t *testing.T) { + t.Parallel() + atBudget := store.MaxSnapshotStoredKeySize + snapshotTSSize + entries := 0 + if _, err := ReadSnapshotWithHeader(bytes.NewReader(oneEntryStream(t, atBudget)), func(SnapshotHeader, SnapshotEntry) error { + entries++ + return nil + }); err != nil { + t.Fatalf("key at the stored-key budget (%d bytes) rejected: %v", atBudget, err) + } + if entries != 1 { + t.Fatalf("entries = %d, want 1", entries) + } + _, err := ReadSnapshotWithHeader(bytes.NewReader(oneEntryStream(t, atBudget+1)), func(SnapshotHeader, SnapshotEntry) error { return nil }) + if !errors.Is(err, ErrSnapshotKeyTooLarge) { + t.Fatalf("key one byte over the budget: err = %v, want ErrSnapshotKeyTooLarge", err) + } +} diff --git a/internal/backup/source_test.go b/internal/backup/source_test.go index 823e5d170..eac33a2d4 100644 --- a/internal/backup/source_test.go +++ b/internal/backup/source_test.go @@ -14,7 +14,11 @@ func TestSnapshotIndexFromPath(t *testing.T) { wantIdx uint64 wantErr bool }{ - {"live-writer zero-padded", "/data/fsm-snap/0000000000000064.fsm", 64, false}, + // The Raft engine names every file %016x (fsmSnapPath in + // internal/raftengine/etcd), so a 16-character stem is hex. + {"live-writer name", "/data/fsm-snap/0000000000000040.fsm", 64, false}, + {"live-writer name with hex letters", "000000000001e240.fsm", 123456, false}, + {"live-writer name is not decimal", "0000000000000064.fsm", 100, false}, {"unpadded hand-rolled", "42.fsm", 42, false}, {"max uint64", "18446744073709551615.fsm", 18446744073709551615, false}, {"missing .fsm suffix", "snapshot.bin", 0, true}, From cd94fcfae168a84ad8582c7fd4e4ec41a8af67d4 Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Fri, 16 Oct 2026 16:03:46 +0900 Subject: [PATCH 3/6] backup: accept the store's stored-key budget and share the entry path The reader capped encoded keys at the 1 MiB logical size plus the timestamp, while a live store streams stored keys up to MaxSnapshotStoredKeySize (1 MiB plus a 32-byte internal envelope, such as a migration-staged key) plus the timestamp; raise MaxSnapshotEncodedKeySize to match. The encoder writes user keys, so it keeps the logical cap under its own constant. Split ReadSnapshotWithHeader into readPebbleStream and emitRawEntry so the framed-file and SST-ingest readers that follow can reuse the EKVPBBL1 loop and the per-entry decode (metadata-key skip included). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01S5GoEpeK4aYNkUxWxutJTz --- internal/backup/encode.go | 22 ++++++++---- internal/backup/snapshot_reader.go | 56 ++++++++++++++++++++++-------- 2 files changed, 56 insertions(+), 22 deletions(-) diff --git a/internal/backup/encode.go b/internal/backup/encode.go index f8f366f8c..dab932c1d 100644 --- a/internal/backup/encode.go +++ b/internal/backup/encode.go @@ -45,11 +45,19 @@ import ( // `.fsm` whose Pebble image depends on insertion order. var ErrEncodeDuplicateKey = errors.New("backup: duplicate encoded key in snapshot build") -// ErrEncodeKeyTooLarge / ErrEncodeValueTooLarge mirror the decode-side -// MaxSnapshotEncodedKeySize / MaxSnapshotEncodedValueSize caps. A -// reconstructed entry exceeding them would produce a `.fsm` the live -// restore path rejects, so the encoder fails closed before writing a -// single byte rather than emit an unloadable file. +// maxEncodedLogicalKeySize caps the keys the encoder writes: a logical key +// up to store.MaxSnapshotKeySize (1 MiB), the largest key guaranteed to +// round-trip through a snapshot, plus the timestamp suffix. The reader's +// MaxSnapshotEncodedKeySize is larger by the store's 32-byte internal +// envelope, which only the store's own staging keys use; the encoder +// writes user keys, so it keeps the logical cap. +const maxEncodedLogicalKeySize = maxSnapshotUserKeySize + snapshotTSSize + +// ErrEncodeKeyTooLarge / ErrEncodeValueTooLarge enforce +// maxEncodedLogicalKeySize and the decode-side MaxSnapshotEncodedValueSize. +// A reconstructed entry exceeding them would produce a `.fsm` the live +// restore path rejects or cannot round-trip, so the encoder fails closed +// before writing a single byte rather than emit an unloadable file. var ErrEncodeKeyTooLarge = errors.New("backup: encoded key length exceeds limit") var ErrEncodeValueTooLarge = errors.New("backup: encoded value length exceeds limit") @@ -170,9 +178,9 @@ func (b *snapshotBuilder) Add(userKey, userValue []byte, expireAt uint64) error // framed buffers — encKey = userKey + snapshotTSSize, encVal = // snapshotValueHeaderSize + userValue — so an oversize record // fails closed without a wasted allocation (claude review nit). - if uint64(len(userKey))+snapshotTSSize > MaxSnapshotEncodedKeySize { + if uint64(len(userKey))+snapshotTSSize > maxEncodedLogicalKeySize { return errors.Wrapf(ErrEncodeKeyTooLarge, - "length %d > %d", len(userKey)+snapshotTSSize, MaxSnapshotEncodedKeySize) + "length %d > %d", len(userKey)+snapshotTSSize, maxEncodedLogicalKeySize) } if uint64(len(userValue))+snapshotValueHeaderSize > MaxSnapshotEncodedValueSize { return errors.Wrapf(ErrEncodeValueTooLarge, diff --git a/internal/backup/snapshot_reader.go b/internal/backup/snapshot_reader.go index 03a41961c..653d23bf9 100644 --- a/internal/backup/snapshot_reader.go +++ b/internal/backup/snapshot_reader.go @@ -63,8 +63,12 @@ const ( // MaxSnapshotEncodedKeySize / MaxSnapshotEncodedValueSize bound the // per-entry allocations made by readExact / readExactGrow. Mirrors - // the live store's `maxPebbleEncodedKeySize` (1 MiB user-key cap + - // 8-byte TS suffix; store/mvcc_store.go:29 + store/lsm_store.go:51) + // the live store's `maxPebbleEncodedKeySize` (the physical stored-key + // budget, a 1 MiB logical key plus 32 bytes of internal envelope such + // as migration staging, plus the 8-byte TS suffix; + // store/mvcc_store.go `maxSnapshotStoredKeySize` + + // store/lsm_store.go `maxPebbleEncodedKeySize`; pinned by + // TestMaxSnapshotEncodedKeySize_MatchesStore) // and `maxSnapshotValueSize + valueHeaderSize + // encryption.EnvelopeOverhead` (256 MiB cleartext + 9-byte header + // 34-byte envelope; store/mvcc_store.go:37 + @@ -83,13 +87,20 @@ const ( // maxSnapshotUserKeySize mirrors store/mvcc_store.go:29 // `maxSnapshotKeySize` = 1 MiB. maxSnapshotUserKeySize = 1 << maxSnapshotUserKeyShift + // maxSnapshotInternalKeyEnvelope / maxSnapshotStoredKeySize mirror + // store/mvcc_store.go `maxSnapshotInternalKeyEnvelope` (32 bytes of + // headroom for internal stored-key envelopes) and + // `maxSnapshotStoredKeySize`: a live store may stream a stored key + // that long, so the offline reader accepts it too. + maxSnapshotInternalKeyEnvelope = 32 + maxSnapshotStoredKeySize = maxSnapshotUserKeySize + maxSnapshotInternalKeyEnvelope // maxSnapshotUserValueSize mirrors store/mvcc_store.go:37 // `maxSnapshotValueSize` = 256 MiB. maxSnapshotUserValueSize = maxSnapshotValueMiB << maxSnapshotValueShift maxSnapshotUserKeyShift = 20 // 1 << 20 == 1 MiB maxSnapshotValueShift = 20 // << 20 == byte count maxSnapshotValueMiB = 256 - MaxSnapshotEncodedKeySize = maxSnapshotUserKeySize + snapshotTSSize + MaxSnapshotEncodedKeySize = maxSnapshotStoredKeySize + snapshotTSSize MaxSnapshotEncodedValueSize = maxSnapshotUserValueSize + snapshotValueHeaderSize + envelopeMaxB // envelopeMaxB mirrors internal/encryption.EnvelopeOverhead (12- @@ -215,8 +226,18 @@ func ReadSnapshot(r io.Reader, fn func(SnapshotHeader, SnapshotEntry) error) err // DecodeSnapshot needs this so MANIFEST.json's `last_commit_ts` // field is populated for an empty-snapshot dump — the per-entry // callback's header argument cannot fire on a header-only file. +// +// It reads the raw store stream only. A file persisted by a node +// (`fsm-snap/.fsm`) or an SST ingest bundle goes through +// ReadSnapshotFile, which strips the framing and dispatches on the +// store format. func ReadSnapshotWithHeader(r io.Reader, fn func(SnapshotHeader, SnapshotEntry) error) (SnapshotHeader, error) { - br := bufio.NewReader(r) + return readPebbleStream(bufio.NewReader(r), fn) +} + +// readPebbleStream reads one EKVPBBL1 stream from br: the header, then +// every entry up to the clean EOF that terminates it. +func readPebbleStream(br *bufio.Reader, fn func(SnapshotHeader, SnapshotEntry) error) (SnapshotHeader, error) { header, err := readSnapshotHeader(br) if err != nil { return SnapshotHeader{}, err @@ -315,22 +336,27 @@ func readOneEntry( if err != nil { return false, err } - // A snapshot taken from a live Pebble store carries the store's - // reserved metadata keys (`_meta_last_commit_ts` and its siblings) - // next to the MVCC versions. They have neither the inverted-timestamp - // key suffix nor the value header, so decoding one fails the whole - // read; they carry no user data, so the reader skips them. + return false, emitRawEntry(key, value, header, fn) +} + +// emitRawEntry decodes one raw (key, value) pair as the live store wrote +// it and hands it to fn. Shared by the EKVPBBL1 stream and the SST ingest +// bundle, which carry the same physical keys and values. +// +// A snapshot taken from a live Pebble store carries the store's reserved +// metadata keys (`_meta_last_commit_ts` and its siblings) next to the +// MVCC versions. They have neither the inverted-timestamp key suffix nor +// the value header, so decoding one fails the whole read; they carry no +// user data, so the reader skips them. +func emitRawEntry(key, value []byte, header SnapshotHeader, fn func(SnapshotHeader, SnapshotEntry) error) error { if isPebbleOperationalKey(key) { - return false, nil + return nil } entry, err := decodeSnapshotEntry(key, value) if err != nil { - return false, err - } - if err := fn(header, entry); err != nil { - return false, err + return err } - return false, nil + return fn(header, entry) } // readEntryKey reads the per-entry length-prefix-then-bytes for the From 3f097df6c1e2a5c0c4c06b6cc521ddf3d4c8132b Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Fri, 16 Oct 2026 16:06:36 +0900 Subject: [PATCH 4/6] backup: read on-disk FSM framing, SST ingest bundles, and keep the newest version ReadSnapshotFile accepts a store payload with or without the FSM snapshot header (EKVTHLC1/EKVTHLC2) a node writes in front of it, and withholds and verifies the engine's CRC32C footer. It dispatches the payload to the EKVPBBL1 stream reader or to the new EKVSSTI1 bundle reader, which checks the manifest and per-file SHA-256 before parsing each SST. latestVersionFilter keeps the newest version of each user key from a store that streams every physical MVCC version. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01S5GoEpeK4aYNkUxWxutJTz --- internal/backup/latest_version.go | 61 ++++++ internal/backup/snapshot_file.go | 314 ++++++++++++++++++++++++++++++ internal/backup/snapshot_sst.go | 230 ++++++++++++++++++++++ 3 files changed, 605 insertions(+) create mode 100644 internal/backup/latest_version.go create mode 100644 internal/backup/snapshot_file.go create mode 100644 internal/backup/snapshot_sst.go diff --git a/internal/backup/latest_version.go b/internal/backup/latest_version.go new file mode 100644 index 000000000..061be3689 --- /dev/null +++ b/internal/backup/latest_version.go @@ -0,0 +1,61 @@ +package backup + +import ( + "bytes" + "encoding/binary" + + cockroachdberr "github.com/cockroachdb/errors" +) + +// ErrSnapshotUnsorted is returned when a snapshot's entries are not in +// the store's key order, which the newest-version filter depends on. +var ErrSnapshotUnsorted = cockroachdberr.New("backup: snapshot entries are not in key order") + +// latestVersionFilter keeps the newest version of each user key and +// hides the older ones, which is what a logical dump wants from a live +// store's snapshot: the store streams every physical MVCC version +// (writePebbleSnapshotEntries, and the SST ingest export likewise), while +// an encoder-produced snapshot has one version per key. +// +// The store encodes a version as userKey || BE(^commitTS) and orders keys +// bytewise, so a key's versions arrive newest first. They need not be +// adjacent: another user key that extends this one (userKey || x) can +// sort between two of its versions. Every encoded key that sorts between +// two versions of userKey starts with userKey, so the filter keeps a key +// "open" while the current encoded key still starts with it, and an entry +// whose user key is already open is an older version. The open set holds +// only user keys that are prefixes of the current encoded key, so it stays +// as small as the deepest such nesting. +type latestVersionFilter struct { + open [][]byte + enc []byte + previous []byte +} + +// keep reports whether e is the newest version of its user key. It fails +// with ErrSnapshotUnsorted when e does not sort after the previous entry. +func (f *latestVersionFilter) keep(e SnapshotEntry) (bool, error) { + f.enc = append(f.enc[:0], e.UserKey...) + f.enc = binary.BigEndian.AppendUint64(f.enc, ^e.CommitTS) + if f.previous != nil && bytes.Compare(f.enc, f.previous) <= 0 { + return false, cockroachdberr.Wrapf(ErrSnapshotUnsorted, "key %q at commit_ts %d", e.UserKey, e.CommitTS) + } + f.previous = append(f.previous[:0], f.enc...) + + newest := true + open := f.open[:0] + for _, k := range f.open { + if !bytes.HasPrefix(f.enc, k) { + continue // every version of k is behind us + } + if bytes.Equal(k, e.UserKey) { + newest = false + } + open = append(open, k) + } + f.open = open + if newest { + f.open = append(f.open, bytes.Clone(e.UserKey)) + } + return newest, nil +} diff --git a/internal/backup/snapshot_file.go b/internal/backup/snapshot_file.go new file mode 100644 index 000000000..b3489821d --- /dev/null +++ b/internal/backup/snapshot_file.go @@ -0,0 +1,314 @@ +package backup + +import ( + "bufio" + "bytes" + "encoding/binary" + "hash" + "hash/crc32" + "io" + + cockroachdberr "github.com/cockroachdb/errors" +) + +// snapshot_file.go reads a snapshot in every form it takes on disk or in +// transit and hands the store payload to the reader for its format: +// +// file a node persists, fsm-snap/.fsm (internal/raftengine/etcd +// writeFSMSnapshotPayload): +// [FSM snapshot header: EKVTHLC1 | EKVTHLC2] kv/snapshot.go +// [store payload: EKVPBBL1 stream | EKVSSTI1 bundle] store/ +// [CRC32C footer: 4 bytes BE, Castagnoli, over everything before it] +// +// encoder output, or a store payload taken from a running process: +// [store payload: EKVPBBL1 stream | EKVSSTI1 bundle] +// [optional CRC32C footer] (a headerless payload persisted by an +// older node carries one) +// +// The constants mirror kv/snapshot.go and +// internal/raftengine/etcd/fsm_snapshot_file.go; they are duplicated +// here, like the store constants in snapshot_reader.go, so this package +// links no live-cluster machinery. The CLI test that persists a real +// FSM snapshot through the engine's writer pins the mirror. + +const ( + // fsmHeaderMagicPrefix is the common prefix of every FSM snapshot + // header magic (kv/snapshot.go hlcSnapshotMagicPrefix). + fsmHeaderMagicPrefix = "EKVTHLC" + // FSMSnapshotHeaderV1 / FSMSnapshotHeaderV2 are the FSM snapshot + // header magics kv/snapshot.go writes in front of the store payload. + FSMSnapshotHeaderV1 = "EKVTHLC1" + FSMSnapshotHeaderV2 = "EKVTHLC2" + // fsmHeaderV1Len is the v1 header: magic(8) + ceiling ms(8). + fsmHeaderV1Len = 16 + // fsmHeaderV2LenBytes is the v2 header's BE uint16 payload length, + // after the magic; the payload holds ceiling ms(8) + cutover(8) and + // optional trailing extension bytes, bounded like the live reader. + fsmHeaderV2LenBytes = 2 + fsmHeaderV2MinPayload = 16 + fsmHeaderV2MaxPayload = 1024 + // fsmFooterSize is the engine's CRC32C footer. + fsmFooterSize = 4 + // snapshotMagicLen is the length of every magic above. + snapshotMagicLen = 8 + // fsmFooterChunk sizes the footer reader's refill buffer. + fsmFooterChunk = 64 << 10 +) + +// SSTIngestSnapshotMagic introduces the SST ingest bundle a live store +// emits when ELASTICKV_PEBBLE_SST_INGEST_SNAPSHOT is on +// (store/snapshot_pebble_sst.go pebbleSSTIngestSnapshotMagic). +const SSTIngestSnapshotMagic = "EKVSSTI1" + +var fsmFooterTable = crc32.MakeTable(crc32.Castagnoli) + +// ErrSnapshotFSMHeaderUnknown is returned for an EKVTHLC* magic whose +// version byte this decoder does not know. Like the live restore path, the +// decoder fails closed rather than guess at the header's length. +var ErrSnapshotFSMHeaderUnknown = cockroachdberr.New("backup: snapshot carries an FSM header version this decoder does not know") + +// ErrSnapshotFSMHeaderInvalid is returned for an EKVTHLC2 header whose +// length prefix is outside the bounds the live reader accepts. +var ErrSnapshotFSMHeaderInvalid = cockroachdberr.New("backup: snapshot FSM header has an invalid length prefix") + +// ErrSnapshotFooterMismatch is returned when a file framed by an FSM +// snapshot header does not end with a CRC32C footer matching the bytes +// before it: the file is truncated or corrupt. +var ErrSnapshotFooterMismatch = cockroachdberr.New("backup: snapshot CRC32C footer does not match its payload") + +// ErrSnapshotTrailingBytes is returned when bytes follow a store payload +// whose format defines its own end (the SST ingest bundle). +var ErrSnapshotTrailingBytes = cockroachdberr.New("backup: snapshot has trailing bytes after its store payload") + +// SnapshotFraming records how ReadSnapshotFile found the input framed. +type SnapshotFraming struct { + // FSMHeader is FSMSnapshotHeaderV1 or FSMSnapshotHeaderV2 for a file a + // node persisted, and empty for an input that starts at the store + // payload. + FSMHeader string + // StoreFormat is PebbleSnapshotMagic or SSTIngestSnapshotMagic. + StoreFormat string + // FooterVerified reports that the input ended with a CRC32C footer + // matching the bytes before it; FooterCRC32C is its value. + FooterVerified bool + FooterCRC32C uint32 +} + +// ReadSnapshotFile reads a snapshot in any of the forms described at the +// top of this file and yields every physical entry through fn, in key +// order: every MVCC version of a key, newest first, tombstones included, +// with the store's reserved metadata keys skipped. Callers that want one +// record per key (the logical dump) filter the versions themselves; see +// DecodeSnapshot. +// +// fn receives transient byte slices, as with ReadSnapshot. +func ReadSnapshotFile(r io.Reader, fn func(SnapshotHeader, SnapshotEntry) error) (SnapshotHeader, SnapshotFraming, error) { + var magic [snapshotMagicLen]byte + if _, err := io.ReadFull(r, magic[:]); err != nil { + return SnapshotHeader{}, SnapshotFraming{}, cockroachdberr.WithStack(err) + } + var framing SnapshotFraming + fsmHeader, err := classifySnapshotMagic(magic[:]) + if err != nil { + return SnapshotHeader{}, framing, err + } + framing.FSMHeader = fsmHeader + + // A node always closes a file framed by an FSM header with the + // footer; an input that starts at the store payload may carry one. + footer := newFooterReader(io.MultiReader(bytes.NewReader(magic[:]), r), fsmHeader != "") + br := bufio.NewReader(footer) + if framing.FSMHeader != "" { + if err := skipFSMSnapshotHeader(br, framing.FSMHeader); err != nil { + return SnapshotHeader{}, framing, err + } + } + header, storeFormat, err := readStorePayload(br, fn) + framing.StoreFormat = storeFormat + if err != nil { + return header, framing, err + } + framing.FooterVerified = footer.verified + framing.FooterCRC32C = footer.sum + return header, framing, nil +} + +// classifySnapshotMagic returns the FSM header magic when the input is a +// file a node persisted, "" when it starts at the store payload, and an +// error for anything else. +func classifySnapshotMagic(magic []byte) (string, error) { + switch m := string(magic); { + case m == FSMSnapshotHeaderV1 || m == FSMSnapshotHeaderV2: + return m, nil + case m == PebbleSnapshotMagic || m == SSTIngestSnapshotMagic: + return "", nil + case bytes.HasPrefix(magic, []byte(fsmHeaderMagicPrefix)): + return "", cockroachdberr.Wrapf(ErrSnapshotFSMHeaderUnknown, "got %q", magic) + default: + return "", cockroachdberr.Wrapf(ErrSnapshotBadMagic, + "got %q, want %s, %s, %s, or %s", magic, + FSMSnapshotHeaderV1, FSMSnapshotHeaderV2, PebbleSnapshotMagic, SSTIngestSnapshotMagic) + } +} + +// readStorePayload dispatches on the store payload's magic. +func readStorePayload(br *bufio.Reader, fn func(SnapshotHeader, SnapshotEntry) error) (SnapshotHeader, string, error) { + magic, err := br.Peek(snapshotMagicLen) + if err != nil { + return SnapshotHeader{}, "", cockroachdberr.WithStack(err) + } + switch string(magic) { + case PebbleSnapshotMagic: + header, err := readPebbleStream(br, fn) + return header, PebbleSnapshotMagic, err + case SSTIngestSnapshotMagic: + header, err := readSSTIngestBundle(br, fn) + if err != nil { + return header, SSTIngestSnapshotMagic, err + } + return header, SSTIngestSnapshotMagic, expectEndOfPayload(br) + default: + return SnapshotHeader{}, "", cockroachdberr.Wrapf(ErrSnapshotBadMagic, + "store payload magic %q, want %s or %s", magic, PebbleSnapshotMagic, SSTIngestSnapshotMagic) + } +} + +// expectEndOfPayload fails when bytes remain after a self-delimiting store +// payload. The footer reader has already withheld and checked the footer, +// so anything left is not part of the snapshot. +func expectEndOfPayload(br *bufio.Reader) error { + if _, err := br.ReadByte(); err == nil { + return cockroachdberr.WithStack(ErrSnapshotTrailingBytes) + } else if !cockroachdberr.Is(err, io.EOF) { + return cockroachdberr.WithStack(err) + } + return nil +} + +// skipFSMSnapshotHeader consumes the FSM snapshot header. Its fields (the +// HLC ceiling and the Raft envelope cutover index) matter to a restoring +// node, not to the logical dump. +func skipFSMSnapshotHeader(br *bufio.Reader, magic string) error { + if magic == FSMSnapshotHeaderV1 { + if _, err := br.Discard(fsmHeaderV1Len); err != nil { + return cockroachdberr.Wrap(truncatedIfEOF(err), "FSM header v1") + } + return nil + } + if _, err := br.Discard(snapshotMagicLen); err != nil { + return cockroachdberr.Wrap(truncatedIfEOF(err), "FSM header v2 magic") + } + var lenBuf [fsmHeaderV2LenBytes]byte + if _, err := io.ReadFull(br, lenBuf[:]); err != nil { + return cockroachdberr.Wrap(truncatedIfEOF(err), "FSM header v2 length") + } + payloadLen := int(binary.BigEndian.Uint16(lenBuf[:])) + if payloadLen < fsmHeaderV2MinPayload || payloadLen > fsmHeaderV2MaxPayload { + return cockroachdberr.Wrapf(ErrSnapshotFSMHeaderInvalid, "payload length %d not in [%d, %d]", + payloadLen, fsmHeaderV2MinPayload, fsmHeaderV2MaxPayload) + } + if _, err := br.Discard(payloadLen); err != nil { + return cockroachdberr.Wrap(truncatedIfEOF(err), "FSM header v2 payload") + } + return nil +} + +func truncatedIfEOF(err error) error { + if cockroachdberr.Is(err, io.EOF) || cockroachdberr.Is(err, io.ErrUnexpectedEOF) { + return cockroachdberr.WithStack(ErrSnapshotTruncated) + } + return cockroachdberr.WithStack(err) +} + +// footerReader passes its source through while withholding the last +// fsmFooterSize bytes, so the consumer never sees the engine's CRC32C +// footer. At the source's end it compares the withheld bytes with the +// CRC32C of everything before them. When they match, the footer is +// verified and dropped. When they do not, a required footer (a file +// framed by an FSM header, which the engine always closes with one) is +// ErrSnapshotFooterMismatch, while an optional one (an input that starts +// at the store payload, such as encoder output) means there was no +// footer, and the withheld bytes are released as data. An optional +// footer is therefore misread only when the last four bytes of a +// footer-less payload happen to equal the CRC32C of the rest; the payload +// then reads as truncated, never as different data. +type footerReader struct { + src io.Reader + required bool + crc hash.Hash32 + buf []byte // bytes read from src and not yet returned + srcDone bool + release bool // the withheld bytes turned out to be data + verified bool + sum uint32 + err error +} + +func newFooterReader(src io.Reader, required bool) *footerReader { + return &footerReader{src: src, required: required, crc: crc32.New(fsmFooterTable)} +} + +func (f *footerReader) Read(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + for { + if f.err != nil { + return 0, f.err + } + withhold := fsmFooterSize + if f.release { + withhold = 0 + } + if avail := len(f.buf) - withhold; avail > 0 { + n := copy(p, f.buf[:min(avail, len(p))]) + if !f.release { + _, _ = f.crc.Write(p[:n]) + } + f.buf = f.buf[n:] + return n, nil + } + if f.srcDone { + f.finish() + continue + } + f.fill() + } +} + +// fill appends the next chunk of the source to buf. +func (f *footerReader) fill() { + if cap(f.buf)-len(f.buf) < fsmFooterChunk { + grown := make([]byte, len(f.buf), len(f.buf)+fsmFooterChunk) + copy(grown, f.buf) + f.buf = grown + } + n, err := f.src.Read(f.buf[len(f.buf):cap(f.buf)]) + f.buf = f.buf[:len(f.buf)+n] + switch { + case cockroachdberr.Is(err, io.EOF): + f.srcDone = true + case err != nil: + f.err = cockroachdberr.WithStack(err) + } +} + +// finish decides what the withheld bytes were once the source has ended. +func (f *footerReader) finish() { + if f.release { + f.err = io.EOF + return + } + if len(f.buf) == fsmFooterSize && binary.BigEndian.Uint32(f.buf) == f.crc.Sum32() { + f.verified = true + f.sum = binary.BigEndian.Uint32(f.buf) + f.buf = nil + f.err = io.EOF + return + } + if f.required { + f.err = cockroachdberr.Wrapf(ErrSnapshotFooterMismatch, "computed %08x over the payload", f.crc.Sum32()) + return + } + f.release = true +} diff --git a/internal/backup/snapshot_sst.go b/internal/backup/snapshot_sst.go new file mode 100644 index 000000000..a3f68281a --- /dev/null +++ b/internal/backup/snapshot_sst.go @@ -0,0 +1,230 @@ +package backup + +import ( + "bufio" + "bytes" + "crypto/sha256" + "encoding/binary" + "encoding/hex" + "encoding/json" + "io" + + cockroachdberr "github.com/cockroachdb/errors" + "github.com/cockroachdb/pebble/v2/sstable" +) + +// snapshot_sst.go reads the SST ingest bundle a live store emits when +// ELASTICKV_PEBBLE_SST_INGEST_SNAPSHOT is on +// (store/snapshot_pebble_sst.go pebbleSSTIngestBundle.WriteTo): +// +// [8 bytes] magic "EKVSSTI1" +// [8 bytes] manifest length (BE int64) +// [32 bytes] SHA-256 of the manifest +// [n bytes] manifest JSON: version, last_commit_ts, files[] +// repeated per manifest file, in order: +// [size bytes] one SST file written with Pebble's default comparer +// +// The SSTs hold the same physical keys and values as the EKVPBBL1 stream +// (the store's reserved metadata keys included), in key order across +// files, so each entry goes through emitRawEntry. The bounds mirror the +// store's; each file is buffered whole (the store's target file size is +// 64 MiB) and checked against its SHA-256 before it is parsed. + +const ( + sstIngestVersion = 1 + sstIngestDigestSize = sha256.Size + sstIngestManifestMaxBytes = 64 << 20 + sstIngestMaxFiles = 1 << 20 + // sstIngestMaxFileBytes bounds one buffered file: the store's 64 MiB + // target plus room for the entry that crosses it, up to the largest + // stored key and value. + sstIngestMaxFileBytes = 1 << 30 +) + +// ErrSnapshotSSTInvalid is returned for an SST ingest bundle whose +// manifest or files fail validation (digest, bounds, ordering, kinds). +var ErrSnapshotSSTInvalid = cockroachdberr.New("backup: invalid SST ingest snapshot") + +type sstIngestManifest struct { + Version int `json:"version"` + LastCommitTS uint64 `json:"last_commit_ts"` + Files []sstIngestFile `json:"files"` +} + +type sstIngestFile struct { + Name string `json:"name"` + Size int64 `json:"size"` + SHA256 string `json:"sha256"` + Smallest []byte `json:"smallest"` + Largest []byte `json:"largest"` +} + +// readSSTIngestBundle reads the bundle from br and yields every entry of +// every SST, in order. +func readSSTIngestBundle(br *bufio.Reader, fn func(SnapshotHeader, SnapshotEntry) error) (SnapshotHeader, error) { + manifest, err := readSSTIngestManifest(br) + if err != nil { + return SnapshotHeader{}, err + } + header := SnapshotHeader{LastCommitTS: manifest.LastCommitTS} + var previousLargest []byte + for i, file := range manifest.Files { + if err := validateSSTIngestFile(file, i, previousLargest); err != nil { + return header, err + } + data, err := readSSTIngestFile(br, file) + if err != nil { + return header, err + } + if err := emitSSTEntries(data, file, header, fn); err != nil { + return header, err + } + previousLargest = file.Largest + } + return header, nil +} + +func readSSTIngestManifest(r io.Reader) (sstIngestManifest, error) { + manifestJSON, err := readSSTIngestManifestJSON(r) + if err != nil { + return sstIngestManifest{}, err + } + var manifest sstIngestManifest + if err := json.Unmarshal(manifestJSON, &manifest); err != nil { + return sstIngestManifest{}, cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "manifest JSON: %v", err) + } + if manifest.Version != sstIngestVersion { + return sstIngestManifest{}, cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "manifest version %d, want %d", manifest.Version, sstIngestVersion) + } + if len(manifest.Files) > sstIngestMaxFiles { + return sstIngestManifest{}, cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "%d files exceed the limit %d", len(manifest.Files), sstIngestMaxFiles) + } + return manifest, nil +} + +// readSSTIngestManifestJSON reads the magic, the manifest length, and the +// manifest bytes, and checks them against the manifest's SHA-256. +func readSSTIngestManifestJSON(r io.Reader) ([]byte, error) { + var magic [snapshotMagicLen]byte + if _, err := io.ReadFull(r, magic[:]); err != nil { + return nil, truncatedIfEOF(err) + } + var manifestLen int64 + if err := binary.Read(r, binary.BigEndian, &manifestLen); err != nil { + return nil, truncatedIfEOF(err) + } + if manifestLen <= 0 || manifestLen > sstIngestManifestMaxBytes { + return nil, cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "manifest length %d", manifestLen) + } + var digest [sstIngestDigestSize]byte + if _, err := io.ReadFull(r, digest[:]); err != nil { + return nil, truncatedIfEOF(err) + } + manifestJSON := make([]byte, manifestLen) + if _, err := io.ReadFull(r, manifestJSON); err != nil { + return nil, truncatedIfEOF(err) + } + if sha256.Sum256(manifestJSON) != digest { + return nil, cockroachdberr.Wrap(ErrSnapshotSSTInvalid, "manifest SHA-256 mismatch") + } + return manifestJSON, nil +} + +// validateSSTIngestFile mirrors store's validateSSTIngestFileMeta: a +// positive size within the decoder's buffer bound, a SHA-256, ordered key +// bounds, and no overlap with the previous file, which is what makes the +// concatenated files one key-ordered stream. +func validateSSTIngestFile(file sstIngestFile, index int, previousLargest []byte) error { + if file.Size <= 0 || file.Size > sstIngestMaxFileBytes { + return cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "file %d (%s) size %d", index, file.Name, file.Size) + } + if len(file.Smallest) == 0 || len(file.Largest) == 0 || bytes.Compare(file.Smallest, file.Largest) > 0 { + return cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "file %d (%s) key bounds", index, file.Name) + } + if index > 0 && bytes.Compare(previousLargest, file.Smallest) >= 0 { + return cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "file %d (%s) overlaps the previous file", index, file.Name) + } + if digest, err := hex.DecodeString(file.SHA256); err != nil || len(digest) != sstIngestDigestSize { + return cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "file %d (%s) SHA-256 %q", index, file.Name, file.SHA256) + } + return nil +} + +func readSSTIngestFile(r io.Reader, file sstIngestFile) ([]byte, error) { + data := make([]byte, file.Size) + if _, err := io.ReadFull(r, data); err != nil { + return nil, cockroachdberr.Wrapf(truncatedIfEOF(err), "SST file %s", file.Name) + } + digest := sha256.Sum256(data) + if hex.EncodeToString(digest[:]) != file.SHA256 { + return nil, cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "file %s SHA-256 mismatch", file.Name) + } + return data, nil +} + +// emitSSTEntries yields every point entry of one SST. +func emitSSTEntries(data []byte, file sstIngestFile, header SnapshotHeader, fn func(SnapshotHeader, SnapshotEntry) error) error { + return withSSTIterator(data, file, func(iter sstable.Iterator) error { + var valBuf []byte + for kv := iter.First(); kv != nil; kv = iter.Next() { + value, err := sstEntryValue(kv.K, kv.Value, &valBuf, file) + if err != nil { + return err + } + if err := emitRawEntry(kv.K.UserKey, value, header, fn); err != nil { + return err + } + } + return cockroachdberr.WithStack(iter.Error()) + }) +} + +// sstEntryValue checks one SST entry and returns its value. The store +// writes only Set entries (pebbleSSTExporter.Add), so any other kind means +// the file is not one of the store's; the size bounds match the EKVPBBL1 +// reader's. valBuf is reused for values the reader copies out; a value +// that is not caller-owned points into the reader's block and must not be +// written to. +func sstEntryValue(key sstable.InternalKey, value func([]byte) ([]byte, bool, error), valBuf *[]byte, file sstIngestFile) ([]byte, error) { + if kind := key.Kind(); kind != sstable.InternalKeyKindSet { + return nil, cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "file %s holds a %s entry", file.Name, kind) + } + if len(key.UserKey) > MaxSnapshotEncodedKeySize { + return nil, cockroachdberr.Wrapf(ErrSnapshotKeyTooLarge, "length %d > %d", len(key.UserKey), MaxSnapshotEncodedKeySize) + } + v, callerOwned, err := value((*valBuf)[:0]) + if err != nil { + return nil, cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "value in %s: %v", file.Name, err) + } + if len(v) > MaxSnapshotEncodedValueSize { + return nil, cockroachdberr.Wrapf(ErrSnapshotValueTooLarge, "length %d > %d", len(v), MaxSnapshotEncodedValueSize) + } + if callerOwned { + *valBuf = v + } + return v, nil +} + +// withSSTIterator opens an in-memory reader and a point iterator over one +// SST, runs body, and closes both. +func withSSTIterator(data []byte, file sstIngestFile, body func(sstable.Iterator) error) (retErr error) { + reader, err := sstable.NewMemReader(data, sstable.ReaderOptions{}) + if err != nil { + return cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "open %s: %v", file.Name, err) + } + defer func() { + if err := reader.Close(); err != nil && retErr == nil { + retErr = cockroachdberr.WithStack(err) + } + }() + iter, err := reader.NewIter(sstable.NoTransforms, nil, nil, sstable.TableBlobContext{}) + if err != nil { + return cockroachdberr.Wrapf(ErrSnapshotSSTInvalid, "iterate %s: %v", file.Name, err) + } + defer func() { + if err := iter.Close(); err != nil && retErr == nil { + retErr = cockroachdberr.WithStack(err) + } + }() + return body(iter) +} From bfd7ff9cb5829f336f59bed54be25457e3cbefdf Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Fri, 16 Oct 2026 16:08:38 +0900 Subject: [PATCH 5/6] backup: decode the newest version of each key and parse hex .fsm names DecodeSnapshot now reads through ReadSnapshotFile, so it accepts the fsm-snap/.fsm file a node persists and the SST ingest bundle, and filters versions through latestVersionFilter: a key's older MVCC versions count as Superseded instead of overwriting the newest one. DecodeResult carries the input's framing, including the CRC32C footer the reader verified. SnapshotIndexFromPath parses the live writer's %016x stem as hex; any other stem still parses as decimal. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01S5GoEpeK4aYNkUxWxutJTz --- internal/backup/decode.go | 61 +++++++++++++++++++++++++-------------- internal/backup/source.go | 49 ++++++++++++++++++++++++------- 2 files changed, 78 insertions(+), 32 deletions(-) diff --git a/internal/backup/decode.go b/internal/backup/decode.go index 3c4014076..2cf42fc35 100644 --- a/internal/backup/decode.go +++ b/internal/backup/decode.go @@ -107,7 +107,10 @@ type DecodeOptions struct { // The breakdown is: // // Total - every snapshot entry that reached the dispatcher -// Tombstone - entries whose value-header flagged a delete; skipped +// Superseded - older MVCC versions hidden by a newer version of the +// same key; a live store's snapshot streams every +// version, an encoder-produced one has none +// Tombstone - keys whose newest version is a delete; skipped // before adapter dispatch // DynamoDB - entries routed to a DDBEncoder method // S3 - entries routed to an S3Encoder method @@ -121,34 +124,40 @@ type DecodeOptions struct { // so a malformed or version-skewed snapshot does not // silently round-trip into an empty dump // -// Total = Tombstone + DynamoDB + S3 + Redis + SQS + Internal + Unknown. +// Total = Superseded + Tombstone + DynamoDB + S3 + Redis + SQS + +// Internal + Unknown. type DecodeCounters struct { - Total uint64 - Tombstone uint64 - DynamoDB uint64 - S3 uint64 - Redis uint64 - SQS uint64 - Internal uint64 - Unknown uint64 + Total uint64 + Superseded uint64 + Tombstone uint64 + DynamoDB uint64 + S3 uint64 + Redis uint64 + SQS uint64 + Internal uint64 + Unknown uint64 } // DecodeResult is returned by DecodeSnapshot after a successful // run. Header carries the snapshot's `last_commit_ts` for the -// caller's MANIFEST.json. +// caller's MANIFEST.json; Framing records how the input was framed, +// including the CRC32C footer the reader verified, if any. type DecodeResult struct { Header SnapshotHeader Counters DecodeCounters + Framing SnapshotFraming } // ErrDecodeOptionsInvalid is returned when DecodeSnapshot is called // with an under-populated DecodeOptions (missing OutRoot, etc.). var ErrDecodeOptionsInvalid = errors.New("backup: DecodeOptions invalid") -// DecodeSnapshot reads a `.fsm`-format stream from r, dispatches -// every non-tombstone entry to the appropriate adapter encoder -// rooted under opts.OutRoot, and runs Finalize on each enabled -// adapter at end of stream. +// DecodeSnapshot reads a snapshot from r in any form ReadSnapshotFile +// accepts (a `fsm-snap/.fsm` file a node persisted, an encoder- +// produced stream, or an SST ingest bundle), keeps the newest version of +// each key, dispatches every key whose newest version is not a delete to +// the appropriate adapter encoder rooted under opts.OutRoot, and runs +// Finalize on each enabled adapter at end of stream. // // Caller closes r. DecodeSnapshot does NOT write MANIFEST.json or // CHECKSUMS — both are emitted by the cmd/ wrapper from the returned @@ -158,7 +167,17 @@ func DecodeSnapshot(r io.Reader, opts DecodeOptions) (DecodeResult, error) { if err != nil { return DecodeResult{}, err } - hdr, err := ReadSnapshotWithHeader(r, func(_ SnapshotHeader, e SnapshotEntry) error { + var versions latestVersionFilter + hdr, framing, err := ReadSnapshotFile(r, func(_ SnapshotHeader, e SnapshotEntry) error { + newest, err := versions.keep(e) + if err != nil { + return err + } + if !newest { + d.counters.Total++ + d.counters.Superseded++ + return nil + } return d.handleEntry(e) }) // finalize() runs unconditionally so encoders (notably the S3 @@ -174,7 +193,7 @@ func DecodeSnapshot(r io.Reader, opts DecodeOptions) (DecodeResult, error) { if err != nil { return DecodeResult{}, err } - return DecodeResult{Header: hdr, Counters: d.counters}, nil + return DecodeResult{Header: hdr, Counters: d.counters, Framing: framing}, nil } // dispatcher is the in-memory state carried across the ReadSnapshot @@ -271,10 +290,10 @@ func newS3DecodeEncoder(opts DecodeOptions) (*S3Encoder, error) { WithWarnSink(opts.WarnSink), nil } -// handleEntry is the per-entry hook ReadSnapshot calls. Tombstones -// are counted and dropped before any prefix matching runs — Phase 0a -// dumps reflect the *current* user-visible state, not the -// crash-consistent MVCC tombstone fan. +// handleEntry is the per-key hook DecodeSnapshot calls with each key's +// newest version. Tombstones are counted and dropped before any prefix +// matching runs — Phase 0a dumps reflect the *current* user-visible +// state, not the crash-consistent MVCC tombstone fan. func (d *dispatcher) handleEntry(e SnapshotEntry) error { d.counters.Total++ if e.Tombstone { diff --git a/internal/backup/source.go b/internal/backup/source.go index 0dfb760c7..79117a134 100644 --- a/internal/backup/source.go +++ b/internal/backup/source.go @@ -14,18 +14,19 @@ import ( // the snapshot filename convention surfaces in one place. // SnapshotIndexFromPath extracts the integer applied-index encoded -// in a `.fsm` filename. The live writer (internal/raftengine/etcd/ -// engine.go) names every snapshot `.fsm` where is -// the FSM's applied_index at the moment the snapshot was taken. -// Phase 0a's MANIFEST.json carries that value as `snapshot_index` -// so a restore-time operator knows how stale the dump is relative -// to the cluster's last activity. +// in a `.fsm` filename. The live writer (fsmSnapPath in +// internal/raftengine/etcd/fsm_snapshot_file.go) names every snapshot +// `%016x.fsm`: the FSM's applied_index at the moment the snapshot was +// taken, as 16 zero-padded lowercase hex digits. Phase 0a's +// MANIFEST.json carries that value as `snapshot_index` so a +// restore-time operator knows how stale the dump is relative to the +// cluster's last activity. // // The numeric portion is everything between the final path -// separator (or string start) and the `.fsm` suffix. Lengths and -// zero-padding are not enforced — the live writer happens to pad -// to 16 digits, but a hand-rolled snapshot named `42.fsm` decodes -// correctly too. +// separator (or string start) and the `.fsm` suffix. A stem of +// exactly 16 hex digits is the live writer's name and parses as hex +// (`0000000000000040.fsm` is index 64); any other stem parses as +// decimal, so a hand-rolled snapshot named `42.fsm` decodes as 42. // // Returns ErrSnapshotIndexUnparseable when path lacks the .fsm // suffix or when the basename is not a parseable uint64. @@ -41,7 +42,11 @@ func SnapshotIndexFromPath(path string) (uint64, error) { return 0, errors.Wrapf(ErrSnapshotIndexUnparseable, "path %q: empty stem before %q", path, fsmSuffix) } - idx, err := strconv.ParseUint(stem, 10, 64) + numBase := decimalBase + if isLiveWriterFSMStem(stem) { + numBase = hexBase + } + idx, err := strconv.ParseUint(stem, numBase, 64) if err != nil { return 0, errors.Wrapf(ErrSnapshotIndexUnparseable, "path %q: stem %q is not a uint64", path, stem) @@ -49,6 +54,28 @@ func SnapshotIndexFromPath(path string) (uint64, error) { return idx, nil } +// liveWriterFSMStemLen is the width of the live writer's `%016x` name; +// decimalBase and hexBase are the strconv bases for the two stem shapes. +const ( + liveWriterFSMStemLen = 16 + decimalBase = 10 + hexBase = 16 +) + +// isLiveWriterFSMStem reports whether stem has the live writer's shape: +// exactly 16 lowercase hex digits. +func isLiveWriterFSMStem(stem string) bool { + if len(stem) != liveWriterFSMStemLen { + return false + } + for _, c := range stem { + if (c < '0' || c > '9') && (c < 'a' || c > 'f') { + return false + } + } + return true +} + // ErrSnapshotIndexUnparseable is returned by SnapshotIndexFromPath // when the input filename does not match the live writer's // `.fsm` convention. The Phase 0a CLI surfaces this as a From c5b0889ccba48347a13927ec742bf07b8c569ea7 Mon Sep 17 00:00:00 2001 From: "Yoshiaki Ueda (bootjp)" Date: Fri, 16 Oct 2026 16:11:50 +0900 Subject: [PATCH 6/6] cmd/elastickv-snapshot-decode: report framing and stamp the verified footer The CLI logs the superseded count and how the input was framed, and stamps source.fsm_crc32c in MANIFEST.json from the CRC32C footer the reader verified. Adds unit and property tests for the framing reader, the SST ingest bundle's integrity checks, the newest-version filter, and the Superseded counter. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01S5GoEpeK4aYNkUxWxutJTz --- cmd/elastickv-snapshot-decode/main.go | 21 ++- internal/backup/latest_version_test.go | 133 +++++++++++++++ internal/backup/snapshot_file_test.go | 217 +++++++++++++++++++++++++ 3 files changed, 366 insertions(+), 5 deletions(-) create mode 100644 internal/backup/latest_version_test.go create mode 100644 internal/backup/snapshot_file_test.go diff --git a/cmd/elastickv-snapshot-decode/main.go b/cmd/elastickv-snapshot-decode/main.go index 3c697584d..c58570f2f 100644 --- a/cmd/elastickv-snapshot-decode/main.go +++ b/cmd/elastickv-snapshot-decode/main.go @@ -2,9 +2,12 @@ // decoder described in // docs/design/2026_04_29_proposed_snapshot_logical_decoder.md. // -// It reads a Pebble snapshot (`.fsm`) emitted by the live FSM -// (store/lsm_store.go) and writes a vendor-independent per-adapter -// directory tree rooted at --output. The output tree carries a +// It reads a snapshot in any form backup.ReadSnapshotFile accepts: the +// `fsm-snap/.fsm` file a node persists (FSM snapshot header, +// store payload, CRC32C footer, which is verified), an encoder-produced +// stream, or a store's SST ingest bundle. It keeps the newest version +// of each key and writes a vendor-independent per-adapter directory +// tree rooted at --output. The output tree carries a // MANIFEST.json describing the dump and a sha256sum(1)-compatible // CHECKSUMS file the operator can verify with stock UNIX tooling // — `sha256sum -c CHECKSUMS` from the dump root suffices, with no @@ -241,10 +244,14 @@ func decodeOne(cfg *config, logger *slog.Logger) error { "s3", res.Counters.S3, "redis", res.Counters.Redis, "sqs", res.Counters.SQS, + "superseded", res.Counters.Superseded, "tombstone", res.Counters.Tombstone, "internal", res.Counters.Internal, "unknown", res.Counters.Unknown, - "last_commit_ts", res.Header.LastCommitTS) + "last_commit_ts", res.Header.LastCommitTS, + "fsm_header", res.Framing.FSMHeader, + "store_format", res.Framing.StoreFormat, + "footer_verified", res.Framing.FooterVerified) if err := emitManifest(cfg, res); err != nil { return err @@ -258,7 +265,8 @@ func decodeOne(cfg *config, logger *slog.Logger) error { // emitManifest writes MANIFEST.json under cfg.outputRoot. Source // metadata (FSM path and parsed snapshot_index) come from the CLI // invocation; last_commit_ts comes from the snapshot header the -// dispatcher surfaces. +// dispatcher surfaces, and source.fsm_crc32c from the CRC32C footer +// the reader verified, when the input carried one. func emitManifest(cfg *config, res backup.DecodeResult) error { m := backup.NewPhase0SnapshotManifest(time.Now()) m.ElastickvVersion = version @@ -268,6 +276,9 @@ func emitManifest(cfg *config, res backup.DecodeResult) error { m.SnapshotIndex = idx } m.Source = &backup.Source{FSMPath: cfg.inputPath} + if res.Framing.FooterVerified { + m.Source.FSMCRC32C = fmt.Sprintf("%08x", res.Framing.FooterCRC32C) + } m.Adapters = populateAdapterScopes(cfg.adapters, res) m.Exclusions = &backup.Exclusions{ IncludeIncompleteUploads: cfg.includeIncompleteUploads, diff --git a/internal/backup/latest_version_test.go b/internal/backup/latest_version_test.go new file mode 100644 index 000000000..1203210ed --- /dev/null +++ b/internal/backup/latest_version_test.go @@ -0,0 +1,133 @@ +package backup + +import ( + "bytes" + "encoding/binary" + "sort" + "testing" + + "github.com/cockroachdb/errors" + "pgregory.net/rapid" +) + +// versionedKey is one physical version as the store writes it. +type versionedKey struct { + userKey []byte + commitTS uint64 +} + +func (v versionedKey) encoded() []byte { + return binary.BigEndian.AppendUint64(bytes.Clone(v.userKey), ^v.commitTS) +} + +// sortedLikeStore orders versions the way a store snapshot streams them: +// bytewise over userKey || BE(^commitTS). +func sortedLikeStore(versions []versionedKey) []versionedKey { + out := append([]versionedKey(nil), versions...) + sort.Slice(out, func(i, j int) bool { return bytes.Compare(out[i].encoded(), out[j].encoded()) < 0 }) + return out +} + +// keptByFilter runs the filter over versions in store order and returns +// the commit timestamp it kept for each user key. +func keptByFilter(t *rapid.T, versions []versionedKey) map[string]uint64 { + var f latestVersionFilter + kept := map[string]uint64{} + for _, v := range sortedLikeStore(versions) { + newest, err := f.keep(SnapshotEntry{UserKey: v.userKey, CommitTS: v.commitTS}) + if err != nil { + t.Fatalf("keep(%q@%d): %v", v.userKey, v.commitTS, err) + } + if !newest { + continue + } + if prev, dup := kept[string(v.userKey)]; dup { + t.Fatalf("kept two versions of %q: %d and %d", v.userKey, prev, v.commitTS) + } + kept[string(v.userKey)] = v.commitTS + } + return kept +} + +// TestLatestVersionFilter_InterleavedVersions pins the case adjacency +// alone gets wrong: a user key that extends another one can sort between +// two versions of the shorter key, and the shorter key's older version +// must still be hidden. +func TestLatestVersionFilter_InterleavedVersions(t *testing.T) { + t.Parallel() + short := []byte("k") + // ^20 and ^10 begin with seven 0xFF bytes; a key "k" + 0xFF*7 + 0xF0 + // sorts between "k"||^20 (…0xEB) and "k"||^10 (…0xF5). + long := append([]byte("k"), 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xF0) + versions := sortedLikeStore([]versionedKey{ + {userKey: short, commitTS: 20}, + {userKey: short, commitTS: 10}, + {userKey: long, commitTS: 15}, + }) + if !bytes.Equal(versions[1].userKey, long) { + t.Fatalf("fixture does not interleave: order %q, %q, %q", versions[0].userKey, versions[1].userKey, versions[2].userKey) + } + var f latestVersionFilter + var got []bool + for _, v := range versions { + newest, err := f.keep(SnapshotEntry{UserKey: v.userKey, CommitTS: v.commitTS}) + if err != nil { + t.Fatalf("keep: %v", err) + } + got = append(got, newest) + } + if want := []bool{true, true, false}; got[0] != want[0] || got[1] != want[1] || got[2] != want[2] { + t.Fatalf("kept = %v, want %v (k@20, the long key, then k@10 hidden)", got, want) + } +} + +// TestLatestVersionFilter_KeepsNewestPerKey checks, over random key sets +// with shared prefixes and 0xFF bytes, that exactly the newest version of +// every user key survives. +func TestLatestVersionFilter_KeepsNewestPerKey(t *testing.T) { + t.Parallel() + rapid.Check(t, func(t *rapid.T) { + alphabet := []byte{'a', 'b', 0x00, 0xEB, 0xF0, 0xFE, 0xFF} + keyGen := rapid.SliceOfN(rapid.SampledFrom(alphabet), 1, 10) + keys := rapid.SliceOfNDistinct(keyGen, 1, 12, func(k []byte) string { return string(k) }).Draw(t, "keys") + var versions []versionedKey + want := map[string]uint64{} + for i, k := range keys { + tss := rapid.SliceOfNDistinct(rapid.Uint64Range(1, 40), 1, 4, rapid.ID[uint64]).Draw(t, "ts"+string(rune('0'+i))) + for _, ts := range tss { + versions = append(versions, versionedKey{userKey: k, commitTS: ts}) + if ts > want[string(k)] { + want[string(k)] = ts + } + } + } + got := keptByFilter(t, versions) + if len(got) != len(want) { + t.Fatalf("kept %d keys, want %d", len(got), len(want)) + } + for k, ts := range want { + if got[k] != ts { + t.Fatalf("key %q: kept commit_ts %d, want the newest %d", k, got[k], ts) + } + } + }) +} + +// TestLatestVersionFilter_RejectsUnsortedInput pins the fail-closed +// order check: out of order, the first version seen need not be the +// newest. +func TestLatestVersionFilter_RejectsUnsortedInput(t *testing.T) { + t.Parallel() + var f latestVersionFilter + if _, err := f.keep(SnapshotEntry{UserKey: []byte("b"), CommitTS: 1}); err != nil { + t.Fatalf("first entry: %v", err) + } + for _, e := range []SnapshotEntry{ + {UserKey: []byte("a"), CommitTS: 1}, // sorts before b + {UserKey: []byte("b"), CommitTS: 1}, // the same physical key again + } { + if _, err := f.keep(e); !errors.Is(err, ErrSnapshotUnsorted) { + t.Fatalf("keep(%q@%d): err = %v, want ErrSnapshotUnsorted", e.UserKey, e.CommitTS, err) + } + } +} diff --git a/internal/backup/snapshot_file_test.go b/internal/backup/snapshot_file_test.go new file mode 100644 index 000000000..4a2457489 --- /dev/null +++ b/internal/backup/snapshot_file_test.go @@ -0,0 +1,217 @@ +package backup + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "hash/crc32" + "io" + "testing" + "testing/iotest" + + "github.com/cockroachdb/errors" +) + +// withFooter appends the engine's CRC32C footer (BE, Castagnoli, over +// everything before it) to payload. +func withFooter(payload []byte) []byte { + sum := crc32.Checksum(payload, crc32.MakeTable(crc32.Castagnoli)) + return binary.BigEndian.AppendUint32(bytes.Clone(payload), sum) +} + +// fsmHeaderV1 is kv/snapshot.go's v1 header: magic + BE ceiling ms. +func fsmHeaderV1(ceilingMs uint64) []byte { + return binary.BigEndian.AppendUint64([]byte(FSMSnapshotHeaderV1), ceilingMs) +} + +// fsmHeaderV2 is kv/snapshot.go's v2 header with extLen bytes in its +// forward-compatible extension area. +func fsmHeaderV2(ceilingMs, cutover uint64, extLen uint16) []byte { + hdr := binary.BigEndian.AppendUint16([]byte(FSMSnapshotHeaderV2), fsmHeaderV2MinPayload+extLen) + hdr = binary.BigEndian.AppendUint64(hdr, ceilingMs) + hdr = binary.BigEndian.AppendUint64(hdr, cutover) + return append(hdr, bytes.Repeat([]byte{0xAB}, int(extLen))...) +} + +// redisStringStream is a one-entry EKVPBBL1 stream for !redis|str|. +func redisStringStream(t *testing.T, key, body string, commitTS uint64) []byte { + t.Helper() + var buf bytes.Buffer + buf.WriteString(PebbleSnapshotMagic) + mustBinaryWrite(t, &buf, commitTS) + encKey := binary.BigEndian.AppendUint64([]byte(RedisStringPrefix+key), ^commitTS) + mustBinaryWrite(t, &buf, uint64(len(encKey))) + buf.Write(encKey) + encVal := append(make([]byte, snapshotValueHeaderSize), liveRedisStringValue(body)...) + mustBinaryWrite(t, &buf, uint64(len(encVal))) + buf.Write(encVal) + return buf.Bytes() +} + +type readResult struct { + keys []string + framing SnapshotFraming + err error +} + +func readFile(input []byte, r func([]byte) io.Reader) readResult { + var res readResult + _, res.framing, res.err = ReadSnapshotFile(r(input), func(_ SnapshotHeader, e SnapshotEntry) error { + res.keys = append(res.keys, string(e.UserKey)) + return nil + }) + return res +} + +// TestReadSnapshotFile_Framings covers every accepted framing, each read +// through a whole-buffer reader and a one-byte reader so the footer +// reader's withholding is exercised at every boundary. +func TestReadSnapshotFile_Framings(t *testing.T) { + t.Parallel() + stream := redisStringStream(t, "greeting", "hello", 7) + cases := []struct { + name string + input []byte + wantHeader string + wantFooter bool + }{ + {"raw stream", stream, "", false}, + {"raw stream with footer", withFooter(stream), "", true}, + {"v1 header and footer", withFooter(append(fsmHeaderV1(1234), stream...)), FSMSnapshotHeaderV1, true}, + {"v2 header with extension and footer", withFooter(append(fsmHeaderV2(1234, 99, 6), stream...)), FSMSnapshotHeaderV2, true}, + } + readers := map[string]func([]byte) io.Reader{ + "whole": func(b []byte) io.Reader { return bytes.NewReader(b) }, + "one-byte": func(b []byte) io.Reader { return iotest.OneByteReader(bytes.NewReader(b)) }, + } + for _, tc := range cases { + for rname, r := range readers { + t.Run(tc.name+"/"+rname, func(t *testing.T) { + t.Parallel() + assertOneGreetingRead(t, tc.input, readFile(tc.input, r), tc.wantHeader, tc.wantFooter) + }) + } + } +} + +func assertOneGreetingRead(t *testing.T, input []byte, res readResult, wantHeader string, wantFooter bool) { + t.Helper() + if res.err != nil { + t.Fatalf("ReadSnapshotFile: %v", res.err) + } + if len(res.keys) != 1 || res.keys[0] != RedisStringPrefix+"greeting" { + t.Fatalf("keys = %q", res.keys) + } + f := res.framing + if f.FSMHeader != wantHeader || f.StoreFormat != PebbleSnapshotMagic || f.FooterVerified != wantFooter { + t.Fatalf("framing = %+v, want header %q, store %s, footer %v", f, wantHeader, PebbleSnapshotMagic, wantFooter) + } + if wantFooter && f.FooterCRC32C != binary.BigEndian.Uint32(input[len(input)-fsmFooterSize:]) { + t.Fatalf("FooterCRC32C = %08x, want the file's footer", f.FooterCRC32C) + } +} + +// TestReadSnapshotFile_Rejects covers the inputs the reader must refuse. +func TestReadSnapshotFile_Rejects(t *testing.T) { + t.Parallel() + stream := redisStringStream(t, "greeting", "hello", 7) + framed := withFooter(append(fsmHeaderV1(1234), stream...)) + corrupt := bytes.Clone(framed) + corrupt[len(corrupt)-1] ^= 0xFF + payloadFlip := bytes.Clone(framed) + payloadFlip[len(fsmHeaderV1(0))+len(stream)-1] ^= 0x01 + cases := []struct { + name string + input []byte + want error + }{ + {"framed file without footer", append(fsmHeaderV1(1234), stream...), ErrSnapshotFooterMismatch}, + {"footer bytes corrupted", corrupt, ErrSnapshotFooterMismatch}, + {"payload corrupted under a footer", payloadFlip, ErrSnapshotFooterMismatch}, + {"unknown FSM header version", append([]byte("EKVTHLC9"), stream...), ErrSnapshotFSMHeaderUnknown}, + {"v2 header length below minimum", withFooter(append(append([]byte(FSMSnapshotHeaderV2), 0, 8), stream...)), ErrSnapshotFSMHeaderInvalid}, + {"v2 header length above maximum", withFooter(append(append([]byte(FSMSnapshotHeaderV2), 0x04, 0x01), stream...)), ErrSnapshotFSMHeaderInvalid}, + {"unknown magic", []byte("NOTASNAP0000000000000000"), ErrSnapshotBadMagic}, + {"unknown store payload under a header", withFooter(append(fsmHeaderV1(1), []byte("NOTASNAP00000000")...)), ErrSnapshotBadMagic}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + res := readFile(tc.input, func(b []byte) io.Reader { return bytes.NewReader(b) }) + if !errors.Is(res.err, tc.want) { + t.Fatalf("err = %v, want %v", res.err, tc.want) + } + }) + } +} + +// TestReadSnapshotFile_SSTIngestBundleIntegrity corrupts a real SST ingest +// bundle in the places its own checks cover and adds a byte after it. +func TestReadSnapshotFile_SSTIngestBundleIntegrity(t *testing.T) { + t.Setenv(sstIngestSnapshotEnv, "true") + bundle := liveStoreSnapshotWithHistory(t) + if !bytes.HasPrefix(bundle, []byte(SSTIngestSnapshotMagic)) { + t.Fatalf("expected an %s bundle", SSTIngestSnapshotMagic) + } + // The manifest JSON follows magic(8), length(8), and its SHA-256; its + // length is that of the one JSON value there. + manifestStart := 16 + sstIngestDigestSize + var manifestJSON json.RawMessage + if err := json.NewDecoder(bytes.NewReader(bundle[manifestStart:])).Decode(&manifestJSON); err != nil { + t.Fatalf("locate manifest: %v", err) + } + manifestLen := len(manifestJSON) + firstFileByte := manifestStart + manifestLen + + flip := func(at int) []byte { + out := bytes.Clone(bundle) + out[at] ^= 0x01 + return out + } + cases := []struct { + name string + input []byte + want error + }{ + {"manifest byte flipped", flip(manifestStart + manifestLen/2), ErrSnapshotSSTInvalid}, + {"SST byte flipped", flip(firstFileByte + 1), ErrSnapshotSSTInvalid}, + {"trailing byte after the bundle", append(bytes.Clone(bundle), 0x00), ErrSnapshotTrailingBytes}, + {"framed bundle with a bad footer", append(append(fsmHeaderV1(1), bundle...), 0, 0, 0, 0), ErrSnapshotFooterMismatch}, + } + for _, tc := range cases { + res := readFile(tc.input, func(b []byte) io.Reader { return bytes.NewReader(b) }) + if !errors.Is(res.err, tc.want) { + t.Fatalf("%s: err = %v, want %v", tc.name, res.err, tc.want) + } + } + + res := readFile(withFooter(append(fsmHeaderV1(1), bundle...)), func(b []byte) io.Reader { return bytes.NewReader(b) }) + if res.err != nil { + t.Fatalf("framed bundle: %v", res.err) + } + if res.framing.StoreFormat != SSTIngestSnapshotMagic || !res.framing.FooterVerified { + t.Fatalf("framed bundle framing = %+v", res.framing) + } +} + +// TestDecodeSnapshot_CountsSupersededVersions pins the counters for the +// live history: the overwritten version of "updated" and the version under +// the delete of "deleted" are superseded, the delete itself is the one +// tombstone, and the invariant on DecodeCounters holds. +func TestDecodeSnapshot_CountsSupersededVersions(t *testing.T) { + t.Parallel() + res, err := DecodeSnapshot(bytes.NewReader(liveStoreSnapshotWithHistory(t)), DecodeOptions{ + OutRoot: t.TempDir(), + Adapters: AdapterSet{Redis: true}, + }) + if err != nil { + t.Fatalf("DecodeSnapshot: %v", err) + } + c := res.Counters + if c.Superseded != 2 || c.Tombstone != 1 || c.Redis != 2 { + t.Fatalf("counters = %+v, want superseded 2, tombstone 1, redis 2", c) + } + if sum := c.Superseded + c.Tombstone + c.DynamoDB + c.S3 + c.Redis + c.SQS + c.Internal + c.Unknown; sum != c.Total { + t.Fatalf("Total %d != sum of classes %d (%+v)", c.Total, sum, c) + } +}