feat(snapshot-skip B2): plumb metaAppliedIndex through raft-Apply + both snapshot persist sites #915
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,148 @@ | ||
| package etcd | ||
|
|
||
| import ( | ||
| "io" | ||
| "sync" | ||
| "testing" | ||
|
|
||
| "github.com/bootjp/elastickv/internal/raftengine" | ||
| "github.com/coreos/go-semver/semver" | ||
| "github.com/stretchr/testify/require" | ||
| raftpb "go.etcd.io/raft/v3/raftpb" | ||
| ) | ||
|
|
||
| // applyIndexOrderRecorder is shared between the recording FSM and | ||
| // the recording persist storage so the test can assert the | ||
| // crash-ordering invariant (SetDurableAppliedIndex MUST run before | ||
| // persist.SaveSnap). Both record into a single ordered slice keyed | ||
| // by event kind; the test reads it back to verify the sequence. | ||
| type applyIndexOrderRecorder struct { | ||
| mu sync.Mutex | ||
| events []orderEvent | ||
| } | ||
|
|
||
| type orderEvent struct { | ||
| kind string // "bump" | "save" | ||
| index uint64 | ||
| } | ||
|
|
||
| func (r *applyIndexOrderRecorder) record(kind string, idx uint64) { | ||
| r.mu.Lock() | ||
| defer r.mu.Unlock() | ||
| r.events = append(r.events, orderEvent{kind: kind, index: idx}) | ||
| } | ||
|
|
||
| func (r *applyIndexOrderRecorder) snapshot() []orderEvent { | ||
| r.mu.Lock() | ||
| defer r.mu.Unlock() | ||
| out := make([]orderEvent, len(r.events)) | ||
| copy(out, r.events) | ||
| return out | ||
| } | ||
|
|
||
| // recordingAppliedIndexFSM implements StateMachine + | ||
| // raftengine.AppliedIndexWriter. It records every | ||
| // SetDurableAppliedIndex call into the shared recorder. | ||
| type recordingAppliedIndexFSM struct { | ||
| rec *applyIndexOrderRecorder | ||
| failNext bool | ||
| failErr error | ||
| } | ||
|
|
||
| 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 } | ||
|
|
||
| func (f *recordingAppliedIndexFSM) SetDurableAppliedIndex(idx uint64) error { | ||
| if f.failNext { | ||
| f.failNext = false | ||
| return f.failErr | ||
| } | ||
| f.rec.record("bump", idx) | ||
| return nil | ||
| } | ||
|
|
||
| // recordingPersistStorage is a minimal etcdstorage.Storage stand-in | ||
| // that records SaveSnap calls into the shared recorder. The hook | ||
| // only calls SaveSnap + Release; the rest are stubs. | ||
| type recordingPersistStorage struct { | ||
| rec *applyIndexOrderRecorder | ||
| } | ||
|
|
||
| func (p *recordingPersistStorage) SaveSnap(snap raftpb.Snapshot) error { | ||
| p.rec.record("save", snap.Metadata.Index) | ||
| return nil | ||
| } | ||
|
|
||
| func (p *recordingPersistStorage) Save(_ raftpb.HardState, _ []raftpb.Entry) error { return nil } | ||
| func (p *recordingPersistStorage) Release(_ raftpb.Snapshot) error { return nil } | ||
| func (p *recordingPersistStorage) Sync() error { return nil } | ||
| func (p *recordingPersistStorage) Close() error { return nil } | ||
| func (p *recordingPersistStorage) MinimalEtcdVersion() *semver.Version { return nil } | ||
|
|
||
| // 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 | ||
| // (snap.Metadata.Index) BEFORE SaveSnap. | ||
| func TestPersistCreatedSnapshot_BumpsAppliedIndex(t *testing.T) { | ||
| rec := &applyIndexOrderRecorder{} | ||
| fsm := &recordingAppliedIndexFSM{rec: rec} | ||
| persist := &recordingPersistStorage{rec: rec} | ||
| e := &Engine{fsm: fsm, persist: persist} | ||
|
|
||
| snap := raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 42, Term: 1}} | ||
| require.NoError(t, e.persistCreatedSnapshot(snap)) | ||
|
|
||
| require.Equal(t, []orderEvent{ | ||
| {kind: "bump", index: 42}, | ||
| {kind: "save", index: 42}, | ||
| }, rec.snapshot(), | ||
| "hook MUST call SetDurableAppliedIndex BEFORE SaveSnap") | ||
| } | ||
|
|
||
| // TestPersistCreatedSnapshot_NilFSMNoOp covers the legacy / test- | ||
| // fake case: an FSM that does NOT implement AppliedIndexWriter | ||
| // silently no-ops; snapshot persist still runs. | ||
| func TestPersistCreatedSnapshot_NilFSMNoOp(t *testing.T) { | ||
| rec := &applyIndexOrderRecorder{} | ||
| persist := &recordingPersistStorage{rec: rec} | ||
| // testStateMachine (defined in engine_test.go) is the canonical | ||
| // non-AppliedIndexWriter FSM used by other tests in this package. | ||
| e := &Engine{fsm: &testStateMachine{}, persist: persist} | ||
|
|
||
| snap := raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 17, Term: 1}} | ||
| require.NoError(t, e.persistCreatedSnapshot(snap)) | ||
|
|
||
| require.Equal(t, []orderEvent{ | ||
| {kind: "save", index: 17}, | ||
| }, rec.snapshot(), | ||
| "legacy FSM path: snapshot persist still happens, just without the meta-key bump") | ||
| } | ||
|
|
||
| // TestPersistCreatedSnapshot_BumpErrorAborts checks the ordering | ||
| // invariant under failure: if SetDurableAppliedIndex returns an | ||
| // error, the engine MUST surface it AND NOT call SaveSnap. This | ||
| // preserves the (metaAppliedIndex < snapshot pointer impossible) | ||
| // crash invariant from PR #910 design §6. | ||
| func TestPersistCreatedSnapshot_BumpErrorAborts(t *testing.T) { | ||
| rec := &applyIndexOrderRecorder{} | ||
| fsm := &recordingAppliedIndexFSM{rec: rec, failNext: true, failErr: io.ErrShortBuffer} | ||
| persist := &recordingPersistStorage{rec: rec} | ||
| e := &Engine{fsm: fsm, persist: persist} | ||
|
|
||
| snap := raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 99, Term: 1}} | ||
| err := e.persistCreatedSnapshot(snap) | ||
| require.Error(t, err, "bump failure MUST be surfaced to caller") | ||
| require.Empty(t, rec.snapshot(), | ||
| "failed bump MUST NOT have recorded; SaveSnap MUST NOT have run") | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When the config-snapshot path reaches this hook,
persistConfigSnapshotPayloadLockedhas already obtainedpayloadfromsnapshotPayload, which writes/closes the.fsmfile (or serializes the legacy bytes) before callingpersistCreatedSnapshot. Bumping_meta_applied_indexhere only updates the live store, so a peer or restart that restores this saved config snapshot gets the older/missing meta key while the raft snapshot is atsnap.Metadata.Index; after WAL entries up to that index are released, the advertisedLastAppliedIndex >= snapshot indexinvariant does not hold for the snapshot artifact. This is separate from the steady-state snapshot site: the same ordering occurs throughpersistConfigSnapshotPayloadLocked -> snapshotPayload -> persistCreatedSnapshot.Useful? React with 👍 / 👎.