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

sweepbatcher: add mode with presigned transactions

Browse source 
In this mode sweepbatcher uses transactions provided by the Presigned helper.
Transactions are signed upon adding an input to a batch.

A single Batcher instance can handle both presigned and regular batches.
Currently presigned and non-presigned sweeps never appear in the same batch.
This commit is contained in:
Boris Nagaev 2025-04-01 12:52:57 -03:00
parent a4a2bfbee3
commit b9b256f4b3
No known key found for this signature in database
7 changed files with 4355 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

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

292
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.
@ -466,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.
@ -586,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.
@ -880,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)
}
@ -997,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 (
@ -1022,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)
@ -1820,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 {
@ -1834,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(
@ -1907,19 +2099,20 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
// Iterate over the purge list and feed the sweeps back to the
// batcher.
for _, sweep := range purgeList {
err := b.purger(ctx, &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
}
@ -1932,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
@ -1976,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

212
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
@ -232,6 +282,14 @@ type addSweepsRequest struct {
// 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 {
@ -366,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.
@ -406,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.
@ -479,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,
@ -533,6 +608,7 @@ func NewBatcher(wallet lndclient.WalletKitClient,
txLabeler: cfg.txLabeler,
customMuSig2Signer: cfg.customMuSig2Signer,
publishErrorHandler: cfg.publishErrorHandler,
presignedHelper: cfg.presignedHelper,
}
}
@ -570,7 +646,10 @@ func (b *Batcher) Run(ctx context.Context) error {
for {
select {
case req := <-b.addSweepsChan:
err = b.handleSweeps(runCtx, req.sweeps, req.notifier)
err = b.handleSweeps(
runCtx, req.sweeps, req.notifier, req.completed,
req.parentBatch,
)
if err != nil {
warnf("handleSweeps failed: %v.", err)
@ -594,6 +673,47 @@ func (b *Batcher) Run(ctx context.Context) 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
@ -613,9 +733,61 @@ func (b *Batcher) AddSweep(ctx context.Context, sweepReq *SweepRequest) error {
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,
sweeps: sweeps,
notifier: sweepReq.Notifier,
completed: completed,
parentBatch: parentBatch,
}
select {
@ -660,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 {
@ -733,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
@ -761,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)
@ -1216,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
}
@ -1226,7 +1378,9 @@ 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,
}
}

1685
sweepbatcher/sweep_batcher_presigned_test.go Normal file
View file

File diff suppressed because it is too large Load diff

10
sweepbatcher/sweep_batcher_test.go
View file

@ -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.