Remove dead code

Author: algorandskiy

components/mocks/mockNetwork.go

@@ -91,8 +91,8 @@ func (network *MockNetwork) RegisterHandlers(dispatch []network.TaggedMessageHan
 func (network *MockNetwork) ClearHandlers() {
 }
 
-// RegisterProcessors - empty implementation.
-func (network *MockNetwork) RegisterProcessors(dispatch []network.TaggedMessageProcessor) {
+// RegisterValidatorHandlers - empty implementation.
+func (network *MockNetwork) RegisterValidatorHandlers(dispatch []network.TaggedMessageValidatorHandler) {
 }
 
 // ClearProcessors - empty implementation

data/txHandler.go

@@ -262,16 +262,14 @@ func (handler *TxHandler) Start() {
 
 	// libp2p pubsub validator and handler abstracted as TaggedMessageProcessor
 	// TODO: rename to validators
-	handler.net.RegisterProcessors([]network.TaggedMessageProcessor{
+	handler.net.RegisterValidatorHandlers([]network.TaggedMessageValidatorHandler{
 		{
 			Tag: protocol.TxnTag,
 			// create anonymous struct to hold the two functions and satisfy the network.MessageProcessor interface
 			MessageHandler: struct {
-				network.ProcessorValidateFunc
-				network.ProcessorHandleFunc
+				network.ValidateHandleFunc
 			}{
-				network.ProcessorValidateFunc(handler.validateIncomingTxMessage),
-				network.ProcessorHandleFunc(handler.processIncomingTxMessage),
+				network.ValidateHandleFunc(handler.validateIncomingTxMessage),
 			},
 		},
 	})
@@ -788,31 +786,24 @@ func (handler *TxHandler) processIncomingTxn(rawmsg network.IncomingMessage) net
 	return network.OutgoingMessage{Action: network.Ignore}
 }
 
-type validatedIncomingTxMessage struct {
-	rawmsg            network.IncomingMessage
-	unverifiedTxGroup []transactions.SignedTxn
-	msgKey            *crypto.Digest
-	canonicalKey      *crypto.Digest
-}
-
 // validateIncomingTxMessage is the validator for the MessageProcessor implementation used by P2PNetwork.
-func (handler *TxHandler) validateIncomingTxMessage(rawmsg network.IncomingMessage) network.ValidatedMessage {
+func (handler *TxHandler) validateIncomingTxMessage(rawmsg network.IncomingMessage) network.OutgoingMessage {
 	msgKey, isDup := handler.incomingMsgDupCheck(rawmsg.Data)
 	if isDup {
-		return network.ValidatedMessage{Action: network.Ignore, ValidatedMessage: nil}
+		return network.OutgoingMessage{Action: network.Ignore}
 	}
 
 	unverifiedTxGroup, consumed, invalid := decodeMsg(rawmsg.Data)
 	if invalid {
 		// invalid encoding or exceeding txgroup, disconnect from this peer
-		return network.ValidatedMessage{Action: network.Disconnect, ValidatedMessage: nil}
+		return network.OutgoingMessage{Action: network.Disconnect}
 	}
 
 	canonicalKey, drop := handler.incomingTxGroupDupRateLimit(unverifiedTxGroup, consumed, rawmsg.Sender)
 	if drop {
 		// this re-serialized txgroup was detected as a duplicate by the canonical message cache,
 		// or it was rate-limited by the per-app rate limiter
-		return network.ValidatedMessage{Action: network.Ignore, ValidatedMessage: nil}
+		return network.OutgoingMessage{Action: network.Ignore}
 	}
 
 	// apply backlog worker logic
@@ -827,9 +818,8 @@ func (handler *TxHandler) validateIncomingTxMessage(rawmsg network.IncomingMessa
 
 	if handler.checkAlreadyCommitted(wi) {
 		transactionMessagesAlreadyCommitted.Inc(nil)
-		return network.ValidatedMessage{
-			Action:           network.Ignore,
-			ValidatedMessage: nil,
+		return network.OutgoingMessage{
+			Action: network.Ignore,
 		}
 	}
 
@@ -842,9 +832,8 @@ func (handler *TxHandler) validateIncomingTxMessage(rawmsg network.IncomingMessa
 		if wi.Err != nil {
 			handler.postProcessReportErrors(wi.Err)
 			logging.Base().Warnf("Received a malformed tx group %v: %v", m.unverifiedTxGroup, wi.Err)
-			return network.ValidatedMessage{
-				Action:           network.Disconnect,
-				ValidatedMessage: nil,
+			return network.OutgoingMessage{
+				Action: network.Disconnect,
 			}
 		}
 		// at this point, we've verified the transaction, so we can safely treat the transaction as a verified transaction.
@@ -855,9 +844,8 @@ func (handler *TxHandler) validateIncomingTxMessage(rawmsg network.IncomingMessa
 		if err != nil {
 			handler.rememberReportErrors(err)
 			logging.Base().Debugf("could not remember tx: %v", err)
-			return network.ValidatedMessage{
-				Action:           network.Ignore,
-				ValidatedMessage: nil,
+			return network.OutgoingMessage{
+				Action: network.Ignore,
 			}
 		}
 
@@ -868,32 +856,23 @@ func (handler *TxHandler) validateIncomingTxMessage(rawmsg network.IncomingMessa
 		if err != nil {
 			logging.Base().Infof("unable to pin transaction: %v", err)
 		}
-		return network.ValidatedMessage{
-			Action:           network.Accept,
-			ValidatedMessage: nil,
+		return network.OutgoingMessage{
+			Action: network.Accept,
 		}
 
 	case <-handler.streamVerifierDropped2:
 		transactionMessagesDroppedFromBacklog.Inc(nil)
-		return network.ValidatedMessage{
-			Action:           network.Ignore,
-			ValidatedMessage: nil,
+		return network.OutgoingMessage{
+			Action: network.Ignore,
 		}
 	case <-handler.ctx.Done():
 		transactionMessagesDroppedFromBacklog.Inc(nil)
-		return network.ValidatedMessage{
-			Action:           network.Ignore,
-			ValidatedMessage: nil,
+		return network.OutgoingMessage{
+			Action: network.Ignore,
 		}
 	}
 }
 
-// processIncomingTxMessage is the handler for the MessageProcessor implementation used by P2PNetwork.
-func (handler *TxHandler) processIncomingTxMessage(validatedMessage network.ValidatedMessage) network.OutgoingMessage {
-	// process is noop, all work is done in validateIncomingTxMessage above
-	return network.OutgoingMessage{Action: network.Ignore}
-}
-
 var errBackLogFullLocal = errors.New("backlog full")
 
 // LocalTransaction is a special shortcut handler for local transactions and intended to be used

network/gossipNode.go

@@ -80,8 +80,10 @@ type GossipNode interface {
 	// ClearHandlers deregisters all the existing message handlers.
 	ClearHandlers()
 
-	// RegisterProcessors adds to the set of given message processors.
-	RegisterProcessors(dispatch []TaggedMessageProcessor)
+	// RegisterValidatorHandlers adds to the set of given message validation handlers.
+	// A difference with regular handlers is validation ones perform synchronous validation.
+	// Currently used as p2p pubsub topic validators.
+	RegisterValidatorHandlers(dispatch []TaggedMessageValidatorHandler)
 
 	// ClearProcessors deregisters all the existing message processors.
 	ClearProcessors()
@@ -156,14 +158,6 @@ type OutgoingMessage struct {
 	OnRelease func()
 }
 
-// ValidatedMessage is a message that has been validated and is ready to be processed.
-// Think as an intermediate one between IncomingMessage and OutgoingMessage
-type ValidatedMessage struct {
-	Action           ForwardingPolicy
-	Tag              Tag
-	ValidatedMessage interface{}
-}
-
 // ForwardingPolicy is an enum indicating to whom we should send a message
 //
 //msgp:ignore ForwardingPolicy
@@ -202,28 +196,19 @@ func (f HandlerFunc) Handle(message IncomingMessage) OutgoingMessage {
 	return f(message)
 }
 
-// MessageProcessor takes a IncomingMessage (e.g., vote, transaction), processes it, and returns what (if anything)
+// MessageValidatorHandler takes a IncomingMessage (e.g., vote, transaction), processes it, and returns what (if anything)
 // to send to the network in response.
-// This is an extension of the MessageHandler that works in two stages: validate ->[result]-> handle.
-type MessageProcessor interface {
-	Validate(message IncomingMessage) ValidatedMessage
-	Handle(message ValidatedMessage) OutgoingMessage
+// it supposed to perform synchronous validation and return the result of the validation
+// so that network knows immediately if the message should be be broadcasted or not.
+type MessageValidatorHandler interface {
+	ValidateHandle(message IncomingMessage) OutgoingMessage
 }
 
-// ProcessorValidateFunc represents an implementation of the MessageProcessor interface
-type ProcessorValidateFunc func(message IncomingMessage) ValidatedMessage
-
-// ProcessorHandleFunc represents an implementation of the MessageProcessor interface
-type ProcessorHandleFunc func(message ValidatedMessage) OutgoingMessage
-
-// Validate implements MessageProcessor.Validate, calling the validator with the IncomingMessage and returning the action
-// and validation extra data that can be use as the handler input.
-func (f ProcessorValidateFunc) Validate(message IncomingMessage) ValidatedMessage {
-	return f(message)
-}
+// ValidateHandleFunc represents an implementation of the MessageProcessor interface
+type ValidateHandleFunc func(message IncomingMessage) OutgoingMessage
 
-// Handle implements MessageProcessor.Handle calling the handler with the ValidatedMessage and returning the OutgoingMessage
-func (f ProcessorHandleFunc) Handle(message ValidatedMessage) OutgoingMessage {
+// ValidateHandle implements MessageValidatorHandler.ValidateHandle, calling the validator with the IncomingMessage and returning the action.
+func (f ValidateHandleFunc) ValidateHandle(message IncomingMessage) OutgoingMessage {
 	return f(message)
 }
 
@@ -235,9 +220,9 @@ type taggedMessageDispatcher[T any] struct {
 // TaggedMessageHandler receives one type of broadcast messages
 type TaggedMessageHandler = taggedMessageDispatcher[MessageHandler]
 
-// TaggedMessageProcessor receives one type of broadcast messages
+// TaggedMessageValidatorHandler receives one type of broadcast messages
 // and performs two stage processing: validating and handling
-type TaggedMessageProcessor = taggedMessageDispatcher[MessageProcessor]
+type TaggedMessageValidatorHandler = taggedMessageDispatcher[MessageValidatorHandler]
 
 // Propagate is a convenience function to save typing in the common case of a message handler telling us to propagate an incoming message
 // "return network.Propagate(msg)" instead of "return network.OutgoingMsg{network.Broadcast, msg.Tag, msg.Data}"

network/hybridNetwork.go

@@ -180,10 +180,10 @@ func (n *HybridP2PNetwork) ClearHandlers() {
 	n.wsNetwork.ClearHandlers()
 }
 
-// RegisterProcessors adds to the set of given message processors.
-func (n *HybridP2PNetwork) RegisterProcessors(dispatch []TaggedMessageProcessor) {
-	n.p2pNetwork.RegisterProcessors(dispatch)
-	n.wsNetwork.RegisterProcessors(dispatch)
+// RegisterValidatorHandlers adds to the set of given message processors.
+func (n *HybridP2PNetwork) RegisterValidatorHandlers(dispatch []TaggedMessageValidatorHandler) {
+	n.p2pNetwork.RegisterValidatorHandlers(dispatch)
+	n.wsNetwork.RegisterValidatorHandlers(dispatch)
 }
 
 // ClearProcessors deregisters all the existing message processors.

network/multiplexer.go

@@ -61,8 +61,8 @@ func (m *Multiplexer) getHandler(tag Tag) (MessageHandler, bool) {
 }
 
 // Retrieves the processor for the given message Tag from the processors array.
-func (m *Multiplexer) getProcessor(tag Tag) (MessageProcessor, bool) {
-	return getHandler[MessageProcessor](&m.msgProcessors, tag)
+func (m *Multiplexer) getProcessor(tag Tag) (MessageValidatorHandler, bool) {
+	return getHandler[MessageValidatorHandler](&m.msgProcessors, tag)
 }
 
 // Handle is the "input" side of the multiplexer. It dispatches the message to the previously defined handler.
@@ -74,17 +74,9 @@ func (m *Multiplexer) Handle(msg IncomingMessage) OutgoingMessage {
 }
 
 // Validate is an alternative "input" side of the multiplexer. It dispatches the message to the previously defined validator.
-func (m *Multiplexer) Validate(msg IncomingMessage) ValidatedMessage {
+func (m *Multiplexer) Validate(msg IncomingMessage) OutgoingMessage {
 	if handler, ok := m.getProcessor(msg.Tag); ok {
-		return handler.Validate(msg)
-	}
-	return ValidatedMessage{}
-}
-
-// Process is the second step of message handling after validation. It dispatches the message to the previously defined processor.
-func (m *Multiplexer) Process(msg ValidatedMessage) OutgoingMessage {
-	if handler, ok := m.getProcessor(msg.Tag); ok {
-		return handler.Handle(msg)
+		return handler.ValidateHandle(msg)
 	}
 	return OutgoingMessage{}
 }
@@ -110,8 +102,8 @@ func (m *Multiplexer) RegisterHandlers(dispatch []TaggedMessageHandler) {
 	registerMultiplexer(&m.msgHandlers, dispatch)
 }
 
-// RegisterProcessors registers the set of given message handlers.
-func (m *Multiplexer) RegisterProcessors(dispatch []TaggedMessageProcessor) {
+// RegisterValidatorHandlers registers the set of given message handlers.
+func (m *Multiplexer) RegisterValidatorHandlers(dispatch []TaggedMessageValidatorHandler) {
 	registerMultiplexer(&m.msgProcessors, dispatch)
 }
 
@@ -145,5 +137,5 @@ func (m *Multiplexer) ClearHandlers(excludeTags []Tag) {
 
 // ClearProcessors deregisters all the existing message handlers other than the one provided in the excludeTags list
 func (m *Multiplexer) ClearProcessors(excludeTags []Tag) {
-	clearMultiplexer[MessageProcessor](&m.msgProcessors, excludeTags)
+	clearMultiplexer[MessageValidatorHandler](&m.msgProcessors, excludeTags)
 }

network/p2pNetwork.go

@@ -685,9 +685,9 @@ func (n *P2PNetwork) ClearHandlers() {
 	n.handler.ClearHandlers([]Tag{})
 }
 
-// RegisterProcessors adds to the set of given message handlers.
-func (n *P2PNetwork) RegisterProcessors(dispatch []TaggedMessageProcessor) {
-	n.handler.RegisterProcessors(dispatch)
+// RegisterValidatorHandlers adds to the set of given message handlers.
+func (n *P2PNetwork) RegisterValidatorHandlers(dispatch []TaggedMessageValidatorHandler) {
+	n.handler.RegisterValidatorHandlers(dispatch)
 }
 
 // ClearProcessors deregisters all the existing message handlers.
@@ -880,7 +880,8 @@ func (n *P2PNetwork) txTopicHandleLoop() {
 	n.log.Debugf("Subscribed to topic %s", p2p.TXTopicName)
 
 	for {
-		msg, err := sub.Next(n.ctx)
+		// msg from sub.Next not used since all work done by txTopicValidator
+		_, err := sub.Next(n.ctx)
 		if err != nil {
 			if err != pubsub.ErrSubscriptionCancelled && err != context.Canceled {
 				n.log.Errorf("Error reading from subscription %v, peerId %s", err, n.service.ID())
@@ -889,13 +890,6 @@ func (n *P2PNetwork) txTopicHandleLoop() {
 			sub.Cancel()
 			return
 		}
-		// if there is a self-sent the message no need to process it.
-		if msg.ReceivedFrom == n.service.ID() {
-			continue
-		}
-
-		_ = n.handler.Process(msg.ValidatorData.(ValidatedMessage))
-
 		// participation or configuration change, cancel subscription and quit
 		if !n.wantTXGossip.Load() {
 			n.log.Debugf("Cancelling subscription to topic %s due participation change", p2p.TXTopicName)