From 4bb281bd4afefdf4da9741a3d6636d39c5ea2575 Mon Sep 17 00:00:00 2001 From: Garry Sharp <> Date: Fri, 1 Aug 2025 22:06:53 +0100 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Resend=20stuck=20txs=20if=20nonce?= =?UTF-8?q?=20is=20the=20same=20as=20pending?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- etc/vultisig/fee.yml | 3 +- internal/types/fees.go | 9 ++ plugin/fees/config.go | 43 +++++-- plugin/fees/fees.go | 24 +++- plugin/fees/helper.go | 121 +++++++++++++----- plugin/fees/post_tx.go | 93 +++++++++++++- plugin/fees/transaction.go | 78 ++++++----- storage/db.go | 11 +- storage/postgres/fees.go | 49 ++++++- .../plugin/20250630152230_fee_runs.sql | 10 ++ storage/postgres/schema/schema.sql | 16 +++ 11 files changed, 362 insertions(+), 95 deletions(-) diff --git a/etc/vultisig/fee.yml b/etc/vultisig/fee.yml index 3d9d759..e77af6b 100644 --- a/etc/vultisig/fee.yml +++ b/etc/vultisig/fee.yml @@ -13,4 +13,5 @@ jobs: max_concurrent_jobs: 10 post: cronexpr: "@every 5m" - max_concurrent_jobs: 10 \ No newline at end of file + max_concurrent_jobs: 10 +dry_run: false \ No newline at end of file diff --git a/internal/types/fees.go b/internal/types/fees.go index 78c1310..64e3a53 100644 --- a/internal/types/fees.go +++ b/internal/types/fees.go @@ -37,3 +37,12 @@ type FeeRun struct { FeeCount int `db:"fee_count"` Fees []Fee `db:"fees"` } + +type FeeRunTx struct { + ID uuid.UUID `db:"id"` + FeeRunID uuid.UUID `db:"fee_run_id"` + Tx string `db:"tx"` + Hash string `db:"hash"` + BlockNumber uint64 `db:"block_number"` + CreatedAt time.Time `db:"created_at"` +} 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/fees.go b/plugin/fees/fees.go index ba978f7..4f61609 100644 --- a/plugin/fees/fees.go +++ b/plugin/fees/fees.go @@ -203,9 +203,26 @@ func (fp *FeePlugin) executeFeeLoading(ctx context.Context, feePolicy vtypes.Plu } // If no draft run is found, create a new one and add the fee to it + dbTx, err := fp.db.Pool().Begin(ctx) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + + var txErr error + defer func() { + if txErr != nil { + if err := dbTx.Rollback(ctx); err != nil { + fp.logger.WithError(err).Error("Failed to rollback transaction") + } + } else { + if err := dbTx.Commit(ctx); err != nil { + fp.logger.WithError(err).Error("Failed to commit transaction") + } + } + }() if run == nil { - run, err = fp.db.CreateFeeRun(ctx, feePolicy.ID, types.FeeRunStateDraft, fee) - if err != nil { + run, txErr = fp.db.CreateFeeRun(ctx, dbTx, feePolicy.ID, types.FeeRunStateDraft, fee) + if txErr != nil { return fmt.Errorf("failed to create fee run: %w", err) } fp.logger.WithFields(logrus.Fields{ @@ -216,7 +233,7 @@ func (fp *FeePlugin) executeFeeLoading(ctx context.Context, feePolicy vtypes.Plu // If a draft run is found, add the fee to it } else { - if err := fp.db.CreateFee(ctx, run.ID, fee); err != nil { + if txErr = fp.db.CreateFee(ctx, dbTx, run.ID, fee); err != nil { return fmt.Errorf("failed to create fee: %w", err) } fp.logger.WithFields(logrus.Fields{ @@ -225,6 +242,7 @@ func (fp *FeePlugin) executeFeeLoading(ctx context.Context, feePolicy vtypes.Plu "runId": run.ID, }).Info("Fee added to fee run") } + } } diff --git a/plugin/fees/helper.go b/plugin/fees/helper.go index 77948c8..a3c6a73 100644 --- a/plugin/fees/helper.go +++ b/plugin/fees/helper.go @@ -1,6 +1,7 @@ package fees import ( + "context" "encoding/hex" "fmt" "math/big" @@ -11,8 +12,13 @@ import ( "github.com/ethereum/go-ethereum/common/hexutil" etypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/rlp" + "github.com/google/uuid" + "github.com/vultisig/plugin/common" reth "github.com/vultisig/recipes/ethereum" "github.com/vultisig/recipes/sdk/evm" + "github.com/vultisig/verifier/address" + vcommon "github.com/vultisig/verifier/common" + vtypes "github.com/vultisig/verifier/types" ) func getHash(inTx evm.UnsignedTx, r, s, v []byte, chainID *big.Int) (*etypes.Transaction, error) { @@ -34,76 +40,87 @@ func getHash(inTx evm.UnsignedTx, r, s, v []byte, chainID *big.Int) (*etypes.Tra return outTx, nil } -type erc20tx struct { +type erc20Data struct { to ecommon.Address `json:"to"` amount *big.Int `json:"amount"` token ecommon.Address `json:"token"` } -func decodeTx(rawHex string) (*erc20tx, error) { - type unsignedDynamicFeeTx struct { - ChainID *big.Int - Nonce uint64 - GasTipCap *big.Int - GasFeeCap *big.Int - Gas uint64 - To *ecommon.Address - Value *big.Int - Data []byte - AccessList etypes.AccessList - } +type unsignedDynamicFeeTx struct { + ChainID *big.Int + Nonce uint64 + GasTipCap *big.Int + GasFeeCap *big.Int + Gas uint64 + To *ecommon.Address + Value *big.Int + Data []byte + AccessList etypes.AccessList +} +func decodeUnsignedTx(rawHex string) (*unsignedDynamicFeeTx, error) { rawHex = strings.TrimPrefix(rawHex, "0x") - rawBytes, err := hex.DecodeString(rawHex) if err != nil { return nil, fmt.Errorf("hex decode failed: %w", err) } - // Check transaction type (EIP-1559 is 0x02) if len(rawBytes) == 0 || rawBytes[0] != 0x02 { return nil, fmt.Errorf("unsupported transaction type: 0x%02x", rawBytes[0]) } + // Decode RLP (strip type byte) tx := new(unsignedDynamicFeeTx) - err = rlp.DecodeBytes(rawBytes[1:], tx) - if err != nil { + if err := rlp.DecodeBytes(rawBytes[1:], tx); err != nil { return nil, fmt.Errorf("rlp decode failed: %w", err) } - // Parse the ERC20 transfer ABI + return tx, nil +} + +// pass in a types.Transaction or a unsignedDynamicFeeTx +func parseErc20Tx[T any](transaction T) (*erc20Data, error) { + // Decode ERC20 transfer ABI const transferABI = `[{"name":"transfer","type":"function","inputs":[{"name":"to","type":"address"},{"name":"value","type":"uint256"}],"outputs":[{"name":"","type":"bool"}]}]` - parsedABI, err := abi.JSON(strings.NewReader(transferABI)) - if err != nil { - return nil, fmt.Errorf("failed to parse ABI") + parsedABI, _ := abi.JSON(strings.NewReader(transferABI)) + + var data []byte + var to *ecommon.Address + + switch tx := any(transaction).(type) { + case *etypes.Transaction: + data = tx.Data() + to = tx.To() + case *unsignedDynamicFeeTx: + data = tx.Data + to = tx.To + default: + return nil, fmt.Errorf("unsupported transaction struct type") } - // Get the method by selector - method, err := parsedABI.MethodById(tx.Data[:4]) + method, err := parsedABI.MethodById(data[:4]) if err != nil { return nil, fmt.Errorf("unknown method ID") } - // Decode the arguments args := make(map[string]interface{}) - if err := method.Inputs.UnpackIntoMap(args, tx.Data[4:]); err != nil { - return nil, fmt.Errorf("failed get recipient and amount from tx") + if err := method.Inputs.UnpackIntoMap(args, data[4:]); err != nil { + return nil, fmt.Errorf("failed to unpack args") } recipient, ok := args["to"].(ecommon.Address) if !ok { - return nil, fmt.Errorf("invalid recipient address in tx data") + return nil, fmt.Errorf("invalid recipient type") } - amount, ok := args["value"].(*big.Int) if !ok { - return nil, fmt.Errorf("invalid amount in tx data") + return nil, fmt.Errorf("invalid amount type") } - return &erc20tx{ + return &erc20Data{ to: recipient, amount: amount, - token: *tx.To, + token: *to, }, nil } @@ -113,3 +130,45 @@ func hexutilDecode(hexStr string) ([]byte, error) { } return hexutil.Decode(hexStr) } + +func appendSignature(inTx evm.UnsignedTx, r, s, v []byte, chainID *big.Int) (*etypes.Transaction, error) { + var sig []byte + sig = append(sig, r...) + sig = append(sig, s...) + sig = append(sig, v...) + + inTxDecoded, err := reth.DecodeUnsignedPayload(inTx) + if err != nil { + return nil, fmt.Errorf("reth.DecodeUnsignedPayload: %w", err) + } + + outTx, err := etypes.NewTx(inTxDecoded).WithSignature(etypes.LatestSignerForChainID(chainID), sig) + if err != nil { + return nil, fmt.Errorf("types.NewTx.WithSignature: %w", err) + } + + return outTx, nil +} + +func (fp *FeePlugin) getEthAddressFromFeePolicy(policy vtypes.PluginPolicy) (string, error) { + vault, err := common.GetVaultFromPolicy(fp.vaultStorage, policy, fp.encryptionSecret) + if err != nil { + return "", fmt.Errorf("failed to get vault from policy: %v", err) + } + + // Get the ethereum derived addresses from the vaults master public key + fromAddress, _, _, err := address.GetAddress(vault.PublicKeyEcdsa, vault.HexChainCode, vcommon.Ethereum) + if err != nil { + return "", fmt.Errorf("failed to get eth address: %v", err) + } + + return fromAddress, nil +} + +func (fp *FeePlugin) getEthAddressFromFeePolicyId(policyId uuid.UUID) (string, error) { + policy, err := fp.db.GetPluginPolicy(context.Background(), policyId) + if err != nil { + return "", fmt.Errorf("failed to get policy: %v", err) + } + return fp.getEthAddressFromFeePolicy(*policy) +} diff --git a/plugin/fees/post_tx.go b/plugin/fees/post_tx.go index 718e9fa..59ba497 100644 --- a/plugin/fees/post_tx.go +++ b/plugin/fees/post_tx.go @@ -7,6 +7,8 @@ import ( "github.com/ethereum/go-ethereum" ecommon "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" @@ -51,6 +53,7 @@ func (fp *FeePlugin) HandlePostTx(ctx context.Context, task *asynq.Task) error { } wg.Wait() if err := eg.Wait(); err != nil { + fp.logger.WithError(err).Error("failed to execute fee run status check") return fmt.Errorf("failed to execute fee run status check: %w", err) } fp.logger.Info("Fee run status check completed") @@ -66,9 +69,71 @@ func (fp *FeePlugin) updateStatus(ctx context.Context, run types.FeeRun, current receipt, err := fp.ethClient.TransactionReceipt(ctx, hash) if err == ethereum.NotFound { - // TODO rebroadcast logic - fp.logger.WithFields(logrus.Fields{"run_id": run.ID}).Info("tx not found on chain, rebroadcasting") - return nil + lastTxs, err := fp.db.GetFeeRunTxs(ctx, run.ID) + if err != nil { + return fmt.Errorf("failed to get fee run txs: %w", err) + } + if len(lastTxs) == 0 { + return fmt.Errorf("no tx found for run %s", run.ID) + } + txHex := lastTxs[0].Tx + + var tx etypes.Transaction + txBytes, err := hexutil.Decode(txHex) + if err != nil { + return fmt.Errorf("failed to decode tx: %w", err) + } + if err := tx.UnmarshalBinary(txBytes); err != nil { + return fmt.Errorf("failed to unmarshal tx: %w", err) + } + + fromAddress, err := fp.getEthAddressFromFeePolicyId(run.PolicyID) + if err != nil { + return fmt.Errorf("failed to get eth address from fee policy: %w", err) + } + + nonce, err := fp.ethClient.PendingNonceAt(ctx, ecommon.HexToAddress(fromAddress)) + if err != nil { + return fmt.Errorf("failed to get pending nonce: %w", err) + } + + // Rebroadcast if pending nonce is the same as the nonce in the tx + if tx.Nonce() == nonce { + var err error + dbTx, err := fp.db.Pool().Begin(ctx) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + txBytes, err := tx.MarshalBinary() + if err != nil { + return fmt.Errorf("failed to marshal tx: %w", err) + } + defer func() { + if err != nil { + if err := dbTx.Rollback(ctx); err != nil { + fp.logger.WithError(err).Error("failed to rollback transaction") + } + } else { + if err := dbTx.Commit(ctx); err != nil { + fp.logger.WithError(err).Error("failed to commit transaction") + } + } + }() + err = fp.db.CreateFeeRunTx(ctx, dbTx, run.ID, txBytes, tx.Hash().Hex(), 0, fp.config.ChainId) + if err != nil { + return fmt.Errorf("failed to create fee run tx: %w", err) + } + err = fp.ethClient.SendTransaction(ctx, &tx) + if err != nil { + return fmt.Errorf("failed to send tx: %w", err) + } + + fp.logger.WithFields(logrus.Fields{"run_id": run.ID}).Info("tx rebroadcasted") + return nil + } else { + //TODO handle an earlier nonce or later nonce + return nil + } } if receipt.Status == 1 { if currentBlock > receipt.BlockNumber.Uint64()+fp.config.Jobs.Post.SuccessConfirmations { @@ -79,14 +144,28 @@ func (fp *FeePlugin) updateStatus(ctx context.Context, run types.FeeRun, current ids = append(ids, fee.ID) } - if err = fp.verifierApi.MarkFeeAsCollected(*run.TxHash, run.CreatedAt, ids...); err != nil { - return fmt.Errorf("failed to mark fee as collected on verifier: %w", err) + dbTx, err := fp.db.Pool().Begin(ctx) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) } + var txErr error + defer func() { + if txErr != nil { + dbTx.Rollback(ctx) + } else { + dbTx.Commit(ctx) + } + }() - // This is semi critical code as it could create a state mismatch between the verifier and the database. - if err = fp.db.SetFeeRunSuccess(ctx, run.ID); err != nil { + if txErr = fp.db.SetFeeRunSuccess(ctx, dbTx, run.ID); err != nil { return fmt.Errorf("failed to set fee run success: %w", err) } + + // include this in the errors too, as if the verifier is not updated, the there will be a state mismatch between the verifier and the plugin. + if txErr = fp.verifierApi.MarkFeeAsCollected(*run.TxHash, run.CreatedAt, ids...); err != nil { + return fmt.Errorf("failed to mark fee as collected on verifier: %w", err) + } + } else { fp.logger.WithFields(logrus.Fields{"run_id": run.ID}).Info("tx successful, but not enough confirmations, waiting for more") return nil diff --git a/plugin/fees/transaction.go b/plugin/fees/transaction.go index 39cc1ca..40367d6 100644 --- a/plugin/fees/transaction.go +++ b/plugin/fees/transaction.go @@ -12,14 +12,12 @@ import ( "github.com/google/uuid" "github.com/sirupsen/logrus" "github.com/vultisig/mobile-tss-lib/tss" - "github.com/vultisig/plugin/common" rcommon "github.com/vultisig/recipes/common" "github.com/vultisig/recipes/engine" reth "github.com/vultisig/recipes/ethereum" resolver "github.com/vultisig/recipes/resolver" rtypes "github.com/vultisig/recipes/types" - "github.com/vultisig/verifier/address" vcommon "github.com/vultisig/verifier/common" vtypes "github.com/vultisig/verifier/types" @@ -35,9 +33,9 @@ func (fp *FeePlugin) proposeTransactions(ctx context.Context, policy vtypes.Plug return nil, fmt.Errorf("policy id does not match run policy id") } - vault, err := common.GetVaultFromPolicy(fp.vaultStorage, policy, fp.encryptionSecret) + fromAddress, err := fp.getEthAddressFromFeePolicy(policy) if err != nil { - return nil, fmt.Errorf("failed to get vault from policy: %v", err) + return nil, fmt.Errorf("failed to get eth address from fee policy: %v", err) } // ERC20 USDC Token List @@ -49,12 +47,6 @@ func (fp *FeePlugin) proposeTransactions(ctx context.Context, policy vtypes.Plug Decimals: 6, } - // Get the ethereum derived addresses from the vaults master public key - ethAddress, _, _, err := address.GetAddress(vault.PublicKeyEcdsa, vault.HexChainCode, vcommon.Ethereum) - if err != nil { - return nil, fmt.Errorf("failed to get eth address: %v", err) - } - // Get some consts and types needed for later recipe, err := policy.GetRecipe() if err != nil { @@ -90,7 +82,6 @@ func (fp *FeePlugin) proposeTransactions(ctx context.Context, policy vtypes.Plug 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 { @@ -100,28 +91,25 @@ func (fp *FeePlugin) proposeTransactions(ctx context.Context, policy vtypes.Plug if gcommon.HexToAddress(token) != gcommon.HexToAddress(usdc.Address) { return nil, fmt.Errorf("token address does not match usdc address") } - amount := run.TotalAmount - tx, err := fp.eth.MakeAnyTransfer(ctx, - gcommon.HexToAddress(ethAddress), + gcommon.HexToAddress(fromAddress), 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) } 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)) + txHashToSign := etypes.LatestSignerForChainID(fp.config.ChainId).Hash(etypes.NewTx(txData)) msgHash := sha256.Sum256(txHashToSign.Bytes()) - signRequest := vtypes.PluginKeysignRequest{ KeysignRequest: vtypes.KeysignRequest{ PublicKey: policy.PublicKey, @@ -141,10 +129,8 @@ func (fp *FeePlugin) proposeTransactions(ctx context.Context, policy vtypes.Plug }, Transaction: txHex, } - txs = append(txs, signRequest) } - return txs, nil } @@ -191,10 +177,14 @@ func (fp *FeePlugin) initSign( return fmt.Errorf("failed to get hash: %w", err) } - erc20tx, err := decodeTx(req.Transaction) + erc20tx := new(erc20Data) + unsignedTx, err := decodeUnsignedTx(req.Transaction) + if err != nil { + return fmt.Errorf("failed to decode unsigned tx: %w", err) + } + erc20tx, err = parseErc20Tx(unsignedTx) if err != nil { - fp.logger.WithError(err).Error("failed to decode tx") - return fmt.Errorf("failed to decode tx: %w", err) + return fmt.Errorf("failed to parse erc20 tx: %w", err) } fp.logger.WithFields(logrus.Fields{ @@ -205,19 +195,49 @@ func (fp *FeePlugin) initSign( "public_key": pluginPolicy.PublicKey, }).Info("fee collection transaction") - tx, err := fp.eth.Send(ctx, txBytes, r, s, v) + txWithSig, err := appendSignature(txBytes, r, s, v, fp.config.ChainId) + if err != nil { + return fmt.Errorf("failed to append signature: %w", err) + } + + signedTxBytes, err := txWithSig.MarshalBinary() + if err != nil { + return fmt.Errorf("failed to marshal transaction: %w", err) + } + + dbTx, err := fp.db.Pool().Begin(ctx) 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) + return fmt.Errorf("failed to begin 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 + var txdbErr error + defer func() { + if txdbErr != nil { + dbTx.Rollback(ctx) + } + }() + + if txdbErr = fp.db.SetFeeRunSent(ctx, dbTx, runId, txWithSig.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") + if txdbErr = fp.db.CreateFeeRunTx(ctx, dbTx, runId, signedTxBytes, txWithSig.Hash().Hex(), 0, fp.config.ChainId); err != nil { + return fmt.Errorf("failed to create fee run tx: %w", err) + } + if !fp.config.DryRun { + if err = fp.ethClient.SendTransaction(ctx, txWithSig); err != nil { + return fmt.Errorf("failed to send transaction: %w", err) + } + txDataHex := hexutil.Encode(txWithSig.Data()) + fp.logger.WithFields(logrus.Fields{"hash": txWithSig.Hash().Hex(), "tx_data": txDataHex}).Info("fee collection transaction successfully broadcasted") + } else { + fp.logger.WithField("tx_hash", txHash.Hash().Hex()).Info("dry-run: fee collection transaction would have been broadcasted") + } + // This would be a CRITICAL error, as the transaction would be recorded as draft, even after it's been broadcasted + if err = dbTx.Commit(ctx); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) + } + return nil } diff --git a/storage/db.go b/storage/db.go index 8a98027..a1e8484 100644 --- a/storage/db.go +++ b/storage/db.go @@ -2,6 +2,7 @@ package storage import ( "context" + "math/big" "github.com/google/uuid" "github.com/jackc/pgx/v5" @@ -22,14 +23,16 @@ type DatabaseStorage interface { InsertPluginPolicyTx(ctx context.Context, dbTx pgx.Tx, policy vtypes.PluginPolicy) (*vtypes.PluginPolicy, error) UpdatePluginPolicyTx(ctx context.Context, dbTx pgx.Tx, policy vtypes.PluginPolicy) (*vtypes.PluginPolicy, error) - CreateFeeRun(ctx context.Context, policyId uuid.UUID, state types.FeeRunState, fees ...verifierapi.FeeDto) (*types.FeeRun, error) - SetFeeRunSent(ctx context.Context, runId uuid.UUID, txHash string) error - SetFeeRunSuccess(ctx context.Context, runId uuid.UUID) error + CreateFeeRun(ctx context.Context, dbTx pgx.Tx, policyId uuid.UUID, state types.FeeRunState, fees ...verifierapi.FeeDto) (*types.FeeRun, error) + SetFeeRunSent(ctx context.Context, dbTx pgx.Tx, runId uuid.UUID, txHash string) error + SetFeeRunSuccess(ctx context.Context, dbTx pgx.Tx, runId uuid.UUID) error GetAllFeeRuns(ctx context.Context, statuses ...types.FeeRunState) ([]types.FeeRun, error) // If no statuses are provided, all fee runs are returned. GetFees(ctx context.Context, feeIds ...uuid.UUID) ([]types.Fee, error) GetPendingFeeRun(ctx context.Context, policyId uuid.UUID) (*types.FeeRun, error) - CreateFee(ctx context.Context, runId uuid.UUID, fee verifierapi.FeeDto) error + CreateFee(ctx context.Context, dbTx pgx.Tx, runId uuid.UUID, fee verifierapi.FeeDto) error GetFeeRuns(ctx context.Context, state types.FeeRunState) ([]types.FeeRun, error) + CreateFeeRunTx(ctx context.Context, dbTx pgx.Tx, runId uuid.UUID, tx []byte, hash string, blockNumber uint64, chainID *big.Int) error + GetFeeRunTxs(ctx context.Context, runId uuid.UUID) ([]types.FeeRunTx, error) Pool() *pgxpool.Pool } diff --git a/storage/postgres/fees.go b/storage/postgres/fees.go index afec256..f426e2a 100644 --- a/storage/postgres/fees.go +++ b/storage/postgres/fees.go @@ -4,7 +4,9 @@ import ( "context" "errors" "fmt" + "math/big" + "github.com/ethereum/go-ethereum/common/hexutil" "github.com/google/uuid" "github.com/jackc/pgx/v5" @@ -12,7 +14,7 @@ import ( "github.com/vultisig/plugin/internal/verifierapi" ) -func (p *PostgresBackend) CreateFeeRun(ctx context.Context, policyId uuid.UUID, state types.FeeRunState, fees ...verifierapi.FeeDto) (*types.FeeRun, error) { +func (p *PostgresBackend) CreateFeeRun(ctx context.Context, dbTx pgx.Tx, policyId uuid.UUID, state types.FeeRunState, fees ...verifierapi.FeeDto) (*types.FeeRun, error) { // Check policy id is valid query := `select plugin_id from plugin_policies where id = $1` policyrows := p.pool.QueryRow(ctx, query, policyId) @@ -57,16 +59,16 @@ func (p *PostgresBackend) CreateFeeRun(ctx context.Context, policyId uuid.UUID, return &run, nil } -func (p *PostgresBackend) SetFeeRunSent(ctx context.Context, runId uuid.UUID, txHash string) error { - _, err := p.pool.Exec(ctx, `update fee_run set status = $1, tx_hash = $2 where id = $3`, types.FeeRunStateSent, txHash, runId) +func (p *PostgresBackend) SetFeeRunSent(ctx context.Context, dbTx pgx.Tx, runId uuid.UUID, txHash string) error { + _, err := dbTx.Exec(ctx, `update fee_run set status = $1, tx_hash = $2 where id = $3`, types.FeeRunStateSent, txHash, runId) if err != nil { return fmt.Errorf("failed to update fee run: %w", err) } return nil } -func (p *PostgresBackend) SetFeeRunSuccess(ctx context.Context, runId uuid.UUID) error { - _, err := p.pool.Exec(ctx, `update fee_run set status = $1 where id = $2`, types.FeeRunStateSuccess, runId) +func (p *PostgresBackend) SetFeeRunSuccess(ctx context.Context, dbTx pgx.Tx, runId uuid.UUID) error { + _, err := dbTx.Exec(ctx, `update fee_run set status = $1 where id = $2`, types.FeeRunStateSuccess, runId) if err != nil { return fmt.Errorf("failed to update fee run: %w", err) } @@ -170,8 +172,8 @@ func (p *PostgresBackend) GetPendingFeeRun(ctx context.Context, policyId uuid.UU return &run, nil } -func (p *PostgresBackend) CreateFee(ctx context.Context, runId uuid.UUID, fee verifierapi.FeeDto) error { - _, err := p.pool.Exec(ctx, `insert into fee (id, fee_run_id, amount) values ($1, $2, $3)`, fee.ID, runId, fee.Amount) +func (p *PostgresBackend) CreateFee(ctx context.Context, dbTx pgx.Tx, runId uuid.UUID, fee verifierapi.FeeDto) error { + _, err := dbTx.Exec(ctx, `insert into fee (id, fee_run_id, amount) values ($1, $2, $3)`, fee.ID, runId, fee.Amount) if err != nil { return fmt.Errorf("failed to insert fee: %w", err) } @@ -197,3 +199,36 @@ func (p *PostgresBackend) GetFeeRuns(ctx context.Context, state types.FeeRunStat } return runs, nil } + +func (p *PostgresBackend) CreateFeeRunTx(ctx context.Context, dbTx pgx.Tx, runId uuid.UUID, tx []byte, hash string, blockNumber uint64, chainID *big.Int) error { + txHex := hexutil.Encode(tx) + var chainIDInt uint64 + if chainID != nil { + chainIDInt = chainID.Uint64() + } + _, err := dbTx.Exec(ctx, `insert into fee_run_tx (fee_run_id, tx, hash, block_number, chain_id) values ($1, $2, $3, $4, $5)`, runId, txHex, hash, blockNumber, chainIDInt) + if err != nil { + return fmt.Errorf("failed to insert fee run tx: %w", err) + } + return nil +} + +// Get a list of run txs most recent first +func (p *PostgresBackend) GetFeeRunTxs(ctx context.Context, runId uuid.UUID) ([]types.FeeRunTx, error) { + query := `select id, fee_run_id, tx, hash, block_number, created_at from fee_run_tx where fee_run_id = $1 order by created_at desc` + rows, err := p.pool.Query(ctx, query, runId) + if err != nil { + return nil, err + } + defer rows.Close() + txs := []types.FeeRunTx{} + for rows.Next() { + var tx types.FeeRunTx + err := rows.Scan(&tx.ID, &tx.FeeRunID, &tx.Tx, &tx.Hash, &tx.BlockNumber, &tx.CreatedAt) + if err != nil { + return nil, err + } + txs = append(txs, tx) + } + return txs, nil +} diff --git a/storage/postgres/migrations/plugin/20250630152230_fee_runs.sql b/storage/postgres/migrations/plugin/20250630152230_fee_runs.sql index dedfe34..0cf5ad5 100644 --- a/storage/postgres/migrations/plugin/20250630152230_fee_runs.sql +++ b/storage/postgres/migrations/plugin/20250630152230_fee_runs.sql @@ -38,6 +38,16 @@ CREATE INDEX IF NOT EXISTS idx_fee_run_status ON fee_run(status); CREATE INDEX IF NOT EXISTS idx_fee_run_created_at ON fee_run(created_at); CREATE INDEX IF NOT EXISTS idx_fee_id_fee_run_id ON fee(fee_run_id); +CREATE TABLE IF NOT EXISTS fee_run_tx ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + fee_run_id UUID NOT NULL REFERENCES fee_run(id) ON DELETE CASCADE, + chain_id BIGINT NOT NULL, + tx TEXT NOT NULL, + hash VARCHAR(66) NOT NULL, + block_number BIGINT NOT NULL, + created_at TIMESTAMP DEFAULT NOW() +); + -- Create trigger to update updated_at timestamp CREATE OR REPLACE FUNCTION update_updated_at_column() RETURNS TRIGGER AS $$ diff --git a/storage/postgres/schema/schema.sql b/storage/postgres/schema/schema.sql index 8ed668c..e259bb0 100644 --- a/storage/postgres/schema/schema.sql +++ b/storage/postgres/schema/schema.sql @@ -77,6 +77,16 @@ CREATE TABLE "fee_run" ( CONSTRAINT "fee_run_status_check" CHECK ((("status")::"text" = ANY ((ARRAY['draft'::character varying, 'sent'::character varying, 'completed'::character varying, 'failed'::character varying])::"text"[]))) ); +CREATE TABLE "fee_run_tx" ( + "id" "uuid" DEFAULT "gen_random_uuid"() NOT NULL, + "fee_run_id" "uuid" NOT NULL, + "chain_id" bigint NOT NULL, + "tx" "text" NOT NULL, + "hash" character varying(66) NOT NULL, + "block_number" bigint NOT NULL, + "created_at" timestamp without time zone DEFAULT "now"() +); + CREATE VIEW "fee_run_with_totals" AS SELECT "fr"."id", "fr"."status", @@ -133,6 +143,9 @@ ALTER TABLE ONLY "fee" ALTER TABLE ONLY "fee_run" ADD CONSTRAINT "fee_run_pkey" PRIMARY KEY ("id"); +ALTER TABLE ONLY "fee_run_tx" + ADD CONSTRAINT "fee_run_tx_pkey" PRIMARY KEY ("id"); + ALTER TABLE ONLY "plugin_policies" ADD CONSTRAINT "plugin_policies_pkey" PRIMARY KEY ("id"); @@ -174,3 +187,6 @@ ALTER TABLE ONLY "fee" ALTER TABLE ONLY "fee_run" ADD CONSTRAINT "fee_run_policy_id_fkey" FOREIGN KEY ("policy_id") REFERENCES "plugin_policies"("id") ON DELETE CASCADE; +ALTER TABLE ONLY "fee_run_tx" + ADD CONSTRAINT "fee_run_tx_fee_run_id_fkey" FOREIGN KEY ("fee_run_id") REFERENCES "fee_run"("id") ON DELETE CASCADE; +