GlcoinCore/lnd
1
0
Fork 0
mirror of https://github.com/lightningnetwork/lnd.git synced 2026-08-13 12:32:48 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge pull request #9446 from yyforyongyu/yy-prepare-fee-replace

Browse source 
sweeper: rename `Failed` to `Fatal` and minor refactor
This commit is contained in:
Yong 2025-02-05 22:48:41 +08:00 • committed by GitHub
commit bac699df8f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 266 additions and 158 deletions
Download patch file Download diff file Expand all files Collapse all files

5
docs/release-notes/release-notes-0.19.0.md
View file

@ -87,6 +87,7 @@
## Functional Enhancements
* [Add ability](https://github.com/lightningnetwork/lnd/pull/8998) to paginate
wallet transactions.
## RPC Additions
* [Add a new rpc endpoint](https://github.com/lightningnetwork/lnd/pull/8843)
@ -317,6 +318,10 @@ The underlying functionality between those two options remain the same.
StateMachine](https://github.com/lightningnetwork/lnd/pull/9342) to use the
new GoroutineManager API along with structured logging.
* A minor [refactor](https://github.com/lightningnetwork/lnd/pull/9446) is done
to the sweeper to improve code quality, with a renaming of the internal state
(`Failed` -> `Fatal`) used by the inputs tracked in the sweeper.
## Tooling and Documentation
* [Improved `lncli create` command help text](https://github.com/lightningnetwork/lnd/pull/9077)

164
sweep/fee_bumper.go
View file

@ -410,34 +410,50 @@ func (t *TxPublisher) Broadcast(req *BumpRequest) <-chan *BumpResult {
lnutils.SpewLogClosure(req))
// Store the request.
requestID, record := t.storeInitialRecord(req)
record := t.storeInitialRecord(req)
// Create a chan to send the result to the caller.
subscriber := make(chan *BumpResult, 1)
t.subscriberChans.Store(requestID, subscriber)
t.subscriberChans.Store(record.requestID, subscriber)
// Publish the tx immediately if specified.
if req.Immediate {
t.handleInitialBroadcast(record, requestID)
t.handleInitialBroadcast(record)
}
return subscriber
}
// storeInitialRecord initializes a monitor record and saves it in the map.
func (t *TxPublisher) storeInitialRecord(req *BumpRequest) (
uint64, *monitorRecord) {
func (t *TxPublisher) storeInitialRecord(req *BumpRequest) *monitorRecord {
// Increase the request counter.
//
// NOTE: this is the only place where we increase the counter.
requestID := t.requestCounter.Add(1)
// Register the record.
record := &monitorRecord{req: req}
record := &monitorRecord{
requestID: requestID,
req: req,
}
t.records.Store(requestID, record)
return requestID, record
return record
}
// updateRecord updates the given record's tx and fee, and saves it in the
// records map.
func (t *TxPublisher) updateRecord(r *monitorRecord,
sweepCtx *sweepTxCtx) *monitorRecord {
r.tx = sweepCtx.tx
r.fee = sweepCtx.fee
r.outpointToTxIndex = sweepCtx.outpointToTxIndex
// Register the record.
t.records.Store(r.requestID, r)
return r
}
// NOTE: part of the `chainio.Consumer` interface.
@ -447,21 +463,30 @@ func (t *TxPublisher) Name() string {
// initializeTx initializes a fee function and creates an RBF-compliant tx. If
// succeeded, the initial tx is stored in the records map.
func (t *TxPublisher) initializeTx(requestID uint64, req *BumpRequest) error {
func (t *TxPublisher) initializeTx(r *monitorRecord) (*monitorRecord, error) {
// Create a fee bumping algorithm to be used for future RBF.
feeAlgo, err := t.initializeFeeFunction(req)
feeAlgo, err := t.initializeFeeFunction(r.req)
if err != nil {
return fmt.Errorf("init fee function: %w", err)
return nil, fmt.Errorf("init fee function: %w", err)
}
// Attach the newly created fee function.
//
// TODO(yy): current we'd initialize a monitorRecord before creating the
// fee function, while we could instead create the fee function first
// then save it to the record. To make this happen we need to change the
// conf target calculation below since we would be initializing the fee
// function one block before.
r.feeFunction = feeAlgo
// Create the initial tx to be broadcasted. This tx is guaranteed to
// comply with the RBF restrictions.
err = t.createRBFCompliantTx(requestID, req, feeAlgo)
record, err := t.createRBFCompliantTx(r)
if err != nil {
return fmt.Errorf("create RBF-compliant tx: %w", err)
return nil, fmt.Errorf("create RBF-compliant tx: %w", err)
}
return nil
return record, nil
}
// initializeFeeFunction initializes a fee function to be used for this request
@ -497,29 +522,28 @@ func (t *TxPublisher) initializeFeeFunction(
// so by creating a tx, validate it using `TestMempoolAccept`, and bump its fee
// and redo the process until the tx is valid, or return an error when non-RBF
// related errors occur or the budget has been used up.
func (t *TxPublisher) createRBFCompliantTx(requestID uint64, req *BumpRequest,
f FeeFunction) error {
func (t *TxPublisher) createRBFCompliantTx(
r *monitorRecord) (*monitorRecord, error) {
f := r.feeFunction
for {
// Create a new tx with the given fee rate and check its
// mempool acceptance.
sweepCtx, err := t.createAndCheckTx(req, f)
sweepCtx, err := t.createAndCheckTx(r.req, f)
switch {
case err == nil:
// The tx is valid, store it.
t.storeRecord(
requestID, sweepCtx.tx, req, f, sweepCtx.fee,
sweepCtx.outpointToTxIndex,
)
record := t.updateRecord(r, sweepCtx)
log.Infof("Created initial sweep tx=%v for %v inputs: "+
"feerate=%v, fee=%v, inputs:\n%v",
sweepCtx.tx.TxHash(), len(req.Inputs),
sweepCtx.tx.TxHash(), len(r.req.Inputs),
f.FeeRate(), sweepCtx.fee,
inputTypeSummary(req.Inputs))
inputTypeSummary(r.req.Inputs))
return nil
return record, nil
// If the error indicates the fees paid is not enough, we will
// ask the fee function to increase the fee rate and retry.
@ -550,7 +574,7 @@ func (t *TxPublisher) createRBFCompliantTx(requestID uint64, req *BumpRequest,
// cluster these inputs differetly.
increased, err = f.Increment()
if err != nil {
return err
return nil, err
}
}
@ -560,26 +584,11 @@ func (t *TxPublisher) createRBFCompliantTx(requestID uint64, req *BumpRequest,
// mempool acceptance.
default:
log.Debugf("Failed to create RBF-compliant tx: %v", err)
return err
return nil, err
}
}
}
// storeRecord stores the given record in the records map.
func (t *TxPublisher) storeRecord(requestID uint64, tx *wire.MsgTx,
req *BumpRequest, f FeeFunction, fee btcutil.Amount,
outpointToTxIndex map[wire.OutPoint]int) {
// Register the record.
t.records.Store(requestID, &monitorRecord{
tx: tx,
req: req,
feeFunction: f,
fee: fee,
outpointToTxIndex: outpointToTxIndex,
})
}
// createAndCheckTx creates a tx based on the given inputs, change output
// script, and the fee rate. In addition, it validates the tx's mempool
// acceptance before returning a tx that can be published directly, along with
@ -638,13 +647,7 @@ func (t *TxPublisher) createAndCheckTx(req *BumpRequest,
// the event channel to the record. Any broadcast-related errors will not be
// returned here, instead, they will be put inside the `BumpResult` and
// returned to the caller.
func (t *TxPublisher) broadcast(requestID uint64) (*BumpResult, error) {
// Get the record being monitored.
record, ok := t.records.Load(requestID)
if !ok {
return nil, fmt.Errorf("tx record %v not found", requestID)
}
func (t *TxPublisher) broadcast(record *monitorRecord) (*BumpResult, error) {
txid := record.tx.TxHash()
tx := record.tx
@ -691,7 +694,7 @@ func (t *TxPublisher) broadcast(requestID uint64) (*BumpResult, error) {
Fee: record.fee,
FeeRate: record.feeFunction.FeeRate(),
Err: err,
requestID: requestID,
requestID: record.requestID,
}
return result, nil
@ -777,6 +780,9 @@ func (t *TxPublisher) handleResult(result *BumpResult) {
// monitorRecord is used to keep track of the tx being monitored by the
// publisher internally.
type monitorRecord struct {
// requestID is the ID of the request that created this record.
requestID uint64
// tx is the tx being monitored.
tx *wire.MsgTx
@ -919,35 +925,35 @@ func (t *TxPublisher) processRecords() {
t.records.ForEach(visitor)
// Handle the initial broadcast.
for requestID, r := range initialRecords {
t.handleInitialBroadcast(r, requestID)
for _, r := range initialRecords {
t.handleInitialBroadcast(r)
}
// For records that are confirmed, we'll notify the caller about this
// result.
for requestID, r := range confirmedRecords {
for _, r := range confirmedRecords {
log.Debugf("Tx=%v is confirmed", r.tx.TxHash())
t.wg.Add(1)
go t.handleTxConfirmed(r, requestID)
go t.handleTxConfirmed(r)
}
// Get the current height to be used in the following goroutines.
currentHeight := t.currentHeight.Load()
// For records that are not confirmed, we perform a fee bump if needed.
for requestID, r := range feeBumpRecords {
for _, r := range feeBumpRecords {
log.Debugf("Attempting to fee bump Tx=%v", r.tx.TxHash())
t.wg.Add(1)
go t.handleFeeBumpTx(requestID, r, currentHeight)
go t.handleFeeBumpTx(r, currentHeight)
}
// For records that are failed, we'll notify the caller about this
// result.
for requestID, r := range failedRecords {
for _, r := range failedRecords {
log.Debugf("Tx=%v has inputs been spent by a third party, "+
"failing it now", r.tx.TxHash())
t.wg.Add(1)
go t.handleThirdPartySpent(r, requestID)
go t.handleThirdPartySpent(r)
}
}
@ -955,7 +961,7 @@ func (t *TxPublisher) processRecords() {
// notify the subscriber then remove the record from the maps .
//
// NOTE: Must be run as a goroutine to avoid blocking on sending the result.
func (t *TxPublisher) handleTxConfirmed(r *monitorRecord, requestID uint64) {
func (t *TxPublisher) handleTxConfirmed(r *monitorRecord) {
defer t.wg.Done()
// Create a result that will be sent to the resultChan which is
@ -963,7 +969,7 @@ func (t *TxPublisher) handleTxConfirmed(r *monitorRecord, requestID uint64) {
result := &BumpResult{
Event: TxConfirmed,
Tx: r.tx,
requestID: requestID,
requestID: r.requestID,
Fee: r.fee,
FeeRate: r.feeFunction.FeeRate(),
}
@ -1021,10 +1027,8 @@ func (t *TxPublisher) handleInitialTxError(requestID uint64, err error) {
// 1. init a fee function based on the given strategy.
// 2. create an RBF-compliant tx and monitor it for confirmation.
// 3. notify the initial broadcast result back to the caller.
func (t *TxPublisher) handleInitialBroadcast(r *monitorRecord,
requestID uint64) {
log.Debugf("Initial broadcast for requestID=%v", requestID)
func (t *TxPublisher) handleInitialBroadcast(r *monitorRecord) {
log.Debugf("Initial broadcast for requestID=%v", r.requestID)
var (
result *BumpResult
@ -1035,18 +1039,18 @@ func (t *TxPublisher) handleInitialBroadcast(r *monitorRecord,
// RBF rules.
//
// Create the initial tx to be broadcasted.
err = t.initializeTx(requestID, r.req)
record, err := t.initializeTx(r)
if err != nil {
log.Errorf("Initial broadcast failed: %v", err)
// We now handle the initialization error and exit.
t.handleInitialTxError(requestID, err)
t.handleInitialTxError(r.requestID, err)
return
}
// Successfully created the first tx, now broadcast it.
result, err = t.broadcast(requestID)
result, err = t.broadcast(record)
if err != nil {
// The broadcast failed, which can only happen if the tx record
// cannot be found or the aux sweeper returns an error. In
@ -1055,7 +1059,7 @@ func (t *TxPublisher) handleInitialBroadcast(r *monitorRecord,
result = &BumpResult{
Event: TxFailed,
Err: err,
requestID: requestID,
requestID: r.requestID,
}
}
@ -1066,9 +1070,7 @@ func (t *TxPublisher) handleInitialBroadcast(r *monitorRecord,
// attempt to bump the fee of the tx.
//
// NOTE: Must be run as a goroutine to avoid blocking on sending the result.
func (t *TxPublisher) handleFeeBumpTx(requestID uint64, r *monitorRecord,
currentHeight int32) {
func (t *TxPublisher) handleFeeBumpTx(r *monitorRecord, currentHeight int32) {
defer t.wg.Done()
oldTxid := r.tx.TxHash()
@ -1099,7 +1101,7 @@ func (t *TxPublisher) handleFeeBumpTx(requestID uint64, r *monitorRecord,
// The fee function now has a new fee rate, we will use it to bump the
// fee of the tx.
resultOpt := t.createAndPublishTx(requestID, r)
resultOpt := t.createAndPublishTx(r)
// If there's a result, we will notify the caller about the result.
resultOpt.WhenSome(func(result BumpResult) {
@ -1113,9 +1115,7 @@ func (t *TxPublisher) handleFeeBumpTx(requestID uint64, r *monitorRecord,
// and send a TxFailed event to the subscriber.
//
// NOTE: Must be run as a goroutine to avoid blocking on sending the result.
func (t *TxPublisher) handleThirdPartySpent(r *monitorRecord,
requestID uint64) {
func (t *TxPublisher) handleThirdPartySpent(r *monitorRecord) {
defer t.wg.Done()
// Create a result that will be sent to the resultChan which is
@ -1127,7 +1127,7 @@ func (t *TxPublisher) handleThirdPartySpent(r *monitorRecord,
result := &BumpResult{
Event: TxFailed,
Tx: r.tx,
requestID: requestID,
requestID: r.requestID,
Err: ErrThirdPartySpent,
}
@ -1138,7 +1138,7 @@ func (t *TxPublisher) handleThirdPartySpent(r *monitorRecord,
// createAndPublishTx creates a new tx with a higher fee rate and publishes it
// to the network. It will update the record with the new tx and fee rate if
// successfully created, and return the result when published successfully.
func (t *TxPublisher) createAndPublishTx(requestID uint64,
func (t *TxPublisher) createAndPublishTx(
r *monitorRecord) fn.Option[BumpResult] {
// Fetch the old tx.
@ -1189,22 +1189,16 @@ func (t *TxPublisher) createAndPublishTx(requestID uint64,
Event: TxFailed,
Tx: oldTx,
Err: err,
requestID: requestID,
requestID: r.requestID,
})
}
// The tx has been created without any errors, we now register a new
// record by overwriting the same requestID.
t.records.Store(requestID, &monitorRecord{
tx: sweepCtx.tx,
req: r.req,
feeFunction: r.feeFunction,
fee: sweepCtx.fee,
outpointToTxIndex: sweepCtx.outpointToTxIndex,
})
record := t.updateRecord(r, sweepCtx)
// Attempt to broadcast this new tx.
result, err := t.broadcast(requestID)
result, err := t.broadcast(record)
if err != nil {
log.Infof("Failed to broadcast replacement tx %v: %v",
sweepCtx.tx.TxHash(), err)

182
sweep/fee_bumper_test.go
View file

@ -313,9 +313,9 @@ func TestInitializeFeeFunction(t *testing.T) {
require.Equal(t, feerate, f.FeeRate())
}
// TestStoreRecord correctly increases the request counter and saves the
// TestUpdateRecord correctly updates the fields fee and tx, and saves the
// record.
func TestStoreRecord(t *testing.T) {
func TestUpdateRecord(t *testing.T) {
t.Parallel()
// Create a test input.
@ -351,8 +351,22 @@ func TestStoreRecord(t *testing.T) {
op: 0,
}
// Create a sweepTxCtx.
sweepCtx := &sweepTxCtx{
tx: tx,
fee: fee,
outpointToTxIndex: utxoIndex,
}
// Create a test record.
record := &monitorRecord{
requestID: initialCounter,
req: req,
feeFunction: feeFunc,
}
// Call the method under test.
tp.storeRecord(initialCounter, tx, req, feeFunc, fee, utxoIndex)
tp.updateRecord(record, sweepCtx)
// Read the saved record and compare.
record, ok := tp.records.Load(initialCounter)
@ -657,14 +671,31 @@ func TestCreateRBFCompliantTx(t *testing.T) {
tc := tc
rid := requestCounter.Add(1)
// Create a test record.
record := &monitorRecord{
requestID: rid,
req: req,
feeFunction: m.feeFunc,
}
t.Run(tc.name, func(t *testing.T) {
tc.setupMock()
// Call the method under test.
err := tp.createRBFCompliantTx(rid, req, m.feeFunc)
rec, err := tp.createRBFCompliantTx(record)
// Check the result is as expected.
require.ErrorIs(t, err, tc.expectedErr)
if tc.expectedErr != nil {
return
}
// Assert the returned record has the following fields
// populated.
require.NotEmpty(t, rec.tx)
require.NotEmpty(t, rec.fee)
})
}
}
@ -698,13 +729,21 @@ func TestTxPublisherBroadcast(t *testing.T) {
// Create a testing record and put it in the map.
fee := btcutil.Amount(1000)
requestID := uint64(1)
tp.storeRecord(requestID, tx, req, m.feeFunc, fee, utxoIndex)
// Quickly check when the requestID cannot be found, an error is
// returned.
result, err := tp.broadcast(uint64(1000))
require.Error(t, err)
require.Nil(t, result)
// Create a sweepTxCtx.
sweepCtx := &sweepTxCtx{
tx: tx,
fee: fee,
outpointToTxIndex: utxoIndex,
}
// Create a test record.
record := &monitorRecord{
requestID: requestID,
req: req,
feeFunction: m.feeFunc,
}
rec := tp.updateRecord(record, sweepCtx)
testCases := []struct {
name string
@ -759,7 +798,7 @@ func TestTxPublisherBroadcast(t *testing.T) {
tc.setupMock()
// Call the method under test.
result, err := tp.broadcast(requestID)
result, err := tp.broadcast(rec)
// Check the result is as expected.
require.ErrorIs(t, err, tc.expectedErr)
@ -796,6 +835,13 @@ func TestRemoveResult(t *testing.T) {
// Create a test request ID counter.
requestCounter := atomic.Uint64{}
// Create a sweepTxCtx.
sweepCtx := &sweepTxCtx{
tx: tx,
fee: fee,
outpointToTxIndex: utxoIndex,
}
testCases := []struct {
name string
setupRecord func() uint64
@ -808,9 +854,15 @@ func TestRemoveResult(t *testing.T) {
name: "remove on TxConfirmed",
setupRecord: func() uint64 {
rid := requestCounter.Add(1)
tp.storeRecord(
rid, tx, req, m.feeFunc, fee, utxoIndex,
)
// Create a test record.
record := &monitorRecord{
requestID: rid,
req: req,
feeFunction: m.feeFunc,
}
tp.updateRecord(record, sweepCtx)
tp.subscriberChans.Store(rid, nil)
return rid
@ -826,9 +878,15 @@ func TestRemoveResult(t *testing.T) {
name: "remove on TxFailed",
setupRecord: func() uint64 {
rid := requestCounter.Add(1)
tp.storeRecord(
rid, tx, req, m.feeFunc, fee, utxoIndex,
)
// Create a test record.
record := &monitorRecord{
requestID: rid,
req: req,
feeFunction: m.feeFunc,
}
tp.updateRecord(record, sweepCtx)
tp.subscriberChans.Store(rid, nil)
return rid
@ -845,9 +903,15 @@ func TestRemoveResult(t *testing.T) {
name: "noop when tx is not confirmed or failed",
setupRecord: func() uint64 {
rid := requestCounter.Add(1)
tp.storeRecord(
rid, tx, req, m.feeFunc, fee, utxoIndex,
)
// Create a test record.
record := &monitorRecord{
requestID: rid,
req: req,
feeFunction: m.feeFunc,
}
tp.updateRecord(record, sweepCtx)
tp.subscriberChans.Store(rid, nil)
return rid
@ -906,7 +970,21 @@ func TestNotifyResult(t *testing.T) {
// Create a testing record and put it in the map.
fee := btcutil.Amount(1000)
requestID := uint64(1)
tp.storeRecord(requestID, tx, req, m.feeFunc, fee, utxoIndex)
// Create a sweepTxCtx.
sweepCtx := &sweepTxCtx{
tx: tx,
fee: fee,
outpointToTxIndex: utxoIndex,
}
// Create a test record.
record := &monitorRecord{
requestID: requestID,
req: req,
feeFunction: m.feeFunc,
}
tp.updateRecord(record, sweepCtx)
// Create a subscription to the event.
subscriber := make(chan *BumpResult, 1)
@ -1058,6 +1136,7 @@ func TestCreateAnPublishFail(t *testing.T) {
// Overwrite the budget to make it smaller than the fee.
req.Budget = 100
record := &monitorRecord{
requestID: requestID,
req: req,
feeFunction: m.feeFunc,
tx: &wire.MsgTx{},
@ -1073,7 +1152,7 @@ func TestCreateAnPublishFail(t *testing.T) {
mock.Anything).Return(script, nil)
// Call the createAndPublish method.
resultOpt := tp.createAndPublishTx(requestID, record)
resultOpt := tp.createAndPublishTx(record)
result := resultOpt.UnwrapOrFail(t)
// We expect the result to be TxFailed and the error is set in the
@ -1092,7 +1171,7 @@ func TestCreateAnPublishFail(t *testing.T) {
mock.Anything).Return(lnwallet.ErrMempoolFee).Once()
// Call the createAndPublish method and expect a none option.
resultOpt = tp.createAndPublishTx(requestID, record)
resultOpt = tp.createAndPublishTx(record)
require.True(t, resultOpt.IsNone())
// Mock the testmempoolaccept to return a fee related error that should
@ -1101,7 +1180,7 @@ func TestCreateAnPublishFail(t *testing.T) {
mock.Anything).Return(chain.ErrInsufficientFee).Once()
// Call the createAndPublish method and expect a none option.
resultOpt = tp.createAndPublishTx(requestID, record)
resultOpt = tp.createAndPublishTx(record)
require.True(t, resultOpt.IsNone())
}
@ -1123,6 +1202,7 @@ func TestCreateAnPublishSuccess(t *testing.T) {
// Create a testing monitor record.
req := createTestBumpRequest()
record := &monitorRecord{
requestID: requestID,
req: req,
feeFunction: m.feeFunc,
tx: &wire.MsgTx{},
@ -1145,7 +1225,7 @@ func TestCreateAnPublishSuccess(t *testing.T) {
mock.Anything, mock.Anything).Return(errDummy).Once()
// Call the createAndPublish method and expect a failure result.
resultOpt := tp.createAndPublishTx(requestID, record)
resultOpt := tp.createAndPublishTx(record)
result := resultOpt.UnwrapOrFail(t)
// We expect the result to be TxFailed and the error is set.
@ -1166,7 +1246,7 @@ func TestCreateAnPublishSuccess(t *testing.T) {
mock.Anything, mock.Anything).Return(nil).Once()
// Call the createAndPublish method and expect a success result.
resultOpt = tp.createAndPublishTx(requestID, record)
resultOpt = tp.createAndPublishTx(record)
result = resultOpt.UnwrapOrFail(t)
require.True(t, resultOpt.IsSome())
@ -1208,7 +1288,22 @@ func TestHandleTxConfirmed(t *testing.T) {
// Create a testing record and put it in the map.
fee := btcutil.Amount(1000)
requestID := uint64(1)
tp.storeRecord(requestID, tx, req, m.feeFunc, fee, utxoIndex)
// Create a sweepTxCtx.
sweepCtx := &sweepTxCtx{
tx: tx,
fee: fee,
outpointToTxIndex: utxoIndex,
}
// Create a test record.
record := &monitorRecord{
requestID: requestID,
req: req,
feeFunction: m.feeFunc,
}
tp.updateRecord(record, sweepCtx)
record, ok := tp.records.Load(requestID)
require.True(t, ok)
@ -1226,7 +1321,7 @@ func TestHandleTxConfirmed(t *testing.T) {
tp.wg.Add(1)
done := make(chan struct{})
go func() {
tp.handleTxConfirmed(record, requestID)
tp.handleTxConfirmed(record)
close(done)
}()
@ -1272,7 +1367,11 @@ func TestHandleFeeBumpTx(t *testing.T) {
// Create a testing monitor record.
req := createTestBumpRequest()
// Create a testing record and put it in the map.
requestID := uint64(1)
record := &monitorRecord{
requestID: requestID,
req: req,
feeFunction: m.feeFunc,
tx: tx,
@ -1285,11 +1384,16 @@ func TestHandleFeeBumpTx(t *testing.T) {
utxoIndex := map[wire.OutPoint]int{
op: 0,
}
// Create a testing record and put it in the map.
fee := btcutil.Amount(1000)
requestID := uint64(1)
tp.storeRecord(requestID, tx, req, m.feeFunc, fee, utxoIndex)
// Create a sweepTxCtx.
sweepCtx := &sweepTxCtx{
tx: tx,
fee: fee,
outpointToTxIndex: utxoIndex,
}
tp.updateRecord(record, sweepCtx)
// Create a subscription to the event.
subscriber := make(chan *BumpResult, 1)
@ -1305,7 +1409,7 @@ func TestHandleFeeBumpTx(t *testing.T) {
// Call the method and expect no result received.
tp.wg.Add(1)
go tp.handleFeeBumpTx(requestID, record, testHeight)
go tp.handleFeeBumpTx(record, testHeight)
// Check there's no result sent back.
select {
@ -1319,7 +1423,7 @@ func TestHandleFeeBumpTx(t *testing.T) {
// Call the method and expect no result received.
tp.wg.Add(1)
go tp.handleFeeBumpTx(requestID, record, testHeight)
go tp.handleFeeBumpTx(record, testHeight)
// Check there's no result sent back.
select {
@ -1351,7 +1455,7 @@ func TestHandleFeeBumpTx(t *testing.T) {
//
// NOTE: must be called in a goroutine in case it blocks.
tp.wg.Add(1)
go tp.handleFeeBumpTx(requestID, record, testHeight)
go tp.handleFeeBumpTx(record, testHeight)
select {
case <-time.After(time.Second):
@ -1397,6 +1501,7 @@ func TestProcessRecords(t *testing.T) {
// Create a monitor record that's confirmed.
recordConfirmed := &monitorRecord{
requestID: requestID1,
req: req1,
feeFunction: m.feeFunc,
tx: tx1,
@ -1410,6 +1515,7 @@ func TestProcessRecords(t *testing.T) {
// Create a monitor record that's not confirmed. We know it's not
// confirmed because the num of confirms is zero.
recordFeeBump := &monitorRecord{
requestID: requestID2,
req: req2,
feeFunction: m.feeFunc,
tx: tx2,
@ -1548,7 +1654,7 @@ func TestHandleInitialBroadcastSuccess(t *testing.T) {
// Call the method under test.
tp.wg.Add(1)
tp.handleInitialBroadcast(rec, rid)
tp.handleInitialBroadcast(rec)
// Check the result is sent back.
select {
@ -1619,7 +1725,7 @@ func TestHandleInitialBroadcastFail(t *testing.T) {
// Call the method under test and expect an error returned.
tp.wg.Add(1)
tp.handleInitialBroadcast(rec, rid)
tp.handleInitialBroadcast(rec)
// Check the result is sent back.
select {
@ -1652,7 +1758,7 @@ func TestHandleInitialBroadcastFail(t *testing.T) {
// Call the method under test.
tp.wg.Add(1)
tp.handleInitialBroadcast(rec, rid)
tp.handleInitialBroadcast(rec)
// Check the result is sent back.
select {

33
sweep/sweeper.go
View file

@ -119,9 +119,12 @@ const (
// sweeping transactions confirmed, the remaining two will be excluded.
Excluded
// Failed is the state when a pending input has too many failed publish
// atttempts or unknown broadcast error is returned.
Failed
// Fatal is the final state of a pending input. Inputs ending in this
// state won't be retried. This could happen,
// - when a pending input has too many failed publish attempts;
// - the input has been spent by another party;
// - unknown broadcast error is returned.
Fatal
)
// String gives a human readable text for the sweep states.
@ -145,8 +148,8 @@ func (s SweepState) String() string {
case Excluded:
return "Excluded"
case Failed:
return "Failed"
case Fatal:
return "Fatal"
default:
return "Unknown"
@ -215,7 +218,7 @@ func (p *SweeperInput) terminated() bool {
// If the input has reached a final state, that it's either
// been swept, or failed, or excluded, we will remove it from
// our sweeper.
case Failed, Swept, Excluded:
case Fatal, Swept, Excluded:
return true
default:
@ -1264,7 +1267,7 @@ func (s *UtxoSweeper) handleNewInput(input *sweepInputMessage) error {
)
if err != nil {
err := fmt.Errorf("wait for spend: %w", err)
s.markInputFailed(pi, err)
s.markInputFatal(pi, err)
return err
}
@ -1477,12 +1480,12 @@ func (s *UtxoSweeper) markInputsSwept(tx *wire.MsgTx, isOurTx bool) {
}
}
// markInputFailed marks the given input as failed and won't be retried. It
// markInputFatal marks the given input as fatal and won't be retried. It
// will also notify all the subscribers of this input.
func (s *UtxoSweeper) markInputFailed(pi *SweeperInput, err error) {
func (s *UtxoSweeper) markInputFatal(pi *SweeperInput, err error) {
log.Errorf("Failed to sweep input: %v, error: %v", pi, err)
pi.state = Failed
pi.state = Fatal
s.signalResult(pi, Result{Err: err})
}
@ -1784,15 +1787,15 @@ func (s *UtxoSweeper) handleBumpEventTxFatal(resp *bumpResp) error {
}
}
// Mark the inputs as failed.
s.markInputsFailed(resp.set, r.Err)
// Mark the inputs as fatal.
s.markInputsFatal(resp.set, r.Err)
return nil
}
// markInputsFailed marks all inputs found in the tx as failed. It will also
// markInputsFatal marks all inputs in the input set as failed. It will also
// notify all the subscribers of these inputs.
func (s *UtxoSweeper) markInputsFailed(set InputSet, err error) {
func (s *UtxoSweeper) markInputsFatal(set InputSet, err error) {
for _, inp := range set.Inputs() {
outpoint := inp.OutPoint()
@ -1816,7 +1819,7 @@ func (s *UtxoSweeper) markInputsFailed(set InputSet, err error) {
continue
}
s.markInputFailed(input, err)
s.markInputFatal(input, err)
}
}

40
sweep/sweeper_test.go
View file

@ -215,7 +215,7 @@ func TestMarkInputsPublishFailed(t *testing.T) {
// published.
// - inputSwept specifies an input that's swept.
// - inputExcluded specifies an input that's excluded.
// - inputFailed specifies an input that's failed.
// - inputFatal specifies an input that's fatal.
var (
inputInit = createMockInput(t, s, Init)
inputPendingPublish = createMockInput(t, s, PendingPublish)
@ -223,13 +223,13 @@ func TestMarkInputsPublishFailed(t *testing.T) {
inputPublishFailed = createMockInput(t, s, PublishFailed)
inputSwept = createMockInput(t, s, Swept)
inputExcluded = createMockInput(t, s, Excluded)
inputFailed = createMockInput(t, s, Failed)
inputFatal = createMockInput(t, s, Fatal)
)
// Gather all inputs.
set.On("Inputs").Return([]input.Input{
inputInit, inputPendingPublish, inputPublished,
inputPublishFailed, inputSwept, inputExcluded, inputFailed,
inputPublishFailed, inputSwept, inputExcluded, inputFatal,
})
// Mark the test inputs. We expect the non-exist input and the
@ -264,7 +264,7 @@ func TestMarkInputsPublishFailed(t *testing.T) {
require.Equal(Excluded, s.inputs[inputExcluded.OutPoint()].state)
// We expect the failed input to stay unchanged.
require.Equal(Failed, s.inputs[inputFailed.OutPoint()].state)
require.Equal(Fatal, s.inputs[inputFatal.OutPoint()].state)
// Assert mocked statements are executed as expected.
mockStore.AssertExpectations(t)
@ -437,7 +437,7 @@ func TestUpdateSweeperInputs(t *testing.T) {
// These inputs won't hit RequiredLockTime so we won't mock.
input4 := &SweeperInput{state: Swept, Input: inp1}
input5 := &SweeperInput{state: Excluded, Input: inp1}
input6 := &SweeperInput{state: Failed, Input: inp1}
input6 := &SweeperInput{state: Fatal, Input: inp1}
// Mock the input to have a locktime in the future so it will NOT be
// returned.
@ -575,7 +575,7 @@ func TestDecideStateAndRBFInfo(t *testing.T) {
require.Equal(Published, state)
}
// TestMarkInputFailed checks that the input is marked as failed as expected.
// TestMarkInputFatal checks that the input is marked as expected.
func TestMarkInputFailed(t *testing.T) {
t.Parallel()
@ -596,10 +596,10 @@ func TestMarkInputFailed(t *testing.T) {
}
// Call the method under test.
s.markInputFailed(pi, errors.New("dummy error"))
s.markInputFatal(pi, errors.New("dummy error"))
// Assert the state is updated.
require.Equal(t, Failed, pi.state)
require.Equal(t, Fatal, pi.state)
}
// TestSweepPendingInputs checks that `sweepPendingInputs` correctly executes
@ -1102,7 +1102,7 @@ func TestMarkInputsFailed(t *testing.T) {
// published.
// - inputSwept specifies an input that's swept.
// - inputExcluded specifies an input that's excluded.
// - inputFailed specifies an input that's failed.
// - inputFatal specifies an input that's fatal.
var (
inputInit = createMockInput(t, s, Init)
inputPendingPublish = createMockInput(t, s, PendingPublish)
@ -1110,33 +1110,33 @@ func TestMarkInputsFailed(t *testing.T) {
inputPublishFailed = createMockInput(t, s, PublishFailed)
inputSwept = createMockInput(t, s, Swept)
inputExcluded = createMockInput(t, s, Excluded)
inputFailed = createMockInput(t, s, Failed)
inputFatal = createMockInput(t, s, Fatal)
)
// Gather all inputs.
set.On("Inputs").Return([]input.Input{
inputInit, inputPendingPublish, inputPublished,
inputPublishFailed, inputSwept, inputExcluded, inputFailed,
inputPublishFailed, inputSwept, inputExcluded, inputFatal,
})
// Mark the test inputs. We expect the non-exist input and
// inputSwept/inputExcluded/inputFailed to be skipped.
s.markInputsFailed(set, errDummy)
// inputSwept/inputExcluded/inputFatal to be skipped.
s.markInputsFatal(set, errDummy)
// We expect unchanged number of pending inputs.
require.Len(s.inputs, 7)
// We expect the init input's to be marked as failed.
require.Equal(Failed, s.inputs[inputInit.OutPoint()].state)
// We expect the init input's to be marked as fatal.
require.Equal(Fatal, s.inputs[inputInit.OutPoint()].state)
// We expect the pending-publish input to be marked as failed.
require.Equal(Failed, s.inputs[inputPendingPublish.OutPoint()].state)
require.Equal(Fatal, s.inputs[inputPendingPublish.OutPoint()].state)
// We expect the published input to be marked as failed.
require.Equal(Failed, s.inputs[inputPublished.OutPoint()].state)
// We expect the published input to be marked as fatal.
require.Equal(Fatal, s.inputs[inputPublished.OutPoint()].state)
// We expect the publish failed input to be markd as failed.
require.Equal(Failed, s.inputs[inputPublishFailed.OutPoint()].state)
require.Equal(Fatal, s.inputs[inputPublishFailed.OutPoint()].state)
// We expect the swept input to stay unchanged.
require.Equal(Swept, s.inputs[inputSwept.OutPoint()].state)
@ -1145,7 +1145,7 @@ func TestMarkInputsFailed(t *testing.T) {
require.Equal(Excluded, s.inputs[inputExcluded.OutPoint()].state)
// We expect the failed input to stay unchanged.
require.Equal(Failed, s.inputs[inputFailed.OutPoint()].state)
require.Equal(Fatal, s.inputs[inputFatal.OutPoint()].state)
}
// TestHandleBumpEventTxFatal checks that `handleBumpEventTxFatal` correctly