txmgr

package
v0.0.0-...-1c5c8e4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Dec 10, 2025 License: MIT Imports: 31 Imported by: 17

Documentation

Index

Constants

View Source 
const (
	// InFlightTransactionRecheckInterval controls how often the Broadcaster
	// will poll the unconfirmed queue to see if it is allowed to send another
	// transaction
	InFlightTransactionRecheckInterval = 1 * time.Second

	// TransmitCheckTimeout controls the maximum amount of time that will be
	// spent on the transmit check.
	TransmitCheckTimeout = 2 * time.Second
)
View Source 
const (
	TxUnstarted               = types.TxState("unstarted")
	TxInProgress              = types.TxState("in_progress")
	TxFatalError              = types.TxState("fatal_error")
	TxUnconfirmed             = types.TxState("unconfirmed")
	TxConfirmed               = types.TxState("confirmed")
	TxConfirmedMissingReceipt = types.TxState("confirmed_missing_receipt")
	TxFinalized               = types.TxState("finalized")
)
View Source 
const (
	// pollInterval is the maximum amount of time in addition to
	// TxResendAfterThreshold that we will wait before resending an attempt
	DefaultResenderPollInterval = 5 * time.Second
)

Variables

View Source 
var ErrTxRemoved = errors.New("tx removed")

Functions

func NewQueueingTxStrategy 

func NewQueueingTxStrategy(subject uuid.UUID, queueSize uint32) (strategy types.TxStrategy)

NewQueueingTxStrategy creates a new TxStrategy that drops the oldest transactions after the queue size is exceeded if a queue size is specified, and otherwise does not drop transactions.

func NewSendEveryStrategy 

func NewSendEveryStrategy() types.TxStrategy

NewSendEveryStrategy creates a new TxStrategy that does not drop transactions.

Types

type Broadcaster 

type Broadcaster[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, SEQ chains.Sequence, FEE fees.Fee] struct {
	services.StateMachine

	types.TxAttemptBuilder[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]
	// contains filtered or unexported fields
}

Broadcaster monitors txes for transactions that need to be broadcast, assigns sequences and ensures that at least one node somewhere has received the transaction successfully.

This does not guarantee delivery! A whole host of other things can subsequently go wrong such as transactions being evicted from the mempool, nodes going offline etc. Responsibility for ensuring eventual inclusion into the chain falls on the shoulders of the confirmer.

What Broadcaster does guarantee is: - a monotonic series of increasing sequences for txes that can all eventually be confirmed if you retry enough times - transition of txes out of unstarted into either fatal_error or unconfirmed - existence of a saved tx_attempt

func NewBroadcaster 

func NewBroadcaster[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, SEQ chains.Sequence, FEE fees.Fee](
	txStore types.TransactionStore[ADDR, CID, THASH, BHASH, SEQ, FEE],
	client types.TransactionClient[CID, ADDR, THASH, BHASH, SEQ, FEE],
	config types.BroadcasterChainConfig,
	feeConfig types.BroadcasterFeeConfig,
	txConfig types.BroadcasterTransactionsConfig,
	listenerConfig types.BroadcasterListenerConfig,
	keystore types.KeyStore[ADDR],
	txAttemptBuilder types.TxAttemptBuilder[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE],
	sequenceTracker types.SequenceTracker[ADDR, SEQ],
	lggr logger.Logger,
	checkerFactory TransmitCheckerFactory[CID, ADDR, THASH, BHASH, SEQ, FEE],
	autoSyncSequence bool,
	chainType string,
	metrics broadcasterMetrics,
) *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Close 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Close() error

Close closes the Broadcaster

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) HealthReport 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) HealthReport() map[string]error

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Name 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Name() string

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) ProcessUnstartedTxs 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) ProcessUnstartedTxs(ctx context.Context, addr ADDR) (retryable bool, err error)

ProcessUnstartedTxs picks up and handles all txes in the queue revive:disable:error-return

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) SetResumeCallback 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) SetResumeCallback(callback ResumeCallback)

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Start 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Start(ctx context.Context) error

