Simplify Epoch Collector to Store One Vote Per Validator

nonvalidator/epochs.go

@@ -145,18 +145,22 @@ type latestValidatorSetRetriever interface {
 type epochDigestCounter struct {
 	logger common.Logger
 
-	// sealingBlockResponses stores sealing blocks that have been received by our non-validator.
-	// It maps the epoch they are creating, to the NodeIds that have sent them to us(and which digest they sent).
+	// sealingBlockResponses stores the latest sealing block each validator has sent us keyed by NodeID.
 	// Once we have collected f+1 messages for a finalization, the sealing block for that epoch is validated.
-	sealingBlockResponses map[uint64]map[string]common.Digest
+	sealingBlockResponses map[string]sealingBlockResponse
 
 	// latestValidatorSetRetriever is used to calculate the threshold of votes needed to validate an epoch
 	latestValidatorSetRetriever latestValidatorSetRetriever
 }
 
+type sealingBlockResponse struct {
+	epoch  uint64
+	digest common.Digest
+}
+
 func newEpochReplicator(logger common.Logger, validatorSetRetriever latestValidatorSetRetriever) *epochDigestCounter {
 	return &epochDigestCounter{
-		sealingBlockResponses:       make(map[uint64]map[string]common.Digest),
+		sealingBlockResponses:       make(map[string]sealingBlockResponse),
 		logger:                      logger,
 		latestValidatorSetRetriever: validatorSetRetriever,
 	}
@@ -179,36 +183,36 @@ func (e *epochDigestCounter) collectedSealingBlockInfo(sealingBlockInfo *common.
 
 	e.logger.Debug("Collected a sealing block", zap.Stringer("QR", sealingBlockInfo), zap.Stringer("From", from))
 
-	threshold := common.F(len(e.latestValidatorSetRetriever.Validators())) + 1
-	newEpoch := bh.Seq
-	epochResponses, ok := e.sealingBlockResponses[newEpoch]
-	digest := bh.Digest
-	if !ok {
-		epochResponses = make(map[string]common.Digest)
-		e.sealingBlockResponses[newEpoch] = epochResponses
+	threshold := common.F(len(validators)) + 1
+	// the sequence number is the epoch the sealing block creates
+	response := sealingBlockResponse{epoch: bh.Seq, digest: bh.Digest}
+	if stored, ok := e.sealingBlockResponses[string(from)]; ok && stored.epoch > response.epoch {
+		e.logger.Debug("Ignoring sealing block for a lower epoch than already collected from this node", zap.Stringer("From", from), zap.Uint64("Stored Epoch", stored.epoch), zap.Uint64("Epoch", response.epoch))
+		return false
 	}
-	epochResponses[string(from)] = digest
+	e.sealingBlockResponses[string(from)] = response
 
 	// check if we have a threshold of responses
-	counts := make(map[common.Digest]uint64)
-	for _, digest := range epochResponses {
-		count := counts[digest]
+	count := 0
+	for _, other := range e.sealingBlockResponses {
+		if other != response {
+			continue
+		}
 		count++
-		counts[digest] = count
-
-		if counts[digest] >= uint64(threshold) {
-			e.logger.Info("We received enough messages to validate a higher epoch", zap.Stringer("EpochInfo", sealingBlockInfo), zap.Int("Threshold", threshold), zap.Uint64("Responses", counts[digest]))
+		if count >= threshold {
+			e.logger.Info("We received enough messages to validate a higher epoch", zap.Stringer("EpochInfo", sealingBlockInfo), zap.Int("Threshold", threshold), zap.Int("Responses", count))
 			return true
 		}
 	}
 
 	return false
 }
 
+// removeOldEpochs deletes all responses for epochs strictly less than minEpochToKeep.
 func (e *epochDigestCounter) removeOldEpochs(minEpochToKeep uint64) {
-	for epoch := range e.sealingBlockResponses {
-		if epoch < minEpochToKeep {
-			delete(e.sealingBlockResponses, epoch)
+	for from, response := range e.sealingBlockResponses {
+		if response.epoch < minEpochToKeep {
+			delete(e.sealingBlockResponses, from)
 		}
 	}
 }

nonvalidator/epochs_test.go

@@ -297,3 +297,26 @@ func TestCollectedQuorumRound(t *testing.T) {
 		})
 	}
 }
+
+// TestCollectedSealingBlockInfoOneResponsePerValidator asserts each validator counts
+// toward one epoch, the highest it has sent, so a single validator can never reach
+// the threshold and re-sending an older sealing block does not discard a newer one.
+func TestCollectedSealingBlockInfoOneResponsePerValidator(t *testing.T) {
+	qr := newSealingQuorumRound(1, 4)
+	info := qr.Block.SealingBlockInfo()
+	validators := info.ValidatorSet
+	e := newEpochReplicator(testutil.MakeLogger(t, 1), &testValidatorSetRetriever{
+		nodes: validators,
+	})
+
+	seq5 := newSealingTestBlock(5, 1, common.Digest{}, info).BlockHeader()
+	seq6 := newSealingTestBlock(6, 1, common.Digest{}, info).BlockHeader()
+
+	// threshold is 2, one validator sending distinct sequences never reaches it
+	require.False(t, e.collectedSealingBlockInfo(info, seq6, validators[0].Id))
+	require.False(t, e.collectedSealingBlockInfo(info, seq5, validators[0].Id))
+
+	// voters[0] still counts toward seq 6, not the older seq 5
+	require.False(t, e.collectedSealingBlockInfo(info, seq5, validators[1].Id))
+	require.True(t, e.collectedSealingBlockInfo(info, seq6, validators[1].Id))
+}