Isolate transaction broadcasting latency from regas logic #465
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,6 +22,8 @@ var ( | |
| gasPricePercentageMultiplier = big.NewInt(10) | ||
| hundred = big.NewInt(100) | ||
| maxSendTransactionRetry = 3 | ||
| queryTickerDuration = 3 * time.Second | ||
| ErrTransactionNotBroadcasted = errors.New("transaction not broadcasted") | ||
| ) | ||
|
|
||
| // TxnManager receives transactions from the caller, sends them to the chain, and monitors their status. | ||
|
|
@@ -36,7 +38,8 @@ type TxnManager interface { | |
|
|
||
| type transaction struct { | ||
| *types.Transaction | ||
| TxID walletsdk.TxID | ||
| requestedAt time.Time | ||
| } | ||
|
|
||
| type TxnRequest struct { | ||
|
|
@@ -70,25 +73,27 @@ type txnManager struct { | |
| requestChan chan *TxnRequest | ||
| logger logging.Logger | ||
|
|
||
| receiptChan chan *ReceiptOrErr | ||
| queueSize int | ||
| txnBroadcastTimeout time.Duration | ||
| txnRefreshInterval time.Duration | ||
| metrics *TxnManagerMetrics | ||
| } | ||
|
|
||
| var _ TxnManager = (*txnManager)(nil) | ||
|
|
||
| func NewTxnManager(ethClient common.EthClient, wallet walletsdk.Wallet, numConfirmations, queueSize int, txnBroadcastTimeout time.Duration, txnRefreshInterval time.Duration, logger logging.Logger, metrics *TxnManagerMetrics) TxnManager { | ||
| return &txnManager{ | ||
| ethClient: ethClient, | ||
| wallet: wallet, | ||
| numConfirmations: numConfirmations, | ||
| requestChan: make(chan *TxnRequest, queueSize), | ||
| logger: logger.With("component", "TxnManager"), | ||
| receiptChan: make(chan *ReceiptOrErr, queueSize), | ||
| queueSize: queueSize, | ||
| txnBroadcastTimeout: txnBroadcastTimeout, | ||
| txnRefreshInterval: txnRefreshInterval, | ||
| metrics: metrics, | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -128,7 +133,7 @@ func (t *txnManager) Start(ctx context.Context) { | |
| t.metrics.UpdateGasUsed(receipt.GasUsed) | ||
| } | ||
| } | ||
| t.metrics.ObserveLatency("total", float64(time.Since(req.requestedAt).Milliseconds())) | ||
| } | ||
| } | ||
| }() | ||
|
|
@@ -141,7 +146,7 @@ func (t *txnManager) Start(ctx context.Context) { | |
| func (t *txnManager) ProcessTransaction(ctx context.Context, req *TxnRequest) error { | ||
| t.mu.Lock() | ||
| defer t.mu.Unlock() | ||
| t.logger.Debug("new transaction", "tag", req.Tag, "nonce", req.Tx.Nonce(), "gasFeeCap", req.Tx.GasFeeCap(), "gasTipCap", req.Tx.GasTipCap()) | ||
|
|
||
| var txn *types.Transaction | ||
| var txID walletsdk.TxID | ||
|
|
@@ -164,13 +169,13 @@ func (t *txnManager) ProcessTransaction(ctx context.Context, req *TxnRequest) er | |
| didTimeout = urlErr.Timeout() | ||
| } | ||
| if didTimeout || errors.Is(err, context.DeadlineExceeded) { | ||
| t.logger.Warn("failed to send txn due to timeout", "tag", req.Tag, "hash", txn.Hash().Hex(), "numRetries", retryFromFailure, "maxRetry", maxSendTransactionRetry, "err", err) | ||
| retryFromFailure++ | ||
| continue | ||
| } else if err != nil { | ||
| return fmt.Errorf("failed to send txn (%s) %s: %w", req.Tag, txn.Hash().Hex(), err) | ||
| } else { | ||
| t.logger.Debug("successfully sent txn", "tag", req.Tag, "txID", txID, "txHash", txn.Hash().Hex()) | ||
| break | ||
| } | ||
| } | ||
|
|
@@ -183,6 +188,7 @@ func (t *txnManager) ProcessTransaction(ctx context.Context, req *TxnRequest) er | |
| req.txAttempts = append(req.txAttempts, &transaction{ | ||
| TxID: txID, | ||
| Transaction: txn, | ||
| requestedAt: time.Now(), | ||
| }) | ||
|
|
||
| t.requestChan <- req | ||
|
|
@@ -194,8 +200,31 @@ func (t *txnManager) ReceiptChan() chan *ReceiptOrErr { | |
| return t.receiptChan | ||
| } | ||
|
|
||
| // ensureAnyTransactionBroadcasted waits until all given transactions are broadcasted to the network. | ||
| func (t *txnManager) ensureAnyTransactionBroadcasted(ctx context.Context, txs []*transaction) error { | ||
| queryTicker := time.NewTicker(queryTickerDuration) | ||
| defer queryTicker.Stop() | ||
|
|
||
| for { | ||
| for _, tx := range txs { | ||
| _, err := t.wallet.GetTransactionReceipt(ctx, tx.TxID) | ||
| if err == nil || errors.Is(err, walletsdk.ErrReceiptNotYetAvailable) { | ||
| t.metrics.ObserveLatency("broadcasted", float64(time.Since(tx.requestedAt).Milliseconds())) | ||
| return nil | ||
| } | ||
| } | ||
|
|
||
| // Wait for the next round. | ||
| select { | ||
| case <-ctx.Done(): | ||
| return ctx.Err() | ||
| case <-queryTicker.C: | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (t *txnManager) ensureAnyTransactionEvaled(ctx context.Context, txs []*transaction) (*types.Receipt, error) { | ||
| queryTicker := time.NewTicker(queryTickerDuration) | ||
| defer queryTicker.Stop() | ||
| var receipt *types.Receipt | ||
| var err error | ||
|
|
@@ -212,25 +241,32 @@ func (t *txnManager) ensureAnyTransactionEvaled(ctx context.Context, txs []*tran | |
| chainTip, err := t.ethClient.BlockNumber(ctx) | ||
| if err == nil { | ||
| if receipt.BlockNumber.Uint64()+uint64(t.numConfirmations) > chainTip { | ||
| t.logger.Debug("transaction has been mined but don't have enough confirmations at current chain tip", "txnBlockNumber", receipt.BlockNumber.Uint64(), "numConfirmations", t.numConfirmations, "chainTip", chainTip) | ||
| break | ||
| } else { | ||
| return receipt, nil | ||
| } | ||
| } else { | ||
| t.logger.Debug("failed to get chain tip while waiting for transaction to mine", "err", err) | ||
| } | ||
| } | ||
|
|
||
| if errors.Is(err, ethereum.NotFound) || errors.Is(err, walletsdk.ErrReceiptNotYetAvailable) { | ||
| t.logger.Debug("Transaction not yet mined", "txID", txID, "txHash", tx.Hash().Hex(), "err", err) | ||
| } else if errors.Is(err, walletsdk.ErrTransactionFailed) { | ||
| t.logger.Debug("Transaction failed", "txID", txID, "txHash", tx.Hash().Hex(), "err", err) | ||
| delete(txnsToQuery, txID) | ||
| } else if errors.Is(err, walletsdk.ErrNotYetBroadcasted) { | ||
| t.logger.Error("Transaction has not been broadcasted to network but attempted to retrieve receipt", "err", err) | ||
| } else { | ||
| t.logger.Debug("Transaction receipt retrieval failed", "err", err) | ||
| } | ||
| } | ||
|
|
||
| if len(txnsToQuery) == 0 { | ||
| return nil, fmt.Errorf("all transactions failed") | ||
| } | ||
|
|
||
| // Wait for the next round. | ||
| select { | ||
| case <-ctx.Done(): | ||
|
|
@@ -251,10 +287,40 @@ func (t *txnManager) monitorTransaction(ctx context.Context, req *TxnRequest) (* | |
| var err error | ||
|
|
||
| rpcCallAttempt := func() error { | ||
| t.logger.Debug("monitoring transaction", "txHash", req.Tx.Hash().Hex(), "tag", req.Tag, "nonce", req.Tx.Nonce()) | ||
|
|
||
| ctxWithTimeout, cancelBroadcastTimeout := context.WithTimeout(ctx, t.txnBroadcastTimeout) | ||
| defer cancelBroadcastTimeout() | ||
|
|
||
| // Ensure transactions are broadcasted to the network before querying the receipt. | ||
| // This is to avoid querying the receipt of a transaction that hasn't been broadcasted yet. | ||
| // For example, when Fireblocks wallet is used, there may be delays in broadcasting the transaction due to latency from cosigning and MPC operations. | ||
| err = t.ensureAnyTransactionBroadcasted(ctxWithTimeout, req.txAttempts) | ||
|
There was a problem hiding this comment. We don't retry here because there wouldn't be any advantage over just waiting longer? There was a problem hiding this comment. Yeah as explained in this comment |
||
| if err != nil && errors.Is(err, context.DeadlineExceeded) { | ||
ian-shim marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| t.logger.Warn("transaction not broadcasted within timeout", "tag", req.Tag, "txHash", req.Tx.Hash().Hex(), "nonce", req.Tx.Nonce()) | ||
| fireblocksWallet, ok := t.wallet.(interface { | ||
| CancelTransactionBroadcast(ctx context.Context, txID walletsdk.TxID) (bool, error) | ||
| }) | ||
| if ok { | ||
| // Consider these transactions failed as they haven't been broadcasted within timeout. | ||
| // Cancel these transactions to avoid blocking the next transactions. | ||
| for _, tx := range req.txAttempts { | ||
| cancelled, err := fireblocksWallet.CancelTransactionBroadcast(ctx, tx.TxID) | ||
| if err != nil { | ||
| t.logger.Warn("failed to cancel Fireblocks transaction broadcast", "txID", tx.TxID, "err", err) | ||
| } else if cancelled { | ||
| t.logger.Info("cancelled Fireblocks transaction broadcast because it didn't get broadcasted within timeout", "txID", tx.TxID, "timeout", t.txnBroadcastTimeout.String()) | ||
| } | ||
| } | ||
| } | ||
| return ErrTransactionNotBroadcasted | ||
| } else if err != nil { | ||
| t.logger.Error("unexpected error while waiting for Fireblocks transaction to broadcast", "txHash", req.Tx.Hash().Hex(), "err", err) | ||
| return err | ||
| } | ||
|
|
||
| ctxWithTimeout, cancelEvaluationTimeout := context.WithTimeout(ctx, t.txnRefreshInterval) | ||
| defer cancelEvaluationTimeout() | ||
| receipt, err = t.ensureAnyTransactionEvaled( | ||
| ctxWithTimeout, | ||
| req.txAttempts, | ||
|
|
@@ -272,38 +338,38 @@ func (t *txnManager) monitorTransaction(ctx context.Context, req *TxnRequest) (* | |
|
|
||
| if errors.Is(err, context.DeadlineExceeded) { | ||
| if receipt != nil { | ||
| t.logger.Warn("transaction has been mined, but hasn't accumulated the required number of confirmations", "tag", req.Tag, "txHash", req.Tx.Hash().Hex(), "nonce", req.Tx.Nonce()) | ||
| continue | ||
| } | ||
| t.logger.Warn("transaction not mined within timeout, resending with higher gas price", "tag", req.Tag, "txHash", req.Tx.Hash().Hex(), "nonce", req.Tx.Nonce()) | ||
| newTx, err := t.speedUpTxn(ctx, req.Tx, req.Tag) | ||
| if err != nil { | ||
| t.logger.Error("failed to speed up transaction", "err", err) | ||
| t.metrics.IncrementTxnCount("failure") | ||
| return nil, err | ||
| } | ||
| txID, err := t.wallet.SendTransaction(ctx, newTx) | ||
| if err != nil { | ||
| if retryFromFailure >= maxSendTransactionRetry { | ||
| t.logger.Warn("failed to send txn - retries exhausted", "tag", req.Tag, "txn", req.Tx.Hash().Hex(), "attempt", retryFromFailure, "maxRetry", maxSendTransactionRetry, "err", err) | ||
| t.metrics.IncrementTxnCount("failure") | ||
| return nil, err | ||
| } else { | ||
| t.logger.Warn("failed to send txn - retrying", "tag", req.Tag, "txn", req.Tx.Hash().Hex(), "attempt", retryFromFailure, "maxRetry", maxSendTransactionRetry, "err", err) | ||
| } | ||
| retryFromFailure++ | ||
| continue | ||
| } | ||
|
|
||
| t.logger.Debug("successfully sent txn", "tag", req.Tag, "txID", txID, "txHash", newTx.Hash().Hex()) | ||
| req.Tx = newTx | ||
| req.txAttempts = append(req.txAttempts, &transaction{ | ||
| TxID: txID, | ||
| Transaction: newTx, | ||
| }) | ||
| numSpeedUps++ | ||
| } else { | ||
| t.logger.Error("transaction failed", "tag", req.Tag, "txHash", req.Tx.Hash().Hex(), "err", err) | ||
| t.metrics.IncrementTxnCount("failure") | ||
| return nil, err | ||
| } | ||
|
|
@@ -335,7 +401,7 @@ func (t *txnManager) speedUpTxn(ctx context.Context, tx *types.Transaction, tag | |
| newGasFeeCap = increasedGasFeeCap | ||
| } | ||
|
|
||
| t.logger.Info("increasing gas price", "tag", tag, "txHash", tx.Hash().Hex(), "nonce", tx.Nonce(), "prevGasTipCap", prevGasTipCap, "prevGasFeeCap", prevGasFeeCap, "newGasTipCap", newGasTipCap, "newGasFeeCap", newGasFeeCap) | ||
| return t.ethClient.UpdateGas(ctx, tx, tx.Value(), newGasTipCap, newGasFeeCap) | ||
| } | ||
|
|
||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Checking error types seems like it's a bit brittle. I guess there's no other option here?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We can change the implementation in eigensdk, but I think this signature should be kept.
One thing we can do is make the error type more expressive and return Fireblocks status as part of the error, but not sure if that will be less brittle
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Maybe add a comment: "Before the transaction is broadcast, this wallet method will return error of type
walletsdk.Whatever. Once the transaction has been broadcast, it will either successfully return the receipt or return an error of typewalletsdk.ErrReceiptNotYetAvailable."