Skip to content

Equivalent messages filter on consensus topic #5666

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 13 commits into from
Dec 21, 2023
Merged

Equivalent messages filter on consensus topic #5666

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
0e51612
added equivalent messages filter + debugger
sstanculeanu Oct 24, 2023
b18146b
added EquivalentMessagesDebuggerStub
sstanculeanu Oct 24, 2023
901309d
fixes after first review
sstanculeanu Oct 25, 2023
0f793fe
fix after review, process invalid equivalent message received
sstanculeanu Nov 13, 2023
0babda5
fix after review, better mutex handling
sstanculeanu Nov 13, 2023
d4e7634
fix after review, extract to new method
sstanculeanu Nov 13, 2023
cba26d3
added activation flag for equivalent messages
sstanculeanu Dec 7, 2023
1ba03ab
fixed epochs config test
sstanculeanu Dec 7, 2023
71ce841
fix after review: small refactor, in order to keep the mutex locked o…
sstanculeanu Dec 13, 2023
7bf2591
Merge branch 'feat/equivalent-messages' of https://github.com/multive…
sstanculeanu Dec 14, 2023
57072fa
fixes after merge
sstanculeanu Dec 14, 2023
918f099
further fixes after review + todos
sstanculeanu Dec 18, 2023
c9fe2aa
added one more todo
sstanculeanu Dec 18, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions consensus/spos/worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,7 @@ func (wrk *Worker) ProcessReceivedMessage(message p2p.MessageP2P, fromConnectedP

err = wrk.consensusMessageValidator.checkConsensusMessageValidity(cnsMsg, message.Peer())
if err != nil {
wrk.processInvalidEquivalentMessage(msgType, cnsMsg.BlockHeaderHash)
return err
}

Expand Down Expand Up @@ -749,6 +750,19 @@ func (wrk *Worker) processEquivalentMessage(msgType consensus.MessageType, block
return nil
}

func (wrk *Worker) processInvalidEquivalentMessage(msgType consensus.MessageType, blockHeaderHash []byte) {
if !wrk.consensusService.IsMessageWithFinalInfo(msgType) {
return
}

hdrHash := string(blockHeaderHash)

wrk.mutEquivalentMessages.Lock()
defer wrk.mutEquivalentMessages.Unlock()

delete(wrk.equivalentMessages, hdrHash)
Copy link
Contributor

@iulianpascalau iulianpascalau Nov 13, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I do not think this is the proper way to do this.
Reason: what happens in this scenario?

  1. receive the correct message
  2. receive the same block final info but for some reason checkConsensusMessageValidity returns an error. We delete the equivalent message for the hdrHash
  3. we receive the correct message again and we re-broadcast it

Copy link
Collaborator Author

@sstanculeanu sstanculeanu Nov 13, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pushed

}

// getEquivalentMessages returns a copy of the equivalent messages
func (wrk *Worker) getEquivalentMessages() map[string]uint64 {
wrk.mutEquivalentMessages.RLock()
Expand Down
34 changes: 34 additions & 0 deletions consensus/spos/worker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -634,6 +634,24 @@ func TestWorker_ProcessReceivedMessageEquivalentMessageShouldReturnError(t *test
)
buffEquiv, _ := wrk.Marshalizer().Marshal(cnsMsgEquiv)

invalidCnsMsg := consensus.NewConsensusMessage(
[]byte("other block header hash"),
nil,
nil,
nil,
pubKey,
bytes.Repeat([]byte("a"), SignatureSize),
int(bls.MtBlockHeaderFinalInfo),
0,
[]byte("invalid chain id"),
[]byte("01"),
signature,
signature,
currentPid,
nil,
)
buffInvalidCnsMsg, _ := wrk.Marshalizer().Marshal(invalidCnsMsg)

err := wrk.ProcessReceivedMessage(
&p2pmocks.P2PMessageMock{
DataField: buff,
Expand Down Expand Up @@ -665,6 +683,22 @@ func TestWorker_ProcessReceivedMessageEquivalentMessageShouldReturnError(t *test
assert.Equal(t, 1, len(equivalentMessages))
assert.Equal(t, uint64(2), equivalentMessages[string(equivalentBlockHeaderHash)])

err = wrk.ProcessReceivedMessage(
&p2pmocks.P2PMessageMock{
DataField: buffInvalidCnsMsg,
PeerField: currentPid,
SignatureField: []byte("signatureEquiv"),
},
equivMsgFrom,
&p2pmocks.MessengerStub{},
)
assert.Error(t, err)

// same state as before, invalid message should have been dropped
equivalentMessages = wrk.GetEquivalentMessages()
assert.Equal(t, 1, len(equivalentMessages))
assert.Equal(t, uint64(2), equivalentMessages[string(equivalentBlockHeaderHash)])

wrk.RemoveAllEquivalentMessages()
equivalentMessages = wrk.GetEquivalentMessages()
assert.Equal(t, 0, len(equivalentMessages))
Expand Down