Repository: matrixorigin/matrixcube (go)

Issue:
Crash when applying config change entry after split
<!-- Please describe your issue in English. -->

#### Can be reproduced ?

Yes.

#### Steps:

Split, try to apply some update entries and then try to apply a config change entry.

#### Expected behavior:

complete successfully.

#### Actual behavior:

crash & complaining unexpected entry index.

#### Environment:
- Version or commit-id (e.g. v0.1.0 or 8b23a93):
- Hardware parameters:
- OS type:
- Others:

#### Configuration file:


#### Additional context:
- Error message from client:
- Server log:
- Other information:

Pull request:
allow entries to be skipped after split
**What type of PR is this?**

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

Fixes #422 

**What this PR does / why we need it:**

Not Available

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available


Patch:
diff --git a/raftstore/replica_destroy.go b/raftstore/replica_destroy.go
index 44e8609..3f8e4d9 100644
--- a/raftstore/replica_destroy.go
+++ b/raftstore/replica_destroy.go
@@ -101,6 +101,12 @@ func (s *store) cleanupTombstones(shards []Shard) {
 // vacuum is the actual method for handling a vacuum task.
 func (s *store) vacuum(t vacuumTask) error {
 	if t.replica != nil {
+		if t.replica.closed() {
+			// this can happen when PD request to remove a splitting shard. two vaccum
+			// tasks will be created, one for splitting and one for removal.
+			t.replica.logger.Info("skip vacuuming already closed replica")
+			return nil
+		}
 		if err := t.replica.destroy(t.shardRemoved, t.reason); err != nil {
 			return err
 		}
diff --git a/raftstore/replica_state_machine.go b/raftstore/replica_state_machine.go
index f562a60..bfeb6fa 100644
--- a/raftstore/replica_state_machine.go
+++ b/raftstore/replica_state_machine.go
@@ -146,6 +146,7 @@ func (d *stateMachine) applyCommittedEntries(entries []raftpb.Entry) {
 	// in the loop below, we are still applying entries one by one.
 	for _, entry := range entries {
 		d.applyCtx.initialize(entry)
+		d.checkEntryIndexTerm(entry)
 		// notify all clients that current shard has been removed or splitted
 		if !d.canApply(entry) {
 			if ce := d.logger.Check(zap.DebugLevel, "apply committed log skipped"); ce != nil {
@@ -154,9 +155,9 @@ func (d *stateMachine) applyCommittedEntries(entries []raftpb.Entry) {
 					log.ReasonField("continue check failed"))
 			}
 			d.notifyShardRemoved(d.applyCtx)
+			d.updateAppliedIndexTerm(entry.Index, entry.Term)
 			continue
 		}
-		d.checkEntryIndexTerm(entry)
 		if len(entry.Data) == 0 {
 			// noop entry with empty payload proposed by the leader at the beginning
 			// of its term


Test patch:
diff --git a/raftstore/replica_destroy_test.go b/raftstore/replica_destroy_test.go
index 0dc4d1d..e83961b 100644
--- a/raftstore/replica_destroy_test.go
+++ b/raftstore/replica_destroy_test.go
@@ -19,6 +19,7 @@ import (
 	"github.com/matrixorigin/matrixcube/components/log"
 	"github.com/matrixorigin/matrixcube/storage"
 	skv "github.com/matrixorigin/matrixcube/storage/kv"
+	"github.com/matrixorigin/matrixcube/util/leaktest"
 	"github.com/matrixorigin/matrixcube/util/stop"
 	"github.com/matrixorigin/matrixcube/util/task"
 	"github.com/stretchr/testify/assert"
@@ -26,8 +27,7 @@ import (
 )
 
 func TestDestroyReplica(t *testing.T) {
-	// FIXME: re-enable the leak test below
-	// defer leaktest.AfterTest(t)()
+	defer leaktest.AfterTest(t)()
 	r := Replica{ID: 1}
 	s := NewSingleTestClusterStore(t).GetStore(0).(*store)
 	kv := s.DataStorageByGroup(0).(storage.KVStorageWrapper).GetKVStorage()
@@ -83,6 +83,11 @@ func TestDestroyReplica(t *testing.T) {
 			break
 		}
 	}
+	// when repeatedly request the the same replica to be destroyed, such requests
+	// should become NOOPs.
+	for i := 0; i < 100; i++ {
+		s.destroyReplica(pr.shardID, true, true, "testing")
+	}
 	wc := s.logdb.NewWorkerContext()
 	pr.handleEvent(wc)
 	pr.waitDestroyed()
diff --git a/raftstore/replica_state_machine_test.go b/raftstore/replica_state_machine_test.go
index 0a9a285..6a2fd01 100644
--- a/raftstore/replica_state_machine_test.go
+++ b/raftstore/replica_state_machine_test.go
@@ -444,6 +444,53 @@ func TestStateMachineRejectsStaleEpochEntries(t *testing.T) {
 	runSimpleStateMachineTest(t, f, h)
 }
 
+func TestStateMachineUpdatesAppliedIndexAfterSkippingEntries(t *testing.T) {
+	h := &testReplicaResultHandler{}
+	f := func(sm *stateMachine) {
+		batch := rpc.RequestBatch{
+			Header: rpc.RequestBatchHeader{
+				ID: []byte{0x1, 0x2, 0x3},
+				// ShardID missing here, this will cause the epoch check to fail
+			},
+			AdminRequest: rpc.AdminRequest{
+				CmdType:      rpc.AdminCmdType_ConfigChange,
+				ConfigChange: &rpc.ConfigChangeRequest{},
+			},
+		}
+		cc := raftpb.ConfChange{
+			Type:    raftpb.ConfChangeAddNode,
+			NodeID:  100,
+			Context: protoc.MustMarshal(&batch),
+		}
+		entry1 := raftpb.Entry{
+			Index: 1,
+			Term:  1,
+			Type:  raftpb.EntryNormal,
+		}
+		entry2 := raftpb.Entry{
+			Index: 2,
+			Term:  1,
+			Type:  raftpb.EntryNormal,
+		}
+		entry3 := raftpb.Entry{
+			Index: 3,
+			Term:  1,
+			Type:  raftpb.EntryConfChange,
+			Data:  protoc.MustMarshal(&cc),
+		}
+
+		// mark as splitted
+		sm.metadataMu.splited = true
+		assert.False(t, sm.canApply(entry1))
+		assert.False(t, sm.canApply(entry2))
+		sm.applyCommittedEntries([]raftpb.Entry{entry1, entry2, entry3})
+		index, term := sm.getAppliedIndexTerm()
+		assert.Equal(t, uint64(3), index)
+		assert.Equal(t, uint64(1), term)
+	}
+	runSimpleStateMachineTest(t, f, h)
+}
+
 func TestStateMachineApplyCommittedEntriesAllowEmptyInput(t *testing.T) {
 	h := &testReplicaResultHandler{}
 	f := func(sm *stateMachine) {