Add configurable rebalance strategy for bridge pegout

    Introduce UTXO_SPLIT strategy that splits bridge conversions into
    multiple transactions based on BridgeTransactionMin, alongside the
    existing ALL_AT_ONCE behavior. Strategy is configured via
    REBALANCE_STRATEGY env variable.

Author: volodymyrzahanych

cmd/application/lps/application.go

@@ -83,7 +83,10 @@ func NewApplication(initCtx context.Context, env environment.Environment, timeou
 	liquidityProvider := registry.NewLiquidityProvider(dbRegistry, rootstockRegistry, btcRegistry, messagingRegistry)
 	mutexes := environment.NewApplicationMutexes()
 
-	useCaseRegistry := registry.NewUseCaseRegistry(env, rootstockRegistry, btcRegistry, dbRegistry, liquidityProvider, messagingRegistry, mutexes)
+	useCaseRegistry, err := registry.NewUseCaseRegistry(env, rootstockRegistry, btcRegistry, dbRegistry, liquidityProvider, messagingRegistry, mutexes)
+	if err != nil {
+		log.Fatal("Error creating use case registry:", err)
+	}
 	watcherRegistry := registry.NewWatcherRegistry(env, useCaseRegistry, rootstockRegistry, btcRegistry, liquidityProvider, messagingRegistry, watcher.NewApplicationTickers(), timeouts)
 	return &Application{
 		env:               env,

docker-compose/local/docker-compose.lps.yml

@@ -87,6 +87,7 @@ services:
       - BTC_RELEASE_WATCHER_START_BLOCK
       - BTC_RELEASE_WATCHER_PAGE_SIZE
       - BTC_RELEASE_CHECK_TIMEOUT
+      - REBALANCE_STRATEGY
     ports:
       - "8080:8080"
     volumes:

internal/adapters/entrypoints/watcher/pegout_bridge_watcher_test.go

@@ -35,7 +35,7 @@ func TestPegoutBridgeWatcher_Start(t *testing.T) {
 	bridge := &mocks.BridgeMock{}
 	bridge.On("GetAddress").Return(test.AnyAddress)
 	mutexes := environment.NewApplicationMutexes()
-	bridgeUseCase := pegout.NewBridgePegoutUseCase(pegoutRepository, providerMock, rskWallet, blockchain.RskContracts{Bridge: bridge}, mutexes.RskWalletMutex())
+	bridgeUseCase := pegout.NewBridgePegoutUseCase(pegoutRepository, providerMock, rskWallet, blockchain.RskContracts{Bridge: bridge}, mutexes.RskWalletMutex(), pegout.AllAtOnce)
 	getUseCase := w.NewGetWatchedPegoutQuoteUseCase(pegoutRepository)
 	bridgeWatcher := watcher.NewPegoutBridgeWatcher(getUseCase, bridgeUseCase, ticker)
 	resetMocks := func() {

internal/configuration/environment/environment.go

@@ -168,6 +168,7 @@ type PegoutEnv struct {
 	DepositCacheStartBlock      uint64 `env:"PEGOUT_DEPOSIT_CACHE_START_BLOCK"`
 	BtcReleaseWatcherStartBlock uint64 `env:"BTC_RELEASE_WATCHER_START_BLOCK"`
 	BtcReleaseWatcherPageSize   uint64 `env:"BTC_RELEASE_WATCHER_PAGE_SIZE"`
+	RebalanceStrategy           string `env:"REBALANCE_STRATEGY" validate:"required,oneof=ALL_AT_ONCE UTXO_SPLIT"`
 }
 
 type CaptchaEnv struct {

internal/configuration/environment/environment_reader_test.go

@@ -34,6 +34,7 @@ func setUpEnv(t *testing.T) {
 		"BTC_RELEASE_WATCHER_START_BLOCK":      "1",
 		"USE_SEGWIT_FEDERATION":                "true",
 		"ALLOWED_ORIGINS":                      "http://example.com,http://example2.com",
+		"REBALANCE_STRATEGY":                   "ALL_AT_ONCE",
 	}
 	const envFilePath = "../../../sample-config.env"
 	envFile, err := os.ReadFile(envFilePath)

internal/configuration/registry/usecase.go

@@ -92,7 +92,11 @@ func NewUseCaseRegistry(
 	liquidityProvider *dataproviders.LocalLiquidityProvider,
 	messaging *Messaging,
 	mutexes entities.ApplicationMutexes,
-) *UseCaseRegistry {
+) (*UseCaseRegistry, error) {
+	rebalanceStrategy, err := pegout.ParseRebalanceStrategy(env.Pegout.RebalanceStrategy)
+	if err != nil {
+		return nil, err
+	}
 	return &UseCaseRegistry{
 		summariesUseCase: reports.NewSummariesUseCase(
 			databaseRegistry.PeginRepository,
@@ -259,6 +263,7 @@ func NewUseCaseRegistry(
 			rskRegistry.Wallet,
 			rskRegistry.Contracts,
 			mutexes.RskWalletMutex(),
+			rebalanceStrategy,
 		),
 		peginStatusUseCase:        pegin.NewStatusUseCase(databaseRegistry.PeginRepository),
 		pegoutStatusUseCase:       pegout.NewStatusUseCase(databaseRegistry.PegoutRepository),
@@ -348,7 +353,7 @@ func NewUseCaseRegistry(
 			messaging.Rpc,
 			utils.Scale,
 		),
-	}
+	}, nil
 }
 
 func (registry *UseCaseRegistry) GetPeginQuoteUseCase() *pegin.GetQuoteUseCase {

internal/configuration/registry/usecase_test.go

@@ -30,6 +30,7 @@ func TestNewUseCaseRegistry(t *testing.T) {
 				BridgeAddress:               "0x0000000000000000000000000000000001000006",
 			},
 			Btc: environment.BtcEnv{Network: "testnet"},
+			Pegout: environment.PegoutEnv{RebalanceStrategy: "ALL_AT_ONCE"},
 		}
 
 		client := &mocks.DbClientBindingMock{}
@@ -55,8 +56,8 @@ func TestNewUseCaseRegistry(t *testing.T) {
 		lp := registry.NewLiquidityProvider(dbRegistry, rskRegistry, btcRegistry, messagingRegistry)
 		mutexes := environment.NewApplicationMutexes()
 
-		useCaseRegistry := registry.NewUseCaseRegistry(env, rskRegistry, btcRegistry, dbRegistry, lp, messagingRegistry, mutexes)
-
+		useCaseRegistry, err := registry.NewUseCaseRegistry(env, rskRegistry, btcRegistry, dbRegistry, lp, messagingRegistry, mutexes)
+		require.NoError(t, err)
 		require.NotNil(t, useCaseRegistry)
 		value := reflect.ValueOf(useCaseRegistry).Elem()
 		for i := 0; i < value.NumField(); i++ {

internal/configuration/registry/watcher_test.go

@@ -29,6 +29,7 @@ func TestNewWatcherRegistry(t *testing.T) {
 				BridgeAddress:               "0x0000000000000000000000000000000001000006",
 			},
 			Btc: environment.BtcEnv{Network: "testnet"},
+			Pegout: environment.PegoutEnv{RebalanceStrategy: "ALL_AT_ONCE"},
 		}
 
 		client := &mocks.DbClientBindingMock{}
@@ -53,7 +54,8 @@ func TestNewWatcherRegistry(t *testing.T) {
 		messagingRegistry := registry.NewMessagingRegistry(context.Background(), environment.Environment{}, rskClient, connection, registry.ExternalRpc{})
 		lp := registry.NewLiquidityProvider(dbRegistry, rskRegistry, btcRegistry, messagingRegistry)
 		mutexes := environment.NewApplicationMutexes()
-		useCaseRegistry := registry.NewUseCaseRegistry(env, rskRegistry, btcRegistry, dbRegistry, lp, messagingRegistry, mutexes)
+		useCaseRegistry, err := registry.NewUseCaseRegistry(env, rskRegistry, btcRegistry, dbRegistry, lp, messagingRegistry, mutexes)
+		require.NoError(t, err)
 
 		watcherRegistry := registry.NewWatcherRegistry(env, useCaseRegistry, rskRegistry, btcRegistry, lp, messagingRegistry, watcher.NewApplicationTickers(), environment.DefaultTimeouts())
 

internal/usecases/pegout/bridge_pegout.go

@@ -3,6 +3,7 @@ package pegout
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 
 	"github.com/rsksmart/liquidity-provider-server/internal/entities"
@@ -20,12 +21,31 @@ const (
 	BridgeConversionGasPrice = 60000000
 )
 
+type RebalanceStrategy string
+
+const (
+	AllAtOnce RebalanceStrategy = "ALL_AT_ONCE"
+	UtxoSplit RebalanceStrategy = "UTXO_SPLIT"
+)
+
+func ParseRebalanceStrategy(s string) (RebalanceStrategy, error) {
+	switch s {
+	case string(AllAtOnce):
+		return AllAtOnce, nil
+	case string(UtxoSplit):
+		return UtxoSplit, nil
+	default:
+		return "", fmt.Errorf("unknown rebalance strategy: %q", s)
+	}
+}
+
 type BridgePegoutUseCase struct {
 	quoteRepository quote.PegoutQuoteRepository
 	pegoutProvider  liquidity_provider.PegoutLiquidityProvider
 	rskWallet       blockchain.RootstockWallet
 	contracts       blockchain.RskContracts
 	rskWalletMutex  sync.Locker
+	strategy        RebalanceStrategy
 }
 
 func NewBridgePegoutUseCase(
@@ -34,19 +54,21 @@ func NewBridgePegoutUseCase(
 	rskWallet blockchain.RootstockWallet,
 	contracts blockchain.RskContracts,
 	rskWalletMutex sync.Locker,
+	strategy RebalanceStrategy,
 ) *BridgePegoutUseCase {
 	return &BridgePegoutUseCase{
 		quoteRepository: quoteRepository,
 		pegoutProvider:  pegoutProvider,
 		rskWallet:       rskWallet,
 		contracts:       contracts,
 		rskWalletMutex:  rskWalletMutex,
+		strategy:        strategy,
 	}
 }
 
 func (useCase *BridgePegoutUseCase) Run(ctx context.Context, watchedQuotes ...quote.WatchedPegoutQuote) error {
 	var err error
-	var balance, totalValue *entities.Wei
+	var totalValue *entities.Wei
 
 	totalValue, err = useCase.calculateTotalToPegout(watchedQuotes)
 	if err != nil {
@@ -66,6 +88,18 @@ func (useCase *BridgePegoutUseCase) Run(ctx context.Context, watchedQuotes ...qu
 	useCase.rskWalletMutex.Lock()
 	defer useCase.rskWalletMutex.Unlock()
 
+	switch useCase.strategy {
+	case UtxoSplit:
+		return useCase.runUtxoSplit(ctx, totalValue, pegoutConfig, watchedQuotes)
+	default:
+		return useCase.runAllAtOnce(ctx, totalValue, watchedQuotes)
+	}
+}
+
+func (useCase *BridgePegoutUseCase) runAllAtOnce(ctx context.Context, totalValue *entities.Wei, watchedQuotes []quote.WatchedPegoutQuote) error {
+	var balance *entities.Wei
+	var err error
+
 	requiredBalance := new(entities.Wei).Add(totalValue, entities.NewWei(BridgeConversionGasLimit*BridgeConversionGasPrice))
 	if balance, err = useCase.rskWallet.GetBalance(ctx); err != nil {
 		return usecases.WrapUseCaseError(usecases.BridgePegoutId, err)
@@ -86,6 +120,77 @@ func (useCase *BridgePegoutUseCase) Run(ctx context.Context, watchedQuotes ...qu
 	return nil
 }
 
+func (useCase *BridgePegoutUseCase) runUtxoSplit(
+	ctx context.Context,
+	totalValue *entities.Wei,
+	pegoutConfig liquidity_provider.PegoutConfiguration,
+	watchedQuotes []quote.WatchedPegoutQuote,
+) error {
+	var balance *entities.Wei
+	var err error
+
+	bridgeMin := pegoutConfig.BridgeTransactionMin
+	numTxs, _ := new(entities.Wei).Div(totalValue, bridgeMin)
+	remainder := new(entities.Wei).Sub(totalValue, new(entities.Wei).Mul(numTxs, bridgeMin))
+
+	gasPerTx := entities.NewWei(BridgeConversionGasLimit * BridgeConversionGasPrice)
+	requiredBalance := new(entities.Wei).Add(totalValue, new(entities.Wei).Mul(numTxs, gasPerTx))
+	if balance, err = useCase.rskWallet.GetBalance(ctx); err != nil {
+		return usecases.WrapUseCaseError(usecases.BridgePegoutId, err)
+	} else if balance.Cmp(requiredBalance) < 0 {
+		return usecases.WrapUseCaseError(usecases.BridgePegoutId, usecases.InsufficientAmountError)
+	}
+
+	bridgeAddress := useCase.contracts.Bridge.GetAddress()
+	n := numTxs.Uint64()
+
+	if n == 1 {
+		config := blockchain.NewTransactionConfig(totalValue, BridgeConversionGasLimit, entities.NewWei(BridgeConversionGasPrice))
+		receipt, txErr := useCase.rskWallet.SendRbtc(ctx, config, bridgeAddress)
+		if txErr == nil {
+			log.Debugf("%s: transaction sent to the bridge successfully (%s)", usecases.BridgePegoutId, receipt.TransactionHash)
+		}
+		err = useCase.updateQuotes(ctx, receipt, txErr, watchedQuotes)
+		if err != nil {
+			return usecases.WrapUseCaseError(usecases.BridgePegoutId, err)
+		}
+		return nil
+	}
+
+	// N >= 2: first chunk absorbs the remainder
+	firstChunk := new(entities.Wei).Add(bridgeMin.Copy(), remainder)
+	var receipt blockchain.TransactionReceipt
+	var txErr error
+
+	config := blockchain.NewTransactionConfig(firstChunk, BridgeConversionGasLimit, entities.NewWei(BridgeConversionGasPrice))
+	receipt, txErr = useCase.rskWallet.SendRbtc(ctx, config, bridgeAddress)
+	if txErr == nil {
+		log.Debugf("%s: split tx 1/%d sent to the bridge successfully (%s)", usecases.BridgePegoutId, n, receipt.TransactionHash)
+	} else {
+		err = useCase.updateQuotes(ctx, receipt, txErr, watchedQuotes)
+		if err != nil {
+			return usecases.WrapUseCaseError(usecases.BridgePegoutId, err)
+		}
+		return usecases.WrapUseCaseError(usecases.BridgePegoutId, txErr)
+	}
+
+	for i := uint64(1); i < n; i++ {
+		config = blockchain.NewTransactionConfig(bridgeMin.Copy(), BridgeConversionGasLimit, entities.NewWei(BridgeConversionGasPrice))
+		receipt, txErr = useCase.rskWallet.SendRbtc(ctx, config, bridgeAddress)
+		if txErr == nil {
+			log.Debugf("%s: split tx %d/%d sent to the bridge successfully (%s)", usecases.BridgePegoutId, i+1, n, receipt.TransactionHash)
+		} else {
+			break
+		}
+	}
+
+	err = useCase.updateQuotes(ctx, receipt, txErr, watchedQuotes)
+	if err != nil {
+		return usecases.WrapUseCaseError(usecases.BridgePegoutId, err)
+	}
+	return nil
+}
+
 func (useCase *BridgePegoutUseCase) updateQuotes(
 	ctx context.Context,
 	receipt blockchain.TransactionReceipt,