Refactor pusher engine: change interface

tim-barry · tim-barry · commit 1fc3a1951baa · 2024-12-03T17:14:07.000-08:00
- Remove pusher engine implementation of network.Engine - Replace with network.MessageProcessor - See: #6747 (comment) - Remove SubmitCollectionGuarantee message type - Was only used between Finalizer and Pusher engine - New interface passes and stores collection guarantees directly, instead of wrapping and then unwrapping them - See: #6747 (comment) - Add GuaranteedCollectionPublisher interface, implemented by pusher engine - Only used by the Finalizer (and intermediate constructors) - Mocks are generated for it, used in Finalizer unit tests - See: #6747 (comment)
diff --git a/engine/collection/epochmgr/factories/builder.go b/engine/collection/epochmgr/factories/builder.go
@@ -6,11 +6,11 @@ import (
 	"github.com/dgraph-io/badger/v2"
 	"github.com/rs/zerolog"
 
+	"github.com/onflow/flow-go/engine/collection"
 	"github.com/onflow/flow-go/module"
 	builder "github.com/onflow/flow-go/module/builder/collection"
 	finalizer "github.com/onflow/flow-go/module/finalizer/collection"
 	"github.com/onflow/flow-go/module/mempool"
-	"github.com/onflow/flow-go/network"
 	clusterstate "github.com/onflow/flow-go/state/cluster"
 	"github.com/onflow/flow-go/state/protocol"
 	"github.com/onflow/flow-go/storage"
@@ -23,7 +23,7 @@ type BuilderFactory struct {
 	trace            module.Tracer
 	opts             []builder.Opt
 	metrics          module.CollectionMetrics
-	pusher           network.Engine // engine for pushing finalized collection to consensus committee
+	pusher           collection.GuaranteedCollectionPublisher // engine for pushing finalized collection to consensus committee
 	log              zerolog.Logger
 }
 
@@ -33,7 +33,7 @@ func NewBuilderFactory(
 	mainChainHeaders storage.Headers,
 	trace module.Tracer,
 	metrics module.CollectionMetrics,
-	pusher network.Engine,
+	pusher collection.GuaranteedCollectionPublisher,
 	log zerolog.Logger,
 	opts ...builder.Opt,
 ) (*BuilderFactory, error) {
diff --git a/engine/collection/guaranteed_collection_publisher.go b/engine/collection/guaranteed_collection_publisher.go
@@ -0,0 +1,15 @@
+package collection
+
+import (
+	"github.com/onflow/flow-go/model/flow"
+)
+
+// GuaranteedCollectionPublisher defines the interface to send collection guarantees
+// from a collection node to consensus nodes. Collection guarantees are broadcast on a best-effort basis,
+// and it is acceptable to discard some guarantees (especially those that are out of date).
+// Implementation is non-blocking and concurrency safe.
+type GuaranteedCollectionPublisher interface {
+	// SubmitCollectionGuarantee adds a guarantee to an internal queue
+	// to be published to consensus nodes.
+	SubmitCollectionGuarantee(guarantee *flow.CollectionGuarantee)
+}
diff --git a/engine/collection/mock/guaranteed_collection_publisher.go b/engine/collection/mock/guaranteed_collection_publisher.go
diff --git a/engine/collection/pusher/engine.go b/engine/collection/pusher/engine.go
@@ -13,7 +13,6 @@ import (
 	"github.com/onflow/flow-go/engine/common/fifoqueue"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
-	"github.com/onflow/flow-go/model/messages"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/component"
 	"github.com/onflow/flow-go/module/irrecoverable"
@@ -44,8 +43,7 @@ type Engine struct {
 	cm *component.ComponentManager
 }
 
-// TODO convert to network.MessageProcessor
-var _ network.Engine = (*Engine)(nil)
+var _ network.MessageProcessor = (*Engine)(nil)
 var _ component.Component = (*Engine)(nil)
 
 // New creates a new pusher engine.
@@ -120,17 +118,17 @@ func (e *Engine) outboundQueueWorker(ctx irrecoverable.SignalerContext, ready co
 // No errors expected during normal operations.
 func (e *Engine) processOutboundMessages(ctx context.Context) error {
 	for {
-		nextMessage, ok := e.queue.Pop()
+		item, ok := e.queue.Pop()
 		if !ok {
 			return nil
 		}
 
-		asSCGMsg, ok := nextMessage.(*messages.SubmitCollectionGuarantee)
+		guarantee, ok := item.(*flow.CollectionGuarantee)
 		if !ok {
-			return fmt.Errorf("invalid message type in pusher engine queue")
+			return fmt.Errorf("invalid type in pusher engine queue")
 		}
 
-		err := e.publishCollectionGuarantee(&asSCGMsg.Guarantee)
+		err := e.publishCollectionGuarantee(guarantee)
 		if err != nil {
 			return err
 		}
@@ -143,44 +141,18 @@ func (e *Engine) processOutboundMessages(ctx context.Context) error {
 	}
 }
 
-// SubmitLocal submits an event originating on the local node.
-func (e *Engine) SubmitLocal(event interface{}) {
-	ev, ok := event.(*messages.SubmitCollectionGuarantee)
-	if ok {
-		e.SubmitCollectionGuarantee(ev)
-	} else {
-		engine.LogError(e.log, fmt.Errorf("invalid message argument to pusher engine"))
-	}
-}
-
-// Submit submits the given event from the node with the given origin ID
-// for processing in a non-blocking manner. It returns instantly and logs
-// a potential processing error internally when done.
-func (e *Engine) Submit(channel channels.Channel, originID flow.Identifier, event interface{}) {
-	engine.LogError(e.log, fmt.Errorf("pusher engine should only receive local messages on the same node"))
-}
-
-// ProcessLocal processes an event originating on the local node.
-func (e *Engine) ProcessLocal(event interface{}) error {
-	ev, ok := event.(*messages.SubmitCollectionGuarantee)
-	if ok {
-		e.SubmitCollectionGuarantee(ev)
-		return nil
-	} else {
-		return fmt.Errorf("invalid message argument to pusher engine")
-	}
-}
-
 // Process processes the given event from the node with the given origin ID in
-// a blocking manner. It returns the potential processing error when done.
+// a non-blocking manner. It returns the potential processing error when done.
+// Because the pusher engine does not accept inputs from the network,
+// always drop any messages and return an error.
 func (e *Engine) Process(channel channels.Channel, originID flow.Identifier, message any) error {
 	return fmt.Errorf("pusher engine should only receive local messages on the same node")
 }
 
 // SubmitCollectionGuarantee adds a collection guarantee to the engine's queue
 // to later be published to consensus nodes.
-func (e *Engine) SubmitCollectionGuarantee(msg *messages.SubmitCollectionGuarantee) {
-	if e.queue.Push(msg) {
+func (e *Engine) SubmitCollectionGuarantee(guarantee *flow.CollectionGuarantee) {
+	if e.queue.Push(guarantee) {
 		e.notifier.Notify()
 	} else {
 		e.engMetrics.OutboundMessageDropped(metrics.EngineCollectionProvider, metrics.MessageCollectionGuarantee)
diff --git a/engine/collection/pusher/engine_test.go b/engine/collection/pusher/engine_test.go
@@ -13,7 +13,6 @@ import (
 	"github.com/onflow/flow-go/engine/collection/pusher"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
-	"github.com/onflow/flow-go/model/messages"
 	"github.com/onflow/flow-go/module/irrecoverable"
 	"github.com/onflow/flow-go/module/metrics"
 	module "github.com/onflow/flow-go/module/mock"
@@ -97,11 +96,9 @@ func (suite *Suite) TestSubmitCollectionGuarantee() {
 	suite.conduit.On("Publish", guarantee, consensus[0].NodeID).
 		Run(func(_ mock.Arguments) { close(done) }).Return(nil).Once()
 
-	msg := &messages.SubmitCollectionGuarantee{
-		Guarantee: *guarantee,
-	}
-	err := suite.engine.ProcessLocal(msg)
-	suite.Require().Nil(err)
+	suite.engine.SubmitCollectionGuarantee(guarantee)
+	// TODO signature?
+	//suite.Require().Nil(err)
 
 	unittest.RequireCloseBefore(suite.T(), done, time.Second, "message not sent")
 
@@ -116,10 +113,7 @@ func (suite *Suite) TestSubmitCollectionGuaranteeNonLocal() {
 	// send from a non-allowed role
 	sender := suite.identities.Filter(filter.HasRole[flow.Identity](flow.RoleVerification))[0]
 
-	msg := &messages.SubmitCollectionGuarantee{
-		Guarantee: *guarantee,
-	}
-	err := suite.engine.Process(channels.PushGuarantees, sender.NodeID, msg)
+	err := suite.engine.Process(channels.PushGuarantees, sender.NodeID, guarantee)
 	suite.Require().Error(err)
 
 	suite.conduit.AssertNumberOfCalls(suite.T(), "Multicast", 0)
diff --git a/model/messages/collection.go b/model/messages/collection.go
@@ -5,12 +5,6 @@ import (
 	"github.com/onflow/flow-go/model/flow"
 )
 
-// SubmitCollectionGuarantee is a request to submit the given collection
-// guarantee to consensus nodes. Only valid as a node-local message.
-type SubmitCollectionGuarantee struct {
-	Guarantee flow.CollectionGuarantee
-}
-
 // CollectionRequest request all transactions from a collection with the given
 // fingerprint.
 type CollectionRequest struct {
diff --git a/module/finalizer/collection/finalizer.go b/module/finalizer/collection/finalizer.go
@@ -5,12 +5,11 @@ import (
 
 	"github.com/dgraph-io/badger/v2"
 
+	"github.com/onflow/flow-go/engine/collection"
 	"github.com/onflow/flow-go/model/cluster"
 	"github.com/onflow/flow-go/model/flow"
-	"github.com/onflow/flow-go/model/messages"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/mempool"
-	"github.com/onflow/flow-go/network"
 	"github.com/onflow/flow-go/storage/badger/operation"
 	"github.com/onflow/flow-go/storage/badger/procedure"
 )
@@ -22,15 +21,15 @@ import (
 type Finalizer struct {
 	db           *badger.DB
 	transactions mempool.Transactions
-	prov         network.Engine
+	prov         collection.GuaranteedCollectionPublisher
 	metrics      module.CollectionMetrics
 }
 
 // NewFinalizer creates a new finalizer for collection nodes.
 func NewFinalizer(
 	db *badger.DB,
 	transactions mempool.Transactions,
-	prov network.Engine,
+	prov collection.GuaranteedCollectionPublisher,
 	metrics module.CollectionMetrics,
 ) *Finalizer {
 	f := &Finalizer{
@@ -160,14 +159,12 @@ func (f *Finalizer) MakeFinal(blockID flow.Identifier) error {
 			// collection.
 
 			// TODO add real signatures here (2711)
-			f.prov.SubmitLocal(&messages.SubmitCollectionGuarantee{
-				Guarantee: flow.CollectionGuarantee{
-					CollectionID:     payload.Collection.ID(),
-					ReferenceBlockID: payload.ReferenceBlockID,
-					ChainID:          header.ChainID,
-					SignerIndices:    step.ParentVoterIndices,
-					Signature:        nil, // TODO: to remove because it's not easily verifiable by consensus nodes
-				},
+			f.prov.SubmitCollectionGuarantee(&flow.CollectionGuarantee{
+				CollectionID:     payload.Collection.ID(),
+				ReferenceBlockID: payload.ReferenceBlockID,
+				ChainID:          header.ChainID,
+				SignerIndices:    step.ParentVoterIndices,
+				Signature:        nil, // TODO: to remove because it's not easily verifiable by consensus nodes
 			})
 		}
 
diff --git a/module/finalizer/collection/finalizer_test.go b/module/finalizer/collection/finalizer_test.go
