add fee checker

witness-service/cmd/explorer/main.go

@@ -161,7 +161,7 @@ func (s *Service) Query(ctx context.Context, request *services.ExplorerQueryRequ
 	case services.Status_CONFIRMING, services.Status_CREATED:
 		queryOpts = append(queryOpts, relayer.StatusQueryOption(relayer.WaitingForWitnesses))
 	case services.Status_FAILED:
-		queryOpts = append(queryOpts, relayer.StatusQueryOption(relayer.ValidationFailed, relayer.ValidationRejected))
+		queryOpts = append(queryOpts, relayer.StatusQueryOption(relayer.ValidationFailed, relayer.ValidationRejected, relayer.InsufficientFeeRejected))
 	}
 	count, err := s.recorder.Count(queryOpts...)
 	if err != nil {
@@ -201,7 +201,7 @@ func (s *Service) convertStatus(status relayer.ValidationStatusType) services.St
 		return services.Status_SUBMITTED
 	case relayer.TransferSettled, relayer.BonusPending:
 		return services.Status_SETTLED
-	case relayer.ValidationFailed, relayer.ValidationRejected:
+	case relayer.ValidationFailed, relayer.ValidationRejected, relayer.InsufficientFeeRejected:
 		return services.Status_FAILED
 	}
 

witness-service/cmd/relayer/main.go

