chore(submitter): refactoring points

btcclient/client_wallet.go

@@ -2,6 +2,7 @@ package btcclient
 
 import (
 	"fmt"
+	notifier "github.com/lightningnetwork/lnd/chainntnfs"
 
 	"github.com/btcsuite/btcd/btcjson"
 	"github.com/btcsuite/btcd/btcutil"
@@ -15,6 +16,17 @@ import (
 	"github.com/babylonlabs-io/vigilante/netparams"
 )
 
+type TxStatus int
+
+const (
+	TxNotFound TxStatus = iota
+	TxInMemPool
+	TxInChain
+)
+const (
+	txNotFoundErrMsgBitcoind = "No such mempool or blockchain transaction"
+)
+
 // NewWallet creates a new BTC wallet
 // used by vigilant submitter
 // a wallet is essentially a BTC client
@@ -125,3 +137,42 @@ func (c *Client) SignRawTransactionWithWallet(tx *wire.MsgTx) (*wire.MsgTx, bool
 func (c *Client) GetRawTransaction(txHash *chainhash.Hash) (*btcutil.Tx, error) {
 	return c.Client.GetRawTransaction(txHash)
 }
+
+func notifierStateToWalletState(state notifier.TxConfStatus) TxStatus {
+	switch state {
+	case notifier.TxNotFoundIndex:
+		return TxNotFound
+	case notifier.TxFoundMempool:
+		return TxInMemPool
+	case notifier.TxFoundIndex:
+		return TxInChain
+	case notifier.TxNotFoundManually:
+		return TxNotFound
+	case notifier.TxFoundManually:
+		return TxInChain
+	default:
+		panic(fmt.Sprintf("unknown notifier state: %s", state))
+	}
+}
+
+func (c *Client) getTxDetails(req notifier.ConfRequest, msg string) (*notifier.TxConfirmation, TxStatus, error) {
+	res, state, err := notifier.ConfDetailsFromTxIndex(c.Client, req, msg)
+
+	if err != nil {
+		return nil, TxNotFound, err
+	}
+
+	return res, notifierStateToWalletState(state), nil
+}
+
+// TxDetails Fetch info about transaction from mempool or blockchain, requires node to have enabled  transaction index
+func (c *Client) TxDetails(txHash *chainhash.Hash, pkScript []byte) (*notifier.TxConfirmation, TxStatus, error) {
+	req, err := notifier.NewConfRequest(txHash, pkScript)
+
+	if err != nil {
+		return nil, TxNotFound, err
+	}
+
+	return c.getTxDetails(req, txNotFoundErrMsgBitcoind)
+
+}

btcclient/interface.go

@@ -6,6 +6,7 @@ import (
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
+	notifier "github.com/lightningnetwork/lnd/chainntnfs"
 
 	"github.com/babylonlabs-io/vigilante/config"
 	"github.com/babylonlabs-io/vigilante/types"
@@ -39,4 +40,5 @@ type BTCWallet interface {
 	FundRawTransaction(tx *wire.MsgTx, opts btcjson.FundRawTransactionOpts, isWitness *bool) (*btcjson.FundRawTransactionResult, error)
 	SignRawTransactionWithWallet(tx *wire.MsgTx) (*wire.MsgTx, bool, error)
 	GetRawTransaction(txHash *chainhash.Hash) (*btcutil.Tx, error)
+	TxDetails(txHash *chainhash.Hash, pkScript []byte) (*notifier.TxConfirmation, TxStatus, error)
 }

e2etest/reporter_e2e_test.go

@@ -54,7 +54,7 @@ func (tm *TestManager) GenerateAndSubmitBlockNBlockStartingFromDepth(t *testing.
 }
 
 func TestReporter_BoostrapUnderFrequentBTCHeaders(t *testing.T) {
-	t.Parallel()
+	//t.Parallel() // todo(lazar): this test when run in parallel is very flaky, investigate why
 	// no need to much mature outputs, we are not going to submit transactions in this test
 	numMatureOutputs := uint32(150)
 

e2etest/submitter_e2e_test.go

@@ -5,6 +5,7 @@ package e2etest
 
 import (
 	"github.com/babylonlabs-io/vigilante/testutil"
+	promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"math/rand"
 	"testing"
 	"time"
@@ -188,4 +189,5 @@ func TestSubmitterSubmissionReplace(t *testing.T) {
 	blockWithOpReturnTransactions := tm.mineBlock(t)
 	// block should have 2 transactions, 1 from submitter and 1 coinbase
 	require.Equal(t, len(blockWithOpReturnTransactions.Transactions), 3)
+	require.True(t, promtestutil.ToFloat64(vigilantSubmitter.Metrics().ResentCheckpointsCounter) == 1)
 }

submitter/relayer/relayer.go

@@ -33,6 +33,10 @@ const (
 	dustThreshold  btcutil.Amount = 546
 )
 
+var (
+	TxNotFoundErr = errors.New("-5: No such mempool or blockchain transaction. Use gettransaction for wallet transactions")
+)
+
 type GetLatestCheckpointFunc func() (*store.StoredCheckpoint, bool, error)
 type GetRawTransactionFunc func(txHash *chainhash.Hash) (*btcutil.Tx, error)
 type SendTransactionFunc func(tx *wire.MsgTx) (*chainhash.Hash, error)
@@ -149,6 +153,18 @@ func (rl *Relayer) SendCheckpointToBTC(ckpt *ckpttypes.RawCheckpointWithMetaResp
 		return nil
 	}
 
+	return nil
+}
+
+// MaybeResubmitSecondCheckpointTx based on the resend interval attempts to resubmit 2nd ckpt tx with a bumped fee
+func (rl *Relayer) MaybeResubmitSecondCheckpointTx(ckpt *ckpttypes.RawCheckpointWithMetaResponse) error {
+	ckptEpoch := ckpt.Ckpt.EpochNum
+	if ckpt.Status != ckpttypes.Sealed {
+		rl.logger.Errorf("The checkpoint for epoch %v is not sealed", ckptEpoch)
+		rl.metrics.InvalidCheckpointCounter.Inc()
+		return nil
+	}
+
 	lastSubmittedEpoch := rl.lastSubmittedCheckpoint.Epoch
 	if ckptEpoch < lastSubmittedEpoch {
 		rl.logger.Errorf("The checkpoint for epoch %v is lower than the last submission for epoch %v",
@@ -158,55 +174,51 @@ func (rl *Relayer) SendCheckpointToBTC(ckpt *ckpttypes.RawCheckpointWithMetaResp
 		return nil
 	}
 
-	// now that the checkpoint has been sent, we should try to resend it
-	// if the resend interval has passed
 	durSeconds := uint(time.Since(rl.lastSubmittedCheckpoint.Ts).Seconds())
-	if durSeconds >= rl.config.ResendIntervalSeconds {
-		rl.logger.Debugf("The checkpoint for epoch %v was sent more than %v seconds ago but not included on BTC",
-			ckptEpoch, rl.config.ResendIntervalSeconds)
+	if durSeconds < rl.config.ResendIntervalSeconds {
+		return nil
+	}
 
-		bumpedFee := rl.calculateBumpedFee(rl.lastSubmittedCheckpoint)
+	rl.logger.Debugf("The checkpoint for epoch %v was sent more than %v seconds ago but not included on BTC",
+		ckptEpoch, rl.config.ResendIntervalSeconds)
 
-		// make sure the bumped fee is effective
-		if !rl.shouldResendCheckpoint(rl.lastSubmittedCheckpoint, bumpedFee) {
-			return nil
-		}
+	bumpedFee := rl.calculateBumpedFee(rl.lastSubmittedCheckpoint)
 
-		rl.logger.Debugf("Resending the second tx of the checkpoint %v, old fee of the second tx: %v Satoshis, txid: %s",
-			ckptEpoch, rl.lastSubmittedCheckpoint.Tx2.Fee, rl.lastSubmittedCheckpoint.Tx2.TxId.String())
+	// make sure the bumped fee is effective
+	if !rl.shouldResendCheckpoint(rl.lastSubmittedCheckpoint, bumpedFee) {
+		return nil
+	}
 
-		resubmittedTx2, err := rl.resendSecondTxOfCheckpointToBTC(rl.lastSubmittedCheckpoint.Tx2, bumpedFee)
-		if err != nil {
-			rl.metrics.FailedResentCheckpointsCounter.Inc()
-			return fmt.Errorf("failed to re-send the second tx of the checkpoint %v: %w", rl.lastSubmittedCheckpoint.Epoch, err)
-		}
+	rl.logger.Debugf("Resending the second tx of the checkpoint %v, old fee of the second tx: %v Satoshis, txid: %s",
+		ckptEpoch, rl.lastSubmittedCheckpoint.Tx2.Fee, rl.lastSubmittedCheckpoint.Tx2.TxId.String())
 
-		// record the metrics of the resent tx2
-		rl.metrics.NewSubmittedCheckpointSegmentGaugeVec.WithLabelValues(
-			strconv.Itoa(int(ckptEpoch)),
-			"1",
-			resubmittedTx2.TxId.String(),
-			strconv.Itoa(int(resubmittedTx2.Fee)),
-		).SetToCurrentTime()
-		rl.metrics.ResentCheckpointsCounter.Inc()
-
-		rl.logger.Infof("Successfully re-sent the second tx of the checkpoint %v, txid: %s, bumped fee: %v Satoshis",
-			rl.lastSubmittedCheckpoint.Epoch, resubmittedTx2.TxId.String(), resubmittedTx2.Fee)
-
-		// update the second tx of the last submitted checkpoint as it is replaced
-		rl.lastSubmittedCheckpoint.Tx2 = resubmittedTx2
-
-		err = storeCkptFunc(
-			rl.lastSubmittedCheckpoint.Tx1.Tx,
-			rl.lastSubmittedCheckpoint.Tx2.Tx,
-			rl.lastSubmittedCheckpoint.Epoch,
-		)
-		if err != nil {
-			return err
-		}
+	resubmittedTx2, err := rl.resendSecondTxOfCheckpointToBTC(rl.lastSubmittedCheckpoint.Tx2, bumpedFee)
+	if err != nil {
+		rl.metrics.FailedResentCheckpointsCounter.Inc()
+		return fmt.Errorf("failed to re-send the second tx of the checkpoint %v: %w", rl.lastSubmittedCheckpoint.Epoch, err)
 	}
 
-	return nil
+	// record the metrics of the resent tx2
+	rl.metrics.NewSubmittedCheckpointSegmentGaugeVec.WithLabelValues(
+		strconv.Itoa(int(ckptEpoch)),
+		"1",
+		resubmittedTx2.TxId.String(),
+		strconv.Itoa(int(resubmittedTx2.Fee)),
+	).SetToCurrentTime()
+	rl.metrics.ResentCheckpointsCounter.Inc()
+
+	rl.logger.Infof("Successfully re-sent the second tx of the checkpoint %v, txid: %s, bumped fee: %v Satoshis",
+		rl.lastSubmittedCheckpoint.Epoch, resubmittedTx2.TxId.String(), resubmittedTx2.Fee)
+
+	// update the second tx of the last submitted checkpoint as it is replaced
+	rl.lastSubmittedCheckpoint.Tx2 = resubmittedTx2
+
+	storedCkpt := store.NewStoredCheckpoint(
+		rl.lastSubmittedCheckpoint.Tx1.Tx,
+		rl.lastSubmittedCheckpoint.Tx2.Tx,
+		rl.lastSubmittedCheckpoint.Epoch,
+	)
+	return rl.store.PutCheckpoint(storedCkpt)
 }
 
 func (rl *Relayer) shouldSendCompleteCkpt(ckptEpoch uint64) bool {
@@ -240,6 +252,19 @@ func (rl *Relayer) calculateBumpedFee(ckptInfo *types.CheckpointInfo) btcutil.Am
 
 // resendSecondTxOfCheckpointToBTC resends the second tx of the checkpoint with bumpedFee
 func (rl *Relayer) resendSecondTxOfCheckpointToBTC(tx2 *types.BtcTxInfo, bumpedFee btcutil.Amount) (*types.BtcTxInfo, error) {
+	_, status, err := rl.TxDetails(rl.lastSubmittedCheckpoint.Tx2.TxId,
+		rl.lastSubmittedCheckpoint.Tx2.Tx.TxOut[changePosition].PkScript)
+	if err != nil {
+		return nil, err
+	}
+
+	// No need to resend, transaction already confirmed
+	if status != btcclient.TxInMemPool && status != btcclient.TxNotFound {
+		rl.logger.Debugf("Transaction %v is already confirmed or has an unexpected state: %v",
+			rl.lastSubmittedCheckpoint.Tx2.TxId, status)
+		return nil, nil
+	}
+
 	// set output value of the second tx to be the balance minus the bumped fee
 	// if the bumped fee is higher than the balance, then set the bumped fee to
 	// be equal to the balance to ensure the output value is not negative

submitter/submitter.go

@@ -202,15 +202,22 @@ func (s *Submitter) processCheckpoints() {
 		select {
 		case ckpt := <-s.poller.GetSealedCheckpointChan():
 			s.logger.Infof("A sealed raw checkpoint for epoch %v is found", ckpt.Ckpt.EpochNum)
-			err := s.relayer.SendCheckpointToBTC(ckpt)
-			if err != nil {
+			if err := s.relayer.SendCheckpointToBTC(ckpt); err != nil {
 				s.logger.Errorf("Failed to submit the raw checkpoint for %v: %v", ckpt.Ckpt.EpochNum, err)
 				s.metrics.FailedCheckpointsCounter.Inc()
 			}
+			if err := s.relayer.MaybeResubmitSecondCheckpointTx(ckpt); err != nil {
+				s.logger.Errorf("Failed to resubmit the raw checkpoint for %v: %v", ckpt.Ckpt.EpochNum, err)
+				s.metrics.FailedCheckpointsCounter.Inc()
+			}
 			s.metrics.SecondsSinceLastCheckpointGauge.Set(0)
 		case <-quit:
 			// We have been asked to stop
 			return
 		}
 	}
 }
+
+func (s *Submitter) Metrics() *metrics.SubmitterMetrics {
+	return s.metrics
+}