From 81ee3b9fbe5f05ceab7bd703b6cfe09cc2e03caa Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Mon, 3 Apr 2023 21:02:23 +0800 Subject: [PATCH 01/16] chainntnfs: introduce mempool notifier This commit adds a mempool notifier which notifies the subscriber the spending event found in the mempool for a given input. --- chainntnfs/mempool.go | 193 +++++++++++++++++++++++++++++++++++++ chainntnfs/mempool_test.go | 1 + 2 files changed, 194 insertions(+) create mode 100644 chainntnfs/mempool.go create mode 100644 chainntnfs/mempool_test.go diff --git a/chainntnfs/mempool.go b/chainntnfs/mempool.go new file mode 100644 index 000000000..9ef209bc0 --- /dev/null +++ b/chainntnfs/mempool.go @@ -0,0 +1,193 @@ +package chainntnfs + +import ( + "sync" + + "github.com/btcsuite/btcd/btcutil" + "github.com/btcsuite/btcd/wire" + "github.com/lightningnetwork/lnd/lnutils" +) + +// inputsWithTx is a map of outpoints to the tx that spends them. +type inputsWithTx map[wire.OutPoint]*SpendDetail + +// MempoolNotifier defines an internal mempool notifier that's used to notify +// the spending of given inputs. This is mounted to either `BitcoindNotifier` +// or `BtcdNotifier` depending on the chain backend. +type MempoolNotifier struct { + wg sync.WaitGroup + + // subscribedInputs stores the inputs that we want to watch their + // spending event for. + subscribedInputs lnutils.SyncMap[wire.OutPoint, *MempoolSpendEvent] + + // quit is closed when the notifier is torn down. + quit chan struct{} +} + +// MempoolSpendEvent is returned to the subscriber to watch for the spending +// event for the requested input. +type MempoolSpendEvent struct { + // Spend is a receive only channel which will be sent upon once the + // target outpoint has been spent. + // + // NOTE: This channel must be buffered. + Spend <-chan *SpendDetail + + // event is the channel that will be sent upon once the target outpoint + // is spent. + event chan *SpendDetail + + // cancel cancels the subscription. + cancel chan struct{} +} + +// newMempoolSpendEvent returns a new instance of MempoolSpendEvent. +func newMempoolSpendEvent() *MempoolSpendEvent { + sub := &MempoolSpendEvent{ + event: make(chan *SpendDetail, 1), + cancel: make(chan struct{}), + } + + // Mount the receive only channel to the event channel. + sub.Spend = (<-chan *SpendDetail)(sub.event) + + return sub +} + +// Cancel cancels the subscription. +func (m *MempoolSpendEvent) Cancel() { + close(m.cancel) +} + +// NewMempoolNotifier takes a chain connection and returns a new mempool +// notifier. +func NewMempoolNotifier() *MempoolNotifier { + return &MempoolNotifier{ + subscribedInputs: lnutils.SyncMap[ + wire.OutPoint, *MempoolSpendEvent, + ]{}, + quit: make(chan struct{}), + } +} + +// SubscribeInput takes an outpoint of the input and returns a channel that the +// subscriber can listen to for the spending event. +func (m *MempoolNotifier) SubscribeInput( + outpoint wire.OutPoint) *MempoolSpendEvent { + + Log.Debugf("Subscribing mempool event for input %s", outpoint) + sub := newMempoolSpendEvent() + + m.subscribedInputs.Store(outpoint, sub) + + return sub +} + +// Unsubscribe removes the subscription for the given outpoint. +func (m *MempoolNotifier) Unsubscribe(outpoint wire.OutPoint) { + Log.Debugf("Unsubscribing MempoolSpendEvent for input %s", outpoint) + m.subscribedInputs.Delete(outpoint) +} + +// ProcessRelevantSpendTx takes a transaction and checks whether it spends any +// of the subscribed inputs. If so, spend notifications are sent to the +// relevant subscribers. +func (m *MempoolNotifier) ProcessRelevantSpendTx(tx *btcutil.Tx) { + Log.Tracef("Processing mempool tx %s", tx.Hash()) + defer Log.Tracef("Finished processing mempool tx %s", tx.Hash()) + + // Get the spent inputs of interest. + spentInputs := m.findRelevantInputs(tx) + + // Notify the subscribers. + m.notifySpent(spentInputs) +} + +// TearDown stops the notifier and cleans up resources. +func (m *MempoolNotifier) TearDown() { + Log.Infof("Stopping mempool notifier") + close(m.quit) + m.wg.Wait() +} + +// findRelevantInputs takes a transaction to find the subscribed inputs and +// returns them. +func (m *MempoolNotifier) findRelevantInputs(tx *btcutil.Tx) inputsWithTx { + txid := tx.Hash() + watchedInputs := make(inputsWithTx) + + // NOTE: we may have found multiple targeted inputs in the same tx. + for i, input := range tx.MsgTx().TxIn { + op := &input.PreviousOutPoint + + // Check whether this input is subscribed. + _, loaded := m.subscribedInputs.Load(*op) + if !loaded { + continue + } + + // If found, save it to watchedInputs to notify the + // subscriber later. + Log.Infof("Found input %s, spent in %s", op, txid) + + // Construct the spend details. + details := &SpendDetail{ + SpentOutPoint: op, + SpenderTxHash: txid, + SpendingTx: tx.MsgTx().Copy(), + SpenderInputIndex: uint32(i), + SpendingHeight: 0, + } + watchedInputs[*op] = details + } + + return watchedInputs +} + +// notifySpent iterates all the spentInputs and notifies the subscribers about +// the spent details. +func (m *MempoolNotifier) notifySpent(spentInputs inputsWithTx) { + // notify is a helper closure that constructs a spend detail and sends + // it to the subscribers. + // + // NOTE: must be used inside a goroutine. + notify := func(detail *SpendDetail, op wire.OutPoint) { + defer m.wg.Done() + + txid := detail.SpendingTx.TxHash() + Log.Debugf("Notifying the spend of %s in tx %s", op, txid) + + // Load the subscriber. + sub, loaded := m.subscribedInputs.Load(op) + if !loaded { + // TODO(yy): this is possible if the subscriber cancels + // the subscription? + Log.Errorf("Sub not found for %s", op) + return + } + + // Send the spend details to the subscriber. + select { + case sub.event <- detail: + Log.Debugf("Notified mempool spent for input %s", op) + + case <-sub.cancel: + Log.Debugf("Subscription canceled, skipped notifying "+ + "mempool spent for input %s", op) + + // Delete the subscriber. + m.subscribedInputs.Delete(op) + + case <-m.quit: + Log.Debugf("Mempool notifier quit, skipped notifying "+ + "mempool spent for input %s", op) + } + } + + // Iterate the spent inputs to notify the subscribers concurrently. + for op, tx := range spentInputs { + m.wg.Add(1) + go notify(tx, op) + } +} diff --git a/chainntnfs/mempool_test.go b/chainntnfs/mempool_test.go new file mode 100644 index 000000000..54a8429f6 --- /dev/null +++ b/chainntnfs/mempool_test.go @@ -0,0 +1 @@ +package chainntnfs From 5743dd9601f763cef740aa2d33e422f37a8a572b Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Thu, 6 Apr 2023 03:39:08 +0800 Subject: [PATCH 02/16] chainntnfs: allow multiple subscriber to the same input This commit changes the `subscribedInputs` to store a map of subscribers so multiple subscribers are allowed to receive events from the same outpoint. --- chainntnfs/mempool.go | 91 ++++++++++++++++++++++++++++++++----------- 1 file changed, 69 insertions(+), 22 deletions(-) diff --git a/chainntnfs/mempool.go b/chainntnfs/mempool.go index 9ef209bc0..132dce744 100644 --- a/chainntnfs/mempool.go +++ b/chainntnfs/mempool.go @@ -2,6 +2,7 @@ package chainntnfs import ( "sync" + "sync/atomic" "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/wire" @@ -19,7 +20,11 @@ type MempoolNotifier struct { // subscribedInputs stores the inputs that we want to watch their // spending event for. - subscribedInputs lnutils.SyncMap[wire.OutPoint, *MempoolSpendEvent] + subscribedInputs *lnutils.SyncMap[wire.OutPoint, + *lnutils.SyncMap[uint64, *MempoolSpendEvent]] + + // sCounter is used to generate unique subscription IDs. + sCounter atomic.Uint64 // quit is closed when the notifier is torn down. quit chan struct{} @@ -64,9 +69,10 @@ func (m *MempoolSpendEvent) Cancel() { // notifier. func NewMempoolNotifier() *MempoolNotifier { return &MempoolNotifier{ - subscribedInputs: lnutils.SyncMap[ - wire.OutPoint, *MempoolSpendEvent, - ]{}, + subscribedInputs: &lnutils.SyncMap[ + wire.OutPoint, *lnutils.SyncMap[ + uint64, *MempoolSpendEvent, + ]]{}, quit: make(chan struct{}), } } @@ -77,9 +83,20 @@ func (m *MempoolNotifier) SubscribeInput( outpoint wire.OutPoint) *MempoolSpendEvent { Log.Debugf("Subscribing mempool event for input %s", outpoint) + + // Get the current subscribers for this input or create a new one. + clients := &lnutils.SyncMap[uint64, *MempoolSpendEvent]{} + clients, _ = m.subscribedInputs.LoadOrStore(outpoint, clients) + + // Create a new subscription. sub := newMempoolSpendEvent() - m.subscribedInputs.Store(outpoint, sub) + // Add the subscriber with a unique id. + subscriptionID := m.sCounter.Add(1) + clients.Store(subscriptionID, sub) + + // Update the subscribed inputs. + m.subscribedInputs.Store(outpoint, clients) return sub } @@ -148,24 +165,16 @@ func (m *MempoolNotifier) findRelevantInputs(tx *btcutil.Tx) inputsWithTx { // notifySpent iterates all the spentInputs and notifies the subscribers about // the spent details. func (m *MempoolNotifier) notifySpent(spentInputs inputsWithTx) { - // notify is a helper closure that constructs a spend detail and sends - // it to the subscribers. + // notifySingle sends a notification to a single subscriber about the + // spending event. // // NOTE: must be used inside a goroutine. - notify := func(detail *SpendDetail, op wire.OutPoint) { + notifySingle := func(id uint64, sub *MempoolSpendEvent, + op wire.OutPoint, detail *SpendDetail) { + defer m.wg.Done() - txid := detail.SpendingTx.TxHash() - Log.Debugf("Notifying the spend of %s in tx %s", op, txid) - - // Load the subscriber. - sub, loaded := m.subscribedInputs.Load(op) - if !loaded { - // TODO(yy): this is possible if the subscriber cancels - // the subscription? - Log.Errorf("Sub not found for %s", op) - return - } + Log.Debugf("Notifying client %d", id) // Send the spend details to the subscriber. select { @@ -176,8 +185,18 @@ func (m *MempoolNotifier) notifySpent(spentInputs inputsWithTx) { Log.Debugf("Subscription canceled, skipped notifying "+ "mempool spent for input %s", op) - // Delete the subscriber. - m.subscribedInputs.Delete(op) + // Find all the subscribers for this outpoint. + clients, loaded := m.subscribedInputs.Load(op) + if !loaded { + Log.Errorf("Client %d not found", id) + return + } + + // Delete the specific subscriber. + clients.Delete(id) + + // Update the subscribers map. + m.subscribedInputs.Store(op, clients) case <-m.quit: Log.Debugf("Mempool notifier quit, skipped notifying "+ @@ -185,9 +204,37 @@ func (m *MempoolNotifier) notifySpent(spentInputs inputsWithTx) { } } + // notifyAll is a helper closure that constructs a spend detail and + // sends it to all the subscribers of that particular input. + // + // NOTE: must be used inside a goroutine. + notifyAll := func(detail *SpendDetail, op wire.OutPoint) { + defer m.wg.Done() + + txid := detail.SpendingTx.TxHash() + Log.Debugf("Notifying the spend of %s in tx %s", op, txid) + + // Load the subscriber. + subs, loaded := m.subscribedInputs.Load(op) + if !loaded { + Log.Errorf("Sub not found for %s", op) + return + } + + // Iterate all the subscribers for this input and notify them. + subs.ForEach(func(id uint64, sub *MempoolSpendEvent) error { + m.wg.Add(1) + go notifySingle(id, sub, op, detail) + + return nil + }) + } + // Iterate the spent inputs to notify the subscribers concurrently. for op, tx := range spentInputs { + op, tx := op, tx + m.wg.Add(1) - go notify(tx, op) + go notifyAll(tx, op) } } From 3b77e9c51f01edf22b660778dcbde5e5c4a0e28f Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Fri, 7 Apr 2023 03:45:22 +0800 Subject: [PATCH 03/16] chainntnfs: add method `UnsubscribeEvent` to cancel a subscription This commit adds the method `UnsubscribeEvent` to cancel a single subscription. --- chainntnfs/mempool.go | 86 ++++++++++++++++++++++++++----------------- 1 file changed, 53 insertions(+), 33 deletions(-) diff --git a/chainntnfs/mempool.go b/chainntnfs/mempool.go index 132dce744..d613e829a 100644 --- a/chainntnfs/mempool.go +++ b/chainntnfs/mempool.go @@ -39,6 +39,12 @@ type MempoolSpendEvent struct { // NOTE: This channel must be buffered. Spend <-chan *SpendDetail + // id is the unique identifier of this subscription. + id uint64 + + // outpoint is the subscribed outpoint. + outpoint wire.OutPoint + // event is the channel that will be sent upon once the target outpoint // is spent. event chan *SpendDetail @@ -48,10 +54,12 @@ type MempoolSpendEvent struct { } // newMempoolSpendEvent returns a new instance of MempoolSpendEvent. -func newMempoolSpendEvent() *MempoolSpendEvent { +func newMempoolSpendEvent(id uint64, op wire.OutPoint) *MempoolSpendEvent { sub := &MempoolSpendEvent{ - event: make(chan *SpendDetail, 1), - cancel: make(chan struct{}), + id: id, + outpoint: op, + event: make(chan *SpendDetail, 1), + cancel: make(chan struct{}), } // Mount the receive only channel to the event channel. @@ -60,11 +68,6 @@ func newMempoolSpendEvent() *MempoolSpendEvent { return sub } -// Cancel cancels the subscription. -func (m *MempoolSpendEvent) Cancel() { - close(m.cancel) -} - // NewMempoolNotifier takes a chain connection and returns a new mempool // notifier. func NewMempoolNotifier() *MempoolNotifier { @@ -82,31 +85,61 @@ func NewMempoolNotifier() *MempoolNotifier { func (m *MempoolNotifier) SubscribeInput( outpoint wire.OutPoint) *MempoolSpendEvent { - Log.Debugf("Subscribing mempool event for input %s", outpoint) - // Get the current subscribers for this input or create a new one. clients := &lnutils.SyncMap[uint64, *MempoolSpendEvent]{} clients, _ = m.subscribedInputs.LoadOrStore(outpoint, clients) + // Increment the subscription counter and return the new value. + subscriptionID := m.sCounter.Add(1) + // Create a new subscription. - sub := newMempoolSpendEvent() + sub := newMempoolSpendEvent(subscriptionID, outpoint) // Add the subscriber with a unique id. - subscriptionID := m.sCounter.Add(1) clients.Store(subscriptionID, sub) // Update the subscribed inputs. m.subscribedInputs.Store(outpoint, clients) + Log.Debugf("Subscribed(id=%v) mempool event for input=%s", + subscriptionID, outpoint) + return sub } -// Unsubscribe removes the subscription for the given outpoint. -func (m *MempoolNotifier) Unsubscribe(outpoint wire.OutPoint) { +// UnsubscribeInput removes all the subscriptions for the given outpoint. +func (m *MempoolNotifier) UnsubscribeInput(outpoint wire.OutPoint) { Log.Debugf("Unsubscribing MempoolSpendEvent for input %s", outpoint) m.subscribedInputs.Delete(outpoint) } +// UnsubscribeEvent removes a given subscriber for the given MempoolSpendEvent. +func (m *MempoolNotifier) UnsubscribeEvent(sub *MempoolSpendEvent) { + Log.Debugf("Unsubscribing(id=%v) MempoolSpendEvent for input=%s", + sub.id, sub.outpoint) + + // Load all the subscribers for this input. + clients, loaded := m.subscribedInputs.Load(sub.outpoint) + if !loaded { + Log.Debugf("No subscribers for input %s", sub.outpoint) + return + } + + // Load the subscriber. + subscriber, loaded := clients.Load(sub.id) + if !loaded { + Log.Debugf("No subscribers for input %s with id %v", + sub.outpoint, sub.id) + return + } + + // Close the cancel channel in case it's been used in a goroutine. + close(subscriber.cancel) + + // Remove the subscriber. + clients.Delete(sub.id) +} + // ProcessRelevantSpendTx takes a transaction and checks whether it spends any // of the subscribed inputs. If so, spend notifications are sent to the // relevant subscribers. @@ -174,29 +207,15 @@ func (m *MempoolNotifier) notifySpent(spentInputs inputsWithTx) { defer m.wg.Done() - Log.Debugf("Notifying client %d", id) - // Send the spend details to the subscriber. select { case sub.event <- detail: - Log.Debugf("Notified mempool spent for input %s", op) + Log.Debugf("Notified(id=%v) mempool spent for input %s", + sub.id, op) case <-sub.cancel: - Log.Debugf("Subscription canceled, skipped notifying "+ - "mempool spent for input %s", op) - - // Find all the subscribers for this outpoint. - clients, loaded := m.subscribedInputs.Load(op) - if !loaded { - Log.Errorf("Client %d not found", id) - return - } - - // Delete the specific subscriber. - clients.Delete(id) - - // Update the subscribers map. - m.subscribedInputs.Store(op, clients) + Log.Debugf("Subscription(id=%v) canceled, skipped "+ + "notifying spent for input %s", sub.id, op) case <-m.quit: Log.Debugf("Mempool notifier quit, skipped notifying "+ @@ -212,7 +231,8 @@ func (m *MempoolNotifier) notifySpent(spentInputs inputsWithTx) { defer m.wg.Done() txid := detail.SpendingTx.TxHash() - Log.Debugf("Notifying the spend of %s in tx %s", op, txid) + Log.Debugf("Notifying all clients for the spend of %s in tx %s", + op, txid) // Load the subscriber. subs, loaded := m.subscribedInputs.Load(op) From 8b42151edcfe0e96d7f11d0f1b25ca798a9a3ad5 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Fri, 7 Apr 2023 03:46:11 +0800 Subject: [PATCH 04/16] chainntnfs: patch unit tests for mempool notifier --- chainntnfs/mempool_test.go | 343 +++++++++++++++++++++++++++++++++++++ 1 file changed, 343 insertions(+) diff --git a/chainntnfs/mempool_test.go b/chainntnfs/mempool_test.go index 54a8429f6..6802ac427 100644 --- a/chainntnfs/mempool_test.go +++ b/chainntnfs/mempool_test.go @@ -1 +1,344 @@ package chainntnfs + +import ( + "testing" + "time" + + "github.com/btcsuite/btcd/btcutil" + "github.com/btcsuite/btcd/wire" + "github.com/stretchr/testify/require" +) + +const testTimeout = 5 * time.Second + +// TestMempoolSubscribeInput tests that we can successfully subscribe an input. +func TestMempoolSubscribeInput(t *testing.T) { + t.Parallel() + + // Create a new mempool notifier instance. + notifier := NewMempoolNotifier() + + // Create a test input. + input := wire.OutPoint{Hash: [32]byte{1}} + + // Create the expected subscription. + expectedSub := newMempoolSpendEvent(1, input) + + // Subscribe to the input. + sub := notifier.SubscribeInput(input) + + // Verify the subscription is returned. + require.Equal(t, expectedSub.id, sub.id) + require.Equal(t, expectedSub.outpoint, sub.outpoint) + + // Verify that the subscription was added to the notifier. + subs, loaded := notifier.subscribedInputs.Load(input) + require.True(t, loaded) + + // Verify the saved subscription is the same as the expected one. + sub, loaded = subs.Load(sub.id) + require.True(t, loaded) + require.Equal(t, expectedSub.id, sub.id) + require.Equal(t, expectedSub.outpoint, sub.outpoint) +} + +// TestMempoolUnsubscribeInput tests that we can successfully unsubscribe an +// input. +func TestMempoolUnsubscribeInput(t *testing.T) { + t.Parallel() + + // Create a new mempool notifier instance. + notifier := NewMempoolNotifier() + + // Register a spend notification for an outpoint. + input := wire.OutPoint{Hash: [32]byte{1}} + notifier.SubscribeInput(input) + + // Verify that the subscription was added to the notifier. + _, loaded := notifier.subscribedInputs.Load(input) + require.True(t, loaded) + + // Unsubscribe the input. + notifier.UnsubscribeInput(input) + + // Verify that the input is gone. + _, loaded = notifier.subscribedInputs.Load(input) + require.False(t, loaded) +} + +// TestMempoolUnsubscribeEvent tests that when a given input has multiple +// subscribers, removing one of them won't affect the others. +func TestMempoolUnsubscribeEvent(t *testing.T) { + t.Parallel() + + // Create a new mempool notifier instance. + notifier := NewMempoolNotifier() + + // Register a spend notification for an outpoint. + input := wire.OutPoint{Hash: [32]byte{1}} + sub1 := notifier.SubscribeInput(input) + sub2 := notifier.SubscribeInput(input) + + // Verify that the subscription was added to the notifier. + subs, loaded := notifier.subscribedInputs.Load(input) + require.True(t, loaded) + + // sub1 should be found. + _, loaded = subs.Load(sub1.id) + require.True(t, loaded) + + // sub2 should be found. + _, loaded = subs.Load(sub2.id) + require.True(t, loaded) + + // Unsubscribe sub1. + notifier.UnsubscribeEvent(sub1) + + // Verify that the subscription was removed from the notifier. + subs, loaded = notifier.subscribedInputs.Load(input) + require.True(t, loaded) + + // sub1 should be gone. + _, loaded = subs.Load(sub1.id) + require.False(t, loaded) + + // sub2 should still be found. + _, loaded = subs.Load(sub2.id) + require.True(t, loaded) +} + +// TestMempoolFindRelevantInputs tests that the mempool notifier can find the +// spend of subscribed inputs from a given transaction. +func TestMempoolFindRelevantInputs(t *testing.T) { + t.Parallel() + + // Create a new mempool notifier instance. + notifier := NewMempoolNotifier() + + // Create two inputs and subscribe to the second one. + input1 := wire.OutPoint{Hash: [32]byte{1}} + input2 := wire.OutPoint{Hash: [32]byte{2}} + + // Make input2 the subscribed input. + notifier.SubscribeInput(input2) + + // Create a transaction that spends the above two inputs. + msgTx := &wire.MsgTx{ + TxIn: []*wire.TxIn{ + {PreviousOutPoint: input1}, + {PreviousOutPoint: input2}, + }, + TxOut: []*wire.TxOut{}, + } + tx := btcutil.NewTx(msgTx) + + // Create the expected spend detail. + detailExp := &SpendDetail{ + SpentOutPoint: &input2, + SpenderTxHash: tx.Hash(), + SpendingTx: msgTx, + SpenderInputIndex: 1, + } + + // Call the method. + result := notifier.findRelevantInputs(tx) + + // Verify that the result is as expected. + require.Contains(t, result, input2) + + // Verify the returned spend details is as expected. + detail := result[input2] + require.Equal(t, detailExp, detail) +} + +// TestMempoolNotifySpentSameInputs tests that the mempool notifier sends +// notifications to all subscribers of the same input. +func TestMempoolNotifySpentSameInputs(t *testing.T) { + t.Parallel() + + // Create a new mempool notifier instance. + notifier := NewMempoolNotifier() + + // Register a spend notification for an outpoint. + input := wire.OutPoint{Hash: [32]byte{1}} + sub1 := notifier.SubscribeInput(input) + sub2 := notifier.SubscribeInput(input) + + // Create a transaction that spends input. + msgTx := &wire.MsgTx{ + TxIn: []*wire.TxIn{ + {PreviousOutPoint: input}, + }, + } + tx := btcutil.NewTx(msgTx) + + // Notify the subscribers about the spent input. + spendDetail := &SpendDetail{ + SpentOutPoint: &input, + SpenderTxHash: tx.Hash(), + SpendingTx: msgTx, + SpenderInputIndex: 0, + } + notifier.notifySpent(inputsWithTx{input: spendDetail}) + + // Verify that sub1 received the spend notification for input1. + select { + case spend := <-sub1.Spend: + require.Equal(t, tx.Hash(), spend.SpenderTxHash) + + case <-time.After(testTimeout): + require.Fail(t, "timeout for sub1 to receive") + } + + // Verify that sub2 received the spend notification for input1. + select { + case spend := <-sub2.Spend: + require.Equal(t, tx.Hash(), spend.SpenderTxHash) + + case <-time.After(testTimeout): + require.Fail(t, "timeout for sub2 to receive") + } +} + +// TestMempoolNotifySpentDifferentInputs tests that the mempool notifier sends +// notifications to different subscribers of different inputs. +func TestMempoolNotifySpentDifferentInputs(t *testing.T) { + t.Parallel() + + // Create a new mempool notifier instance. + notifier := NewMempoolNotifier() + + // Create two inputs and subscribe to them. + input1 := wire.OutPoint{Hash: [32]byte{1}, Index: 0} + input2 := wire.OutPoint{Hash: [32]byte{2}, Index: 0} + sub1 := notifier.SubscribeInput(input1) + sub2 := notifier.SubscribeInput(input2) + + // Create a transaction that spends input1. + msgTx := &wire.MsgTx{ + TxIn: []*wire.TxIn{ + {PreviousOutPoint: input1}, + }, + } + tx := btcutil.NewTx(msgTx) + + spendDetail1 := &SpendDetail{ + SpentOutPoint: &input1, + SpenderTxHash: tx.Hash(), + SpendingTx: msgTx, + SpenderInputIndex: 0, + } + + // Notify the subscribers about the spent input. + notifier.notifySpent(inputsWithTx{input1: spendDetail1}) + + // Verify that sub1 received the spend notification for input1. + select { + case spend := <-sub1.Spend: + require.Equal(t, tx.Hash(), spend.SpenderTxHash) + + case <-time.After(testTimeout): + require.Fail(t, "timeout for sub1 to receive") + } + + // Verify that sub2 did not receive any spend notifications. + select { + case <-sub2.Spend: + require.Fail(t, "Expected sub2 to not receive") + + // Give it one second to NOT receive a spend notification. + case <-time.After(1 * time.Second): + } + + // Create another transaction that spends input1 and input2. + msgTx2 := &wire.MsgTx{ + TxIn: []*wire.TxIn{ + {PreviousOutPoint: input1}, + {PreviousOutPoint: input2}, + }, + } + tx2 := btcutil.NewTx(msgTx) + + spendDetail2 := &SpendDetail{ + SpentOutPoint: &input2, + SpenderTxHash: tx2.Hash(), + SpendingTx: msgTx2, + SpenderInputIndex: 1, + } + + // Notify the subscribers about the spent inputs. + notifier.notifySpent(inputsWithTx{ + input1: spendDetail1, input2: spendDetail2, + }) + + // Verify that sub1 received the spend notification for input1. + select { + case spend := <-sub1.Spend: + require.Equal(t, tx.Hash(), spend.SpenderTxHash) + + case <-time.After(testTimeout): + require.Fail(t, "timeout for sub1 to receive") + } + + // Verify that sub2 received the spend notification for input2. + select { + case spend := <-sub2.Spend: + require.Equal(t, tx2.Hash(), spend.SpenderTxHash) + + case <-time.After(testTimeout): + require.Fail(t, "timeout for sub2 to receive") + } +} + +// TestMempoolNotifySpentCancel tests that once a subscription is canceled, it +// won't get notified and won't affect other subscriptions. +func TestMempoolNotifySpentCancel(t *testing.T) { + t.Parallel() + + // Create a new mempool notifier instance. + notifier := NewMempoolNotifier() + + // Register a spend notification for an outpoint. + input := wire.OutPoint{Hash: [32]byte{1}} + sub1 := notifier.SubscribeInput(input) + sub2 := notifier.SubscribeInput(input) + + // Create a transaction that spends input. + msgTx := &wire.MsgTx{ + TxIn: []*wire.TxIn{ + {PreviousOutPoint: input}, + }, + } + tx := btcutil.NewTx(msgTx) + + // Cancel the second subscription before notify. + notifier.UnsubscribeEvent(sub2) + + // Notify the subscribers about the spent input. + spendDetail := &SpendDetail{ + SpentOutPoint: &input, + SpenderTxHash: tx.Hash(), + SpendingTx: msgTx, + SpenderInputIndex: 0, + } + notifier.notifySpent(inputsWithTx{input: spendDetail}) + + // Verify that sub1 received the spend notification for input1. + select { + case spend := <-sub1.Spend: + require.Equal(t, tx.Hash(), spend.SpenderTxHash) + + case <-time.After(testTimeout): + require.Fail(t, "timeout for sub1 to receive") + } + + // Verify that sub2 did not receive any spend notifications. + select { + case <-sub2.Spend: + require.Fail(t, "expected sub2 to not receive") + + // Give it one second to NOT receive a spend notification. + case <-time.After(1 * time.Second): + // Expected + } +} From 454e76cbb6bdd3321e9fe140590644204365b1fc Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Mon, 3 Apr 2023 21:06:56 +0800 Subject: [PATCH 05/16] chainntnfs+chainreg: add mempool watcher to BitcoindNotifier This commit adds the mempool watcher to bitcoind notifier to allow the notifier managing the starting and stopping of the watcher. --- chainntnfs/bitcoindnotify/bitcoind.go | 47 +++++++++++++++++++++++---- 1 file changed, 41 insertions(+), 6 deletions(-) diff --git a/chainntnfs/bitcoindnotify/bitcoind.go b/chainntnfs/bitcoindnotify/bitcoind.go index 680d7c7d0..78a9aba57 100644 --- a/chainntnfs/bitcoindnotify/bitcoind.go +++ b/chainntnfs/bitcoindnotify/bitcoind.go @@ -69,6 +69,9 @@ type BitcoindNotifier struct { // which the transaction could have confirmed within the chain. confirmHintCache chainntnfs.ConfirmHintCache + // memNotifier notifies clients of events related to the mempool. + memNotifier *chainntnfs.MempoolNotifier + wg sync.WaitGroup quit chan struct{} } @@ -96,7 +99,8 @@ func New(chainConn *chain.BitcoindConn, chainParams *chaincfg.Params, spendHintCache: spendHintCache, confirmHintCache: confirmHintCache, - blockCache: blockCache, + blockCache: blockCache, + memNotifier: chainntnfs.NewMempoolNotifier(), quit: make(chan struct{}), } @@ -113,6 +117,7 @@ func (b *BitcoindNotifier) Start() error { b.start.Do(func() { startErr = b.startNotifier() }) + return startErr } @@ -142,6 +147,9 @@ func (b *BitcoindNotifier) Stop() error { } b.txNotifier.TearDown() + // Stop the mempool notifier. + b.memNotifier.TearDown() + return nil } @@ -441,15 +449,17 @@ out: b.bestBlock = newBestBlock case chain.RelevantTx: - // We only care about notifying on confirmed - // spends, so if this is a mempool spend, we can - // ignore it and wait for the spend to appear in - // on-chain. + tx := btcutil.NewTx(&item.TxRecord.MsgTx) + + // If this is a mempool spend, we'll ask the + // mempool notifier to hanlde it. if item.Block == nil { + b.memNotifier.ProcessRelevantSpendTx(tx) continue } - tx := btcutil.NewTx(&item.TxRecord.MsgTx) + // Otherwise this is a confirmed spend, and + // we'll ask the tx notifier to handle it. err := b.txNotifier.ProcessRelevantSpendTx( tx, uint32(item.Block.Height), ) @@ -993,3 +1003,28 @@ func (b *BitcoindNotifier) GetBlock(hash *chainhash.Hash) (*wire.MsgBlock, return b.blockCache.GetBlock(hash, b.chainConn.GetBlock) } + +// SubscribeMempoolSpent allows the caller to register a subscription to watch +// for a spend of an outpoint in the mempool.The event will be dispatched once +// the outpoint is spent in the mempool. +// +// NOTE: part of the MempoolWatcher interface. +func (b *BitcoindNotifier) SubscribeMempoolSpent( + outpoint wire.OutPoint) (*chainntnfs.MempoolSpendEvent, error) { + + event := b.memNotifier.SubscribeInput(outpoint) + + ops := []*wire.OutPoint{&outpoint} + + return event, b.chainConn.NotifySpent(ops) +} + +// CancelMempoolSpendEvent allows the caller to cancel a subscription to watch +// for a spend of an outpoint in the mempool. +// +// NOTE: part of the MempoolWatcher interface. +func (b *BitcoindNotifier) CancelMempoolSpendEvent( + sub *chainntnfs.MempoolSpendEvent) { + + b.memNotifier.UnsubscribeEvent(sub) +} From ad5e798392a7e79ef1624cdd79f6ca0288937666 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Mon, 3 Apr 2023 21:09:04 +0800 Subject: [PATCH 06/16] chainntnfs+chainreg: add mempool watcher to BtcdNotifier This commit adds the mempool watcher to btcd notifier to allow the notifier managing the starting and stopping of the watcher. --- chainntnfs/btcdnotify/btcd.go | 46 +++++++++++++++++++++++++++++++---- 1 file changed, 41 insertions(+), 5 deletions(-) diff --git a/chainntnfs/btcdnotify/btcd.go b/chainntnfs/btcdnotify/btcd.go index 5dfb82346..ca4b89823 100644 --- a/chainntnfs/btcdnotify/btcd.go +++ b/chainntnfs/btcdnotify/btcd.go @@ -86,6 +86,9 @@ type BtcdNotifier struct { // which the transaction could have confirmed within the chain. confirmHintCache chainntnfs.ConfirmHintCache + // memNotifier notifies clients of events related to the mempool. + memNotifier *chainntnfs.MempoolNotifier + wg sync.WaitGroup quit chan struct{} } @@ -115,7 +118,8 @@ func New(config *rpcclient.ConnConfig, chainParams *chaincfg.Params, spendHintCache: spendHintCache, confirmHintCache: confirmHintCache, - blockCache: blockCache, + blockCache: blockCache, + memNotifier: chainntnfs.NewMempoolNotifier(), quit: make(chan struct{}), } @@ -146,6 +150,7 @@ func (b *BtcdNotifier) Start() error { b.start.Do(func() { startErr = b.startNotifier() }) + return startErr } @@ -183,6 +188,9 @@ func (b *BtcdNotifier) Stop() error { } b.txNotifier.TearDown() + // Stop the mempool notifier. + b.memNotifier.TearDown() + return nil } @@ -492,16 +500,19 @@ out: case item := <-b.txUpdates.ChanOut(): newSpend := item.(*txUpdate) + tx := newSpend.tx - // We only care about notifying on confirmed spends, so - // if this is a mempool spend, we can ignore it and wait - // for the spend to appear in on-chain. + // If this is a mempool spend, we'll ask the mempool + // notifier to hanlde it. if newSpend.details == nil { + b.memNotifier.ProcessRelevantSpendTx(tx) continue } + // Otherwise this is a confirmed spend, and we'll ask + // the tx notifier to handle it. err := b.txNotifier.ProcessRelevantSpendTx( - newSpend.tx, uint32(newSpend.details.Height), + tx, uint32(newSpend.details.Height), ) if err != nil { chainntnfs.Log.Errorf("Unable to process "+ @@ -1051,3 +1062,28 @@ func (b *BtcdNotifier) GetBlock(hash *chainhash.Hash) (*wire.MsgBlock, return b.blockCache.GetBlock(hash, b.chainConn.GetBlock) } + +// SubscribeMempoolSpent allows the caller to register a subscription to watch +// for a spend of an outpoint in the mempool.The event will be dispatched once +// the outpoint is spent in the mempool. +// +// NOTE: part of the MempoolWatcher interface. +func (b *BtcdNotifier) SubscribeMempoolSpent( + outpoint wire.OutPoint) (*chainntnfs.MempoolSpendEvent, error) { + + event := b.memNotifier.SubscribeInput(outpoint) + + ops := []*wire.OutPoint{&outpoint} + + return event, b.chainConn.NotifySpent(ops) +} + +// CancelMempoolSpendEvent allows the caller to cancel a subscription to watch +// for a spend of an outpoint in the mempool. +// +// NOTE: part of the MempoolWatcher interface. +func (b *BtcdNotifier) CancelMempoolSpendEvent( + sub *chainntnfs.MempoolSpendEvent) { + + b.memNotifier.UnsubscribeEvent(sub) +} From e81d62fc707b13e7720637475380158ee6b7d53d Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Mon, 3 Apr 2023 21:21:07 +0800 Subject: [PATCH 07/16] chainntnfs+chainreg: add interface `MempoolWatcher` This commit adds the interface `MempoolWatcher` and uses it in the chain registry. --- chainntnfs/bitcoindnotify/bitcoind.go | 4 ++++ chainntnfs/btcdnotify/btcd.go | 3 +++ chainntnfs/interface.go | 13 +++++++++++++ chainreg/chainregistry.go | 16 ++++++++++++++-- 4 files changed, 34 insertions(+), 2 deletions(-) diff --git a/chainntnfs/bitcoindnotify/bitcoind.go b/chainntnfs/bitcoindnotify/bitcoind.go index 78a9aba57..e9393ff6d 100644 --- a/chainntnfs/bitcoindnotify/bitcoind.go +++ b/chainntnfs/bitcoindnotify/bitcoind.go @@ -80,6 +80,10 @@ type BitcoindNotifier struct { // time. var _ chainntnfs.ChainNotifier = (*BitcoindNotifier)(nil) +// Ensure BitcoindNotifier implements the MempoolWatcher interface at compile +// time. +var _ chainntnfs.MempoolWatcher = (*BitcoindNotifier)(nil) + // New returns a new BitcoindNotifier instance. This function assumes the // bitcoind node detailed in the passed configuration is already running, and // willing to accept RPC requests and new zmq clients. diff --git a/chainntnfs/btcdnotify/btcd.go b/chainntnfs/btcdnotify/btcd.go index ca4b89823..32f776fe8 100644 --- a/chainntnfs/btcdnotify/btcd.go +++ b/chainntnfs/btcdnotify/btcd.go @@ -96,6 +96,9 @@ type BtcdNotifier struct { // Ensure BtcdNotifier implements the ChainNotifier interface at compile time. var _ chainntnfs.ChainNotifier = (*BtcdNotifier)(nil) +// Ensure BtcdNotifier implements the MempoolWatcher interface at compile time. +var _ chainntnfs.MempoolWatcher = (*BtcdNotifier)(nil) + // New returns a new BtcdNotifier instance. This function assumes the btcd node // detailed in the passed configuration is already running, and willing to // accept new websockets clients. diff --git a/chainntnfs/interface.go b/chainntnfs/interface.go index 7d160de13..3362a9d2d 100644 --- a/chainntnfs/interface.go +++ b/chainntnfs/interface.go @@ -808,3 +808,16 @@ type ConfirmHintCache interface { // the cache. PurgeConfirmHint(confRequests ...ConfRequest) error } + +// MempoolWatcher defines an interface that allows the caller to query +// information in the mempool. +type MempoolWatcher interface { + // SubscribeMempoolSpent allows the caller to register a subscription + // to watch for a spend of an outpoint in the mempool.The event will be + // dispatched once the outpoint is spent in the mempool. + SubscribeMempoolSpent(op wire.OutPoint) (*MempoolSpendEvent, error) + + // CancelMempoolSpendEvent allows the caller to cancel a subscription to + // watch for a spend of an outpoint in the mempool. + CancelMempoolSpendEvent(sub *MempoolSpendEvent) +} diff --git a/chainreg/chainregistry.go b/chainreg/chainregistry.go index 667c20d38..c4a5caecd 100644 --- a/chainreg/chainregistry.go +++ b/chainreg/chainregistry.go @@ -182,6 +182,10 @@ type PartialChainControl struct { // interested in. ChainNotifier chainntnfs.ChainNotifier + // MempoolNotifier is used to watch for spending events happened in + // mempool. + MempoolNotifier chainntnfs.MempoolWatcher + // ChainView is used in the router for maintaining an up-to-date graph. ChainView chainview.FilteredChainView @@ -433,10 +437,14 @@ func NewPartialChainControl(cfg *Config) (*PartialChainControl, func(), error) { "bitcoind: %v", err) } - cc.ChainNotifier = bitcoindnotify.New( + chainNotifier := bitcoindnotify.New( bitcoindConn, cfg.ActiveNetParams.Params, hintCache, hintCache, cfg.BlockCache, ) + + cc.ChainNotifier = chainNotifier + cc.MempoolNotifier = chainNotifier + cc.ChainView = chainview.NewBitcoindFilteredChainView( bitcoindConn, cfg.BlockCache, ) @@ -655,7 +663,8 @@ func NewPartialChainControl(cfg *Config) (*PartialChainControl, func(), error) { DisableConnectOnNew: true, DisableAutoReconnect: false, } - cc.ChainNotifier, err = btcdnotify.New( + + chainNotifier, err := btcdnotify.New( rpcConfig, cfg.ActiveNetParams.Params, hintCache, hintCache, cfg.BlockCache, ) @@ -663,6 +672,9 @@ func NewPartialChainControl(cfg *Config) (*PartialChainControl, func(), error) { return nil, nil, err } + cc.ChainNotifier = chainNotifier + cc.MempoolNotifier = chainNotifier + // Finally, we'll create an instance of the default chain view // to be used within the routing layer. cc.ChainView, err = chainview.NewBtcdFilteredChainView( From 7ef8072724386f889a3fbb32760353115e9e8421 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Wed, 5 Apr 2023 08:40:18 +0800 Subject: [PATCH 08/16] contractcourt: refactor spendHtlcOutput for clarity Also fixes the docs and rename `isSuccessSpend` to `isPreimageSpend`. --- contractcourt/htlc_timeout_resolver.go | 140 +++++++++++++++---------- 1 file changed, 87 insertions(+), 53 deletions(-) diff --git a/contractcourt/htlc_timeout_resolver.go b/contractcourt/htlc_timeout_resolver.go index 061cf0483..7848494f5 100644 --- a/contractcourt/htlc_timeout_resolver.go +++ b/contractcourt/htlc_timeout_resolver.go @@ -192,13 +192,10 @@ func (h *htlcTimeoutResolver) claimCleanUp( // chainDetailsToWatch returns the output and script which we use to watch for // spends from the direct HTLC output on the commitment transaction. -// -// TODO(joostjager): output already set properly in -// lnwallet.newOutgoingHtlcResolution? And script too? func (h *htlcTimeoutResolver) chainDetailsToWatch() (*wire.OutPoint, []byte, error) { - // If there's no timeout transaction, then the claim output is the - // output directly on the commitment transaction, so we'll just use - // that. + // If there's no timeout transaction, it means we are spending from a + // remote commit, then the claim output is the output directly on the + // commitment transaction, so we'll just use that. if h.htlcResolution.SignedTimeoutTx == nil { outPointToWatch := h.htlcResolution.ClaimOutpoint scriptToWatch := h.htlcResolution.SweepSignDesc.Output.PkScript @@ -206,11 +203,11 @@ func (h *htlcTimeoutResolver) chainDetailsToWatch() (*wire.OutPoint, []byte, err return &outPointToWatch, scriptToWatch, nil } - // If this is the remote party's commitment, then we'll need to grab - // watch the output that our timeout transaction points to. We can - // directly grab the outpoint, then also extract the witness script - // (the last element of the witness stack) to re-construct the pkScript - // we need to watch. + // If SignedTimeoutTx is not nil, this is the local party's commitment, + // and we'll need to grab watch the output that our timeout transaction + // points to. We can directly grab the outpoint, then also extract the + // witness script (the last element of the witness stack) to + // re-construct the pkScript we need to watch. outPointToWatch := h.htlcResolution.SignedTimeoutTx.TxIn[0].PreviousOutPoint witness := h.htlcResolution.SignedTimeoutTx.TxIn[0].Witness scriptToWatch, err := input.WitnessScriptHash(witness[len(witness)-1]) @@ -221,9 +218,9 @@ func (h *htlcTimeoutResolver) chainDetailsToWatch() (*wire.OutPoint, []byte, err return &outPointToWatch, scriptToWatch, nil } -// isSuccessSpend returns true if the passed spend on the specified commitment +// isPreimageSpend returns true if the passed spend on the specified commitment // is a success spend that reveals the pre-image or not. -func isSuccessSpend(spend *chainntnfs.SpendDetail, localCommit bool) bool { +func isPreimageSpend(spend *chainntnfs.SpendDetail, localCommit bool) bool { // Based on the spending input index and transaction, obtain the // witness that tells us what type of spend this is. spenderIndex := spend.SpenderInputIndex @@ -281,7 +278,10 @@ func (h *htlcTimeoutResolver) Resolve() (ContractResolver, error) { // If the spend reveals the pre-image, then we'll enter the clean up // workflow to pass the pre-image back to the incoming link, add it to // the witness cache, and exit. - if isSuccessSpend(commitSpend, h.htlcResolution.SignedTimeoutTx != nil) { + if isPreimageSpend( + commitSpend, h.htlcResolution.SignedTimeoutTx != nil, + ) { + log.Infof("%T(%v): HTLC has been swept with pre-image by "+ "remote party during timeout flow! Adding pre-image to "+ "witness cache", h.htlcResolution.ClaimOutpoint) @@ -310,6 +310,49 @@ func (h *htlcTimeoutResolver) Resolve() (ContractResolver, error) { return h.handleCommitSpend(commitSpend) } +// sweepSecondLevelTx sends a second level timeout transaction to the sweeper. +// This transaction uses the SINLGE|ANYONECANPAY flag. +func (h *htlcTimeoutResolver) sweepSecondLevelTx() error { + log.Infof("%T(%x): offering second-layer timeout tx to sweeper: %v", + h, h.htlc.RHash[:], + spew.Sdump(h.htlcResolution.SignedTimeoutTx)) + + inp := input.MakeHtlcSecondLevelTimeoutAnchorInput( + h.htlcResolution.SignedTimeoutTx, + h.htlcResolution.SignDetails, + h.broadcastHeight, + ) + _, err := h.Sweeper.SweepInput( + &inp, sweep.Params{ + Fee: sweep.FeePreference{ + ConfTarget: secondLevelConfTarget, + }, + }, + ) + + // TODO(yy): checkpoint here? + return err +} + +// sendSecondLevelTxLegacy sends a second level timeout transaction to the utxo +// nursery. This transaction uses the legacy SIGHASH_ALL flag. +func (h *htlcTimeoutResolver) sendSecondLevelTxLegacy() error { + log.Debugf("%T(%v): incubating htlc output", h, + h.htlcResolution.ClaimOutpoint) + + err := h.IncubateOutputs( + h.ChanPoint, &h.htlcResolution, nil, + h.broadcastHeight, + ) + if err != nil { + return err + } + + h.outputIncubating = true + + return h.Checkpoint(h) +} + // spendHtlcOutput handles the initial spend of an HTLC output via the timeout // clause. If this is our local commitment, the second-level timeout TX will be // used to spend the output into the next stage. If this is the remote @@ -322,45 +365,16 @@ func (h *htlcTimeoutResolver) spendHtlcOutput() (*chainntnfs.SpendDetail, error) // (the case for anchor type channels). In this case we can re-sign it // and attach fees at will. We let the sweeper handle this job. case h.htlcResolution.SignDetails != nil && !h.outputIncubating: - log.Infof("%T(%x): offering second-layer timeout tx to "+ - "sweeper: %v", h, h.htlc.RHash[:], - spew.Sdump(h.htlcResolution.SignedTimeoutTx)) - - inp := input.MakeHtlcSecondLevelTimeoutAnchorInput( - h.htlcResolution.SignedTimeoutTx, - h.htlcResolution.SignDetails, - h.broadcastHeight, - ) - _, err := h.Sweeper.SweepInput( - &inp, - sweep.Params{ - Fee: sweep.FeePreference{ - ConfTarget: secondLevelConfTarget, - }, - }, - ) - if err != nil { + if err := h.sweepSecondLevelTx(); err != nil { + log.Errorf("Sending timeout tx to sweeper: %v", err) return nil, err } // If we have no SignDetails, and we haven't already sent the output to // the utxo nursery, then we'll do so now. case h.htlcResolution.SignDetails == nil && !h.outputIncubating: - log.Debugf("%T(%v): incubating htlc output", h, - h.htlcResolution.ClaimOutpoint) - - err := h.IncubateOutputs( - h.ChanPoint, &h.htlcResolution, nil, - h.broadcastHeight, - ) - if err != nil { - return nil, err - } - - h.outputIncubating = true - - if err := h.Checkpoint(h); err != nil { - log.Errorf("unable to Checkpoint: %v", err) + if err := h.sendSecondLevelTxLegacy(); err != nil { + log.Errorf("Sending timeout tx to nursery: %v", err) return nil, err } } @@ -369,36 +383,56 @@ func (h *htlcTimeoutResolver) spendHtlcOutput() (*chainntnfs.SpendDetail, error) // watch for a spend of the output, and make our next move off of that. // Depending on if this is our commitment, or the remote party's // commitment, we'll be watching a different outpoint and script. + // + // TODO(yy): outpointToWatch is always h.HtlcOutpoint(), can refactor + // to remove the redundancy. outpointToWatch, scriptToWatch, err := h.chainDetailsToWatch() if err != nil { return nil, err } + return h.waitForConfirmedSpend(outpointToWatch, scriptToWatch) +} + +// waitForConfirmedSpend waits for the HTLC output to be spent and confirmed in +// a block, returns the spend details. +func (h *htlcTimeoutResolver) waitForConfirmedSpend(op *wire.OutPoint, + pkScript []byte) (*chainntnfs.SpendDetail, error) { + log.Infof("%T(%v): waiting for spent of HTLC output %v to be "+ - "fully confirmed", h, h.htlcResolution.ClaimOutpoint, - outpointToWatch) + "fully confirmed", h, h.htlcResolution.ClaimOutpoint, op) // We'll block here until either we exit, or the HTLC output on the // commitment transaction has been spent. spend, err := waitForSpend( - outpointToWatch, scriptToWatch, h.broadcastHeight, - h.Notifier, h.quit, + op, pkScript, h.broadcastHeight, h.Notifier, h.quit, ) if err != nil { return nil, err } + // Once confirmed, persist the state on disk. + if err := h.checkPointSecondLevelTx(); err != nil { + return nil, err + } + + return spend, err +} + +// checkPointSecondLevelTx persists the state of a second level HTLC tx to disk +// if it's published by the sweeper. +func (h *htlcTimeoutResolver) checkPointSecondLevelTx() error { // If this was the second level transaction published by the sweeper, // we can checkpoint the resolver now that it's confirmed. if h.htlcResolution.SignDetails != nil && !h.outputIncubating { h.outputIncubating = true if err := h.Checkpoint(h); err != nil { log.Errorf("unable to Checkpoint: %v", err) - return nil, err + return err } } - return spend, err + return nil } // handleCommitSpend handles the spend of the HTLC output on the commitment From 556c38fefd653f54852fd8b511cd1bb5af4901c2 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Thu, 6 Apr 2023 03:40:26 +0800 Subject: [PATCH 09/16] contractcourt+lnd: watch for preimage reveal in mempool This commit extends the current htlc timeout resolver to also watch for preimage spend in mempool for a full node backend. If mempool enabled, the resolver will watch the spend of the htlc output in mempool and blocks **concurrently**, as if they are independent. Ideally, a transaction will first appear in mempool then in a block. However, there's no guarantee it will appear in **our** mempool since there's no global mempool, thus we need to watch the spend in two places in case it doesn't go through our mempool. The current design favors the spend event found in blocks, that is, when the tx is confirmed, we'd abort the monitoring and conitnue since the outpoint cannot be double spent and re-appear in mempool again. This is not true in the rare case of reorg, and we will handle reorg seperately. --- contractcourt/chain_arbitrator.go | 4 + contractcourt/htlc_timeout_resolver.go | 179 ++++++++++++++++++++++++- server.go | 1 + 3 files changed, 182 insertions(+), 2 deletions(-) diff --git a/contractcourt/chain_arbitrator.go b/contractcourt/chain_arbitrator.go index b2f3f7242..de06078e8 100644 --- a/contractcourt/chain_arbitrator.go +++ b/contractcourt/chain_arbitrator.go @@ -130,6 +130,10 @@ type ChainArbitratorConfig struct { // certain on-chain events. Notifier chainntnfs.ChainNotifier + // Mempool is the a mempool watcher that allows us to watch for events + // happened in mempool. + Mempool chainntnfs.MempoolWatcher + // Signer is a signer backed by the active lnd node. This should be // capable of producing a signature as specified by a valid // SignDescriptor. diff --git a/contractcourt/htlc_timeout_resolver.go b/contractcourt/htlc_timeout_resolver.go index 7848494f5..06941da04 100644 --- a/contractcourt/htlc_timeout_resolver.go +++ b/contractcourt/htlc_timeout_resolver.go @@ -383,7 +383,15 @@ func (h *htlcTimeoutResolver) spendHtlcOutput() (*chainntnfs.SpendDetail, error) // watch for a spend of the output, and make our next move off of that. // Depending on if this is our commitment, or the remote party's // commitment, we'll be watching a different outpoint and script. - // + return h.watchHtlcSpend() +} + +// watchHtlcSpend watches for a spend of the HTLC output. For neutrino backend, +// it will check blocks for the confirmed spend. For btcd and bitcoind, it will +// check both the mempool and the blocks. +func (h *htlcTimeoutResolver) watchHtlcSpend() (*chainntnfs.SpendDetail, + error) { + // TODO(yy): outpointToWatch is always h.HtlcOutpoint(), can refactor // to remove the redundancy. outpointToWatch, scriptToWatch, err := h.chainDetailsToWatch() @@ -391,7 +399,14 @@ func (h *htlcTimeoutResolver) spendHtlcOutput() (*chainntnfs.SpendDetail, error) return nil, err } - return h.waitForConfirmedSpend(outpointToWatch, scriptToWatch) + // If there's no mempool configured, which is the case for SPV node + // such as neutrino, then we will watch for confirmed spend only. + if h.Mempool == nil { + return h.waitForConfirmedSpend(outpointToWatch, scriptToWatch) + } + + // Watch for a spend of the HTLC output in both the mempool and blocks. + return h.waitForMempoolOrBlockSpend(*outpointToWatch, scriptToWatch) } // waitForConfirmedSpend waits for the HTLC output to be spent and confirmed in @@ -730,3 +745,163 @@ func (h *htlcTimeoutResolver) HtlcPoint() wire.OutPoint { // A compile time assertion to ensure htlcTimeoutResolver meets the // ContractResolver interface. var _ htlcContractResolver = (*htlcTimeoutResolver)(nil) + +// spendResult is used to hold the result of a spend event from either a +// mempool spend or a block spend. +type spendResult struct { + // spend contains the details of the spend. + spend *chainntnfs.SpendDetail + + // err is the error that occurred during the spend notification. + err error +} + +// waitForMempoolOrBlockSpend waits for the htlc output to be spent by a +// transaction that's either be found in the mempool or in a block. +func (h *htlcTimeoutResolver) waitForMempoolOrBlockSpend(op wire.OutPoint, + pkScript []byte) (*chainntnfs.SpendDetail, error) { + + log.Infof("%T(%v): waiting for spent of HTLC output %v to be found "+ + "in mempool or block", h, h.htlcResolution.ClaimOutpoint, op) + + // Subscribe for block spent(confirmed). + blockSpent, err := h.Notifier.RegisterSpendNtfn( + &op, pkScript, h.broadcastHeight, + ) + if err != nil { + return nil, fmt.Errorf("register spend: %w", err) + } + + // Subscribe for mempool spent(unconfirmed). + mempoolSpent, err := h.Mempool.SubscribeMempoolSpent(op) + if err != nil { + return nil, fmt.Errorf("register mempool spend: %w", err) + } + + // Create a result chan that will be used to receive the spending + // events. + result := make(chan *spendResult, 2) + + // Create a goroutine that will wait for either a mempool spend or a + // block spend. + // + // NOTE: no need to use waitgroup here as when the resolver exits, the + // goroutine will return on the quit channel. + go h.consumeSpendEvents(result, blockSpent.Spend, mempoolSpent.Spend) + + // Wait for the spend event to be received. + select { + case event := <-result: + // Cancel the mempool subscription as we don't need it anymore. + h.Mempool.CancelMempoolSpendEvent(mempoolSpent) + + return event.spend, event.err + + case <-h.quit: + return nil, errResolverShuttingDown + } +} + +// consumeSpendEvents consumes the spend events from the block and mempool +// subscriptions. It exits when a spend event is received from the block, or +// the resolver itself quits. When a spend event is received from the mempool, +// however, it won't exit but continuing to wait for a spend event from the +// block subscription. +// +// NOTE: there could be a case where we found the preimage in the mempool, +// which will be added to our preimage beacon and settle the incoming link, +// meanwhile the timeout sweep tx confirms. This outgoing HTLC is "free" money +// and is not swept here. +// +// TODO(yy): sweep the outgoing htlc if it's confirmed. +func (h *htlcTimeoutResolver) consumeSpendEvents(resultChan chan *spendResult, + blockSpent, mempoolSpent <-chan *chainntnfs.SpendDetail) { + + op := h.HtlcPoint() + + // Create a result chan to hold the results. + result := &spendResult{} + + // Wait for a spend event to arrive. + for { + select { + // If a spend event is received from the block, this outgoing + // htlc is spent either by the remote via the preimage or by us + // via the timeout. We can exit the loop and `claimCleanUp` + // will feed the preimage to the beacon if found. This treats + // the block as the final judge and the preimage spent won't + // appear in the mempool afterwards. + // + // NOTE: if a reorg happens, the preimage spend can appear in + // the mempool again. Though a rare case, we should handle it + // in a dedicated reorg system. + case spendDetail, ok := <-blockSpent: + if !ok { + result.err = fmt.Errorf("block spent err: %w", + errResolverShuttingDown) + } else { + log.Debugf("Found confirmed spend of HTLC "+ + "output %s in tx=%s", op, + spendDetail.SpenderTxHash) + + result.spend = spendDetail + + // Once confirmed, persist the state on disk. + result.err = h.checkPointSecondLevelTx() + } + + // Send the result and exit the loop. + resultChan <- result + + return + + // If a spend event is received from the mempool, this can be + // either the 2nd stage timeout tx or a preimage spend from the + // remote. We will further check whether the spend reveals the + // preimage and add it to the preimage beacon to settle the + // incoming link. + // + // NOTE: we won't exit the loop here so we can continue to + // watch for the block spend to check point the resolution. + case spendDetail, ok := <-mempoolSpent: + if !ok { + result.err = fmt.Errorf("mempool spent err: %w", + errResolverShuttingDown) + + // This is an internal error so we exit. + resultChan <- result + + return + } + + log.Debugf("Found mempool spend of HTLC output %s "+ + "in tx=%s", op, spendDetail.SpenderTxHash) + + // Check whether the spend reveals the preimage, if not + // continue the loop. + hasPreimage := isPreimageSpend( + spendDetail, + h.htlcResolution.SignedTimeoutTx != nil, + ) + if !hasPreimage { + log.Debugf("HTLC output %s spent doesn't "+ + "reveal preimage", op) + continue + } + + // Found the preimage spend, send the result and + // continue the loop. + result.spend = spendDetail + resultChan <- result + + continue + + // If the resolver exits, we exit the goroutine. + case <-h.quit: + result.err = errResolverShuttingDown + resultChan <- result + + return + } + } +} diff --git a/server.go b/server.go index 7d76353de..b9c8867d1 100644 --- a/server.go +++ b/server.go @@ -1148,6 +1148,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr, }, PreimageDB: s.witnessBeacon, Notifier: cc.ChainNotifier, + Mempool: cc.MempoolNotifier, Signer: cc.Wallet.Cfg.Signer, FeeEstimator: cc.FeeEstimator, ChainIO: cc.ChainIO, From e9c10bf0a19282d750ac2d2b31d3b005b006a704 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Fri, 7 Apr 2023 04:35:27 +0800 Subject: [PATCH 10/16] chainntnfs: remove subscriptions when the relevant tx is confirmed This commit removes the subscribed inputs from mempool notifier when the relevant transaction is confirmed. --- chainntnfs/bitcoindnotify/bitcoind.go | 57 ++++++++++++++++++++------- chainntnfs/btcdnotify/btcd.go | 57 ++++++++++++++++++++------- chainntnfs/mempool.go | 17 ++++++++ chainntnfs/mempool_test.go | 43 ++++++++++++++++++++ 4 files changed, 146 insertions(+), 28 deletions(-) diff --git a/chainntnfs/bitcoindnotify/bitcoind.go b/chainntnfs/bitcoindnotify/bitcoind.go index e9393ff6d..ef29c4503 100644 --- a/chainntnfs/bitcoindnotify/bitcoind.go +++ b/chainntnfs/bitcoindnotify/bitcoind.go @@ -455,23 +455,19 @@ out: case chain.RelevantTx: tx := btcutil.NewTx(&item.TxRecord.MsgTx) - // If this is a mempool spend, we'll ask the - // mempool notifier to hanlde it. + // Init values. + isMempool := false + height := uint32(0) + + // Unwrap values. if item.Block == nil { - b.memNotifier.ProcessRelevantSpendTx(tx) - continue + isMempool = true + } else { + height = uint32(item.Block.Height) } - // Otherwise this is a confirmed spend, and - // we'll ask the tx notifier to handle it. - err := b.txNotifier.ProcessRelevantSpendTx( - tx, uint32(item.Block.Height), - ) - if err != nil { - chainntnfs.Log.Errorf("Unable to "+ - "process transaction %v: %v", - tx.Hash(), err) - } + // Handle the transaction. + b.handleRelevantTx(tx, isMempool, height) } case <-b.quit: @@ -480,6 +476,39 @@ out: } } +// handleRelevantTx handles a new transaction that has been seen either in a +// block or in the mempool. If in mempool, it will ask the mempool notifier to +// handle it. If in a block, it will ask the txNotifier to handle it, and +// cancel any relevant subscriptions made in the mempool. +func (b *BitcoindNotifier) handleRelevantTx(tx *btcutil.Tx, + mempool bool, height uint32) { + + // If this is a mempool spend, we'll ask the mempool notifier to hanlde + // it. + if mempool { + b.memNotifier.ProcessRelevantSpendTx(tx) + return + } + + // Otherwise this is a confirmed spend, and we'll ask the tx notifier + // to handle it. + err := b.txNotifier.ProcessRelevantSpendTx(tx, height) + if err != nil { + chainntnfs.Log.Errorf("Unable to process transaction %v: %v", + tx.Hash(), err) + + return + } + + // Once the tx is processed, we will ask the memNotifier to unsubscribe + // the input. + // + // NOTE(yy): we could build it into txNotifier.ProcessRelevantSpendTx, + // but choose to implement it here so we can easily decouple the two + // notifiers in the future. + b.memNotifier.UnsubsribeConfirmedSpentTx(tx) +} + // historicalConfDetails looks up whether a confirmation request (txid/output // script) has already been included in a block in the active chain and, if so, // returns details about said block. diff --git a/chainntnfs/btcdnotify/btcd.go b/chainntnfs/btcdnotify/btcd.go index 32f776fe8..2c2619b2c 100644 --- a/chainntnfs/btcdnotify/btcd.go +++ b/chainntnfs/btcdnotify/btcd.go @@ -505,23 +505,19 @@ out: newSpend := item.(*txUpdate) tx := newSpend.tx - // If this is a mempool spend, we'll ask the mempool - // notifier to hanlde it. + // Init values. + isMempool := false + height := uint32(0) + + // Unwrap values. if newSpend.details == nil { - b.memNotifier.ProcessRelevantSpendTx(tx) - continue + isMempool = true + } else { + height = uint32(newSpend.details.Height) } - // Otherwise this is a confirmed spend, and we'll ask - // the tx notifier to handle it. - err := b.txNotifier.ProcessRelevantSpendTx( - tx, uint32(newSpend.details.Height), - ) - if err != nil { - chainntnfs.Log.Errorf("Unable to process "+ - "transaction %v: %v", - newSpend.tx.Hash(), err) - } + // Handle the transaction. + b.handleRelevantTx(tx, isMempool, height) case <-b.quit: break out @@ -529,6 +525,39 @@ out: } } +// handleRelevantTx handles a new transaction that has been seen either in a +// block or in the mempool. If in mempool, it will ask the mempool notifier to +// handle it. If in a block, it will ask the txNotifier to handle it, and +// cancel any relevant subscriptions made in the mempool. +func (b *BtcdNotifier) handleRelevantTx(tx *btcutil.Tx, + mempool bool, height uint32) { + + // If this is a mempool spend, we'll ask the mempool notifier to hanlde + // it. + if mempool { + b.memNotifier.ProcessRelevantSpendTx(tx) + return + } + + // Otherwise this is a confirmed spend, and we'll ask the tx notifier + // to handle it. + err := b.txNotifier.ProcessRelevantSpendTx(tx, height) + if err != nil { + chainntnfs.Log.Errorf("Unable to process transaction %v: %v", + tx.Hash(), err) + + return + } + + // Once the tx is processed, we will ask the memNotifier to unsubscribe + // the input. + // + // NOTE(yy): we could build it into txNotifier.ProcessRelevantSpendTx, + // but choose to implement it here so we can easily decouple the two + // notifiers in the future. + b.memNotifier.UnsubsribeConfirmedSpentTx(tx) +} + // historicalConfDetails looks up whether a confirmation request (txid/output // script) has already been included in a block in the active chain and, if so, // returns details about said block. diff --git a/chainntnfs/mempool.go b/chainntnfs/mempool.go index d613e829a..59ce28f7b 100644 --- a/chainntnfs/mempool.go +++ b/chainntnfs/mempool.go @@ -140,6 +140,23 @@ func (m *MempoolNotifier) UnsubscribeEvent(sub *MempoolSpendEvent) { clients.Delete(sub.id) } +// UnsubsribeConfirmedSpentTx takes a transaction and removes the subscriptions +// identified using its inputs. +func (m *MempoolNotifier) UnsubsribeConfirmedSpentTx(tx *btcutil.Tx) { + Log.Tracef("Unsubscribe confirmed tx %s", tx.Hash()) + + // Get the spent inputs of interest. + spentInputs := m.findRelevantInputs(tx) + + // Unsubscribe the subscribers. + for outpoint := range spentInputs { + m.UnsubscribeInput(outpoint) + } + + Log.Tracef("Finished unsubscribing confirmed tx %s, found %d inputs", + tx.Hash(), len(spentInputs)) +} + // ProcessRelevantSpendTx takes a transaction and checks whether it spends any // of the subscribed inputs. If so, spend notifications are sent to the // relevant subscribers. diff --git a/chainntnfs/mempool_test.go b/chainntnfs/mempool_test.go index 6802ac427..c5ee25d81 100644 --- a/chainntnfs/mempool_test.go +++ b/chainntnfs/mempool_test.go @@ -342,3 +342,46 @@ func TestMempoolNotifySpentCancel(t *testing.T) { // Expected } } + +// TestMempoolUnsubscribeConfirmedSpentTx tests that the subscriptions for a +// confirmed tx are removed when calling the method. +func TestMempoolUnsubscribeConfirmedSpentTx(t *testing.T) { + t.Parallel() + + // Create a new mempool notifier instance. + notifier := NewMempoolNotifier() + + // Create two inputs and subscribe to them. + input1 := wire.OutPoint{Hash: [32]byte{1}, Index: 0} + input2 := wire.OutPoint{Hash: [32]byte{2}, Index: 0} + + // sub1 and sub2 are subscribed to the same input. + notifier.SubscribeInput(input1) + notifier.SubscribeInput(input1) + + // sub3 is subscribed to a different input. + sub3 := notifier.SubscribeInput(input2) + + // Create a transaction that spends input1. + msgTx := &wire.MsgTx{ + TxIn: []*wire.TxIn{ + {PreviousOutPoint: input1}, + }, + } + tx := btcutil.NewTx(msgTx) + + // Unsubscribe the relevant transaction. + notifier.UnsubsribeConfirmedSpentTx(tx) + + // Verify that the sub1 and sub2 are removed from the notifier. + _, loaded := notifier.subscribedInputs.Load(input1) + require.False(t, loaded) + + // Verify that the sub3 is not affected. + subs, loaded := notifier.subscribedInputs.Load(input2) + require.True(t, loaded) + + // sub3 should still be found. + _, loaded = subs.Load(sub3.id) + require.True(t, loaded) +} From dccf9c77ca075b59e73b2c36ab00eabaddfc6de8 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Fri, 7 Apr 2023 13:47:38 +0800 Subject: [PATCH 11/16] lnd+contractcourt: add more debug logs This commit adds more debug logs for witness beacon and channel arbitrator. --- contractcourt/channel_arbitrator.go | 12 ++++++++++-- contractcourt/htlc_incoming_contest_resolver.go | 3 +++ witness_beacon.go | 13 ++++++++++++- 3 files changed, 25 insertions(+), 3 deletions(-) diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index 57d6675a3..4df8364fb 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -1589,11 +1589,15 @@ func (c *ChannelArbitrator) checkCommitChainActions(height uint32, ) if toChain { + // Convert to int64 in case of overflow. + remainingBlocks := int64(htlc.RefundTimeout) - + int64(height) + log.Infof("ChannelArbitrator(%v): go to chain for "+ "outgoing htlc %x: timeout=%v, "+ "blocks_until_expiry=%v, broadcast_delta=%v", c.cfg.ChanPoint, htlc.RHash[:], - htlc.RefundTimeout, htlc.RefundTimeout-height, + htlc.RefundTimeout, remainingBlocks, c.cfg.OutgoingBroadcastDelta, ) } @@ -1620,11 +1624,15 @@ func (c *ChannelArbitrator) checkCommitChainActions(height uint32, ) if toChain { + // Convert to int64 in case of overflow. + remainingBlocks := int64(htlc.RefundTimeout) - + int64(height) + log.Infof("ChannelArbitrator(%v): go to chain for "+ "incoming htlc %x: timeout=%v, "+ "blocks_until_expiry=%v, broadcast_delta=%v", c.cfg.ChanPoint, htlc.RHash[:], - htlc.RefundTimeout, htlc.RefundTimeout-height, + htlc.RefundTimeout, remainingBlocks, c.cfg.IncomingBroadcastDelta, ) } diff --git a/contractcourt/htlc_incoming_contest_resolver.go b/contractcourt/htlc_incoming_contest_resolver.go index a8576a8d1..79f29e494 100644 --- a/contractcourt/htlc_incoming_contest_resolver.go +++ b/contractcourt/htlc_incoming_contest_resolver.go @@ -140,6 +140,9 @@ func (h *htlcIncomingContestResolver) Resolve() (ContractResolver, error) { return nil, errResolverShuttingDown } + log.Debugf("%T(%v): Resolving incoming HTLC(expiry=%v, height=%v)", h, + h.htlcResolution.ClaimOutpoint, h.htlcExpiry, currentHeight) + // We'll first check if this HTLC has been timed out, if so, we can // return now and mark ourselves as resolved. If we're past the point of // expiry of the HTLC, then at this point the sender can sweep it, so diff --git a/witness_beacon.go b/witness_beacon.go index dbc4c83f9..4fd44e283 100644 --- a/witness_beacon.go +++ b/witness_beacon.go @@ -1,6 +1,7 @@ package lnd import ( + "errors" "sync" "github.com/lightningnetwork/lnd/channeldb" @@ -127,6 +128,11 @@ func (p *preimageBeacon) LookupPreimage( // Otherwise, we'll perform a final check using the witness cache. preimage, err := p.wCache.LookupSha256Witness(payHash) + if errors.Is(err, channeldb.ErrNoWitnesses) { + ltndLog.Debugf("No witness for payment %v", payHash) + return lntypes.Preimage{}, false + } + if err != nil { ltndLog.Errorf("Unable to lookup witness: %v", err) return lntypes.Preimage{}, false @@ -147,7 +153,9 @@ func (p *preimageBeacon) AddPreimages(preimages ...lntypes.Preimage) error { // the caller when delivering notifications. preimageCopies := make([]lntypes.Preimage, 0, len(preimages)) for _, preimage := range preimages { - srvrLog.Infof("Adding preimage=%v to witness cache", preimage) + srvrLog.Infof("Adding preimage=%v to witness cache for %v", + preimage, preimage.Hash()) + preimageCopies = append(preimageCopies, preimage) } @@ -174,6 +182,9 @@ func (p *preimageBeacon) AddPreimages(preimages ...lntypes.Preimage) error { }(client) } + srvrLog.Debugf("Added %d preimage(s) to witness cache", + len(preimageCopies)) + return nil } From 774db0dfe11122f16002bc55bfc5ebae9c860d68 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Tue, 11 Apr 2023 15:22:05 +0800 Subject: [PATCH 12/16] lntest: add more methods to assert mempool state --- lntest/harness.go | 51 +++++++++++ lntest/harness_assertion.go | 169 +++++++++++++++++++++++++++++++++++- lntest/harness_miner.go | 34 ++++++++ 3 files changed, 252 insertions(+), 2 deletions(-) diff --git a/lntest/harness.go b/lntest/harness.go index 55c113dfd..b57471193 100644 --- a/lntest/harness.go +++ b/lntest/harness.go @@ -1610,6 +1610,57 @@ func (h *HarnessTest) MineBlocksAndAssertNumTxes(num uint32, return blocks } +// cleanMempool mines blocks till the mempool is empty and asserts all active +// nodes have synced to the chain. +func (h *HarnessTest) cleanMempool() { + _, startHeight := h.Miner.GetBestBlock() + + // Mining the blocks slow to give `lnd` more time to sync. + var bestBlock *wire.MsgBlock + err := wait.NoError(func() error { + // If mempool is empty, exit. + mem := h.Miner.GetRawMempool() + if len(mem) == 0 { + _, height := h.Miner.GetBestBlock() + h.Logf("Mined %d blocks when cleanup the mempool", + height-startHeight) + + return nil + } + + // Otherwise mine a block. + blocks := h.Miner.MineBlocksSlow(1) + bestBlock = blocks[len(blocks)-1] + + return fmt.Errorf("still have %d txes in mempool", len(mem)) + }, wait.MinerMempoolTimeout) + require.NoError(h, err, "timeout cleaning up mempool") + + // Exit early if the best block is nil, which means we haven't mined + // any blocks during the cleanup. + if bestBlock == nil { + return + } + + // Make sure all the active nodes are synced. + h.AssertActiveNodesSyncedTo(bestBlock) +} + +// CleanShutDown is used to quickly end a test by shutting down all non-standby +// nodes and mining blocks to empty the mempool. +// +// NOTE: this method provides a faster exit for a test that involves force +// closures as the caller doesn't need to mine all the blocks to make sure the +// mempool is empty. +func (h *HarnessTest) CleanShutDown() { + // First, shutdown all non-standby nodes to prevent new transactions + // being created and fed into the mempool. + h.shutdownNonStandbyNodes() + + // Now mine blocks till the mempool is empty. + h.cleanMempool() +} + // MineEmptyBlocks mines a given number of empty blocks. // // NOTE: this differs from miner's `MineEmptyBlocks` as it requires the nodes diff --git a/lntest/harness_assertion.go b/lntest/harness_assertion.go index 47a1fa38b..927559e98 100644 --- a/lntest/harness_assertion.go +++ b/lntest/harness_assertion.go @@ -887,6 +887,15 @@ func (h *HarnessTest) Random32Bytes() []byte { return randBuf } +// RandomPreimage generates a random preimage which can be used as a payment +// preimage. +func (h *HarnessTest) RandomPreimage() lntypes.Preimage { + var preimage lntypes.Preimage + copy(preimage[:], h.Random32Bytes()) + + return preimage +} + // DecodeAddress decodes a given address and asserts there's no error. func (h *HarnessTest) DecodeAddress(addr string) btcutil.Address { resp, err := btcutil.DecodeAddress(addr, harnessNetParams) @@ -1256,6 +1265,111 @@ func (h *HarnessTest) AssertActiveHtlcs(hn *node.HarnessNode, require.NoError(h, err, "timeout checking active HTLCs") } +// AssertIncomingHTLCActive asserts the node has a pending incoming HTLC in the +// given channel. Returns the HTLC if found and active. +func (h *HarnessTest) AssertIncomingHTLCActive(hn *node.HarnessNode, + cp *lnrpc.ChannelPoint, payHash []byte) *lnrpc.HTLC { + + return h.assertHLTCActive(hn, cp, payHash, true) +} + +// AssertOutgoingHTLCActive asserts the node has a pending outgoing HTLC in the +// given channel. Returns the HTLC if found and active. +func (h *HarnessTest) AssertOutgoingHTLCActive(hn *node.HarnessNode, + cp *lnrpc.ChannelPoint, payHash []byte) *lnrpc.HTLC { + + return h.assertHLTCActive(hn, cp, payHash, false) +} + +// assertHLTCActive asserts the node has a pending HTLC in the given channel. +// Returns the HTLC if found and active. +func (h *HarnessTest) assertHLTCActive(hn *node.HarnessNode, + cp *lnrpc.ChannelPoint, payHash []byte, incoming bool) *lnrpc.HTLC { + + var result *lnrpc.HTLC + target := hex.EncodeToString(payHash) + + err := wait.NoError(func() error { + // We require the RPC call to be succeeded and won't wait for + // it as it's an unexpected behavior. + ch := h.GetChannelByChanPoint(hn, cp) + + // Check all payment hashes active for this channel. + for _, htlc := range ch.PendingHtlcs { + h := hex.EncodeToString(htlc.HashLock) + if h != target { + continue + } + + // If the payment hash is found, check the incoming + // field. + if htlc.Incoming == incoming { + // Found it and return. + result = htlc + return nil + } + + // Otherwise we do have the HTLC but its direction is + // not right. + have, want := "outgoing", "incoming" + if htlc.Incoming { + have, want = "incoming", "outgoing" + } + + return fmt.Errorf("node[%s] have htlc(%v), want: %s, "+ + "have: %s", hn.Name(), payHash, want, have) + } + + return fmt.Errorf("node [%s:%x] didn't have: the payHash %v", + hn.Name(), hn.PubKey[:], payHash) + }, DefaultTimeout) + require.NoError(h, err, "timeout checking pending HTLC") + + return result +} + +// AssertHLTCNotActive asserts the node doesn't have a pending HTLC in the +// given channel, which mean either the HTLC never exists, or it was pending +// and now settled. Returns the HTLC if found and active. +// +// NOTE: to check a pending HTLC becoming settled, first use AssertHLTCActive +// then follow this check. +func (h *HarnessTest) AssertHLTCNotActive(hn *node.HarnessNode, + cp *lnrpc.ChannelPoint, payHash []byte) *lnrpc.HTLC { + + var result *lnrpc.HTLC + target := hex.EncodeToString(payHash) + + err := wait.NoError(func() error { + // We require the RPC call to be succeeded and won't wait for + // it as it's an unexpected behavior. + ch := h.GetChannelByChanPoint(hn, cp) + + // Check all payment hashes active for this channel. + for _, htlc := range ch.PendingHtlcs { + h := hex.EncodeToString(htlc.HashLock) + + // Break if found the htlc. + if h == target { + result = htlc + break + } + } + + // If we've found nothing, we're done. + if result == nil { + return nil + } + + // Otherwise return an error. + return fmt.Errorf("node [%s:%x] still has: the payHash %x", + hn.Name(), hn.PubKey[:], payHash) + }, DefaultTimeout) + require.NoError(h, err, "timeout checking pending HTLC") + + return result +} + // ReceiveSingleInvoice waits until a message is received on the subscribe // single invoice stream or the timeout is reached. func (h *HarnessTest) ReceiveSingleInvoice( @@ -1450,16 +1564,17 @@ func (h *HarnessTest) AssertPaymentStatus(hn *node.HarnessNode, status lnrpc.Payment_PaymentStatus) *lnrpc.Payment { var target *lnrpc.Payment + payHash := preimage.Hash() err := wait.NoError(func() error { - p := h.findPayment(hn, preimage.Hash().String()) + p := h.findPayment(hn, payHash.String()) if status == p.Status { target = p return nil } return fmt.Errorf("payment: %v status not match, want %s "+ - "got %s", preimage, status, p.Status) + "got %s", payHash, status, p.Status) }, DefaultTimeout) require.NoError(h, err, "timeout checking payment status") @@ -2349,3 +2464,53 @@ func (h *HarnessTest) AssertWalletAccountBalance(hn *node.HarnessNode, }, DefaultTimeout) require.NoError(h, err, "timeout checking wallet account balance") } + +// AssertClosingTxInMempool assert that the closing transaction of the given +// channel point can be found in the mempool. If the channel has anchors, it +// will assert the anchor sweep tx is also in the mempool. +func (h *HarnessTest) AssertClosingTxInMempool(cp *lnrpc.ChannelPoint, + c lnrpc.CommitmentType) *wire.MsgTx { + + // Get expected number of txes to be found in the mempool. + expectedTxes := 1 + hasAnchors := CommitTypeHasAnchors(c) + if hasAnchors { + expectedTxes = 2 + } + + // Wait for the expected txes to be found in the mempool. + h.Miner.AssertNumTxsInMempool(expectedTxes) + + // Get the closing tx from the mempool. + op := h.OutPointFromChannelPoint(cp) + closeTx := h.Miner.AssertOutpointInMempool(op) + + return closeTx +} + +// AssertClosingTxInMempool assert that the closing transaction of the given +// channel point can be found in the mempool. If the channel has anchors, it +// will assert the anchor sweep tx is also in the mempool. +func (h *HarnessTest) MineClosingTx(cp *lnrpc.ChannelPoint, + c lnrpc.CommitmentType) *wire.MsgTx { + + // Get expected number of txes to be found in the mempool. + expectedTxes := 1 + hasAnchors := CommitTypeHasAnchors(c) + if hasAnchors { + expectedTxes = 2 + } + + // Wait for the expected txes to be found in the mempool. + h.Miner.AssertNumTxsInMempool(expectedTxes) + + // Get the closing tx from the mempool. + op := h.OutPointFromChannelPoint(cp) + closeTx := h.Miner.AssertOutpointInMempool(op) + + // Mine a block to confirm the closing transaction and potential anchor + // sweep. + h.MineBlocksAndAssertNumTxes(1, expectedTxes) + + return closeTx +} diff --git a/lntest/harness_miner.go b/lntest/harness_miner.go index bdbabe35b..abbf2877d 100644 --- a/lntest/harness_miner.go +++ b/lntest/harness_miner.go @@ -301,6 +301,40 @@ func (h *HarnessMiner) AssertTxInMempool(txid *chainhash.Hash) *wire.MsgTx { return msgTx } +// AssertTxNotInMempool asserts a given transaction cannot be found in the +// mempool. It assumes the mempool is not empty. +// +// NOTE: this should be used after `AssertTxInMempool` to ensure the tx has +// entered the mempool before. Otherwise it might give false positive and the +// tx may enter the mempool after the check. +func (h *HarnessMiner) AssertTxNotInMempool(txid chainhash.Hash) *wire.MsgTx { + var msgTx *wire.MsgTx + + err := wait.NoError(func() error { + // We require the RPC call to be succeeded and won't wait for + // it as it's an unexpected behavior. + mempool := h.GetRawMempool() + + if len(mempool) == 0 { + return fmt.Errorf("empty mempool") + } + + for _, memTx := range mempool { + // Check the values are equal. + if txid.IsEqual(memTx) { + return fmt.Errorf("expect txid %v to be NOT "+ + "found in mempool", txid) + } + } + + return nil + }, wait.MinerMempoolTimeout) + + require.NoError(h, err, "timeout checking tx not in mempool") + + return msgTx +} + // SendOutputsWithoutChange uses the miner to send the given outputs using the // specified fee rate and returns the txid. func (h *HarnessMiner) SendOutputsWithoutChange(outputs []*wire.TxOut, From e715b8ed1fd6d34dcdaae0b4087d6640a5e88ac1 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Fri, 7 Apr 2023 19:10:27 +0800 Subject: [PATCH 13/16] itest+lntest: add itest for mempool preimage watch --- itest/list_on_test.go | 8 + itest/lnd_multi-hop_test.go | 288 ++++++++++++++++++++++++++++++++++++ lntest/node/config.go | 4 + 3 files changed, 300 insertions(+) diff --git a/itest/list_on_test.go b/itest/list_on_test.go index 1814d6292..ade6a80f6 100644 --- a/itest/list_on_test.go +++ b/itest/list_on_test.go @@ -523,4 +523,12 @@ var allTestCases = []*lntest.TestCase{ Name: "channel fundmax", TestFunc: testChannelFundMax, }, + { + Name: "htlc timeout resolver extract preimage remote", + TestFunc: testHtlcTimeoutResolverExtractPreimageRemote, + }, + { + Name: "htlc timeout resolver extract preimage local", + TestFunc: testHtlcTimeoutResolverExtractPreimageLocal, + }, } diff --git a/itest/lnd_multi-hop_test.go b/itest/lnd_multi-hop_test.go index 0f562ed9b..61b5cf9fa 100644 --- a/itest/lnd_multi-hop_test.go +++ b/itest/lnd_multi-hop_test.go @@ -1838,3 +1838,291 @@ func createThreeHopNetwork(ht *lntest.HarnessTest, return aliceChanPoint, bobChanPoint, carol } + +// testHtlcTimeoutResolverExtractPreimageRemote tests that in the multi-hop +// setting, Alice->Bob->Carol, when Bob's outgoing HTLC is swept by Carol using +// the 2nd level success tx2nd level success tx, Bob's timeout resolver will +// extract the preimage from the sweep tx found in mempool or blocks(for +// neutrino). The 2nd level success tx is broadcast by Carol and spends the +// outpoint on her commit tx. +func testHtlcTimeoutResolverExtractPreimageRemote(ht *lntest.HarnessTest) { + runMultiHopHtlcClaimTest(ht, runExtraPreimageFromRemoteCommit) +} + +// runExtraPreimageFromRemoteCommit checks that Bob's htlc timeout resolver +// will extract the preimage from the 2nd level success tx broadcast by Carol +// which spends the htlc output on her commitment tx. +func runExtraPreimageFromRemoteCommit(ht *lntest.HarnessTest, + alice, bob *node.HarnessNode, c lnrpc.CommitmentType, zeroConf bool) { + + // First, we'll create a three hop network: Alice -> Bob -> Carol, with + // Carol refusing to actually settle or directly cancel any HTLC's + // self. + aliceChanPoint, bobChanPoint, carol := createThreeHopNetwork( + ht, alice, bob, false, c, zeroConf, + ) + + // With the network active, we'll now add a new hodl invoice at Carol's + // end. Make sure the cltv expiry delta is large enough, otherwise Bob + // won't send out the outgoing htlc. + preimage := ht.RandomPreimage() + payHash := preimage.Hash() + invoiceReq := &invoicesrpc.AddHoldInvoiceRequest{ + Value: 100_000, + CltvExpiry: finalCltvDelta, + Hash: payHash[:], + } + eveInvoice := carol.RPC.AddHoldInvoice(invoiceReq) + + // Subscribe the invoice. + stream := carol.RPC.SubscribeSingleInvoice(payHash[:]) + + // Now that we've created the invoice, we'll send a single payment from + // Alice to Carol. We won't wait for the response however, as Carol + // will not immediately settle the payment. + req := &routerrpc.SendPaymentRequest{ + PaymentRequest: eveInvoice.PaymentRequest, + TimeoutSeconds: 60, + FeeLimitMsat: noFeeLimitMsat, + } + alice.RPC.SendPayment(req) + + // Once the payment sent, Alice should have one outgoing HTLC active. + ht.AssertOutgoingHTLCActive(alice, aliceChanPoint, payHash[:]) + + // Bob should have two HTLCs active. One incoming HTLC from Alice, and + // one outgoing to Carol. + ht.AssertIncomingHTLCActive(bob, aliceChanPoint, payHash[:]) + htlc := ht.AssertOutgoingHTLCActive(bob, bobChanPoint, payHash[:]) + + // Carol should have one incoming HTLC from Bob. + ht.AssertIncomingHTLCActive(carol, bobChanPoint, payHash[:]) + + // Wait for Carol to mark invoice as accepted. There is a small gap to + // bridge between adding the htlc to the channel and executing the exit + // hop logic. + ht.AssertInvoiceState(stream, lnrpc.Invoice_ACCEPTED) + + // Bob now goes offline so the link between Bob and Carol is broken. + restartBob := ht.SuspendNode(bob) + + // Carol now settles the invoice, since her link with Bob is broken, + // Bob won't know the preimage. + carol.RPC.SettleInvoice(preimage[:]) + + // We'll now mine enough blocks to trigger Carol's broadcast of her + // commitment transaction due to the fact that the HTLC is about to + // timeout. With the default incoming broadcast delta of 10, this + // will be the htlc expiry height minus 10. + numBlocks := padCLTV(uint32( + invoiceReq.CltvExpiry - lncfg.DefaultIncomingBroadcastDelta, + )) + ht.MineBlocks(numBlocks) + + // Carol's force close transaction should now be found in the mempool. + // If there are anchors, we also expect Carol's anchor sweep. We now + // mine a block to confirm Carol's closing transaction. + ht.MineClosingTx(bobChanPoint, c) + + // With the closing transaction confirmed, we should expect Carol's + // HTLC success transaction to be broadcast. + ht.Miner.AssertNumTxsInMempool(1) + + // Restart Bob. Once he finishes syncing the channel state, he should + // notice the force close from Carol. + require.NoError(ht, restartBob()) + + // Get the current height to compute number of blocks to mine to + // trigger the htlc timeout resolver from Bob. + _, height := ht.Miner.GetBestBlock() + + // We'll now mine enough blocks to trigger Bob's timeout resolver. + numBlocks = htlc.ExpirationHeight - uint32(height) - + lncfg.DefaultOutgoingBroadcastDelta + + // Mine empty blocks so Carol's htlc success tx stays in mempool. Once + // the height is reached, Bob's timeout resolver will resolve the htlc + // by extracing the preimage from the mempool. + ht.MineEmptyBlocks(int(numBlocks)) + + // For neutrino backend, the timeout resolver needs to extract the + // preimage from the blocks. + if ht.IsNeutrinoBackend() { + // Mine a block to confirm Carol's 2nd level success tx. + ht.MineBlocksAndAssertNumTxes(1, 1) + } + + // Finally, check that the Alice's payment is marked as succeeded as + // Bob has settled the htlc using the preimage extracted from Carol's + // 2nd level success tx. + ht.AssertPaymentStatus(alice, preimage, lnrpc.Payment_SUCCEEDED) + + // NOTE: for non-standby nodes there's no need to clean up the force + // close as long as the mempool is cleaned. + ht.CleanShutDown() +} + +// testHtlcTimeoutResolverExtractPreimage tests that in the multi-hop setting, +// Alice->Bob->Carol, when Bob's outgoing HTLC is swept by Carol using the +// direct preimage spend, Bob's timeout resolver will extract the preimage from +// the sweep tx found in mempool or blocks(for neutrino). The direct spend tx +// is broadcast by Carol and spends the outpoint on Bob's commit tx. +func testHtlcTimeoutResolverExtractPreimageLocal(ht *lntest.HarnessTest) { + runMultiHopHtlcClaimTest(ht, runExtraPreimageFromLocalCommit) +} + +// runExtraPreimageFromLocalCommit checks that Bob's htlc timeout resolver will +// extract the preimage from the direct spend broadcast by Carol which spends +// the htlc output on Bob's commitment tx. +func runExtraPreimageFromLocalCommit(ht *lntest.HarnessTest, + alice, bob *node.HarnessNode, c lnrpc.CommitmentType, zeroConf bool) { + + // First, we'll create a three hop network: Alice -> Bob -> Carol, with + // Carol refusing to actually settle or directly cancel any HTLC's + // self. + aliceChanPoint, bobChanPoint, carol := createThreeHopNetwork( + ht, alice, bob, false, c, zeroConf, + ) + + // With the network active, we'll now add a new hodl invoice at Carol's + // end. Make sure the cltv expiry delta is large enough, otherwise Bob + // won't send out the outgoing htlc. + preimage := ht.RandomPreimage() + payHash := preimage.Hash() + invoiceReq := &invoicesrpc.AddHoldInvoiceRequest{ + Value: 100_000, + CltvExpiry: finalCltvDelta, + Hash: payHash[:], + } + eveInvoice := carol.RPC.AddHoldInvoice(invoiceReq) + + // Subscribe the invoice. + stream := carol.RPC.SubscribeSingleInvoice(payHash[:]) + + // Now that we've created the invoice, we'll send a single payment from + // Alice to Carol. We won't wait for the response however, as Carol + // will not immediately settle the payment. + req := &routerrpc.SendPaymentRequest{ + PaymentRequest: eveInvoice.PaymentRequest, + TimeoutSeconds: 60, + FeeLimitMsat: noFeeLimitMsat, + } + alice.RPC.SendPayment(req) + + // Once the payment sent, Alice should have one outgoing HTLC active. + ht.AssertOutgoingHTLCActive(alice, aliceChanPoint, payHash[:]) + + // Bob should have two HTLCs active. One incoming HTLC from Alice, and + // one outgoing to Carol. + ht.AssertIncomingHTLCActive(bob, aliceChanPoint, payHash[:]) + htlc := ht.AssertOutgoingHTLCActive(bob, bobChanPoint, payHash[:]) + + // Carol should have one incoming HTLC from Bob. + ht.AssertIncomingHTLCActive(carol, bobChanPoint, payHash[:]) + + // Wait for Carol to mark invoice as accepted. There is a small gap to + // bridge between adding the htlc to the channel and executing the exit + // hop logic. + ht.AssertInvoiceState(stream, lnrpc.Invoice_ACCEPTED) + + // Bob now goes offline so the link between Bob and Carol is broken. + restartBob := ht.SuspendNode(bob) + + // Carol now settles the invoice, since her link with Bob is broken, + // Bob won't know the preimage. + carol.RPC.SettleInvoice(preimage[:]) + + // Stop Carol so it's easier to check the mempool's state since she + // will broadcast the anchor sweeping once Bob force closes. + restartCarol := ht.SuspendNode(carol) + + // Restart Bob to force close the channel. + require.NoError(ht, restartBob()) + + // Bob force closes the channel, which gets his commitment tx into the + // mempool. + ht.CloseChannelAssertPending(bob, bobChanPoint, true) + + // Mine Bob's force close tx. + closeTx := ht.MineClosingTx(bobChanPoint, c) + + // We'll now mine enough blocks to trigger Carol's sweeping of the htlc + // via the direct spend. With the default incoming broadcast delta of + // 10, this will be the htlc expiry height minus 10. + // + // NOTE: we need to mine 1 fewer block as we've already mined one to + // confirm Bob's force close tx. + numBlocks := padCLTV(uint32( + invoiceReq.CltvExpiry - lncfg.DefaultIncomingBroadcastDelta - 1, + )) + + // Mine empty blocks so it's easier to check Bob's sweeping txes below. + ht.MineEmptyBlocks(int(numBlocks)) + + // Increase the fee rate used by the sweeper so Carol's direct spend tx + // won't be replaced by Bob's timeout tx. + ht.SetFeeEstimate(30000) + + // Restart Carol to sweep the htlc output. + require.NoError(ht, restartCarol()) + + // Construct the htlc output on Bob's commitment tx, and decide its + // index based on the commit type below. + htlcOutpoint := wire.OutPoint{Hash: closeTx.TxHash()} + + // Check the current mempool state and we should see, + // - Carol's direct spend tx. + // - Bob's local output sweep tx, if this is NOT script enforced lease. + // - Carol's anchor sweep tx, if the commitment type is anchor. + switch c { + case lnrpc.CommitmentType_LEGACY: + htlcOutpoint.Index = 0 + ht.Miner.AssertNumTxsInMempool(2) + + case lnrpc.CommitmentType_ANCHORS: + htlcOutpoint.Index = 2 + ht.Miner.AssertNumTxsInMempool(3) + + case lnrpc.CommitmentType_SCRIPT_ENFORCED_LEASE: + htlcOutpoint.Index = 2 + ht.Miner.AssertNumTxsInMempool(2) + } + + // Get the current height to compute number of blocks to mine to + // trigger the timeout resolver from Bob. + _, height := ht.Miner.GetBestBlock() + + // We'll now mine enough blocks to trigger Bob's htlc timeout resolver + // to act. Once his timeout resolver starts, it will extract the + // preimage from Carol's direct spend tx found in the mempool. + numBlocks = htlc.ExpirationHeight - uint32(height) - + lncfg.DefaultOutgoingBroadcastDelta + + // Decrease the fee rate used by the sweeper so Bob's timeout tx will + // not replace Carol's direct spend tx. + ht.SetFeeEstimate(1000) + + // Mine empty blocks so Carol's direct spend tx stays in mempool. Once + // the height is reached, Bob's timeout resolver will resolve the htlc + // by extracing the preimage from the mempool. + ht.MineEmptyBlocks(int(numBlocks)) + + // For neutrino backend, the timeout resolver needs to extract the + // preimage from the blocks. + if ht.IsNeutrinoBackend() { + // Make sure the direct spend tx is still in the mempool. + ht.Miner.AssertOutpointInMempool(htlcOutpoint) + + // Mine a block to confirm Carol's direct spend tx. + ht.MineBlocks(1) + } + + // Finally, check that the Alice's payment is marked as succeeded as + // Bob has settled the htlc using the preimage extracted from Carol's + // direct spend tx. + ht.AssertPaymentStatus(alice, preimage, lnrpc.Payment_SUCCEEDED) + + // NOTE: for non-standby nodes there's no need to clean up the force + // close as long as the mempool is cleaned. + ht.CleanShutDown() +} diff --git a/lntest/node/config.go b/lntest/node/config.go index d69c7f80d..f044475e7 100644 --- a/lntest/node/config.go +++ b/lntest/node/config.go @@ -236,6 +236,10 @@ func (cfg *BaseNodeConfig) GenArgs() []string { // Use a small cache duration so the `DescribeGraph` can be // updated quicker. "--caches.rpc-graph-cache-duration=100ms", + + // Speed up the tests for bitcoind backend. + "--bitcoind.blockpollinginterval=100ms", + "--bitcoind.txpollinginterval=100ms", } args = append(args, nodeArgs...) From f72d6dfaf5e53ef7f3383731656b9589b8fb6c0f Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Fri, 7 Apr 2023 19:14:34 +0800 Subject: [PATCH 14/16] docs: update release notes for mempool watch --- docs/release-notes/release-notes-0.16.1.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/release-notes/release-notes-0.16.1.md b/docs/release-notes/release-notes-0.16.1.md index 771997056..4b62a1c8b 100644 --- a/docs/release-notes/release-notes-0.16.1.md +++ b/docs/release-notes/release-notes-0.16.1.md @@ -50,6 +50,10 @@ https://github.com/lightningnetwork/lnd/pull/7359) * Optimize script allocation size in order to save [memory](https://github.com/lightningnetwork/lnd/pull/7464). +* When resolving outgoing HTLCs onchain, the HTLC timeout resolver will now + [monitor mempool](https://github.com/lightningnetwork/lnd/pull/7564) for + faster preimage extraction. + ## Spec * [Add test vectors for From d5db1950cd7f222b31a133cf09ece181fc5c2faf Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Thu, 13 Apr 2023 22:28:18 +0800 Subject: [PATCH 15/16] mod: update btcwallet version --- go.mod | 18 +++++++++--------- go.sum | 36 ++++++++++++++++++------------------ 2 files changed, 27 insertions(+), 27 deletions(-) diff --git a/go.mod b/go.mod index 7d052a910..7ed96a52c 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/btcsuite/btcd/btcutil/psbt v1.1.8 github.com/btcsuite/btcd/chaincfg/chainhash v1.0.2 github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f - github.com/btcsuite/btcwallet v0.16.8-0.20230324081040-520f650ce045 + github.com/btcsuite/btcwallet v0.16.8 github.com/btcsuite/btcwallet/wallet/txauthor v1.3.2 github.com/btcsuite/btcwallet/wallet/txrules v1.2.0 github.com/btcsuite/btcwallet/walletdb v1.4.0 @@ -50,12 +50,12 @@ require ( github.com/urfave/cli v1.22.9 go.etcd.io/etcd/client/pkg/v3 v3.5.7 go.etcd.io/etcd/client/v3 v3.5.7 - golang.org/x/crypto v0.1.0 + golang.org/x/crypto v0.7.0 golang.org/x/exp v0.0.0-20221111094246-ab4555d3164f golang.org/x/mobile v0.0.0-20190719004257-d2bd2a29d028 - golang.org/x/net v0.7.0 - golang.org/x/sync v0.0.0-20220819030929-7fc1605a5dde - golang.org/x/term v0.5.0 + golang.org/x/net v0.8.0 + golang.org/x/sync v0.1.0 + golang.org/x/term v0.6.0 golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba google.golang.org/grpc v1.41.0 google.golang.org/protobuf v1.27.1 @@ -149,10 +149,10 @@ require ( go.uber.org/atomic v1.7.0 // indirect go.uber.org/multierr v1.6.0 // indirect go.uber.org/zap v1.17.0 // indirect - golang.org/x/mod v0.6.0 // indirect - golang.org/x/sys v0.5.0 // indirect - golang.org/x/text v0.7.0 // indirect - golang.org/x/tools v0.2.0 // indirect + golang.org/x/mod v0.8.0 // indirect + golang.org/x/sys v0.6.0 // indirect + golang.org/x/text v0.8.0 // indirect + golang.org/x/tools v0.6.0 // indirect google.golang.org/genproto v0.0.0-20210617175327-b9e0b3197ced // indirect gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect gopkg.in/errgo.v1 v1.0.1 // indirect diff --git a/go.sum b/go.sum index eddeb1133..4dd227d45 100644 --- a/go.sum +++ b/go.sum @@ -89,8 +89,8 @@ github.com/btcsuite/btcd/chaincfg/chainhash v1.0.2/go.mod h1:7SFka0XMvUgj3hfZtyd github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f h1:bAs4lUbRJpnnkd9VhRV3jjAVU7DJVjMaK+IsvSeZvFo= github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f/go.mod h1:TdznJufoqS23FtqVCzL0ZqgP5MqXbb4fg/WgDys70nA= github.com/btcsuite/btcutil v0.0.0-20190425235716-9e5f4b9a998d/go.mod h1:+5NJ2+qvTyV9exUAL/rxXi3DcLg2Ts+ymUAY5y4NvMg= -github.com/btcsuite/btcwallet v0.16.8-0.20230324081040-520f650ce045 h1:302lmdYbONzrLACkm9IlCobMYuYOjF8to+tUhaQsTzc= -github.com/btcsuite/btcwallet v0.16.8-0.20230324081040-520f650ce045/go.mod h1:WBcb0CNEgzjF2EdatcERzG7cd/lhHXtNJ4VjIePOdXM= +github.com/btcsuite/btcwallet v0.16.8 h1:ADMormFSOBMANB/ZWvU00VDnFelZXY9Uyc9wW0dNFzM= +github.com/btcsuite/btcwallet v0.16.8/go.mod h1:ynC16HgMU03dYyUit6b0+IZjDg9KVnr2VNLie9ZXG1I= github.com/btcsuite/btcwallet/wallet/txauthor v1.3.2 h1:etuLgGEojecsDOYTII8rYiGHjGyV5xTqsXi+ZQ715UU= github.com/btcsuite/btcwallet/wallet/txauthor v1.3.2/go.mod h1:Zpk/LOb2sKqwP2lmHjaZT9AdaKsHPSbNLm2Uql5IQ/0= github.com/btcsuite/btcwallet/wallet/txrules v1.2.0 h1:BtEN5Empw62/RVnZ0VcJaVtVlBijnLlJY+dwjAye2Bg= @@ -626,8 +626,8 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh golang.org/x/crypto v0.0.0-20201203163018-be400aefbc4c/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I= golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= -golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU= -golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw= +golang.org/x/crypto v0.7.0 h1:AvwMYaRytfdeVt3u6mLaxYtErKYjxA2OXjJ1HHq6t3A= +golang.org/x/crypto v0.7.0/go.mod h1:pYwdfH91IfpZVANVyUOhSIPZaFoJGxTFbZhFTx+dXZU= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8= @@ -663,8 +663,8 @@ golang.org/x/mod v0.1.1-0.20191107180719-034126e5016b/go.mod h1:QqPTAvyqsEbceGzB golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/mod v0.6.0 h1:b9gGHsz9/HhJ3HF5DHQytPpuwocVTChQJK3AvoLRD5I= -golang.org/x/mod v0.6.0/go.mod h1:4mET923SAdbXp2ki8ey+zGs1SLqsuM2Y0uvdZR/fUNI= +golang.org/x/mod v0.8.0 h1:LUYupSeNrTNCGzR/hVBk2NHZO4hXcVaW1k4Qx7rjPx8= +golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/net v0.0.0-20180719180050-a680a1efc54d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -703,8 +703,8 @@ golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwY golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g= -golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ= +golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -723,8 +723,8 @@ golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20220819030929-7fc1605a5dde h1:ejfdSekXMDxDLbRrJMwUk6KnSLZ2McaUCVcIKM+N6jc= -golang.org/x/sync v0.0.0-20220819030929-7fc1605a5dde/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180816055513-1c9583448a9c/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -782,12 +782,12 @@ golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU= -golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= -golang.org/x/term v0.5.0 h1:n2a8QNdAb0sZNpU9R1ALUXBbY+w51fCQDN+7EdxNBsY= -golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/term v0.6.0 h1:clScbb1cHjoCkyRbWwBEUZ5H/tIFu5TAXIqaZD0Gcjw= +golang.org/x/term v0.6.0/go.mod h1:m6U89DPEgQRMq3DNkDClhWw02AUbt2daBVO4cn4Hv9U= golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= @@ -796,8 +796,8 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo= -golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.8.0 h1:57P1ETyNKtuIjB4SRd15iJxuhj8Gc416Y78H3qgMh68= +golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= @@ -851,8 +851,8 @@ golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d/go.mod h1:njjCfa9FT2d7l9Bc golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.3/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= -golang.org/x/tools v0.2.0 h1:G6AHpWxTMGY1KyEYoAQ5WTtIekUUvDNjan3ugu60JvE= -golang.org/x/tools v0.2.0/go.mod h1:y4OqIKeOV/fWJetJ8bXPU1sEVniLMIyDAZWeHdV+NTA= +golang.org/x/tools v0.6.0 h1:BOw41kyTf3PuCW1pVQf8+Cyg8pMlkYB1oo9iJ6D/lKM= +golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= golang.org/x/xerrors v0.0.0-20190410155217-1f06c39b4373/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= From 66d1392f7f1ac75b94c657f3d9b37989086de117 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Wed, 19 Apr 2023 01:18:04 +0800 Subject: [PATCH 16/16] chainreg: update bitcoind polling config --- chainreg/chainregistry.go | 15 ++++++++++----- lncfg/bitcoind.go | 6 ++++++ 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/chainreg/chainregistry.go b/chainreg/chainregistry.go index c4a5caecd..f025dbb64 100644 --- a/chainreg/chainregistry.go +++ b/chainreg/chainregistry.go @@ -253,6 +253,8 @@ func GenDefaultBtcConstraints() channeldb.ChannelConstraints { // NewPartialChainControl creates a new partial chain control that contains all // the parts that can be purely constructed from the passed in global // configuration and doesn't need any wallet instance yet. +// +//nolint:lll func NewPartialChainControl(cfg *Config) (*PartialChainControl, func(), error) { // Set the RPC config from the "home" chain. Multi-chain isn't yet // active, so we'll restrict usage to a particular chain for now. @@ -414,14 +416,17 @@ func NewPartialChainControl(cfg *Config) (*PartialChainControl, func(), error) { if bitcoindMode.RPCPolling { bitcoindCfg.PollingConfig = &chain.PollingConfig{ - BlockPollingInterval: bitcoindMode.BlockPollingInterval, - TxPollingInterval: bitcoindMode.TxPollingInterval, + BlockPollingInterval: bitcoindMode.BlockPollingInterval, + TxPollingInterval: bitcoindMode.TxPollingInterval, + TxPollingIntervalJitter: lncfg.DefaultTxPollingJitter, } } else { bitcoindCfg.ZMQConfig = &chain.ZMQConfig{ - ZMQBlockHost: bitcoindMode.ZMQPubRawBlock, - ZMQTxHost: bitcoindMode.ZMQPubRawTx, - ZMQReadDeadline: bitcoindMode.ZMQReadDeadline, + ZMQBlockHost: bitcoindMode.ZMQPubRawBlock, + ZMQTxHost: bitcoindMode.ZMQPubRawTx, + ZMQReadDeadline: bitcoindMode.ZMQReadDeadline, + MempoolPollingInterval: bitcoindMode.TxPollingInterval, + PollingIntervalJitter: lncfg.DefaultTxPollingJitter, } } diff --git a/lncfg/bitcoind.go b/lncfg/bitcoind.go index a54aefa27..b574aef16 100644 --- a/lncfg/bitcoind.go +++ b/lncfg/bitcoind.go @@ -2,6 +2,12 @@ package lncfg import "time" +const ( + // DefaultTxPollingJitter defines the default TxPollingIntervalJitter + // to be used for bitcoind backend. + DefaultTxPollingJitter = 0.5 +) + // Bitcoind holds the configuration options for the daemon's connection to // bitcoind. //