sweepbatcher: presign transactions in parallel

If a SignTx call is slow, the whole presign() function could timeout if all
the calls are done sequentially. In this commit each call is done in a separate
goroutine to reduce total latency.

Author: starius

sweepbatcher/presigned.go

@@ -12,6 +12,7 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/lnwallet/chainfee"
+	"golang.org/x/sync/errgroup"
 )
 
 // ensurePresigned checks that there is a presigned transaction spending the
@@ -371,6 +372,8 @@ func presign(ctx context.Context, presigner presigner, destAddr btcutil.Address,
 	// Set LockTime to 0. It is not critical.
 	const currentHeight = 0
 
+	eg, grCtx := errgroup.WithContext(ctx)
+
 	for fr := start; fr <= stop; fr = (fr * factorPPM) / 1_000_000 {
 		// Construct an unsigned transaction for this fee rate.
 		tx, _, feeForWeight, fee, err := constructUnsignedTx(
@@ -389,21 +392,31 @@ func presign(ctx context.Context, presigner presigner, destAddr btcutil.Address,
 
 		// Try to presign this transaction.
 		const loadOnly = false
-		_, err = presigner.SignTx(
-			ctx, primarySweepID, tx, batchAmt, minRelayFeeRate, fr,
-			loadOnly,
-		)
-		if err != nil {
-			return fmt.Errorf("failed to presign unsigned tx %v "+
-				"for feeRate %v: %w", tx.TxHash(), fr, err)
-		}
+		eg.Go(func() error {
+			_, err := presigner.SignTx(
+				grCtx, primarySweepID, tx, batchAmt,
+				minRelayFeeRate, fr, loadOnly,
+			)
+			if err != nil {
+				return fmt.Errorf("failed to presign unsigned "+
+					"tx %v for feeRate %v: %w", tx.TxHash(),
+					fr, err)
+			}
+
+			return nil
+		})
 
 		// If fee was clamped, stop here, because fee rate won't grow.
 		if fee < feeForWeight {
 			break
 		}
 	}
 
+	if err := eg.Wait(); err != nil {
+		return fmt.Errorf("presigning of batch of primarySweepID %v "+
+			"failed: %w", primarySweepID, err)
+	}
+
 	return nil
 }
 

sweepbatcher/presigned_test.go

@@ -3,6 +3,7 @@ package sweepbatcher
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 
 	"github.com/btcsuite/btcd/btcutil"
@@ -630,6 +631,9 @@ type mockPresigner struct {
 
 	// failAt is optional index of a call at which it fails, 1 based.
 	failAt int
+
+	// mu protects the state.
+	mu sync.Mutex
 }
 
 // SignTx memorizes the value of the output and fails if the number of
@@ -639,6 +643,9 @@ func (p *mockPresigner) SignTx(ctx context.Context,
 	minRelayFee, feeRate chainfee.SatPerKWeight,
 	loadOnly bool) (*wire.MsgTx, error) {
 
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	if ctx.Err() != nil {
 		return nil, ctx.Err()
 	}
@@ -1037,7 +1044,8 @@ func TestPresign(t *testing.T) {
 		},
 
 		{
-			name:           "small amount => fewer steps until clamped",
+			name: "small amount => fewer steps until " +
+				"clamped",
 			presigner:      &mockPresigner{},
 			primarySweepID: op1,
 			sweeps: []sweep{
@@ -1085,10 +1093,29 @@ func TestPresign(t *testing.T) {
 			destAddr:         destAddr,
 			nextBlockFeeRate: chainfee.FeePerKwFloor,
 			minRelayFeeRate:  chainfee.FeePerKwFloor,
-			wantErr:          "for feeRate 363 sat/kw",
+			wantErr:          "test error in SignTx",
 		},
 	}
 
+	type pair struct {
+		output   btcutil.Amount
+		lockTime uint32
+	}
+
+	zip := func(outputs []btcutil.Amount, lockTimes []uint32) []pair {
+		require.Equal(t, len(outputs), len(lockTimes))
+
+		pairs := make([]pair, len(outputs))
+		for i, output := range outputs {
+			pairs[i] = pair{
+				output:   output,
+				lockTime: lockTimes[i],
+			}
+		}
+
+		return pairs
+	}
+
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			err := presign(
@@ -1102,8 +1129,11 @@ func TestPresign(t *testing.T) {
 			} else {
 				require.NoError(t, err)
 				p := tc.presigner.(*mockPresigner)
-				require.Equal(t, tc.wantOutputs, p.outputs)
-				require.Equal(t, tc.wantLockTimes, p.lockTimes)
+				require.ElementsMatch(
+					t,
+					zip(tc.wantOutputs, tc.wantLockTimes),
+					zip(p.outputs, p.lockTimes),
+				)
 			}
 		})
 	}