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.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/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/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 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/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/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/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.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_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) + } +} 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/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 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) +} 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 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},