Message validation optimization #6462

cristure · 2024-09-16T12:23:58Z

Reasoning behind the pull request

Sometimes a message could be processed multiple times with no real benefit. Thus we need to introduce a mechanism that will check if the message has been previously processed in a set time span and exit early on if that is the case.

Proposed changes

Add a map of timeCachers. Before initializing every interceptor we add the timeCacher to the map. When they will try to process the message they will retrieve their cacher from the global map and check if the message has been processed before.

Testing procedure

TBD

Pre-requisites

Based on the Contributing Guidelines the PR author and the reviewers must check the following requirements are met:

was the PR targeted to the correct branch?
if this is a larger feature that probably needs more than one PR, is there a feat branch created?
if this is a feat branch merging, do all satellite projects have a proper tag inside go.mod?

…ization

…essage_validation_optimization # Conflicts: # process/factory/interceptorscontainer/baseInterceptorsContainerFactory.go # process/factory/interceptorscontainer/metaInterceptorsContainerFactory.go # process/factory/interceptorscontainer/shardInterceptorsContainerFactory.go

AdoAdoAdo · 2024-09-18T08:42:19Z

epochStart/bootstrap/syncEpochStartMeta.go

+			WhiteListRequest:          args.WhitelistHandler,
+			CurrentPeerId:             args.Messenger.ID(),
+			PreferredPeersHolder:      disabled.NewPreferredPeersHolder(),
+			ProcessedMessagesCacheMap: make(map[string]storage.Cacher),


I think this should not be allocated here, but received through the arguments.

AdoAdoAdo · 2024-09-18T08:44:18Z

factory/processing/processComponents.go

@@ -1501,6 +1501,8 @@ func (pcf *processComponentsFactory) newInterceptorContainerFactory(
 		nodeOperationMode = common.FullArchiveMode
 	}

+	processedMessagesCacheMap := make(map[string]storage.Cacher)


I think this should not be initialized here, but received in the arguments so we can also mock it in tests.

AdoAdoAdo · 2024-09-18T08:52:40Z

process/interceptors/multiDataInterceptor.go

 		listInterceptedData[index] = interceptedData
 		if err != nil {
 			mdi.throttler.EndProcessing()
 			return err
 		}

+		errCache := mdi.checkIfMessageHasBeenProcessed(interceptedData)


we need to check also what happens in the following scenario:

receive an invalid message

receive again the same invalid message from another source (before the cache expires)

What we want: - do not accept/propagate the message if it was initially invalidated (need to check if the validation returns true or false in this case, and we need to make sure it returns false)

We could solve this if we save also the verification status inside the cache.
In this case we might still need the mutex per message key, so that we do not process it twice while not having the status inside the cache.

AdoAdoAdo · 2024-09-18T09:03:31Z

epochStart/bootstrap/factory/epochStartInterceptorsContainerFactory.go

@@ -108,6 +110,7 @@ func NewEpochStartInterceptorsContainer(args ArgsEpochStartInterceptorContainer)
 		FullArchivePeerShardMapper:   fullArchivePeerShardMapper,
 		HardforkTrigger:              hardforkTrigger,
 		NodeOperationMode:            args.NodeOperationMode,
+		ProcessedMessagesCacheMap:    make(map[string]storage.Cacher),


Need to check all intializations of this map, as we might end up with multiple versions of this.

AdoAdoAdo · 2024-09-18T10:09:17Z

process/interceptors/multiDataInterceptor.go

