Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update TxStore to use parent context created at initialization #10735

Merged
merged 10 commits into from
Oct 4, 2023
Merged

Update TxStore to use parent context created at initialization #10735

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
855a34e
Removed pg.Opts and updated TxStore methods to use both the provided …
amit-momin Sep 22, 2023
61e6bc3
Updated upstream contexts
amit-momin Sep 22, 2023
510d34d
Updated TxStore to maintain one context in scope at a time
amit-momin Sep 25, 2023
cd7bde3
Updated Fluxmonitor context usage
amit-momin Sep 25, 2023
cb164f6
Wired remaining TxStore methods to upstream contexts
amit-momin Sep 26, 2023
7e7d086
Merge branch 'develop' into txm/BCI-1585
amit-momin Sep 27, 2023
c2919ae
Merge branch 'develop' into txm/BCI-1585
amit-momin Oct 2, 2023
97ca72e
Merge branch 'develop' into txm/BCI-1585
amit-momin Oct 3, 2023
8d9fa20
Merge branch 'develop' into txm/BCI-1585
amit-momin Oct 3, 2023
07fb236
Merge branch 'develop' into txm/BCI-1585
prashantkumar1982 Oct 4, 2023
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
18 changes: 11 additions & 7 deletions common/txmgr/broadcaster.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -444,12 +444,12 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) proc
for {
maxInFlightTransactions := eb.txConfig.MaxInFlight()
if maxInFlightTransactions > 0 {
nUnconfirmed, err := eb.txStore.CountUnconfirmedTransactions(fromAddress, eb.chainID)
nUnconfirmed, err := eb.txStore.CountUnconfirmedTransactions(ctx, fromAddress, eb.chainID)
if err != nil {
return true, errors.Wrap(err, "CountUnconfirmedTransactions failed")
}
if nUnconfirmed >= maxInFlightTransactions {
nUnstarted, err := eb.txStore.CountUnstartedTransactions(fromAddress, eb.chainID)
nUnstarted, err := eb.txStore.CountUnstartedTransactions(ctx, fromAddress, eb.chainID)
if err != nil {
return true, errors.Wrap(err, "CountUnstartedTransactions failed")
}
Expand Down Expand Up @@ -477,7 +477,7 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) proc
return retryable, errors.Wrap(err, "processUnstartedTxs failed on NewAttempt")
}

if err := eb.txStore.UpdateTxUnstartedToInProgress(etx, &a); errors.Is(err, ErrTxRemoved) {
if err := eb.txStore.UpdateTxUnstartedToInProgress(ctx, etx, &a); errors.Is(err, ErrTxRemoved) {
eb.logger.Debugw("tx removed", "txID", etx.ID, "subject", etx.Subject)
continue
} else if err != nil {
Expand All @@ -493,7 +493,7 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) proc
// handleInProgressTx checks if there is any transaction
// in_progress and if so, finishes the job
func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) handleAnyInProgressTx(ctx context.Context, fromAddress ADDR) (err error, retryable bool) {
etx, err := eb.txStore.GetTxInProgress(fromAddress)
etx, err := eb.txStore.GetTxInProgress(ctx, fromAddress)
if err != nil {
return errors.Wrap(err, "handleAnyInProgressTx failed"), true
}
Expand Down Expand Up @@ -668,8 +668,10 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) hand
// Finds next transaction in the queue, assigns a sequence, and moves it to "in_progress" state ready for broadcast.
// Returns nil if no transactions are in queue
func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) nextUnstartedTransactionWithSequence(fromAddress ADDR) (*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], error) {
ctx, cancel := eb.chStop.NewCtx()
defer cancel()
etx := &txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]{}
if err := eb.txStore.FindNextUnstartedTransactionFromAddress(etx, fromAddress, eb.chainID); err != nil {
if err := eb.txStore.FindNextUnstartedTransactionFromAddress(ctx, etx, fromAddress, eb.chainID); err != nil {
if errors.Is(err, sql.ErrNoRows) {
// Finish. No more transactions left to process. Hoorah!
return nil, nil
Expand Down Expand Up @@ -722,14 +724,16 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) tryA
}

func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) saveTryAgainAttempt(ctx context.Context, lgr logger.Logger, etx txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], attempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], replacementAttempt txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], initialBroadcastAt time.Time, newFee FEE, newFeeLimit uint32) (err error, retyrable bool) {
if err = eb.txStore.SaveReplacementInProgressAttempt(attempt, &replacementAttempt); err != nil {
if err = eb.txStore.SaveReplacementInProgressAttempt(ctx, attempt, &replacementAttempt); err != nil {
return errors.Wrap(err, "tryAgainWithNewFee failed"), true
}
lgr.Debugw("Bumped fee on initial send", "oldFee", attempt.TxFee.String(), "newFee", newFee.String(), "newFeeLimit", newFeeLimit)
return eb.handleInProgressTx(ctx, etx, replacementAttempt, initialBroadcastAt)
}

func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) saveFatallyErroredTransaction(lgr logger.Logger, etx *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) error {
ctx, cancel := eb.chStop.NewCtx()
defer cancel()
if etx.State != TxInProgress {
return errors.Errorf("can only transition to fatal_error from in_progress, transaction is currently %s", etx.State)
}
Expand All @@ -756,7 +760,7 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) save
return errors.Wrap(err, "failed to resume pipeline")
}
}
return eb.txStore.UpdateTxFatalError(etx)
return eb.txStore.UpdateTxFatalError(ctx, etx)
}

