Mitigate the occurrences of 'failed to receive candidate' error

pkg/core/candidate/collector.go

@@ -0,0 +1,153 @@
+// This Source Code Form is subject to the terms of the MIT License.
+// If a copy of the MIT License was not distributed with this
+// file, you can obtain one at https://opensource.org/licenses/MIT.
+//
+// Copyright (c) DUSK NETWORK. All rights reserved.
+
+package candidate
+
+import (
+	"bytes"
+	"errors"
+
+	"github.com/dusk-network/dusk-blockchain/pkg/config"
+	"github.com/dusk-network/dusk-blockchain/pkg/core/consensus/committee"
+	"github.com/dusk-network/dusk-blockchain/pkg/core/consensus/header"
+	"github.com/dusk-network/dusk-blockchain/pkg/core/consensus/msg"
+	"github.com/dusk-network/dusk-blockchain/pkg/core/database"
+	"github.com/dusk-network/dusk-blockchain/pkg/p2p/wire/message"
+	"github.com/dusk-network/dusk-blockchain/pkg/p2p/wire/topics"
+	"github.com/dusk-network/dusk-blockchain/pkg/util/nativeutils/eventbus"
+	"github.com/sirupsen/logrus"
+)
+
+var (
+	// ErrInvalidNewBlock error invalid newblock message.
+	ErrInvalidNewBlock = errors.New("invalid newblock message")
+	// ErrInvalidBlockHash ...
+	ErrInvalidBlockHash = errors.New("invalid block hash")
+	// ErrNotBlockGenerator ...
+	ErrNotBlockGenerator = errors.New("message not signed by generator")
+	// ErrInvalidMsgRound ...
+	ErrInvalidMsgRound = errors.New("invalid message round")
+	// ErrInvalidStep ...
+	ErrInvalidStep = errors.New("invalid newblock step")
+	// ErrMaxStepExceeded ...
+	ErrMaxStepExceeded = errors.New("max step exceeded")
+)
+
+// Collector implements a procedure of collecting candidate block from a wire message (newblock).
+type Collector struct {
+	eventbus *eventbus.EventBus
+	handler  *committee.Handler
+	db       database.DB
+
+	round    uint64
+	step     uint8
+	stepName string
+}
+
+// NewCollector instantiates Collector.
+func NewCollector(e *eventbus.EventBus, h *committee.Handler, db database.DB, round uint64) *Collector {
+	return &Collector{
+		eventbus: e,
+		handler:  h,
+		db:       db,
+		round:    round,
+	}
+}
+
+// UpdateStep set step/stepName.
+func (c *Collector) UpdateStep(step uint8, name string) {
+	c.step = step
+	c.stepName = name
+}
+
+// Collect put a candidate block from message.Block to DB, if message is valid.
+func (c *Collector) Collect(msg message.NewBlock, msgHeader []byte) error {
+	if msg.State().Round != c.round {
+		return ErrInvalidMsgRound
+	}
+
+	msgStep := msg.State().Step
+
+	if msgStep >= config.ConsensusMaxStep {
+		return ErrMaxStepExceeded
+	}
+
+	// Check msg.State().Step belongs to current iteration
+	if msgStep == 0 ||
+		(msgStep-1)/3 != (c.step-1)/3 {
+		return errors.New("invalid newblock step")
+	}
+
+	log := logrus.WithField("process", c.stepName)
+
+	if err := c.verify(msg); err != nil {
+		return ErrInvalidNewBlock
+	}
+
+	// Persist Candidate Block on disk.
+	if err := c.db.Update(func(t database.Transaction) error {
+		// TODO: Check a candidate from this BLSKey for this iteration has been registered
+		return t.StoreCandidateMessage(msg.Candidate)
+	}); err != nil {
+		log.WithError(err).Errorln("could not store candidate")
+	}
+
+	// Once the event is verified, and has passed all preliminary checks,
+	// we can republish it to the network.
+	m := message.NewWithHeader(topics.NewBlock, msg, msgHeader)
+
+	buf, err := message.Marshal(m)
+	if err != nil {
+		return err
+	}
+
+	serialized := message.NewWithHeader(m.Category(), buf, m.Header())
+	_ = c.eventbus.Publish(topics.Kadcast, serialized)
+
+	return nil
+}
+
+// verify executes a set of check points to ensure the hash of the candidate
+// block has been signed by the single committee member of selection step of
+// this iteration.
+func (c *Collector) verify(msg message.NewBlock) error {
+	if !c.handler.IsMember(msg.State().PubKeyBLS, msg.State().Round, msg.State().Step, config.ConsensusSelectionMaxCommitteeSize) {
+		return ErrNotBlockGenerator
+	}
+
+	// Verify message signagure
+	if err := verifySignature(msg); err != nil {
+		return err
+	}
+
+	// Sanity-check the candidate block
+	if err := SanityCheckCandidate(msg.Candidate); err != nil {
+		return err
+	}
+
+	// Ensure candidate block hash is equal to the BlockHash of the msg.header
+	hash, err := msg.Candidate.CalculateHash()
+	if err != nil {
+		return err
+	}
+
+	if !bytes.Equal(msg.State().BlockHash, hash) {
+		return ErrInvalidBlockHash
+	}
+
+	return nil
+}
+
+func verifySignature(scr message.NewBlock) error {
+	packet := new(bytes.Buffer)
+
+	hdr := scr.State()
+	if err := header.MarshalSignableVote(packet, hdr); err != nil {
+		return err
+	}
+
+	return msg.VerifyBLSSignature(hdr.PubKeyBLS, scr.SignedHash, packet.Bytes())
+}

