Repository: matrixorigin/matrixcube (go)

Issue:
Refactor the raft ready processing code
saveRaftState should be in its own step, not as a part of the append entries step. 

handleEvent should return error. 

Pull request:
Resolve #390 refactor how raft ready is processed
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [x] Code Refactoring

**Which issue(s) this PR fixes:**

Fixes #390 

**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_event_loop.go b/raftstore/replica_event_loop.go
index 0b28d7e..29aca3d 100644
--- a/raftstore/replica_event_loop.go
+++ b/raftstore/replica_event_loop.go
@@ -93,16 +93,19 @@ func (pr *replica) addAdminRequest(req rpc.AdminRequest) error {
 	return pr.addRequest(newAdminReqCtx(req))
 }
 
+func (pr *replica) addRaftTick() {
+	if err := pr.ticks.Put(struct{}{}); err != nil {
+		pr.logger.Info("raft tick stopped")
+		return
+	}
+	pr.notifyWorker()
+}
+
 func (pr *replica) onRaftTick(arg interface{}) {
 	if !pr.stopRaftTick {
-		if err := pr.ticks.Put(struct{}{}); err != nil {
-			pr.logger.Info("raft tick stopped")
-			return
-		}
+		pr.addRaftTick()
 		metric.SetRaftTickQueueMetric(pr.ticks.Len())
-		pr.notifyWorker()
 	}
-
 	util.DefaultTimeoutWheel().Schedule(pr.cfg.Raft.TickInterval.Duration, pr.onRaftTick, nil)
 }
 
@@ -149,14 +152,14 @@ func (pr *replica) shutdown() {
 	pr.logger.Info("replica shutdown completed")
 }
 