func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) getNextSequence(address ADDR) (sequence SEQ, err error) {
Expand Down
37 changes: 18 additions & 19 deletions common/txmgr/confirmer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
"github.com/smartcontractkit/chainlink/v2/common/types"
"github.com/smartcontractkit/chainlink/v2/core/chains/evm/label"
"github.com/smartcontractkit/chainlink/v2/core/logger"
"github.com/smartcontractkit/chainlink/v2/core/services/pg"
"github.com/smartcontractkit/chainlink/v2/core/utils"
)

Expand Down Expand Up @@ -279,7 +278,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) pro

ec.lggr.Debugw("processHead start", "headNum", head.BlockNumber(), "id", "confirmer")

if err := ec.txStore.SetBroadcastBeforeBlockNum(head.BlockNumber(), ec.chainID); err != nil {
if err := ec.txStore.SetBroadcastBeforeBlockNum(ctx, head.BlockNumber(), ec.chainID); err != nil {
return errors.Wrap(err, "SetBroadcastBeforeBlockNum failed")
}
if err := ec.CheckConfirmedMissingReceipt(ctx); err != nil {
Expand Down Expand Up @@ -340,7 +339,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) pro
//
// This scenario might sound unlikely but has been observed to happen multiple times in the wild on Polygon.
func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) CheckConfirmedMissingReceipt(ctx context.Context) (err error) {
attempts, err := ec.txStore.FindTxAttemptsConfirmedMissingReceipt(ec.chainID)
attempts, err := ec.txStore.FindTxAttemptsConfirmedMissingReceipt(ctx, ec.chainID)
if err != nil {
return err
}
Expand All @@ -351,7 +350,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Che
txCodes, txErrs, broadcastTime, txIDs, err := ec.client.BatchSendTransactions(ctx, attempts, int(ec.chainConfig.RPCDefaultBatchSize()), ec.lggr)
// update broadcast times before checking additional errors
if len(txIDs) > 0 {
if updateErr := ec.txStore.UpdateBroadcastAts(broadcastTime, txIDs); updateErr != nil {
if updateErr := ec.txStore.UpdateBroadcastAts(ctx, broadcastTime, txIDs); updateErr != nil {
err = fmt.Errorf("%w: failed to update broadcast time: %w", err, updateErr)
}
}
Expand All @@ -369,7 +368,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Che

txIDsToUnconfirm = append(txIDsToUnconfirm, attempts[idx].TxID)
}
err = ec.txStore.UpdateTxsUnconfirmed(txIDsToUnconfirm)
err = ec.txStore.UpdateTxsUnconfirmed(ctx, txIDsToUnconfirm)

if err != nil {
return err
Expand All @@ -379,7 +378,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Che

// CheckForReceipts finds attempts that are still pending and checks to see if a receipt is present for the given block number
func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) CheckForReceipts(ctx context.Context, blockNum int64) error {
attempts, err := ec.txStore.FindTxAttemptsRequiringReceiptFetch(ec.chainID)
attempts, err := ec.txStore.FindTxAttemptsRequiringReceiptFetch(ctx, ec.chainID)
if err != nil {
return errors.Wrap(err, "FindTxAttemptsRequiringReceiptFetch failed")
}
Expand Down Expand Up @@ -421,11 +420,11 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Che
}
}

