experiment: verify in channel (#215)

Author: CassOnMars

node/consensus/master/broadcast_messaging.go

@@ -181,25 +181,16 @@ func (e *MasterClockConsensusEngine) handleSelfTestReport(
 		for i := 0; i < len(proofs); i++ {
 			proofs[i] = proof[i*516 : (i+1)*516]
 		}
-		if !e.frameProver.VerifyChallengeProof(
-			challenge,
-			int64(timestamp),
-			report.DifficultyMetric,
-			proofs,
-		) {
-			e.logger.Warn(
-				"received invalid proof from peer",
-				zap.String("peer_id", peer.ID(peerID).String()),
-			)
-			e.pubSub.SetPeerScore(peerID, -1000)
-
-			return errors.Wrap(
-				errors.New("invalid report"),
-				"handle self test report",
-			)
-		}
+		go func() {
+			e.verifyTestCh <- verifyChallenge{
+				peerID:           peerID,
+				challenge:        challenge,
+				timestamp:        int64(timestamp),
+				difficultyMetric: report.DifficultyMetric,
+				proofs:           proofs,
+			}
+		}()
 
-		info.LastSeen = time.Now().UnixMilli()
 		return nil
 	}
 

node/consensus/master/master_clock_consensus_engine.go

@@ -11,6 +11,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/mr-tron/base58"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
@@ -62,6 +63,7 @@ type MasterClockConsensusEngine struct {
 	report                      *protobufs.SelfTestReport
 	frameValidationCh           chan *protobufs.ClockFrame
 	bandwidthTestCh             chan []byte
+	verifyTestCh                chan verifyChallenge
 	currentReceivingSyncPeers   int
 	currentReceivingSyncPeersMx sync.Mutex
 }
@@ -126,6 +128,7 @@ func NewMasterClockConsensusEngine(
 		report:              report,
 		frameValidationCh:   make(chan *protobufs.ClockFrame),
 		bandwidthTestCh:     make(chan []byte),
+		verifyTestCh:        make(chan verifyChallenge),
 	}
 
 	e.addPeerManifestReport(e.pubSub.GetPeerID(), report)
@@ -186,6 +189,8 @@ func (e *MasterClockConsensusEngine) Start() <-chan error {
 				e.masterTimeReel.Insert(newFrame, false)
 			case peerId := <-e.bandwidthTestCh:
 				e.performBandwidthTest(peerId)
+			case verifyTest := <-e.verifyTestCh:
+				e.performVerifyTest(verifyTest)
 			}
 		}
 	}()
@@ -355,6 +360,34 @@ func (e *MasterClockConsensusEngine) Stop(force bool) <-chan error {
 	return errChan
 }
 
+type verifyChallenge struct {
+	peerID           []byte
+	challenge        []byte
+	timestamp        int64
+	difficultyMetric int64
+	proofs           [][]byte
+}
+
+func (e *MasterClockConsensusEngine) performVerifyTest(
+	challenge verifyChallenge,
+) {
+	if !e.frameProver.VerifyChallengeProof(
+		challenge.challenge,
+		challenge.timestamp,
+		challenge.difficultyMetric,
+		challenge.proofs,
+	) {
+		e.logger.Warn(
+			"received invalid proof from peer",
+			zap.String("peer_id", peer.ID(challenge.peerID).String()),
+		)
+		e.pubSub.SetPeerScore(challenge.peerID, -1000)
+	} else {
+		info := e.peerInfoManager.GetPeerInfo(challenge.peerID)
+		info.LastSeen = time.Now().UnixMilli()
+	}
+}
+
 func (e *MasterClockConsensusEngine) performBandwidthTest(peerID []byte) {
 	result := e.pubSub.GetMultiaddrOfPeer(peerID)
 	if result == "" {