Fix merge issues

Author: bruce-riley

node/cmd/guardiand/node.go

@@ -240,6 +240,8 @@ var (
 	chainGovernorEnabled      *bool
 	governorFlowCancelEnabled *bool
 
+	processorWorkerFactor *float64
+
 	ccqEnabled           *bool
 	ccqAllowedRequesters *string
 	ccqP2pPort           *uint
@@ -450,6 +452,7 @@ func init() {
 
 	chainGovernorEnabled = NodeCmd.Flags().Bool("chainGovernorEnabled", false, "Run the chain governor")
 	governorFlowCancelEnabled = NodeCmd.Flags().Bool("governorFlowCancelEnabled", false, "Enable flow cancel on the governor")
+	processorWorkerFactor = NodeCmd.Flags().Float64("processorWorkerFactor", 0.0, "Multiplied by the number of available CPUs on the system to determine the number of workers that the processor uses. 0.0 means single worker")
 
 	ccqEnabled = NodeCmd.Flags().Bool("ccqEnabled", false, "Enable cross chain query support")
 	ccqAllowedRequesters = NodeCmd.Flags().String("ccqAllowedRequesters", "", "Comma separated list of signers allowed to submit cross chain queries")
@@ -1639,7 +1642,7 @@ func runNode(cmd *cobra.Command, args []string) {
 		node.GuardianOptionAdminService(*adminSocketPath, ethRPC, ethContract, rpcMap),
 		node.GuardianOptionP2P(p2pKey, *p2pNetworkID, *p2pBootstrap, *nodeName, *subscribeToVAAs, *disableHeartbeatVerify, *p2pPort, *ccqP2pBootstrap, *ccqP2pPort, *ccqAllowedPeers, *gossipAdvertiseAddress, ibc.GetFeatures),
 		node.GuardianOptionStatusServer(*statusAddr),
-		node.GuardianOptionProcessor(*p2pNetworkID),
+		node.GuardianOptionProcessor(*p2pNetworkID, *processorWorkerFactor),
 	}
 
 	if shouldStart(publicGRPCSocketPath) {

node/pkg/node/node_test.go

@@ -14,6 +14,7 @@ import (
 	"net/http"
 	"os"
 	"regexp"
+	"runtime"
 	"strconv"
 	"strings"
 	"testing"
@@ -196,7 +197,7 @@ func mockGuardianRunnable(t testing.TB, gs []*mockGuardian, mockGuardianIndex ui
 			GuardianOptionPublicWeb(cfg.publicWeb, cfg.publicSocket, "", false, ""),
 			GuardianOptionAdminService(cfg.adminSocket, nil, nil, rpcMap),
 			GuardianOptionStatusServer(fmt.Sprintf("[::]:%d", cfg.statusPort)),
-			GuardianOptionProcessor(networkID),
+			GuardianOptionProcessor(networkID, 3.0/float64(runtime.NumCPU())), // Create three workers.
 		}
 
 		guardianNode := NewGuardianNode(

node/pkg/node/options.go

@@ -575,7 +575,7 @@ func GuardianOptionDatabase(db *db.Database) *GuardianOption {
 
 // GuardianOptionProcessor enables the default processor, which is required to make consensus on messages.
 // Dependencies: db, governor, accountant
-func GuardianOptionProcessor(networkId string) *GuardianOption {
+func GuardianOptionProcessor(networkId string, workerFactor float64) *GuardianOption {
 	return &GuardianOption{
 		name: "processor",
 		// governor and accountant may be set to nil, but that choice needs to be made before the processor is configured
@@ -600,6 +600,7 @@ func GuardianOptionProcessor(networkId string) *GuardianOption {
 				g.acctC.readC,
 				g.gatewayRelayer,
 				networkId,
+				workerFactor,
 			).Run
 
 			return nil

node/pkg/processor/benchmark_test.go

@@ -174,11 +174,10 @@ func createProcessorForTest(b *testing.B, numVAAs int, ctx context.Context, db *
 		gossipAttestationSendC: pd.gossipAttestationSendC,
 		gossipVaaSendC:         pd.gossipVaaSendC,
 		guardianSigner:         ourSigner,
-		gs:                     gs,
 		gst:                    gst,
 		db:                     db,
 		logger:                 logger,
-		state:                  &aggregationState{observationMap{}},
+		state:                  &aggregationState{signatures: observationMap{}},
 		ourAddr:                crypto.PubkeyToAddress(ourSigner.PublicKey()),
 		pythnetVaas:            make(map[string]PythNetVaaEntry),
 		updatedVAAs:            make(map[string]*updateVaaEntry),

node/pkg/processor/cleanup.go

@@ -2,7 +2,6 @@
 package processor
 
 import (
-	"context"
 	"encoding/hex"
 	"fmt"
 	"time"
@@ -69,7 +68,7 @@ var (
 )
 
 // handleCleanup handles periodic retransmissions and cleanup of observations
-func (p *Processor) handleCleanup(ctx context.Context) {
+func (p *Processor) handleCleanup() {
 	p.cleanupState()
 	p.cleanupPythnetVaas()
 }
@@ -129,37 +128,36 @@ func (p *Processor) cleanUpStateEntry(hash string, s *state) bool {
 		gs := s.gs
 		if gs == nil {
 			gs = p.gst.Get()
-			if gs == nil {
-				return false
-			}
 		}
 
-		hasSigs := len(s.signatures)
-		wantSigs := vaa.CalculateQuorum(len(gs.Keys))
-		quorum := hasSigs >= wantSigs
+		if gs != nil {
+			if p.logger.Level().Enabled(zapcore.DebugLevel) {
+				hasSigs := len(s.signatures)
+				wantSigs := vaa.CalculateQuorum(len(gs.Keys))
+				quorum := hasSigs >= wantSigs
 
-		var chain vaa.ChainID
-		if s.ourObservation != nil {
-			chain = s.ourObservation.GetEmitterChain()
-		}
+				var chain vaa.ChainID
+				if s.ourObservation != nil {
+					chain = s.ourObservation.GetEmitterChain()
+				}
 
-		if p.logger.Level().Enabled(zapcore.DebugLevel) {
-			p.logger.Debug("observation considered settled",
-				zap.String("message_id", s.LoggingID()),
-				zap.String("digest", hash),
-				zap.Duration("delta", delta),
-				zap.Int("have_sigs", hasSigs),
-				zap.Int("required_sigs", wantSigs),
-				zap.Bool("quorum", quorum),
-				zap.Stringer("emitter_chain", chain),
-			)
-		}
+				p.logger.Debug("observation considered settled",
+					zap.String("message_id", s.LoggingID()),
+					zap.String("digest", hash),
+					zap.Duration("delta", delta),
+					zap.Int("have_sigs", hasSigs),
+					zap.Int("required_sigs", wantSigs),
+					zap.Bool("quorum", quorum),
+					zap.Stringer("emitter_chain", chain),
+				)
+			}
 
-		for _, k := range gs.Keys {
-			if _, ok := s.signatures[k]; ok {
-				aggregationStateFulfillment.WithLabelValues(k.Hex(), s.source, "present").Inc()
-			} else {
-				aggregationStateFulfillment.WithLabelValues(k.Hex(), s.source, "missing").Inc()
+			for _, k := range gs.Keys {
+				if _, ok := s.signatures[k]; ok {
+					aggregationStateFulfillment.WithLabelValues(k.Hex(), s.source, "present").Inc()
+				} else {
+					aggregationStateFulfillment.WithLabelValues(k.Hex(), s.source, "missing").Inc()
+				}
 			}
 		}
 	case s.submitted && delta.Hours() >= 1:
@@ -176,13 +174,13 @@ func (p *Processor) cleanUpStateEntry(hash string, s *state) bool {
 		}
 		aggregationStateExpiration.Inc()
 		return true
-	case !s.submitted && ((s.ourMsg != nil && delta > retryLimitOurs) || (s.ourMsg == nil && delta > retryLimitNotOurs)):
+	case !s.submitted && ((s.ourObs != nil && delta > retryLimitOurs) || (s.ourObs == nil && delta > retryLimitNotOurs)):
 		// Clearly, this horse is dead and continued beatings won't bring it closer to quorum.
 		p.logger.Info("expiring unsubmitted observation after exhausting retries",
 			zap.String("message_id", s.LoggingID()),
 			zap.String("digest", hash),
 			zap.Duration("delta", delta),
-			zap.Bool("weObserved", s.ourMsg != nil),
+			zap.Bool("weObserved", s.ourObs != nil),
 		)
 		aggregationStateTimeout.Inc()
 		return true
@@ -193,7 +191,7 @@ func (p *Processor) cleanUpStateEntry(hash string, s *state) bool {
 		// sig. If we do not have an observation, it means we either never observed it, or it got
 		// revived by a malfunctioning guardian node, in which case, we can't do anything about it
 		// and just delete it to keep our state nice and lean.
-		if s.ourMsg != nil {
+		if s.ourObs != nil {
 			// Unreliable observations cannot be resubmitted and can be considered failed after 5 minutes
 			if !s.ourObservation.IsReliable() {
 				p.logger.Info("expiring unsubmitted unreliable observation",
@@ -247,7 +245,12 @@ func (p *Processor) cleanUpStateEntry(hash string, s *state) bool {
 				if err := common.PostObservationRequest(p.obsvReqSendC, req); err != nil {
 					p.logger.Warn("failed to broadcast re-observation request", zap.String("message_id", s.LoggingID()), zap.Error(err))
 				}
-				p.gossipSendC <- s.ourMsg
+				if s.ourMsg != nil {
+					// This is the case for immediately published messages (as well as anything still pending from before the cutover).
+					p.gossipAttestationSendC <- s.ourMsg
+				} else {
+					p.postObservationToBatch(s.ourObs)
+				}
 				s.retryCtr++
 				s.nextRetry = time.Now().Add(nextRetryDuration(s.retryCtr))
 				aggregationStateRetries.Inc()
@@ -256,23 +259,19 @@ func (p *Processor) cleanUpStateEntry(hash string, s *state) bool {
 			// For nil state entries, we log the quorum to determine whether the
 			// network reached consensus without us. We don't know the correct guardian
 			// set, so we simply use the most recent one.
-			hasSigs := len(s.signatures)
-			gs := p.gst.Get()
-			if gs != nil {
-				wantSigs := vaa.CalculateQuorum(len(gs.Keys))
-
-				if p.logger.Level().Enabled(zapcore.DebugLevel) {
+			if p.logger.Level().Enabled(zapcore.DebugLevel) {
+				hasSigs := len(s.signatures)
+				gs := p.gst.Get()
+				if gs != nil {
 					p.logger.Debug("expiring unsubmitted nil observation",
 						zap.String("message_id", s.LoggingID()),
 						zap.String("digest", hash),
 						zap.Duration("delta", delta),
 						zap.Int("have_sigs", hasSigs),
-						zap.Int("required_sigs", p.gs.Quorum()),
-						zap.Bool("quorum", hasSigs >= p.gs.Quorum()),
+						zap.Int("required_sigs", gs.Quorum()),
+						zap.Bool("quorum", hasSigs >= gs.Quorum()),
 					)
-				}
-			} else {
-				if p.logger.Level().Enabled(zapcore.DebugLevel) {
+				} else {
 					p.logger.Debug("expiring unsubmitted nil observation, gs is nil",
 						zap.String("message_id", s.LoggingID()),
 						zap.String("digest", hash),

node/pkg/processor/message.go

@@ -9,7 +9,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 
-	ethCommon "github.com/ethereum/go-ethereum/common"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
@@ -100,31 +99,27 @@ func (p *Processor) handleMessage(k *common.MessagePublication) {
 	observationsReceivedByGuardianAddressTotal.WithLabelValues(p.ourAddr.Hex()).Inc()
 
 	// Get / create our state entry.
-	s := p.state.signatures[hash]
-	if s == nil {
-		s = &state{
-			firstObserved: time.Now(),
-			nextRetry:     time.Now().Add(nextRetryDuration(0)),
-			signatures:    map[ethCommon.Address][]byte{},
-			source:        "loopback",
-		}
-
-		p.state.signatures[hash] = s
+	s, created := p.state.getOrCreateState(hash)
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	if created {
+		s.source = "loopback"
 	}
 
 	// Update our state.
 	s.ourObservation = v
 	s.txHash = k.TxHash.Bytes()
 	s.source = v.GetEmitterChain().String()
-	s.gs = p.gs // guaranteed to match ourObservation - there's no concurrent access to p.gs
+	s.gs = p.gst.Get() // guaranteed to match ourObservation - there's no concurrent access to p.gs
 	s.signatures[p.ourAddr] = signature
 	s.ourObs = ourObs
 	s.ourMsg = msg
 
 	// Fast path for our own signature.
 	if !s.submitted {
 		start := time.Now()
-		p.checkForQuorum(ourObs, s, s.gs, hash)
+		p.checkForQuorumAlreadyLocked(ourObs, s, hash)
 		timeToHandleObservation.Observe(float64(time.Since(start).Microseconds()))
 	}
 }