if err := ec.txStore.MarkAllConfirmedMissingReceipt(ec.chainID); err != nil {
if err := ec.txStore.MarkAllConfirmedMissingReceipt(ctx, ec.chainID); err != nil {
return errors.Wrap(err, "unable to mark txes as 'confirmed_missing_receipt'")
}

if err := ec.txStore.MarkOldTxesMissingReceiptAsErrored(blockNum, ec.chainConfig.FinalityDepth(), ec.chainID); err != nil {
if err := ec.txStore.MarkOldTxesMissingReceiptAsErrored(ctx, blockNum, ec.chainConfig.FinalityDepth(), ec.chainID); err != nil {
return errors.Wrap(err, "unable to confirm buried unconfirmed txes")
}
return nil
Expand Down Expand Up @@ -489,7 +488,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) fet
if err != nil {
return errors.Wrap(err, "batchFetchReceipts failed")
}
if err := ec.txStore.SaveFetchedReceipts(receipts, ec.chainID); err != nil {
if err := ec.txStore.SaveFetchedReceipts(ctx, receipts, ec.chainID); err != nil {
return errors.Wrap(err, "saveFetchedReceipts failed")
}
promNumConfirmedTxs.WithLabelValues(ec.chainID.String()).Add(float64(len(receipts)))
Expand All @@ -511,7 +510,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) get
func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) batchFetchReceipts(ctx context.Context, attempts []txmgrtypes.TxAttempt[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], blockNum int64) (receipts []R, err error) {
// Metadata is required to determine whether a tx is forwarded or not.
if ec.txConfig.ForwardersEnabled() {
err = ec.txStore.PreloadTxes(attempts)
err = ec.txStore.PreloadTxes(ctx, attempts)
if err != nil {
return nil, errors.Wrap(err, "Confirmer#batchFetchReceipts error loading txs for attempts")
}
Expand Down Expand Up @@ -648,7 +647,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) reb

lggr.Debugw("Rebroadcasting transaction", "nPreviousAttempts", len(etx.TxAttempts), "fee", attempt.TxFee)

if err := ec.txStore.SaveInProgressAttempt(&attempt); err != nil {
if err := ec.txStore.SaveInProgressAttempt(ctx, &attempt); err != nil {
return errors.Wrap(err, "saveInProgressAttempt failed")
}

Expand Down Expand Up @@ -687,7 +686,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) han
func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) FindTxsRequiringRebroadcast(ctx context.Context, lggr logger.Logger, address ADDR, blockNum, gasBumpThreshold, bumpDepth int64, maxInFlightTransactions uint32, chainID CHAIN_ID) (etxs []*txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE], err error) {
// NOTE: These two queries could be combined into one using union but it
// becomes harder to read and difficult to test in isolation. KISS principle
etxInsufficientFunds, err := ec.txStore.FindTxsRequiringResubmissionDueToInsufficientFunds(address, chainID, pg.WithParentCtx(ctx))
etxInsufficientFunds, err := ec.txStore.FindTxsRequiringResubmissionDueToInsufficientFunds(ctx, address, chainID)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -825,7 +824,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) han
ec.lggr.Warnw("Got terminally underpriced error for gas bump, this should never happen unless the remote RPC node changed its configuration on the fly, or you are using multiple RPC nodes with different minimum gas price requirements. This is not recommended", "err", sendError, "attempt", attempt)
// "Lazily" load attempts here since the overwhelmingly common case is
// that we don't need them unless we enter this path
if err := ec.txStore.LoadTxAttempts(&etx, pg.WithParentCtx(ctx)); err != nil {
if err := ec.txStore.LoadTxAttempts(ctx, &etx); err != nil {
return errors.Wrap(err, "failed to load TxAttempts while bumping on terminally underpriced error")
}
if len(etx.TxAttempts) == 0 {
Expand All @@ -850,7 +849,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) han
"replacementAttempt", replacementAttempt,
).Errorf("gas price was rejected by the node for being too low. Node returned: '%s'", sendError.Error())

