txHandler: add more metric

data/transactions/verify/txn.go

@@ -69,6 +69,52 @@ type GroupContext struct {
 	ledger           logic.LedgerForSignature
 }
 
+var errTxGroupInvalidFee = errors.New("txgroup fee requirement overflow")
+var errTxnSigHasNoSig = errors.New("signedtxn has no sig")
+var errTxnSigNotWellFormed = errors.New("signedtxn should only have one of Sig or Msig or LogicSig")
+var errRekeyingNotSupported = errors.New("nonempty AuthAddr but rekeying is not supported")
+var errUnknownSignature = errors.New("has one mystery sig. WAT?")
+
+// TxGroupErrorReason is reason code for ErrTxGroupError
+type TxGroupErrorReason int
+
+const (
+	// TxGroupErrorReasonGeneric is a generic (not tracked) reason code
+	TxGroupErrorReasonGeneric TxGroupErrorReason = iota
+	// TxGroupErrorReasonNotWellFormed is txn.WellFormed failure
+	TxGroupErrorReasonNotWellFormed
+	// TxGroupErrorReasonInvalidFee is invalid fee pooling in transaction group
+	TxGroupErrorReasonInvalidFee
+	// TxGroupErrorReasonHasNoSig is for transaction without any signature
+	TxGroupErrorReasonHasNoSig
+	// TxGroupErrorReasonSigNotWellFormed defines signature format errors
+	TxGroupErrorReasonSigNotWellFormed
+	// TxGroupErrorReasonMsigNotWellFormed defines multisig format errors
+	TxGroupErrorReasonMsigNotWellFormed
+	// TxGroupErrorReasonLogicSigFailed defines logic sig validation errors
+	TxGroupErrorReasonLogicSigFailed
+
+	// TxGroupErrorReasonNumValues is number of enum values
+	TxGroupErrorReasonNumValues
+)
+
+// ErrTxGroupError is an error from txn pre-validation (well form-ness, signature format, etc).
+// It can be unwrapped into underlying error, as well as has a specific failure reason code.
+type ErrTxGroupError struct {
+	err    error
+	Reason TxGroupErrorReason
+}
+
+// Error returns an error message from the underlying error
+func (e *ErrTxGroupError) Error() string {
+	return e.err.Error()
+}
+
+// Unwrap returns an underlying error
+func (e *ErrTxGroupError) Unwrap() error {
+	return e.err
+}
+
 // PrepareGroupContext prepares a verification group parameter object for a given transaction
 // group.
 func PrepareGroupContext(group []transactions.SignedTxn, contextHdr bookkeeping.BlockHeader, ledger logic.LedgerForSignature) (*GroupContext, error) {
@@ -101,14 +147,14 @@ func (g *GroupContext) Equal(other *GroupContext) bool {
 
 // txnBatchPrep verifies a SignedTxn having no obviously inconsistent data.
 // Block-assembly time checks of LogicSig and accounting rules may still block the txn.
-// it is the caller responsibility to call batchVerifier.Verify()
-func txnBatchPrep(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContext, verifier *crypto.BatchVerifier) error {
+// It is the caller responsibility to call batchVerifier.Verify().
+func txnBatchPrep(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContext, verifier *crypto.BatchVerifier) *ErrTxGroupError {
 	if !groupCtx.consensusParams.SupportRekeying && (s.AuthAddr != basics.Address{}) {
-		return errors.New("nonempty AuthAddr but rekeying is not supported")
+		return &ErrTxGroupError{err: errRekeyingNotSupported, Reason: TxGroupErrorReasonGeneric}
 	}
 
 	if err := s.Txn.WellFormed(groupCtx.specAddrs, groupCtx.consensusParams); err != nil {
-		return err
+		return &ErrTxGroupError{err: err, Reason: TxGroupErrorReasonNotWellFormed}
 	}
 
 	return stxnCoreChecks(s, txnIdx, groupCtx, verifier)
@@ -144,10 +190,14 @@ func txnGroupBatchPrep(stxs []transactions.SignedTxn, contextHdr bookkeeping.Blo
 	minFeeCount := uint64(0)
 	feesPaid := uint64(0)
 	for i, stxn := range stxs {
-		err = txnBatchPrep(&stxn, i, groupCtx, verifier)
-		if err != nil {
-			err = fmt.Errorf("transaction %+v invalid : %w", stxn, err)
-			return
+		prepErr := txnBatchPrep(&stxn, i, groupCtx, verifier)
+		if prepErr != nil {
+			// re-wrap the error, take underlying one and copy the reason code
+			err = &ErrTxGroupError{
+				err:    fmt.Errorf("transaction %+v invalid : %w", stxn, prepErr.err),
+				Reason: prepErr.Reason,
+			}
+			return nil, err
 		}
 		if stxn.Txn.Type != protocol.StateProofTx {
 			minFeeCount++
@@ -156,22 +206,27 @@ func txnGroupBatchPrep(stxs []transactions.SignedTxn, contextHdr bookkeeping.Blo
 	}
 	feeNeeded, overflow := basics.OMul(groupCtx.consensusParams.MinTxnFee, minFeeCount)
 	if overflow {
-		err = fmt.Errorf("txgroup fee requirement overflow")
-		return
+		err = &ErrTxGroupError{err: errTxGroupInvalidFee, Reason: TxGroupErrorReasonInvalidFee}
+		return nil, err
 	}
 	// feesPaid may have saturated. That's ok. Since we know
 	// feeNeeded did not overflow, simple comparison tells us
 	// feesPaid was enough.
 	if feesPaid < feeNeeded {
-		err = fmt.Errorf("txgroup had %d in fees, which is less than the minimum %d * %d",
-			feesPaid, minFeeCount, groupCtx.consensusParams.MinTxnFee)
-		return
+		err = &ErrTxGroupError{
+			err: fmt.Errorf(
+				"txgroup had %d in fees, which is less than the minimum %d * %d",
+				feesPaid, minFeeCount, groupCtx.consensusParams.MinTxnFee),
+			Reason: TxGroupErrorReasonInvalidFee,
+		}
+		return nil, err
 	}
 
-	return
+	return groupCtx, nil
 }
 
-func stxnCoreChecks(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContext, batchVerifier *crypto.BatchVerifier) error {
+// stxnCoreChecks runs signatures validity checks and enqueues signature into batchVerifier for verification.
+func stxnCoreChecks(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContext, batchVerifier *crypto.BatchVerifier) *ErrTxGroupError {
 	numSigs := 0
 	hasSig := false
 	hasMsig := false
@@ -196,11 +251,10 @@ func stxnCoreChecks(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContex
 		if s.Txn.Sender == transactions.StateProofSender && s.Txn.Type == protocol.StateProofTx {
 			return nil
 		}
-
-		return errors.New("signedtxn has no sig")
+		return &ErrTxGroupError{err: errTxnSigHasNoSig, Reason: TxGroupErrorReasonHasNoSig}
 	}
 	if numSigs > 1 {
-		return errors.New("signedtxn should only have one of Sig or Msig or LogicSig")
+		return &ErrTxGroupError{err: errTxnSigNotWellFormed, Reason: TxGroupErrorReasonSigNotWellFormed}
 	}
 
 	if hasSig {
@@ -209,14 +263,17 @@ func stxnCoreChecks(s *transactions.SignedTxn, txnIdx int, groupCtx *GroupContex
 	}
 	if hasMsig {
 		if err := crypto.MultisigBatchPrep(s.Txn, crypto.Digest(s.Authorizer()), s.Msig, batchVerifier); err != nil {
-			return fmt.Errorf("multisig validation failed: %w", err)
+			return &ErrTxGroupError{err: fmt.Errorf("multisig validation failed: %w", err), Reason: TxGroupErrorReasonMsigNotWellFormed}
 		}
 		return nil
 	}
 	if hasLogicSig {
-		return logicSigVerify(s, txnIdx, groupCtx)
+		if err := logicSigVerify(s, txnIdx, groupCtx); err != nil {
+			return &ErrTxGroupError{err: err, Reason: TxGroupErrorReasonLogicSigFailed}
+		}
+		return nil
 	}
-	return errors.New("has one mystery sig. WAT?")
+	return &ErrTxGroupError{err: errUnknownSignature, Reason: TxGroupErrorReasonGeneric}
 }
 
 // LogicSigSanityCheck checks that the signature is valid and that the program is basically well formed.
@@ -254,7 +311,7 @@ func logicSigSanityCheckBatchPrep(txn *transactions.SignedTxn, groupIndex int, g
 	}
 
 	if groupIndex < 0 {
-		return errors.New("Negative groupIndex")
+		return errors.New("negative groupIndex")
 	}
 	txngroup := transactions.WrapSignedTxnsWithAD(groupCtx.signedGroupTxns)
 	ep := logic.EvalParams{
@@ -310,7 +367,7 @@ func logicSigVerify(txn *transactions.SignedTxn, groupIndex int, groupCtx *Group
 	}
 
 	if groupIndex < 0 {
-		return errors.New("Negative groupIndex")
+		return errors.New("negative groupIndex")
 	}
 	ep := logic.EvalParams{
 		Proto:         &groupCtx.consensusParams,

data/txHandler.go

@@ -19,9 +19,11 @@ package data
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"sync"
+	"time"
 
 	"github.com/algorand/go-algorand/config"
 	"github.com/algorand/go-algorand/crypto"
@@ -45,6 +47,14 @@ var txBacklogSize = config.Consensus[protocol.ConsensusCurrentVersion].MaxTxnByt
 var transactionMessagesHandled = metrics.MakeCounter(metrics.TransactionMessagesHandled)
 var transactionMessagesDroppedFromBacklog = metrics.MakeCounter(metrics.TransactionMessagesDroppedFromBacklog)
 var transactionMessagesDroppedFromPool = metrics.MakeCounter(metrics.TransactionMessagesDroppedFromPool)
+var transactionMessagesAlreadyCommitted = metrics.MakeCounter(metrics.TransactionMessagesAlreadyCommitted)
+var transactionMessagesTxGroupInvalidFee = metrics.MakeCounter(metrics.TransactionMessagesTxGroupInvalidFee)
+var transactionMessagesTxnNotWellFormed = metrics.MakeCounter(metrics.TransactionMessagesTxnNotWellFormed)
+var transactionMessagesTxnSigNotWellFormed = metrics.MakeCounter(metrics.TransactionMessagesTxnSigNotWellFormed)
+var transactionMessagesTxnMsigNotWellFormed = metrics.MakeCounter(metrics.TransactionMessagesTxnMsigNotWellFormed)
+var transactionMessagesTxnLogicSig = metrics.MakeCounter(metrics.TransactionMessagesTxnLogicSig)
+var transactionMessagesTxnSigVerificationFailed = metrics.MakeCounter(metrics.TransactionMessagesTxnSigVerificationFailed)
+var transactionMessagesBacklogSizeGauge = metrics.MakeGauge(metrics.TransactionMessagesBacklogSize)
 
 // The txBacklogMsg structure used to track a single incoming transaction from the gossip network,
 type txBacklogMsg struct {
@@ -101,8 +111,9 @@ func (handler *TxHandler) Start() {
 	handler.net.RegisterHandlers([]network.TaggedMessageHandler{
 		{Tag: protocol.TxnTag, MessageHandler: network.HandlerFunc(handler.processIncomingTxn)},
 	})
-	handler.backlogWg.Add(1)
+	handler.backlogWg.Add(2)
 	go handler.backlogWorker()
+	go handler.backlogGaugeThread()
 }
 
 // Stop suspends the processing of incoming messages at the transaction handler
@@ -113,12 +124,26 @@ func (handler *TxHandler) Stop() {
 
 func reencode(stxns []transactions.SignedTxn) []byte {
 	var result [][]byte
-	for _, stxn := range stxns {
-		result = append(result, protocol.Encode(&stxn))
+	for i := range stxns {
+		result = append(result, protocol.Encode(&stxns[i]))
 	}
 	return bytes.Join(result, nil)
 }
 
+func (handler *TxHandler) backlogGaugeThread() {
+	defer handler.backlogWg.Done()
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			transactionMessagesBacklogSizeGauge.Set(float64(len(handler.backlogQueue)))
+		case <-handler.ctx.Done():
+			return
+		}
+	}
+}
+
 // backlogWorker is the worker go routine that process the incoming messages from the postVerificationQueue and backlogQueue channels
 // and dispatches them further.
 func (handler *TxHandler) backlogWorker() {
@@ -132,7 +157,7 @@ func (handler *TxHandler) backlogWorker() {
 			if !ok {
 				return
 			}
-			handler.postprocessCheckedTxn(wi)
+			handler.postProcessCheckedTxn(wi)
 
 			// restart the loop so that we could empty out the post verification queue.
 			continue
@@ -146,6 +171,7 @@ func (handler *TxHandler) backlogWorker() {
 				return
 			}
 			if handler.checkAlreadyCommitted(wi) {
+				transactionMessagesAlreadyCommitted.Inc(nil)
 				continue
 			}
 
@@ -156,17 +182,45 @@ func (handler *TxHandler) backlogWorker() {
 			if !ok {
 				return
 			}
-			handler.postprocessCheckedTxn(wi)
+			handler.postProcessCheckedTxn(wi)
 
 		case <-handler.ctx.Done():
 			return
 		}
 	}
 }
 
-func (handler *TxHandler) postprocessCheckedTxn(wi *txBacklogMsg) {
+func (handler *TxHandler) postProcessReportErrors(err error) {
+	if errors.Is(err, crypto.ErrBatchVerificationFailed) {
+		transactionMessagesTxnSigVerificationFailed.Inc(nil)
+		return
+	}
+
+	var txGroupErr *verify.ErrTxGroupError
+	if errors.As(err, &txGroupErr) {
+		// txGroupErr = err.(*verify.ErrTxGroupError)
+		switch txGroupErr.Reason {
+		case verify.TxGroupErrorReasonNotWellFormed:
+			transactionMessagesTxnNotWellFormed.Inc(nil)
+		case verify.TxGroupErrorReasonInvalidFee:
+			transactionMessagesTxGroupInvalidFee.Inc(nil)
+		case verify.TxGroupErrorReasonHasNoSig:
+			fallthrough
+		case verify.TxGroupErrorReasonSigNotWellFormed:
+			transactionMessagesTxnSigNotWellFormed.Inc(nil)
+		case verify.TxGroupErrorReasonMsigNotWellFormed:
+			transactionMessagesTxnMsigNotWellFormed.Inc(nil)
+		case verify.TxGroupErrorReasonLogicSigFailed:
+			transactionMessagesTxnLogicSig.Inc(nil)
+		default:
+		}
+	}
+}
+
+func (handler *TxHandler) postProcessCheckedTxn(wi *txBacklogMsg) {
 	if wi.verificationErr != nil {
 		// disconnect from peer.
+		handler.postProcessReportErrors(wi.verificationErr)
 		logging.Base().Warnf("Received a malformed tx group %v: %v", wi.unverifiedTxGroup, wi.verificationErr)
 		handler.net.Disconnect(wi.rawmsg.Sender)
 		return
@@ -203,7 +257,7 @@ func (handler *TxHandler) asyncVerifySignature(arg interface{}) interface{} {
 	latest := handler.ledger.Latest()
 	latestHdr, err := handler.ledger.BlockHdr(latest)
 	if err != nil {
-		tx.verificationErr = fmt.Errorf("Could not get header for previous block %d: %w", latest, err)
+		tx.verificationErr = fmt.Errorf("could not get header for previous block %d: %w", latest, err)
 		logging.Base().Warnf("Could not get header for previous block %d: %v", latest, err)
 	} else {
 		// we can't use PaysetGroups here since it's using a execpool like this go-routine and we don't want to deadlock.