chore: add more logs and metrics

message/validation/consensus_validation.go

@@ -34,51 +34,51 @@ func (mv *messageValidator) validateConsensusMessage(
 		e := ErrSSVDataTooBig
 		e.got = len(ssvMessage.Data)
 		e.want = maxEncodedConsensusMsgSize
-		return nil, e
+		return nil, withValidationStage(SSVValidationStageConsensus, e)
 	}
 
 	consensusMessage, err := specqbft.DecodeMessage(ssvMessage.Data)
 	if err != nil {
 		e := ErrUndecodableMessageData
 		e.innerErr = err
-		return nil, e
+		return nil, withValidationStage(SSVValidationStageConsensus, e)
 	}
 
 	if err := mv.validateConsensusMessageSemantics(signedSSVMessage, consensusMessage, committeeInfo.committee); err != nil {
-		return consensusMessage, err
+		return consensusMessage, withValidationStage(SSVValidationStageConsensus, err)
 	}
 
 	state := mv.validatorState(ssvMessage.GetID(), committeeInfo)
 
 	if err := mv.validateQBFTLogic(signedSSVMessage, consensusMessage, committeeInfo, receivedFrom, receivedAt, state); err != nil {
-		return consensusMessage, err
+		return consensusMessage, withValidationStage(SSVValidationStageConsensus, err)
 	}
 
 	if err := mv.validateQBFTMessageByDutyLogic(signedSSVMessage, consensusMessage, committeeInfo, receivedAt, state); err != nil {
-		return consensusMessage, err
+		return consensusMessage, withValidationStage(SSVValidationStageConsensus, err)
 	}
 
 	if err := ctx.Err(); err != nil {
-		return consensusMessage, err
+		return consensusMessage, withValidationStage(SSVValidationStageContext, err)
 	}
 
 	for i := range signedSSVMessage.Signatures {
 		operatorID := signedSSVMessage.OperatorIDs[i]
 		signature := signedSSVMessage.Signatures[i]
 
 		if err := ctx.Err(); err != nil {
-			return consensusMessage, err
+			return consensusMessage, withValidationStage(SSVValidationStageContext, err)
 		}
 
 		if err := mv.signatureVerifier.VerifySignature(operatorID, ssvMessage, signature); err != nil {
 			e := ErrSignatureVerification
 			e.innerErr = fmt.Errorf("verify opid: %v signature: %w", operatorID, err)
-			return consensusMessage, e
+			return consensusMessage, withValidationStage(SSVValidationStageSignatureVerify, e)
 		}
 	}
 
 	if err := mv.updateConsensusState(signedSSVMessage, consensusMessage, committeeInfo, receivedFrom, state); err != nil {
-		return consensusMessage, err
+		return consensusMessage, withValidationStage(SSVValidationStageStateUpdate, err)
 	}
 
 	return consensusMessage, nil

message/validation/errors.go

@@ -143,8 +143,11 @@ var (
 	ErrZeroRound                               = Error{text: "zero round", reject: true}
 )
 
-func (mv *messageValidator) handleValidationError(ctx context.Context, peerID peer.ID, decodedMessage *queue.SSVMessage, err error) pubsub.ValidationResult {
+func (mv *messageValidator) handleValidationError(ctx context.Context, peerID peer.ID, decodedMessage *queue.SSVMessage, pmsg *pubsub.Message, err error) pubsub.ValidationResult {
 	loggerFields := mv.buildLoggerFields(decodedMessage)
+	stage := validationStageFromError(err)
+	topic := pubsubMessageTopic(pmsg)
+	payloadSize := pubsubMessagePayloadSize(pmsg)
 
 	logger := mv.logger.
 		With(loggerFields.AsZapFields()...).
@@ -153,20 +156,20 @@ func (mv *messageValidator) handleValidationError(ctx context.Context, peerID pe
 	switch {
 	case errors.Is(err, context.DeadlineExceeded):
 		recordIgnoredMessage(ctx, loggerFields.Role, validationTimeoutReason)
-		mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, validationTimeoutReason, err)
+		mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, validationTimeoutReason, stage, topic, payloadSize, err)
 		logger.Debug("ignoring message due to validation timeout", zap.Error(err))
 		return pubsub.ValidationIgnore
 	case errors.Is(err, context.Canceled):
 		recordIgnoredMessage(ctx, loggerFields.Role, validationCanceledReason)
-		mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, validationCanceledReason, err)
+		mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, validationCanceledReason, stage, topic, payloadSize, err)
 		logger.Debug("ignoring message due to validation cancellation", zap.Error(err))
 		return pubsub.ValidationIgnore
 	}
 
 	var valErr Error
 	if !errors.As(err, &valErr) {
 		recordIgnoredMessage(ctx, loggerFields.Role, err.Error())
-		mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, err.Error(), err)
+		mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, err.Error(), stage, topic, payloadSize, err)
 		logger.Debug("ignoring invalid message", zap.Error(err))
 		return pubsub.ValidationIgnore
 	}