Start starts Broadcaster service. The provided context can be used to terminate Start sequence.

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Trigger 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Trigger(addr ADDR)

Trigger forces the monitor for a particular address to recheck for new txes Logs error and does nothing if address was not registered on startup

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) XXXTestCloseInternal 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) XXXTestCloseInternal() error

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) XXXTestDisableUnstartedTxAutoProcessing 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) XXXTestDisableUnstartedTxAutoProcessing()

func (*Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) XXXTestStartInternal 

func (eb *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) XXXTestStartInternal(ctx context.Context) error

type Confirmer 

type Confirmer[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH chains.Hashable,
	BHASH chains.Hashable, R types.ChainReceipt[THASH, BHASH], SEQ chains.Sequence, FEE fees.Fee,
] struct {
	services.StateMachine

	types.TxAttemptBuilder[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]
	// contains filtered or unexported fields
}

Confirmer is a broad service which performs four different tasks in sequence on every new longest chain Step 1: Mark that all currently pending transaction attempts were broadcast before this block Step 2: Check pending transactions for confirmation and confirmed transactions for re-org Step 3: Check if any pending transaction is stuck in the mempool. If so, mark for purge. Step 4: See if any transactions have exceeded the gas bumping block threshold and, if so, bump them

func NewConfirmer 

func NewConfirmer[
	CHAIN_ID chains.ID,
	HEAD chains.Head[BLOCK_HASH],
	ADDR chains.Hashable,
	TX_HASH chains.Hashable,
	BLOCK_HASH chains.Hashable,
	R types.ChainReceipt[TX_HASH, BLOCK_HASH],
	SEQ chains.Sequence,
	FEE fees.Fee,
](
	txStore types.TxStore[ADDR, CHAIN_ID, TX_HASH, BLOCK_HASH, R, SEQ, FEE],
	client types.TxmClient[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE],
	feeConfig types.ConfirmerFeeConfig,
	txConfig types.ConfirmerTransactionsConfig,
	dbConfig types.ConfirmerDatabaseConfig,
	keystore types.KeyStore[ADDR],
	txAttemptBuilder types.TxAttemptBuilder[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
	lggr logger.Logger,
	isReceiptNil func(R) bool,
	stuckTxDetector types.StuckTxDetector[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
	metrics confimerMetrics,
) *Confirmer[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) CheckForConfirmation 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) CheckForConfirmation(ctx context.Context, head chains.Head[BHASH]) error

CheckForConfirmation fetches the mined transaction count for each enabled address and marks transactions with a lower sequence as confirmed and ones with equal or higher sequence as unconfirmed

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Close 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Close() error

Close is a comment to appease the linter

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxsRequiringRebroadcast 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxsRequiringRebroadcast(ctx context.Context, lggr logger.Logger, address ADDR, blockNum, gasBumpThreshold, bumpDepth int64, maxInFlightTransactions uint32, chainID CID) (etxs []*types.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

FindTxsRequiringRebroadcast returns attempts that hit insufficient native tokens, and attempts that need bumping, in sequence ASC order

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ForceRebroadcast 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ForceRebroadcast(ctx context.Context, seqs []SEQ, fee FEE, address ADDR, overrideGasLimit uint64) error

ForceRebroadcast sends a transaction for every sequence in the given sequence range at the given gas price. If an tx exists for this sequence, we re-send the existing tx with the supplied parameters. If an tx doesn't exist for this sequence, we send a zero transaction. This operates completely orthogonal to the normal Confirmer and can result in untracked attempts! Only for emergency usage. This is in case of some unforeseen scenario where the node is refusing to release the lock. KISS.

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) HealthReport 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) HealthReport() map[string]error

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Name 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Name() string

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ProcessHead 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ProcessHead(ctx context.Context, head chains.Head[BHASH]) error

ProcessHead takes all required transactions for the confirmer on a new head

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ProcessIncludedTxs 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ProcessIncludedTxs(ctx context.Context, includedTxs []*types.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], head chains.Head[BHASH]) error

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ProcessReorgTxs 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ProcessReorgTxs(ctx context.Context, reorgTxs []*types.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], head chains.Head[BHASH]) error

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ProcessStuckTransactions 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) ProcessStuckTransactions(ctx context.Context, blockNum int64) error

