Skip to content
This repository was archived by the owner on Feb 8, 2026. It is now read-only.

Fee code location move #158

Open
garry-sharp wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fee code location move #158

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e8aff03
🐛 fee config bug fix and target added to tx fee validations
Aug 16, 2025
e7ef35f
🚚 Move some fee code around into more sensible places
Aug 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 30 additions & 13 deletions plugin/fees/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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 {
Expand Down Expand Up @@ -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")

Expand All @@ -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
}
}
Expand All @@ -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 == "" {
Expand Down
244 changes: 19 additions & 225 deletions plugin/fees/fees.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"
Expand All @@ -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"
)
Expand Down Expand Up @@ -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)
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

Expand Down
Loading