[Consensus] Approval Processing Core -> Sealing Core

Makefile

@@ -112,6 +112,7 @@ generate-mocks:
 	GO111MODULE=on mockery -name '.*' -dir="state/protocol/events" -case=underscore -output="./state/protocol/events/mock" -outpkg="mock"
 	GO111MODULE=on mockery -name '.*' -dir=engine/execution/computation/computer -case=underscore -output="./engine/execution/computation/computer/mock" -outpkg="mock"
 	GO111MODULE=on mockery -name '.*' -dir=engine/execution/state -case=underscore -output="./engine/execution/state/mock" -outpkg="mock"
+	GO111MODULE=on mockery -name '.*' -dir=engine/consensus -case=underscore -output="./engine/consensus/mock" -outpkg="mock"
 	GO111MODULE=on mockery -name '.*' -dir=fvm -case=underscore -output="./fvm/mock" -outpkg="mock"
 	GO111MODULE=on mockery -name '.*' -dir=fvm/state -case=underscore -output="./fvm/mock/state" -outpkg="mock"
 	GO111MODULE=on mockery -name '.*' -dir=ledger -case=underscore -output="./ledger/mock" -outpkg="mock"

cmd/consensus/main.go

@@ -15,6 +15,7 @@ import (
 	"github.com/onflow/flow-go/consensus/hotstuff"
 	"github.com/onflow/flow-go/consensus/hotstuff/blockproducer"
 	"github.com/onflow/flow-go/consensus/hotstuff/committees"
+	"github.com/onflow/flow-go/consensus/hotstuff/notifications/pubsub"
 	"github.com/onflow/flow-go/consensus/hotstuff/pacemaker/timeout"
 	"github.com/onflow/flow-go/consensus/hotstuff/persister"
 	"github.com/onflow/flow-go/consensus/hotstuff/verification"
@@ -24,6 +25,7 @@ import (
 	synceng "github.com/onflow/flow-go/engine/common/synchronization"
 	"github.com/onflow/flow-go/engine/consensus/compliance"
 	"github.com/onflow/flow-go/engine/consensus/ingestion"
+	"github.com/onflow/flow-go/engine/consensus/matching"
 	"github.com/onflow/flow-go/engine/consensus/provider"
 	"github.com/onflow/flow-go/engine/consensus/sealing"
 	"github.com/onflow/flow-go/model/bootstrap"
@@ -73,24 +75,22 @@ func main() {
 		requiredApprovalsForSealConstruction   uint
 		emergencySealing                       bool
 
-		err               error
-		mutableState      protocol.MutableState
-		privateDKGData    *bootstrap.DKGParticipantPriv
-		guarantees        mempool.Guarantees
-		results           mempool.IncorporatedResults
-		receipts          mempool.ExecutionTree
-		approvals         mempool.Approvals
-		seals             mempool.IncorporatedResultSeals
-		pendingReceipts   mempool.PendingReceipts
-		prov              *provider.Engine
-		receiptRequester  *requester.Engine
-		syncCore          *synchronization.Core
-		comp              *compliance.Engine
-		conMetrics        module.ConsensusMetrics
-		mainMetrics       module.HotstuffMetrics
-		receiptValidator  module.ReceiptValidator
-		approvalValidator module.ApprovalValidator
-		chunkAssigner     *chmodule.ChunkAssigner
+		err                     error
+		mutableState            protocol.MutableState
+		privateDKGData          *bootstrap.DKGParticipantPriv
+		guarantees              mempool.Guarantees
+		receipts                mempool.ExecutionTree
+		seals                   mempool.IncorporatedResultSeals
+		pendingReceipts         mempool.PendingReceipts
+		prov                    *provider.Engine
+		receiptRequester        *requester.Engine
+		syncCore                *synchronization.Core
+		comp                    *compliance.Engine
+		conMetrics              module.ConsensusMetrics
+		mainMetrics             module.HotstuffMetrics
+		receiptValidator        module.ReceiptValidator
+		chunkAssigner           *chmodule.ChunkAssigner
+		finalizationDistributor *pubsub.FinalizationDistributor
 	)
 
 	cmd.FlowNode(flow.RoleConsensus.String()).
@@ -150,10 +150,6 @@ func main() {
 
 			resultApprovalSigVerifier := signature.NewAggregationVerifier(encoding.ResultApprovalTag)
 
-			approvalValidator = validation.NewApprovalValidator(
-				node.State,
-				resultApprovalSigVerifier)
-
 			sealValidator := validation.NewSealValidator(
 				node.State,
 				node.Storage.Headers,
@@ -183,10 +179,6 @@ func main() {
 			guarantees, err = stdmap.NewGuarantees(guaranteeLimit)
 			return err
 		}).
-		Module("execution results mempool", func(node *cmd.FlowNodeBuilder) error {
-			results, err = stdmap.NewIncorporatedResults(resultLimit)
-			return err
-		}).
 		Module("execution receipts mempool", func(node *cmd.FlowNodeBuilder) error {
 			receipts = consensusMempools.NewExecutionTree()
 			// registers size method of backend for metrics
@@ -196,10 +188,6 @@ func main() {
 			}
 			return nil
 		}).
-		Module("result approvals mempool", func(node *cmd.FlowNodeBuilder) error {
-			approvals, err = stdmap.NewApprovals(approvalLimit)
-			return err
-		}).
 		Module("block seals mempool", func(node *cmd.FlowNodeBuilder) error {
 			// use a custom ejector so we don't eject seals that would break
 			// the chain of seals
@@ -223,8 +211,43 @@ func main() {
 			syncCore, err = synchronization.New(node.Logger, synchronization.DefaultConfig())
 			return err
 		}).
+		Module("finalization distributor", func(node *cmd.FlowNodeBuilder) error {
+			finalizationDistributor = pubsub.NewFinalizationDistributor()
+			return nil
+		}).
 		Component("sealing engine", func(node *cmd.FlowNodeBuilder) (module.ReadyDoneAware, error) {
 
+			resultApprovalSigVerifier := signature.NewAggregationVerifier(encoding.ResultApprovalTag)
+
+			config := sealing.DefaultConfig()
+			config.EmergencySealingActive = emergencySealing
+			config.RequiredApprovalsForSealConstruction = requiredApprovalsForSealConstruction
+
+			e, err := sealing.NewEngine(
+				node.Logger,
+				node.Tracer,
+				conMetrics,
+				node.Metrics.Engine,
+				node.Metrics.Mempool,
+				node.Network,
+				node.Me,
+				node.Storage.Headers,
+				node.Storage.Payloads,
+				node.State,
+				node.Storage.Seals,
+				chunkAssigner,
+				resultApprovalSigVerifier,
+				seals,
+				config,
+			)
+
+			// subscribe for finalization events from hotstuff
+			finalizationDistributor.AddOnBlockFinalizedConsumer(e.OnFinalizedBlock)
+			finalizationDistributor.AddOnBlockIncorporatedConsumer(e.OnBlockIncorporated)
+
+			return e, err
+		}).
+		Component("matching engine", func(node *cmd.FlowNodeBuilder) (module.ReadyDoneAware, error) {
 			receiptRequester, err = requester.New(
 				node.Logger,
 				node.Metrics.Engine,
@@ -241,34 +264,39 @@ func main() {
 				return nil, err
 			}
 
-			match, err := sealing.NewEngine(
+			core := matching.NewCore(
 				node.Logger,
-				node.Metrics.Engine,
 				node.Tracer,
-				node.Metrics.Mempool,
 				conMetrics,
-				node.Network,
+				node.Metrics.Mempool,
 				node.State,
-				node.Me,
-				receiptRequester,
-				node.Storage.Receipts,
 				node.Storage.Headers,
-				node.Storage.Index,
-				results,
+				node.Storage.Receipts,
 				receipts,
-				approvals,
-				seals,
 				pendingReceipts,
-				chunkAssigner,
+				seals,
 				receiptValidator,
-				approvalValidator,
-				requiredApprovalsForSealConstruction,
-				emergencySealing,
+				receiptRequester,
+				matching.DefaultConfig(),
 			)
 
-			receiptRequester.WithHandle(match.HandleReceipt)
+			e, err := matching.NewEngine(
+				node.Logger,
+				node.Network,
+				node.Me,
+				node.Metrics.Engine,
+				node.Metrics.Mempool,
+				core,
+			)
+			if err != nil {
+				return nil, err
+			}
 
-			return match, err
+			// subscribe engine to inputs from other node-internal components
+			receiptRequester.WithHandle(e.HandleReceipt)
+			finalizationDistributor.AddOnBlockFinalizedConsumer(e.OnFinalizedBlock)
+
+			return e, err
 		}).
 		Component("provider engine", func(node *cmd.FlowNodeBuilder) (module.ReadyDoneAware, error) {
 			prov, err = provider.New(
@@ -404,7 +432,9 @@ func main() {
 				node.Storage.Index,
 				node.RootChainID,
 			)
-			// make compliance engine as a FinalizationConsumer
+
+			notifier.AddConsumer(finalizationDistributor)
+
 			// initialize the persister
 			persist := persister.New(node.DB, node.RootChainID)
 

cmd/consensus/notifier.go

@@ -3,7 +3,6 @@ package main
 import (
 	"github.com/rs/zerolog"
 
-	"github.com/onflow/flow-go/consensus/hotstuff"
 	"github.com/onflow/flow-go/consensus/hotstuff/notifications"
 	"github.com/onflow/flow-go/consensus/hotstuff/notifications/pubsub"
 	"github.com/onflow/flow-go/model/flow"
@@ -13,7 +12,7 @@ import (
 )
 
 func createNotifier(log zerolog.Logger, metrics module.HotstuffMetrics, tracer module.Tracer, index storage.Index, chain flow.ChainID,
-) hotstuff.Consumer {
+) *pubsub.Distributor {
 	telemetryConsumer := notifications.NewTelemetryConsumer(log, chain)
 	tracingConsumer := notifications.NewConsensusTracingConsumer(log, tracer, index)
 	metricsConsumer := metricsconsumer.NewMetricsConsumer(metrics)

consensus/hotstuff/notifications/pubsub/finalization_distributor.go

@@ -0,0 +1,83 @@
+package pubsub
+
+import (
+	"sync"
+
+	"github.com/onflow/flow-go/consensus/hotstuff/model"
+	"github.com/onflow/flow-go/model/flow"
+)
+
+type OnBlockFinalizedConsumer = func(finalizedBlockID flow.Identifier)
+type OnBlockIncorporatedConsumer = func(incorporatedBlockID flow.Identifier)
+
+// FinalizationDistributor subscribes for finalization events from hotstuff and distributes it to subscribers
+type FinalizationDistributor struct {
+	blockFinalizedConsumers    []OnBlockFinalizedConsumer
+	blockIncorporatedConsumers []OnBlockIncorporatedConsumer
+	lock                       sync.RWMutex
+}
+
+func NewFinalizationDistributor() *FinalizationDistributor {
+	return &FinalizationDistributor{
+		blockFinalizedConsumers:    make([]OnBlockFinalizedConsumer, 0),
+		blockIncorporatedConsumers: make([]OnBlockIncorporatedConsumer, 0),
+		lock:                       sync.RWMutex{},
+	}
+}
+
+func (p *FinalizationDistributor) AddOnBlockFinalizedConsumer(consumer OnBlockFinalizedConsumer) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	p.blockFinalizedConsumers = append(p.blockFinalizedConsumers, consumer)
+}
+func (p *FinalizationDistributor) AddOnBlockIncorporatedConsumer(consumer OnBlockIncorporatedConsumer) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	p.blockIncorporatedConsumers = append(p.blockIncorporatedConsumers, consumer)
+}
+
+func (p *FinalizationDistributor) OnEventProcessed() {}
+
+func (p *FinalizationDistributor) OnBlockIncorporated(block *model.Block) {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+	for _, consumer := range p.blockIncorporatedConsumers {
+		consumer(block.BlockID)
+	}
+}
+
+func (p *FinalizationDistributor) OnFinalizedBlock(block *model.Block) {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+	for _, consumer := range p.blockFinalizedConsumers {
+		consumer(block.BlockID)
+	}
+}
+
+func (p *FinalizationDistributor) OnDoubleProposeDetected(*model.Block, *model.Block) {}
+
+func (p *FinalizationDistributor) OnReceiveVote(uint64, *model.Vote) {}
+
+func (p *FinalizationDistributor) OnReceiveProposal(uint64, *model.Proposal) {}
+
+func (p *FinalizationDistributor) OnEnteringView(uint64, flow.Identifier) {}
+
+func (p *FinalizationDistributor) OnQcTriggeredViewChange(*flow.QuorumCertificate, uint64) {}
+
+func (p *FinalizationDistributor) OnProposingBlock(*model.Proposal) {}
+
+func (p *FinalizationDistributor) OnVoting(*model.Vote) {}
+
+func (p *FinalizationDistributor) OnQcConstructedFromVotes(*flow.QuorumCertificate) {}
+
+func (p *FinalizationDistributor) OnStartingTimeout(*model.TimerInfo) {}
+
+func (p *FinalizationDistributor) OnReachedTimeout(*model.TimerInfo) {}
+
+func (p *FinalizationDistributor) OnQcIncorporated(*flow.QuorumCertificate) {}
+
+func (p *FinalizationDistributor) OnForkChoiceGenerated(uint64, *flow.QuorumCertificate) {}
+
+func (p *FinalizationDistributor) OnDoubleVotingDetected(*model.Vote, *model.Vote) {}
+
+func (p *FinalizationDistributor) OnInvalidVoteDetected(*model.Vote) {}

engine/consensus/approvals/approval_collector.go

@@ -13,13 +13,12 @@ import (
 // collecting aggregated signatures for chunks that reached seal construction threshold,
 // creating and submitting seal candidates once signatures for every chunk are aggregated.
 type ApprovalCollector struct {
-	incorporatedBlock                    *flow.Header                    // block that incorporates execution result
-	incorporatedResult                   *flow.IncorporatedResult        // incorporated result that is being sealed
-	chunkCollectors                      []*ChunkApprovalCollector       // slice of chunk collectorTree that is created on construction and doesn't change
-	aggregatedSignatures                 *AggregatedSignatures           // aggregated signature for each chunk
-	seals                                mempool.IncorporatedResultSeals // holds candidate seals for incorporated results that have acquired sufficient approvals; candidate seals are constructed  without consideration of the sealability of parent results
-	numberOfChunks                       int                             // number of chunks for execution result, remains constant
-	requiredApprovalsForSealConstruction uint                            // min number of approvals required for constructing a candidate seal
+	incorporatedBlock    *flow.Header                    // block that incorporates execution result
+	incorporatedResult   *flow.IncorporatedResult        // incorporated result that is being sealed
+	chunkCollectors      []*ChunkApprovalCollector       // slice of chunk collectorTree that is created on construction and doesn't change
+	aggregatedSignatures *AggregatedSignatures           // aggregated signature for each chunk
+	seals                mempool.IncorporatedResultSeals // holds candidate seals for incorporated results that have acquired sufficient approvals; candidate seals are constructed  without consideration of the sealability of parent results
+	numberOfChunks       int                             // number of chunks for execution result, remains constant
 }
 
 func NewApprovalCollector(result *flow.IncorporatedResult, incorporatedBlock *flow.Header, assignment *chunks.Assignment, seals mempool.IncorporatedResultSeals, requiredApprovalsForSealConstruction uint) *ApprovalCollector {
@@ -32,13 +31,12 @@ func NewApprovalCollector(result *flow.IncorporatedResult, incorporatedBlock *fl
 
 	numberOfChunks := result.Result.Chunks.Len()
 	return &ApprovalCollector{
-		incorporatedResult:                   result,
-		incorporatedBlock:                    incorporatedBlock,
-		numberOfChunks:                       numberOfChunks,
-		chunkCollectors:                      chunkCollectors,
-		requiredApprovalsForSealConstruction: requiredApprovalsForSealConstruction,
-		aggregatedSignatures:                 NewAggregatedSignatures(uint64(numberOfChunks)),
-		seals:                                seals,
+		incorporatedResult:   result,
+		incorporatedBlock:    incorporatedBlock,
+		numberOfChunks:       numberOfChunks,
+		chunkCollectors:      chunkCollectors,
+		aggregatedSignatures: NewAggregatedSignatures(uint64(numberOfChunks)),
+		seals:                seals,
 	}
 }
 

engine/consensus/approvals/approval_collector_test.go

@@ -48,11 +48,17 @@ func (s *ApprovalCollectorTestSuite) TestProcessApproval_ValidApproval() {
 // met for each chunk.
 func (s *ApprovalCollectorTestSuite) TestProcessApproval_SealResult() {
 	expectedSignatures := make([]flow.AggregatedSignature, s.IncorporatedResult.Result.Chunks.Len())
-	s.sealsPL.On("Add", mock.Anything).Return(true, nil).Once()
+	s.sealsPL.On("Add", mock.Anything).Run(
+		func(args mock.Arguments) {
+			seal := args.Get(0).(*flow.IncorporatedResultSeal)
+			require.Equal(s.T(), s.Block.ID(), seal.Seal.BlockID)
+			require.Equal(s.T(), s.IncorporatedResult.Result.ID(), seal.Seal.ResultID)
+		},
+	).Return(true, nil).Once()
 
 	for i, chunk := range s.Chunks {
 		var err error
-		sigCollector := flow.NewSignatureCollector()
+		sigCollector := NewSignatureCollector()
 		for verID := range s.AuthorizedVerifiers {
 			approval := unittest.ResultApprovalFixture(unittest.WithChunk(chunk.Index), unittest.WithApproverID(verID))
 			err = s.collector.ProcessApproval(approval)