pkg/core/consensus/reduction/firststep/step.go

@@ -87,13 +87,17 @@ func (p *Phase) Run(ctx context.Context, queue *consensus.Queue, evChan chan mes
 		tlog.Traceln("ending first reduction step")
 	}()
 
+	p.handler = reduction.NewHandler(p.Keys, r.P, r.Seed)
+
+	collector := candidate.NewCollector(p.EventBus, p.handler.Handler, p.db, r.Round)
+	collector.UpdateStep(step, "1st_reduction")
+
 	if log.GetLevel() >= logrus.DebugLevel {
 		c := p.selectionResult.Candidate
 		tlog.WithField("hash", util.StringifyBytes(c.Header.Hash)).
 			Debug("initialized")
 	}
 
-	p.handler = reduction.NewHandler(p.Keys, r.P, r.Seed)
 	// first we send our own Selection
 	if p.handler.AmMember(r.Round, step) {
 		m, _ := p.SendReduction(r.Round, step, &p.selectionResult.Candidate)
@@ -131,19 +135,27 @@ func (p *Phase) Run(ctx context.Context, queue *consensus.Queue, evChan chan mes
 	for {
 		select {
 		case ev := <-evChan:
-			if reduction.ShouldProcess(ev, r.Round, step, queue) {
-				rMsg := ev.Payload().(message.Reduction)
-				if !p.handler.IsMember(rMsg.Sender(), r.Round, step) {
-					continue
-				}
-
-				sv := p.collectReduction(ctx, rMsg, r.Round, step, ev.Header())
-				if sv != nil {
-					// preventing timeout leakage
-					go func() {
-						<-timeoutChan
-					}()
-					return p.gotoNextPhase(sv)
+			switch ev.Category() {
+			case topics.NewBlock:
+				// Collect and repropagate the candidate block of this iteration.
+				b := ev.Payload().(message.NewBlock)
+				_ = collector.Collect(b, ev.Header())
+
+			case topics.Reduction:
+				if reduction.ShouldProcess(ev, r.Round, step, queue) {
+					rMsg := ev.Payload().(message.Reduction)
+					if !p.handler.IsMember(rMsg.Sender(), r.Round, step) {
+						continue
+					}
+
+					sv := p.collectReduction(ctx, rMsg, r.Round, step, ev.Header())
+					if sv != nil {
+						// preventing timeout leakage
+						go func() {
+							<-timeoutChan
+						}()
+						return p.gotoNextPhase(sv)
+					}
 				}
 			}
 

pkg/core/consensus/reduction/reduction.go

@@ -179,14 +179,23 @@ func ShouldProcess(m message.Message, round uint64, step uint8, queue *consensus
 
 	cmp := hdr.CompareRoundAndStep(round, step)
 	if cmp == header.Before {
-		lg.
+		l := lg.
 			WithFields(log.Fields{
-				"topic":          m.Category(),
-				"round":          hdr.Round,
-				"step":           hdr.Step,
+				"topic":          m.Category().String(),
+				"msg_round":      hdr.Round,
+				"msg_step":       hdr.Step,
 				"expected round": round,
-			}).
-			Debugln("discarding obsolete event")
+				"expected step":  step,
+			})
+
+		// Report as a warning an event of discarding newblock while in a
+		// reduction phase.
+		if m.Category() == topics.NewBlock && round == hdr.Round {
+			l.Warnln("discarding newblock event")
+		} else {
+			l.Debugln("discarding obsolete event")
+		}
+
 		return false
 	}
 

pkg/core/consensus/reduction/secondstep/reduction_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/dusk-network/dusk-blockchain/pkg/core/consensus"
 	"github.com/dusk-network/dusk-blockchain/pkg/core/consensus/reduction"
 	"github.com/dusk-network/dusk-blockchain/pkg/core/data/block"
+	"github.com/dusk-network/dusk-blockchain/pkg/core/database/lite"
 	"github.com/dusk-network/dusk-blockchain/pkg/p2p/wire/message"
 	"github.com/dusk-network/dusk-blockchain/pkg/p2p/wire/topics"
 	"github.com/dusk-network/dusk-blockchain/pkg/util/nativeutils/eventbus"
@@ -34,10 +35,13 @@ func TestSendReduction(t *testing.T) {
 	hash, err := crypto.RandEntropy(32)
 	require.NoError(t, err)
 
+	_, db := lite.CreateDBConnection()
+	defer db.Close()
+
 	timeout := time.Second
 
 	hlp := reduction.NewHelper(messageToSpawn, timeout)
-	secondStep := New(hlp.Emitter, verifyFn, 10*time.Second)
+	secondStep := New(hlp.Emitter, verifyFn, 10*time.Second, db)
 
 	// Generate second StepVotes
 	svs := message.GenVotes(hash, []byte{0, 0, 0, 0}, 1, 2, hlp.ProvisionersKeys, hlp.P)
@@ -140,6 +144,9 @@ func TestSecondStepReduction(t *testing.T) {
 	hash, err := crypto.RandEntropy(32)
 	require.NoError(t, err)
 
+	_, db := lite.CreateDBConnection()
+	defer db.Close()
+
 	timeout := time.Second
 
 	table := initiateTableTest(timeout, hash, round, step)
@@ -173,7 +180,7 @@ func TestSecondStepReduction(t *testing.T) {
 			}
 
 			// spin secondStepVotes
-			secondStepReduction := New(hlp.Emitter, verifyFn, timeout)
+			secondStepReduction := New(hlp.Emitter, verifyFn, timeout, db)
 
 			// Generate second StepVotes
 			svs := message.GenVotes(hash, []byte{0, 0, 0, 0}, 1, 2, hlp.ProvisionersKeys, hlp.P)

pkg/core/consensus/reduction/secondstep/step.go

@@ -13,10 +13,12 @@ import (
 	"time"
 
 	"github.com/dusk-network/dusk-blockchain/pkg/config"
+	"github.com/dusk-network/dusk-blockchain/pkg/core/candidate"
 	"github.com/dusk-network/dusk-blockchain/pkg/core/consensus"
 	"github.com/dusk-network/dusk-blockchain/pkg/core/consensus/header"
 	"github.com/dusk-network/dusk-blockchain/pkg/core/consensus/reduction"
 	"github.com/dusk-network/dusk-blockchain/pkg/core/data/block"
+	"github.com/dusk-network/dusk-blockchain/pkg/core/database"
 	"github.com/dusk-network/dusk-blockchain/pkg/p2p/wire/message"
 	"github.com/dusk-network/dusk-blockchain/pkg/p2p/wire/topics"
 	"github.com/dusk-network/dusk-blockchain/pkg/util"
@@ -39,6 +41,7 @@ type Phase struct {
 	*reduction.Reduction
 	handler    *reduction.Handler
 	aggregator *reduction.Aggregator
+	db         database.DB
 
 	firstStepVotesMsg message.StepVotesMsg
 
@@ -51,13 +54,14 @@ type Phase struct {
 // NB: we cannot push the agreement directly within the agreementChannel
 // until we have a way to deduplicate it from the peer (the dupemap will not be
 // notified of duplicates).
-func New(e *consensus.Emitter, verifyFn consensus.CandidateVerificationFunc, timeOut time.Duration) *Phase {
+func New(e *consensus.Emitter, verifyFn consensus.CandidateVerificationFunc, timeOut time.Duration, db database.DB) *Phase {
 	return &Phase{
 		Reduction: &reduction.Reduction{
 			Emitter:  e,
 			TimeOut:  timeOut,
 			VerifyFn: verifyFn,
 		},
+		db: db,
 	}
 }
 
@@ -94,6 +98,10 @@ func (p *Phase) Run(ctx context.Context, queue *consensus.Queue, evChan chan mes
 	}
 
 	p.handler = reduction.NewHandler(p.Keys, r.P, r.Seed)
+
+	collector := candidate.NewCollector(p.EventBus, p.handler.Handler, p.db, r.Round)
+	collector.UpdateStep(step, "2nd_reduction")
+
 	// first we send our own Selection
 
 	if p.handler.AmMember(r.Round, step) {