chore(submitter): refactoring points

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

@@ -10,6 +10,7 @@ import (
 	"github.com/lightningnetwork/lnd/kvdb"
 	"math"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/babylonlabs-io/babylon/btctxformatter"
@@ -33,6 +34,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 +154,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 +175,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 +253,16 @@ 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) {
+	found, err := rl.inMempool()
+	if err != nil {
+		return nil, err
+	}
+
+	// No need to resend, transaction already confirmed
+	if !found {
+		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
@@ -272,6 +295,20 @@ func (rl *Relayer) resendSecondTxOfCheckpointToBTC(tx2 *types.BtcTxInfo, bumpedF
 	return tx2, nil
 }
 
+func (rl *Relayer) inMempool() (bool, error) {
+	_, err := rl.GetRawTransaction(rl.lastSubmittedCheckpoint.Tx2.TxId)
+	if err != nil {
+		if strings.Contains(err.Error(), TxNotFoundErr.Error()) {
+			return false, nil
+		}
+		// Return the error if it's not a "not found" error
+		return false, err
+	}
+
+	// the transaction is in the mempool
+	return true, nil
+}
+
 // calcMinRelayFee returns the minimum transaction fee required for a
 // transaction with the passed serialized size to be accepted into the memory
 // pool and relayed.

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
+}