backup: decode the .fsm files a running node persists

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

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/<index>.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,

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"))
 

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/<index>.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

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/<index>.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 {

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,