Repository: matrixorigin/matrixcube (go)

Issue:
Crash when trying to close the replica twice
<!-- Please describe your issue in English. -->

#### Can be reproduced ?

Yes

#### Steps:

Run all tests with race enabled, on a macbook pro, it is pretty easy to reproduce it. 

#### Expected behavior:

all tests to pass.

#### Actual behavior:

crash with the log.

```
{"level":"info","ts":"2021-11-24 12:16:45.363","logger":"cube.store","caller":"raftstore/replica_destroy.go:130","msg":"delete shard data returned","case":"TestSplitWithMultiNodesCluster","store":"node-0","store-id":1,"shard-id":2}
{"level":"info","ts":"2021-11-24 12:16:45.363","logger":"cube.store","caller":"raftstore/replica_destroy.go:160","msg":"going to mark replica as destroyed","case":"TestSplitWithMultiNodesCluster","store":"node-0","store-id":1,"shard-id":2,"replica-id":3}
panic: close of closed channel

goroutine 22258 [running]:
github.com/matrixorigin/matrixcube/raftstore.(*replica).confirmDestroyed(...)
  /Users/lni/src/matrixcube/raftstore/replica_destroy.go:161
github.com/matrixorigin/matrixcube/raftstore.(*store).vacuum(_, {{0x2, {0x0, 0x0, 0x0}, {0x0, 0x0, 0x0}, {0x5, 0x3, ...}, ...}, ...})
  /Users/lni/src/matrixcube/raftstore/replica_destroy.go:135 +0xb71
github.com/matrixorigin/matrixcube/raftstore.(*vacuumCleaner).vacuum(0xc0031b1c40)
  /Users/lni/src/matrixcube/raftstore/vacuum_cleaner.go:100 +0x175
github.com/matrixorigin/matrixcube/raftstore.(*vacuumCleaner).start.func1()
  /Users/lni/src/matrixcube/raftstore/vacuum_cleaner.go:60 +0xfd
github.com/lni/goutils/syncutil.(*Stopper).runWorker.func1()
  /Users/lni/go/pkg/mod/github.com/lni/goutils@v1.3.0/syncutil/stopper.go:79 +0x12f
created by github.com/lni/goutils/syncutil.(*Stopper).runWorker
  /Users/lni/go/pkg/mod/github.com/lni/goutils@v1.3.0/syncutil/stopper.go:74 +0x199
FAIL  github.com/matrixorigin/matrixcube/raftstore  289.524s
FAIL
```

#### 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:
Fix #420 handle vacuuming the replica multiple times
**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 #420 

**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
 		}


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