core/consensus: integrate qbft

app/app.go

@@ -50,6 +50,7 @@ import (
 	"github.com/obolnetwork/charon/core"
 	"github.com/obolnetwork/charon/core/aggsigdb"
 	"github.com/obolnetwork/charon/core/bcast"
+	"github.com/obolnetwork/charon/core/consensus"
 	"github.com/obolnetwork/charon/core/dutydb"
 	"github.com/obolnetwork/charon/core/fetcher"
 	"github.com/obolnetwork/charon/core/leadercast"
@@ -151,7 +152,16 @@ func Run(ctx context.Context, conf Config) (err error) {
 	}
 	lockHashHex := hex.EncodeToString(lockHash[:])[:7]
 
-	tcpNode, localEnode, err := wireP2P(ctx, life, conf, lock)
+	p2pKey := conf.TestConfig.P2PKey
+	if p2pKey == nil {
+		var err error
+		p2pKey, err = p2p.LoadPrivKey(conf.DataDir)
+		if err != nil {
+			return err
+		}
+	}
+
+	tcpNode, localEnode, err := wireP2P(ctx, life, conf, lock, p2pKey)
 	if err != nil {
 		return err
 	}
@@ -182,7 +192,7 @@ func Run(ctx context.Context, conf Config) (err error) {
 
 	wireMonitoringAPI(life, conf.MonitoringAddr, localEnode)
 
-	if err := wireCoreWorkflow(ctx, life, conf, lock, nodeIdx, tcpNode); err != nil {
+	if err := wireCoreWorkflow(ctx, life, conf, lock, nodeIdx, tcpNode, p2pKey); err != nil {
 		return err
 	}
 
@@ -191,16 +201,8 @@ func Run(ctx context.Context, conf Config) (err error) {
 }
 
 // wireP2P constructs the p2p tcp (libp2p) and udp (discv5) nodes and registers it with the life cycle manager.
-func wireP2P(ctx context.Context, life *lifecycle.Manager, conf Config, lock cluster.Lock,
+func wireP2P(ctx context.Context, life *lifecycle.Manager, conf Config, lock cluster.Lock, p2pKey *ecdsa.PrivateKey,
 ) (host.Host, *enode.LocalNode, error) {
-	p2pKey := conf.TestConfig.P2PKey
-	if p2pKey == nil {
-		var err error
-		p2pKey, err = p2p.LoadPrivKey(conf.DataDir)
-		if err != nil {
-			return nil, nil, err
-		}
-	}
 
 	peers, err := lock.Peers()
 	if err != nil {
@@ -260,7 +262,7 @@ func wireP2P(ctx context.Context, life *lifecycle.Manager, conf Config, lock clu
 
 // wireCoreWorkflow wires the core workflow components.
 func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
-	lock cluster.Lock, nodeIdx cluster.NodeIdx, tcpNode host.Host,
+	lock cluster.Lock, nodeIdx cluster.NodeIdx, tcpNode host.Host, p2pKey *ecdsa.PrivateKey,
 ) error {
 	// Convert and prep public keys and public shares
 	var (
@@ -345,15 +347,6 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 		return err
 	}
 
-	var lcastTransport leadercast.Transport
-	if conf.TestConfig.LcastTransportFunc != nil {
-		lcastTransport = conf.TestConfig.LcastTransportFunc()
-	} else {
-		lcastTransport = leadercast.NewP2PTransport(tcpNode, nodeIdx.PeerIdx, peerIDs)
-	}
-
-	consensus := leadercast.New(lcastTransport, nodeIdx.PeerIdx, len(peerIDs))
-
 	dutyDB := dutydb.NewMemDB()
 
 	vapi, err := validatorapi.NewComponent(eth2Cl, pubSharesByKey, nodeIdx.ShareIdx)
@@ -388,7 +381,12 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 		return err
 	}
 
-	core.Wire(sched, fetch, consensus, dutyDB, vapi,
+	cons, startCons, err := newConsensus(conf, lock, tcpNode, p2pKey, nodeIdx)
+	if err != nil {
+		return err
+	}
+
+	core.Wire(sched, fetch, cons, dutyDB, vapi,
 		parSigDB, parSigEx, sigAgg, aggSigDB, broadcaster,
 		core.WithTracing(),
 		core.WithAsyncRetry(retryer),
@@ -403,8 +401,8 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 		sigAgg.Subscribe(conf.TestConfig.BroadcastCallback)
 	}
 
-	life.RegisterStart(lifecycle.AsyncAppCtx, lifecycle.StartLeaderCast, lifecycle.HookFunc(consensus.Run))
 	life.RegisterStart(lifecycle.AsyncBackground, lifecycle.StartScheduler, lifecycle.HookFuncErr(sched.Run))
+	life.RegisterStart(lifecycle.AsyncAppCtx, lifecycle.StartP2PConsensus, startCons)
 	life.RegisterStart(lifecycle.AsyncAppCtx, lifecycle.StartAggSigDB, lifecycle.HookFuncCtx(aggSigDB.Run))
 	life.RegisterStop(lifecycle.StopScheduler, lifecycle.HookFuncMin(sched.Stop))
 	life.RegisterStop(lifecycle.StopDutyDB, lifecycle.HookFuncMin(dutyDB.Shutdown))
@@ -413,6 +411,40 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 	return nil
 }
 
+// newConsensus returns a new consensus component and its start lifecycle hook.
+func newConsensus(conf Config, lock cluster.Lock, tcpNode host.Host, p2pKey *ecdsa.PrivateKey,
+	nodeIdx cluster.NodeIdx,
+) (core.Consensus, lifecycle.IHookFunc, error) {
+	peers, err := lock.Peers()
+	if err != nil {
+		return nil, nil, err
+	}
+	peerIDs, err := lock.PeerIDs()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if featureset.Enabled(featureset.QBFTConsensus) {
+		comp, err := consensus.New(tcpNode, peers, p2pKey)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		return comp, lifecycle.HookFuncCtx(comp.Start), nil
+	}
+
+	var lcastTransport leadercast.Transport
+	if conf.TestConfig.LcastTransportFunc != nil {
+		lcastTransport = conf.TestConfig.LcastTransportFunc()
+	} else {
+		lcastTransport = leadercast.NewP2PTransport(tcpNode, nodeIdx.PeerIdx, peerIDs)
+	}
+
+	lcast := leadercast.New(lcastTransport, nodeIdx.PeerIdx, len(peerIDs))
+
+	return lcast, lifecycle.HookFunc(lcast.Run), nil
+}
+
 // createMockValidators creates mock validators identified by their public shares.
 func createMockValidators(pubkeys []eth2p0.BLSPubKey) beaconmock.ValidatorSet {
 	resp := make(beaconmock.ValidatorSet)

app/lifecycle/order.go

@@ -31,7 +31,7 @@ const (
 	StartMonitoringAPI
 	StartValidatorAPI
 	StartP2PPing
-	StartLeaderCast
+	StartP2PConsensus
 	StartSimulator
 	StartScheduler
 )

core/consensus/component.go

@@ -42,10 +42,8 @@ const (
 	protocolID = "/charon/consensus/qbft/1.0.0"
 )
 
-// NewComponent returns a new consensus QBFT component.
-func NewComponent(tcpNode host.Host, peers []p2p.Peer,
-	peerIdx int64, p2pKey *ecdsa.PrivateKey,
-) (*Component, error) {
+// New returns a new consensus QBFT component.
+func New(tcpNode host.Host, peers []p2p.Peer, p2pKey *ecdsa.PrivateKey) (*Component, error) {
 	// Extract peer pubkeys.
 	keys := make(map[int64]*ecdsa.PublicKey)
 	for i, p := range peers {
@@ -72,7 +70,7 @@ func NewComponent(tcpNode host.Host, peers []p2p.Peer,
 		// IsLeader is a deterministic leader election function.
 		IsLeader: func(duty core.Duty, round, process int64) bool {
 			mod := ((duty.Slot) + int64(duty.Type) + round) % int64(len(peers))
-			return mod == peerIdx
+			return mod == process
 		},
 
 		// Decide sends consensus output to subscribers.
@@ -115,7 +113,6 @@ type Component struct {
 	peers   []p2p.Peer
 	pubkeys map[int64]*ecdsa.PublicKey
 	privkey *ecdsa.PrivateKey
-	peerIdx int64
 	def     qbft.Definition[core.Duty, [32]byte]
 	subs    []func(ctx context.Context, duty core.Duty, set core.UnsignedDataSet) error
 
@@ -130,7 +127,7 @@ func (c *Component) Subscribe(fn func(ctx context.Context, duty core.Duty, set c
 	c.subs = append(c.subs, fn)
 }
 
-// Start registers the libp2p receive handler.
+// Start registers the libp2p receive handler. This should only be called once.
 func (c *Component) Start(ctx context.Context) {
 	c.tcpNode.SetStreamHandler(protocolID, c.makeHandler(ctx))
 }
@@ -142,6 +139,8 @@ func (c *Component) Propose(ctx context.Context, duty core.Duty, data core.Unsig
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
+	log.Debug(ctx, "Starting qbft consensus instance", z.Any("duty", duty))
+
 	// Hash the proposed data, since qbft ony supports simple comparable values.
 	value := core.UnsignedDataSetToProto(data)
 	hash, err := hashProto(value)
@@ -166,12 +165,18 @@ func (c *Component) Propose(ctx context.Context, duty core.Duty, data core.Unsig
 		Receive:   t.recvBuffer,
 	}
 
+	peerIdx, err := c.getPeerIdx()
+	if err != nil {
+		return err
+	}
+
 	// Run the algo, blocking until the context is cancelled.
-	return qbft.Run[core.Duty, [32]byte](ctx, c.def, qt, duty, c.peerIdx, hash)
+	return qbft.Run[core.Duty, [32]byte](ctx, c.def, qt, duty, peerIdx, hash)
 }
 
 // makeHandler returns a consensus libp2p handler.
 func (c *Component) makeHandler(ctx context.Context) func(s network.Stream) {
+	ctx = log.WithTopic(ctx, "qbft")
 	return func(s network.Stream) {
 		defer s.Close()
 
@@ -188,13 +193,13 @@ func (c *Component) makeHandler(ctx context.Context) func(s network.Stream) {
 		}
 
 		if pbMsg.Msg == nil || pbMsg.Msg.Duty == nil {
-			log.Error(ctx, "Invalid consensus message", err)
+			log.Error(ctx, "Invalid consensus message", errors.New("nil msg"))
 			return
 		}
 
 		duty := core.DutyFromProto(pbMsg.Msg.Duty)
 		if !duty.Type.Valid() {
-			log.Error(ctx, "Invalid duty type", err)
+			log.Error(ctx, "Invalid duty type", errors.New("", z.Str("type", duty.Type.String())))
 			return
 		}
 
@@ -247,3 +252,18 @@ func (c *Component) deleteRecvChan(duty core.Duty) {
 
 	delete(c.recvBuffers, duty)
 }
+
+// getPeerIdx returns the local peer index.
+func (c *Component) getPeerIdx() (int64, error) {
+	peerIdx := int64(-1)
+	for i, p := range c.peers {
+		if c.tcpNode.ID() == p.ID {
+			peerIdx = int64(i)
+		}
+	}
+	if peerIdx == -1 {
+		return 0, errors.New("local libp2p host not in peer list")
+	}
+
+	return peerIdx, nil
+}