From 01ec486846db3d38dbb70e13aa69adc6c928a56d Mon Sep 17 00:00:00 2001 From: bootjp Date: Wed, 28 Oct 2026 15:24:31 +0900 Subject: [PATCH] raftengine: test live config persistence and bootstrap guards Signed-off-by: bootjp --- internal/encryption/envelope_test.go | 8 ---- internal/encryption/fsmwire/wire_test.go | 20 ++++++--- internal/raftengine/etcd/engine.go | 21 ---------- .../etcd/engine_applied_index_test.go | 12 +----- internal/raftengine/etcd/engine_test.go | 41 +++++++++++-------- 5 files changed, 39 insertions(+), 63 deletions(-) diff --git a/internal/encryption/envelope_test.go b/internal/encryption/envelope_test.go index 49a550b6a..42deb6897 100644 --- a/internal/encryption/envelope_test.go +++ b/internal/encryption/envelope_test.go @@ -190,14 +190,6 @@ func TestEnvelope_Decode_DoesNotAliasInput(t *testing.T) { } } -func TestHeaderAADBytes_Layout(t *testing.T) { - got := encryption.HeaderAADBytes(encryption.EnvelopeVersionV2, encryption.FlagCompressed, 0x12345678) - want := []byte{0x02, 0x01, 0x12, 0x34, 0x56, 0x78} - if !bytes.Equal(got, want) { - t.Fatalf("HeaderAADBytes layout mismatch:\n got %x\n want %x", got, want) - } -} - func TestAppendHeaderAADBytes_AppendsAndAllocFree(t *testing.T) { // Append onto an existing buffer. The result should be the original // bytes followed by the 6-byte header. diff --git a/internal/encryption/fsmwire/wire_test.go b/internal/encryption/fsmwire/wire_test.go index 1454c797f..896757a40 100644 --- a/internal/encryption/fsmwire/wire_test.go +++ b/internal/encryption/fsmwire/wire_test.go @@ -2,6 +2,7 @@ package fsmwire_test import ( "bytes" + "strings" "testing" "github.com/bootjp/elastickv/internal/encryption/fsmwire" @@ -138,7 +139,7 @@ func TestBootstrap_RejectsTruncated(t *testing.T) { } } -// TestBootstrap_RejectsOverCapBatchCount is the codex P1 round-3 +// TestBootstrap_RejectsOverCapBatchCount is the P1 round-3 // regression for PR748: the wire-bytes check (count * // registrationSize ≤ remaining) is necessary but NOT sufficient // because RegistrationPayload occupies more bytes in memory (24 @@ -165,21 +166,24 @@ func TestBootstrap_RejectsOverCapBatchCount(t *testing.T) { t.Fatalf("encoded bootstrap shorter than expected: %d bytes, want >= %d", got, countOffset+4) } // Overwrite count with maxBootstrapBatchCount + 1 = 16385. - // The cap check fires before the wire-bytes check, so we do - // NOT need to extend the payload with 16385 * 14 trailing - // bytes — the decoder must reject on the cap regardless. + // Supply all registration bytes so the wire-bytes check cannot + // mask a missing cap check. good[countOffset+0] = 0x00 good[countOffset+1] = 0x00 good[countOffset+2] = 0x40 // 0x4001 = 16385 = (1<<14) + 1 good[countOffset+3] = 0x01 + good = append(good, make([]byte, 16385*(4+8+2))...) _, err := fsmwire.DecodeBootstrap(good) if !errors.Is(err, fsmwire.ErrFSMWireMalformed) { t.Fatalf("expected ErrFSMWireMalformed for over-cap batch count, got %v", err) } + if !strings.Contains(err.Error(), "bootstrap: batch count 16385 exceeds cap 16384") { + t.Fatalf("expected explicit batch-count cap error, got %v", err) + } } // TestBootstrap_RejectsHugeBatchCount is the regression for the -// codex P1 / gemini security-high finding: a malformed bootstrap +// security-high finding in PR748: a malformed bootstrap // payload that names `count = 0xffffffff` would, before the guard, // trigger `make([]RegistrationPayload, count)` (4G × 14 bytes ≈ // 56 GiB) and OOM the process before the loop tried to read the @@ -188,7 +192,8 @@ func TestBootstrap_RejectsOverCapBatchCount(t *testing.T) { func TestBootstrap_RejectsHugeBatchCount(t *testing.T) { t.Parallel() // Hand-roll a payload whose batch count is huge but which has - // only a tiny tail to serve registration rows from. We start + // no registration rows. The cap must take precedence over the + // remaining-bytes check without allocating a huge payload. We start // from the encoded form of a minimal valid bootstrap, then // rewrite the count prefix in place. good := fsmwire.EncodeBootstrap(fsmwire.BootstrapPayload{ @@ -211,6 +216,9 @@ func TestBootstrap_RejectsHugeBatchCount(t *testing.T) { if !errors.Is(err, fsmwire.ErrFSMWireMalformed) { t.Fatalf("expected ErrFSMWireMalformed for huge batch count, got %v", err) } + if !strings.Contains(err.Error(), "bootstrap: batch count 4294967295 exceeds cap 16384") { + t.Fatalf("expected explicit batch-count cap error, got %v", err) + } } func TestBootstrap_RejectsTrailingBytes(t *testing.T) { diff --git a/internal/raftengine/etcd/engine.go b/internal/raftengine/etcd/engine.go index cf5760f9c..31a25e423 100644 --- a/internal/raftengine/etcd/engine.go +++ b/internal/raftengine/etcd/engine.go @@ -3419,27 +3419,6 @@ func applyUpdatedPeerToMap(peers map[uint64]Peer, peer Peer, ok bool) { } } -func (e *Engine) persistConfigSnapshot(index uint64, confState raftpb.ConfState) error { - // Fast path (lock-free): avoid unnecessary FSM snapshot when config is already current. - if upToDate, err := e.configSnapshotUpToDate(index, confState); err != nil { - return err - } else if upToDate { - return nil - } - - e.snapshotMu.Lock() - defer e.snapshotMu.Unlock() - - payload, err := e.snapshotPayloadLocked(index) - if err != nil { - if snapshotDeferred(err, "config snapshot", index) { - return nil - } - return err - } - return e.persistConfigSnapshotPayloadLocked(index, confState, payload) -} - func (e *Engine) persistConfigState(index uint64, confState raftpb.ConfState, peers []Peer) error { // Config changes are committed in strictly increasing raft index order. Use // the cached durable config index to avoid re-reading peer metadata from diff --git a/internal/raftengine/etcd/engine_applied_index_test.go b/internal/raftengine/etcd/engine_applied_index_test.go index 107dcaefa..e6626e3ee 100644 --- a/internal/raftengine/etcd/engine_applied_index_test.go +++ b/internal/raftengine/etcd/engine_applied_index_test.go @@ -51,6 +51,8 @@ type recordingAppliedIndexFSM struct { failErr error } +var _ raftengine.AppliedIndexWriter = (*recordingAppliedIndexFSM)(nil) + func (f *recordingAppliedIndexFSM) Apply(_ []byte) any { return nil } func (f *recordingAppliedIndexFSM) Snapshot() (Snapshot, error) { return nil, io.EOF } func (f *recordingAppliedIndexFSM) Restore(_ io.Reader) error { return nil } @@ -337,16 +339,6 @@ func TestReleaseIgnoredReceivedFSMSnapshotStepsKeepsSnapshotReadyProtected(t *te require.FileExists(t, fsmSnapPath(fsmSnapDir, 10)) } -// TestRecordingFSM_SatisfiesAppliedIndexWriter is a compile-time- -// adjacent assertion: the recording FSM MUST satisfy the writer -// seam so the engine hook actually fires for it. -func TestRecordingFSM_SatisfiesAppliedIndexWriter(t *testing.T) { - rec := &applyIndexOrderRecorder{} - var f any = &recordingAppliedIndexFSM{rec: rec} - _, ok := f.(raftengine.AppliedIndexWriter) - require.True(t, ok, "recordingAppliedIndexFSM must implement raftengine.AppliedIndexWriter") -} - // TestPersistCreatedSnapshot_BumpsAppliedIndex exercises Site 1 of // the persist hook. We invoke (*Engine).persistCreatedSnapshot // directly; the engine MUST call SetDurableAppliedIndex diff --git a/internal/raftengine/etcd/engine_test.go b/internal/raftengine/etcd/engine_test.go index d09af4448..b3b582aa5 100644 --- a/internal/raftengine/etcd/engine_test.go +++ b/internal/raftengine/etcd/engine_test.go @@ -1204,7 +1204,7 @@ func TestMaybePersistLocalSnapshotReturnsWhenWorkerIsStopped(t *testing.T) { } } -func TestPersistConfigSnapshotSkipsSerializationWhenNewerSnapshotExists(t *testing.T) { +func TestPersistConfigStateSkipsSerializationWhenNewerSnapshotExists(t *testing.T) { storage := etcdraft.NewMemoryStorage() snap := raftTestSnapshot(9, 2, []uint64{1, 2}, []byte("newer")) require.NoError(t, storage.ApplySnapshot(&snap)) @@ -1213,13 +1213,21 @@ func TestPersistConfigSnapshotSkipsSerializationWhenNewerSnapshotExists(t *testi engine := &Engine{ storage: storage, fsm: fsm, + dataDir: t.TempDir(), } - require.NoError(t, engine.persistConfigSnapshot(7, raftpb.ConfState{Voters: []uint64{1}})) + peers := []Peer{{NodeID: 1, ID: "n1", Address: "127.0.0.1:7001", Suffrage: SuffrageVoter}} + require.NoError(t, engine.persistConfigState(7, raftpb.ConfState{Voters: []uint64{1}}, peers)) require.Zero(t, fsm.snapshotCalls.Load()) + state, exists, err := loadPersistedPeersState(engine.dataDir) + require.NoError(t, err) + require.True(t, exists, "live config path must publish peer metadata") + require.Equal(t, uint64(7), state.Index) + require.Equal(t, peers, state.Peers) + require.Equal(t, uint64(7), engine.currentConfigIndex()) } -func TestPersistConfigSnapshotDoesNotCompactLogs(t *testing.T) { +func TestPersistConfigStateDoesNotCompactLogs(t *testing.T) { storage := etcdraft.NewMemoryStorage() snap := raftTestSnapshot(1, 1, []uint64{1}, nil) require.NoError(t, storage.ApplySnapshot(&snap)) @@ -1233,16 +1241,20 @@ func TestPersistConfigSnapshotDoesNotCompactLogs(t *testing.T) { storage: storage, persist: mockstorage.NewStorageRecorder(""), fsm: &testStateMachine{}, + dataDir: t.TempDir(), } - require.NoError(t, engine.persistConfigSnapshot(4, raftpb.ConfState{Voters: []uint64{1, 2}})) + require.NoError(t, engine.persistConfigState(4, raftpb.ConfState{Voters: []uint64{1, 2}}, []Peer{ + {NodeID: 1, ID: "n1", Address: "127.0.0.1:7001"}, + {NodeID: 2, ID: "n2", Address: "127.0.0.1:7002"}, + })) firstIndex, err := storage.FirstIndex() require.NoError(t, err) require.Equal(t, uint64(2), firstIndex) } -func TestPersistConfigSnapshotWaitsForDurablePublication(t *testing.T) { +func TestPersistConfigStateWaitsForDurablePublication(t *testing.T) { storage := etcdraft.NewMemoryStorage() snap := raftTestSnapshot(1, 1, []uint64{1}, nil) require.NoError(t, storage.ApplySnapshot(&snap)) @@ -1270,7 +1282,10 @@ func TestPersistConfigSnapshotWaitsForDurablePublication(t *testing.T) { done := make(chan error, 1) go func() { - done <- engine.persistConfigSnapshot(4, raftpb.ConfState{Voters: []uint64{1, 2}}) + done <- engine.persistConfigState(4, raftpb.ConfState{Voters: []uint64{1, 2}}, []Peer{ + {NodeID: 1, ID: "n1", Address: "127.0.0.1:7001"}, + {NodeID: 2, ID: "n2", Address: "127.0.0.1:7002"}, + }) }() select { @@ -1282,7 +1297,7 @@ func TestPersistConfigSnapshotWaitsForDurablePublication(t *testing.T) { select { case err := <-done: require.NoError(t, err) - t.Fatal("persistConfigSnapshot returned before snapshot serialization completed") + t.Fatal("persistConfigState returned before snapshot serialization completed") case <-time.After(100 * time.Millisecond): } @@ -1293,7 +1308,7 @@ func TestPersistConfigSnapshotWaitsForDurablePublication(t *testing.T) { case err := <-done: require.NoError(t, err) case <-time.After(time.Second): - t.Fatal("persistConfigSnapshot did not finish after snapshot serialization was released") + t.Fatal("persistConfigState did not finish after snapshot serialization was released") } actions := persist.Action() @@ -1707,16 +1722,6 @@ func TestEnqueueDispatchMessagePreservesIncomingReadIndexHeartbeatWhenRespLaneFu require.Equal(t, []byte("read-index-c"), req.msg.Context) } -func TestMaxAppliedIndexStartsFromSnapshotIndex(t *testing.T) { - storage := etcdraft.NewMemoryStorage() - snap := raftTestSnapshot(5, 2, []uint64{1}, nil) - require.NoError(t, storage.ApplySnapshot(&snap)) - hs := testHardState(2, 9) - require.NoError(t, storage.SetHardState(&hs)) - - require.Equal(t, uint64(5), maxAppliedIndex(raftTestSnapshot(5, 0, nil, nil))) -} - func TestOpenRestoresLegacySnapshotState(t *testing.T) { dir := t.TempDir() require.NoError(t, saveStateFile(stateFilePath(dir), persistedState{