Merge pull request onflow#910 from onflow/yurii/5615-synchronization-…

…engine-queue

[Consensus] Queue for `synchronization.Engine`

cmd/access/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/onflow/flow-go/cmd"
 	"github.com/onflow/flow-go/consensus"
 	"github.com/onflow/flow-go/consensus/hotstuff/committees"
+	"github.com/onflow/flow-go/consensus/hotstuff/notifications/pubsub"
 	"github.com/onflow/flow-go/consensus/hotstuff/verification"
 	recovery "github.com/onflow/flow-go/consensus/recovery/protocol"
 	"github.com/onflow/flow-go/engine"
@@ -59,6 +60,7 @@ func main() {
 		syncCore                     *synchronization.Core
 		rpcConf                      rpc.Config
 		rpcEng                       *rpc.Engine
+		finalizationDistributor      *pubsub.FinalizationDistributor
 		collectionRPC                access.AccessAPIClient
 		executionNodeAddress         string // deprecated
 		historicalAccessRPCs         []access.AccessAPIClient
@@ -275,9 +277,12 @@ func main() {
 				return nil, fmt.Errorf("could not find latest finalized block and pending blocks to recover consensus follower: %w", err)
 			}
 
+			finalizationDistributor = pubsub.NewFinalizationDistributor()
+			finalizationDistributor.AddConsumer(ingestEng)
+
 			// creates a consensus follower with ingestEngine as the notifier
 			// so that it gets notified upon each new finalized block
-			followerCore, err := consensus.NewFollower(node.Logger, committee, node.Storage.Headers, final, verifier, ingestEng, node.RootBlock.Header, node.RootQC, finalized, pending)
+			followerCore, err := consensus.NewFollower(node.Logger, committee, node.Storage.Headers, final, verifier, finalizationDistributor, node.RootBlock.Header, node.RootQC, finalized, pending)
 			if err != nil {
 				return nil, fmt.Errorf("could not initialize follower core: %w", err)
 			}
@@ -316,6 +321,9 @@ func main() {
 			if err != nil {
 				return nil, fmt.Errorf("could not create synchronization engine: %w", err)
 			}
+
+			finalizationDistributor.AddOnBlockFinalizedConsumer(sync.OnFinalizedBlock)
+
 			return sync, nil
 		}).
 		Component("ping engine", func(node *cmd.FlowNodeBuilder) (module.ReadyDoneAware, error) {

cmd/collection/main.go

@@ -9,7 +9,7 @@ import (
 	"github.com/onflow/flow-go/cmd"
 	"github.com/onflow/flow-go/consensus"
 	"github.com/onflow/flow-go/consensus/hotstuff/committees"
-	"github.com/onflow/flow-go/consensus/hotstuff/notifications"
+	"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/verification"
 	recovery "github.com/onflow/flow-go/consensus/recovery/protocol"
@@ -64,8 +64,9 @@ func main() {
 		ingestConf    ingest.Config
 		ingressConf   ingress.Config
 
-		pools          *epochpool.TransactionPools // epoch-scoped transaction pools
-		followerBuffer *buffer.PendingBlocks       // pending block cache for follower
+		pools                   *epochpool.TransactionPools // epoch-scoped transaction pools
+		followerBuffer          *buffer.PendingBlocks       // pending block cache for follower
+		finalizationDistributor *pubsub.FinalizationDistributor
 
 		push              *pusher.Engine
 		ing               *ingest.Engine
@@ -184,8 +185,7 @@ func main() {
 			// initialize the verifier for the protocol consensus
 			verifier := verification.NewCombinedVerifier(mainConsensusCommittee, staking, beacon, merger)
 
-			// use proper engine for notifier to follower
-			notifier := notifications.NewNoopConsumer()
+			finalizationDistributor = pubsub.NewFinalizationDistributor()
 
 			finalized, pending, err := recovery.FindLatest(node.State, node.Storage.Headers)
 			if err != nil {
@@ -199,7 +199,7 @@ func main() {
 				node.Storage.Headers,
 				finalizer,
 				verifier,
-				notifier,
+				finalizationDistributor,
 				node.RootBlock.Header,
 				node.RootQC,
 				finalized,
@@ -246,6 +246,8 @@ func main() {
 				return nil, fmt.Errorf("could not create synchronization engine: %w", err)
 			}
 
+			finalizationDistributor.AddOnBlockFinalizedConsumer(sync.OnFinalizedBlock)
+
 			return sync, nil
 		}).
 		Component("ingestion engine", func(node *cmd.FlowNodeBuilder) (module.ReadyDoneAware, error) {

cmd/consensus/main.go

@@ -494,6 +494,8 @@ func main() {
 				return nil, fmt.Errorf("could not initialize synchronization engine: %w", err)
 			}
 
+			finalizationDistributor.AddOnBlockFinalizedConsumer(sync.OnFinalizedBlock)
+
 			return sync, nil
 		}).
 		Component("receipt requester engine", func(node *cmd.FlowNodeBuilder) (module.ReadyDoneAware, error) {

cmd/execution/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/onflow/flow-go/cmd"
 	"github.com/onflow/flow-go/consensus"
 	"github.com/onflow/flow-go/consensus/hotstuff/committees"
+	"github.com/onflow/flow-go/consensus/hotstuff/notifications/pubsub"
 	"github.com/onflow/flow-go/consensus/hotstuff/verification"
 	recovery "github.com/onflow/flow-go/consensus/recovery/protocol"
 	"github.com/onflow/flow-go/engine"
@@ -70,6 +71,7 @@ func main() {
 		computationManager          *computation.Manager
 		collectionRequester         *requester.Engine
 		ingestionEng                *ingestion.Engine
+		finalizationDistributor     *pubsub.FinalizationDistributor
 		rpcConf                     rpc.Config
 		err                         error
 		executionState              state.ExecutionState
@@ -380,9 +382,12 @@ func main() {
 				return nil, fmt.Errorf("could not find latest finalized block and pending blocks to recover consensus follower: %w", err)
 			}
 
+			finalizationDistributor = pubsub.NewFinalizationDistributor()
+			finalizationDistributor.AddConsumer(checkerEng)
+
 			// creates a consensus follower with ingestEngine as the notifier
 			// so that it gets notified upon each new finalized block
-			followerCore, err := consensus.NewFollower(node.Logger, committee, node.Storage.Headers, final, verifier, checkerEng, node.RootBlock.Header, node.RootQC, finalized, pending)
+			followerCore, err := consensus.NewFollower(node.Logger, committee, node.Storage.Headers, final, verifier, finalizationDistributor, node.RootBlock.Header, node.RootQC, finalized, pending)
 			if err != nil {
 				return nil, fmt.Errorf("could not create follower core logic: %w", err)
 			}
@@ -427,7 +432,7 @@ func main() {
 			)
 			return eng, err
 		}).
-		Component("sychronization engine", func(node *cmd.FlowNodeBuilder) (module.ReadyDoneAware, error) {
+		Component("synchronization engine", func(node *cmd.FlowNodeBuilder) (module.ReadyDoneAware, error) {
 			// initialize the synchronization engine
 			syncEngine, err = synchronization.New(
 				node.Logger,
@@ -443,6 +448,8 @@ func main() {
 				return nil, fmt.Errorf("could not initialize synchronization engine: %w", err)
 			}
 
+			finalizationDistributor.AddOnBlockFinalizedConsumer(syncEngine.OnFinalizedBlock)
+
 			return syncEngine, nil
 		}).
 		Component("grpc server", func(node *cmd.FlowNodeBuilder) (module.ReadyDoneAware, error) {

cmd/verification/main.go

@@ -9,6 +9,7 @@ import (
 	"github.com/onflow/flow-go/cmd"
 	"github.com/onflow/flow-go/consensus"
 	"github.com/onflow/flow-go/consensus/hotstuff/committees"
+	"github.com/onflow/flow-go/consensus/hotstuff/notifications/pubsub"
 	"github.com/onflow/flow-go/consensus/hotstuff/verification"
 	recovery "github.com/onflow/flow-go/consensus/recovery/protocol"
 	followereng "github.com/onflow/flow-go/engine/common/follower"
@@ -60,14 +61,15 @@ func main() {
 		processedBlockHeight *storage.ConsumerProgress // used in block consumer
 		chunkQueue           *storage.ChunksQueue      // used in chunk consumer
 
-		syncCore        *synchronization.Core // used in follower engine
-		pendingBlocks   *buffer.PendingBlocks // used in follower engine
-		assignerEngine  *assigner.Engine      // the assigner engine
-		fetcherEngine   *fetcher.Engine       // the fetcher engine
-		requesterEngine *vereq.Engine         // the requester engine
-		verifierEng     *verifier.Engine      // the verifier engine
-		chunkConsumer   *chunkconsumer.ChunkConsumer
-		blockConsumer   *blockconsumer.BlockConsumer
+		syncCore                *synchronization.Core // used in follower engine
+		pendingBlocks           *buffer.PendingBlocks // used in follower engine
+		assignerEngine          *assigner.Engine      // the assigner engine
+		fetcherEngine           *fetcher.Engine       // the fetcher engine
+		requesterEngine         *vereq.Engine         // the requester engine
+		verifierEng             *verifier.Engine      // the verifier engine
+		chunkConsumer           *chunkconsumer.ChunkConsumer
+		blockConsumer           *blockconsumer.BlockConsumer
+		finalizationDistributor *pubsub.FinalizationDistributor
 
 		followerEng *followereng.Engine        // the follower engine
 		collector   module.VerificationMetrics // used to collect metrics of all engines
@@ -297,9 +299,12 @@ func main() {
 				return nil, fmt.Errorf("could not find latest finalized block and pending blocks to recover consensus follower: %w", err)
 			}
 
+			finalizationDistributor = pubsub.NewFinalizationDistributor()
+			finalizationDistributor.AddConsumer(blockConsumer)
+
 			// creates a consensus follower with ingestEngine as the notifier
 			// so that it gets notified upon each new finalized block
-			followerCore, err := consensus.NewFollower(node.Logger, committee, node.Storage.Headers, final, verifier, blockConsumer, node.RootBlock.Header,
+			followerCore, err := consensus.NewFollower(node.Logger, committee, node.Storage.Headers, final, verifier, finalizationDistributor, node.RootBlock.Header,
 				node.RootQC, finalized, pending)
 			if err != nil {
 				return nil, fmt.Errorf("could not create follower core logic: %w", err)
@@ -339,6 +344,9 @@ func main() {
 			if err != nil {
 				return nil, fmt.Errorf("could not create synchronization engine: %w", err)
 			}
+
+			finalizationDistributor.AddOnBlockFinalizedConsumer(sync.OnFinalizedBlock)
+
 			return sync, nil
 		}).
 		Run()

consensus/hotstuff/notifications/pubsub/finalization_distributor.go

@@ -1,6 +1,7 @@
 package pubsub
 
 import (
+	"github.com/onflow/flow-go/consensus/hotstuff"
 	"sync"
 
 	"github.com/onflow/flow-go/consensus/hotstuff/model"
@@ -12,9 +13,10 @@ 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
+	blockFinalizedConsumers       []OnBlockFinalizedConsumer
+	blockIncorporatedConsumers    []OnBlockIncorporatedConsumer
+	hotStuffFinalizationConsumers []hotstuff.FinalizationConsumer
+	lock                          sync.RWMutex
 }
 
 func NewFinalizationDistributor() *FinalizationDistributor {
@@ -30,12 +32,19 @@ func (p *FinalizationDistributor) AddOnBlockFinalizedConsumer(consumer OnBlockFi
 	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) AddConsumer(consumer hotstuff.FinalizationConsumer) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	p.hotStuffFinalizationConsumers = append(p.hotStuffFinalizationConsumers, consumer)
+}
+
 func (p *FinalizationDistributor) OnEventProcessed() {}
 
 func (p *FinalizationDistributor) OnBlockIncorporated(block *model.Block) {
@@ -44,6 +53,9 @@ func (p *FinalizationDistributor) OnBlockIncorporated(block *model.Block) {
 	for _, consumer := range p.blockIncorporatedConsumers {
 		consumer(block.BlockID)
 	}
+	for _, consumer := range p.hotStuffFinalizationConsumers {
+		consumer.OnBlockIncorporated(block)
+	}
 }
 
 func (p *FinalizationDistributor) OnFinalizedBlock(block *model.Block) {
@@ -52,9 +64,18 @@ func (p *FinalizationDistributor) OnFinalizedBlock(block *model.Block) {
 	for _, consumer := range p.blockFinalizedConsumers {
 		consumer(block.BlockID)
 	}
+	for _, consumer := range p.hotStuffFinalizationConsumers {
+		consumer.OnFinalizedBlock(block)
+	}
 }
 
-func (p *FinalizationDistributor) OnDoubleProposeDetected(*model.Block, *model.Block) {}
+func (p *FinalizationDistributor) OnDoubleProposeDetected(block1, block2 *model.Block) {
+	p.lock.RLock()
+	defer p.lock.RUnlock()
+	for _, consumer := range p.hotStuffFinalizationConsumers {
+		consumer.OnDoubleProposeDetected(block1, block2)
+	}
+}
 
 func (p *FinalizationDistributor) OnReceiveVote(uint64, *model.Vote) {}