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.
This commit is contained in:
Boris Nagaev 2025-10-14 20:37:20 -03:00
parent 03de0a8ed7
commit abf3bb4abf
No known key found for this signature in database
2 changed files with 55 additions and 12 deletions

View file

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

View file

@ -3,6 +3,7 @@ package sweepbatcher
import ( import (
"context" "context"
"fmt" "fmt"
"sync"
"testing" "testing"
"github.com/btcsuite/btcd/btcutil" "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 is optional index of a call at which it fails, 1 based.
failAt int failAt int
// mu protects the state.
mu sync.Mutex
} }
// SignTx memorizes the value of the output and fails if the number of // 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, minRelayFee, feeRate chainfee.SatPerKWeight,
loadOnly bool) (*wire.MsgTx, error) { loadOnly bool) (*wire.MsgTx, error) {
p.mu.Lock()
defer p.mu.Unlock()
if ctx.Err() != nil { if ctx.Err() != nil {
return nil, ctx.Err() 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{}, presigner: &mockPresigner{},
primarySweepID: op1, primarySweepID: op1,
sweeps: []sweep{ sweeps: []sweep{
@ -1085,10 +1093,29 @@ func TestPresign(t *testing.T) {
destAddr: destAddr, destAddr: destAddr,
nextBlockFeeRate: chainfee.FeePerKwFloor, nextBlockFeeRate: chainfee.FeePerKwFloor,
minRelayFeeRate: 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 { for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
err := presign( err := presign(
@ -1102,8 +1129,11 @@ func TestPresign(t *testing.T) {
} else { } else {
require.NoError(t, err) require.NoError(t, err)
p := tc.presigner.(*mockPresigner) p := tc.presigner.(*mockPresigner)
require.Equal(t, tc.wantOutputs, p.outputs) require.ElementsMatch(
require.Equal(t, tc.wantLockTimes, p.lockTimes) t,
zip(tc.wantOutputs, tc.wantLockTimes),
zip(p.outputs, p.lockTimes),
)
} }
}) })
} }