Refactor pusher engine: error on non-local messages

Author: tim-barry

engine/collection/pusher/engine.go

@@ -139,10 +139,7 @@ func (e *Engine) SubmitLocal(event interface{}) {
 // 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{}) {
-	err := e.messageHandler.Process(originID, event)
-	if err != nil {
-		engine.LogError(e.log, err)
-	}
+	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.
@@ -153,16 +150,7 @@ func (e *Engine) ProcessLocal(event interface{}) error {
 // 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.
 func (e *Engine) Process(channel channels.Channel, originID flow.Identifier, message any) error {
-	err := e.messageHandler.Process(originID, message)
-	if err != nil {
-		if errors.Is(err, engine.IncompatibleInputTypeError) {
-			e.log.Warn().Bool(logging.KeySuspicious, true).Msgf("%v delivered unsupported message %T through %v", originID, message, channel)
-			return nil
-		}
-		// TODO add comment about Process errors...
-		return fmt.Errorf("unexpected failure to process inbound pusher message")
-	}
-	return nil
+	return fmt.Errorf("pusher engine should only receive local messages on the same node")
 }
 
 // process processes events for the pusher engine on the collection node.