@@ -152,12 +157,19 @@ func (mdi *MultiDataInterceptor) ProcessReceivedMessage(message p2p.MessageP2P,
 	for index, dataBuff := range multiDataBuff {
 		var interceptedData process.InterceptedData
 		interceptedData, err = mdi.interceptedData(dataBuff, message.Peer(), fromConnectedPeer)


seems like at line 159 inside mdi.interceptedData happens all the intensive processing -> signature verifications, etc.
so checking the cache at line 167 does not save much in this case.
I would suggest moving the cache check inside mdi.InterceptedData, before interceptedData.CheckValidity
Check the same for singleDataInterceptors.

…ization

AdoAdoAdo · 2024-09-24T11:38:03Z

process/interceptors/interceptedDataVerifier.go

+		return ErrInvalidInterceptedData
+	}
+
+	err := interceptedData.CheckValidity()


in case the checkValidity lasts a long time, there can be multiple verifications being done in parallel
these can be avoided if we use a per-resource (hash) mutex.while checking the validity & writing in the cache.

hope I fixed it in the next push

AdoAdoAdo · 2024-09-24T11:41:08Z

process/interceptors/multiDataInterceptor.go

@@ -152,7 +159,10 @@ func (mdi *MultiDataInterceptor) ProcessReceivedMessage(message p2p.MessageP2P,
 	for index, dataBuff := range multiDataBuff {
 		var interceptedData process.InterceptedData
 		interceptedData, err = mdi.interceptedData(dataBuff, message.Peer(), fromConnectedPeer)
-		listInterceptedData[index] = interceptedData
+
+		if !errors.Is(err, ErrInvalidInterceptedData) {


no need for this error handling here, as if the error is not nil then at line 167 it will anyway exit and not do anything with the listInterceptedData.

alright, removed the check.

AdoAdoAdo · 2024-09-24T11:53:38Z

process/interceptors/interceptedDataVerifier.go

+// It will retrieve the status in the cache if it exists, otherwise it will validate it and store the status of the
+// validation in the cache. Note that the entries are stored for a set period of time
+func (idv *interceptedDataVerifier) Verify(interceptedData process.InterceptedData) error {
+	if len(interceptedData.Hash()) == 0 {


this early exit here for len(hash == 0) is bypassing the CheckValidity check which was previously called even if there was no hash computed.

can you return instead interceptedData.CheckValidity() ?

AdoAdoAdo · 2024-09-24T12:26:40Z

process/block/interceptedBlocks/interceptedMetaBlockHeader_test.go

@@ -98,7 +99,7 @@ func TestNewInterceptedMetaHeader_ShouldWork(t *testing.T) {
 	assert.Nil(t, err)
 }

-//------- CheckValidity


maybe wrong replace?

AdoAdoAdo · 2024-09-24T12:27:04Z

process/block/interceptedBlocks/interceptedMetaBlockHeader.go

@@ -99,7 +100,7 @@ func (imh *InterceptedMetaHeader) CheckValidity() error {
 		}

 		if imh.isMetaHeaderEpochOutOfRange() {
-			log.Trace("InterceptedMetaHeader.CheckValidity",
+			log.Trace("InterceptedMetaHeader.Verify",


wrong replace?

AdoAdoAdo · 2024-09-24T12:27:21Z

process/block/interceptedBlocks/interceptedBlockHeader_test.go

@@ -173,7 +174,7 @@ func TestNewInterceptedHeader_MetachainForThisShardShouldWork(t *testing.T) {
 	assert.True(t, inHdr.IsForCurrentShard())
 }

-//------- CheckValidity


wrong replace?

AdoAdoAdo · 2024-09-24T12:33:14Z

process/factory/interceptorscontainer/baseInterceptorsContainerFactory.go

@@ -29,6 +31,8 @@ const (
 	minTimespanDurationInSec        = int64(1)
 	errorOnMainNetworkString        = "on main network"
 	errorOnFullArchiveNetworkString = "on full archive network"
+	cacheDefaultSpan                = 30 * time.Second


todo: maybe move the two constants to config

AdoAdoAdo · 2024-09-24T12:34:54Z

process/interface.go

@@ -1401,3 +1401,8 @@ type SentSignaturesTracker interface {
 	ResetCountersForManagedBlockSigner(signerPk []byte)
 	IsInterfaceNil() bool
 }
+
+type InterceptedDataVerifier interface {


add comment on exported type

AdoAdoAdo · 2024-09-24T12:35:15Z

process/transaction/interceptedTransaction_test.go

@@ -612,7 +612,7 @@ func TestNewInterceptedTransaction_ShouldWork(t *testing.T) {
 	assert.Equal(t, tx, txi.Transaction())
 }

-// ------- CheckValidity


wrong replace?

AdoAdoAdo · 2024-09-24T12:35:24Z

process/unsigned/interceptedUnsignedTransaction_test.go

@@ -170,7 +171,7 @@ func TestNewInterceptedUnsignedTransaction_ShouldWork(t *testing.T) {
 	assert.Nil(t, err)
 }

-// ------- CheckValidity
+// ------- Verify


wrong replace?

…essage_validation_optimization

… into message_validation_optimization

cristure and others added 14 commits September 12, 2024 12:47

initial commit.

10afdba

added test for processing message.

aaa9d4b

added message validation cache.

603bd69

fix some tests.

57f009e

fixed missing topic.

f5a6f2e

fix some more tests.

8eec330

fix nil pointer dereferences in tests.

a733097

more tests fixed.

bab3b05

add map for cacher in more unit tests.

c3f7120

fix nil map for more tests.

b62e37b

fix process tests.

c00f8ee

Merge branch 'feat/equivalent-messages' into message_validation_optim…

94a64c1

…ization

fix conflicts with target branch.

d59c5bd

AdoAdoAdo requested changes Sep 18, 2024

View reviewed changes

cristure and others added 12 commits September 19, 2024 18:22

refactored message validation to a different component.

396cd8e

cosmetic changes.

a44129f

added intercepted data verifier stub in multi data tests.

afdceb1

fix single data interceptor tests.

d8d8113

commit debug strings for CI.

fe2082c

moved map of cachers in node processor.

c6b7a0f

cosmetic changes.

f59fc85

fix integration tests.

d00c7ab

fix some more tests.

f21e24f

fix some unit tests.

f40d222

fix nil map in tests.

3a327c4

Merge branch 'feat/equivalent-messages' into message_validation_optim…

a756a1f

…ization

AdoAdoAdo reviewed Sep 24, 2024

View reviewed changes

cristure added 2 commits September 25, 2024 15:21

Merge remote-tracking branch 'origin/feat/equivalent-messages' into m…

ff8eaa9

…essage_validation_optimization

cosmetic changes.

cde6c80

cristure added 6 commits September 25, 2024 16:20

Merge remote-tracking branch 'origin/message_validation_optimization'…

e0cf24b

… into message_validation_optimization

refactored map of caches for intercepted data into a new component.

9504263

added mock for interceptedDataVerifierFactory and injected into tests.

b622085

more test fixes.

f0b34ef

increase wait time for CI run.

b05c650

bring back rw mutex on interceptedDataVerifier.

5dc6bba

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Message validation optimization #6462

Message validation optimization #6462

cristure commented Sep 16, 2024

AdoAdoAdo Sep 18, 2024

AdoAdoAdo Sep 18, 2024

AdoAdoAdo Sep 18, 2024

AdoAdoAdo Sep 18, 2024

AdoAdoAdo Sep 18, 2024

AdoAdoAdo Sep 18, 2024

AdoAdoAdo Sep 24, 2024

cristure Sep 25, 2024

AdoAdoAdo Sep 24, 2024

cristure Sep 25, 2024

AdoAdoAdo Sep 24, 2024

cristure Sep 25, 2024

AdoAdoAdo Sep 24, 2024

cristure Sep 25, 2024

AdoAdoAdo Sep 24, 2024

cristure Sep 25, 2024

AdoAdoAdo Sep 24, 2024

cristure Sep 25, 2024

AdoAdoAdo Sep 24, 2024

AdoAdoAdo Sep 24, 2024

AdoAdoAdo Sep 24, 2024

AdoAdoAdo Sep 24, 2024

Message validation optimization #6462

Are you sure you want to change the base?

Message validation optimization #6462

Conversation

cristure commented Sep 16, 2024

Reasoning behind the pull request

Proposed changes

Testing procedure

Pre-requisites

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment