fix(swe): wait for stacking tx to be k-deep (#106)

Wait for the stacking transaction to be k-deep before submitting the
inclusion proof

CHANGELOG.md

@@ -38,7 +38,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ## Unreleased
 
 ### Improvements
+
 * [#105](https://github.com/babylonlabs-io/vigilante/pull/105) Measure latency
+* [#106](https://github.com/babylonlabs-io/vigilante/pull/106) Wait for stacking tx to be k-deep
+
 
 ## v0.16.0
 

btcstaking-tracker/stakingeventwatcher/expected_babylon_client.go

@@ -2,19 +2,31 @@ package stakingeventwatcher
 
 import (
 	"context"
+	sdkerrors "cosmossdk.io/errors"
 	"fmt"
-
-	btcctypes "github.com/babylonlabs-io/babylon/x/btccheckpoint/types"
-
-	"cosmossdk.io/errors"
+	"github.com/avast/retry-go/v4"
+	btclctypes "github.com/babylonlabs-io/babylon/x/btclightclient/types"
+	"github.com/babylonlabs-io/vigilante/config"
+	"github.com/cosmos/cosmos-sdk/client"
+	"strings"
+	"time"
+
+	"errors"
 	bbnclient "github.com/babylonlabs-io/babylon/client/client"
 	bbn "github.com/babylonlabs-io/babylon/types"
+	btcctypes "github.com/babylonlabs-io/babylon/x/btccheckpoint/types"
 	btcstakingtypes "github.com/babylonlabs-io/babylon/x/btcstaking/types"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/cosmos/cosmos-sdk/types/query"
 )
 
+var (
+	ErrHeaderNotKnownToBabylon       = errors.New("btc header not known to babylon")
+	ErrHeaderOnBabylonLCFork         = errors.New("btc header is on babylon btc light client fork")
+	ErrBabylonBtcLightClientNotReady = errors.New("babylon btc light client is not ready to receive delegation")
+)
+
 type Delegation struct {
 	StakingTx             *wire.MsgTx
 	StakingOutputIdx      uint32
@@ -23,6 +35,10 @@ type Delegation struct {
 	HasProof              bool
 }
 
+type BabylonParams struct {
+	ConfirmationTimeBlocks uint32 // K-deep
+}
+
 type BabylonNodeAdapter interface {
 	DelegationsByStatus(status btcstakingtypes.BTCDelegationStatus, offset uint64, limit uint64) ([]Delegation, error)
 	IsDelegationActive(stakingTxHash chainhash.Hash) (bool, error)
@@ -31,17 +47,21 @@ type BabylonNodeAdapter interface {
 		inclusionProof *btcstakingtypes.InclusionProof) error
 	BtcClientTipHeight() (uint32, error)
 	ActivateDelegation(ctx context.Context, stakingTxHash chainhash.Hash, proof *btcctypes.BTCSpvProof) error
+	QueryHeaderDepth(headerHash *chainhash.Hash) (uint32, error)
+	Params() (*BabylonParams, error)
 }
 
 type BabylonClientAdapter struct {
 	babylonClient *bbnclient.Client
+	cfg           *config.BTCStakingTrackerConfig
 }
 
 var _ BabylonNodeAdapter = (*BabylonClientAdapter)(nil)
 
-func NewBabylonClientAdapter(babylonClient *bbnclient.Client) *BabylonClientAdapter {
+func NewBabylonClientAdapter(babylonClient *bbnclient.Client, cfg *config.BTCStakingTrackerConfig) *BabylonClientAdapter {
 	return &BabylonClientAdapter{
 		babylonClient: babylonClient,
+		cfg:           cfg,
 	}
 }
 
@@ -127,7 +147,7 @@ func (bca *BabylonClientAdapter) ReportUnbonding(
 		StakeSpendingTxInclusionProof: inclusionProof,
 	}
 
-	resp, err := bca.babylonClient.ReliablySendMsg(ctx, &msg, []*errors.Error{}, []*errors.Error{})
+	resp, err := bca.babylonClient.ReliablySendMsg(ctx, &msg, []*sdkerrors.Error{}, []*sdkerrors.Error{})
 	if err != nil && resp != nil {
 		return fmt.Errorf("msg MsgBTCUndelegate failed exeuction with code %d and error %w", resp.Code, err)
 	}
@@ -161,7 +181,7 @@ func (bca *BabylonClientAdapter) ActivateDelegation(
 		StakingTxInclusionProof: btcstakingtypes.NewInclusionProofFromSpvProof(proof),
 	}
 
-	resp, err := bca.babylonClient.ReliablySendMsg(ctx, &msg, []*errors.Error{}, []*errors.Error{})
+	resp, err := bca.babylonClient.ReliablySendMsg(ctx, &msg, []*sdkerrors.Error{}, []*sdkerrors.Error{})
 
 	if err != nil && resp != nil {
 		return fmt.Errorf("msg MsgAddBTCDelegationInclusionProof failed exeuction with code %d and error %w", resp.Code, err)
@@ -173,3 +193,52 @@ func (bca *BabylonClientAdapter) ActivateDelegation(
 
 	return nil
 }
+
+func (bca *BabylonClientAdapter) QueryHeaderDepth(headerHash *chainhash.Hash) (uint32, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	clientCtx := client.Context{Client: bca.babylonClient.RPCClient}
+	queryClient := btclctypes.NewQueryClient(clientCtx)
+
+	var response *btclctypes.QueryHeaderDepthResponse
+	if err := retry.Do(func() error {
+		depthResponse, err := queryClient.HeaderDepth(ctx, &btclctypes.QueryHeaderDepthRequest{Hash: headerHash.String()})
+		if err != nil {
+			return err
+		}
+		response = depthResponse
+		return nil
+	},
+		retry.Attempts(5),
+		retry.MaxDelay(bca.cfg.RetrySubmitUnbondingTxInterval),
+		retry.MaxJitter(bca.cfg.RetryJitter),
+		retry.LastErrorOnly(true)); err != nil {
+		if strings.Contains(err.Error(), btclctypes.ErrHeaderDoesNotExist.Error()) {
+			return 0, fmt.Errorf("%s: %w", err.Error(), ErrHeaderNotKnownToBabylon)
+		}
+		return 0, err
+	}
+
+	return response.Depth, nil
+}
+
+func (bca *BabylonClientAdapter) Params() (*BabylonParams, error) {
+	var bccParams *btcctypes.Params
+	if err := retry.Do(func() error {
+		response, err := bca.babylonClient.BTCCheckpointParams()
+		if err != nil {
+			return err
+		}
+		bccParams = &response.Params
+
+		return nil
+	}, retry.Attempts(5),
+		retry.MaxDelay(bca.cfg.RetrySubmitUnbondingTxInterval),
+		retry.MaxJitter(bca.cfg.RetryJitter),
+		retry.LastErrorOnly(true)); err != nil {
+		return nil, err
+	}
+
+	return &BabylonParams{ConfirmationTimeBlocks: bccParams.BtcConfirmationDepth}, nil
+}

btcstaking-tracker/stakingeventwatcher/stakingeventwatcher.go

@@ -577,6 +577,12 @@ func (sew *StakingEventWatcher) handlerVerifiedDelegations() {
 // checkBtcForStakingTx gets a snapshot of current Delegations in cache
 // checks if staking tx is in BTC, generates a proof and invokes sending of MsgAddBTCDelegationInclusionProof
 func (sew *StakingEventWatcher) checkBtcForStakingTx() {
+	params, err := sew.babylonNodeAdapter.Params()
+	if err != nil {
+		sew.logger.Errorf("error getting tx params %v", err)
+		return
+	}
+
 	for del := range sew.pendingTracker.DelegationsIter() {
 		if del.ActivationInProgress {
 			continue
@@ -602,13 +608,17 @@ func (sew *StakingEventWatcher) checkBtcForStakingTx() {
 			continue
 		}
 
-		go sew.activateBtcDelegation(txHash, proof)
+		go sew.activateBtcDelegation(txHash, proof, details.Block.BlockHash(), params.ConfirmationTimeBlocks)
 	}
 }
 
 // activateBtcDelegation invokes bbn client and send MsgAddBTCDelegationInclusionProof
 func (sew *StakingEventWatcher) activateBtcDelegation(
-	stakingTxHash chainhash.Hash, proof *btcctypes.BTCSpvProof) {
+	stakingTxHash chainhash.Hash,
+	proof *btcctypes.BTCSpvProof,
+	inclusionBlockHash chainhash.Hash,
+	requiredDepth uint32,
+) {
 	ctx, cancel := sew.quitContext()
 	defer cancel()
 
@@ -618,6 +628,8 @@ func (sew *StakingEventWatcher) activateBtcDelegation(
 		sew.logger.Debugf("skipping tx %s is not in pending tracker, err: %v", stakingTxHash, err)
 	}
 
+	sew.waitForRequiredDepth(ctx, stakingTxHash, &inclusionBlockHash, requiredDepth)
+
 	_ = retry.Do(func() error {
 		verified, err := sew.babylonNodeAdapter.IsDelegationVerified(stakingTxHash)
 		if err != nil {
@@ -652,13 +664,52 @@ func (sew *StakingEventWatcher) activateBtcDelegation(
 	)
 }
 
+func (sew *StakingEventWatcher) waitForRequiredDepth(
+	ctx context.Context,
+	stakingTxHash chainhash.Hash,
+	inclusionBlockHash *chainhash.Hash,
+	requiredDepth uint32,
+) {
+	var depth uint32
+	_ = retry.Do(func() error {
+		var err error
+		depth, err = sew.babylonNodeAdapter.QueryHeaderDepth(inclusionBlockHash)
+		if err != nil {
+			// If the header is not known to babylon, or it is on LCFork, then most probably
+			// lc is not up to date, we should retry sending delegation after some time.
+			if errors.Is(err, ErrHeaderNotKnownToBabylon) || errors.Is(err, ErrHeaderOnBabylonLCFork) {
+				return fmt.Errorf("btc light client error %s: %w", err.Error(), ErrBabylonBtcLightClientNotReady)
+			}
+
+			return fmt.Errorf("error while getting delegation data: %w", err)
+		}
+
+		if depth < requiredDepth {
+			return fmt.Errorf("btc lc not ready, required depth: %d, current depth: %d: %w",
+				requiredDepth, depth, ErrBabylonBtcLightClientNotReady)
+		}
+
+		return nil
+	},
+		retry.Context(ctx),
+		retryForever,
+		fixedDelyTypeWithJitter,
+		retry.MaxDelay(sew.cfg.RetrySubmitUnbondingTxInterval),
+		retry.MaxJitter(sew.cfg.RetryJitter),
+		retry.OnRetry(func(n uint, err error) {
+			sew.logger.Debugf("waiting for staking tx: %s to be k-deep. Current[%d], required[%d]. "+
+				"Attempt: %d. Err: %v", stakingTxHash, depth, requiredDepth, n, err)
+		}),
+	)
+}
+
 func (sew *StakingEventWatcher) latency(method string) func() {
 	startTime := time.Now()
 	return func() {
 		duration := time.Since(startTime)
 		sew.logger.Debug("execution time",
 			zap.String("method", method),
-			zap.Duration("latency", duration))
+			zap.String("latency", duration.String()))
 		sew.metrics.MethodExecutionLatency.WithLabelValues(method).Observe(duration.Seconds())
 	}
 }

btcstaking-tracker/tracker.go

@@ -68,7 +68,7 @@ func NewBTCStakingTracker(
 	logger := parentLogger.With(zap.String("module", "btcstaking-tracker"))
 
 	// watcher routine
-	babylonAdapter := uw.NewBabylonClientAdapter(bbnClient)
+	babylonAdapter := uw.NewBabylonClientAdapter(bbnClient, cfg)
 	watcher := uw.NewStakingEventWatcher(
 		btcNotifier,
 		btcClient,

e2etest/unbondingwatcher_e2e_test.go

@@ -4,8 +4,11 @@
 package e2etest
 
 import (
+	"fmt"
 	btcstakingtypes "github.com/babylonlabs-io/babylon/x/btcstaking/types"
+	promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"go.uber.org/zap"
+	"sync"
 	"testing"
 	"time"
 
@@ -180,15 +183,31 @@ func TestActivatingDelegation(t *testing.T) {
 		return err == nil
 	}, eventuallyWaitTimeOut, eventuallyPollTime)
 
-	// insert k empty blocks to Bitcoin
-	btccParamsResp, err := tm.BabylonClient.BTCCheckpointParams()
-	require.NoError(t, err)
-	btccParams := btccParamsResp.Params
-	for i := 0; i < int(btccParams.BtcConfirmationDepth); i++ {
-		tm.mineBlock(t)
-	}
-
-	tm.CatchUpBTCLightClient(t)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		// We want to introduce a latency to make sure that we are not trying to submit inclusion proof while the
+		// staking tx is not yet K-deep
+		time.Sleep(10 * time.Second)
+		// Insert k empty blocks to Bitcoin
+		btccParamsResp, err := tm.BabylonClient.BTCCheckpointParams()
+		if err != nil {
+			fmt.Println("Error fetching BTCCheckpointParams:", err)
+			return
+		}
+		for i := 0; i < int(btccParamsResp.Params.BtcConfirmationDepth); i++ {
+			tm.mineBlock(t)
+		}
+		tm.CatchUpBTCLightClient(t)
+	}()
+
+	wg.Wait()
+
+	// make sure we didn't submit any "invalid" incl proof
+	require.Eventually(t, func() bool {
+		return promtestutil.ToFloat64(stakingTrackerMetrics.FailedReportedActivateDelegations) == 0
+	}, eventuallyWaitTimeOut, eventuallyPollTime)
 
 	// created delegation lacks inclusion proof, once created it will be in
 	// pending status, once convenant signatures are added it will be in verified status,