@@ -13,7 +13,6 @@ import (
1313 "github.com/onflow/flow-go/engine/common/fifoqueue"
1414 "github.com/onflow/flow-go/model/flow"
1515 "github.com/onflow/flow-go/model/flow/filter"
16- "github.com/onflow/flow-go/model/messages"
1716 "github.com/onflow/flow-go/module"
1817 "github.com/onflow/flow-go/module/component"
1918 "github.com/onflow/flow-go/module/irrecoverable"
@@ -44,8 +43,7 @@ type Engine struct {
4443 cm * component.ComponentManager
4544}
4645
47- // TODO convert to network.MessageProcessor
48- var _ network.Engine = (* Engine )(nil )
46+ var _ network.MessageProcessor = (* Engine )(nil )
4947var _ component.Component = (* Engine )(nil )
5048
5149// New creates a new pusher engine.
@@ -120,17 +118,17 @@ func (e *Engine) outboundQueueWorker(ctx irrecoverable.SignalerContext, ready co
120118// No errors expected during normal operations.
121119func (e * Engine ) processOutboundMessages (ctx context.Context ) error {
122120 for {
123- nextMessage , ok := e .queue .Pop ()
121+ item , ok := e .queue .Pop ()
124122 if ! ok {
125123 return nil
126124 }
127125
128- asSCGMsg , ok := nextMessage .(* messages. SubmitCollectionGuarantee )
126+ guarantee , ok := item .(* flow. CollectionGuarantee )
129127 if ! ok {
130- return fmt .Errorf ("invalid message type in pusher engine queue" )
128+ return fmt .Errorf ("invalid type in pusher engine queue" )
131129 }
132130
133- err := e .publishCollectionGuarantee (& asSCGMsg . Guarantee )
131+ err := e .publishCollectionGuarantee (guarantee )
134132 if err != nil {
135133 return err
136134 }
@@ -143,44 +141,32 @@ func (e *Engine) processOutboundMessages(ctx context.Context) error {
143141 }
144142}
145143
146- // SubmitLocal submits an event originating on the local node.
147- func (e * Engine ) SubmitLocal (event interface {}) {
148- ev , ok := event .(* messages.SubmitCollectionGuarantee )
149- if ok {
150- e .SubmitCollectionGuarantee (ev )
151- } else {
152- engine .LogError (e .log , fmt .Errorf ("invalid message argument to pusher engine" ))
153- }
154- }
155-
156- // Submit submits the given event from the node with the given origin ID
157- // for processing in a non-blocking manner. It returns instantly and logs
158- // a potential processing error internally when done.
159- func (e * Engine ) Submit (channel channels.Channel , originID flow.Identifier , event interface {}) {
160- engine .LogError (e .log , fmt .Errorf ("pusher engine should only receive local messages on the same node" ))
161- }
162-
163- // ProcessLocal processes an event originating on the local node.
164- func (e * Engine ) ProcessLocal (event interface {}) error {
165- ev , ok := event .(* messages.SubmitCollectionGuarantee )
166- if ok {
167- e .SubmitCollectionGuarantee (ev )
168- return nil
169- } else {
170- return fmt .Errorf ("invalid message argument to pusher engine" )
171- }
172- }
173-
174- // Process processes the given event from the node with the given origin ID in
175- // a blocking manner. It returns the potential processing error when done.
144+ // Process is called by the networking layer, when peers broadcast messages with this node
145+ // as one of the recipients. The protocol specifies that Collector nodes broadcast Collection
146+ // Guarantees to Consensus Nodes and _only_ those. When the pusher engine (running only on
147+ // Collectors) receives a message, this message is evidence of byzantine behavior.
148+ // Byzantine inputs are internally handled by the pusher.Engine and do *not* result in
149+ // error returns. No errors expected during normal operation (including byzantine inputs).
176150func (e * Engine ) Process (channel channels.Channel , originID flow.Identifier , message any ) error {
177- return fmt .Errorf ("pusher engine should only receive local messages on the same node" )
151+ // Targeting a collector node's pusher.Engine with messages could be considered as a slashable offense.
152+ // Though, for generating cryptographic evidence, we need Message Forensics - see reference [1].
153+ // Much further into the future, when we are implementing slashing challenges, we'll probably implement a
154+ // dedicated consumer to post-process evidence of protocol violations into slashing challenges. For now,
155+ // we just log this with the `KeySuspicious` to alert the node operator.
156+ // [1] Message Forensics FLIP https://github.com/onflow/flips/pull/195)
157+ errs := fmt .Errorf ("collector node's pusher.Engine was targeted by message %T on channel %v" , message , channel )
158+ e .log .Warn ().
159+ Err (errs ).
160+ Bool (logging .KeySuspicious , true ).
161+ Str ("peer_id" , originID .String ()).
162+ Msg ("potentially byzantine networking traffic detected" )
163+ return nil
178164}
179165
180166// SubmitCollectionGuarantee adds a collection guarantee to the engine's queue
181167// to later be published to consensus nodes.
182- func (e * Engine ) SubmitCollectionGuarantee (msg * messages. SubmitCollectionGuarantee ) {
183- if e .queue .Push (msg ) {
168+ func (e * Engine ) SubmitCollectionGuarantee (guarantee * flow. CollectionGuarantee ) {
169+ if e .queue .Push (guarantee ) {
184170 e .notifier .Notify ()
185171 } else {
186172 e .engMetrics .OutboundMessageDropped (metrics .EngineCollectionProvider , metrics .MessageCollectionGuarantee )
0 commit comments