From e8aff03b330c8650c81a087c6fea6ae484d87b91 Mon Sep 17 00:00:00 2001 From: Garry Sharp <> Date: Fri, 15 Aug 2025 20:23:54 -0500 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=90=9B=20fee=20config=20bug=20fix=20a?= =?UTF-8?q?nd=20target=20added=20to=20tx=20fee=20validations?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugin/fees/config.go | 43 ++++++++---- plugin/fees/transaction.go | 137 +++++++++++++++++++------------------ 2 files changed, 99 insertions(+), 81 deletions(-) diff --git a/plugin/fees/config.go b/plugin/fees/config.go index d4a6218..f7914b0 100644 --- a/plugin/fees/config.go +++ b/plugin/fees/config.go @@ -11,14 +11,13 @@ import ( // These are properties and parameters specific to the fee plugin config. They should be distinct from system/core config type FeeConfig struct { - Type string `mapstructure:"type"` - Version string `mapstructure:"version"` - MaxFeeAmount uint64 `mapstructure:"max_fee_amount"` // Policies that are created/submitted which do not have this amount will be rejected. - UsdcAddress string `mapstructure:"usdc_address"` // The address of the USDC token on the Ethereum blockchain. - VerifierToken string `mapstructure:"verifier_token"` // The token to use for the verifier API. - chainId uint64 `mapstructure:"chain_id"` // The chain ID of the Ethereum blockchain. - ChainId *big.Int - EthProvider string `mapstructure:"eth_provider"` // The Ethereum provider to use for the fee plugin. + Type string `mapstructure:"type"` + Version string `mapstructure:"version"` + MaxFeeAmount uint64 `mapstructure:"max_fee_amount"` // Policies that are created/submitted which do not have this amount will be rejected. + UsdcAddress string `mapstructure:"usdc_address"` // The address of the USDC token on the Ethereum blockchain. + VerifierToken string `mapstructure:"verifier_token"` // The token to use for the verifier API. + ChainId *big.Int // The chain ID as a big.Int (initialized from ChainIdRaw). + EthProvider string `mapstructure:"eth_provider"` // The Ethereum provider to use for the fee plugin. Jobs struct { Load struct { MaxConcurrentJobs uint64 `mapstructure:"max_concurrent_jobs"` //How many consecutive tasks can take place @@ -34,6 +33,12 @@ type FeeConfig struct { MaxConcurrentJobs uint64 `mapstructure:"max_concurrent_jobs"` } `mapstructure:"post"` } + DryRun bool `mapstructure:"dry_run"` +} + +type FeeConfigFileWrapper struct { + FeeConfig `mapstructure:",squash"` + ChainIdRaw uint64 `mapstructure:"chain_id"` } type ConfigOption func(*FeeConfig) error @@ -52,6 +57,7 @@ func withDefaults(c *FeeConfig) { c.Jobs.Load.Cronexpr = "@every 2m" c.Jobs.Transact.Cronexpr = "0 12 * * 5" c.Jobs.Post.Cronexpr = "@every 5m" + c.DryRun = false } func WithMaxFeeAmount(maxFeeAmount uint64) ConfigOption { @@ -100,9 +106,16 @@ func WithCronexpr(load, transact, post string) ConfigOption { } } -func WithFileConfig(basePath string) ConfigOption { +func WithDryRun(dryRun bool) ConfigOption { return func(c *FeeConfig) error { + c.DryRun = dryRun + return nil + } +} + +func WithFileConfig(basePath string) ConfigOption { + return func(c *FeeConfig) error { v := viper.New() v.SetConfigName("fee") @@ -122,11 +135,15 @@ func WithFileConfig(basePath string) ConfigOption { return fmt.Errorf("failed to read config: %w", err) } - if err := v.Unmarshal(c); err != nil { + var wrappedConfig FeeConfigFileWrapper + if err := v.Unmarshal(&wrappedConfig); err != nil { return fmt.Errorf("failed to unmarshal config: %w", err) } - c.ChainId = big.NewInt(0).SetUint64(c.chainId) + // Copy all values from the wrapped config to the original config + *c = wrappedConfig.FeeConfig + + c.ChainId = big.NewInt(0).SetUint64(wrappedConfig.ChainIdRaw) return nil } } @@ -149,8 +166,8 @@ func NewFeeConfig(fns ...ConfigOption) (*FeeConfig, error) { return c, errors.New("verifier_token is required") } - if c.ChainId == nil { - return c, errors.New("chain_id is required") + if c.ChainId == nil || c.ChainId.Uint64() == 0 { + return c, errors.New("chain_id is required and must not be 0") } if c.EthProvider == "" { diff --git a/plugin/fees/transaction.go b/plugin/fees/transaction.go index e3fb291..cad4ba8 100644 --- a/plugin/fees/transaction.go +++ b/plugin/fees/transaction.go @@ -61,88 +61,89 @@ func (fp *FeePlugin) proposeTransactions(ctx context.Context, policy vtypes.Plug txs := []vtypes.PluginKeysignRequest{} var magicConstantRecipientValue rtypes.MagicConstant = rtypes.MagicConstant_UNSPECIFIED - var token string + // This should only return one rule, but in case there are more/fewer rules, we'll loop through them all and error if it's the case. - for _, rule := range recipe.Rules { - - // This section of code goes through the rules in the fee policy. It looks for the recipient of the fee collection policy and extracts it. If other data is found throws an error as they're unsupported rules. - var recipient string // The address specified in the fee policy. - switch rule.Resource { - case "ethereum.erc20.transfer": - for _, constraint := range rule.ParameterConstraints { - if constraint.ParameterName == "recipient" { - if constraint.Constraint.Type != rtypes.ConstraintType_CONSTRAINT_TYPE_MAGIC_CONSTANT { - return nil, fmt.Errorf("recipient constraint is not a magic constant") - } - iv, err := strconv.ParseInt(constraint.Constraint.GetFixedValue(), 10, 64) - if err != nil { - return nil, fmt.Errorf("failed to parse fixed value: %v", err) - } - magicConstantRecipientValue = rtypes.MagicConstant(iv) - } + if len(recipe.Rules) != 1 { + return nil, fmt.Errorf("expected 1 rule, got %d", len(recipe.Rules)) + } + rule := recipe.Rules[0] + + resourceName := "ethereum.erc20.transfer" + if rule.Resource != resourceName { + return nil, fmt.Errorf("rule resource expected to be %s", resourceName) + } + + for _, constraint := range rule.ParameterConstraints { + if constraint.ParameterName == "recipient" { + if constraint.Constraint.Type != rtypes.ConstraintType_CONSTRAINT_TYPE_MAGIC_CONSTANT { + return nil, fmt.Errorf("recipient constraint is not a magic constant") + } + iv, err := strconv.ParseInt(constraint.Constraint.GetFixedValue(), 10, 64) + if err != nil { + return nil, fmt.Errorf("failed to parse fixed value: %v", err) } - default: - return nil, fmt.Errorf("unsupported rule: %v", rule.Id) + magicConstantRecipientValue = rtypes.MagicConstant(iv) } + } - if magicConstantRecipientValue != rtypes.MagicConstant_VULTISIG_TREASURY { - return nil, fmt.Errorf("recipient constraint is not a treasury magic constant") - } + if magicConstantRecipientValue != rtypes.MagicConstant_VULTISIG_TREASURY { + return nil, fmt.Errorf("recipient constraint is not a treasury magic constant") + } - treasuryResolver := resolver.NewDefaultTreasuryResolver() - recipient, _, err := treasuryResolver.Resolve(magicConstantRecipientValue, "ethereum", "usdc") - if err != nil { - return nil, fmt.Errorf("failed to resolve treasury address: %v", err) - } + treasuryResolver := resolver.NewDefaultTreasuryResolver() + recipient, _, err := treasuryResolver.Resolve(magicConstantRecipientValue, "ethereum", "usdc") + if err != nil { + return nil, fmt.Errorf("failed to resolve treasury address: %v", err) + } - if gcommon.HexToAddress(token) != gcommon.HexToAddress(usdc.Address) { - return nil, fmt.Errorf("token address does not match usdc address") - } + token := rule.Target.GetAddress() - amount := run.TotalAmount + if gcommon.HexToAddress(token) != gcommon.HexToAddress(usdc.Address) { + return nil, fmt.Errorf("token address does not match usdc address") + } - tx, err := fp.eth.MakeAnyTransfer(ctx, - gcommon.HexToAddress(ethAddress), - gcommon.HexToAddress(recipient), - gcommon.HexToAddress(usdc.Address), - big.NewInt(int64(amount))) - if err != nil { - return nil, fmt.Errorf("failed to generate unsigned transaction: %w", err) - } + amount := run.TotalAmount - txHex := hexutil.Encode(tx) + tx, err := fp.eth.MakeAnyTransfer(ctx, + gcommon.HexToAddress(ethAddress), + gcommon.HexToAddress(recipient), + gcommon.HexToAddress(usdc.Address), + big.NewInt(int64(amount))) + if err != nil { + return nil, fmt.Errorf("failed to generate unsigned transaction: %w", err) + } - txData, e := reth.DecodeUnsignedPayload(tx) - if e != nil { - return nil, fmt.Errorf("ethereum.DecodeUnsignedPayload: %w", e) - } - txHashToSign := etypes.LatestSignerForChainID(fp.config.ChainId).Hash(etypes.NewTx(txData)) - - msgHash := sha256.Sum256(txHashToSign.Bytes()) - - signRequest := vtypes.PluginKeysignRequest{ - KeysignRequest: vtypes.KeysignRequest{ - PublicKey: policy.PublicKey, - Messages: []vtypes.KeysignMessage{ - { - Message: base64.StdEncoding.EncodeToString(txHashToSign.Bytes()), - RawMessage: txHex, - Chain: vgcommon.Ethereum, - Hash: base64.StdEncoding.EncodeToString(msgHash[:]), - HashFunction: vtypes.HashFunction_SHA256, - }, + txHex := hexutil.Encode(tx) + + txData, e := reth.DecodeUnsignedPayload(tx) + if e != nil { + return nil, fmt.Errorf("ethereum.DecodeUnsignedPayload: %w", e) + } + + txHashToSign := etypes.LatestSignerForChainID(fp.config.ChainId).Hash(etypes.NewTx(txData)) + msgHash := sha256.Sum256(txHashToSign.Bytes()) + signRequest := vtypes.PluginKeysignRequest{ + KeysignRequest: vtypes.KeysignRequest{ + PublicKey: policy.PublicKey, + Messages: []vtypes.KeysignMessage{ + { + Message: base64.StdEncoding.EncodeToString(txHashToSign.Bytes()), + RawMessage: txHex, + Chain: vgcommon.Ethereum, + Hash: base64.StdEncoding.EncodeToString(msgHash[:]), + HashFunction: vtypes.HashFunction_SHA256, }, - SessionID: "", - HexEncryptionKey: "", - PolicyID: policy.ID, - PluginID: policy.PluginID.String(), }, - Transaction: base64.StdEncoding.EncodeToString(tx), - } - - txs = append(txs, signRequest) + SessionID: "", + HexEncryptionKey: "", + PolicyID: policy.ID, + PluginID: policy.PluginID.String(), + }, + Transaction: txHex, } + txs = append(txs, signRequest) + return txs, nil } From e7ef35fd337965ce2c3a568d9da53aa20029237f Mon Sep 17 00:00:00 2001 From: Garry Sharp <> Date: Sat, 16 Aug 2025 18:22:32 -0500 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=9A=9A=20Move=20some=20fee=20code=20a?= =?UTF-8?q?round=20into=20more=20sensible=20places?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugin/fees/fees.go | 244 +++--------------------------- plugin/fees/helper.go | 14 ++ plugin/fees/load.go | 139 +++++++++++++++++ plugin/fees/transaction.go | 300 +++++++++++++++++++++++-------------- 4 files changed, 357 insertions(+), 340 deletions(-) create mode 100644 plugin/fees/load.go diff --git a/plugin/fees/fees.go b/plugin/fees/fees.go index 7373863..87362b0 100644 --- a/plugin/fees/fees.go +++ b/plugin/fees/fees.go @@ -1,14 +1,14 @@ package fees import ( - "context" + "encoding/base64" "fmt" "sync" "github.com/ethereum/go-ethereum/ethclient" - "github.com/google/uuid" "github.com/hibiken/asynq" "github.com/sirupsen/logrus" + "github.com/vultisig/recipes/engine" "github.com/vultisig/recipes/sdk/evm" rtypes "github.com/vultisig/recipes/types" "github.com/vultisig/verifier/plugin" @@ -17,10 +17,7 @@ import ( vtypes "github.com/vultisig/verifier/types" "github.com/vultisig/verifier/vault" vgcommon "github.com/vultisig/vultisig-go/common" - "golang.org/x/sync/errgroup" - "golang.org/x/sync/semaphore" - "github.com/vultisig/plugin/internal/types" "github.com/vultisig/plugin/internal/verifierapi" "github.com/vultisig/plugin/storage" ) @@ -114,238 +111,35 @@ func NewFeePlugin(db storage.DatabaseStorage, }, nil } -/* ------------------------------------------------------------------------------------------------ -LOADING FEES -here we pull in a list of fees (amounts and ids) that are pending collection and add them to a fee run ------------------------------------------------------------------------------------------------- */ - -func (fp *FeePlugin) LoadFees(ctx context.Context, task *asynq.Task) error { - fp.transactingMutex.Lock() - defer fp.transactingMutex.Unlock() - - fp.logger.Info("Starting Fee Loading Job") - - feePolicies, err := fp.db.GetAllFeePolicies(ctx) +func (fp *FeePlugin) ValidateProposedTransactions(policy vtypes.PluginPolicy, txs []vtypes.PluginKeysignRequest) error { + // First validate the plugin policy itself + err := fp.ValidatePluginPolicy(policy) if err != nil { - return fmt.Errorf("failed to get plugin policy: %w", err) - } - - // We limit the number of concurrent fee loading operations to 10 - sem := semaphore.NewWeighted(int64(fp.config.Jobs.Load.MaxConcurrentJobs)) - var wg sync.WaitGroup - var eg errgroup.Group - - for _, feePolicy := range feePolicies { - wg.Add(1) - feePolicy = feePolicy - eg.Go(func() error { - defer wg.Done() - if err := sem.Acquire(ctx, 1); err != nil { - return fmt.Errorf("failed to acquire semaphore: %w", err) - } - defer sem.Release(1) - return fp.executeFeeLoading(ctx, feePolicy) - }) - } - - wg.Wait() - if err := eg.Wait(); err != nil { - return fmt.Errorf("failed to execute fee loading: %w", err) + return fmt.Errorf("failed to validate plugin policy: %v", err) } - return nil -} - -func (fp *FeePlugin) executeFeeLoading(ctx context.Context, feePolicy vtypes.PluginPolicy) error { - // Get list of fees from the verifier connected to the fee policy - feesResponse, err := fp.verifierApi.GetPublicKeysFees(feePolicy.PublicKey) + // Get the recipe from the policy for transaction validation + recipe, err := policy.GetRecipe() if err != nil { - return fmt.Errorf("failed to get plugin policy fees: %w", err) + return fmt.Errorf("failed to get recipe from policy: %v", err) } - // Early return if no fees to collect - if feesResponse.FeesPendingCollection <= 0 { - fp.logger.WithField("publicKey", feePolicy.PublicKey).Info("No fees pending collection") - return nil - } - - // If fees are greater than 0, we need to collect them - fp.logger.WithFields(logrus.Fields{ - "publicKey": feePolicy.PublicKey, - }).Info("Fees pending collection: ", feesResponse.FeesPendingCollection) + // Create a recipe engine for evaluating transactions + eng := engine.NewEngine() - checkAmount := 0 - for _, fee := range feesResponse.Fees { - if !fee.Collected { - checkAmount += fee.Amount - } - } - if checkAmount != feesResponse.FeesPendingCollection { - return fmt.Errorf("fees pending collection amount does not match the sum of the fees") - } - - for _, fee := range feesResponse.Fees { - if !fee.Collected { - - // Check if the fee has already been loaded and added to a fee run, if so, skip it - existingFee, err := fp.db.GetFees(ctx, fee.ID) - if err != nil { - return fmt.Errorf("failed to get fee: %w", err) - } - if len(existingFee) > 0 { - fp.logger.WithFields(logrus.Fields{ - "publicKey": feePolicy.PublicKey, - "feeId": fee.ID, - "runId": existingFee[0].FeeRunID, - }).Info("Fee already added to a fee run") - continue - } - - // If the fee hasn't been loaded, look for a draft run and add it to it - run, err := fp.db.GetPendingFeeRun(ctx, feePolicy.ID) + // Validate each proposed transaction + for _, tx := range txs { + for _, keysignMessage := range tx.Messages { + txBytes, err := base64.StdEncoding.DecodeString(keysignMessage.Message) if err != nil { - return fmt.Errorf("failed to get pending fee run: %w", err) - } - - // If no draft run is found, create a new one and add the fee to it - if run == nil { - run, err = fp.db.CreateFeeRun(ctx, feePolicy.ID, types.FeeRunStateDraft, fee) - if err != nil { - return fmt.Errorf("failed to create fee run: %w", err) - } - fp.logger.WithFields(logrus.Fields{ - "publicKey": feePolicy.PublicKey, - "feeIds": []uuid.UUID{fee.ID}, - "runId": run.ID, - }).Info("Fee run created") - - // If a draft run is found, add the fee to it - } else { - if err := fp.db.CreateFee(ctx, run.ID, fee); err != nil { - return fmt.Errorf("failed to create fee: %w", err) - } - fp.logger.WithFields(logrus.Fields{ - "publicKey": feePolicy.PublicKey, - "feeIds": []uuid.UUID{fee.ID}, - "runId": run.ID, - }).Info("Fee added to fee run") - } - } - } - - return nil -} - -/* ------------------------------------------------------------------------------------------------ -HANDLING TRANSACTIONS -here we handle the transactions for a fee run ------------------------------------------------------------------------------------------------- */ - -func (fp *FeePlugin) HandleTransactions(ctx context.Context, task *asynq.Task) error { - fp.logger.Info("Starting Fee Transaction Job. Acquiring mutex") - fp.transactingMutex.Lock() - fp.logger.Info("Mutex acquired") - - defer func() { - fp.transactingMutex.Unlock() - fp.logger.Info("Mutex released") - }() - - fp.logger.Info("Getting all fee runs") - runs, err := fp.db.GetAllFeeRuns(ctx) - if err != nil { - fp.logger.WithError(err).Error("Failed to get fee runs") - return fmt.Errorf("failed to get fee runs: %w", err) - } - - sem := semaphore.NewWeighted(int64(fp.config.Jobs.Transact.MaxConcurrentJobs)) - var wg sync.WaitGroup - var eg errgroup.Group - for _, run := range runs { - run = run - eg.Go(func() error { - //TODO also check failed runs - if run.Status != types.FeeRunStateDraft { - return nil + return fmt.Errorf("failed to decode transaction: %w", err) } - if run.TxHash != nil { - return nil - } - - if run.FeeCount == 0 || run.TotalAmount == 0 { - return nil - } - - fp.logger.WithFields(logrus.Fields{"runId": run.ID}).Info("Processing fee run") - feePolicy, err := fp.db.GetPluginPolicy(ctx, run.PolicyID) + // Evaluate if the transaction is allowed by the policy + _, err = eng.Evaluate(recipe, vgcommon.Chain(keysignMessage.Chain), txBytes) if err != nil { - return fmt.Errorf("failed to get fee policy: %w", err) + return fmt.Errorf("failed to evaluate transaction: %w", err) } - wg.Add(1) - fp.logger.WithFields(logrus.Fields{"runId": run.ID, "policyId": run.PolicyID}).Info("Retrieved fee policy") - - defer wg.Done() - if err := sem.Acquire(ctx, 1); err != nil { - return fmt.Errorf("failed to acquire semaphore: %w", err) - } - defer sem.Release(1) - if err := fp.executeFeeTransaction(ctx, run, *feePolicy); err != nil { - fp.logger.WithFields(logrus.Fields{ - "runId": run.ID, - }).Error("Failed to execute fee transaction") - return err - } - return nil - }) - } - - wg.Wait() - if err := eg.Wait(); err != nil { - return fmt.Errorf("failed to execute fee transaction: %w", err) - } - - return nil -} - -func (fp *FeePlugin) executeFeeTransaction(ctx context.Context, run types.FeeRun, feePolicy vtypes.PluginPolicy) error { - - fp.logger.WithFields(logrus.Fields{ - "runId": run.ID, - "policyId": feePolicy.ID, - }).Info("Checking if fee run policy id matches fee policy id") - if run.PolicyID != feePolicy.ID { - return fmt.Errorf("fee run policy id does not match fee policy id") - } - - // Get a vault and sign the transactions - fp.logger.WithFields(logrus.Fields{ - "publicKey": feePolicy.PublicKey, - }).Info("Getting vault") - vaultFileName := vgcommon.GetVaultBackupFilename(feePolicy.PublicKey, vtypes.PluginVultisigFees_feee.String()) - vaultContent, err := fp.vaultStorage.GetVault(vaultFileName) - if err != nil { - return fmt.Errorf("failed to get vault: %w", err) - } - if vaultContent == nil { - return fmt.Errorf("vault not found") - } - - // Propose the transactions - fp.logger.WithFields(logrus.Fields{ - "publicKey": feePolicy.PublicKey, - }).Info("Proposing transactions") - keySignRequests, err := fp.proposeTransactions(ctx, feePolicy, run) - if err != nil { - return fmt.Errorf("failed to propose transactions: %w", err) - } - fp.logger.WithFields(logrus.Fields{ - "publicKey": feePolicy.PublicKey, - }).Info("Key sign requests proposed") - for _, keySignRequest := range keySignRequests { - req := keySignRequest - if err := fp.initSign(ctx, req, feePolicy, run.ID); err != nil { - return fmt.Errorf("failed to init sign: %w", err) } } diff --git a/plugin/fees/helper.go b/plugin/fees/helper.go index 77948c8..8e0a02e 100644 --- a/plugin/fees/helper.go +++ b/plugin/fees/helper.go @@ -1,7 +1,9 @@ package fees import ( + "context" "encoding/hex" + "errors" "fmt" "math/big" "strings" @@ -11,8 +13,10 @@ import ( "github.com/ethereum/go-ethereum/common/hexutil" etypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/rlp" + "github.com/vultisig/mobile-tss-lib/tss" reth "github.com/vultisig/recipes/ethereum" "github.com/vultisig/recipes/sdk/evm" + vtypes "github.com/vultisig/verifier/types" ) func getHash(inTx evm.UnsignedTx, r, s, v []byte, chainID *big.Int) (*etypes.Transaction, error) { @@ -113,3 +117,13 @@ func hexutilDecode(hexStr string) ([]byte, error) { } return hexutil.Decode(hexStr) } + +// deprecated, use proposeTransactions instead as it relies on a fee run and a context +func (fp *FeePlugin) ProposeTransactions(policy vtypes.PluginPolicy) ([]vtypes.PluginKeysignRequest, error) { + return nil, errors.New("not implemented") +} + +// deprecated, no longer part of the flow. initSign handles the transaction signing, sending and recording of initial state. The process thereafter is handled by the post_tx flow +func (fp *FeePlugin) SigningComplete(ctx context.Context, signature tss.KeysignResponse, signRequest vtypes.PluginKeysignRequest, policy vtypes.PluginPolicy) error { + return fmt.Errorf("not implemented") +} diff --git a/plugin/fees/load.go b/plugin/fees/load.go new file mode 100644 index 0000000..beb622b --- /dev/null +++ b/plugin/fees/load.go @@ -0,0 +1,139 @@ +package fees + +import ( + "context" + "fmt" + "sync" + + "github.com/google/uuid" + "github.com/hibiken/asynq" + "github.com/sirupsen/logrus" + "github.com/vultisig/plugin/internal/types" + vtypes "github.com/vultisig/verifier/types" + "golang.org/x/sync/errgroup" + "golang.org/x/sync/semaphore" +) + +/* This code section is concerned with pulling fees into the plugin server */ + +/* ------------------------------------------------------------------------------------------------ +LOADING FEES +here we pull in a list of fees (amounts and ids) that are pending collection and add them to a fee run +------------------------------------------------------------------------------------------------ */ + +func (fp *FeePlugin) LoadFees(ctx context.Context, task *asynq.Task) error { + fp.transactingMutex.Lock() + defer fp.transactingMutex.Unlock() + + fp.logger.Info("Starting Fee Loading Job") + + feePolicies, err := fp.db.GetAllFeePolicies(ctx) + if err != nil { + return fmt.Errorf("failed to get plugin policy: %w", err) + } + + // We limit the number of concurrent fee loading operations to 10 + sem := semaphore.NewWeighted(int64(fp.config.Jobs.Load.MaxConcurrentJobs)) + var wg sync.WaitGroup + var eg errgroup.Group + + for _, feePolicy := range feePolicies { + wg.Add(1) + feePolicy = feePolicy + eg.Go(func() error { + defer wg.Done() + if err := sem.Acquire(ctx, 1); err != nil { + return fmt.Errorf("failed to acquire semaphore: %w", err) + } + defer sem.Release(1) + return fp.executeFeeLoading(ctx, feePolicy) + }) + } + + wg.Wait() + if err := eg.Wait(); err != nil { + return fmt.Errorf("failed to execute fee loading: %w", err) + } + return nil +} + +func (fp *FeePlugin) executeFeeLoading(ctx context.Context, feePolicy vtypes.PluginPolicy) error { + + // Get list of fees from the verifier connected to the fee policy + feesResponse, err := fp.verifierApi.GetPublicKeysFees(feePolicy.PublicKey) + if err != nil { + return fmt.Errorf("failed to get plugin policy fees: %w", err) + } + + // Early return if no fees to collect + if feesResponse.FeesPendingCollection <= 0 { + fp.logger.WithField("publicKey", feePolicy.PublicKey).Info("No fees pending collection") + return nil + } + + // If fees are greater than 0, we need to collect them + fp.logger.WithFields(logrus.Fields{ + "publicKey": feePolicy.PublicKey, + }).Info("Fees pending collection: ", feesResponse.FeesPendingCollection) + + checkAmount := 0 + for _, fee := range feesResponse.Fees { + if !fee.Collected { + checkAmount += fee.Amount + } + } + if checkAmount != feesResponse.FeesPendingCollection { + return fmt.Errorf("fees pending collection amount does not match the sum of the fees") + } + + for _, fee := range feesResponse.Fees { + if !fee.Collected { + + // Check if the fee has already been loaded and added to a fee run, if so, skip it + existingFee, err := fp.db.GetFees(ctx, fee.ID) + if err != nil { + return fmt.Errorf("failed to get fee: %w", err) + } + if len(existingFee) > 0 { + fp.logger.WithFields(logrus.Fields{ + "publicKey": feePolicy.PublicKey, + "feeId": fee.ID, + "runId": existingFee[0].FeeRunID, + }).Info("Fee already added to a fee run") + continue + } + + // If the fee hasn't been loaded, look for a draft run and add it to it + run, err := fp.db.GetPendingFeeRun(ctx, feePolicy.ID) + if err != nil { + return fmt.Errorf("failed to get pending fee run: %w", err) + } + + // If no draft run is found, create a new one and add the fee to it + if run == nil { + run, err = fp.db.CreateFeeRun(ctx, feePolicy.ID, types.FeeRunStateDraft, fee) + if err != nil { + return fmt.Errorf("failed to create fee run: %w", err) + } + fp.logger.WithFields(logrus.Fields{ + "publicKey": feePolicy.PublicKey, + "feeIds": []uuid.UUID{fee.ID}, + "runId": run.ID, + }).Info("Fee run created") + + // If a draft run is found, add the fee to it + } else { + if err := fp.db.CreateFee(ctx, run.ID, fee); err != nil { + return fmt.Errorf("failed to create fee: %w", err) + } + fp.logger.WithFields(logrus.Fields{ + "publicKey": feePolicy.PublicKey, + "feeIds": []uuid.UUID{fee.ID}, + "runId": run.ID, + }).Info("Fee added to fee run") + } + } + } + + return nil +} diff --git a/plugin/fees/transaction.go b/plugin/fees/transaction.go index cad4ba8..d42db0c 100644 --- a/plugin/fees/transaction.go +++ b/plugin/fees/transaction.go @@ -8,25 +8,209 @@ import ( "fmt" "math/big" "strconv" + "sync" gcommon "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" etypes "github.com/ethereum/go-ethereum/core/types" "github.com/google/uuid" + "github.com/hibiken/asynq" "github.com/sirupsen/logrus" "github.com/vultisig/mobile-tss-lib/tss" - "github.com/vultisig/recipes/engine" reth "github.com/vultisig/recipes/ethereum" "github.com/vultisig/recipes/resolver" rtypes "github.com/vultisig/recipes/types" vtypes "github.com/vultisig/verifier/types" "github.com/vultisig/vultisig-go/address" vgcommon "github.com/vultisig/vultisig-go/common" + "golang.org/x/sync/errgroup" + "golang.org/x/sync/semaphore" "github.com/vultisig/plugin/common" "github.com/vultisig/plugin/internal/types" ) +/* ------------------------------------------------------------------------------------------------ +HANDLING TRANSACTIONS +here we handle the transactions for a fee run +------------------------------------------------------------------------------------------------ */ + +func (fp *FeePlugin) HandleTransactions(ctx context.Context, task *asynq.Task) error { + fp.logger.Info("Starting Fee Transaction Job. Acquiring mutex") + fp.transactingMutex.Lock() + fp.logger.Info("Mutex acquired") + + defer func() { + fp.transactingMutex.Unlock() + fp.logger.Info("Mutex released") + }() + + fp.logger.Info("Getting all fee runs") + runs, err := fp.db.GetAllFeeRuns(ctx) + if err != nil { + fp.logger.WithError(err).Error("Failed to get fee runs") + return fmt.Errorf("failed to get fee runs: %w", err) + } + + sem := semaphore.NewWeighted(int64(fp.config.Jobs.Transact.MaxConcurrentJobs)) + var wg sync.WaitGroup + var eg errgroup.Group + for _, run := range runs { + run = run + eg.Go(func() error { + //TODO also check failed runs + if run.Status != types.FeeRunStateDraft { + return nil + } + + if run.TxHash != nil { + return nil + } + + if run.FeeCount == 0 || run.TotalAmount == 0 { + return nil + } + + fp.logger.WithFields(logrus.Fields{"runId": run.ID}).Info("Processing fee run") + feePolicy, err := fp.db.GetPluginPolicy(ctx, run.PolicyID) + if err != nil { + return fmt.Errorf("failed to get fee policy: %w", err) + } + wg.Add(1) + fp.logger.WithFields(logrus.Fields{"runId": run.ID, "policyId": run.PolicyID}).Info("Retrieved fee policy") + + defer wg.Done() + if err := sem.Acquire(ctx, 1); err != nil { + return fmt.Errorf("failed to acquire semaphore: %w", err) + } + defer sem.Release(1) + if err := fp.executeFeeTransaction(ctx, run, *feePolicy); err != nil { + fp.logger.WithError(err).WithFields(logrus.Fields{ + "runId": run.ID, + }).Error("Failed to execute fee transaction") + return err + } + return nil + }) + } + + wg.Wait() + if err := eg.Wait(); err != nil { + return fmt.Errorf("failed to execute fee transaction: %w", err) + } + + return nil +} + +func (fp *FeePlugin) executeFeeTransaction(ctx context.Context, run types.FeeRun, feePolicy vtypes.PluginPolicy) error { + + fp.logger.WithFields(logrus.Fields{ + "runId": run.ID, + "policyId": feePolicy.ID, + }).Info("Checking if fee run policy id matches fee policy id") + if run.PolicyID != feePolicy.ID { + return fmt.Errorf("fee run policy id does not match fee policy id") + } + + // Get a vault and sign the transactions + fp.logger.WithFields(logrus.Fields{ + "publicKey": feePolicy.PublicKey, + }).Info("Getting vault") + _, err := common.GetVaultFromPolicy(fp.vaultStorage, feePolicy, fp.encryptionSecret) + if err != nil { + return fmt.Errorf("failed to get vault: %w", err) + } + + // Propose the transactions + fp.logger.WithFields(logrus.Fields{ + "publicKey": feePolicy.PublicKey, + }).Info("Proposing transactions") + keySignRequests, err := fp.proposeTransactions(ctx, feePolicy, run) + if err != nil { + return fmt.Errorf("failed to propose transactions: %w", err) + } + fp.logger.WithFields(logrus.Fields{ + "publicKey": feePolicy.PublicKey, + }).Info("Key sign requests proposed") + for _, keySignRequest := range keySignRequests { + req := keySignRequest + if err := fp.initSign(ctx, req, feePolicy, run.ID); err != nil { + return fmt.Errorf("failed to init sign: %w", err) + } + } + + return nil +} + +func (fp *FeePlugin) initSign( + ctx context.Context, + req vtypes.PluginKeysignRequest, + pluginPolicy vtypes.PluginPolicy, + runId uuid.UUID, +) error { + + sigs, err := fp.signer.Sign(ctx, req) + if err != nil { + fp.logger.WithError(err).Error("Keysign failed") + return fmt.Errorf("failed to sign transaction: %w", err) + } + + if len(sigs) != 1 { + fp.logger. + WithField("sigs_count", len(sigs)). + Error("expected only 1 message+sig per request for evm") + return fmt.Errorf("failed to sign transaction: invalid signature count: %d", len(sigs)) + } + + var sig tss.KeysignResponse + for _, s := range sigs { + sig = s + } + + txBytes, txErr := hexutilDecode(req.Transaction) + r, rErr := hexutilDecode(sig.R) + s, sErr := hexutilDecode(sig.S) + v, vErr := hexutilDecode(sig.RecoveryID) + if txErr != nil || rErr != nil || sErr != nil || vErr != nil { + return fmt.Errorf("error decoding tx or sigs: %w", errors.Join(txErr, rErr, sErr, vErr)) + } + + txHash, err := getHash(txBytes, r, s, v, fp.config.ChainId) + if err != nil { + return fmt.Errorf("failed to get hash: %w", err) + } + + erc20tx, err := decodeTx(req.Transaction) + if err != nil { + fp.logger.WithError(err).Error("failed to decode tx") + return fmt.Errorf("failed to decode tx: %w", err) + } + + fp.logger.WithFields(logrus.Fields{ + "tx_hash": txHash.Hash().Hex(), + "tx_to": erc20tx.to.Hex(), + "tx_amount": erc20tx.amount.String(), + "tx_token": erc20tx.token.Hex(), + "public_key": pluginPolicy.PublicKey, + }).Info("fee collection transaction") + + tx, err := fp.eth.Send(ctx, txBytes, r, s, v) + if err != nil { + fp.logger.WithError(err).WithField("tx_hex", req.Transaction).Error("fp.eth.Send") + return fmt.Errorf("failed to send transaction: %w", err) + } + + // This is exceptionally important, as if it errors, the transaction will internally be recorded as draft, even after it's been broadcasted + if err := fp.db.SetFeeRunSent(ctx, runId, tx.Hash().Hex()); err != nil { //TODO pass the real tx id + return fmt.Errorf("failed to set fee run sent: %w", err) + } + + // Log successful transaction broadcast + fp.logger.WithField("hash", tx.Hash().Hex()).Info("fee collection transaction successfully broadcasted") + return nil + +} + func (fp *FeePlugin) proposeTransactions(ctx context.Context, policy vtypes.PluginPolicy, run types.FeeRun) ([]vtypes.PluginKeysignRequest, error) { if policy.ID != run.PolicyID { @@ -146,117 +330,3 @@ func (fp *FeePlugin) proposeTransactions(ctx context.Context, policy vtypes.Plug return txs, nil } - -// deprecated, use proposeTransactions instead as it relies on a fee run and a context -func (fp *FeePlugin) ProposeTransactions(policy vtypes.PluginPolicy) ([]vtypes.PluginKeysignRequest, error) { - return nil, errors.New("not implemented") -} - -func (fp *FeePlugin) initSign( - ctx context.Context, - req vtypes.PluginKeysignRequest, - pluginPolicy vtypes.PluginPolicy, - runId uuid.UUID, -) error { - - sigs, err := fp.signer.Sign(ctx, req) - if err != nil { - fp.logger.WithError(err).Error("Keysign failed") - return fmt.Errorf("failed to sign transaction: %w", err) - } - - if len(sigs) != 1 { - fp.logger. - WithField("sigs_count", len(sigs)). - Error("expected only 1 message+sig per request for evm") - return fmt.Errorf("failed to sign transaction: invalid signature count: %d", len(sigs)) - } - - var sig tss.KeysignResponse - for _, s := range sigs { - sig = s - } - - txBytes, txErr := hexutilDecode(req.Transaction) - r, rErr := hexutilDecode(sig.R) - s, sErr := hexutilDecode(sig.S) - v, vErr := hexutilDecode(sig.RecoveryID) - if txErr != nil || rErr != nil || sErr != nil || vErr != nil { - return fmt.Errorf("error decoding tx or sigs: %w", errors.Join(txErr, rErr, sErr, vErr)) - } - - txHash, err := getHash(txBytes, r, s, v, fp.config.ChainId) - if err != nil { - return fmt.Errorf("failed to get hash: %w", err) - } - - erc20tx, err := decodeTx(req.Transaction) - if err != nil { - fp.logger.WithError(err).Error("failed to decode tx") - return fmt.Errorf("failed to decode tx: %w", err) - } - - fp.logger.WithFields(logrus.Fields{ - "tx_hash": txHash.Hash().Hex(), - "tx_to": erc20tx.to.Hex(), - "tx_amount": erc20tx.amount.String(), - "tx_token": erc20tx.token.Hex(), - "public_key": pluginPolicy.PublicKey, - }).Info("fee collection transaction") - - tx, err := fp.eth.Send(ctx, txBytes, r, s, v) - if err != nil { - fp.logger.WithError(err).WithField("tx_hex", req.Transaction).Error("fp.eth.Send") - return fmt.Errorf("failed to send transaction: %w", err) - } - - // This is exceptionally important, as if it errors, the transaction will internally be recorded as draft, even after it's been broadcasted - if err := fp.db.SetFeeRunSent(ctx, runId, tx.Hash().Hex()); err != nil { //TODO pass the real tx id - return fmt.Errorf("failed to set fee run sent: %w", err) - } - - // Log successful transaction broadcast - fp.logger.WithField("hash", tx.Hash().Hex()).Info("fee collection transaction successfully broadcasted") - return nil - -} - -func (fp *FeePlugin) ValidateProposedTransactions(policy vtypes.PluginPolicy, txs []vtypes.PluginKeysignRequest) error { - // First validate the plugin policy itself - err := fp.ValidatePluginPolicy(policy) - if err != nil { - return fmt.Errorf("failed to validate plugin policy: %v", err) - } - - // Get the recipe from the policy for transaction validation - recipe, err := policy.GetRecipe() - if err != nil { - return fmt.Errorf("failed to get recipe from policy: %v", err) - } - - // Create a recipe engine for evaluating transactions - eng := engine.NewEngine() - - // Validate each proposed transaction - for _, tx := range txs { - for _, keysignMessage := range tx.Messages { - txBytes, err := base64.StdEncoding.DecodeString(keysignMessage.Message) - if err != nil { - return fmt.Errorf("failed to decode transaction: %w", err) - } - - // Evaluate if the transaction is allowed by the policy - _, err = eng.Evaluate(recipe, keysignMessage.Chain, txBytes) - if err != nil { - return fmt.Errorf("failed to evaluate transaction: %w", err) - } - } - } - - return nil -} - -// deprecated, no longer part of the flow. initSign handles the transaction signing, sending and recording of initial state. The process thereafter is handled by the post_tx flow -func (fp *FeePlugin) SigningComplete(ctx context.Context, signature tss.KeysignResponse, signRequest vtypes.PluginKeysignRequest, policy vtypes.PluginPolicy) error { - return fmt.Errorf("not implemented") -}