@@ -176,7 +179,7 @@ func (mv *messageValidator) handleValidationError(ctx context.Context, peerID pe
 			logger.Debug("ignoring invalid message", zap.Error(valErr))
 		}
 		recordIgnoredMessage(ctx, loggerFields.Role, valErr.Text())
-		mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, valErr.Text(), valErr)
+		mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationIgnored, valErr.Text(), stage, topic, payloadSize, valErr)
 		return pubsub.ValidationIgnore
 	}
 
@@ -185,16 +188,17 @@ func (mv *messageValidator) handleValidationError(ctx context.Context, peerID pe
 	}
 
 	recordRejectedMessage(ctx, loggerFields.Role, valErr.Text())
-	mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationRejected, valErr.Text(), valErr)
+	mv.observeSSVValidation(ctx, peerID, loggerFields, SSVValidationRejected, valErr.Text(), stage, topic, payloadSize, valErr)
 	return pubsub.ValidationReject
 }
 
-func (mv *messageValidator) handleValidationSuccess(ctx context.Context, peerID peer.ID, decodedMessage *queue.SSVMessage) pubsub.ValidationResult {
+func (mv *messageValidator) handleValidationSuccess(ctx context.Context, peerID peer.ID, decodedMessage *queue.SSVMessage, pmsg *pubsub.Message) pubsub.ValidationResult {
 	recordAcceptedMessage(ctx, messageRole(decodedMessage))
 	if mv.observer == nil {
 		return pubsub.ValidationAccept
 	}
-	mv.observeSSVValidation(ctx, peerID, mv.buildLoggerFields(decodedMessage), SSVValidationAccepted, "valid", nil)
+	mv.observeSSVValidation(ctx, peerID, mv.buildLoggerFields(decodedMessage), SSVValidationAccepted, "valid",
+		SSVValidationStageComplete, pubsubMessageTopic(pmsg), pubsubMessagePayloadSize(pmsg), nil)
 	return pubsub.ValidationAccept
 }
 
@@ -204,6 +208,9 @@ func (mv *messageValidator) observeSSVValidation(
 	loggerFields *LoggerFields,
 	outcome string,
 	reason string,
+	stage string,
+	topic string,
+	payloadSize int,
 	err error,
 ) {
 	if mv.observer == nil {
@@ -217,6 +224,9 @@ func (mv *messageValidator) observeSSVValidation(
 		PeerID:         peerID,
 		Outcome:        outcome,
 		Reason:         reason,
+		Stage:          stage,
+		Topic:          topic,
+		PayloadSize:    payloadSize,
 		Role:           loggerFields.Role,
 		SSVMessageType: loggerFields.SSVMessageType,
 		Slot:           loggerFields.Slot,
@@ -232,3 +242,17 @@ func (mv *messageValidator) observeSSVValidation(
 	}
 	mv.observer.ObserveSSVValidation(ctx, mv.logger, event)
 }
+
+func pubsubMessageTopic(pmsg *pubsub.Message) string {
+	if pmsg == nil {
+		return ""
+	}
+	return pmsg.GetTopic()
+}
+
+func pubsubMessagePayloadSize(pmsg *pubsub.Message) int {
+	if pmsg == nil {
+		return 0
+	}
+	return len(pmsg.GetData())
+}

message/validation/observer.go

@@ -16,12 +16,31 @@ const (
 	SSVValidationRejected = "rejected"
 )
 
+const (
+	SSVValidationStageUnknown         = "unknown"
+	SSVValidationStageContext         = "context"
+	SSVValidationStagePubsubBasic     = "pubsub_basic"
+	SSVValidationStageDecodeSigned    = "decode_signed"
+	SSVValidationStageSignedSemantics = "signed_semantics"
+	SSVValidationStageSSVSemantics    = "ssv_semantics"
+	SSVValidationStageCommitteeLookup = "committee_lookup"
+	SSVValidationStageCommitteeChecks = "committee_checks"
+	SSVValidationStageConsensus       = "consensus_validation"
+	SSVValidationStagePartial         = "partial_validation"
+	SSVValidationStageSignatureVerify = "signature_verification"
+	SSVValidationStageStateUpdate     = "state_update"
+	SSVValidationStageComplete        = "complete"
+)
+
 // SSVValidationEvent describes the SSV-level validation decision before it
 // is reduced to libp2p's accept/reject/ignore validation result.
 type SSVValidationEvent struct {
 	PeerID         peer.ID
 	Outcome        string
 	Reason         string
+	Stage          string
+	Topic          string
+	PayloadSize    int
 	Role           spectypes.RunnerRole
 	SSVMessageType spectypes.MsgType
 	Slot           phase0.Slot

message/validation/observer_event_test.go

@@ -0,0 +1,54 @@
+package validation
+
+import (
+	"context"
+	"testing"
+
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	pspb "github.com/libp2p/go-libp2p-pubsub/pb"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+type captureSSVValidationObserver struct {
+	event SSVValidationEvent
+}
+
+func (o *captureSSVValidationObserver) ObserveSSVValidation(_ context.Context, _ *zap.Logger, event SSVValidationEvent) {
+	o.event = event
+}
+
+func TestHandleValidationErrorObservesStageTopicAndPayload(t *testing.T) {
+	observer := &captureSSVValidationObserver{}
+	mv := &messageValidator{
+		logger:   zap.NewNop(),
+		observer: observer,
+	}
+
+	topic := "ssv.v2.42"
+	pmsg := &pubsub.Message{
+		Message: &pspb.Message{
+			Topic: &topic,
+			Data:  []byte{1, 2, 3, 4},
+		},
+	}
+	pid := peer.ID("peer-a")
+
+	result := mv.handleValidationError(t.Context(), pid, nil, pmsg, withValidationStage(SSVValidationStageDecodeSigned, ErrMalformedPubSubMessage))
+
+	require.Equal(t, pubsub.ValidationReject, result)
+	require.Equal(t, pid, observer.event.PeerID)
+	require.Equal(t, SSVValidationRejected, observer.event.Outcome)
+	require.Equal(t, ErrMalformedPubSubMessage.Text(), observer.event.Reason)
+	require.Equal(t, SSVValidationStageDecodeSigned, observer.event.Stage)
+	require.Equal(t, topic, observer.event.Topic)
+	require.Equal(t, 4, observer.event.PayloadSize)
+}
+
+func TestWithValidationStagePreservesErrorMatching(t *testing.T) {
+	err := withValidationStage(SSVValidationStagePubsubBasic, ErrPubSubMessageHasNoData)
+
+	require.ErrorIs(t, err, ErrPubSubMessageHasNoData)
+	require.Equal(t, SSVValidationStagePubsubBasic, validationStageFromError(err))
+}

message/validation/partial_validation.go

@@ -31,39 +31,39 @@ func (mv *messageValidator) validatePartialSignatureMessage(
 		e := ErrSSVDataTooBig
 		e.got = len(ssvMessage.Data)
 		e.want = maxEncodedPartialSignatureSize
-		return nil, e
+		return nil, withValidationStage(SSVValidationStagePartial, e)
 	}
 
 	partialSignatureMessages := &spectypes.PartialSignatureMessages{}
 	if err := partialSignatureMessages.Decode(ssvMessage.Data); err != nil {
 		e := ErrUndecodableMessageData
 		e.innerErr = err
-		return nil, e
+		return nil, withValidationStage(SSVValidationStagePartial, e)
 	}
 
 	if err := mv.validatePartialSignatureMessageSemantics(signedSSVMessage, partialSignatureMessages, committeeInfo.validatorIndices); err != nil {
-		return partialSignatureMessages, err
+		return partialSignatureMessages, withValidationStage(SSVValidationStagePartial, err)
 	}
 
 	state := mv.validatorState(ssvMessage.GetID(), committeeInfo)
 	if err := mv.validatePartialSigMessagesByDutyLogic(signedSSVMessage, partialSignatureMessages, committeeInfo, receivedFrom, receivedAt, state); err != nil {
-		return partialSignatureMessages, err
+		return partialSignatureMessages, withValidationStage(SSVValidationStagePartial, err)
 	}
 
 	if err := ctx.Err(); err != nil {
-		return partialSignatureMessages, err
+		return partialSignatureMessages, withValidationStage(SSVValidationStageContext, err)
 	}
 
 	signature := signedSSVMessage.Signatures[0]
 	signer := signedSSVMessage.OperatorIDs[0]
 	if err := mv.signatureVerifier.VerifySignature(signer, ssvMessage, signature); err != nil {
 		e := ErrSignatureVerification
 		e.innerErr = fmt.Errorf("verify opid: %v signature: %w", signer, err)
-		return partialSignatureMessages, e
+		return partialSignatureMessages, withValidationStage(SSVValidationStageSignatureVerify, e)
 	}
 
 	if err := mv.updatePartialSignatureState(partialSignatureMessages, receivedFrom, state, signer, committeeInfo); err != nil {
-		return partialSignatureMessages, err
+		return partialSignatureMessages, withValidationStage(SSVValidationStageStateUpdate, err)
 	}
 
 	return partialSignatureMessages, nil

message/validation/stage.go

@@ -0,0 +1,34 @@
+package validation
+
+import "errors"
+
+type validationStageError struct {
+	stage string
+	err   error
+}
+
+func (e validationStageError) Error() string {
+	return e.err.Error()
+}
+
+func (e validationStageError) Unwrap() error {
+	return e.err
+}
+
+func withValidationStage(stage string, err error) error {
+	if err == nil {
+		return nil
+	}
+	if validationStageFromError(err) != SSVValidationStageUnknown {
+		return err
+	}
+	return validationStageError{stage: stage, err: err}
+}
+
+func validationStageFromError(err error) string {
+	var staged validationStageError
+	if errors.As(err, &staged) {
+		return staged.stage
+	}
+	return SSVValidationStageUnknown
+}