block certifier: fix certifier concurrent map access
## Motivation
race condition, added test that failed with -race -count 5 and pass after the fix
MODIFY
blocks/certifier.go
blocks/certifier.go
@@ -298,6 +298,8 @@ func (c *Certifier) HandleSyncedCertificate(ctx context.Context, lid types.Layer
}
func (c *Certifier) certified(lid types.LayerID, bid types.BlockID) bool {
+ c.mu.Lock()
+ defer c.mu.Unlock()
if _, ok := c.certifyMsgs[lid]; ok {
if _, ok := c.certifyMsgs[lid][bid]; ok {
return c.certifyMsgs[lid][bid].done

MODIFY
blocks/certifier_test.go
blocks/certifier_test.go
@@ -3,6 +3,7 @@ package blocks
import (
"context"
"errors"
+ "sync"
"testing"
"github.com/golang/mock/gomock"
@@ -231,24 +232,56 @@ func Test_HandleCertifyMessage(t *testing.T) {
}
func Test_HandleCertifyMessage_Certified(t *testing.T) {
- tc := newTestCertifier(t)
- numMsgs := tc.cfg.CommitteeSize
- cutoff := tc.cfg.CertifyThreshold / int(defaultCnt)
- b := generateBlock(t, tc.db)
- tc.mClk.EXPECT().GetCurrentLayer().Return(b.LayerIndex).AnyTimes()
- tc.mb.EXPECT().GetBeacon(b.LayerIndex.GetEpoch()).Return(types.RandomBeacon(), nil).AnyTimes()
- require.NoError(t, tc.RegisterForCert(context.TODO(), b.LayerIndex, b.ID()))
- tc.mtortoise.EXPECT().OnHareOutput(b.LayerIndex, b.ID())
+ tt := []struct {
+ name string
+ concurrent bool
+ }{
+ {
+ name: "sequential",
+ },
+ {
+ name: "concurrent",
+ concurrent: true,
+ },
+ }
+ for _, tc := range tt {
+ tc := tc
+ t.Run(tc.name, func(t *testing.T) {
+ tcc := newTestCertifier(t)
+ numMsgs := tcc.cfg.CommitteeSize
+ cutoff := tcc.cfg.CertifyThreshold / int(defaultCnt)
+ b := generateBlock(t, tcc.db)
+ tcc.mClk.EXPECT().GetCurrentLayer().Return(b.LayerIndex).AnyTimes()
+ tcc.mb.EXPECT().GetBeacon(b.LayerIndex.GetEpoch()).Return(types.RandomBeacon(), nil).AnyTimes()
+ require.NoError(t, tcc.RegisterForCert(context.TODO(), b.LayerIndex, b.ID()))
+ if tc.concurrent {
+ tcc.mOracle.EXPECT().Validate(gomock.Any(), b.LayerIndex, eligibility.CertifyRound, tcc.cfg.CommitteeSize, gomock.Any(), gomock.Any(), defaultCnt).
+ Return(true, nil).MinTimes(cutoff).MaxTimes(numMsgs)
+ }
+ tcc.mtortoise.EXPECT().OnHareOutput(b.LayerIndex, b.ID())
+ var wg sync.WaitGroup
for i := 0; i < numMsgs; i++ {
nid, msg, encoded := genEncodedMsg(t, b.LayerIndex, b.ID())
- if i < cutoff {
- tc.mOracle.EXPECT().Validate(gomock.Any(), b.LayerIndex, eligibility.CertifyRound, tc.cfg.CommitteeSize, nid, msg.Proof, defaultCnt).
+ if !tc.concurrent && i < cutoff { // we know exactly which msgs will be validated
+ tcc.mOracle.EXPECT().Validate(gomock.Any(), b.LayerIndex, eligibility.CertifyRound, tcc.cfg.CommitteeSize, nid, msg.Proof, defaultCnt).
Return(true, nil)
}
- res := tc.HandleCertifyMessage(context.TODO(), "peer", encoded)
+ if tc.concurrent {
+ wg.Add(1)
+ go func(data []byte) {
+ res := tcc.HandleCertifyMessage(context.TODO(), "peer", data)
+ require.Equal(t, pubsub.ValidationAccept, res)
+ wg.Done()
+ }(encoded)
+ } else {
+ res := tcc.HandleCertifyMessage(context.TODO(), "peer", encoded)
require.Equal(t, pubsub.ValidationAccept, res)
}
- verifySaved(t, tc.db, b.LayerIndex, b.ID(), cutoff)
+ }
+ wg.Wait()
+ verifySaved(t, tcc.db, b.LayerIndex, b.ID(), cutoff)
+ })
+ }
}
func Test_HandleCertifyMessage_NotRegistered(t *testing.T) {

block certifier: fix certifier concurrent map access (#3601)

## Motivation

race condition, added test that failed with -race -count 5 and pass after the fix