loop: integrate sweepbatcher to loopout flow

Author: GeorgeTsagk

client.go

@@ -17,6 +17,7 @@ import (
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/swap"
 	"github.com/lightninglabs/loop/sweep"
+	"github.com/lightninglabs/loop/sweepbatcher"
 	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/routing/route"
@@ -61,7 +62,7 @@ var (
 	// probeTimeout is the maximum time until a probe is allowed to take.
 	probeTimeout = 3 * time.Minute
 
-	republishDelay = 10 * time.Second
+	repushDelay = 1 * time.Second
 
 	// MinerFeeEstimationFailed is a magic number that is returned in a
 	// quote call as the miner fee if the fee estimation in lnd's wallet
@@ -134,7 +135,8 @@ type ClientConfig struct {
 
 // NewClient returns a new instance to initiate swaps with.
 func NewClient(dbDir string, loopDB loopdb.SwapStore,
-	cfg *ClientConfig) (*Client, func(), error) {
+	sweeperDb sweepbatcher.BatcherStore, cfg *ClientConfig) (
+	*Client, func(), error) {
 
 	lsatStore, err := lsat.NewFileStore(dbDir)
 	if err != nil {
@@ -162,27 +164,36 @@ func NewClient(dbDir string, loopDB loopdb.SwapStore,
 		Lnd: cfg.Lnd,
 	}
 
+	verifySchnorrSig := func(pubKey *btcec.PublicKey, hash, sig []byte) error {
+		schnorrSig, err := schnorr.ParseSignature(sig)
+		if err != nil {
+			return err
+		}
+
+		if !schnorrSig.Verify(hash, pubKey) {
+			return fmt.Errorf("invalid signature")
+		}
+
+		return nil
+	}
+
+	batcher := sweepbatcher.NewBatcher(
+		cfg.Lnd.WalletKit, cfg.Lnd.ChainNotifier, cfg.Lnd.Signer,
+		swapServerClient.MultiMuSig2SignSweep, verifySchnorrSig,
+		cfg.Lnd.ChainParams, sweeperDb, loopDB,
+	)
+
 	executor := newExecutor(&executorConfig{
 		lnd:                 cfg.Lnd,
 		store:               loopDB,
 		sweeper:             sweeper,
+		batcher:             batcher,
 		createExpiryTimer:   config.CreateExpiryTimer,
 		loopOutMaxParts:     cfg.LoopOutMaxParts,
 		totalPaymentTimeout: cfg.TotalPaymentTimeout,
 		maxPaymentRetries:   cfg.MaxPaymentRetries,
 		cancelSwap:          swapServerClient.CancelLoopOutSwap,
-		verifySchnorrSig: func(pubKey *btcec.PublicKey, hash, sig []byte) error {
-			schnorrSig, err := schnorr.ParseSignature(sig)
-			if err != nil {
-				return err
-			}
-
-			if !schnorrSig.Verify(hash, pubKey) {
-				return fmt.Errorf("invalid signature")
-			}
-
-			return nil
-		},
+		verifySchnorrSig:    verifySchnorrSig,
 	})
 
 	client := &Client{

executor.go

@@ -13,6 +13,7 @@ import (
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/sweep"
+	"github.com/lightninglabs/loop/sweepbatcher"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/queue"
 )
@@ -23,6 +24,8 @@ type executorConfig struct {
 
 	sweeper *sweep.Sweeper
 
+	batcher *sweepbatcher.Batcher
+
 	store loopdb.SwapStore
 
 	createExpiryTimer func(expiry time.Duration) <-chan time.Time
@@ -71,6 +74,7 @@ func (s *executor) run(mainCtx context.Context,
 		err            error
 		blockEpochChan <-chan int32
 		blockErrorChan <-chan error
+		batcherErrChan chan error
 	)
 
 	for {
@@ -121,6 +125,21 @@ func (s *executor) run(mainCtx context.Context,
 		return mainCtx.Err()
 	}
 
+	batcherErrChan = make(chan error, 1)
+
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+
+		err := s.batcher.Run(mainCtx)
+		if err != nil {
+			select {
+			case batcherErrChan <- err:
+			case <-mainCtx.Done():
+			}
+		}
+	}()
+
 	// Start main event loop.
 	log.Infof("Starting event loop at height %v", height)
 
@@ -156,6 +175,7 @@ func (s *executor) run(mainCtx context.Context,
 				err := newSwap.execute(mainCtx, &executeConfig{
 					statusChan:          statusChan,
 					sweeper:             s.sweeper,
+					batcher:             s.batcher,
 					blockEpochChan:      queue.ChanOut(),
 					timerFactory:        s.executorConfig.createExpiryTimer,
 					loopOutMaxParts:     s.executorConfig.loopOutMaxParts,
@@ -211,6 +231,9 @@ func (s *executor) run(mainCtx context.Context,
 		case err := <-blockErrorChan:
 			return fmt.Errorf("block error: %v", err)
 
+		case err := <-batcherErrChan:
+			return fmt.Errorf("batcher error: %v", err)
+
 		case <-mainCtx.Done():
 			return mainCtx.Err()
 		}

loopd/daemon.go

@@ -18,6 +18,7 @@ import (
 	"github.com/lightninglabs/loop"
 	"github.com/lightninglabs/loop/loopd/perms"
 	"github.com/lightninglabs/loop/loopdb"
+	"github.com/lightninglabs/loop/sweepbatcher"
 
 	"github.com/lightninglabs/loop/instantout/reservation"
 	loop_looprpc "github.com/lightninglabs/loop/looprpc"
@@ -412,9 +413,11 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 		return err
 	}
 
+	sweeperDb := sweepbatcher.NewSQLStore(baseDb, chainParams)
+
 	// Create an instance of the loop client library.
 	swapClient, clientCleanup, err := getClient(
-		d.cfg, swapDb, &d.lnd.LndServices,
+		d.cfg, swapDb, sweeperDb, &d.lnd.LndServices,
 	)
 	if err != nil {
 		return err

loopd/utils.go

@@ -11,13 +11,15 @@ import (
 	"github.com/lightninglabs/loop/liquidity"
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/swap"
+	"github.com/lightninglabs/loop/sweepbatcher"
 	"github.com/lightningnetwork/lnd/clock"
 	"github.com/lightningnetwork/lnd/ticker"
 )
 
 // getClient returns an instance of the swap client.
 func getClient(cfg *Config, swapDb loopdb.SwapStore,
-	lnd *lndclient.LndServices) (*loop.Client, func(), error) {
+	sweeperDb sweepbatcher.BatcherStore, lnd *lndclient.LndServices) (
+	*loop.Client, func(), error) {
 
 	clientConfig := &loop.ClientConfig{
 		ServerAddress:       cfg.Server.Host,
@@ -33,7 +35,7 @@ func getClient(cfg *Config, swapDb loopdb.SwapStore,
 	}
 
 	swapClient, cleanUp, err := loop.NewClient(
-		cfg.DataDir, swapDb, clientConfig,
+		cfg.DataDir, swapDb, sweeperDb, clientConfig,
 	)
 	if err != nil {
 		return nil, nil, err

loopd/view.go

@@ -8,6 +8,7 @@ import (
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop"
 	"github.com/lightninglabs/loop/loopdb"
+	"github.com/lightninglabs/loop/sweepbatcher"
 	"github.com/lightninglabs/loop/utils"
 )
 
@@ -26,12 +27,16 @@ func view(config *Config, lisCfg *ListenerCfg) error {
 		return err
 	}
 
-	swapDb, _, err := openDatabase(config, chainParams)
+	swapDb, baseDb, err := openDatabase(config, chainParams)
 	if err != nil {
 		return err
 	}
 
-	swapClient, cleanup, err := getClient(config, swapDb, &lnd.LndServices)
+	sweeperDb := sweepbatcher.NewSQLStore(baseDb, chainParams)
+
+	swapClient, cleanup, err := getClient(
+		config, swapDb, sweeperDb, &lnd.LndServices,
+	)
 	if err != nil {
 		return err
 	}