Merge branch 'master' into fix-unit-tests

.github/workflows/go.yml

@@ -0,0 +1,27 @@
+# This workflow will build a golang project
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-go
+
+name: Go
+
+on: [push , pull_request]
+
+jobs:
+
+  build:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Set up Go
+      uses: actions/setup-go@v4
+      with:
+        go-version: '1.20'
+
+#    - name: Run init
+#      run: go run build/ci.go lint
+
+    - name: Build
+      run: make geth
+
+    - name: Test
+      run: go test -timeout=40m -tags=ckzg -p 1 ./...

accounts/keystore/keystore.go

@@ -240,7 +240,7 @@ func (ks *KeyStore) Delete(a accounts.Account, passphrase string) error {
 	// Decrypting the key isn't really necessary, but we do
 	// it anyway to check the password and zero out the key
 	// immediately afterwards.
-	a, key, err := ks.getDecryptedKey(a, passphrase)
+	a, key, err := ks.GetDecryptedKey(a, passphrase)
 	if key != nil {
 		zeroKey(key.PrivateKey)
 	}
@@ -292,7 +292,7 @@ func (ks *KeyStore) SignTx(a accounts.Account, tx *types.Transaction, chainID *b
 // can be decrypted with the given passphrase. The produced signature is in the
 // [R || S || V] format where V is 0 or 1.
 func (ks *KeyStore) SignHashWithPassphrase(a accounts.Account, passphrase string, hash []byte) (signature []byte, err error) {
-	_, key, err := ks.getDecryptedKey(a, passphrase)
+	_, key, err := ks.GetDecryptedKey(a, passphrase)
 	if err != nil {
 		return nil, err
 	}
@@ -303,7 +303,7 @@ func (ks *KeyStore) SignHashWithPassphrase(a accounts.Account, passphrase string
 // SignTxWithPassphrase signs the transaction if the private key matching the
 // given address can be decrypted with the given passphrase.
 func (ks *KeyStore) SignTxWithPassphrase(a accounts.Account, passphrase string, tx *types.Transaction, chainID *big.Int) (*types.Transaction, error) {
-	_, key, err := ks.getDecryptedKey(a, passphrase)
+	_, key, err := ks.GetDecryptedKey(a, passphrase)
 	if err != nil {
 		return nil, err
 	}
@@ -338,7 +338,7 @@ func (ks *KeyStore) Lock(addr common.Address) error {
 // shortens the active unlock timeout. If the address was previously unlocked
 // indefinitely the timeout is not altered.
 func (ks *KeyStore) TimedUnlock(a accounts.Account, passphrase string, timeout time.Duration) error {
-	a, key, err := ks.getDecryptedKey(a, passphrase)
+	a, key, err := ks.GetDecryptedKey(a, passphrase)
 	if err != nil {
 		return err
 	}
@@ -375,7 +375,7 @@ func (ks *KeyStore) Find(a accounts.Account) (accounts.Account, error) {
 	return a, err
 }
 
-func (ks *KeyStore) getDecryptedKey(a accounts.Account, auth string) (accounts.Account, *Key, error) {
+func (ks *KeyStore) GetDecryptedKey(a accounts.Account, auth string) (accounts.Account, *Key, error) {
 	a, err := ks.Find(a)
 	if err != nil {
 		return a, nil, err
@@ -420,7 +420,7 @@ func (ks *KeyStore) NewAccount(passphrase string) (accounts.Account, error) {
 
 // Export exports as a JSON key, encrypted with newPassphrase.
 func (ks *KeyStore) Export(a accounts.Account, passphrase, newPassphrase string) (keyJSON []byte, err error) {
-	_, key, err := ks.getDecryptedKey(a, passphrase)
+	_, key, err := ks.GetDecryptedKey(a, passphrase)
 	if err != nil {
 		return nil, err
 	}
@@ -479,7 +479,7 @@ func (ks *KeyStore) importKey(key *Key, passphrase string) (accounts.Account, er
 
 // Update changes the passphrase of an existing account.
 func (ks *KeyStore) Update(a accounts.Account, passphrase, newPassphrase string) error {
-	a, key, err := ks.getDecryptedKey(a, passphrase)
+	a, key, err := ks.GetDecryptedKey(a, passphrase)
 	if err != nil {
 		return err
 	}

consensus/consensus.go

@@ -169,12 +169,6 @@ type HotStuff interface {
 
 	// FillHeader fulfill the header with extra which contains epoch change info
 	FillHeader(state *state.StateDB, header *types.Header) error
-
-	// IsSystemCall return method id and true if the tx is an system transaction
-	IsSystemTransaction(tx *types.Transaction, header *types.Header) (string, bool)
-
-	// HasSystemTxHook return true if systemTxHook is not nil
-	HasSystemTxHook() bool
 }
 
 // Handler should be implemented is the consensus needs to handle and send peer's message

consensus/hotstuff/backend.go

@@ -33,7 +33,7 @@ type Backend interface {
 	Address() common.Address
 
 	// Validators returns current epoch participants
-	Validators(height uint64, inConsensus bool) ValidatorSet
+	Validators(height uint64, inConsensus bool) (ValidatorSet, error)
 
 	// EventMux returns the event mux in backend
 	EventMux() *event.TypeMux
@@ -70,7 +70,7 @@ type Backend interface {
 	// CheckPoint retrieve the flag of epoch change and new epoch start height
 	CheckPoint(height uint64) (uint64, bool)
 
-	ReStart()
+	Reset()
 
 	Close() error
 }

consensus/hotstuff/backend/engine.go

@@ -288,24 +288,27 @@ func (s *backend) Close() error {
 	return nil
 }
 
-func (s *backend) ReStart() {
+func (s *backend) Reset() {
+	if !s.coreStarted {
+		log.Errorf("Try to reset stopped core engine")
+		return
+	}
+	log.Debug("Reset consensus engine...")
+
 	next, err := s.newEpochValidators()
 	if err != nil {
-		panic(fmt.Errorf("Restart consensus engine failed, err: %v ", err))
+		panic(fmt.Errorf("Reset consensus engine failed, err: %v ", err))
 	}
-
 	if next.Equal(s.vals.Copy()) {
-		log.Trace("Restart Consensus engine, validators not changed.", "origin", s.vals.AddressList(), "current", next.AddressList())
+		log.Trace("Reset Consensus engine, validators not changed.", "origin", s.vals.AddressList(), "current", next.AddressList())
 		return
 	}
 
-	if s.coreStarted {
-		s.Stop()
-		// waiting for last engine instance free resource, e.g: unsubscribe...
-		time.Sleep(2 * time.Second)
-		log.Debug("Restart consensus engine...")
-		s.Start(s.chain, s.hasBadBlock)
-	}
+	// reset validator set
+	s.vals = next.Copy()
+
+	// p2p module connect nodes directly
+	s.nodesFeed.Send(consensus.StaticNodesEvent{Validators: s.vals.AddressList()})
 }
 
 // verifyHeader checks whether a header conforms to the consensus rules.The

consensus/hotstuff/backend/governance.go

@@ -95,20 +95,20 @@ func (s *backend) CheckPoint(height uint64) (uint64, bool) {
 }
 
 // Validators get validators from backend by `consensus core`, param of `mining` is false denote need last epoch validators.
-func (s *backend) Validators(height uint64, mining bool) hotstuff.ValidatorSet {
+func (s *backend) Validators(height uint64, mining bool) (hotstuff.ValidatorSet, error) {
 	if mining {
-		return s.vals.Copy()
+		return s.vals.Copy(), nil
 	}
 
 	header := s.chain.GetHeaderByNumber(height)
 	if header == nil {
-		return nil
+		return nil, fmt.Errorf("GetHeaderByNumber, header is nil")
 	}
 	_, vals, err := s.getValidatorsByHeader(header, nil, s.chain)
 	if err != nil {
-		return nil
+		return nil, err
 	}
-	return vals
+	return vals, nil
 }
 
 // getValidatorsByHeader check if current header height is an new epoch start and retrieve the validators.

consensus/hotstuff/core/core.go

@@ -58,6 +58,8 @@ type core struct {
 	validateFn   func(common.Hash, []byte) (common.Address, error)
 	checkPointFn func(uint64) (uint64, bool)
 	isRunning    bool
+
+	wg sync.WaitGroup
 }
 
 // New creates an HotStuff consensus core
@@ -125,30 +127,30 @@ func (c *core) startNewRound(round *big.Int) {
 		Height: new(big.Int).Add(lastProposal.Number(), common.Big1),
 		Round:  new(big.Int),
 	}
+	var changeEpoch bool
 	if changeView {
 		newView.Height = new(big.Int).Set(c.current.Height())
 		newView.Round = new(big.Int).Set(round)
-	} else if c.checkPoint(newView) {
-		logger.Trace("Stop engine after check point.")
-		return
+	} else {
+		changeEpoch = c.checkPoint(newView)
 	}
 
 	// calculate validator set
-	c.valSet = c.backend.Validators(newView.HeightU64(), true)
+	var err error
+	if c.valSet, err = c.backend.Validators(newView.HeightU64(), true); err != nil {
+		logger.Error("get validator set failed", "err", err)
+		return
+	}
 	c.valSet.CalcProposer(lastProposer, newView.Round.Uint64())
 
 	// update smr and try to unlock at the round0
 	if err := c.updateRoundState(lastProposal, newView); err != nil {
 		logger.Error("Update round state failed", "state", c.currentState(), "newView", newView, "err", err)
 		return
 	}
-	if !changeView {
-		if err := c.current.Unlock(); err != nil {
-			logger.Error("Unlock node failed", "newView", newView, "err", err)
-			return
-		}
+	if changeEpoch {
+		c.current.Unlock()
 	}
-
 	logger.Debug("New round", "state", c.currentState(), "newView", newView, "new_proposer", c.valSet.GetProposer(), "valSet", c.valSet.List(), "size", c.valSet.Size(), "IsProposer", c.IsProposer())
 
 	// stop last timer and regenerate new timer
@@ -170,9 +172,7 @@ func (c *core) checkPoint(view *View) bool {
 		c.point = epochStart
 		c.lastVals = c.valSet.Copy()
 		c.logger.Trace("CheckPoint done", "view", view, "point", c.point)
-		c.backend.ReStart()
-	}
-	if !c.isRunning {
+		c.backend.Reset()
 		return true
 	}
 	return false

consensus/hotstuff/core/decide.go

@@ -29,8 +29,10 @@ import (
 // handleCommitVote implement description as follow:
 // ```
 // leader wait for (n n f) votes: V ← {v | matchingMsg(v, commit, curView)}
+//
 //	commitQC ← QC(V )
 //	broadcast Msg(decide, ⊥, commitQC )
+//
 // ```
 func (c *core) handleCommitVote(data *Message) error {
 	var (
@@ -200,6 +202,9 @@ func (c *core) handleDecide(data *Message) error {
 		}
 	}
 
+	//prepare for new round
+	c.current.Unlock()
+
 	c.startNewRound(common.Big0)
 	return nil
 }

consensus/hotstuff/core/handler.go

@@ -32,17 +32,19 @@ func (c *core) Start(chain consensus.ChainReader) {
 	c.current = nil
 
 	c.subscribeEvents()
-	go c.handleEvents()
 
 	// Start a new round from last sequence + 1
 	c.startNewRound(common.Big0)
+	c.wg.Add(1)
+	go c.handleEvents()
 }
 
 // Stop implements core.Engine.Stop
 func (c *core) Stop() {
 	c.stopTimer()
 	c.unsubscribeEvents()
 	c.isRunning = false
+	c.wg.Wait()
 }
 
 // Address implement core.Engine.Address
@@ -100,6 +102,7 @@ func (c *core) unsubscribeEvents() {
 }
 
 func (c *core) handleEvents() {
+	defer c.wg.Done()
 	logger := c.logger.New("handleEvents")
 
 	for {

consensus/hotstuff/core/prepare.go

@@ -53,6 +53,10 @@ func (c *core) sendPrepare() {
 		request := c.current.PendingRequest()
 		if request == nil || request.block == nil || request.block.NumberU64() != c.HeightU64() {
 			logger.Trace("Pending request invalid", "msg", code)
+			if request != nil && request.block != nil {
+				logger.Trace("Pending request invalid", "msg", code, "request.block.Number", request.block.NumberU64(),
+					"c.Height", c.HeightU64(), "request.block.hash", request.block.SealHash())
+			}
 			return
 		} else {
 			block = c.current.PendingRequest().block
@@ -91,10 +95,12 @@ func (c *core) sendPrepare() {
 
 // handlePrepare implement description as follow:
 // ```
-//  repo wait for message m : matchingMsg(m, prepare, curView) from leader(curView)
-//	if m.node extends from m.justify.node ∧
-//	safeNode(m.node, m.justify) then
-//	send voteMsg(prepare, m.node, ⊥) to leader(curView)
+//
+//	 repo wait for message m : matchingMsg(m, prepare, curView) from leader(curView)
+//		if m.node extends from m.justify.node ∧
+//		safeNode(m.node, m.justify) then
+//		send voteMsg(prepare, m.node, ⊥) to leader(curView)
+//
 // ```
 func (c *core) handlePrepare(data *Message) error {
 	var (
@@ -127,17 +133,19 @@ func (c *core) handlePrepare(data *Message) error {
 
 	// ensure remote block is legal.
 	block := node.Block
-	if err := c.checkBlock(block); err != nil {
-		logger.Trace("Failed to check block", "msg", code, "src", src, "err", err)
-		return err
-	}
-	if duration, err := c.backend.Verify(block, false); err != nil {
-		logger.Trace("Failed to verify unsealed proposal", "msg", code, "src", src, "err", err, "duration", duration)
-		return errVerifyUnsealedProposal
-	}
-	if err := c.executeBlock(block); err != nil {
-		logger.Trace("Failed to execute block", "msg", code, "src", src, "err", err)
-		return err
+	if c.current.executed == nil || c.current.executed.Block != nil || c.current.executed.Block.SealHash() != block.SealHash() {
+		if err := c.checkBlock(block); err != nil {
+			logger.Trace("Failed to check block", "msg", code, "src", src, "err", err)
+			return err
+		}
+		if duration, err := c.backend.Verify(block, false); err != nil {
+			logger.Trace("Failed to verify unsealed proposal", "msg", code, "src", src, "err", err, "duration", duration)
+			return errVerifyUnsealedProposal
+		}
+		if err := c.executeBlock(block); err != nil {
+			logger.Trace("Failed to execute block", "msg", code, "src", src, "err", err)
+			return err
+		}
 	}
 
 	// safety and liveness rules judgement.

consensus/hotstuff/core/request.go

@@ -46,6 +46,7 @@ func (c *core) handleRequest(request *Request) error {
 		if c.current.PendingRequest() == nil ||
 			c.current.PendingRequest().block.NumberU64() < c.current.HeightU64() {
 			c.current.SetPendingRequest(request)
+			logger.Trace("Set PendingRequest", "number", request.block.NumberU64(), "hash", request.block.SealHash())
 			c.sendPrepare()
 		} else {
 			logger.Trace("PendingRequest exist")

consensus/hotstuff/core/round_state.go

@@ -149,9 +149,7 @@ func (s *roundState) LastChainedBlock() *types.Block {
 
 // accept pending request from miner only for once.
 func (s *roundState) SetPendingRequest(req *Request) {
-	if s.pendingRequest == nil {
-		s.pendingRequest = req
-	}
+	s.pendingRequest = req
 }
 
 func (s *roundState) PendingRequest() *Request {
@@ -207,13 +205,12 @@ func (s *roundState) LockQC() *QuorumCert {
 }
 
 // Unlock it's happened at the start of new round, new state is `StateAcceptRequest`, and `lockQC` keep to judge safety rule
-func (s *roundState) Unlock() error {
+func (s *roundState) Unlock() {
 	s.pendingRequest = nil
 	s.proposalLocked = false
 	s.lockedBlock = nil
 	s.node.temp = nil
 	s.executed = nil
-	return nil
 }
 
 func (s *roundState) LockedBlock() *types.Block {