GlcoinCore/loop
1
0
Fork 0
mirror of https://github.com/lightninglabs/loop.git synced 2026-08-13 12:33:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #891 from starius/presigned

Browse source 
sweepbatcher: add mode with presigned transactions
This commit is contained in:
Boris Nagaev 2025-05-08 17:43:58 -03:00 • committed by GitHub
commit 125d83f760
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 4910 additions and 133 deletions
Download patch file Download diff file Expand all files Collapse all files

2
loopout.go
View file

@ -1269,7 +1269,7 @@ func (s *loopOutSwap) waitForHtlcSpendConfirmedV2(globalCtx context.Context,
}
// Send the sweep to the sweeper.
err := s.batcher.AddSweep(&sweepReq)
err := s.batcher.AddSweep(ctx, &sweepReq)
if err != nil {
return nil, err
}

5
sweepbatcher/log.go
View file

@ -50,8 +50,3 @@ func infof(format string, params ...interface{}) {
func warnf(format string, params ...interface{}) {
log().Warnf(format, params...)
}
// errorf logs a message with level ERROR.
func errorf(format string, params ...interface{}) {
log().Errorf(format, params...)
}

630
sweepbatcher/presigned.go Normal file
View file

@ -0,0 +1,630 @@
package sweepbatcher
import (
"bytes"
"context"
"fmt"
"github.com/btcsuite/btcd/blockchain"
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcd/txscript"
"github.com/btcsuite/btcd/wire"
"github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwallet/chainfee"
)
// ensurePresigned checks that there is a presigned transaction spending the
// inputs of this group only. If allowNonEmptyBatch is false, the batch must be
// empty.
func (b *batch) ensurePresigned(ctx context.Context, newSweeps []*sweep,
allowNonEmptyBatch bool) error {
if b.cfg.presignedHelper == nil {
return fmt.Errorf("presignedHelper is not installed")
}
if len(b.sweeps) != 0 && !allowNonEmptyBatch {
return fmt.Errorf("ensurePresigned should be done when " +
"adding to an empty batch")
}
return ensurePresigned(
ctx, newSweeps, b.cfg.presignedHelper, b.cfg.chainParams,
)
}
// presignedTxChecker has methods to check if the inputs are presigned.
type presignedTxChecker interface {
destPkScripter
// SignTx signs an unsigned transaction or returns a pre-signed tx.
// It is only called with loadOnly=true by ensurePresigned.
SignTx(ctx context.Context, primarySweepID wire.OutPoint,
tx *wire.MsgTx, inputAmt btcutil.Amount,
minRelayFee, feeRate chainfee.SatPerKWeight,
loadOnly bool) (*wire.MsgTx, error)
}
// ensurePresigned checks that there is a presigned transaction spending the
// inputs of this group only.
func ensurePresigned(ctx context.Context, newSweeps []*sweep,
presignedTxChecker presignedTxChecker,
chainParams *chaincfg.Params) error {
sweeps := make([]sweep, len(newSweeps))
for i, s := range newSweeps {
sweeps[i] = sweep{
outpoint: s.outpoint,
value: s.value,
presigned: s.presigned,
}
}
// The sweeps are ordered inside the group, the first one is the primary
// outpoint in the batch.
primarySweepID := sweeps[0].outpoint
// Cache the destination address.
destAddr, err := getPresignedSweepsDestAddr(
ctx, presignedTxChecker, primarySweepID, chainParams,
)
if err != nil {
return fmt.Errorf("failed to find destination address: %w", err)
}
// Set LockTime to 0. It is not critical.
const currentHeight = 0
// Check if we can sign with minimum fee rate.
const feeRate = chainfee.FeePerKwFloor
tx, _, _, _, err := constructUnsignedTx(
sweeps, destAddr, currentHeight, feeRate,
)
if err != nil {
return fmt.Errorf("failed to construct unsigned tx "+
"for feeRate %v: %w", feeRate, err)
}
// Check of a presigned transaction exists.
var batchAmt btcutil.Amount
for _, sweep := range newSweeps {
batchAmt += sweep.value
}
const loadOnly = true
signedTx, err := presignedTxChecker.SignTx(
ctx, primarySweepID, tx, batchAmt, feeRate, feeRate, loadOnly,
)
if err != nil {
return fmt.Errorf("failed to find a presigned transaction "+
"for feeRate %v, txid of the template is %v, inputs: %d, "+
"outputs: %d: %w", feeRate, tx.TxHash(),
len(tx.TxIn), len(tx.TxOut), err)
}
// Check the SignTx worked correctly.
err = CheckSignedTx(tx, signedTx, batchAmt, feeRate)
if err != nil {
return fmt.Errorf("signed tx doesn't correspond the "+
"unsigned tx: %w", err)
}
return nil
}
// getOrderedSweeps returns the sweeps of the batch in the order they were
// added. The method must be called from the event loop of the batch.
func (b *batch) getOrderedSweeps(ctx context.Context) ([]sweep, error) {
// We use the DB just to know the order. Sweeps are copied from RAM.
utxo2sweep := make(map[wire.OutPoint]sweep, len(b.sweeps))
for _, s := range b.sweeps {
utxo2sweep[s.outpoint] = s
}
dbSweeps, err := b.store.FetchBatchSweeps(ctx, b.id)
if err != nil {
return nil, fmt.Errorf("FetchBatchSweeps(%d) failed: %w", b.id,
err)
}
if len(dbSweeps) != len(utxo2sweep) {
return nil, fmt.Errorf("FetchBatchSweeps(%d) returned %d "+
"sweeps, len(b.sweeps) is %d", b.id, len(dbSweeps),
len(utxo2sweep))
}
orderedSweeps := make([]sweep, len(dbSweeps))
for i, dbSweep := range dbSweeps {
// Sanity check: make sure dbSweep.ID grows.
if i > 0 && dbSweep.ID <= dbSweeps[i-1].ID {
return nil, fmt.Errorf("sweep ID does not grow: %d->%d",
dbSweeps[i-1].ID, dbSweep.ID)
}
s, has := utxo2sweep[dbSweep.Outpoint]
if !has {
return nil, fmt.Errorf("FetchBatchSweeps(%d) returned "+
"unknown sweep %v", b.id, dbSweep.Outpoint)
}
orderedSweeps[i] = s
}
return orderedSweeps, nil
}
// getSweepsGroups returns groups in which sweeps were added to the batch.
// All the sweeps are sorted by addition order and grouped by swap.
// The method must be called from the event loop of the batch.
func (b *batch) getSweepsGroups(ctx context.Context) ([][]sweep, error) {
orderedSweeps, err := b.getOrderedSweeps(ctx)
if err != nil {
return nil, fmt.Errorf("getOrderedSweeps(%d) failed: %w", b.id,
err)
}
groups := [][]sweep{}
for _, s := range orderedSweeps {
index := len(groups) - 1
// Start new group if there are no groups or new swap starts.
if len(groups) == 0 || s.swapHash != groups[index][0].swapHash {
groups = append(groups, []sweep{})
index++
}
groups[index] = append(groups[index], s)
}
// Sanity check: make sure the number of groups is the same as the
// number of distinct swaps.
swapsSet := make(map[lntypes.Hash]struct{}, len(groups))
for _, s := range orderedSweeps {
swapsSet[s.swapHash] = struct{}{}
}
if len(swapsSet) != len(groups) {
return nil, fmt.Errorf("batch %d: there are %d groups of "+
"sweeps and %d distinct swaps", b.id, len(groups),
len(swapsSet))
}
return groups, nil
}
// presign tries to presign batch sweep transactions composed of this batch and
// the sweep. In addition to that it presigns sweep transactions for any subset
// of sweeps that could remain if one of the sweep transactions gets confirmed.
// This can be done efficiently, since we keep track of the order in which
// sweeps are added and the associated swap hashes. So we presign transactions
// sweeping all the sweeps starting at some past sweeps group. For each inputs
// layout it presigns many transactions with different fee rates.
func (b *batch) presign(ctx context.Context, newSweeps []*sweep) error {
if b.cfg.presignedHelper == nil {
return fmt.Errorf("presignedHelper is not installed")
}
if len(b.sweeps) == 0 {
return fmt.Errorf("presigning should be done when adding to " +
"a non-empty batch")
}
// priorityConfTarget defines the confirmation target for quick
// inclusion in a block. A value of 2, rather than 1, is used to prevent
// fee estimator from failing.
// See https://github.com/lightninglabs/loop/issues/898
const priorityConfTarget = 2
// Find the feerate needed to get into next block.
nextBlockFeeRate, err := b.wallet.EstimateFeeRate(
ctx, priorityConfTarget,
)
if err != nil {
return fmt.Errorf("failed to get nextBlockFeeRate: %w", err)
}
b.Infof("nextBlockFeeRate is %v", nextBlockFeeRate)
// We need to restore previously added groups. We can do it by reading
// all the sweeps from DB (they must be ordered) and grouping by swap.
groups, err := b.getSweepsGroups(ctx)
if err != nil {
return fmt.Errorf("getSweepsGroups failed: %w", err)
}
if len(groups) == 0 {
return fmt.Errorf("getSweepsGroups returned no sweeps groups")
}
// Now presign a transaction spending a suffix of groups as well as new
// sweeps. Any non-empty suffix of groups may remain non-swept after
// some past tx is confirmed.
for len(groups) != 0 {
// Create the list of sweeps from the remaining groups and new
// sweeps.
sweeps := make([]sweep, 0, len(b.sweeps)+len(newSweeps))
for _, group := range groups {
sweeps = append(sweeps, group...)
}
for _, sweep := range newSweeps {
sweeps = append(sweeps, *sweep)
}
// The primarySweepID is the first sweep from the list of
// remaining sweeps if previous groups are confirmed.
primarySweepID := sweeps[0].outpoint
// Cache the destination address.
destAddr, err := getPresignedSweepsDestAddr(
ctx, b.cfg.presignedHelper, b.primarySweepID,
b.cfg.chainParams,
)
if err != nil {
return fmt.Errorf("failed to find destination "+
"address: %w", err)
}
err = presign(
ctx, b.cfg.presignedHelper, destAddr, primarySweepID,
sweeps, nextBlockFeeRate,
)
if err != nil {
return fmt.Errorf("failed to presign a transaction "+
"of %d sweeps: %w", len(sweeps), err)
}
// Cut a group to proceed to next suffix of original groups.
groups = groups[1:]
}
// Ensure that a batch spending new sweeps only has been presigned by
// PresignSweepsGroup.
const allowNonEmptyBatch = true
err = b.ensurePresigned(ctx, newSweeps, allowNonEmptyBatch)
if err != nil {
return fmt.Errorf("new sweeps were not presigned; this means "+
"that PresignSweepsGroup was not called prior to "+
"AddSweep for the group: %w", err)
}
return nil
}
// presigner tries to presign a batch transaction.
type presigner interface {
// Presign tries to presign a batch transaction. If the method returns
// nil, it is guaranteed that future calls to SignTx on this set of
// sweeps return valid signed transactions.
Presign(ctx context.Context, primarySweepID wire.OutPoint,
tx *wire.MsgTx, inputAmt btcutil.Amount) error
}
// presign tries to presign batch sweep transactions of the sweeps. It signs
// multiple versions of the transaction to make sure there is a transaction to
// be published if minRelayFee grows. If feerate is high, then a presigned tx
// gets LockTime equal to timeout minus 50 blocks, as a precautionary measure.
// A feerate is considered high if it is at least 100 sat/vbyte AND is at least
// 10x of the current next block feerate.
func presign(ctx context.Context, presigner presigner, destAddr btcutil.Address,
primarySweepID wire.OutPoint, sweeps []sweep,
nextBlockFeeRate chainfee.SatPerKWeight) error {
if presigner == nil {
return fmt.Errorf("presigner is not installed")
}
if len(sweeps) == 0 {
return fmt.Errorf("there are no sweeps")
}
if nextBlockFeeRate == 0 {
return fmt.Errorf("nextBlockFeeRate is not set")
}
// Keep track of the total amount this batch is sweeping back.
batchAmt := btcutil.Amount(0)
for _, sweep := range sweeps {
batchAmt += sweep.value
}
// Find the sweep with the earliest expiry.
timeout := sweeps[0].timeout
for _, sweep := range sweeps[1:] {
timeout = min(timeout, sweep.timeout)
}
if timeout <= 0 {
return fmt.Errorf("timeout is invalid: %d", timeout)
}
// Go from the floor (1.01 sat/vbyte) to 2k sat/vbyte with step of 1.2x.
const (
start = chainfee.FeePerKwFloor
stop = chainfee.AbsoluteFeePerKwFloor * 2_000
factorPPM = 1_200_000
timeoutThreshold = 50
)
// Calculate the locktime value to use for high feerate transactions.
// If timeout <= timeoutThreshold, don't set LockTime (keep value 0).
var highFeeRateLockTime uint32
if timeout > timeoutThreshold {
highFeeRateLockTime = uint32(timeout - timeoutThreshold)
}
// Calculate which feerate to consider high. At least 100 sat/vbyte and
// at least 10x of current nextBlockFeeRate.
highFeeRate := max(100*chainfee.FeePerKwFloor, 10*nextBlockFeeRate)
// Set LockTime to 0. It is not critical.
const currentHeight = 0
for fr := start; fr <= stop; fr = (fr * factorPPM) / 1_000_000 {
// Construct an unsigned transaction for this fee rate.
tx, _, feeForWeight, fee, err := constructUnsignedTx(
sweeps, destAddr, currentHeight, fr,
)
if err != nil {
return fmt.Errorf("failed to construct unsigned tx "+
"for feeRate %v: %w", fr, err)
}
// If the feerate is high enough, set locktime to prevent
// broadcasting such a transaction too early by mistake.
if fr >= highFeeRate {
tx.LockTime = highFeeRateLockTime
}
// Try to presign this transaction.
err = presigner.Presign(ctx, primarySweepID, tx, batchAmt)
if err != nil {
return fmt.Errorf("failed to presign unsigned tx %v "+
"for feeRate %v: %w", tx.TxHash(), fr, err)
}
// If fee was clamped, stop here, because fee rate won't grow.
if fee < feeForWeight {
break
}
}
return nil
}
// publishPresigned creates sweep transaction using a custom transaction signer
// and publishes it. It returns the fee of the transaction, and an error (if
// signing and/or publishing failed) and a boolean flag indicating signing
// success. This mode is incompatible with an external address.
func (b *batch) publishPresigned(ctx context.Context) (btcutil.Amount, error,
bool) {
// Sanity check, there should be at least 1 sweep in this batch.
if len(b.sweeps) == 0 {
return 0, fmt.Errorf("no sweeps in batch"), false
}
// Make sure that no external address is used.
for _, sweep := range b.sweeps {
if sweep.isExternalAddr {
return 0, fmt.Errorf("external address was used with " +
"a custom transaction signer"), false
}
}
// Cache current height and desired feerate of the batch.
currentHeight := b.currentHeight
feeRate := b.rbfCache.FeeRate
// Append this sweep to an array of sweeps. This is needed to keep the
// order of sweeps stored, as iterating the sweeps map does not
// guarantee same order.
sweeps := make([]sweep, 0, len(b.sweeps))
for _, sweep := range b.sweeps {
sweeps = append(sweeps, sweep)
}
// Cache the destination address.
address, err := getPresignedSweepsDestAddr(
ctx, b.cfg.presignedHelper, b.primarySweepID,
b.cfg.chainParams,
)
if err != nil {
return 0, fmt.Errorf("failed to find destination address: %w",
err), false
}
// Construct unsigned batch transaction.
tx, weight, _, fee, err := constructUnsignedTx(
sweeps, address, currentHeight, feeRate,
)
if err != nil {
return 0, fmt.Errorf("failed to construct tx: %w", err),
false
}
// Adjust feeRate, because it may have been clamped.
feeRate = chainfee.NewSatPerKWeight(fee, weight)
// Calculate total input amount.
batchAmt := btcutil.Amount(0)
for _, sweep := range sweeps {
batchAmt += sweep.value
}
// Determine the current minimum relay fee based on our chain backend.
minRelayFee, err := b.wallet.MinRelayFee(ctx)
if err != nil {
return 0, fmt.Errorf("failed to get minRelayFee: %w", err),
false
}
// Get a pre-signed transaction.
const loadOnly = false
signedTx, err := b.cfg.presignedHelper.SignTx(
ctx, b.primarySweepID, tx, batchAmt, minRelayFee, feeRate,
loadOnly,
)
if err != nil {
return 0, fmt.Errorf("failed to sign tx: %w", err),
false
}
// Run sanity checks to make sure presignedHelper.SignTx complied with
// all the invariants.
err = CheckSignedTx(tx, signedTx, batchAmt, minRelayFee)
if err != nil {
return 0, fmt.Errorf("signed tx doesn't correspond the "+
"unsigned tx: %w", err), false
}
tx = signedTx
txHash := tx.TxHash()
// Make sure tx weight matches the expected value.
realWeight := lntypes.WeightUnit(
blockchain.GetTransactionWeight(btcutil.NewTx(tx)),
)
if realWeight != weight {
b.Warnf("actual weight of tx %v is %v, estimated as %d",
txHash, realWeight, weight)
}
// Find actual fee rate of the signed transaction. It may differ from
// the desired fee rate, because SignTx may return a presigned tx.
output := btcutil.Amount(tx.TxOut[0].Value)
fee = batchAmt - output
signedFeeRate := chainfee.NewSatPerKWeight(fee, realWeight)
numSweeps := len(tx.TxIn)
b.Infof("attempting to publish custom signed tx=%v, desiredFeerate=%v,"+
" signedFeeRate=%v, weight=%v, fee=%v, sweeps=%d, destAddr=%s",
txHash, feeRate, signedFeeRate, realWeight, fee, numSweeps,
address)
b.debugLogTx("serialized batch", tx)
// Publish the transaction.
err = b.wallet.PublishTransaction(ctx, tx, b.cfg.txLabeler(b.id))
if err != nil {
return 0, fmt.Errorf("publishing tx failed: %w", err), true
}
// Store the batch transaction's txid and pkScript, for monitoring
// purposes.
b.batchTxid = &txHash
b.batchPkScript = tx.TxOut[0].PkScript
return fee, nil, true
}
// destPkScripter returns destination pkScript used by the sweep batch.
type destPkScripter interface {
// DestPkScript returns destination pkScript used by the sweep batch
// with the primary outpoint specified. Returns an error, if such tx
// doesn't exist. If there are many such transactions, returns any of
// pkScript's; all of them should have the same destination pkScript.
DestPkScript(ctx context.Context,
primarySweepID wire.OutPoint) ([]byte, error)
}
// getPresignedSweepsDestAddr returns the destination address used by the
// primary outpoint. The function must be used in presigned mode only.
func getPresignedSweepsDestAddr(ctx context.Context, helper destPkScripter,
primarySweepID wire.OutPoint,
chainParams *chaincfg.Params) (btcutil.Address, error) {
// Load pkScript from the presigned helper.
pkScriptBytes, err := helper.DestPkScript(ctx, primarySweepID)
if err != nil {
return nil, fmt.Errorf("presignedHelper.DestPkScript failed "+
"for primarySweepID %v: %w", primarySweepID, err)
}
// Convert pkScript to btcutil.Address.
pkScript, err := txscript.ParsePkScript(pkScriptBytes)
if err != nil {
return nil, fmt.Errorf("txscript.ParsePkScript failed for "+
"pkScript %x returned for primarySweepID %v: %w",
pkScriptBytes, primarySweepID, err)
}
address, err := pkScript.Address(chainParams)
if err != nil {
return nil, fmt.Errorf("pkScript.Address failed for "+
"pkScript %x returned for primarySweepID %v: %w",
pkScriptBytes, primarySweepID, err)
}
return address, nil
}
// CheckSignedTx makes sure that signedTx matches the unsignedTx. It checks
// according to criteria specified in the description of PresignedHelper.SignTx.
func CheckSignedTx(unsignedTx, signedTx *wire.MsgTx, inputAmt btcutil.Amount,
minRelayFee chainfee.SatPerKWeight) error {
// Make sure all inputs of signedTx have a non-empty witness.
for _, txIn := range signedTx.TxIn {
if len(txIn.Witness) == 0 {
return fmt.Errorf("input %s of signed tx is not signed",
txIn.PreviousOutPoint)
}
}
// Make sure the set of inputs is the same.
unsignedMap := make(map[wire.OutPoint]uint32, len(unsignedTx.TxIn))
for _, txIn := range unsignedTx.TxIn {
unsignedMap[txIn.PreviousOutPoint] = txIn.Sequence
}
for _, txIn := range signedTx.TxIn {
seq, has := unsignedMap[txIn.PreviousOutPoint]
if !has {
return fmt.Errorf("input %s is new in signed tx",
txIn.PreviousOutPoint)
}
if seq != txIn.Sequence {
return fmt.Errorf("sequence mismatch in input %s: "+
"%d in unsigned, %d in signed",
txIn.PreviousOutPoint, seq, txIn.Sequence)
}
delete(unsignedMap, txIn.PreviousOutPoint)
}
for outpoint := range unsignedMap {
return fmt.Errorf("input %s is missing in signed tx", outpoint)
}
// Compare outputs.
if len(unsignedTx.TxOut) != 1 {
return fmt.Errorf("unsigned tx has %d outputs, want 1",
len(unsignedTx.TxOut))
}
if len(signedTx.TxOut) != 1 {
return fmt.Errorf("the signed tx has %d outputs, want 1",
len(signedTx.TxOut))
}
unsignedOut := unsignedTx.TxOut[0]
signedOut := signedTx.TxOut[0]
if !bytes.Equal(unsignedOut.PkScript, signedOut.PkScript) {
return fmt.Errorf("mismatch of output pkScript: %v, %v",
unsignedOut.PkScript, signedOut.PkScript)
}
// Find the feerate of signedTx.
fee := inputAmt - btcutil.Amount(signedOut.Value)
weight := lntypes.WeightUnit(
blockchain.GetTransactionWeight(btcutil.NewTx(signedTx)),
)
feeRate := chainfee.NewSatPerKWeight(fee, weight)
if feeRate < minRelayFee {
return fmt.Errorf("feerate (%v) of signed tx is lower than "+
"minRelayFee (%v)", feeRate, minRelayFee)
}
// Check LockTime.
if signedTx.LockTime > unsignedTx.LockTime {
return fmt.Errorf("locktime (%d) of signed tx is higher than "+
"locktime of unsigned tx (%d)", signedTx.LockTime,
unsignedTx.LockTime)
}
// Check Version.
if signedTx.Version != unsignedTx.Version {
return fmt.Errorf("version (%d) of signed tx is not equal to "+
"version of unsigned tx (%d)", signedTx.Version,
unsignedTx.Version)
}
return nil
}

1579
sweepbatcher/presigned_test.go Normal file
View file

File diff suppressed because it is too large Load diff

14
sweepbatcher/store_mock.go
View file

@ -15,6 +15,7 @@ type StoreMock struct {
batches map[int32]dbBatch
sweeps map[wire.OutPoint]dbSweep
mu sync.Mutex
sweepID int32
}
// NewStoreMock instantiates a new mock store.
@ -122,7 +123,18 @@ func (s *StoreMock) UpsertSweep(ctx context.Context, sweep *dbSweep) error {
s.mu.Lock()
defer s.mu.Unlock()
s.sweeps[sweep.Outpoint] = *sweep
sweepCopy := *sweep
if old, exists := s.sweeps[sweep.Outpoint]; exists {
// Preserve existing sweep ID.
sweepCopy.ID = old.ID
} else {
// Assign fresh sweep ID.
sweepCopy.ID = s.sweepID
s.sweepID++
}
s.sweeps[sweep.Outpoint] = sweepCopy
return nil
}

328
sweepbatcher/sweep_batch.go
View file

@ -17,6 +17,7 @@ import (
"github.com/btcsuite/btcd/btcec/v2/schnorr/musig2"
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/btcutil/psbt"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/txscript"
"github.com/btcsuite/btcd/wire"
@ -121,6 +122,9 @@ type sweep struct {
// but it failed. We try to spend a sweep cooperatively only once. This
// status is not persisted in the DB.
coopFailed bool
// presigned is set, if the sweep should be handled in presigned mode.
presigned bool
}
// batchState is the state of the batch.
@ -176,6 +180,14 @@ type batchConfig struct {
// Note that musig2SignSweep must be nil in this case, however signer
// client must still be provided, as it is used for non-coop spendings.
customMuSig2Signer SignMuSig2
// presignedHelper provides methods used when presigned batches are
// enabled.
presignedHelper PresignedHelper
// chainParams are the chain parameters of the chain that is used by
// batches.
chainParams *chaincfg.Params
}
// rbfCache stores data related to our last fee bump.
@ -306,7 +318,7 @@ type batch struct {
// Purger is a function that takes a sweep request and feeds it back to the
// batcher main entry point. The name is inspired by its purpose, which is to
// purge the batch from sweeps that didn't make it to the confirmed tx.
type Purger func(sweepReq *SweepRequest) error
type Purger func(ctx context.Context, sweepReq *SweepRequest) error
// batchKit is a kit of dependencies that are used to initialize a batch. This
// struct is only used as a wrapper for the arguments that are required to
@ -319,7 +331,6 @@ type batchKit struct {
primaryID wire.OutPoint
sweeps map[wire.OutPoint]sweep
rbfCache rbfCache
returnChan chan SweepRequest
wallet lndclient.WalletKitClient
chainNotifier lndclient.ChainNotifierClient
signerClient lndclient.SignerClient
@ -467,7 +478,10 @@ func (b *batch) Errorf(format string, params ...interface{}) {
// checkSweepToAdd checks if a sweep can be added or updated in the batch. The
// caller must lock the event loop using scheduleNextCall. The function returns
// if the sweep already exists in the batch.
// if the sweep already exists in the batch. If presigned mode is enabled, the
// result depends on the outcome of the method presignedHelper.Presign for a
// non-empty batch. For an empty batch, the input needs to pass
// PresignSweepsGroup.
func (b *batch) checkSweepToAdd(_ context.Context, sweep *sweep) (bool, error) {
// If the provided sweep is nil, we can't proceed with any checks, so
// we just return early.
@ -587,6 +601,84 @@ func (b *batch) addSweeps(ctx context.Context, sweeps []*sweep) (bool, error) {
outpointsSet[s.outpoint] = struct{}{}
}
// Track if there is a presigned and a regular sweep.
var addingPresigned, addingRegular bool
for _, s := range sweeps {
if s.presigned {
addingPresigned = true
} else {
addingRegular = true
}
}
if addingPresigned && addingRegular {
b.Warnf("There are presigned and regular sweeps in the group")
return false, nil
}
// If presigned mode is enabled, we should first presign the new version
// of batch transaction. Also ensure that all the sweeps in the batch
// use the same mode (presigned or regular).
if addingPresigned {
// Ensure that all the sweeps in the batch use presigned mode.
for _, s := range b.sweeps {
if !s.presigned {
b.Warnf("Failed to add presigned sweep %x to "+
"the batch, because the batch has "+
"non-presigned sweep %x",
sweeps[0].swapHash[:6], s.swapHash[:6])
return false, nil
}
}
switch {
// We don't need to run checks if existing sweeps are updated.
case numExisting == len(sweeps):
// If new sweeps are added to the batch, we need to presign new
// version of batch transaction.
case len(b.sweeps) != 0:
if err := b.presign(ctx, sweeps); err != nil {
b.Warnf("Failed to add sweep %x to the batch, "+
"because failed to presign new version"+
" of batch tx: %v",
sweeps[0].swapHash[:6], err)
return false, nil
}
// If this is a new batch being formed, make sure we already
// have a presigned transaction.
default:
const allowNonEmptyBatch = false
err := b.ensurePresigned(
ctx, sweeps, allowNonEmptyBatch,
)
if err != nil {
b.Warnf("Failed to check signing of input %x,"+
" this means that PresignSweepsGroup "+
"was not called prior to AddSweep for"+
" this input: %v",
sweeps[0].swapHash[:6], err)
return false, nil
}
}
} else {
// Ensure that all the sweeps in the batch don't use presigned.
for _, s := range b.sweeps {
if s.presigned {
b.Warnf("failed to add a non-presigned sweep "+
"%x to the batch, because the batch "+
"has presigned sweep %x",
sweeps[0].swapHash[:6], s.swapHash[:6])
return false, nil
}
}
}
// Past this point we know that a new incoming sweep passes the
// acceptance criteria and is now ready to be added to this batch.
@ -881,8 +973,8 @@ func (b *batch) Run(ctx context.Context) error {
return fmt.Errorf("handleSpend error: %w", err)
}
case <-b.confChan:
if err := b.handleConf(runCtx); err != nil {
case conf := <-b.confChan:
if err := b.handleConf(runCtx, conf); err != nil {
return fmt.Errorf("handleConf error: %w", err)
}
@ -998,6 +1090,39 @@ func (b *batch) isUrgent(skipBefore time.Time) bool {
return true
}
// isPresigned returns if the batch uses presigned mode. Currently presigned and
// non-presigned sweeps never appear in the same batch. Fails if the batch is
// empty or contains both presigned and regular sweeps.
func (b *batch) isPresigned() (bool, error) {
var (
hasPresigned bool
hasRegular bool
)
for _, sweep := range b.sweeps {
if sweep.presigned {
hasPresigned = true
} else {
hasRegular = true
}
}
switch {
case hasPresigned && !hasRegular:
return true, nil
case !hasPresigned && hasRegular:
return false, nil
case hasPresigned && hasRegular:
return false, fmt.Errorf("the batch has both presigned and " +
"non-presigned sweeps")
default:
return false, fmt.Errorf("the batch is empty")
}
}
// publish creates and publishes the latest batch transaction to the network.
func (b *batch) publish(ctx context.Context) error {
var (
@ -1023,7 +1148,19 @@ func (b *batch) publish(ctx context.Context) error {
b.publishErrorHandler(err, errMsg, b.log())
}
fee, err, signSuccess = b.publishMixedBatch(ctx)
// Determine if we should use presigned mode for the batch.
presigned, err := b.isPresigned()
if err != nil {
return fmt.Errorf("failed to determine if the batch %d uses "+
"presigned mode: %w", b.id, err)
}
if presigned {
fee, err, signSuccess = b.publishPresigned(ctx)
} else {
fee, err, signSuccess = b.publishMixedBatch(ctx)
}
if err != nil {
if signSuccess {
logPublishError("publish error", err)
@ -1094,9 +1231,9 @@ func (b *batch) createPsbt(unsignedTx *wire.MsgTx, sweeps []sweep) ([]byte,
// constructUnsignedTx creates unsigned tx from the sweeps, paying to the addr.
// It also returns absolute fee (from weight and clamped).
func (b *batch) constructUnsignedTx(sweeps []sweep,
address btcutil.Address) (*wire.MsgTx, lntypes.WeightUnit,
btcutil.Amount, btcutil.Amount, error) {
func constructUnsignedTx(sweeps []sweep, address btcutil.Address,
currentHeight int32, feeRate chainfee.SatPerKWeight) (*wire.MsgTx,
lntypes.WeightUnit, btcutil.Amount, btcutil.Amount, error) {
// Sanity check, there should be at least 1 sweep in this batch.
if len(sweeps) == 0 {
@ -1106,7 +1243,7 @@ func (b *batch) constructUnsignedTx(sweeps []sweep,
// Create the batch transaction.
batchTx := &wire.MsgTx{
Version: 2,
LockTime: uint32(b.currentHeight),
LockTime: uint32(currentHeight),
}
// Add transaction inputs and estimate its weight.
@ -1158,7 +1295,14 @@ func (b *batch) constructUnsignedTx(sweeps []sweep,
// Find weight and fee.
weight := weightEstimate.Weight()
feeForWeight := b.rbfCache.FeeRate.FeeForWeight(weight)
feeForWeight := feeRate.FeeForWeight(weight)
// Fee can be rounded towards zero, leading to actual feeRate being
// slightly lower than the requested value. Increase the fee if this is
// the case.
if chainfee.NewSatPerKWeight(feeForWeight, weight) < feeRate {
feeForWeight++
}
// Clamp the calculated fee to the max allowed fee amount for the batch.
fee := clampBatchFee(feeForWeight, batchAmt)
@ -1243,8 +1387,8 @@ func (b *batch) publishMixedBatch(ctx context.Context) (btcutil.Amount, error,
// Construct unsigned batch transaction.
var err error
tx, weight, feeForWeight, fee, err = b.constructUnsignedTx(
sweeps, address,
tx, weight, feeForWeight, fee, err = constructUnsignedTx(
sweeps, address, b.currentHeight, b.rbfCache.FeeRate,
)
if err != nil {
return 0, fmt.Errorf("failed to construct tx: %w", err),
@ -1814,13 +1958,36 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
b.Warnf("transaction %v has no outputs", txHash)
}
// Determine if we should use presigned mode for the batch.
presigned, err := b.isPresigned()
if err != nil {
return fmt.Errorf("failed to determine if the batch %d uses "+
"presigned mode: %w", b.id, err)
}
// Sort sweeps by the addition order. This is important in presigned
// mode to pass them in correct order to purger (AddSweep) so the
// primary sweep is determined correctly and the presigned transaction
// is found. In regular mode the order doesn't matter, but we do it the
// same way for simplicity.
allSweeps, err := b.getOrderedSweeps(ctx)
if err != nil {
return fmt.Errorf("getOrderedSweeps(%d) failed: %w",
b.id, err)
}
// As a previous version of the batch transaction may get confirmed,
// which does not contain the latest sweeps, we need to detect the
// sweeps that did not make it to the confirmed transaction and feed
// them back to the batcher. This will ensure that the sweeps will enter
// a new batch instead of remaining dangling.
var totalSweptAmt btcutil.Amount
for _, sweep := range b.sweeps {
var (
totalSweptAmt btcutil.Amount
confirmedSweeps = []wire.OutPoint{}
purgedSweeps = []wire.OutPoint{}
purgedSwaps = []lntypes.Hash{}
)
for _, sweep := range allSweeps {
found := false
for _, txIn := range spendTx.TxIn {
@ -1828,27 +1995,58 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
found = true
totalSweptAmt += sweep.value
notifyList = append(notifyList, sweep)
confirmedSweeps = append(
confirmedSweeps, sweep.outpoint,
)
break
}
}
// If the sweep's outpoint was not found in the transaction's
// inputs this means it was left out. So we delete it from this
// batch and feed it back to the batcher.
if !found {
newSweep := sweep
delete(b.sweeps, sweep.outpoint)
if found {
continue
}
newSweep := sweep
delete(b.sweeps, sweep.outpoint)
newInput := Input{
Outpoint: newSweep.outpoint,
Value: newSweep.value,
}
// In presigned mode we should form a SweepRequest per swap
// (i.e. per group) and keep them ordered. It should reproduce
// the arguments and the order of the original external AddSweep
// calls.
L := len(purgeList)
if presigned && L != 0 &&
purgeList[L-1].SwapHash == newSweep.swapHash {
// Add the input to existing SweepRequest for this swap.
purgeList[L-1].Inputs = append(
purgeList[L-1].Inputs, newInput,
)
} else {
// Add the current sweep as a new element to purgeList.
// This is possible either in regular mode or in
// presigned mode in the beginning or on new swap.
purgeList = append(purgeList, SweepRequest{
SwapHash: newSweep.swapHash,
Inputs: []Input{
{
Outpoint: newSweep.outpoint,
Value: newSweep.value,
},
},
Inputs: []Input{newInput},
Notifier: newSweep.notifier,
})
}
}
for _, sweepReq := range purgeList {
purgedSwaps = append(purgedSwaps, sweepReq.SwapHash)
for _, input := range sweepReq.Inputs {
purgedSweeps = append(purgedSweeps, input.Outpoint)
}
}
// Calculate the fee portion that each sweep should pay for the batch.
feePortionPaidPerSweep, roundingDifference := getFeePortionForSweep(
@ -1881,7 +2079,13 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
// Dispatch the sweep notifier, we don't care about the outcome
// of this action so we don't wait for it.
go sweep.notifySweepSpend(ctx, &spendDetail)
go func() {
// Make sure this context doesn't expire so we
// successfully notify the caller.
ctx := context.WithoutCancel(ctx)
sweep.notifySweepSpend(ctx, &spendDetail)
}()
}
// Proceed with purging the sweeps. This will feed the sweeps that
@ -1889,21 +2093,26 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
// for re-entry. This batch doesn't care for the outcome of this
// operation so we don't wait for it.
go func() {
// Make sure this context doesn't expire so we successfully
// add the sweeps to the batcher.
ctx := context.WithoutCancel(ctx)
// Iterate over the purge list and feed the sweeps back to the
// batcher.
for _, sweep := range purgeList {
err := b.purger(&sweep)
for _, sweepReq := range purgeList {
err := b.purger(ctx, &sweepReq)
if err != nil {
b.Errorf("unable to purge sweep %x: %v",
sweep.SwapHash[:6], err)
b.Errorf("unable to purge sweep group %x: %v",
sweepReq.SwapHash[:6], err)
}
}
}()
b.Infof("spent, total sweeps: %v, purged sweeps: %v",
len(notifyList), len(purgeList))
b.Infof("spent, confirmed sweeps: %v, purged sweeps: %v, "+
"purged swaps: %v, purged groups: %v", confirmedSweeps,
purgedSweeps, purgedSwaps, len(purgeList))
err := b.monitorConfirmations(ctx)
err = b.monitorConfirmations(ctx)
if err != nil {
return err
}
@ -1916,8 +2125,44 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
}
// handleConf handles a confirmation notification. This is the final step of the
// batch. Here we signal to the batcher that this batch was completed.
func (b *batch) handleConf(ctx context.Context) error {
// batch. Here we signal to the batcher that this batch was completed. We also
// cleanup up presigned transactions whose primarySweepID is one of the sweeps
// that were spent and fully confirmed: such a transaction can't be broadcasted
// since it is either in a block or double-spends one of spent outputs.
func (b *batch) handleConf(ctx context.Context,
conf *chainntnfs.TxConfirmation) error {
spendTx := conf.Tx
txHash := spendTx.TxHash()
if b.batchTxid == nil || *b.batchTxid != txHash {
b.Warnf("Mismatch of batch txid: tx in spend notification had "+
"txid %v, but confirmation notification has txif %v. "+
"Using the later.", b.batchTxid, txHash)
}
b.batchTxid = &txHash
// If the batch is in presigned mode, cleanup presignedHelper.
presigned, err := b.isPresigned()
if err != nil {
return fmt.Errorf("failed to determine if the batch %d uses "+
"presigned mode: %w", b.id, err)
}
if presigned {
b.Infof("Cleaning up presigned store")
inputs := make([]wire.OutPoint, 0, len(spendTx.TxIn))
for _, txIn := range spendTx.TxIn {
inputs = append(inputs, txIn.PreviousOutPoint)
}
err := b.cfg.presignedHelper.CleanupTransactions(ctx, inputs)
if err != nil {
return fmt.Errorf("failed to clean up store for "+
"batch %d, inputs %v: %w", b.id, inputs, err)
}
}
b.Infof("confirmed in txid %s", b.batchTxid)
b.state = Confirmed
@ -1960,7 +2205,22 @@ func (b *batch) persist(ctx context.Context) error {
// getBatchDestAddr returns the batch's destination address. If the batch
// has already generated an address then the same one will be returned.
// The method must not be used in presigned mode. Use getPresignedSweepsDestAddr
// instead.
func (b *batch) getBatchDestAddr(ctx context.Context) (btcutil.Address, error) {
// Determine if we should use presigned mode for the batch.
presigned, err := b.isPresigned()
if err != nil {
return nil, fmt.Errorf("failed to determine if the batch %d "+
"uses presigned mode: %w", b.id, err)
}
// Make sure that the method is not used for presigned batches.
if presigned {
return nil, fmt.Errorf("getBatchDestAddr used in presigned " +
"mode")
}
var address btcutil.Address
// If a batch address is set, use that. Otherwise, generate a

357
sweepbatcher/sweep_batch_test.go Normal file
View file

@ -0,0 +1,357 @@
package sweepbatcher
import (
"fmt"
"testing"
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/txscript"
"github.com/btcsuite/btcd/wire"
"github.com/lightninglabs/loop/loopdb"
"github.com/lightninglabs/loop/utils"
"github.com/lightningnetwork/lnd/input"
"github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwallet/chainfee"
"github.com/stretchr/testify/require"
)
// TestConstructUnsignedTx verifies that the function constructUnsignedTx
// correctly creates unsigned transactions.
func TestConstructUnsignedTx(t *testing.T) {
// Prepare the necessary data for test cases.
op1 := wire.OutPoint{
Hash: chainhash.Hash{1, 1, 1},
Index: 1,
}
op2 := wire.OutPoint{
Hash: chainhash.Hash{2, 2, 2},
Index: 2,
}
batchPkScript, err := txscript.PayToAddrScript(destAddr)
require.NoError(t, err)
p2trAddr := "bcrt1pa38tp2hgjevqv3jcsxeu7v72n0s5a3ck8q2u8r" +
"k6mm67dv7uk26qq8je7e"
p2trAddress, err := btcutil.DecodeAddress(p2trAddr, nil)
require.NoError(t, err)
p2trPkScript, err := txscript.PayToAddrScript(p2trAddress)
require.NoError(t, err)
serializedPubKey := []byte{
0x02, 0x19, 0x2d, 0x74, 0xd0, 0xcb, 0x94, 0x34, 0x4c, 0x95,
0x69, 0xc2, 0xe7, 0x79, 0x01, 0x57, 0x3d, 0x8d, 0x79, 0x03,
0xc3, 0xeb, 0xec, 0x3a, 0x95, 0x77, 0x24, 0x89, 0x5d, 0xca,
0x52, 0xc6, 0xb4,
}
p2pkAddress, err := btcutil.NewAddressPubKey(
serializedPubKey, &chaincfg.RegressionNetParams,
)
require.NoError(t, err)
swapHash := lntypes.Hash{1, 1, 1}
swapContract := &loopdb.SwapContract{
CltvExpiry: 222,
AmountRequested: 2_000_000,
ProtocolVersion: loopdb.ProtocolVersionMuSig2,
HtlcKeys: htlcKeys,
}
htlc, err := utils.GetHtlc(
swapHash, swapContract, &chaincfg.RegressionNetParams,
)
require.NoError(t, err)
estimator := htlc.AddSuccessToEstimator
brokenEstimator := func(*input.TxWeightEstimator) error {
return fmt.Errorf("weight estimator test failure")
}
cases := []struct {
name string
sweeps []sweep
address btcutil.Address
currentHeight int32
feeRate chainfee.SatPerKWeight
wantErr string
wantTx *wire.MsgTx
wantWeight lntypes.WeightUnit
wantFeeForWeight btcutil.Amount
wantFee btcutil.Amount
}{
{
name: "no sweeps error",
wantErr: "no sweeps in batch",
},
{
name: "two coop sweeps",
sweeps: []sweep{
{
outpoint: op1,
value: 1_000_000,
},
{
outpoint: op2,
value: 2_000_000,
},
},
address: destAddr,
currentHeight: 800_000,
feeRate: 1000,
wantTx: &wire.MsgTx{
Version: 2,
LockTime: 800_000,
TxIn: []*wire.TxIn{
{
PreviousOutPoint: op1,
},
{
PreviousOutPoint: op2,
},
},
TxOut: []*wire.TxOut{
{
Value: 2999374,
PkScript: batchPkScript,
},
},
},
wantWeight: 626,
wantFeeForWeight: 626,
wantFee: 626,
},
{
name: "p2tr destination address",
sweeps: []sweep{
{
outpoint: op1,
value: 1_000_000,
},
{
outpoint: op2,
value: 2_000_000,
},
},
address: p2trAddress,
currentHeight: 800_000,
feeRate: 1000,
wantTx: &wire.MsgTx{
Version: 2,
LockTime: 800_000,
TxIn: []*wire.TxIn{
{
PreviousOutPoint: op1,
},
{
PreviousOutPoint: op2,
},
},
TxOut: []*wire.TxOut{
{
Value: 2999326,
PkScript: p2trPkScript,
},
},
},
wantWeight: 674,
wantFeeForWeight: 674,
wantFee: 674,
},
{
name: "unknown kind of address",
sweeps: []sweep{
{
outpoint: op1,
value: 1_000_000,
},
{
outpoint: op2,
value: 2_000_000,
},
},
address: nil,
wantErr: "unsupported address type",
},
{
name: "pay-to-pubkey address",
sweeps: []sweep{
{
outpoint: op1,
value: 1_000_000,
},
{
outpoint: op2,
value: 2_000_000,
},
},
address: p2pkAddress,
wantErr: "unknown address type",
},
{
name: "fee more than 20% clamped",
sweeps: []sweep{
{
outpoint: op1,
value: 1_000_000,
},
{
outpoint: op2,
value: 2_000_000,
},
},
address: destAddr,
currentHeight: 800_000,
feeRate: 1_000_000,
wantTx: &wire.MsgTx{
Version: 2,
LockTime: 800_000,
TxIn: []*wire.TxIn{
{
PreviousOutPoint: op1,
},
{
PreviousOutPoint: op2,
},
},
TxOut: []*wire.TxOut{
{
Value: 2400000,
PkScript: batchPkScript,
},
},
},
wantWeight: 626,
wantFeeForWeight: 626_000,
wantFee: 600_000,
},
{
name: "coop and noncoop",
sweeps: []sweep{
{
outpoint: op1,
value: 1_000_000,
},
{
outpoint: op2,
value: 2_000_000,
nonCoopHint: true,
htlc: *htlc,
htlcSuccessEstimator: estimator,
},
},
address: destAddr,
currentHeight: 800_000,
feeRate: 1000,
wantTx: &wire.MsgTx{
Version: 2,
LockTime: 800_000,
TxIn: []*wire.TxIn{
{
PreviousOutPoint: op1,
},
{
PreviousOutPoint: op2,
Sequence: 1,
},
},
TxOut: []*wire.TxOut{
{
Value: 2999211,
PkScript: batchPkScript,
},
},
},
wantWeight: 789,
wantFeeForWeight: 789,
wantFee: 789,
},
{
name: "weight estimator fails",
sweeps: []sweep{
{
outpoint: op1,
value: 1_000_000,
},
{
outpoint: op2,
value: 2_000_000,
nonCoopHint: true,
htlc: *htlc,
htlcSuccessEstimator: brokenEstimator,
},
},
address: destAddr,
currentHeight: 800_000,
feeRate: 1000,
wantErr: "sweep.htlcSuccessEstimator failed: " +
"weight estimator test failure",
},
{
name: "fix fee rounding",
sweeps: []sweep{
{
outpoint: op1,
value: 1_000_000,
},
{
outpoint: op2,
value: 2_000_000,
},
},
address: destAddr,
currentHeight: 800_000,
feeRate: 253,
wantTx: &wire.MsgTx{
Version: 2,
LockTime: 800_000,
TxIn: []*wire.TxIn{
{
PreviousOutPoint: op1,
},
{
PreviousOutPoint: op2,
},
},
TxOut: []*wire.TxOut{
{
Value: 2999841,
PkScript: batchPkScript,
},
},
},
wantWeight: 626,
wantFeeForWeight: 159,
wantFee: 159,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
tx, weight, feeForW, fee, err := constructUnsignedTx(
tc.sweeps, tc.address, tc.currentHeight,
tc.feeRate,
)
if tc.wantErr != "" {
require.Error(t, err)
require.ErrorContains(t, err, tc.wantErr)
} else {
require.NoError(t, err)
require.Equal(t, tc.wantTx, tx)
require.Equal(t, tc.wantWeight, weight)
require.Equal(t, tc.wantFeeForWeight, feeForW)
require.Equal(t, tc.wantFee, fee)
}
})
}
}

265
sweepbatcher/sweep_batcher.go
View file

@ -132,6 +132,11 @@ type SweepInfo struct {
// has to be spent using preimage. This is only used in fee estimations
// when selecting a batch for the sweep to minimize fees.
NonCoopHint bool
// IsPresigned stores if presigned mode is enabled for the sweep. This
// value should be stable for a sweep. Currently presigned and
// non-presigned sweeps never appear in the same batch.
IsPresigned bool
}
// SweepFetcher is used to get details of a sweep.
@ -156,6 +161,51 @@ type SignMuSig2 func(ctx context.Context, muSig2Version input.MuSig2Version,
swapHash lntypes.Hash, rootHash chainhash.Hash, sigHash [32]byte,
) ([]byte, error)
// PresignedHelper provides methods used when batches are presigned in advance.
// In this mode sweepbatcher uses transactions provided by PresignedHelper,
// which are pre-signed. The helper also memorizes transactions it previously
// produced. It also affects batch selection: presigned inputs and regular
// (non-presigned) inputs never appear in the same batch. Also if presigning
// fails (e.g. because one of the inputs is offline), an input can't be added to
// a batch.
type PresignedHelper interface {
// Presign tries to presign a batch transaction. If the method returns
// nil, it is guaranteed that future calls to SignTx on this set of
// sweeps return valid signed transactions. The implementation should
// first check if this transaction already exists in the store to skip
// cosigning if possible.
Presign(ctx context.Context, primarySweepID wire.OutPoint,
tx *wire.MsgTx, inputAmt btcutil.Amount) error
// DestPkScript returns destination pkScript used by the sweep batch
// with the primary outpoint specified. Returns an error, if such tx
// doesn't exist. If there are many such transactions, returns any of
// pkScript's; all of them should have the same destination pkScript.
DestPkScript(ctx context.Context,
primarySweepID wire.OutPoint) ([]byte, error)
// SignTx signs an unsigned transaction or returns a pre-signed tx.
// It must satisfy the following invariants:
// - the set of inputs is the same, though the order may change;
// - the output is the same, but its amount may be different;
// - feerate is higher or equal to minRelayFee;
// - LockTime may be decreased;
// - transaction version must be the same;
// - witness must not be empty;
// - Sequence numbers in the inputs must be preserved.
// When choosing a presigned transaction, a transaction with fee rate
// closer to the fee rate passed is selected. If loadOnly is set, it
// doesn't try to sign the transaction and only loads a presigned tx.
SignTx(ctx context.Context, primarySweepID wire.OutPoint,
tx *wire.MsgTx, inputAmt btcutil.Amount,
minRelayFee, feeRate chainfee.SatPerKWeight,
loadOnly bool) (*wire.MsgTx, error)
// CleanupTransactions removes all transactions related to any of the
// outpoints. Should be called after sweep batch tx is fully confirmed.
CleanupTransactions(ctx context.Context, inputs []wire.OutPoint) error
}
// VerifySchnorrSig is a function that can be used to verify a schnorr
// signature.
type VerifySchnorrSig func(pubKey *btcec.PublicKey, hash, sig []byte) error
@ -222,6 +272,26 @@ type SweepRequest struct {
Notifier *SpendNotifier
}
// addSweepsRequest is a request to sweep an outpoint or a group of outpoints
// that is used internally by the batcher (between AddSweep and handleSweeps).
type addSweepsRequest struct {
// sweeps is the list of sweeps already loaded from DB and fee rate
// source.
sweeps []*sweep
// Notifier is a notifier that is used to notify the requester of this
// sweep that the sweep was successful.
notifier *SpendNotifier
// completed is set if the sweep is spent and the spending transaction
// is confirmed.
completed bool
// parentBatch is the parent batch of this sweep. It is loaded ony if
// completed is true.
parentBatch *dbBatch
}
type SpendDetail struct {
// Tx is the transaction that spent the outpoint.
Tx *wire.MsgTx
@ -266,8 +336,8 @@ type Batcher struct {
// batches is a map of batch IDs to the currently active batches.
batches map[int32]*batch
// sweepReqs is a channel where sweep requests are received.
sweepReqs chan SweepRequest
// addSweepsChan is a channel where sweep requests are received.
addSweepsChan chan *addSweepsRequest
// testReqs is a channel where test requests are received.
// This is used only in unit tests! The reason to have this is to
@ -354,6 +424,10 @@ type Batcher struct {
// error. By default, it logs all errors as warnings, but "insufficient
// fee" as Info.
publishErrorHandler PublishErrorHandler
// presignedHelper provides methods used when presigned batches are
// enabled.
presignedHelper PresignedHelper
}
// BatcherConfig holds batcher configuration.
@ -394,6 +468,10 @@ type BatcherConfig struct {
// error. By default, it logs all errors as warnings, but "insufficient
// fee" as Info.
publishErrorHandler PublishErrorHandler
// presignedHelper provides methods used when presigned batches are
// enabled.
presignedHelper PresignedHelper
}
// BatcherOption configures batcher behaviour.
@ -467,6 +545,15 @@ func WithPublishErrorHandler(handler PublishErrorHandler) BatcherOption {
}
}
// WithPresignedHelper enables presigned batches in the batcher. When a sweep
// intended for presigning is added, it must be first passed to the
// PresignSweepsGroup method, before first call of the AddSweep method.
func WithPresignedHelper(presignedHelper PresignedHelper) BatcherOption {
return func(cfg *BatcherConfig) {
cfg.presignedHelper = presignedHelper
}
}
// NewBatcher creates a new Batcher instance.
func NewBatcher(wallet lndclient.WalletKitClient,
chainNotifier lndclient.ChainNotifierClient,
@ -501,7 +588,7 @@ func NewBatcher(wallet lndclient.WalletKitClient,
return &Batcher{
batches: make(map[int32]*batch),
sweepReqs: make(chan SweepRequest),
addSweepsChan: make(chan *addSweepsRequest),
testReqs: make(chan *testRequest),
errChan: make(chan error, 1),
quit: make(chan struct{}),
@ -521,6 +608,7 @@ func NewBatcher(wallet lndclient.WalletKitClient,
txLabeler: cfg.txLabeler,
customMuSig2Signer: cfg.customMuSig2Signer,
publishErrorHandler: cfg.publishErrorHandler,
presignedHelper: cfg.presignedHelper,
}
}
@ -557,15 +645,11 @@ func (b *Batcher) Run(ctx context.Context) error {
for {
select {
case sweepReq := <-b.sweepReqs:
sweeps, err := b.fetchSweeps(runCtx, sweepReq)
if err != nil {
warnf("fetchSweeps failed: %v.", err)
return err
}
err = b.handleSweeps(runCtx, sweeps, sweepReq.Notifier)
case req := <-b.addSweepsChan:
err = b.handleSweeps(
runCtx, req.sweeps, req.notifier, req.completed,
req.parentBatch,
)
if err != nil {
warnf("handleSweeps failed: %v.", err)
@ -589,11 +673,125 @@ func (b *Batcher) Run(ctx context.Context) error {
}
}
// AddSweep adds a sweep request to the batcher for handling. This will either
// place the sweep in an existing batch or create a new one.
func (b *Batcher) AddSweep(sweepReq *SweepRequest) error {
// PresignSweepsGroup creates and stores presigned transactions for the sweeps
// group. This method must be called prior to AddSweep if presigned mode is
// enabled, otherwise AddSweep will fail. All the sweeps must belong to the same
// swap. The order of sweeps is important. The first sweep serves as
// primarySweepID if the group starts a new batch.
func (b *Batcher) PresignSweepsGroup(ctx context.Context, inputs []Input,
sweepTimeout int32, destAddress btcutil.Address) error {
if len(inputs) == 0 {
return fmt.Errorf("no inputs passed to PresignSweepsGroup")
}
if b.presignedHelper == nil {
return fmt.Errorf("presignedHelper is not installed")
}
// Find the feerate needed to get into next block. Use conf_target=2,
nextBlockFeeRate, err := b.wallet.EstimateFeeRate(ctx, 2)
if err != nil {
return fmt.Errorf("failed to get nextBlockFeeRate: %w", err)
}
infof("PresignSweepsGroup: nextBlockFeeRate is %v", nextBlockFeeRate)
sweeps := make([]sweep, len(inputs))
for i, input := range inputs {
sweeps[i] = sweep{
outpoint: input.Outpoint,
value: input.Value,
timeout: sweepTimeout,
}
}
// The sweeps are ordered inside the group, the first one is the primary
// outpoint in the batch.
primarySweepID := sweeps[0].outpoint
return presign(
ctx, b.presignedHelper, destAddress, primarySweepID, sweeps,
nextBlockFeeRate,
)
}
// AddSweep loads information about sweeps from the store and fee rate source,
// and adds them to the batcher for handling. This will either place the sweep
// in an existing batch or create a new one. The method can be called multiple
// times, but the sweeps (including the order of them) must be the same. If
// notifier is provided, the batcher sends back sweeping results through it.
func (b *Batcher) AddSweep(ctx context.Context, sweepReq *SweepRequest) error {
// If the batcher is shutting down, quit now.
select {
case b.sweepReqs <- *sweepReq:
case <-b.quit:
return ErrBatcherShuttingDown
default:
}
sweeps, err := b.fetchSweeps(ctx, *sweepReq)
if err != nil {
return fmt.Errorf("fetchSweeps failed: %w", err)
}
if len(sweeps) == 0 {
return fmt.Errorf("trying to add an empty group of sweeps")
}
// Since the whole group is added to the same batch and belongs to
// the same transaction, we use sweeps[0] below where we need any sweep.
sweep := sweeps[0]
completed, err := b.store.GetSweepStatus(ctx, sweep.outpoint)
if err != nil {
return fmt.Errorf("failed to get the status of sweep %v: %w",
sweep.outpoint, err)
}
var (
parentBatch *dbBatch
fullyConfirmed bool
)
if completed {
// Verify that the parent batch is confirmed. Note that a batch
// is only considered confirmed after it has received three
// on-chain confirmations to prevent issues caused by reorgs.
parentBatch, err = b.store.GetParentBatch(ctx, sweep.outpoint)
if err != nil {
return fmt.Errorf("unable to get parent batch for "+
"sweep %x: %w", sweep.swapHash[:6], err)
}
if parentBatch.State == batchConfirmed {
fullyConfirmed = true
}
}
// If this is a presigned mode, make sure PresignSweepsGroup was called.
// We skip the check for fully confirmed sweeps, because their presigned
// transactions were already cleaned up from the store.
if sweep.presigned && !fullyConfirmed {
err := ensurePresigned(
ctx, sweeps, b.presignedHelper, b.chainParams,
)
if err != nil {
return fmt.Errorf("inputs with primarySweep %v were "+
"not presigned (call PresignSweepsGroup "+
"first): %w", sweep.outpoint, err)
}
}
infof("Batcher adding sweep group of %d sweeps with primarySweep %x, "+
"presigned=%v, completed=%v", len(sweeps), sweep.swapHash[:6],
sweep.presigned, completed)
req := &addSweepsRequest{
sweeps: sweeps,
notifier: sweepReq.Notifier,
completed: completed,
parentBatch: parentBatch,
}
select {
case b.addSweepsChan <- req:
return nil
case <-b.quit:
@ -634,39 +832,16 @@ func (b *Batcher) testRunInEventLoop(ctx context.Context, handler func()) {
// handleSweeps handles a sweep request by either placing the group of sweeps in
// an existing batch, or by spinning up a new batch for it.
func (b *Batcher) handleSweeps(ctx context.Context, sweeps []*sweep,
notifier *SpendNotifier) error {
if len(sweeps) == 0 {
return fmt.Errorf("trying to add an empty group of sweeps")
}
notifier *SpendNotifier, completed bool, parentBatch *dbBatch) error {
// Since the whole group is added to the same batch and belongs to
// the same transaction, we use sweeps[0] below where we need any sweep.
sweep := sweeps[0]
completed, err := b.store.GetSweepStatus(ctx, sweep.outpoint)
if err != nil {
return err
}
infof("Batcher handling sweep %x, completed=%v",
sweep.swapHash[:6], completed)
// If the sweep has already been completed in a confirmed batch then we
// can't attach its notifier to the batch as that is no longer running.
// Instead we directly detect and return the spend here.
if completed && *notifier != (SpendNotifier{}) {
// Verify that the parent batch is confirmed. Note that a batch
// is only considered confirmed after it has received three
// on-chain confirmations to prevent issues caused by reorgs.
parentBatch, err := b.store.GetParentBatch(ctx, sweep.outpoint)
if err != nil {
errorf("unable to get parent batch for sweep %x:"+
" %v", sweep.swapHash[:6], err)
return err
}
// The parent batch is indeed confirmed, meaning it is complete
// and we won't be able to attach this sweep to it.
if parentBatch.State == batchConfirmed {
@ -707,7 +882,7 @@ func (b *Batcher) handleSweeps(ctx context.Context, sweeps []*sweep,
}
// Try to run the greedy algorithm of batch selection to minimize costs.
err = b.greedyAddSweeps(ctx, sweeps)
err := b.greedyAddSweeps(ctx, sweeps)
if err == nil {
// The greedy algorithm succeeded.
return nil
@ -735,7 +910,9 @@ func (b *Batcher) handleSweeps(ctx context.Context, sweeps []*sweep,
return b.spinUpNewBatch(ctx, sweeps)
}
// spinUpNewBatch creates new batch, starts it and adds the sweeps to it.
// spinUpNewBatch creates new batch, starts it and adds the sweeps to it. If
// presigned mode is enabled, the result also depends on outcome of
// presignedHelper.Presign.
func (b *Batcher) spinUpNewBatch(ctx context.Context, sweeps []*sweep) error {
// Spin up a fresh batch.
newBatch, err := b.spinUpBatch(ctx)
@ -1190,6 +1367,7 @@ func (b *Batcher) loadSweep(ctx context.Context, swapHash lntypes.Hash,
destAddr: s.DestAddr,
minFeeRate: minFeeRate,
nonCoopHint: s.NonCoopHint,
presigned: s.IsPresigned,
}, nil
}
@ -1200,14 +1378,15 @@ func (b *Batcher) newBatchConfig(maxTimeoutDistance int32) batchConfig {
noBumping: b.customFeeRate != nil,
txLabeler: b.txLabeler,
customMuSig2Signer: b.customMuSig2Signer,
presignedHelper: b.presignedHelper,
clock: b.clock,
chainParams: b.chainParams,
}
}
// newBatchKit creates new batch kit.
func (b *Batcher) newBatchKit() batchKit {
return batchKit{
returnChan: b.sweepReqs,
wallet: b.wallet,
chainNotifier: b.chainNotifier,
signerClient: b.signerClient,

1685
sweepbatcher/sweep_batcher_presigned_test.go Normal file
View file

File diff suppressed because it is too large Load diff

105
sweepbatcher/sweep_batcher_test.go
View file

@ -252,14 +252,14 @@ func testSweepBatcherBatchCreation(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
<-lnd.RegisterSpendChannel
// Insert the same swap twice, this should be a noop.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Once batcher receives sweep request it will eventually spin up a
// batch.
@ -301,7 +301,7 @@ func testSweepBatcherBatchCreation(t *testing.T, store testStore,
require.NoError(t, err)
store.AssertLoopOutStored()
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
// Tick tock next block.
err = lnd.NotifyHeight(601)
@ -347,7 +347,7 @@ func testSweepBatcherBatchCreation(t *testing.T, store testStore,
require.NoError(t, err)
store.AssertLoopOutStored()
require.NoError(t, batcher.AddSweep(&sweepReq3))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq3))
// Since the second batch got created we check that it registered its
// primary sweep's spend.
@ -457,7 +457,7 @@ func testFeeBumping(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -562,7 +562,7 @@ func testTxLabeler(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// When batch is successfully created it will execute it's first step,
// which leads to a spend monitor of the primary sweep.
@ -718,7 +718,7 @@ func testPublishErrorHandler(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// When batch is successfully created it will execute it's first step,
// which leads to a spend monitor of the primary sweep.
@ -802,7 +802,7 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// When batch is successfully created it will execute it's first step,
// which leads to a spend monitor of the primary sweep.
@ -918,6 +918,7 @@ type wrappedLogger struct {
debugMessages []string
infoMessages []string
warnMessages []string
}
// Debugf logs debug message.
@ -938,6 +939,15 @@ func (l *wrappedLogger) Infof(format string, params ...interface{}) {
l.Logger.Infof(format, params...)
}
// Warnf logs a warning message.
func (l *wrappedLogger) Warnf(format string, params ...interface{}) {
l.mu.Lock()
defer l.mu.Unlock()
l.warnMessages = append(l.warnMessages, format)
l.Logger.Warnf(format, params...)
}
// testDelays tests that WithInitialDelay and WithPublishDelay work.
func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) {
// Set initial delay and publish delay.
@ -1016,7 +1026,7 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) {
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq))
// Expect two timers to be set: initialDelay and publishDelay,
// and RegisterSpend to be called. The order is not determined,
@ -1318,7 +1328,7 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) {
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
// Expect the sweep to be added to new batch. Expect two timers:
// largeInitialDelay and publishDelay. RegisterSpend is called in
@ -1399,7 +1409,7 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) {
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq3))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq3))
// Wait for sweep to be added to the batch.
require.EventuallyWithT(t, func(c *assert.CollectT) {
@ -1532,7 +1542,7 @@ func testCustomDelays(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Expect two timers to be set: initialDelay and publishDelay,
// and RegisterSpend to be called. The order is not determined,
@ -1603,7 +1613,7 @@ func testCustomDelays(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
// Expect timer for initialDelay2 to be registered, because
// initialDelay2 is lower than initialDelay1, meaning that swap2
@ -1752,7 +1762,7 @@ func testMaxSweepsPerBatch(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq))
// If this is new batch, expect a spend registration.
if i%MaxSweepsPerBatch == 0 {
@ -1931,7 +1941,7 @@ func testSweepBatcherSweepReentry(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Feed the sweeps to the batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// After inserting the primary (first) sweep, a spend monitor should be
// registered.
@ -1941,7 +1951,7 @@ func testSweepBatcherSweepReentry(t *testing.T, store testStore,
<-lnd.TxPublishChannel
// Add the second sweep.
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
// Add next block to trigger batch publishing.
err = lnd.NotifyHeight(601)
@ -1951,7 +1961,7 @@ func testSweepBatcherSweepReentry(t *testing.T, store testStore,
<-lnd.TxPublishChannel
// Add the third sweep.
require.NoError(t, batcher.AddSweep(&sweepReq3))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq3))
// Add next block to trigger batch publishing.
err = lnd.NotifyHeight(602)
@ -2058,7 +2068,7 @@ func testSweepBatcherSweepReentry(t *testing.T, store testStore,
// Re-add one of remaining sweeps to trigger removing the completed
// batch from the batcher.
require.NoError(t, batcher.AddSweep(&sweepReq3))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq3))
// Eventually the batch receives the confirmation notification,
// gracefully exits and the batcher deletes it.
@ -2144,7 +2154,7 @@ func testSweepBatcherGroup(t *testing.T, store testStore,
},
Notifier: &dummyNotifier,
}
require.NoError(t, batcher.AddSweep(&sweepReq))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq))
// After inserting the primary (first) sweep, a spend monitor should be
// registered.
@ -2225,7 +2235,7 @@ func testSweepBatcherNonWalletAddr(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -2241,7 +2251,7 @@ func testSweepBatcherNonWalletAddr(t *testing.T, store testStore,
<-lnd.TxPublishChannel
// Insert the same swap twice, this should be a noop.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Create a second sweep request that has a timeout distance less than
// our configured threshold.
@ -2274,7 +2284,7 @@ func testSweepBatcherNonWalletAddr(t *testing.T, store testStore,
require.NoError(t, err)
store.AssertLoopOutStored()
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -2320,7 +2330,7 @@ func testSweepBatcherNonWalletAddr(t *testing.T, store testStore,
require.NoError(t, err)
store.AssertLoopOutStored()
require.NoError(t, batcher.AddSweep(&sweepReq3))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq3))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -2594,7 +2604,7 @@ func testSweepBatcherComposite(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -2610,9 +2620,9 @@ func testSweepBatcherComposite(t *testing.T, store testStore,
<-lnd.TxPublishChannel
// Insert the same swap twice, this should be a noop.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
// Batcher should not create a second batch as timeout distance is small
// enough.
@ -2628,7 +2638,7 @@ func testSweepBatcherComposite(t *testing.T, store testStore,
tx := <-lnd.TxPublishChannel
require.Len(t, tx.TxIn, 2)
require.NoError(t, batcher.AddSweep(&sweepReq3))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq3))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -2644,7 +2654,7 @@ func testSweepBatcherComposite(t *testing.T, store testStore,
tx = <-lnd.TxPublishChannel
require.Len(t, tx.TxIn, 1)
require.NoError(t, batcher.AddSweep(&sweepReq4))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq4))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -2660,7 +2670,7 @@ func testSweepBatcherComposite(t *testing.T, store testStore,
tx = <-lnd.TxPublishChannel
require.Len(t, tx.TxIn, 1)
require.NoError(t, batcher.AddSweep(&sweepReq5))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq5))
// Publish a block to trigger batch 3 republishing.
err = lnd.NotifyHeight(601)
@ -2677,7 +2687,7 @@ func testSweepBatcherComposite(t *testing.T, store testStore,
return batcher.numBatches(ctx) == 3
}, test.Timeout, eventuallyCheckFrequency)
require.NoError(t, batcher.AddSweep(&sweepReq6))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq6))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -2845,7 +2855,7 @@ func testRestoringEmptyBatch(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -3063,7 +3073,7 @@ func testHandleSweepTwice(t *testing.T, backend testStore,
store.putLoopOutSwap(sweepReq2.SwapHash, loopOut2)
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Since two batches were created we check that it registered for its
// primary sweep's spend.
@ -3074,7 +3084,7 @@ func testHandleSweepTwice(t *testing.T, backend testStore,
// Deliver the second sweep. It will go to a separate batch,
// since CltvExpiry values are distant enough.
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
<-lnd.RegisterSpendChannel
// Wait for tx to be published.
@ -3117,7 +3127,7 @@ func testHandleSweepTwice(t *testing.T, backend testStore,
// Re-add the second sweep. It is expected to stay in second batch,
// not added to both batches.
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
require.Eventually(t, func() bool {
// Make sure there are two batches.
@ -3232,7 +3242,7 @@ func testRestoringPreservesConfTarget(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -3450,7 +3460,7 @@ func testSweepFetcher(t *testing.T, store testStore,
<-batcher.initDone
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -3571,8 +3581,11 @@ func testSweepBatcherCloseDuringAdding(t *testing.T, store testStore,
}
// Deliver sweep request to batcher.
err := batcher.AddSweep(&sweepReq)
if err == ErrBatcherShuttingDown {
err := batcher.AddSweep(ctx, &sweepReq)
if errors.Is(err, ErrBatcherShuttingDown) {
break
}
if errors.Is(err, context.Canceled) {
break
}
require.NoError(t, err)
@ -3667,7 +3680,7 @@ func testCustomSignMuSig2(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -3830,7 +3843,7 @@ func testWithMixedBatch(t *testing.T, store testStore,
}},
Notifier: &dummyNotifier,
}
require.NoError(t, batcher.AddSweep(&sweepReq))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq))
if i == 0 {
// Since a batch was created we check that it registered
@ -4000,7 +4013,7 @@ func testWithMixedBatchCustom(t *testing.T, store testStore,
}},
Notifier: &dummyNotifier,
}
require.NoError(t, batcher.AddSweep(&sweepReq))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq))
if i == 0 {
// Since a batch was created we check that it registered
@ -4329,7 +4342,7 @@ func testFeeRateGrows(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Since a batch was created we check that it registered for its primary
// sweep's spend.
@ -4346,7 +4359,7 @@ func testFeeRateGrows(t *testing.T, store testStore,
// Now decrease the fee of sweep1.
setFeeRate(swapHash1, feeRateLow)
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Tick tock next block.
err = lnd.NotifyHeight(601)
@ -4395,7 +4408,7 @@ func testFeeRateGrows(t *testing.T, store testStore,
store.AssertLoopOutStored()
// Deliver sweep request to batcher.
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
// Tick tock next block.
err = lnd.NotifyHeight(602)
@ -4412,8 +4425,8 @@ func testFeeRateGrows(t *testing.T, store testStore,
// Now update fee rate of second sweep (which is not primary) to
// feeRateHigh. Fee rate of sweep 1 is still feeRateLow.
setFeeRate(swapHash2, feeRateHigh)
require.NoError(t, batcher.AddSweep(&sweepReq1))
require.NoError(t, batcher.AddSweep(&sweepReq2))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
// Tick tock next block.
err = lnd.NotifyHeight(603)

12
test/lnd_services_mock.go
View file

@ -29,6 +29,7 @@ func NewMockLnd() *LndMockServices {
lightningClient := &mockLightningClient{}
walletKit := &mockWalletKit{
feeEstimates: make(map[int32]chainfee.SatPerKWeight),
minRelayFee: chainfee.FeePerKwFloor,
}
chainNotifier := &mockChainNotifier{}
signer := &mockSigner{}
@ -128,6 +129,11 @@ type SignOutputRawRequest struct {
SignDescriptors []*lndclient.SignDescriptor
}
// PublishHandler is optional transaction handler function called upon calling
// the method PublishTransaction.
type PublishHandler func(ctx context.Context, tx *wire.MsgTx,
label string) error
// LndMockServices provides a full set of mocked lnd services.
type LndMockServices struct {
lndclient.LndServices
@ -173,6 +179,8 @@ type LndMockServices struct {
WaitForFinished func()
PublishHandler PublishHandler
lock sync.Mutex
}
@ -278,3 +286,7 @@ func (s *LndMockServices) SetFeeEstimate(confTarget int32,
confTarget, feeEstimate,
)
}
func (s *LndMockServices) SetMinRelayFee(feeEstimate chainfee.SatPerKWeight) {
s.LndServices.WalletKit.(*mockWalletKit).setMinRelayFee(feeEstimate)
}

61
test/walletkit_mock.go
View file

@ -1,6 +1,7 @@
package test
import (
"bytes"
"context"
"errors"
"fmt"
@ -34,6 +35,7 @@ type mockWalletKit struct {
feeEstimateLock sync.Mutex
feeEstimates map[int32]chainfee.SatPerKWeight
minRelayFee chainfee.SatPerKWeight
}
var _ lndclient.WalletKitClient = (*mockWalletKit)(nil)
@ -111,7 +113,13 @@ func (m *mockWalletKit) NextAddr(context.Context, string, walletrpc.AddressType,
}
func (m *mockWalletKit) PublishTransaction(ctx context.Context, tx *wire.MsgTx,
_ string) error {
label string) error {
if m.lnd.PublishHandler != nil {
if err := m.lnd.PublishHandler(ctx, tx, label); err != nil {
return err
}
}
m.lnd.AddTx(tx)
m.lnd.TxPublishChannel <- tx
@ -169,6 +177,24 @@ func (m *mockWalletKit) EstimateFeeRate(ctx context.Context,
return feeEstimate, nil
}
func (m *mockWalletKit) setMinRelayFee(fee chainfee.SatPerKWeight) {
m.feeEstimateLock.Lock()
defer m.feeEstimateLock.Unlock()
m.minRelayFee = fee
}
// MinRelayFee returns the current minimum relay fee based on our chain backend
// in sat/kw. It can be set with setMinRelayFee.
func (m *mockWalletKit) MinRelayFee(
ctx context.Context) (chainfee.SatPerKWeight, error) {
m.feeEstimateLock.Lock()
defer m.feeEstimateLock.Unlock()
return m.minRelayFee, nil
}
// ListSweeps returns a list of the sweep transaction ids known to our node.
func (m *mockWalletKit) ListSweeps(_ context.Context, _ int32) (
[]string, error) {
@ -227,6 +253,25 @@ func (m *mockWalletKit) FundPsbt(_ context.Context,
return nil, 0, nil, nil
}
// finalScriptWitness is a sample signature suitable to put into PSBT.
var finalScriptWitness = func() []byte {
const pver = 0
var buf bytes.Buffer
// Write the number of witness elements.
if err := wire.WriteVarInt(&buf, pver, 1); err != nil {
panic(err)
}
// Write a single witness element with a signature.
signature := make([]byte, 64)
if err := wire.WriteVarBytes(&buf, pver, signature); err != nil {
panic(err)
}
return buf.Bytes()
}()
// SignPsbt expects a partial transaction with all inputs and outputs
// fully declared and tries to sign all unsigned inputs that have all
// required fields (UTXO information, BIP32 derivation information,
@ -239,9 +284,19 @@ func (m *mockWalletKit) FundPsbt(_ context.Context,
// locking or input/output/fee value validation, PSBT finalization). Any
// input that is incomplete will be skipped.
func (m *mockWalletKit) SignPsbt(_ context.Context,
_ *psbt.Packet) (*psbt.Packet, error) {
packet *psbt.Packet) (*psbt.Packet, error) {
return nil, nil
inputs := make([]psbt.PInput, len(packet.Inputs))
copy(inputs, packet.Inputs)
for i := range inputs {
inputs[i].FinalScriptWitness = finalScriptWitness
}
signedPacket := *packet
signedPacket.Inputs = inputs
return &signedPacket, nil
}
// FinalizePsbt expects a partial transaction with all inputs and