Feat: Add a coin reservation abstraction to prevent equivocation within the same relayer (#304)

* add a GasCoinManager abstraction

* add locked coin error parsing to reserve coins

* allow setting an expiry when reserving coins

* update GasCoinManager

* fix txm tests

* release coins on confirmer marking transaction as failed or succeeded

* error checking

* add separate txm error strategy for simple coin refreshing (used for locked coins)

* nit

---------

Co-authored-by: stackman27 <[email protected]>

Author: faisal-chainlink

relayer/client/suierrors/errors.go

@@ -13,6 +13,8 @@ determine if an error is considered retryable (e.g. IsRetryable) according to Su
 package suierrors
 
 import (
+	"regexp"
+	"strconv"
 	"strings"
 )
 
@@ -28,6 +30,7 @@ const (
 	CheckpointAndConsensusErrors
 	PublishingErrors
 	SoftBundleErrors
+	LockCoinErrors
 	UnknownErrors
 )
 
@@ -47,6 +50,8 @@ func (c ErrorCategory) String() string {
 		return "Publishing Errors"
 	case SoftBundleErrors:
 		return "Soft Bundle Errors"
+	case LockCoinErrors:
+		return "Lock Coins Errors"
 	default:
 		return "Unknown Error Category"
 	}
@@ -146,6 +151,9 @@ var ErrNoSharedObjectError = NewSuiError(SoftBundleErrors, "NoSharedObjectError"
 var ErrAlreadyExecutedError = NewSuiError(SoftBundleErrors, "AlreadyExecutedError")
 var ErrCertificateAlreadyProcessed = NewSuiError(SoftBundleErrors, "CertificateAlreadyProcessed")
 
+// Lock Coins error
+var ErrLockCoins = NewSuiError(LockCoinErrors, "already locked by a different transaction")
+
 // Unknown Error
 var ErrUnknownError = NewSuiError(UnknownErrors, "UnknownError")
 
@@ -176,6 +184,7 @@ var suiErrorMappings = []struct {
 	{ErrUnsupported.Error(), ErrUnsupported},
 	{ErrMoveFunctionInputError.Error(), ErrMoveFunctionInputError},
 	{ErrPostRandomCommandRestrictions.Error(), ErrPostRandomCommandRestrictions},
+	{ErrObjectVersionUnavailableForConsumption.Error(), ErrObjectVersionUnavailableForConsumption},
 
 	// Gas Errors
 	{ErrMissingGasPayment.Error(), ErrMissingGasPayment},
@@ -225,10 +234,17 @@ var suiErrorMappings = []struct {
 	{ErrAlreadyExecutedError.Error(), ErrAlreadyExecutedError},
 	{ErrCertificateAlreadyProcessed.Error(), ErrCertificateAlreadyProcessed},
 
+	// Lock Coins error
+	{ErrLockCoins.Error(), ErrLockCoins},
+
 	// Unknown Error
 	{ErrUnknownError.Error(), ErrUnknownError},
 }
 
+var lockedObjectRe = regexp.MustCompile(
+	`Object\s+\((0x[0-9a-fA-F]+),\s*SequenceNumber\((\d+)\)`,
+)
+
 // ParseSuiErrorMessage maps a raw RPC error message to a structured error.
 // It iterates over the known substrings in suiErrorMappings. If a substring is found,
 // the corresponding error is returned. Otherwise, it returns an "Unknown error".
@@ -242,6 +258,25 @@ func ParseSuiErrorMessage(msg string) *SuiError {
 	return NewSuiError(UnknownErrors, msg)
 }
 
+// ExtractLockedObjectRef parses a Sui equivocation / lock error message and returns
+// (objectID, version, ok).
+func ExtractLockedObjectRef(msg string) (string, uint64, bool) {
+	m := lockedObjectRe.FindStringSubmatch(msg)
+	if len(m) != 3 {
+		return "", 0, false
+	}
+
+	objID := m[1]
+	verStr := m[2]
+
+	ver, err := strconv.ParseUint(verStr, 10, 64)
+	if err != nil {
+		return objID, 0, false
+	}
+
+	return objID, ver, true
+}
+
 var ExponentialBackoffErrors = []error{
 	ErrPackageVerificationTimeout,
 	ErrVerifiedCheckpointNotFound,
@@ -262,3 +297,7 @@ var GasBumpErrors = []error{
 	ErrInsufficientGas,
 	ErrGasPriceTooHigh,
 }
+
+var CoinRefreshErrors = []error{
+	ErrLockCoins,
+}

relayer/txm/broadcaster.go

@@ -92,6 +92,13 @@ func broadcastTransactions(loopCtx context.Context, txm *SuiTxm, transactions []
 			// Default to retrying the transaction
 			newState := StateRetriable
 
+			// Attach the transaction submission error to the transaction object. If it remains marked as retriable, 
+			// the "confirmer" loop will pick it up and potentially retry it.
+			err = txm.transactionRepository.UpdateTransactionBroadcastError(tx.TransactionID, err.Error())
+			if err != nil {
+				txm.lggr.Errorw("Failed to update transaction broadcast error", "txID", tx.TransactionID, "error", err)
+			}
+
 			if resp.Effects.Status.Status != "" && resp.TxDigest == "" {
 				// Update the transaction state to Failed if the digest is empty
 				// An empty digest indicates a total failure of the transaction
@@ -102,7 +109,6 @@ func broadcastTransactions(loopCtx context.Context, txm *SuiTxm, transactions []
 			// Attempt updating the state if it has changed
 			if tx.State != newState {
 				err = txm.transactionRepository.ChangeState(tx.TransactionID, newState)
-
 				if err != nil {
 					txm.lggr.Errorw("Failed to change transaction state", "txID", tx.TransactionID, "error", err)
 				}
@@ -118,6 +124,8 @@ func broadcastTransactions(loopCtx context.Context, txm *SuiTxm, transactions []
 			continue
 		}
 
+		// Update the transaction state to submitted as we have not yet confirmed its status.
+		// The "confirmer" loop checks the transactions statuses and possibly marks them as finalized.
 		err = txm.transactionRepository.ChangeState(tx.TransactionID, StateSubmitted)
 		if err != nil {
 			txm.lggr.Errorw("Failed to change transaction state to Submitted", "txID", tx.TransactionID, "error", err)

relayer/txm/confirmer.go

@@ -5,6 +5,8 @@ import (
 	"math"
 	"time"
 
+	"github.com/block-vision/sui-go-sdk/models"
+	"github.com/block-vision/sui-go-sdk/transaction"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 
 	"github.com/smartcontractkit/chainlink-sui/relayer/client"
@@ -56,14 +58,26 @@ func checkConfirmations(loopCtx context.Context, txm *SuiTxm) {
 
 	for _, tx := range inFlightTransactions {
 		txm.lggr.Debugw("Checking transaction confirmations", "transactionID", tx.TransactionID)
-		if tx.State != StateSubmitted {
-			continue
-		}
 
-		txm.lggr.Debugw("Transaction is in submitted state", "transactionID", tx.TransactionID)
-		resp, err := txm.suiGateway.GetTransactionStatus(loopCtx, tx.Digest)
-		if err != nil {
-			txm.lggr.Errorw("Error getting transaction status", "transactionID", tx.TransactionID, "error", err)
+		var resp client.TransactionResult
+		var err error
+
+		if tx.State == StateSubmitted {
+			txm.lggr.Debugw("Transaction is in submitted state", "transactionID", tx.TransactionID)
+			resp, err = txm.suiGateway.GetTransactionStatus(loopCtx, tx.Digest)
+			if err != nil {
+				txm.lggr.Errorw("Error getting transaction status", "transactionID", tx.TransactionID, "error", err)
+				continue
+			}
+		} else if tx.State == StateRetriable {
+			txm.lggr.Debugw("Transaction is in retriable state", "transactionID", tx.TransactionID)
+			// Check if it's a broadcast error (never made it onchain)
+			if tx.BroadcastError == "" {
+				continue
+			}
+			resp.Status = failure
+			resp.Error = tx.BroadcastError
+		} else {
 			continue
 		}
 
@@ -88,15 +102,53 @@ func handleSuccess(txm *SuiTxm, tx SuiTx) error {
 		return err
 	}
 	txm.lggr.Infow("Transaction finalized", "transactionID", tx.TransactionID)
+
+	if err := txm.coinManager.ReleaseCoins(tx.TransactionID); err != nil {
+		// This error is not critical, can be safely ignored as the coins will auto-release after the default TTL
+		txm.lggr.Debugw("Failed to release coins", "transactionID", tx.TransactionID, "error", err)
+	}
+
 	return nil
 }
 
 func handleTransactionError(ctx context.Context, txm *SuiTxm, tx SuiTx, result *client.TransactionResult) error {
 	txm.lggr.Debugw("Handling transaction error", "transactionID", tx.TransactionID, "error", result.Error)
 
 	txError := suierrors.ParseSuiErrorMessage(result.Error)
-	if txError == nil {
-		txError = suierrors.NewSuiError(suierrors.UnknownErrors, result.Error)
+
+	// Check if the error is a locked object error, mark the coin as reserved if it is not already
+	// to avoid other transactions from using it
+	if objectID, version, ok := suierrors.ExtractLockedObjectRef(result.Error); ok {
+		txm.lggr.Infow("Detected locked coin at confirmation time",
+			"txID", tx.TransactionID,
+			"objectID", objectID,
+			"version", version,
+		)
+
+		coinID, err := transaction.ConvertSuiAddressStringToBytes(models.SuiAddress(objectID))
+		if err == nil && !txm.coinManager.IsCoinReserved(*coinID) {
+			// Coin lock duration
+			expiry := DefaultLockedCoinTTL
+
+			// The coin is not recorded is not marked as reserved, mark it as reserved
+			err = txm.coinManager.TryReserveCoins(ctx, tx.TransactionID, []transaction.SuiObjectRef{
+				{
+					ObjectId: *coinID,
+					Version:  0,
+					Digest:   nil,
+				},
+			}, &expiry)
+
+			if err != nil {
+				// This is not a critical error, so we continue
+				txm.lggr.Debugw(
+					"Failed to mark locked coin as reserved",
+					"transactionID", tx.TransactionID,
+					"objectID", objectID,
+					"error", err,
+				)
+			}
+		}
 	}
 
 	isRetryable, strategy := txm.retryManager.IsRetryable(&tx, result.Error)
@@ -111,6 +163,8 @@ func handleTransactionError(ctx context.Context, txm *SuiTxm, tx SuiTx, result *
 		return handleExponentialBackoffRetry(txm, tx)
 	case GasBump:
 		return handleGasBumpRetry(ctx, txm, tx, txError)
+	case CoinRefresh:
+		return handleCoinRefreshRetry(ctx, txm, tx, txError)
 	case NoRetry:
 		return markTransactionFailed(txm, tx, txError)
 	default:
@@ -147,6 +201,40 @@ func handleGasBumpRetry(ctx context.Context, txm *SuiTxm, tx SuiTx, txError *sui
 	return nil
 }
 
+func handleCoinRefreshRetry(ctx context.Context, txm *SuiTxm, tx SuiTx, txError *suierrors.SuiError) error {
+	txm.lggr.Infow("Coin refresh strategy - refreshing coins for locked coin error", "transactionID", tx.TransactionID)
+
+	// Release the old coins that are locked
+	if err := txm.coinManager.ReleaseCoins(tx.TransactionID); err != nil {
+		// This is not critical - coins will auto-release after TTL
+		txm.lggr.Debugw("Failed to release old coins", "transactionID", tx.TransactionID, "error", err)
+	}
+
+	// Get the current transaction to ensure we have the latest state
+	currentTx, err := txm.transactionRepository.GetTransaction(tx.TransactionID)
+	if err != nil {
+		txm.lggr.Errorw("Failed to get current transaction", "transactionID", tx.TransactionID, "error", err)
+		return err
+	}
+
+	// Calling UpdateTransactionGas will also update the gas coins used as the transaction gets re-built
+	// with new (unlocked) coins.
+	// Call chain: UpdateTransactionGas -> UpdateBSCPayload -> preparePTBTransaction (this refreshes the coins).
+	if err := txm.transactionRepository.UpdateTransactionGas(ctx, txm.keystoreService, txm.suiGateway, tx.TransactionID, currentTx.Metadata.GasLimit); err != nil {
+		txm.lggr.Errorw("Failed to update transaction with refreshed coins", "transactionID", tx.TransactionID, "error", err)
+		return err
+	}
+
+	if err := txm.transactionRepository.ChangeState(tx.TransactionID, StateRetriable); err != nil {
+		txm.lggr.Errorw("Failed to update transaction state", "transactionID", tx.TransactionID, "error", err)
+		return err
+	}
+
+	txm.lggr.Infow("Transaction refreshed with new coins", "transactionID", tx.TransactionID)
+	txm.broadcastChannel <- tx.TransactionID
+	return nil
+}
+
 func handleExponentialBackoffRetry(txm *SuiTxm, tx SuiTx) error {
 	delaySeconds := float64(defaultExponentialBackoffDelaySeconds) * math.Pow(2, float64(tx.Attempt))
 
@@ -185,5 +273,11 @@ func markTransactionFailed(txm *SuiTxm, tx SuiTx, txError *suierrors.SuiError) e
 	}
 
 	txm.lggr.Infow("Transaction failed", "transactionID", tx.TransactionID)
+
+	if err := txm.coinManager.ReleaseCoins(tx.TransactionID); err != nil {
+		// This error is not critical, can be safely ignored as the coins will auto-release after the default TTL
+		txm.lggr.Debugw("Failed to release coins", "transactionID", tx.TransactionID, "error", err)
+	}
+
 	return nil
 }

relayer/txm/confirmer_test.go

@@ -56,6 +56,7 @@ func TestConfirmerRoutine_GasBump(t *testing.T) {
 	// Create a fake gas manager that returns an updated gas value.
 	maxGasBudget := big.NewInt(12000000)
 	gasManager := txm.NewSuiGasManager(lggr, fakeClient, *maxGasBudget, 0)
+	coinManager := txm.NewGasCoinManager(lggr, fakeClient)
 
 	// For the confirmer, the keystore is not used; create a dummy signer.
 	keystoreInstance := testutils.NewTestKeystore(t)
@@ -119,6 +120,7 @@ func TestConfirmerRoutine_GasBump(t *testing.T) {
 		TxError:       nil,
 		GasBudget:     maxGasBudget.Uint64(),
 		Ptb:           ptb,
+		CoinManager:   coinManager,
 	}
 	err = store.AddTransaction(tx)
 	require.NoError(t, err)
@@ -184,6 +186,7 @@ func TestConfirmerRoutine_SuccessfulGasBumpAfterTwoAttempts(t *testing.T) {
 	maxGasBudget := big.NewInt(10000000)
 	percentIncrease := int64(120) // 120% (20% increase) per bump
 	gasManager := txm.NewSuiGasManager(lggr, fakeClient, *maxGasBudget, percentIncrease)
+	coinManager := txm.NewGasCoinManager(lggr, fakeClient)
 
 	// Create keystore
 	keystoreInstance := testutils.NewTestKeystore(t)
@@ -248,6 +251,7 @@ func TestConfirmerRoutine_SuccessfulGasBumpAfterTwoAttempts(t *testing.T) {
 		TxError:       nil,
 		GasBudget:     maxGasBudget.Uint64(), // Use max budget to allow for gas bumps
 		Ptb:           ptb,
+		CoinManager:   coinManager,
 	}
 	err = store.AddTransaction(tx)
 	require.NoError(t, err)
@@ -360,6 +364,7 @@ func TestConfirmerRoutine_ExponentialBackoffRetry(t *testing.T) {
 
 	// Add a transaction in StateSubmitted with a known digest
 	txID := "tx-exponential-backoff-retry-test"
+	coinManager := txm.NewGasCoinManager(lggr, fakeClient)
 	tx := txm.SuiTx{
 		TransactionID: txID,
 		Sender:        address,
@@ -376,6 +381,7 @@ func TestConfirmerRoutine_ExponentialBackoffRetry(t *testing.T) {
 		TxError:       nil,
 		GasBudget:     initialGasBudget,
 		Ptb:           ptb,
+		CoinManager:   coinManager,
 	}
 	err = store.AddTransaction(tx)
 	require.NoError(t, err)