if err := ec.txStore.SaveReplacementInProgressAttempt(attempt, &replacementAttempt); err != nil {
if err := ec.txStore.SaveReplacementInProgressAttempt(ctx, attempt, &replacementAttempt); err != nil {
return errors.Wrap(err, "saveReplacementInProgressAttempt failed")
}
return ec.handleInProgressAttempt(ctx, lggr, etx, replacementAttempt, blockHeight)
Expand Down Expand Up @@ -882,11 +881,11 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) han
return ec.txStore.SaveConfirmedMissingReceiptAttempt(ctx, timeout, &attempt, now)
case clienttypes.InsufficientFunds:
timeout := ec.dbConfig.DefaultQueryTimeout()
return ec.txStore.SaveInsufficientFundsAttempt(timeout, &attempt, now)
return ec.txStore.SaveInsufficientFundsAttempt(ctx, timeout, &attempt, now)
case clienttypes.Successful:
lggr.Debugw("Successfully broadcast transaction", "txAttemptID", attempt.ID, "txHash", attempt.Hash.String())
timeout := ec.dbConfig.DefaultQueryTimeout()
return ec.txStore.SaveSentAttempt(timeout, &attempt, now)
return ec.txStore.SaveSentAttempt(ctx, timeout, &attempt, now)
case clienttypes.Unknown:
// Every error that doesn't fall under one of the above categories will be treated as Unknown.
fallthrough
Expand Down Expand Up @@ -921,7 +920,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Ens
} else {
ec.nConsecutiveBlocksChainTooShort = 0
}
etxs, err := ec.txStore.FindTransactionsConfirmedInBlockRange(head.BlockNumber(), head.EarliestHeadInChain().BlockNumber(), ec.chainID)
etxs, err := ec.txStore.FindTransactionsConfirmedInBlockRange(ctx, head.BlockNumber(), head.EarliestHeadInChain().BlockNumber(), ec.chainID)
if err != nil {
return errors.Wrap(err, "findTransactionsConfirmedInBlockRange failed")
}
Expand Down Expand Up @@ -1015,7 +1014,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) mar
ec.lggr.Infow(fmt.Sprintf("Re-org detected. Rebroadcasting transaction %s which may have been re-org'd out of the main chain", attempt.Hash.String()), logValues...)

// Put it back in progress and delete all receipts (they do not apply to the new chain)
err := ec.txStore.UpdateTxForRebroadcast(etx, attempt)
err := ec.txStore.UpdateTxForRebroadcast(ec.ctx, etx, attempt)
return errors.Wrap(err, "markForRebroadcast failed")
}

Expand All @@ -1034,7 +1033,7 @@ func (ec *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) For

for _, seq := range seqs {

etx, err := ec.txStore.FindTxWithSequence(address, seq)
etx, err := ec.txStore.FindTxWithSequence(context.TODO(), address, seq)
if err != nil {
return errors.Wrap(err, "ForceRebroadcast failed")
}
Expand Down
45 changes: 18 additions & 27 deletions common/txmgr/mocks/tx_manager.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 3 additions & 1 deletion common/txmgr/reaper.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,8 @@ func (r *Reaper[CHAIN_ID]) SetLatestBlockNum(latestBlockNum int64) {

// ReapTxes deletes old txes
func (r *Reaper[CHAIN_ID]) ReapTxes(headNum int64) error {
ctx, cancel := utils.StopChan(r.chStop).NewCtx()
defer cancel()
threshold := r.txConfig.ReaperThreshold()
if threshold == 0 {
r.log.Debug("Transactions.ReaperThreshold set to 0; skipping ReapTxes")
Expand All @@ -108,7 +110,7 @@ func (r *Reaper[CHAIN_ID]) ReapTxes(headNum int64) error {

r.log.Debugw(fmt.Sprintf("reaping old txes created before %s", timeThreshold.Format(time.RFC3339)), "ageThreshold", threshold, "timeThreshold", timeThreshold, "minBlockNumberToKeep", minBlockNumberToKeep)

if err := r.store.ReapTxHistory(minBlockNumberToKeep, timeThreshold, r.chainID); err != nil {
if err := r.store.ReapTxHistory(ctx, minBlockNumberToKeep, timeThreshold, r.chainID); err != nil {
return err
}

Expand Down
Loading
Loading