expired tx changes without tests

docs/relay/README.md

@@ -43,7 +43,8 @@ chainlink nodes solana create --name=<node-name> --chain-id=<chain-id> --url=<ur
 | `OCR2CacheTTL`        | stale OCR2 cache deadline                                                                                                                                                                                          | 1m          |                                       |
 | `TxTimeout`           | timeout to send tx to rpc endpoint                                                                                                                                                                                 | 1m          |                                       |
 | `TxRetryTimeout`      | duration for tx to be rebroadcast to rpc, txm stops rebroadcast after timeout                                                                                                                                      | 10s          |                                       |
-| `TxConfirmTimeout`    | duration when confirming a tx signature before signature is discarded as unconfirmed                                                                                                                               | 30s         |                                       |
+| `TxConfirmTimeout`    | duration when confirming a tx signature before signature is discarded as unconfirmed                                                                                                                               | 30s         |  
+| `TxExpirationRebroadcast`       | enable or disable tx rebroadcast if expired. Expiration check is performed every `TxConfirmTimeout` interval                                                                                                                                                                 | `false`      | `true`, `false`                                      |
 | `SkipPreflight`       | enable or disable preflight checks when sending tx                                                                                                                                                                 | `true`      | `true`, `false`                       |
 | `Commitment`          | Confirmation level for solana state and transactions. ([documentation](https://docs.solana.com/developing/clients/jsonrpc-api#configuring-state-commitment))                                                       | `confirmed` | `processed`, `confirmed`, `finalized` |
 | `MaxRetries`          | Parameter when sending transactions, how many times the RPC node will automatically rebroadcast a tx, default = `0` for custom txm rebroadcasting method, set to `-1` to use the RPC node's default retry strategy | `0`         |                                       |

pkg/solana/config/config.go

@@ -11,17 +11,18 @@ import (
 
 // Global solana defaults.
 var defaultConfigSet = Chain{
-	BalancePollPeriod:   config.MustNewDuration(5 * time.Second),        // poll period for balance monitoring
-	ConfirmPollPeriod:   config.MustNewDuration(500 * time.Millisecond), // polling for tx confirmation
-	OCR2CachePollPeriod: config.MustNewDuration(time.Second),            // cache polling rate
-	OCR2CacheTTL:        config.MustNewDuration(time.Minute),            // stale cache deadline
-	TxTimeout:           config.MustNewDuration(time.Minute),            // timeout for send tx method in client
-	TxRetryTimeout:      config.MustNewDuration(10 * time.Second),       // duration for tx rebroadcasting to RPC node
-	TxConfirmTimeout:    config.MustNewDuration(30 * time.Second),       // duration before discarding tx as unconfirmed. Set to 0 to disable discarding tx.
-	TxRetentionTimeout:  config.MustNewDuration(0 * time.Second),        // duration to retain transactions after being marked as finalized or errored. Set to 0 to immediately drop transactions.
-	SkipPreflight:       ptr(true),                                      // to enable or disable preflight checks
-	Commitment:          ptr(string(rpc.CommitmentConfirmed)),
-	MaxRetries:          ptr(int64(0)), // max number of retries (default = 0). when config.MaxRetries < 0), interpreted as MaxRetries = nil and rpc node will do a reasonable number of retries
+	BalancePollPeriod:       config.MustNewDuration(5 * time.Second),        // poll period for balance monitoring
+	ConfirmPollPeriod:       config.MustNewDuration(500 * time.Millisecond), // polling for tx confirmation
+	OCR2CachePollPeriod:     config.MustNewDuration(time.Second),            // cache polling rate
+	OCR2CacheTTL:            config.MustNewDuration(time.Minute),            // stale cache deadline
+	TxTimeout:               config.MustNewDuration(time.Minute),            // timeout for send tx method in client
+	TxRetryTimeout:          config.MustNewDuration(10 * time.Second),       // duration for tx rebroadcasting to RPC node
+	TxConfirmTimeout:        config.MustNewDuration(30 * time.Second),       // duration before discarding tx as unconfirmed. Set to 0 to disable discarding tx.
+	TxExpirationRebroadcast: ptr(false),
+	TxRetentionTimeout:      config.MustNewDuration(0 * time.Second), // duration to retain transactions after being marked as finalized or errored. Set to 0 to immediately drop transactions.
+	SkipPreflight:           ptr(true),                               // to enable or disable preflight checks
+	Commitment:              ptr(string(rpc.CommitmentConfirmed)),
+	MaxRetries:              ptr(int64(0)), // max number of retries (default = 0). when config.MaxRetries < 0), interpreted as MaxRetries = nil and rpc node will do a reasonable number of retries
 
 	// fee estimator
 	FeeEstimatorMode:         ptr("fixed"),
@@ -43,6 +44,7 @@ type Config interface {
 	TxTimeout() time.Duration
 	TxRetryTimeout() time.Duration
 	TxConfirmTimeout() time.Duration
+	TxExpirationRebroadcast() bool
 	TxRetentionTimeout() time.Duration
 	SkipPreflight() bool
 	Commitment() rpc.CommitmentType
@@ -68,6 +70,7 @@ type Chain struct {
 	TxTimeout                *config.Duration
 	TxRetryTimeout           *config.Duration
 	TxConfirmTimeout         *config.Duration
+	TxExpirationRebroadcast  *bool
 	TxRetentionTimeout       *config.Duration
 	SkipPreflight            *bool
 	Commitment               *string
@@ -105,6 +108,9 @@ func (c *Chain) SetDefaults() {
 	if c.TxConfirmTimeout == nil {
 		c.TxConfirmTimeout = defaultConfigSet.TxConfirmTimeout
 	}
+	if c.TxExpirationRebroadcast == nil {
+		c.TxExpirationRebroadcast = defaultConfigSet.TxExpirationRebroadcast
+	}
 	if c.TxRetentionTimeout == nil {
 		c.TxRetentionTimeout = defaultConfigSet.TxRetentionTimeout
 	}

pkg/solana/config/toml.go

@@ -155,6 +155,9 @@ func setFromChain(c, f *Chain) {
 	if f.TxConfirmTimeout != nil {
 		c.TxConfirmTimeout = f.TxConfirmTimeout
 	}
+	if f.TxExpirationRebroadcast != nil {
+		c.TxExpirationRebroadcast = f.TxExpirationRebroadcast
+	}
 	if f.TxRetentionTimeout != nil {
 		c.TxRetentionTimeout = f.TxRetentionTimeout
 	}
@@ -241,6 +244,10 @@ func (c *TOMLConfig) TxConfirmTimeout() time.Duration {
 	return c.Chain.TxConfirmTimeout.Duration()
 }
 
+func (c *TOMLConfig) TxExpirationRebroadcast() bool {
+	return *c.Chain.TxExpirationRebroadcast
+}
+
 func (c *TOMLConfig) TxRetentionTimeout() time.Duration {
 	return c.Chain.TxRetentionTimeout.Duration()
 }

pkg/solana/txm/pendingtx.go

@@ -28,6 +28,8 @@ type PendingTxContext interface {
 	Remove(sig solana.Signature) (string, error)
 	// ListAll returns all of the signatures being tracked for all transactions not yet finalized or errored
 	ListAll() []solana.Signature
+	// ListAllExpiredBroadcastedTxs returns all the expired broadcasted that are in broadcasted state and have expired for given slot height.
+	ListAllExpiredBroadcastedTxs(currHeight uint64) []PendingTx
 	// Expired returns whether or not confirmation timeout amount of time has passed since creation
 	Expired(sig solana.Signature, confirmationTimeout time.Duration) bool
 	// OnProcessed marks transactions as Processed
@@ -216,6 +218,19 @@ func (c *pendingTxContext) ListAll() []solana.Signature {
 	return maps.Keys(c.sigToID)
 }
 
+// ListAllExpiredBroadcastedTxs returns all the expired broadcasted that are in broadcasted state and have expired for given slot height.
+func (c *pendingTxContext) ListAllExpiredBroadcastedTxs(currHeight uint64) []PendingTx {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	broadcastedTxes := make([]PendingTx, 0, len(c.broadcastedProcessedTxs)) // worst case, all of them
+	for _, tx := range c.broadcastedProcessedTxs {
+		if tx.state == Broadcasted && tx.lastValidBlockHeight < currHeight {
+			broadcastedTxes = append(broadcastedTxes, tx)
+		}
+	}
+	return broadcastedTxes
+}
+
 // Expired returns if the timeout for trying to confirm a signature has been reached
 func (c *pendingTxContext) Expired(sig solana.Signature, confirmationTimeout time.Duration) bool {
 	c.lock.RLock()
@@ -604,6 +619,10 @@ func (c *pendingTxContextWithProm) ListAll() []solana.Signature {
 	return sigs
 }
 
+func (c *pendingTxContextWithProm) ListAllExpiredBroadcastedTxs(currHeight uint64) []PendingTx {
+	return c.pendingTx.ListAllExpiredBroadcastedTxs(currHeight)
+}
+
 func (c *pendingTxContextWithProm) Expired(sig solana.Signature, lifespan time.Duration) bool {
 	return c.pendingTx.Expired(sig, lifespan)
 }

pkg/solana/txm/txm.go

@@ -464,6 +464,9 @@ func (txm *Txm) confirm() {
 				txm.lggr.Errorw("failed to get client in txm.confirm", "error", err)
 				return
 			}
+			if txm.cfg.TxExpirationRebroadcast() {
+				txm.rebroadcastExpiredTxs(ctx, client)
+			}
 			txm.processConfirmations(ctx, client, sigs)
 		}
 		tick = time.After(utils.WithJitter(txm.cfg.ConfirmPollPeriod()))
@@ -497,6 +500,46 @@ func (txm *Txm) processConfirmations(ctx context.Context, client client.ReaderWr
 	wg.Wait() // wait for processing to finish
 }
 
+func (txm *Txm) rebroadcastExpiredTxs(ctx context.Context, client client.ReaderWriter) {
+	// Get current slot height to check if txes have expired when compared against their lastValidBlockHeight
+	currHeight, err := client.SlotHeight(ctx)
+	if err != nil {
+		txm.lggr.Errorw("failed to get current slot height", "error", err)
+		return
+	}
+
+	// Rebroadcast all expired txes
+	for _, tx := range txm.txs.ListAllExpiredBroadcastedTxs(currHeight) {
+		txm.lggr.Infow("transaction expired, rebroadcasting", "id", tx.UUID, "signature", tx.signatures)
+		if len(tx.signatures) == 0 { // prevent panic, shouldn't happen.
+			txm.lggr.Errorw("no signatures found for expired transaction", "id", tx.UUID)
+			continue
+		}
+		_, err := txm.txs.Remove(tx.signatures[0]) // only picking signature[0] because remove func removes all remaining signatures.
+		if err != nil {
+			txm.lggr.Errorw("failed to remove expired transaction", "id", tx.UUID, "error", err)
+			continue
+		}
+
+		rebroadcastTx := &PendingTx{
+			Tx:           tx.Tx,
+			UUID:         tx.UUID, // same id to handle case where set by caller. Previous ID has already been removed.
+			BalanceCheck: tx.BalanceCheck,
+			Amount:       tx.Amount,
+			From:         tx.From,
+		}
+
+		// Re-enqueue the transaction for rebroadcasting
+		err = txm.Enqueue(ctx, rebroadcastTx)
+		if err != nil {
+			txm.lggr.Errorw("failed to enqueue rebroadcast transaction", "id", tx.UUID, "error", err)
+			continue
+		}
+
+		txm.lggr.Infow("rebroadcast transaction enqueued", "id", tx.UUID)
+	}
+}
+
 func (txm *Txm) processSignatureStatuses(sigs []solanaGo.Signature, res []*rpc.SignatureStatusesResult) {
 	// Sort signatures and results process successful first
 	sortedSigs, sortedRes, err := SortSignaturesAndResults(sigs, res)

pkg/solana/txm/txm_internal_test.go

@@ -132,7 +132,7 @@ func TestTxm(t *testing.T) {
 				},
 			}, nil)
 			mc.On("GetLatestBlock", mock.Anything).Return(&rpc.GetBlockResult{}, nil).Maybe()
-			mc.On("SlotHeight", mock.Anything).Return(uint64(0), nil).Maybe()
+			mc.On("SlotHeight", mock.Anything).Return(uint64(0), nil)
 
 			// mock solana keystore
 			mkey := keyMocks.NewSimpleKeystore(t)
@@ -775,6 +775,7 @@ func TestTxm_disabled_confirm_timeout_with_retention(t *testing.T) {
 		},
 	}, nil)
 	mc.On("GetLatestBlock", mock.Anything).Return(&rpc.GetBlockResult{}, nil).Maybe()
+	mc.On("SlotHeight", mock.Anything).Return(uint64(0), nil)
 
 	computeUnitLimitDefault := fees.ComputeUnitLimit(cfg.ComputeUnitLimitDefault())
 
@@ -980,6 +981,7 @@ func TestTxm_compute_unit_limit_estimation(t *testing.T) {
 		},
 	}, nil)
 	mc.On("GetLatestBlock", mock.Anything).Return(&rpc.GetBlockResult{}, nil).Maybe()
+	mc.On("SlotHeight", mock.Anything).Return(uint64(0), nil)
 
 	// mock solana keystore
 	mkey := keyMocks.NewSimpleKeystore(t)