Determines if any of the unconfirmed transactions are terminally stuck for each enabled address If any transaction is found to be terminally stuck, this method sends an empty attempt with bumped gas in an attempt to purge the stuck transaction

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) RebroadcastWhereNecessary 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) RebroadcastWhereNecessary(ctx context.Context, blockHeight int64) error

RebroadcastWhereNecessary bumps gas or resends transactions that were previously out-of-funds

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) SetResumeCallback 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) SetResumeCallback(callback ResumeCallback)

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Start 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Start(ctx context.Context) error

Start is a comment to appease the linter

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestCloseInternal 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestCloseInternal() error

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestSetClient 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestSetClient(client types.TxmClient[CID, ADDR, THASH, BHASH, R, SEQ, FEE])

func (*Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestStartInternal 

func (ec *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestStartInternal() error

type DropOldestStrategy 

type DropOldestStrategy struct {
	// contains filtered or unexported fields
}

DropOldestStrategy will send the newest N transactions, older ones will be removed from the queue

func NewDropOldestStrategy 

func NewDropOldestStrategy(subject uuid.UUID, queueSize uint32) DropOldestStrategy

NewDropOldestStrategy creates a new TxStrategy that drops the oldest transactions after the queue size is exceeded.

func (DropOldestStrategy) PruneQueue 

func (s DropOldestStrategy) PruneQueue(ctx context.Context, pruneService types.UnstartedTxQueuePruner) (ids []int64, err error)

func (DropOldestStrategy) Subject 

func (s DropOldestStrategy) Subject() uuid.NullUUID

type FeeParts 

type FeeParts struct {
	GasUsed           uint64
	EffectiveGasPrice *big.Int
	L1Fee             *big.Int
}

type NewErrorClassifier 

type NewErrorClassifier func(err error) txmgrtypes.ErrorClassifier

type NotEnabledError 

type NotEnabledError[ADDR chains.Hashable] struct {
	FromAddress ADDR
	Err         error
}

func (*NotEnabledError[ADDR]) As 

func (e *NotEnabledError[ADDR]) As(err error) bool

func (NotEnabledError[ADDR]) Error 

func (e NotEnabledError[ADDR]) Error() string

func (NotEnabledError[ADDR]) Is 

func (e NotEnabledError[ADDR]) Is(err error) bool

func (NotEnabledError[ADDR]) Unwrap 

func (e NotEnabledError[ADDR]) Unwrap() error

type NullTxManager deprecated 

type NullTxManager[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH, BHASH chains.Hashable, SEQ chains.Sequence, FEE fees.Fee] struct {
	ErrMsg string
}

Deprecated: use txmgrtest.ErrTxManager

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) CalculateFee 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) CalculateFee(feeParts FeeParts) *big.Int

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Close 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Close() error

