spectests: align with robust message replay

qbft/commit.go

@@ -5,6 +5,7 @@ import (
 	"sort"
 
 	"github.com/pkg/errors"
+
 	"github.com/ssvlabs/ssv-spec/types"
 )
 
@@ -127,7 +128,7 @@ func baseCommitValidationIgnoreSignature(
 		return errors.New("commit msg type is wrong")
 	}
 	if msg.QBFTMessage.Height != height {
-		return errors.New("wrong msg height")
+		return ErrWrongMsgHeight
 	}
 
 	if !msg.SignedMessage.CheckSignersInCommittee(operators) {
@@ -170,7 +171,7 @@ func validateCommit(
 	}
 
 	if msg.QBFTMessage.Round != round {
-		return errors.New("wrong msg round")
+		return ErrWrongMsgRound
 	}
 
 	if !bytes.Equal(proposedMsg.QBFTMessage.Root[:], msg.QBFTMessage.Root[:]) {

qbft/controller.go

@@ -89,18 +89,16 @@ func (c *Controller) ProcessMsg(signedMessage *types.SignedSSVMessage) (*types.S
 		return nil, err
 	}
 	if isFuture {
-		return nil, fmt.Errorf("future msg from height, could not process")
+		return nil, ErrFutureMsg
 	}
 
 	return c.UponExistingInstanceMsg(msg)
-
 }
 
 func (c *Controller) UponExistingInstanceMsg(msg *ProcessingMessage) (*types.SignedSSVMessage, error) {
-
 	inst := c.InstanceForHeight(msg.QBFTMessage.Height)
 	if inst == nil {
-		return nil, errors.New("instance not found")
+		return nil, ErrInstanceNotFound
 	}
 
 	prevDecided, _ := inst.IsDecided()

qbft/error.go

@@ -0,0 +1,13 @@
+package qbft
+
+import (
+	"fmt"
+)
+
+var (
+	ErrInstanceNotFound   = fmt.Errorf("instance not found")
+	ErrFutureMsg          = fmt.Errorf("future msg from height, could not process")
+	ErrNoProposalForRound = fmt.Errorf("no proposal for round")
+	ErrWrongMsgRound      = fmt.Errorf("wrong msg round")
+	ErrWrongMsgHeight     = fmt.Errorf("wrong msg height")
+)

qbft/instance.go

@@ -140,7 +140,7 @@ func (i *Instance) BaseMsgValidation(msg *ProcessingMessage) error {
 	case PrepareMsgType:
 		proposedMsg := i.State.ProposalAcceptedForCurrentRound
 		if proposedMsg == nil {
-			return errors.New("did not receive proposal for this round")
+			return ErrNoProposalForRound
 		}
 		return validSignedPrepareForHeightRoundAndRootIgnoreSignature(
 			msg,
@@ -152,7 +152,7 @@ func (i *Instance) BaseMsgValidation(msg *ProcessingMessage) error {
 	case CommitMsgType:
 		proposedMsg := i.State.ProposalAcceptedForCurrentRound
 		if proposedMsg == nil {
-			return errors.New("did not receive proposal for this round")
+			return ErrNoProposalForRound
 		}
 		return validateCommit(
 			msg,

qbft/prepare.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 
 	"github.com/pkg/errors"
+
 	"github.com/ssvlabs/ssv-spec/types"
 )
 
@@ -90,10 +91,10 @@ func validSignedPrepareForHeightRoundAndRootIgnoreSignature(
 		return errors.New("prepare msg type is wrong")
 	}
 	if msg.QBFTMessage.Height != height {
-		return errors.New("wrong msg height")
+		return ErrWrongMsgHeight
 	}
 	if msg.QBFTMessage.Round != round {
-		return errors.New("wrong msg round")
+		return ErrWrongMsgRound
 	}
 
 	if err := msg.Validate(); err != nil {

qbft/proposal.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 
 	"github.com/pkg/errors"
+
 	"github.com/ssvlabs/ssv-spec/types"
 )
 
@@ -56,7 +57,7 @@ func isValidProposal(
 		return errors.New("msg type is not proposal")
 	}
 	if msg.QBFTMessage.Height != state.Height {
-		return errors.New("wrong msg height")
+		return ErrWrongMsgHeight
 	}
 	if len(msg.SignedMessage.OperatorIDs) != 1 {
 		return errors.New("msg allows 1 signer")

qbft/round_change.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 
 	"github.com/pkg/errors"
+
 	"github.com/ssvlabs/ssv-spec/types"
 )
 
@@ -240,15 +241,14 @@ func validRoundChangeForDataIgnoreSignature(
 	round Round,
 	fullData []byte,
 ) error {
-
 	if msg.QBFTMessage.MsgType != RoundChangeMsgType {
 		return errors.New("round change msg type is wrong")
 	}
 	if msg.QBFTMessage.Height != height {
-		return errors.New("wrong msg height")
+		return ErrWrongMsgHeight
 	}
 	if msg.QBFTMessage.Round != round {
-		return errors.New("wrong msg round")
+		return ErrWrongMsgRound
 	}
 	if len(msg.SignedMessage.OperatorIDs) != 1 {
 		return errors.New("msg allows 1 signer")

qbft/spectest/generate/tests/tests.ControllerSpecTest_qbft_controller_decide_no_quorum.json

@@ -33,7 +33,7 @@
     }
   ],
   "OutputMessages": null,
-  "ExpectedError": "could not process msg: invalid signed message: did not receive proposal for this round",
+  "ExpectedError": "no proposal for round: invalid signed message: did not receive proposal for this round",
   "PrivateKeys": {
     "ValidatorSK": "3515c7d08e5affd729e9579f7588d30f2342ee6f6a9334acf006345262162c6f",
     "ValidatorPK": "8e80066551a81b318258709edaf7dd1f63cd686a0e4db8b29bbb7acfe65608677af5a527d9448ee47835485e02b50bc0",

qbft/spectest/tests/controller/decided/no_quorum.go

@@ -30,8 +30,7 @@ func NoQuorum() tests.SpecTest {
 			},
 		},
 		nil,
-		// TODO: before merge ask engineering how often they see such message in production
-		"could not process msg: invalid signed message: did not receive proposal for this round",
+		"no proposal for round: invalid signed message: did not receive proposal for this round",
 		nil,
 		ks,
 	)

ssv/error.go

@@ -0,0 +1,30 @@
+package ssv
+
+import (
+	"fmt"
+)
+
+var (
+	// Below is a list of retryable errors a runner might encounter.
+
+	// ErrNoRunningDuty means we might not have started the duty yet, while another operator already did + sent this
+	// message to us.
+	ErrNoRunningDuty = fmt.Errorf("no running duty")
+	// ErrFuturePartialSigMsg means the message we've got is "from the future"; it can happen if we haven't advanced
+	// the runner to the slot the message is targeting yet, while another operator already did + sent this message
+	// to us.
+	ErrFuturePartialSigMsg = fmt.Errorf("future partial sig msg")
+	// ErrInstanceNotFound means we might not have started the QBFT instance yet, while another operator already did
+	// + sent this message to us.
+	ErrInstanceNotFound = fmt.Errorf("instance not found")
+	// ErrFutureConsensusMsg means the message we've got is "from the future"; it can happen if we haven't started
+	// QBFT instance for the slot the message is targeting yet, while another operator already did + sent this
+	// message to us.
+	ErrFutureConsensusMsg = fmt.Errorf("future consensus msg")
+	// ErrNoProposalForRound means we might not have received a proposal-message yet, while another operator already
+	// did and started preparing it.
+	ErrNoProposalForRound = fmt.Errorf("no proposal for round")
+	// ErrWrongMsgRound means we might not have changed the round to the next one yet, while another operator already
+	// did + sent this message to us.
+	ErrWrongMsgRound = fmt.Errorf("wrong msg round")
+)

ssv/runner.go

@@ -1,9 +1,12 @@
 package ssv
 
 import (
+	"fmt"
+
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	ssz "github.com/ferranbt/fastssz"
 	"github.com/pkg/errors"
+
 	"github.com/ssvlabs/ssv-spec/qbft"
 	"github.com/ssvlabs/ssv-spec/types"
 )
@@ -122,6 +125,18 @@ func (b *BaseRunner) baseConsensusMsgProcessing(runner Runner, msg *types.Signed
 	}
 
 	decidedSignedMsg, err := b.QBFTController.ProcessMsg(msg)
+	if errors.Is(err, qbft.ErrInstanceNotFound) {
+		return false, nil, fmt.Errorf("%w: %v", ErrInstanceNotFound, err)
+	}
+	if errors.Is(err, qbft.ErrFutureMsg) {
+		return false, nil, fmt.Errorf("%w: %v", ErrFutureConsensusMsg, err)
+	}
+	if errors.Is(err, qbft.ErrNoProposalForRound) {
+		return false, nil, fmt.Errorf("%w: %v", ErrNoProposalForRound, err)
+	}
+	if errors.Is(err, qbft.ErrWrongMsgRound) {
+		return false, nil, fmt.Errorf("%w: %v", ErrWrongMsgRound, err)
+	}
 	if err != nil {
 		return false, nil, err
 	}

ssv/runner_signatures.go

@@ -1,10 +1,13 @@
 package ssv
 
 import (
+	"fmt"
+
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	ssz "github.com/ferranbt/fastssz"
 	"github.com/herumi/bls-eth-go-binary/bls"
 	"github.com/pkg/errors"
+
 	"github.com/ssvlabs/ssv-spec/types"
 )
 
@@ -32,15 +35,19 @@ func (b *BaseRunner) signBeaconObject(runner Runner, duty *types.ValidatorDuty,
 }
 
 // Validate message content without verifying signatures
-func (b *BaseRunner) validatePartialSigMsgForSlot(
-	psigMsgs *types.PartialSignatureMessages,
-	slot phase0.Slot,
-) error {
+func (b *BaseRunner) validatePartialSigMsg(psigMsgs *types.PartialSignatureMessages, expectedSlot phase0.Slot) error {
 	if err := psigMsgs.Validate(); err != nil {
 		return errors.Wrap(err, "PartialSignatureMessages invalid")
 	}
-	if psigMsgs.Slot != slot {
-		return errors.New("invalid partial sig slot")
+
+	if psigMsgs.Slot < expectedSlot {
+		// this message is targeting a slot that's already passed - our runner has advanced to the next slot already,
+		// and we cannot process it anymore
+		return fmt.Errorf("invalid partial sig slot: %d, want: %d", psigMsgs.Slot, expectedSlot)
+	}
+
+	if psigMsgs.Slot > expectedSlot {
+		return fmt.Errorf("%w, message slot: %d, want slot: %d", ErrFuturePartialSigMsg, psigMsgs.Slot, expectedSlot)
 	}
 
 	// Get signer

ssv/runner_validations.go

@@ -7,15 +7,16 @@ import (
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	ssz "github.com/ferranbt/fastssz"
 	"github.com/pkg/errors"
+
 	"github.com/ssvlabs/ssv-spec/types"
 )
 
 func (b *BaseRunner) ValidatePreConsensusMsg(runner Runner, psigMsgs *types.PartialSignatureMessages) error {
 	if !b.hasRunningDuty() {
-		return errors.New("no running duty")
+		return ErrNoRunningDuty
 	}
 
-	if err := b.validatePartialSigMsgForSlot(psigMsgs, b.State.StartingDuty.DutySlot()); err != nil {
+	if err := b.validatePartialSigMsg(psigMsgs, b.State.StartingDuty.DutySlot()); err != nil {
 		return err
 	}
 
@@ -46,7 +47,7 @@ func (b *BaseRunner) FallBackAndVerifyEachSignature(container *PartialSigContain
 
 func (b *BaseRunner) ValidatePostConsensusMsg(runner Runner, psigMsgs *types.PartialSignatureMessages) error {
 	if !b.hasRunningDuty() {
-		return errors.New("no running duty")
+		return ErrNoRunningDuty
 	}
 
 	// TODO https://github.com/ssvlabs/ssv-spec/issues/142 need to fix with this issue solution instead.
@@ -70,14 +71,14 @@ func (b *BaseRunner) ValidatePostConsensusMsg(runner Runner, psigMsgs *types.Par
 			return errors.Wrap(err, "failed to parse decided value to BeaconData")
 		}
 
-		return b.validatePartialSigMsgForSlot(psigMsgs, b.State.StartingDuty.DutySlot())
+		return b.validatePartialSigMsg(psigMsgs, b.State.StartingDuty.DutySlot())
 	default:
 		decidedValue := &types.ValidatorConsensusData{}
 		if err := decidedValue.Decode(decidedValueBytes); err != nil {
 			return errors.Wrap(err, "failed to parse decided value to ValidatorConsensusData")
 		}
 
-		if err := b.validatePartialSigMsgForSlot(psigMsgs, decidedValue.Duty.Slot); err != nil {
+		if err := b.validatePartialSigMsg(psigMsgs, decidedValue.Duty.Slot); err != nil {
 			return err
 		}
 

ssv/spectest/generate/tests/tests.MultiMsgProcessingSpecTest_consensus_future_message.json

@@ -96,7 +96,7 @@
       "OutputMessages": null,
       "BeaconBroadcastedRoots": null,
       "DontStartDuty": true,
-      "ExpectedError": "failed processing consensus message: future msg from height, could not process",
+      "ExpectedError": "failed processing consensus message: future consensus msg: future msg from height, could not process",
       "ValidatorDuty": {
         "Type": 4,
         "PubKey": "0x8e80066551a81b318258709edaf7dd1f63cd686a0e4db8b29bbb7acfe65608677af5a527d9448ee47835485e02b50bc0",
@@ -206,7 +206,7 @@
       "OutputMessages": null,
       "BeaconBroadcastedRoots": null,
       "DontStartDuty": true,
-      "ExpectedError": "failed processing consensus message: future msg from height, could not process",
+      "ExpectedError": "failed processing consensus message: future consensus msg: future msg from height, could not process",
       "ValidatorDuty": {
         "Type": 2,
         "PubKey": "0x8e80066551a81b318258709edaf7dd1f63cd686a0e4db8b29bbb7acfe65608677af5a527d9448ee47835485e02b50bc0",
@@ -312,7 +312,7 @@
       "OutputMessages": null,
       "BeaconBroadcastedRoots": null,
       "DontStartDuty": true,
-      "ExpectedError": "failed processing consensus message: future msg from height, could not process",
+      "ExpectedError": "failed processing consensus message: future consensus msg: future msg from height, could not process",
       "ValidatorDuty": {
         "Type": 2,
         "PubKey": "0x8e80066551a81b318258709edaf7dd1f63cd686a0e4db8b29bbb7acfe65608677af5a527d9448ee47835485e02b50bc0",
@@ -598,7 +598,7 @@
       "OutputMessages": null,
       "BeaconBroadcastedRoots": null,
       "DontStartDuty": true,
-      "ExpectedError": "failed processing consensus message: future msg from height, could not process",
+      "ExpectedError": "failed processing consensus message: future consensus msg: future msg from height, could not process",
       "ValidatorDuty": {
         "Type": 1,
         "PubKey": "0x8e80066551a81b318258709edaf7dd1f63cd686a0e4db8b29bbb7acfe65608677af5a527d9448ee47835485e02b50bc0",
@@ -704,7 +704,7 @@
       "OutputMessages": null,
       "BeaconBroadcastedRoots": null,
       "DontStartDuty": true,
-      "ExpectedError": "failed processing consensus message: future msg from height, could not process",
+      "ExpectedError": "failed processing consensus message: future consensus msg: future msg from height, could not process",
       "ValidatorDuty": {
         "Type": 1,
         "PubKey": "0x8e80066551a81b318258709edaf7dd1f63cd686a0e4db8b29bbb7acfe65608677af5a527d9448ee47835485e02b50bc0",