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: allow adding groups of inputs

Browse source 
A group of inputs can be added by passing it in SweepRequest.Inputs field.
All the inputs belong to the same swap and are added to the same batch.
This commit is contained in:
Boris Nagaev 2025-04-01 00:26:01 -03:00
parent dc7261be4a
commit e8cf6386cc
No known key found for this signature in database
6 changed files with 763 additions and 397 deletions
Download patch file Download diff file Expand all files Collapse all files

8
loopout.go
View file

@ -1159,8 +1159,12 @@ func (s *loopOutSwap) waitForHtlcSpendConfirmedV2(globalCtx context.Context,
sweepReq := sweepbatcher.SweepRequest{
SwapHash: s.hash,
Outpoint: htlcOutpoint,
Value: htlcValue,
Inputs: []sweepbatcher.Input{
{
Outpoint: htlcOutpoint,
Value: htlcValue,
},
},
Notifier: &notifier,
}

108
sweepbatcher/greedy_batch_selection.go
View file

@ -15,28 +15,34 @@ import (
"github.com/lightningnetwork/lnd/lnwallet/chainfee"
)
// greedyAddSweep selects a batch for the sweep using the greedy algorithm,
// which minimizes costs, and adds the sweep to the batch. To accomplish this,
// it first collects fee details about the sweep being added, about a potential
// new batch composed of this sweep only, and about all existing batches. It
// greedyAddSweeps selects a batch for the sweeps using the greedy algorithm,
// which minimizes costs, and adds the sweeps to the batch. To accomplish this,
// it first collects fee details about the sweeps being added, about a potential
// new batch composed of these sweeps only, and about all existing batches. It
// skips batches with at least MaxSweepsPerBatch swaps to keep tx standard. Then
// it passes the data to selectBatches() function, which emulates adding the
// sweep to each batch and creating new batch for the sweep, and calculates the
// sweep to each batch and creating new batch for the sweeps, and calculates the
// costs of each alternative. Based on the estimates of selectBatches(), this
// method adds the sweep to the batch that results in the least overall fee
// increase, or creates new batch for it. If the sweep is not accepted by an
// method adds the sweeps to the batch that results in the least overall fee
// increase, or creates new batch for it. If the sweeps are not accepted by an
// existing batch (may happen because of too distant timeouts), next batch is
// tried in the list returned by selectBatches(). If adding fails or new batch
// creation fails, this method returns an error. If this method fails for any
// reason, the caller falls back to the simple algorithm (method handleSweep).
func (b *Batcher) greedyAddSweep(ctx context.Context, sweep *sweep) error {
func (b *Batcher) greedyAddSweeps(ctx context.Context, sweeps []*sweep) error {
if len(sweeps) == 0 {
return fmt.Errorf("trying to greedy add an empty sweeps group")
}
swap := sweeps[0].swapHash
// Collect weight and fee rate info about the sweep and new batch.
sweepFeeDetails, newBatchFeeDetails, err := estimateSweepFeeIncrement(
sweep,
sweeps,
)
if err != nil {
return fmt.Errorf("failed to estimate tx weight for "+
"sweep %x: %w", sweep.swapHash[:6], err)
"sweep %x: %w", swap[:6], err)
}
// Collect weight and fee rate info about existing batches.
@ -64,30 +70,30 @@ func (b *Batcher) greedyAddSweep(ctx context.Context, sweep *sweep) error {
)
if err != nil {
return fmt.Errorf("batch selection algorithm failed for sweep "+
"%x: %w", sweep.swapHash[:6], err)
"%x: %w", swap[:6], err)
}
// Try batches, starting with the best.
for _, batchId := range batchesIds {
// If the best option is to start new batch, do it.
if batchId == newBatchSignal {
return b.spinUpNewBatch(ctx, sweep)
return b.spinUpNewBatch(ctx, sweeps)
}
// Locate the batch to add the sweep to.
// Locate the batch to add the sweeps to.
bestBatch, has := b.batches[batchId]
if !has {
return fmt.Errorf("batch selection algorithm returned "+
"batch id %d which doesn't exist, for sweep %x",
batchId, sweep.swapHash[:6])
batchId, swap[:6])
}
// Add the sweep to the batch.
accepted, err := bestBatch.addSweep(ctx, sweep)
// Add the sweeps to the batch.
accepted, err := bestBatch.addSweeps(ctx, sweeps)
if err != nil {
return fmt.Errorf("batch selection algorithm returned "+
"batch id %d for sweep %x, but adding failed: "+
"%w", batchId, sweep.swapHash[:6], err)
"%w", batchId, swap[:6], err)
}
if accepted {
return nil
@ -95,23 +101,34 @@ func (b *Batcher) greedyAddSweep(ctx context.Context, sweep *sweep) error {
debugf("Batch selection algorithm returned batch id %d "+
"for sweep %x, but acceptance failed.", batchId,
sweep.swapHash[:6])
swap[:6])
}
return fmt.Errorf("no batch accepted sweep %x", sweep.swapHash[:6])
return fmt.Errorf("no batch accepted sweep group %x", swap[:6])
}
// estimateSweepFeeIncrement returns fee details for adding the sweep to a batch
// and for creating new batch with this sweep only.
func estimateSweepFeeIncrement(s *sweep) (feeDetails, feeDetails, error) {
// Create a fake batch with this sweep.
// estimateSweepFeeIncrement returns fee details for adding the sweeps to
// a batch and for creating new batch with these sweeps only.
func estimateSweepFeeIncrement(
sweeps []*sweep) (feeDetails, feeDetails, error) {
if len(sweeps) == 0 {
return feeDetails{}, feeDetails{}, fmt.Errorf("estimating an " +
"empty group of sweeps")
}
// Create a fake batch with the sweeps.
batch := &batch{
rbfCache: rbfCache{
FeeRate: s.minFeeRate,
},
sweeps: map[wire.OutPoint]sweep{
s.outpoint: *s,
FeeRate: sweeps[0].minFeeRate,
},
sweeps: make(map[wire.OutPoint]sweep, len(sweeps)),
}
for _, s := range sweeps {
batch.sweeps[s.outpoint] = *s
batch.rbfCache.FeeRate = max(
batch.rbfCache.FeeRate, s.minFeeRate,
)
}
// Estimate new batch.
@ -120,14 +137,17 @@ func estimateSweepFeeIncrement(s *sweep) (feeDetails, feeDetails, error) {
return feeDetails{}, feeDetails{}, err
}
// Add the same sweep again to measure weight increments.
outpoint2 := s.outpoint
outpoint2.Hash[0]++
if _, has := batch.sweeps[outpoint2]; has {
return feeDetails{}, feeDetails{}, fmt.Errorf("dummy outpoint "+
"%s is present in the batch", outpoint2)
// Add the same sweeps again with different outpoints to measure weight
// increments.
for _, s := range sweeps {
dummy := s.outpoint
dummy.Hash[0]++
if _, has := batch.sweeps[dummy]; has {
return feeDetails{}, feeDetails{}, fmt.Errorf("dummy "+
"outpoint %s is present in the batch", dummy)
}
batch.sweeps[dummy] = *s
}
batch.sweeps[outpoint2] = *s
// Estimate weight of a batch with two sweeps.
fd2, err := estimateBatchWeight(batch)
@ -137,8 +157,8 @@ func estimateSweepFeeIncrement(s *sweep) (feeDetails, feeDetails, error) {
// Create feeDetails for sweep.
sweepFeeDetails := feeDetails{
FeeRate: s.minFeeRate,
IsExternalAddr: s.isExternalAddr,
FeeRate: batch.rbfCache.FeeRate,
IsExternalAddr: sweeps[0].isExternalAddr,
// Calculate sweep weight as a difference.
Weight: fd2.Weight - fd1.Weight,
@ -252,10 +272,10 @@ func (e1 feeDetails) combine(e2 feeDetails) feeDetails {
// rate and a weight is provided. Also, a hint is provided to signal which
// spending path will be used by the batch.
//
// The same data is also provided for the sweep for which we are selecting a
// batch to add. In case of the sweep weights are weight deltas resulted from
// adding the sweep. Finally, the same data is provided for new batch having
// this sweep only.
// The same data is also provided for the sweep (or sweeps) for which we are
// selecting a batch to add. In case of the sweep weights are weight deltas
// resulted from adding the sweep. Finally, the same data is provided for new
// batch having this sweep(s) only.
//
// The algorithm compares costs of adding the sweep to each existing batch, and
// costs of new batch creation for this sweep and returns BatchId of the winning
@ -265,11 +285,11 @@ func (e1 feeDetails) combine(e2 feeDetails) feeDetails {
// having flag IsExternalAddr must go in individual batches. Cooperative
// spending may only be available for some sweeps supporting it, not for all.
func selectBatches(batches []feeDetails,
sweep, oneSweepBatch feeDetails) ([]int32, error) {
added, newBatch feeDetails) ([]int32, error) {
// If the sweep has IsExternalAddr flag, the sweep can't be added to
// a batch, so create new batch for it.
if sweep.IsExternalAddr {
if added.IsExternalAddr {
return []int32{newBatchSignal}, nil
}
@ -286,7 +306,7 @@ func selectBatches(batches []feeDetails,
// creation with this sweep only in it. The cost is its full fee.
alternatives = append(alternatives, alternative{
batchId: newBatchSignal,
cost: oneSweepBatch.fee(),
cost: newBatch.fee(),
})
// Try to add the sweep to every batch, calculate the costs and
@ -299,7 +319,7 @@ func selectBatches(batches []feeDetails,
}
// Add the sweep to the batch virtually.
combinedBatch := batch.combine(sweep)
combinedBatch := batch.combine(added)
// The cost is the fee increase.
cost := combinedBatch.fee() - batch.fee()

97
sweepbatcher/greedy_batch_selection_test.go
View file

@ -74,17 +74,28 @@ func TestEstimateSweepFeeIncrement(t *testing.T) {
trAddr := (*btcutil.AddressTaproot)(nil)
p2pkhAddr := (*btcutil.AddressPubKeyHash)(nil)
outpoint1 := wire.OutPoint{
Hash: chainhash.Hash{1, 1, 1},
Index: 1,
}
outpoint2 := wire.OutPoint{
Hash: chainhash.Hash{2, 2, 2},
Index: 2,
}
cases := []struct {
name string
sweep *sweep
sweeps []*sweep
wantSweepFeeDetails feeDetails
wantNewBatchFeeDetails feeDetails
}{
{
name: "regular",
sweep: &sweep{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
sweeps: []*sweep{
{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
},
},
wantSweepFeeDetails: feeDetails{
FeeRate: lowFeeRate,
@ -98,9 +109,11 @@ func TestEstimateSweepFeeIncrement(t *testing.T) {
{
name: "high fee rate",
sweep: &sweep{
minFeeRate: highFeeRate,
htlcSuccessEstimator: se3,
sweeps: []*sweep{
{
minFeeRate: highFeeRate,
htlcSuccessEstimator: se3,
},
},
wantSweepFeeDetails: feeDetails{
FeeRate: highFeeRate,
@ -114,11 +127,13 @@ func TestEstimateSweepFeeIncrement(t *testing.T) {
{
name: "isExternalAddr taproot",
sweep: &sweep{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
isExternalAddr: true,
destAddr: trAddr,
sweeps: []*sweep{
{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
isExternalAddr: true,
destAddr: trAddr,
},
},
wantSweepFeeDetails: feeDetails{
FeeRate: lowFeeRate,
@ -134,11 +149,13 @@ func TestEstimateSweepFeeIncrement(t *testing.T) {
{
name: "isExternalAddr P2PKH",
sweep: &sweep{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
isExternalAddr: true,
destAddr: p2pkhAddr,
sweeps: []*sweep{
{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
isExternalAddr: true,
destAddr: p2pkhAddr,
},
},
wantSweepFeeDetails: feeDetails{
FeeRate: lowFeeRate,
@ -155,10 +172,12 @@ func TestEstimateSweepFeeIncrement(t *testing.T) {
{
name: "non-coop",
sweep: &sweep{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
nonCoopHint: true,
sweeps: []*sweep{
{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
nonCoopHint: true,
},
},
wantSweepFeeDetails: feeDetails{
FeeRate: lowFeeRate,
@ -172,10 +191,12 @@ func TestEstimateSweepFeeIncrement(t *testing.T) {
{
name: "coop-failed",
sweep: &sweep{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
coopFailed: true,
sweeps: []*sweep{
{
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
coopFailed: true,
},
},
wantSweepFeeDetails: feeDetails{
FeeRate: lowFeeRate,
@ -186,12 +207,36 @@ func TestEstimateSweepFeeIncrement(t *testing.T) {
Weight: nonCoopNewBatchWeight,
},
},
{
name: "two sweeps",
sweeps: []*sweep{
{
outpoint: outpoint1,
minFeeRate: lowFeeRate,
htlcSuccessEstimator: se3,
},
{
outpoint: outpoint2,
minFeeRate: highFeeRate,
htlcSuccessEstimator: se3,
},
},
wantSweepFeeDetails: feeDetails{
FeeRate: highFeeRate,
Weight: coopInputWeight * 2,
},
wantNewBatchFeeDetails: feeDetails{
FeeRate: highFeeRate,
Weight: coopNewBatchWeight + coopInputWeight,
},
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
gotSweepFeeDetails, gotNewBatchFeeDetails, err :=
estimateSweepFeeIncrement(tc.sweep)
estimateSweepFeeIncrement(tc.sweeps)
require.NoError(t, err)
require.Equal(
t, tc.wantSweepFeeDetails, gotSweepFeeDetails,

252
sweepbatcher/sweep_batch.go
View file

@ -465,55 +465,19 @@ func (b *batch) Errorf(format string, params ...interface{}) {
b.log().Errorf(format, params...)
}
// addSweep tries to add a sweep to the batch. If this is the first sweep being
// added to the batch then it also sets the primary sweep ID.
func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) {
done, err := b.scheduleNextCall()
defer done()
if err != nil {
return false, err
}
// 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.
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.
if sweep == nil {
b.Infof("the sweep is nil")
return false, nil
return false, fmt.Errorf("the sweep is nil")
}
// Before we run through the acceptance checks, let's just see if this
// sweep is already in our batch. In that case, just update the sweep.
oldSweep, ok := b.sweeps[sweep.outpoint]
if ok {
// Preserve coopFailed value not to forget about cooperative
// spending failure in this sweep.
tmp := *sweep
tmp.coopFailed = oldSweep.coopFailed
// If the sweep was resumed from storage, and the swap requested
// to sweep again, a new sweep notifier will be created by the
// swap. By re-assigning to the batch's sweep we make sure that
// everything, including the notifier, is up to date.
b.sweeps[sweep.outpoint] = tmp
// If this is the primary sweep, we also need to update the
// batch's confirmation target and fee rate.
if b.primarySweepID == sweep.outpoint {
b.cfg.batchConfTarget = sweep.confTarget
b.rbfCache.SkipNextBump = true
}
// Update batch's fee rate to be greater than or equal to
// minFeeRate of the sweep. Make sure batch's fee rate does not
// decrease (otherwise it won't pass RBF rules and won't be
// broadcasted) and that it is not lower that minFeeRate of
// other sweeps (so it is applied).
if b.rbfCache.FeeRate < sweep.minFeeRate {
b.rbfCache.FeeRate = sweep.minFeeRate
}
if _, ok := b.sweeps[sweep.outpoint]; ok {
return true, nil
}
@ -521,19 +485,16 @@ func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) {
// the batch, do not add another sweep to prevent the tx from becoming
// non-standard.
if len(b.sweeps) >= MaxSweepsPerBatch {
b.Infof("the batch has already too many sweeps %d >= %d",
len(b.sweeps), MaxSweepsPerBatch)
return false, nil
return false, fmt.Errorf("the batch has already too many "+
"sweeps %d >= %d", len(b.sweeps), MaxSweepsPerBatch)
}
// Since all the actions of the batch happen sequentially, we could
// arrive here after the batch got closed because of a spend. In this
// case we cannot add the sweep to this batch.
if b.state != Open {
b.Infof("the batch state (%v) is not open", b.state)
return false, nil
return false, fmt.Errorf("the batch state (%v) is not open",
b.state)
}
// If this batch contains a single sweep that spends to a non-wallet
@ -541,17 +502,15 @@ func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) {
// we cannot add this sweep to the batch.
for _, s := range b.sweeps {
if s.isExternalAddr {
b.Infof("the batch already has a sweep %x with "+
"an external address", s.swapHash[:6])
return false, nil
return false, fmt.Errorf("the batch already has a "+
"sweep %x with an external address",
s.swapHash[:6])
}
if sweep.isExternalAddr {
b.Infof("the batch is not empty and new sweep %x "+
"has an external address", sweep.swapHash[:6])
return false, nil
return false, fmt.Errorf("the batch is not empty and "+
"new sweep %x has an external address",
sweep.swapHash[:6])
}
}
@ -563,46 +522,165 @@ func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) {
int32(math.Abs(float64(sweep.timeout - s.timeout)))
if timeoutDistance > b.cfg.maxTimeoutDistance {
b.Infof("too long timeout distance between the "+
"batch and sweep %x: %d > %d",
return false, fmt.Errorf("too long timeout distance "+
"between the batch and sweep %x: %d > %d",
sweep.swapHash[:6], timeoutDistance,
b.cfg.maxTimeoutDistance)
}
}
// Everything is ok, the sweep can be added to the batch.
return false, nil
}
// addSweeps tries to add sweeps to the batch. If this is the first sweep being
// added to the batch then it also sets the primary sweep ID. It returns if the
// sweeps were accepted to the batch.
func (b *batch) addSweeps(ctx context.Context, sweeps []*sweep) (bool, error) {
done, err := b.scheduleNextCall()
defer done()
if err != nil {
return false, err
}
// This must be a bug, so log a warning.
if len(sweeps) == 0 {
b.Warnf("An attempt to add zero sweeps.")
return false, nil
}
// Track how many new and existing sweeps are among the sweeps.
var numExisting, numNew int
for _, s := range sweeps {
existing, err := b.checkSweepToAdd(ctx, s)
if err != nil {
b.Infof("Failed to add sweep %v to batch %d: %v",
s.outpoint, b.id, err)
return false, nil
}
if existing {
numExisting++
} else {
numNew++
}
}
// Make sure the whole group is either new or existing. If this is not
// the case, this might be a bug, so print a warning.
if numExisting > 0 && numNew > 0 {
b.Warnf("There are %d existing and %d new sweeps among the "+
"group. They must not be mixed.", numExisting, numNew)
return false, nil
}
// Make sure all the sweeps spend different outpoints.
outpointsSet := make(map[wire.OutPoint]struct{}, len(sweeps))
for _, s := range sweeps {
if _, has := outpointsSet[s.outpoint]; has {
b.Warnf("Multiple sweeps spend outpoint %v", s.outpoint)
return false, nil
}
outpointsSet[s.outpoint] = struct{}{}
}
// Past this point we know that a new incoming sweep passes the
// acceptance criteria and is now ready to be added to this batch.
// If this is the first sweep being added to the batch, make it the
// primary sweep.
if b.primarySweepID == zeroSweepID {
b.primarySweepID = sweep.outpoint
b.cfg.batchConfTarget = sweep.confTarget
b.rbfCache.FeeRate = sweep.minFeeRate
b.rbfCache.SkipNextBump = true
// For an existing group, update the sweeps in the batch.
if numExisting == len(sweeps) {
for _, s := range sweeps {
oldSweep, ok := b.sweeps[s.outpoint]
if !ok {
return false, fmt.Errorf("sweep %v not found "+
"in batch %d", s.outpoint, b.id)
}
// We also need to start the spend monitor for this new primary
// sweep.
err := b.monitorSpend(ctx, *sweep)
if err != nil {
return false, err
// Preserve coopFailed value not to forget about
// cooperative spending failure in this sweep.
tmp := *s
tmp.coopFailed = oldSweep.coopFailed
// If the sweep was resumed from storage, and the swap
// requested to sweep again, a new sweep notifier will
// be created by the swap. By re-assigning to the
// batch's sweep we make sure that everything, including
// the notifier, is up to date.
b.sweeps[s.outpoint] = tmp
// If this is the primary sweep, we also need to update
// the batch's confirmation target and fee rate.
if b.primarySweepID == s.outpoint {
b.cfg.batchConfTarget = s.confTarget
b.rbfCache.SkipNextBump = true
}
// Update batch's fee rate to be greater than or equal
// to minFeeRate of the sweep. Make sure batch's fee
// rate does not decrease (otherwise it won't pass RBF
// rules and won't be broadcasted) and that it is not
// lower that minFeeRate of other sweeps (so it is
// applied).
if b.rbfCache.FeeRate < s.minFeeRate {
b.rbfCache.FeeRate = s.minFeeRate
}
}
return true, nil
} else if numNew != len(sweeps) {
// Sanity check: all the sweeps must be either existing or new.
// We have checked this above, let's check here as well.
return false, fmt.Errorf("bug in numExisting and numNew logic:"+
" numExisting=%d, numNew=%d, len(sweeps)=%d, "+
"len(b.sweeps)=%d", numExisting, numNew, len(sweeps),
len(b.sweeps))
}
// Here is the code to add new sweeps to a batch.
for _, s := range sweeps {
// If this is the first sweep being added to the batch, make it
// the primary sweep.
if b.primarySweepID == zeroSweepID {
b.primarySweepID = s.outpoint
b.cfg.batchConfTarget = s.confTarget
b.rbfCache.FeeRate = s.minFeeRate
b.rbfCache.SkipNextBump = true
// We also need to start the spend monitor for this new
// primary sweep.
err := b.monitorSpend(ctx, *s)
if err != nil {
return false, err
}
}
// Make sure the sweep is not present in the batch. If it is
// present, this is a bug, return an error to stop here.
if _, has := b.sweeps[s.outpoint]; has {
return false, fmt.Errorf("sweep %v is already present "+
"in batch %d", s.outpoint, b.id)
}
// Add the sweep to the batch's sweeps.
b.Infof("adding sweep %v, swap %x", s.outpoint, s.swapHash[:6])
b.sweeps[s.outpoint] = *s
// Update FeeRate. Max(s.minFeeRate) for all the sweeps of
// the batch is the basis for fee bumps.
if b.rbfCache.FeeRate < s.minFeeRate {
b.rbfCache.FeeRate = s.minFeeRate
b.rbfCache.SkipNextBump = true
}
if err := b.persistSweep(ctx, *s, false); err != nil {
return true, err
}
}
// Add the sweep to the batch's sweeps.
b.Infof("adding sweep %x", sweep.swapHash[:6])
b.sweeps[sweep.outpoint] = *sweep
// Update FeeRate. Max(sweep.minFeeRate) for all the sweeps of
// the batch is the basis for fee bumps.
if b.rbfCache.FeeRate < sweep.minFeeRate {
b.rbfCache.FeeRate = sweep.minFeeRate
b.rbfCache.SkipNextBump = true
}
return true, b.persistSweep(ctx, *sweep, false)
return true, nil
}
// sweepExists returns true if the batch contains the sweep with the given
@ -1754,8 +1832,12 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
delete(b.sweeps, sweep.outpoint)
purgeList = append(purgeList, SweepRequest{
SwapHash: newSweep.swapHash,
Outpoint: newSweep.outpoint,
Value: newSweep.value,
Inputs: []Input{
{
Outpoint: newSweep.outpoint,
Value: newSweep.value,
},
},
Notifier: newSweep.notifier,
})
}

81
sweepbatcher/sweep_batcher.go
View file

@ -199,16 +199,23 @@ func defaultPublishErrorLogger(err error, errMsg string, log btclog.Logger) {
log.Warnf("%s: %v", errMsg, err)
}
// SweepRequest is a request to sweep a specific outpoint.
type SweepRequest struct {
// SwapHash is the hash of the swap that is being swept.
SwapHash lntypes.Hash
// Input specifies an UTXO with amount that is added to the batcher.
type Input struct {
// Outpoint is the outpoint that is being swept.
Outpoint wire.OutPoint
// Value is the value of the outpoint that is being swept.
Value btcutil.Amount
}
// SweepRequest is a request to sweep an outpoint or a group of outpoints.
type SweepRequest struct {
// SwapHash is the hash of the swap that is being swept.
SwapHash lntypes.Hash
// Inputs specifies the inputs in the same request. All the inputs
// belong to the same swap and are added to the same batch.
Inputs []Input
// Notifier is a notifier that is used to notify the requester of this
// sweep that the sweep was successful.
@ -551,16 +558,16 @@ func (b *Batcher) Run(ctx context.Context) error {
for {
select {
case sweepReq := <-b.sweepReqs:
sweep, err := b.fetchSweep(runCtx, sweepReq)
sweeps, err := b.fetchSweeps(runCtx, sweepReq)
if err != nil {
warnf("fetchSweep failed: %v.", err)
warnf("fetchSweeps failed: %v.", err)
return err
}
err = b.handleSweep(runCtx, sweep, sweepReq.Notifier)
err = b.handleSweeps(runCtx, sweeps, sweepReq.Notifier)
if err != nil {
warnf("handleSweep failed: %v.", err)
warnf("handleSweeps failed: %v.", err)
return err
}
@ -624,11 +631,19 @@ func (b *Batcher) testRunInEventLoop(ctx context.Context, handler func()) {
}
}
// handleSweep handles a sweep request by either placing it in an existing
// batch, or by spinning up a new batch for it.
func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
// 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")
}
// 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
@ -675,7 +690,7 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
// provide the sweep to that batch and return.
for _, batch := range b.batches {
if batch.sweepExists(sweep.outpoint) {
accepted, err := batch.addSweep(ctx, sweep)
accepted, err := batch.addSweeps(ctx, sweeps)
if err != nil && !errors.Is(err, ErrBatchShuttingDown) {
return err
}
@ -692,7 +707,7 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
}
// Try to run the greedy algorithm of batch selection to minimize costs.
err = b.greedyAddSweep(ctx, sweep)
err = b.greedyAddSweeps(ctx, sweeps)
if err == nil {
// The greedy algorithm succeeded.
return nil
@ -703,7 +718,7 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
// If one of the batches accepts the sweep, we provide it to that batch.
for _, batch := range b.batches {
accepted, err := batch.addSweep(ctx, sweep)
accepted, err := batch.addSweeps(ctx, sweeps)
if err != nil && !errors.Is(err, ErrBatchShuttingDown) {
return err
}
@ -717,28 +732,28 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
// If no batch is capable of accepting the sweep, we spin up a fresh
// batch and hand the sweep over to it.
return b.spinUpNewBatch(ctx, sweep)
return b.spinUpNewBatch(ctx, sweeps)
}
// spinUpNewBatch creates new batch, starts it and adds the sweep to it.
func (b *Batcher) spinUpNewBatch(ctx context.Context, sweep *sweep) error {
// spinUpNewBatch creates new batch, starts it and adds the sweeps to it.
func (b *Batcher) spinUpNewBatch(ctx context.Context, sweeps []*sweep) error {
// Spin up a fresh batch.
newBatch, err := b.spinUpBatch(ctx)
if err != nil {
return err
}
// Add the sweep to the fresh batch.
accepted, err := newBatch.addSweep(ctx, sweep)
// Add the sweeps to the fresh batch.
accepted, err := newBatch.addSweeps(ctx, sweeps)
if err != nil {
return err
}
// If the sweep wasn't accepted by the fresh batch something is wrong,
// If the sweeps weren't accepted by the fresh batch something is wrong,
// we should return the error.
if !accepted {
return fmt.Errorf("sweep %x was not accepted by new batch %d",
sweep.swapHash[:6], newBatch.id)
sweeps[0].swapHash[:6], newBatch.id)
}
return nil
@ -1101,12 +1116,24 @@ func NewSweepFetcherFromSwapStore(swapStore LoopOutFetcher,
}, nil
}
// fetchSweep fetches the sweep related information from the database.
func (b *Batcher) fetchSweep(ctx context.Context,
sweepReq SweepRequest) (*sweep, error) {
// fetchSweeps fetches the sweep related information from the database.
func (b *Batcher) fetchSweeps(ctx context.Context,
sweepReq SweepRequest) ([]*sweep, error) {
return b.loadSweep(ctx, sweepReq.SwapHash, sweepReq.Outpoint,
sweepReq.Value)
sweeps := make([]*sweep, len(sweepReq.Inputs))
for i, utxo := range sweepReq.Inputs {
s, err := b.loadSweep(
ctx, sweepReq.SwapHash, utxo.Outpoint,
utxo.Value,
)
if err != nil {
return nil, fmt.Errorf("failed to load "+
"sweep %v: %w", utxo.Outpoint, err)
}
sweeps[i] = s
}
return sweeps, nil
}
// loadSweep loads inputs of sweep from the database and from FeeRateProvider

614
sweepbatcher/sweep_batcher_test.go
View file

File diff suppressed because it is too large Load diff