-func (pr *replica) handleEvent(wc *logdb.WorkerContext) (hasEvent bool) {
+func (pr *replica) handleEvent(wc *logdb.WorkerContext) (hasEvent bool, err error) {
 	select {
 	case <-pr.closedC:
 		if !pr.unloaded() {
 			pr.shutdown()
 			pr.confirmUnloaded()
 		}
-		return false
+		return false, nil
 	default:
 	}
 
@@ -174,12 +177,14 @@ func (pr *replica) handleEvent(wc *logdb.WorkerContext) (hasEvent bool) {
 	}
 	if pr.rn.HasReady() {
 		hasEvent = true
-		pr.handleReady(wc)
+		if err := pr.handleRaftReady(wc); err != nil {
+			return hasEvent, err
+		}
 	}
 	if pr.handleAction(pr.items) {
 		hasEvent = true
 	}
-	return true
+	return true, nil
 }
 
 func (pr *replica) handleAction(items []interface{}) bool {
diff --git a/raftstore/replica_event_raft_ready.go b/raftstore/replica_event_raft_ready.go
index 4afb820..0902e0b 100644
--- a/raftstore/replica_event_raft_ready.go
+++ b/raftstore/replica_event_raft_ready.go
@@ -32,16 +32,35 @@ var (
 	ErrUnknownReplica = errors.New("unknown replica")
 )
 
-func (pr *replica) handleReady(wc *logdb.WorkerContext) error {
+func (pr *replica) handleRaftReady(wc *logdb.WorkerContext) error {
+	rd := pr.getRaftReady()
+	if err := pr.processReady(rd, wc); err != nil {
+		return err
+	}
+	pr.commitRaftReady(rd)
+	return nil
+}
+
+func (pr *replica) getRaftReady() raft.Ready {
+	return pr.rn.Ready()
+}
+
+func (pr *replica) commitRaftReady(rd raft.Ready) {
+	pr.rn.Advance(rd)
+}
+
+func (pr *replica) processReady(rd raft.Ready, wc *logdb.WorkerContext) error {
 	if ce := pr.logger.Check(zap.DebugLevel, "begin handleReady"); ce != nil {
 		ce.Write(log.ShardIDField(pr.shardID),
 			log.ReplicaIDField(pr.replica.ID))
 	}
 
-	rd := pr.rn.Ready()
 	pr.handleRaftState(rd)
 	pr.sendRaftAppendLogMessages(rd)
-	if err := pr.handleRaftReadyAppend(rd, wc); err != nil {
+	if err := pr.saveRaftState(rd, wc); err != nil {
+		return err
+	}
+	if err := pr.appendEntries(rd); err != nil {
 		return err
 	}
 	pr.sendRaftMessages(rd)
@@ -52,7 +71,6 @@ func (pr *replica) handleReady(wc *logdb.WorkerContext) error {
 	if err := pr.handleRaftCreateSnapshotRequest(); err != nil {
 		return err
 	}
-	pr.rn.Advance(rd)
 
 	if ce := pr.logger.Check(zap.DebugLevel, "handleReady completed"); ce != nil {
 		ce.Write(log.ShardIDField(pr.shardID),
@@ -83,13 +101,6 @@ func (pr *replica) handleRaftState(rd raft.Ready) {
 	}
 }
 
-func (pr *replica) handleRaftReadyAppend(rd raft.Ready,
-	wc *logdb.WorkerContext) error {
-	start := time.Now()
-	defer metric.ObserveRaftLogAppendDuration(start)
-	return pr.handleAppendEntries(rd, wc)
-}
-
 func getEstimatedAppendSize(rd raft.Ready) int {
 	sz := 0
 	for _, e := range rd.Entries {
@@ -99,8 +110,10 @@ func getEstimatedAppendSize(rd raft.Ready) int {
 	return sz
 }
 
-func (pr *replica) handleAppendEntries(rd raft.Ready,
-	wc *logdb.WorkerContext) error {
+func (pr *replica) appendEntries(rd raft.Ready) error {
+	start := time.Now()
+	defer metric.ObserveRaftLogAppendDuration(start)
+
 	if len(rd.Entries) > 0 {
 		if ce := pr.logger.Check(zap.DebugLevel,
 			"begin to append raft log"); ce != nil {
@@ -109,7 +122,7 @@ func (pr *replica) handleAppendEntries(rd raft.Ready,
 				log.IndexField(rd.Entries[0].Index),
 				zap.Int("estimated-size", getEstimatedAppendSize(rd)))
 		}
-		pr.lr.Append(rd.Entries)
+		err := pr.lr.Append(rd.Entries)
 		if ce := pr.logger.Check(zap.DebugLevel,
 			"append raft log completed"); ce != nil {
 			ce.Write(log.ShardIDField(pr.shardID),
@@ -117,8 +130,12 @@ func (pr *replica) handleAppendEntries(rd raft.Ready,
 				log.IndexField(rd.Entries[0].Index))
 		}
 		pr.metrics.ready.append++
+		return err
 	}
+	return nil
+}
 
+func (pr *replica) saveRaftState(rd raft.Ready, wc *logdb.WorkerContext) error {
 	return pr.logdb.SaveRaftState(pr.shardID, pr.replica.ID, rd, wc)
 }
 
diff --git a/raftstore/worker_pool.go b/raftstore/worker_pool.go
index 7894390..a09d45e 100644
--- a/raftstore/worker_pool.go
+++ b/raftstore/worker_pool.go
@@ -43,7 +43,7 @@ var _ replicaLoader = (*storeReplicaLoader)(nil)
 
 type replicaEventHandler interface {
 	getShardID() uint64
-	handleEvent(*logdb.WorkerContext) bool
+	handleEvent(*logdb.WorkerContext) (bool, error)
 }
 
 var _ replicaEventHandler = (*replica)(nil)
@@ -102,7 +102,12 @@ func (w *replicaWorker) completed() {
 func (w *replicaWorker) handleEvent(h replicaEventHandler) error {
 	for {
 		w.wc.Reset()
-		if !h.handleEvent(w.wc) {
+		hasEvent, err := h.handleEvent(w.wc)
+		if err != nil {
+			// TODO: pretty printing the error
+			panic(err)
+		}
+		if !hasEvent {
 			break
 		}
 	}


Test patch:
diff --git a/raftstore/replica_event_loop_test.go b/raftstore/replica_event_loop_test.go
index 68a1757..c5a0d2a 100644
--- a/raftstore/replica_event_loop_test.go
+++ b/raftstore/replica_event_loop_test.go
@@ -88,5 +88,7 @@ func TestReplicaCanBeClosed(t *testing.T) {
 	r, closer := getCloseableReplica()
 	defer r.close()
 	defer closer()
-	assert.True(t, r.handleEvent(nil))
+	hasEvent, err := r.handleEvent(nil)
+	assert.NoError(t, err)
+	assert.True(t, hasEvent)
 }
diff --git a/raftstore/replica_event_raft_ready_test.go b/raftstore/replica_event_raft_ready_test.go
index 83a7d77..2fa9009 100644
--- a/raftstore/replica_event_raft_ready_test.go
+++ b/raftstore/replica_event_raft_ready_test.go
@@ -183,17 +183,16 @@ func TestSendOtherMessages(t *testing.T) {
 func TestIssue386(t *testing.T) {
 	// See https://github.com/matrixorigin/matrixcube/issues/386
 	// Missing save hardstate to logdb if append entries is empty.
-
 	s, closer := newTestStore(t)
 	defer closer()
 
 	pr := newTestReplica(Shard{ID: 1}, Replica{ID: 2}, s)
 
-	assert.NoError(t, pr.handleRaftReadyAppend(raft.Ready{
+	assert.NoError(t, pr.processReady(raft.Ready{
 		Entries: []raftpb.Entry{{Index: 1, Term: 1, Data: []byte("test")}},
 	}, s.logdb.NewWorkerContext()))
 
-	assert.NoError(t, pr.handleRaftReadyAppend(raft.Ready{
+	assert.NoError(t, pr.processReady(raft.Ready{
 		HardState: raftpb.HardState{Commit: 1},
 	}, s.logdb.NewWorkerContext()))
 
diff --git a/raftstore/worker_pool_test.go b/raftstore/worker_pool_test.go
index 9d11eb3..02e69f0 100644
--- a/raftstore/worker_pool_test.go
+++ b/raftstore/worker_pool_test.go
@@ -51,13 +51,13 @@ func (t *testReplicaEventHandler) getShardID() uint64 {
 	return t.shardID
 }
 
-func (t *testReplicaEventHandler) handleEvent(*logdb.WorkerContext) bool {
+func (t *testReplicaEventHandler) handleEvent(*logdb.WorkerContext) (bool, error) {
 	if t.invoked != nil {
 		close(t.invoked)
 		<-t.waitC
 	}
 	atomic.StoreUint64(&t.handled, 1)
-	return false
+	return false, nil
 }
 
 func (t *testReplicaEventHandler) getHandled() bool {