Close does noop for NullTxManager.

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) CountTransactionsByState 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) CountTransactionsByState(ctx context.Context, state txmgrtypes.TxState) (count uint32, err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) CreateTransaction 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) CreateTransaction(ctx context.Context, txRequest txmgrtypes.TxRequest[ADDR, THASH]) (etx txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindEarliestUnconfirmedBroadcastTime 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindEarliestUnconfirmedBroadcastTime(ctx context.Context) (nullv4.Time, error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindEarliestUnconfirmedTxAttemptBlock 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindEarliestUnconfirmedTxAttemptBlock(ctx context.Context) (nullv4.Int, error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindTxesByMetaFieldAndStates 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindTxesByMetaFieldAndStates(ctx context.Context, metaField string, metaValue string, states []txmgrtypes.TxState, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindTxesWithAttemptsAndReceiptsByIdsAndState 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindTxesWithAttemptsAndReceiptsByIdsAndState(ctx context.Context, ids []int64, states []txmgrtypes.TxState, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindTxesWithMetaFieldByReceiptBlockNum 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindTxesWithMetaFieldByReceiptBlockNum(ctx context.Context, metaField string, blockNum int64, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindTxesWithMetaFieldByStates 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) FindTxesWithMetaFieldByStates(ctx context.Context, metaField string, states []txmgrtypes.TxState, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetForwarderForEOA 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetForwarderForEOA(ctx context.Context, addr ADDR) (fwdr ADDR, err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetForwarderForEOAOCR2Feeds 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetForwarderForEOAOCR2Feeds(ctx context.Context, _, _ ADDR) (fwdr ADDR, err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetTransactionFee 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetTransactionFee(ctx context.Context, transactionID string) (fee *evmtypes.TransactionFee, err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetTransactionReceipt 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetTransactionReceipt(ctx context.Context, transactionID string) (receipt *txmgrtypes.ChainReceipt[THASH, BHASH], err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetTransactionStatus 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetTransactionStatus(ctx context.Context, transactionID string) (status commontypes.TransactionStatus, err error)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) HealthReport 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) HealthReport() map[string]error

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Name 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Name() string

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) OnNewLongestChain 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) OnNewLongestChain(context.Context, HEAD)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Ready 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Ready() error

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) RegisterResumeCallback 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) RegisterResumeCallback(fn ResumeCallback)

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Reset 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Reset(addr ADDR, abandon bool) error

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) SendNativeToken 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) SendNativeToken(ctx context.Context, chainID CID, from, to ADDR, value big.Int, gasLimit uint64) (etx txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

SendNativeToken does nothing, null functionality

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Start 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Start(context.Context) error

Start does noop for NullTxManager.

func (*NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Trigger 

func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) Trigger(ADDR)

Trigger does noop for NullTxManager.

type ProcessUnstartedTxs 

type ProcessUnstartedTxs[ADDR chains.Hashable] func(ctx context.Context, fromAddress ADDR) (retryable bool, err error)

type Reaper 

type Reaper[CHAIN_ID chains.ID] struct {
	// contains filtered or unexported fields
}

Reaper handles periodic database cleanup for Txm

func NewReaper 

func NewReaper[CHAIN_ID chains.ID](lggr logger.Logger, store types.TxHistoryReaper[CHAIN_ID], txConfig types.ReaperTransactionsConfig, chainID CHAIN_ID) *Reaper[CHAIN_ID]

NewReaper instantiates a new reaper object

func (*Reaper[CHAIN_ID]) ReapTxes 

func (r *Reaper[CHAIN_ID]) ReapTxes(headNum int64) error

ReapTxes deletes old txes

func (*Reaper[CHAIN_ID]) SetLatestBlockNum 

func (r *Reaper[CHAIN_ID]) SetLatestBlockNum(latestBlockNum int64)

SetLatestBlockNum should be called on every new highest block number

func (*Reaper[CHAIN_ID]) Start 

func (r *Reaper[CHAIN_ID]) Start()

Start the reaper. Should only be called once.

func (*Reaper[CHAIN_ID]) Stop 

func (r *Reaper[CHAIN_ID]) Stop()

Stop the reaper. Should only be called once.

type Resender 

type Resender[CID chains.ID, ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, R txmgrtypes.ChainReceipt[THASH, BHASH], SEQ chains.Sequence, FEE fees.Fee] struct {
	// contains filtered or unexported fields
}

Resender periodically picks up transactions that have been languishing unconfirmed for a configured amount of time without being sent, and sends their highest priced attempt again. This helps to defend against geth/parity silently dropping txes, or txes being ejected from the mempool.

Previously we relied on the bumper to do this for us implicitly but there can occasionally be problems with this (e.g. abnormally long block times, or if gas bumping is disabled)

func NewResender 

func NewResender[
	CHAIN_ID chains.ID,
	ADDR chains.Hashable,
	TX_HASH chains.Hashable,
	BLOCK_HASH chains.Hashable,
	R txmgrtypes.ChainReceipt[TX_HASH, BLOCK_HASH],
	SEQ chains.Sequence,
	FEE fees.Fee,
](
	lggr logger.Logger,
	txStore txmgrtypes.TransactionStore[ADDR, CHAIN_ID, TX_HASH, BLOCK_HASH, SEQ, FEE],
	client txmgrtypes.TransactionClient[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
	tracker *Tracker[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE],
	ks txmgrtypes.KeyStore[ADDR],
	pollInterval time.Duration,
	config txmgrtypes.ResenderChainConfig,
	txConfig txmgrtypes.ResenderTransactionsConfig,
) *Resender[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]

func (*Resender[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) Start 

func (er *Resender[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) Start(ctx context.Context)

Start is a comment which satisfies the linter

func (*Resender[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) Stop 

func (er *Resender[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) Stop()

Stop is a comment which satisfies the linter

func (*Resender[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestResendUnconfirmed 

func (er *Resender[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestResendUnconfirmed() error

type ResumeCallback 

type ResumeCallback func(ctx context.Context, id uuid.UUID, result interface{}, err error) error

ResumeCallback is assumed to be idempotent

type SendEveryStrategy 

type SendEveryStrategy struct{}

SendEveryStrategy will always send the tx

func (SendEveryStrategy) PruneQueue 

func (SendEveryStrategy) PruneQueue(ctx context.Context, pruneService types.UnstartedTxQueuePruner) ([]int64, error)

func (SendEveryStrategy) Subject 

func (SendEveryStrategy) Subject() uuid.NullUUID

type Tracker 

type Tracker[CID chains.ID, ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, R types.ChainReceipt[THASH, BHASH], SEQ chains.Sequence, FEE fees.Fee] struct {
	services.StateMachine
	// contains filtered or unexported fields
}

Tracker tracks all transactions which have abandoned fromAddresses. The fromAddresses can be deleted by Node Operators from the KeyStore. In such cases, existing in-flight transactions for these fromAddresses are considered abandoned too. Since such Txs can still have attempts on chain's mempool, these could still be confirmed. This tracker just tracks such Txs for some time, in case they get confirmed as-is.

func NewTracker 

func NewTracker[CID chains.ID, ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, R types.ChainReceipt[THASH, BHASH], SEQ chains.Sequence, FEE fees.Fee](
	txStore types.TxStore[ADDR, CID, THASH, BHASH, R, SEQ, FEE],
	keyStore types.KeyStore[ADDR],
	chainID CID,
	lggr logger.Logger,
) *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]

func (*Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) AbandonedTxCount 

func (tr *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) AbandonedTxCount() int

AbandonedTxCount returns the number of abandoned txes currently being tracked

func (*Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) Close 

func (tr *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) Close() error

func (*Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) GetAbandonedAddresses 

func (tr *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) GetAbandonedAddresses() []ADDR

func (*Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) IsStarted 

func (tr *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) IsStarted() bool

func (*Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) Start 

func (tr *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) Start(ctx context.Context) error

func (*Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXDeliverBlock 

func (tr *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXDeliverBlock(blockHeight int64)

func (*Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestSetTTL 

func (tr *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestSetTTL(ttl time.Duration)

type TransmitChecker 

type TransmitChecker[CID chains.ID, ADDR chains.Hashable, THASH, BHASH chains.Hashable, SEQ chains.Sequence, FEE fees.Fee] interface {

	// Check the given transaction. If the transaction should not be sent, an error indicating why
	// is returned. Errors should only be returned if the checker can confirm that a transaction
	// should not be sent, other errors (for example connection or other unexpected errors) should
	// be logged and swallowed.
	Check(ctx context.Context, l logger.SugaredLogger, tx types.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], a types.TxAttempt[CID, ADDR, THASH, BHASH, SEQ, FEE]) error
}

TransmitChecker determines whether a transaction should be submitted on-chain.

type TransmitCheckerFactory 

type TransmitCheckerFactory[CID chains.ID, ADDR chains.Hashable, THASH, BHASH chains.Hashable, SEQ chains.Sequence, FEE fees.Fee] interface {
	// BuildChecker builds a new TransmitChecker based on the given spec.
	BuildChecker(spec types.TransmitCheckerSpec[ADDR]) (TransmitChecker[CID, ADDR, THASH, BHASH, SEQ, FEE], error)
}

TransmitCheckerFactory creates a transmit checker based on a spec.

type TxManager 

type TxManager[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, SEQ chains.Sequence, FEE fees.Fee] interface {
	heads.Trackable[HEAD, BHASH]
	services.Service
	Trigger(addr ADDR)
	CreateTransaction(ctx context.Context, txRequest txmgrtypes.TxRequest[ADDR, THASH]) (etx txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)
	GetForwarderForEOA(ctx context.Context, eoa ADDR) (forwarder ADDR, err error)
	GetForwarderForEOAOCR2Feeds(ctx context.Context, eoa, ocr2AggregatorID ADDR) (forwarder ADDR, err error)
	RegisterResumeCallback(fn ResumeCallback)
	SendNativeToken(ctx context.Context, chainID CID, from, to ADDR, value big.Int, gasLimit uint64) (etx txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)
	Reset(addr ADDR, abandon bool) error
	// Find transactions by a field in the TxMeta blob and transaction states
	FindTxesByMetaFieldAndStates(ctx context.Context, metaField string, metaValue string, states []txmgrtypes.TxState, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)
	// Find transactions with a non-null TxMeta field that was provided by transaction states
	FindTxesWithMetaFieldByStates(ctx context.Context, metaField string, states []txmgrtypes.TxState, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)
	// Find transactions with a non-null TxMeta field that was provided and a receipt block number greater than or equal to the one provided
	FindTxesWithMetaFieldByReceiptBlockNum(ctx context.Context, metaField string, blockNum int64, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)
	// Find transactions loaded with transaction attempts and receipts by transaction IDs and states
	FindTxesWithAttemptsAndReceiptsByIdsAndState(ctx context.Context, ids []int64, states []txmgrtypes.TxState, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)
	FindEarliestUnconfirmedBroadcastTime(ctx context.Context) (nullv4.Time, error)
	FindEarliestUnconfirmedTxAttemptBlock(ctx context.Context) (nullv4.Int, error)
	CountTransactionsByState(ctx context.Context, state txmgrtypes.TxState) (count uint32, err error)
	GetTransactionStatus(ctx context.Context, transactionID string) (state commontypes.TransactionStatus, err error)
	GetTransactionFee(ctx context.Context, transactionID string) (fee *evmtypes.TransactionFee, err error)
	GetTransactionReceipt(ctx context.Context, transactionID string) (receipt *txmgrtypes.ChainReceipt[THASH, BHASH], err error)
	CalculateFee(feeParts FeeParts) *big.Int
}

TxManager is the main component of the transaction manager. It is also the interface to external callers.

type Txm 

type Txm[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, R txmgrtypes.ChainReceipt[THASH, BHASH], SEQ chains.Sequence, FEE fees.Fee] struct {
	services.StateMachine
	// contains filtered or unexported fields
}

func NewTxm 

func NewTxm[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, R txmgrtypes.ChainReceipt[THASH, BHASH], SEQ chains.Sequence, FEE fees.Fee](
	chainID CID,
	cfg txmgrtypes.TransactionManagerChainConfig,
	txCfg txmgrtypes.TransactionManagerTransactionsConfig,
	keyStore txmgrtypes.KeyStore[ADDR],
	lggr logger.Logger,
	checkerFactory TransmitCheckerFactory[CID, ADDR, THASH, BHASH, SEQ, FEE],
	fwdMgr txmgrtypes.ForwarderManager[ADDR],
	txAttemptBuilder txmgrtypes.TxAttemptBuilder[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE],
	txStore txmgrtypes.TxStore[ADDR, CID, THASH, BHASH, R, SEQ, FEE],
	broadcaster *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE],
	confirmer *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE],
	resender *Resender[CID, ADDR, THASH, BHASH, R, SEQ, FEE],
	tracker *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE],
	finalizer txmgrtypes.Finalizer[BHASH, HEAD],
	newErrorClassifierFunc NewErrorClassifier,
	txmv2wrapper TxmV2Wrapper[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE],
) *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]

NewTxm creates a new Txm with the given configuration.

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) CalculateFee 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) CalculateFee(feeParts FeeParts) *big.Int

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Close 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Close() (merr error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) CountTransactionsByState 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) CountTransactionsByState(ctx context.Context, state txmgrtypes.TxState) (count uint32, err error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) CreateTransaction 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) CreateTransaction(ctx context.Context, txRequest txmgrtypes.TxRequest[ADDR, THASH]) (tx txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

CreateTransaction inserts a new transaction

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindEarliestUnconfirmedBroadcastTime 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindEarliestUnconfirmedBroadcastTime(ctx context.Context) (nullv4.Time, error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindEarliestUnconfirmedTxAttemptBlock 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindEarliestUnconfirmedTxAttemptBlock(ctx context.Context) (nullv4.Int, error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxesByMetaFieldAndStates 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxesByMetaFieldAndStates(ctx context.Context, metaField string, metaValue string, states []txmgrtypes.TxState, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxesWithAttemptsAndReceiptsByIdsAndState 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxesWithAttemptsAndReceiptsByIdsAndState(ctx context.Context, ids []int64, states []txmgrtypes.TxState, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxesWithMetaFieldByReceiptBlockNum 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxesWithMetaFieldByReceiptBlockNum(ctx context.Context, metaField string, blockNum int64, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxesWithMetaFieldByStates 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) FindTxesWithMetaFieldByStates(ctx context.Context, metaField string, states []txmgrtypes.TxState, chainID *big.Int) (txes []*txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetForwarderForEOA 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetForwarderForEOA(ctx context.Context, eoa ADDR) (forwarder ADDR, err error)

Calls forwarderMgr to get a proper forwarder for a given EOA.

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetForwarderForEOAOCR2Feeds 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetForwarderForEOAOCR2Feeds(ctx context.Context, eoa, ocr2Aggregator ADDR) (forwarder ADDR, err error)

GetForwarderForEOAOCR2Feeds calls forwarderMgr to get a proper forwarder for a given EOA and checks if its set as a transmitter on the OCR2Aggregator contract.

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetTransactionFee 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetTransactionFee(ctx context.Context, transactionID string) (fee *evmtypes.TransactionFee, err error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetTransactionReceipt 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetTransactionReceipt(ctx context.Context, transactionID string) (receipt *txmgrtypes.ChainReceipt[THASH, BHASH], err error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetTransactionStatus 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetTransactionStatus(ctx context.Context, transactionID string) (status commontypes.TransactionStatus, err error)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) HealthReport 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) HealthReport() map[string]error

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Name 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Name() string

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) OnNewLongestChain 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) OnNewLongestChain(ctx context.Context, head HEAD)

OnNewLongestChain conforms to HeadTrackable

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) RegisterResumeCallback 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) RegisterResumeCallback(fn ResumeCallback)

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Reset 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Reset(addr ADDR, abandon bool) (err error)

Reset stops Broadcaster/Confirmer, executes callback, then starts them again

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) SendNativeToken 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) SendNativeToken(ctx context.Context, chainID CID, from, to ADDR, value big.Int, gasLimit uint64) (etx txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)

SendNativeToken creates a transaction that transfers the given value of native tokens

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Start 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Start(ctx context.Context) (merr error)

Start starts Txm service. The provided context can be used to terminate Start sequence.

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Trigger 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) Trigger(addr ADDR)

Trigger forces the Broadcaster to check early for the given address

func (*Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestAbandon 

func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) XXXTestAbandon(addr ADDR) (err error)

type TxmV2Wrapper 

type TxmV2Wrapper[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, SEQ chains.Sequence, FEE fees.Fee] interface {
	services.Service
	CreateTransaction(ctx context.Context, txRequest txmgrtypes.TxRequest[ADDR, THASH]) (etx txmgrtypes.Tx[CID, ADDR, THASH, BHASH, SEQ, FEE], err error)
	Reset(addr ADDR, abandon bool) error
}

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.