@@ -33,12 +33,17 @@ import (
 )
 
 type (
+	// CashierConfig defines the configuration of a cashier
+	CashierConfig struct {
+		Address string                         `json:"address" yaml:"address"`
+		Tokens  map[string]map[string]*big.Int `json:"tokens" yaml:"tokens"`
+	}
 	// ValidatorConfig defines the configuration of a validator
 	ValidatorConfig struct {
-		Address     string   `json:"address" yaml:"address"`
-		Cashiers    []string `json:"cashiers" yaml:"cashiers"`
-		WithPayload bool     `json:"withPayload" yaml:"withPayload"`
-		FromSolana  bool     `json:"fromSolana" yaml:"fromSolana"`
+		Address     string          `json:"address" yaml:"address"`
+		Cashiers    []CashierConfig `json:"cashiers" yaml:"cashiers"`
+		WithPayload bool            `json:"withPayload" yaml:"withPayload"`
+		FromSolana  bool            `json:"fromSolana" yaml:"fromSolana"`
 	}
 	// Configuration defines the configuration of the witness service
 	Configuration struct {
@@ -173,6 +178,7 @@ func main() {
 			}
 		}
 		validators := map[string]relayer.TransferValidator{}
+		checkers := map[string]*relayer.FeeChecker{}
 		for _, vc := range cfg.Validators {
 			validatorAddr, err := util.ParseEthAddress(vc.Address)
 			if err != nil {
@@ -202,7 +208,22 @@ func main() {
 				log.Fatalf("failed to create validator: %+v\n", err)
 			}
 			for _, cashier := range vc.Cashiers {
-				validators[cashier] = validator
+				validators[cashier.Address] = validator
+				checker := relayer.NewFeeChecker()
+				for token, recipients := range cashier.Tokens {
+					tokenAddr, err := util.ParseEthAddress(token)
+					if err != nil {
+						log.Fatalf("failed to parse token address %s: %+v", token, err)
+					}
+					for recipient, fee := range recipients {
+						addr, err := util.ParseEthAddress(recipient)
+						if err != nil {
+							log.Fatalf("failed to parse recipient address %s: %+v", recipient, err)
+						}
+						checker.SetFee(tokenAddr.String(), addr.String(), fee)
+					}
+					checkers[cashier.Address] = checker
+				}
 			}
 		}
 		bonusSender, err := relayer.NewBonusSender(ethClient, privateKeys, cfg.BonusTokens, cfg.Bonus)
@@ -213,6 +234,7 @@ func main() {
 		ethService, err := relayer.NewServiceOnEthereum(
 			validators,
 			unwrappers,
+			checkers,
 			bonusSender,
 			relayer.NewRecorder(
 				storeFactory.NewStore(cfg.Database),

witness-service/relayer/feechecker.go

@@ -0,0 +1,46 @@
+package relayer
+
+import (
+	"math/big"
+
+	"github.com/pkg/errors"
+)
+
+// ErrInsufficientFee is the error of insufficient fee
+var ErrInsufficientFee = errors.New("insufficient fee")
+
+// FeeChecker checks the fee of a transfer
+type FeeChecker struct {
+	tokenFees map[string]map[string]*big.Int
+}
+
+// NewFeeChecker creates a new fee checker
+func NewFeeChecker() *FeeChecker {
+	return &FeeChecker{
+		tokenFees: make(map[string]map[string]*big.Int),
+	}
+}
+
+// SetFee sets the fee of a transfer
+func (fc *FeeChecker) SetFee(token, recipient string, fee *big.Int) {
+	if _, exists := fc.tokenFees[token]; !exists {
+		fc.tokenFees[token] = make(map[string]*big.Int)
+	}
+	fc.tokenFees[token][recipient] = fee
+}
+
+// Check checks the fee of a transfer
+func (fc *FeeChecker) Check(transfer *Transfer) error {
+	requiredFees, exists := fc.tokenFees[transfer.token.String()]
+	if !exists {
+		return nil
+	}
+	requiredFee, exists := requiredFees[transfer.recipient.String()]
+	if !exists {
+		return nil
+	}
+	if transfer.fee.Cmp(requiredFee) < 0 {
+		return errors.Wrapf(ErrInsufficientFee, "fee %d is lower than required %d", transfer.fee, requiredFee)
+	}
+	return nil
+}

witness-service/relayer/recorder.go

@@ -783,6 +783,28 @@ func (recorder *Recorder) UpdateRecord(id common.Hash, txhash common.Hash, relay
 	return recorder.validateResult(result, 1)
 }
 
+// MarkAsInsufficientFee marks a transfer as insufficient fee
+func (recorder *Recorder) MarkAsInsufficientFee(id common.Hash) error {
+	log.Printf("mark transfer %s as insufficient fee\n", id.Hex())
+	recorder.mutex.Lock()
+	defer recorder.mutex.Unlock()
+	result, err := recorder.store.DB().Exec(recorder.updateStatusQuery, InsufficientFeeRejected, id.Hex(), ValidationInProcess)
+	if err != nil {
+		return errors.Wrap(err, "failed to mark as insufficient fee")
+	}
+	if recorder.explorerStore != nil {
+		for {
+			_, err := recorder.explorerStore.DB().Exec(recorder.updateStatusQueryForExplorer, InsufficientFeeRejected, id.Hex())
+			if err == nil {
+				break
+			}
+			log.Println("failed to update explorer db", err)
+		}
+	}
+
+	return recorder.validateResult(result, 1)
+}
+
 // MarkAsValidated marks a transfer as validated
 func (recorder *Recorder) MarkAsValidated(id common.Hash, txhash common.Hash, relayer common.Address, nonce uint64, gasPrice *big.Int) error {
 	log.Printf("mark transfer %s as validated (%s, %s, %d)\n", id.Hex(), txhash.Hex(), relayer.Hex(), nonce)
@@ -991,11 +1013,12 @@ func (recorder *Recorder) NewTXs(count uint32) ([]uint64, [][]byte, error) {
 	recorder.mutex.Lock()
 	defer recorder.mutex.Unlock()
 	rows, err := recorder.store.DB().Query(fmt.Sprintf(
-		"SELECT `txHash`, `blockheight` FROM %s WHERE TIMESTAMPDIFF(HOUR, creationTime, NOW()) <= 48 AND `txHash` NOT IN (SELECT `sourceTxHash` FROM %s WHERE (`status`=?  OR `status`=? OR `status`=?) AND `sourceTxHash` IS NOT NULL) LIMIT ?",
+		"SELECT `txHash`, `blockheight` FROM %s WHERE TIMESTAMPDIFF(HOUR, creationTime, NOW()) <= 48 AND `txHash` NOT IN (SELECT `sourceTxHash` FROM %s WHERE `status` in (?, ?, ?, ?) AND `sourceTxHash` IS NOT NULL) LIMIT ?",
 		recorder.newTXTableName, recorder.transferTableName),
 		TransferSettled,
 		ValidationFailed,
 		ValidationRejected,
+		InsufficientFeeRejected,
 		count)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "failed to query new txs")

witness-service/relayer/service.go

@@ -34,6 +34,7 @@ type (
 		services.UnimplementedRelayServiceServer
 		validators      map[string]TransferValidator
 		unwrappers      map[string]map[string]common.Address
+		checkers        map[string]*FeeChecker
 		bonusSender     BonusSender
 		processor       dispatcher.Runner
 		recorder        *Recorder
@@ -59,6 +60,7 @@ const (
 func NewServiceOnEthereum(
 	validators map[string]TransferValidator,
 	unwrappers map[string]map[string]common.Address,
+	checkers map[string]*FeeChecker,
 	bonusSender BonusSender,
 	recorder *Recorder,
 	interval time.Duration,
@@ -70,6 +72,7 @@ func NewServiceOnEthereum(
 	s := &Service{
 		validators:      validators,
 		unwrappers:      unwrappers,
+		checkers:        checkers,
 		bonusSender:     bonusSender,
 		recorder:        recorder,
 		cache:           cache,
@@ -287,7 +290,7 @@ func (s *Service) List(ctx context.Context, request *services.ListRequest) (*ser
 	case services.Status_CREATED, services.Status_CONFIRMING:
 		queryOpts = append(queryOpts, StatusQueryOption(WaitingForWitnesses))
 	case services.Status_FAILED:
-		queryOpts = append(queryOpts, StatusQueryOption(ValidationFailed, ValidationRejected))
+		queryOpts = append(queryOpts, StatusQueryOption(ValidationFailed, ValidationRejected, InsufficientFeeRejected))
 	}
 	count, err := s.recorder.Count(queryOpts...)
 	if err != nil {
@@ -369,7 +372,7 @@ func (s *Service) convertStatus(status ValidationStatusType) services.Status {
 		return services.Status_SUBMITTED
 	case TransferSettled, BonusPending:
 		return services.Status_SETTLED
-	case ValidationFailed, ValidationRejected:
+	case ValidationFailed, ValidationRejected, InsufficientFeeRejected:
 		return services.Status_FAILED
 	}
 
@@ -477,7 +480,7 @@ func (s *Service) confirmTransfers() error {
 			return errors.Wrap(err, "failed to read transfers to confirm")
 		}
 		for _, transfer := range validatedTransfers {
-			speedup, merged, err := s.confirmTransfer(transfer, validator)
+			speedup, merged, err := s.confirmTransfer(transfer, validator, s.checkers[cashier])
 			switch {
 			case err != nil:
 				log.Printf("failed to confirm transfer %s, %+v\n", transfer.id.String(), err)
@@ -506,7 +509,13 @@ func (s *Service) confirmTransfers() error {
 	return nil
 }
 
-func (s *Service) confirmTransfer(transfer *Transfer, validator TransferValidator) (bool, bool, error) {
+func (s *Service) confirmTransfer(transfer *Transfer, validator TransferValidator, checker *FeeChecker) (bool, bool, error) {
+	if checker != nil {
+		if err := checker.Check(transfer); err != nil {
+			log.Printf("failed to check fee of transfer %s, %v\n", transfer.id.String(), err)
+			return false, true, s.recorder.MarkAsInsufficientFee(transfer.id)
+		}
+	}
 	statusOnChain, err := validator.Check(transfer)
 	switch errors.Cause(err) {
 	case nil:

witness-service/relayer/solanaservice.go

@@ -191,7 +191,7 @@ func (s *SolanaService) List(ctx context.Context, request *services.ListRequest)
 	case services.Status_CREATED, services.Status_CONFIRMING:
 		queryOpts = append(queryOpts, StatusQueryOption(WaitingForWitnesses))
 	case services.Status_FAILED:
-		queryOpts = append(queryOpts, StatusQueryOption(ValidationFailed, ValidationRejected))
+		queryOpts = append(queryOpts, StatusQueryOption(ValidationFailed, ValidationRejected, InsufficientFeeRejected))
 	}
 	count, err := s.abstractRecorder.Count(queryOpts...)
 	if err != nil {
@@ -273,7 +273,7 @@ func (s *SolanaService) convertStatus(status ValidationStatusType) services.Stat
 		return services.Status_SUBMITTED
 	case TransferSettled, BonusPending:
 		return services.Status_SETTLED
-	case ValidationFailed, ValidationRejected:
+	case ValidationFailed, ValidationRejected, InsufficientFeeRejected:
 		return services.Status_FAILED
 	}
 

witness-service/relayer/types.go

@@ -132,6 +132,8 @@ const (
 	WaitingForWitnesses ValidationStatusType = "new"
 	// ValidationInProcess stands for a transfer in process
 	ValidationInProcess = "processing"
+	// InsufficientFeeRejected stands for a transfer with insufficient fee
+	InsufficientFeeRejected = "insufficient"
 	// ValidationSubmitted stands for a transfer with validation submitted
 	ValidationSubmitted = "validated"
 	// BonusPending stands for a transfer with pending bonus