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

Merge pull request #919 from starius/conf-chan

Browse source 
sweepbatcher: notify caller about confirmations
This commit is contained in:
Boris Nagaev 2025-06-12 15:14:15 -03:00 • committed by GitHub
commit 59f67de306
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 694 additions and 193 deletions
Download patch file Download diff file Expand all files Collapse all files

23
sweepbatcher/store.go
View file

@ -213,8 +213,8 @@ type dbBatch struct {
// ID is the unique identifier of the batch.
ID int32
// State is the current state of the batch.
State string
// Confirmed is set when the batch is fully confirmed.
Confirmed bool
// BatchTxid is the txid of the batch transaction.
BatchTxid chainhash.Hash
@ -255,11 +255,8 @@ type dbSweep struct {
// convertBatchRow converts a batch row from db to a sweepbatcher.Batch struct.
func convertBatchRow(row sqlc.SweepBatch) *dbBatch {
batch := dbBatch{
ID: row.ID,
}
if row.Confirmed {
batch.State = batchOpen
ID: row.ID,
Confirmed: row.Confirmed,
}
if row.BatchTxID.Valid {
@ -288,7 +285,7 @@ func convertBatchRow(row sqlc.SweepBatch) *dbBatch {
// it into the database.
func batchToInsertArgs(batch dbBatch) sqlc.InsertBatchParams {
args := sqlc.InsertBatchParams{
Confirmed: false,
Confirmed: batch.Confirmed,
BatchTxID: sql.NullString{
Valid: true,
String: batch.BatchTxid.String(),
@ -305,10 +302,6 @@ func batchToInsertArgs(batch dbBatch) sqlc.InsertBatchParams {
MaxTimeoutDistance: batch.MaxTimeoutDistance,
}
if batch.State == batchConfirmed {
args.Confirmed = true
}
return args
}
@ -317,7 +310,7 @@ func batchToInsertArgs(batch dbBatch) sqlc.InsertBatchParams {
func batchToUpdateArgs(batch dbBatch) sqlc.UpdateBatchParams {
args := sqlc.UpdateBatchParams{
ID: batch.ID,
Confirmed: false,
Confirmed: batch.Confirmed,
BatchTxID: sql.NullString{
Valid: true,
String: batch.BatchTxid.String(),
@ -333,10 +326,6 @@ func batchToUpdateArgs(batch dbBatch) sqlc.UpdateBatchParams {
},
}
if batch.State == batchConfirmed {
args.Confirmed = true
}
return args
}

8
sweepbatcher/store_mock.go
View file

@ -36,7 +36,7 @@ func (s *StoreMock) FetchUnconfirmedSweepBatches(ctx context.Context) (
result := []*dbBatch{}
for _, batch := range s.batches {
if batch.State != "confirmed" {
if !batch.Confirmed {
result = append(result, &batch)
}
}
@ -91,7 +91,7 @@ func (s *StoreMock) ConfirmBatch(ctx context.Context, id int32) error {
return errors.New("batch not found")
}
batch.State = "confirmed"
batch.Confirmed = true
s.batches[batch.ID] = batch
return nil
@ -201,7 +201,7 @@ func (s *StoreMock) TotalSweptAmount(ctx context.Context, batchID int32) (
return 0, errors.New("batch not found")
}
if batch.State != batchConfirmed && batch.State != batchClosed {
if !batch.Confirmed {
return 0, nil
}
@ -212,5 +212,5 @@ func (s *StoreMock) TotalSweptAmount(ctx context.Context, batchID int32) (
}
}
return 0, nil
return total, nil
}

248
sweepbatcher/sweep_batch.go
View file

@ -135,7 +135,9 @@ const (
Open batchState = 0
// Closed is the state in which the batch is no longer able to accept
// new sweeps.
// new sweeps. NOTE: this state exists only in-memory. In the database
// it is stored as Open and converted to Closed after a spend
// notification arrives (quickly after start of Batch.Run).
Closed batchState = 1
// Confirmed is the state in which the batch transaction has reached the
@ -870,8 +872,8 @@ func (b *batch) Run(ctx context.Context) error {
// completes.
timerChan := clock.TickAfter(b.cfg.batchPublishDelay)
b.Infof("started, primary %s, total sweeps %d",
b.primarySweepID, len(b.sweeps))
b.Infof("started, primary %s, total sweeps %d, state: %d",
b.primarySweepID, len(b.sweeps), b.state)
for {
// If the batch is not empty, find earliest initialDelay.
@ -1822,27 +1824,22 @@ func (b *batch) monitorSpend(ctx context.Context, primarySweep sweep) error {
b.Infof("monitoring spend for outpoint %s",
primarySweep.outpoint.String())
for {
select {
case spend := <-spendChan:
select {
case spend := <-spendChan:
select {
case b.spendChan <- spend:
case <-ctx.Done():
}
return
case err := <-spendErr:
b.writeToErrChan(
fmt.Errorf("spend error: %w", err),
)
return
case b.spendChan <- spend:
case <-ctx.Done():
return
}
case err := <-spendErr:
b.writeToSpendErrChan(ctx, err)
b.writeToErrChan(
fmt.Errorf("spend error: %w", err),
)
case <-ctx.Done():
}
}()
@ -1876,39 +1873,33 @@ func (b *batch) monitorConfirmations(ctx context.Context) error {
defer cancel()
defer b.wg.Done()
for {
select {
case conf := <-confChan:
select {
case conf := <-confChan:
select {
case b.confChan <- conf:
case <-ctx.Done():
}
return
case err := <-errChan:
b.writeToErrChan(fmt.Errorf("confirmations "+
"monitoring error: %w", err))
return
case <-reorgChan:
// A re-org has been detected. We set the batch
// state back to open since our batch
// transaction is no longer present in any
// block. We can accept more sweeps and try to
// publish new transactions, at this point we
// need to monitor again for a new spend.
select {
case b.reorgChan <- struct{}{}:
case <-ctx.Done():
}
return
case b.confChan <- conf:
case <-ctx.Done():
return
}
case err := <-errChan:
b.writeToConfErrChan(ctx, err)
b.writeToErrChan(fmt.Errorf("confirmations "+
"monitoring error: %w", err))
case <-reorgChan:
// A re-org has been detected. We set the batch
// state back to open since our batch
// transaction is no longer present in any
// block. We can accept more sweeps and try to
// publish new transactions, at this point we
// need to monitor again for a new spend.
select {
case b.reorgChan <- struct{}{}:
case <-ctx.Done():
}
case <-ctx.Done():
}
}()
@ -2112,16 +2103,19 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
"purged swaps: %v, purged groups: %v", confirmedSweeps,
purgedSweeps, purgedSwaps, len(purgeList))
err = b.monitorConfirmations(ctx)
if err != nil {
return err
}
// We are no longer able to accept new sweeps, so we mark the batch as
// closed and persist on storage.
b.state = Closed
return b.persist(ctx)
if err = b.persist(ctx); err != nil {
return fmt.Errorf("saving batch failed: %w", err)
}
if err = b.monitorConfirmations(ctx); err != nil {
return fmt.Errorf("monitorConfirmations failed: %w", err)
}
return nil
}
// handleConf handles a confirmation notification. This is the final step of the
@ -2166,7 +2160,55 @@ func (b *batch) handleConf(ctx context.Context,
b.Infof("confirmed in txid %s", b.batchTxid)
b.state = Confirmed
return b.store.ConfirmBatch(ctx, b.id)
if err := b.store.ConfirmBatch(ctx, b.id); err != nil {
return fmt.Errorf("failed to store confirmed state: %w", err)
}
// Calculate the fee portion that each sweep should pay for the batch.
// TODO: make sure spendTx matches b.sweeps.
var totalSweptAmt btcutil.Amount
for _, s := range b.sweeps {
totalSweptAmt += s.value
}
feePortionPaidPerSweep, roundingDifference := getFeePortionForSweep(
spendTx, len(b.sweeps), totalSweptAmt,
)
// Send the confirmation to all the notifiers.
for _, s := range b.sweeps {
// If the sweep's notifier is empty then this means that
// a swap is not waiting to read an update from it, so
// we can skip the notification part.
if s.notifier == nil || s.notifier.ConfChan == nil {
continue
}
confDetail := &ConfDetail{
TxConfirmation: conf,
OnChainFeePortion: getFeePortionPaidBySweep(
spendTx, feePortionPaidPerSweep,
roundingDifference, &s,
),
}
// Notify the caller in a goroutine to avoid possible dead-lock.
go func(notifier *SpendNotifier) {
// Note that we don't unblock on ctx, because it will
// expire soon, when batch.Run completes. The caller is
// responsible to consume ConfChan or close QuitChan.
select {
// Try to write the confirmation to the notification
// channel.
case notifier.ConfChan <- confDetail:
// If a quit signal was provided by the swap,
// continue.
case <-notifier.QuitChan:
}
}(s.notifier)
}
return nil
}
// isComplete returns true if the batch is completed. This method is used by the
@ -2189,7 +2231,7 @@ func (b *batch) persist(ctx context.Context) error {
bch := &dbBatch{}
bch.ID = b.id
bch.State = stateEnumToString(b.state)
bch.Confirmed = b.state == Confirmed
if b.batchTxid != nil {
bch.BatchTxid = *b.batchTxid
@ -2248,7 +2290,7 @@ func (b *batch) getBatchDestAddr(ctx context.Context) (btcutil.Address, error) {
func (b *batch) insertAndAcquireID(ctx context.Context) (int32, error) {
bch := &dbBatch{}
bch.State = stateEnumToString(b.state)
bch.Confirmed = b.state == Confirmed
bch.MaxTimeoutDistance = b.cfg.maxTimeoutDistance
id, err := b.store.InsertSweepBatch(ctx, bch)
@ -2285,6 +2327,81 @@ func (b *batch) writeToErrChan(err error) {
}
}
// writeToSpendErrChan sends an error to spend error channels of all the sweeps.
func (b *batch) writeToSpendErrChan(ctx context.Context, spendErr error) {
done, err := b.scheduleNextCall()
if err != nil {
done()
return
}
notifiers := make([]*SpendNotifier, 0, len(b.sweeps))
for _, s := range b.sweeps {
// If the sweep's notifier is empty then this means that a swap
// is not waiting to read an update from it, so we can skip
// the notification part.
if s.notifier == nil || s.notifier.SpendErrChan == nil {
continue
}
notifiers = append(notifiers, s.notifier)
}
done()
for _, notifier := range notifiers {
select {
// Try to write the error to the notification
// channel.
case notifier.SpendErrChan <- spendErr:
// If a quit signal was provided by the swap,
// continue.
case <-notifier.QuitChan:
// If the context was canceled, stop.
case <-ctx.Done():
}
}
}
// writeToConfErrChan sends an error to confirmation error channels of all the
// sweeps.
func (b *batch) writeToConfErrChan(ctx context.Context, confErr error) {
done, err := b.scheduleNextCall()
if err != nil {
done()
return
}
notifiers := make([]*SpendNotifier, 0, len(b.sweeps))
for _, s := range b.sweeps {
// If the sweep's notifier is empty then this means that a swap
// is not waiting to read an update from it, so we can skip
// the notification part.
if s.notifier == nil || s.notifier.ConfErrChan == nil {
continue
}
notifiers = append(notifiers, s.notifier)
}
done()
for _, notifier := range notifiers {
select {
// Try to write the error to the notification
// channel.
case notifier.ConfErrChan <- confErr:
// If a quit signal was provided by the swap,
// continue.
case <-notifier.QuitChan:
// If the context was canceled, stop.
case <-ctx.Done():
}
}
}
func (b *batch) persistSweep(ctx context.Context, sweep sweep,
completed bool) error {
@ -2312,18 +2429,3 @@ func clampBatchFee(fee btcutil.Amount,
return fee
}
func stateEnumToString(state batchState) string {
switch state {
case Open:
return batchOpen
case Closed:
return batchClosed
case Confirmed:
return batchConfirmed
}
return ""
}

259
sweepbatcher/sweep_batcher.go
View file

@ -20,6 +20,7 @@ import (
"github.com/lightninglabs/loop/loopdb"
"github.com/lightninglabs/loop/swap"
"github.com/lightninglabs/loop/utils"
"github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/clock"
"github.com/lightningnetwork/lnd/input"
"github.com/lightningnetwork/lnd/lntypes"
@ -31,18 +32,6 @@ const (
// of sweeps that can appear in the same batch.
defaultMaxTimeoutDistance = 288
// batchOpen is the string representation of the state of a batch that
// is open.
batchOpen = "open"
// batchClosed is the string representation of the state of a batch
// that is closed.
batchClosed = "closed"
// batchConfirmed is the string representation of the state of a batch
// that is confirmed.
batchConfirmed = "confirmed"
// defaultMainnetPublishDelay is the default publish delay that is used
// for mainnet.
defaultMainnetPublishDelay = 5 * time.Second
@ -292,6 +281,8 @@ type addSweepsRequest struct {
parentBatch *dbBatch
}
// SpendDetail is a notification that is send to the user of sweepbatcher when
// a batch gets the first confirmation.
type SpendDetail struct {
// Tx is the transaction that spent the outpoint.
Tx *wire.MsgTx
@ -303,17 +294,38 @@ type SpendDetail struct {
OnChainFeePortion btcutil.Amount
}
// ConfDetail is a notification that is send to the user of sweepbatcher when
// a batch is fully confirmed, i.e. gets batchConfHeight confirmations.
type ConfDetail struct {
// TxConfirmation has data about the confirmation of the transaction.
*chainntnfs.TxConfirmation
// OnChainFeePortion is the fee portion that was paid to get this sweep
// confirmed on chain. This is the difference between the value of the
// outpoint and the value of all sweeps that were included in the batch
// divided by the number of sweeps.
OnChainFeePortion btcutil.Amount
}
// SpendNotifier is a notifier that is used to notify the requester of a sweep
// that the sweep was successful.
type SpendNotifier struct {
// SpendChan is a channel where the spend details are received.
SpendChan chan *SpendDetail
SpendChan chan<- *SpendDetail
// SpendErrChan is a channel where spend errors are received.
SpendErrChan chan error
SpendErrChan chan<- error
// ConfChan is a channel where the confirmation details are received.
// This channel is optional.
ConfChan chan<- *ConfDetail
// ConfErrChan is a channel where confirmation errors are received.
// This channel is optional.
ConfErrChan chan<- error
// QuitChan is a channel that can be closed to stop the notifier.
QuitChan chan bool
QuitChan <-chan bool
}
var (
@ -760,7 +772,7 @@ func (b *Batcher) AddSweep(ctx context.Context, sweepReq *SweepRequest) error {
"sweep %x: %w", sweep.swapHash[:6], err)
}
if parentBatch.State == batchConfirmed {
if parentBatch.Confirmed {
fullyConfirmed = true
}
}
@ -844,7 +856,7 @@ func (b *Batcher) handleSweeps(ctx context.Context, sweeps []*sweep,
if completed && *notifier != (SpendNotifier{}) {
// The parent batch is indeed confirmed, meaning it is complete
// and we won't be able to attach this sweep to it.
if parentBatch.State == batchConfirmed {
if parentBatch.Confirmed {
return b.monitorSpendAndNotify(
ctx, sweep, parentBatch.ID, notifier,
)
@ -1093,15 +1105,18 @@ func (b *Batcher) FetchUnconfirmedBatches(ctx context.Context) ([]*batch,
batch := batch{}
batch.id = bch.ID
switch bch.State {
case batchOpen:
batch.state = Open
case batchClosed:
batch.state = Closed
case batchConfirmed:
if bch.Confirmed {
batch.state = Confirmed
} else {
// We don't store Closed state separately in DB.
// If the batch is closed (included into a block, but
// not fully confirmed), it is now considered Open
// again. It will receive a spending notification as
// soon as it starts, so it is not an issue. If a sweep
// manages to be added during this time, it will be
// detected as missing when analyzing the spend
// notification and will be added to new batch.
batch.state = Open
}
batch.batchTxid = &bch.BatchTxid
@ -1123,16 +1138,19 @@ func (b *Batcher) FetchUnconfirmedBatches(ctx context.Context) ([]*batch,
}
// monitorSpendAndNotify monitors the spend of a specific outpoint and writes
// the response back to the response channel.
// the response back to the response channel. It is called if the batch is fully
// confirmed and we just need to deliver the data back to the caller though
// SpendNotifier.
func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
parentBatchID int32, notifier *SpendNotifier) error {
spendCtx, cancel := context.WithCancel(ctx)
defer cancel()
// Then we get the total amount that was swept by the batch.
totalSwept, err := b.store.TotalSweptAmount(ctx, parentBatchID)
if err != nil {
cancel()
return err
}
@ -1141,65 +1159,170 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
sweep.initiationHeight,
)
if err != nil {
cancel()
return err
}
b.wg.Add(1)
go func() {
defer cancel()
defer b.wg.Done()
infof("Batcher monitoring spend for swap %x",
sweep.swapHash[:6])
for {
select {
case spend := <-spendChan:
spendTx := spend.SpendingTx
// Calculate the fee portion that each sweep
// should pay for the batch.
feePortionPerSweep, roundingDifference :=
getFeePortionForSweep(
spendTx, len(spendTx.TxIn),
totalSwept,
)
onChainFeePortion := getFeePortionPaidBySweep(
spendTx, feePortionPerSweep,
roundingDifference, sweep,
select {
case spend := <-spendChan:
spendTx := spend.SpendingTx
// Calculate the fee portion that each sweep should pay
// for the batch.
feePortionPerSweep, roundingDifference :=
getFeePortionForSweep(
spendTx, len(spendTx.TxIn),
totalSwept,
)
// Notify the requester of the spend
// with the spend details, including the fee
// portion for this particular sweep.
spendDetail := &SpendDetail{
Tx: spendTx,
onChainFeePortion := getFeePortionPaidBySweep(
spendTx, feePortionPerSweep,
roundingDifference, sweep,
)
// Notify the requester of the spend with the spend
// details, including the fee portion for this
// particular sweep.
spendDetail := &SpendDetail{
Tx: spendTx,
OnChainFeePortion: onChainFeePortion,
}
select {
// Try to write the update to the notification channel.
case notifier.SpendChan <- spendDetail:
err := b.monitorConfAndNotify(
ctx, sweep, notifier, spendTx,
onChainFeePortion,
)
if err != nil {
b.writeToErrChan(
ctx, fmt.Errorf("monitor conf "+
"failed: %w", err),
)
}
// If a quit signal was provided by the swap, continue.
case <-notifier.QuitChan:
// If the context was canceled, stop.
case <-ctx.Done():
}
return
case err := <-spendErr:
select {
// Try to write the error to the notification
// channel.
case notifier.SpendErrChan <- err:
// If a quit signal was provided by the swap,
// continue.
case <-notifier.QuitChan:
// If the context was canceled, stop.
case <-ctx.Done():
}
b.writeToErrChan(
ctx, fmt.Errorf("spend error: %w", err),
)
return
// If a quit signal was provided by the swap, continue.
case <-notifier.QuitChan:
return
// If the context was canceled, stop.
case <-ctx.Done():
return
}
}()
return nil
}
// monitorConfAndNotify monitors the confirmation of a specific transaction and
// writes the response back to the response channel. It is called if the batch
// is fully confirmed and we just need to deliver the data back to the caller
// though SpendNotifier.
func (b *Batcher) monitorConfAndNotify(ctx context.Context, sweep *sweep,
notifier *SpendNotifier, spendTx *wire.MsgTx,
onChainFeePortion btcutil.Amount) error {
// If confirmation notifications were not requested, stop.
if notifier.ConfChan == nil && notifier.ConfErrChan == nil {
return nil
}
batchTxid := spendTx.TxHash()
if len(spendTx.TxOut) != 1 {
return fmt.Errorf("unexpected number of outputs in batch: %d, "+
"want %d", len(spendTx.TxOut), 1)
}
batchPkScript := spendTx.TxOut[0].PkScript
reorgChan := make(chan struct{})
confCtx, cancel := context.WithCancel(ctx)
confChan, errChan, err := b.chainNotifier.RegisterConfirmationsNtfn(
confCtx, &batchTxid, batchPkScript, batchConfHeight,
sweep.initiationHeight, lndclient.WithReOrgChan(reorgChan),
)
if err != nil {
cancel()
return err
}
b.wg.Add(1)
go func() {
defer cancel()
defer b.wg.Done()
select {
case conf := <-confChan:
if notifier.ConfChan != nil {
confDetail := &ConfDetail{
TxConfirmation: conf,
OnChainFeePortion: onChainFeePortion,
}
select {
case notifier.SpendChan <- spendDetail:
case notifier.ConfChan <- confDetail:
case <-notifier.QuitChan:
case <-ctx.Done():
}
return
case err := <-spendErr:
select {
case notifier.SpendErrChan <- err:
case <-ctx.Done():
}
b.writeToErrChan(
ctx, fmt.Errorf("spend error: %w", err),
)
return
case <-notifier.QuitChan:
return
case <-ctx.Done():
return
}
case err := <-errChan:
if notifier.ConfErrChan != nil {
select {
case notifier.ConfErrChan <- err:
case <-notifier.QuitChan:
case <-ctx.Done():
}
}
b.writeToErrChan(ctx, fmt.Errorf("confirmations "+
"monitoring error: %w", err))
case <-reorgChan:
// A re-org has been detected, but the batch is fully
// confirmed and this is unexpected. Crash the batcher.
b.writeToErrChan(ctx, fmt.Errorf("unexpected reorg"))
case <-ctx.Done():
}
}()

313
sweepbatcher/sweep_batcher_test.go
View file

@ -762,9 +762,9 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
batcherStore, sweepStore)
runErrChan := make(chan error)
go func() {
err := batcher.Run(ctx)
checkBatcherError(t, err)
runErrChan <- batcher.Run(ctx)
}()
// Create a sweep request.
@ -772,13 +772,24 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
Hash: chainhash.Hash{1, 1},
Index: 1,
}
const (
inputValue = 111
outputValue = 50
fee = inputValue - outputValue
)
spendErrChan := make(chan error, 1)
notifier := &SpendNotifier{
SpendChan: make(chan *SpendDetail, 1),
SpendErrChan: spendErrChan,
QuitChan: make(chan bool, 1),
}
sweepReq1 := SweepRequest{
SwapHash: lntypes.Hash{1, 1, 1},
Inputs: []Input{{
Value: 111,
Value: inputValue,
Outpoint: op1,
}},
Notifier: &dummyNotifier,
Notifier: notifier,
}
const initiationHeight = 550
@ -786,7 +797,7 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
swap1 := &loopdb.LoopOutContract{
SwapContract: loopdb.SwapContract{
CltvExpiry: 111,
AmountRequested: 111,
AmountRequested: inputValue,
ProtocolVersion: loopdb.ProtocolVersionMuSig2,
HtlcKeys: htlcKeys,
InitiationHeight: initiationHeight,
@ -806,33 +817,27 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
// When batch is successfully created it will execute it's first step,
// which leads to a spend monitor of the primary sweep.
<-lnd.RegisterSpendChannel
spendReg := <-lnd.RegisterSpendChannel
// Wait for tx to be published.
<-lnd.TxPublishChannel
// Eventually request will be consumed and a new batch will spin up.
var primarySweepID wire.OutPoint
require.Eventually(t, func() bool {
return batcher.numBatches(ctx) == 1
}, test.Timeout, eventuallyCheckFrequency)
batch := tryGetOnlyBatch(ctx, batcher)
if batch == nil {
return false
}
// Find the batch and assign it to a local variable for easier access.
batch := &batch{}
for _, btch := range getBatches(ctx, batcher) {
btch.testRunInEventLoop(ctx, func() {
if btch.primarySweepID == op1 {
batch = btch
}
})
}
primarySweepID = batch.snapshot(ctx).primarySweepID
require.Eventually(t, func() bool {
// Batch should have the sweep stored.
return batch.numSweeps(ctx) == 1
}, test.Timeout, eventuallyCheckFrequency)
// The primary sweep id should be that of the first inserted sweep.
require.Equal(t, batch.primarySweepID, op1)
// Wait for tx to be published.
<-lnd.TxPublishChannel
require.Equal(t, primarySweepID, op1)
err = lnd.NotifyHeight(601)
require.NoError(t, err)
@ -840,7 +845,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
// After receiving a height notification the batch will step again,
// leading to a new spend monitoring.
require.Eventually(t, func() bool {
batch := batch.snapshot(ctx)
batch := tryGetOnlyBatch(ctx, batcher)
if batch == nil {
return false
}
batch = batch.snapshot(ctx)
return batch.currentHeight == 601
}, test.Timeout, eventuallyCheckFrequency)
@ -848,6 +857,60 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
// Wait for tx to be published.
<-lnd.TxPublishChannel
// Emulate spend error.
testError := errors.New("test error")
spendReg.ErrChan <- testError
// Make sure the caller of AddSweep got the spending error.
notifierErr := <-spendErrChan
require.Error(t, notifierErr)
require.ErrorIs(t, notifierErr, testError)
// Wait for the batcher to crash because of the spending error.
runErr := <-runErrChan
require.ErrorIs(t, runErr, testError)
// Now launch the batcher again.
batcher = NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
batcherStore, sweepStore)
go func() {
runErrChan <- batcher.Run(ctx)
}()
// When batch is successfully created it will execute it's first step,
// which leads to a spend monitor of the primary sweep.
<-lnd.RegisterSpendChannel
// Wait for tx to be published.
<-lnd.TxPublishChannel
// Deliver sweep request to batcher.
spendChan := make(chan *SpendDetail, 1)
confErrChan := make(chan error)
notifier = &SpendNotifier{
SpendChan: spendChan,
SpendErrChan: make(chan error, 1),
ConfErrChan: confErrChan,
QuitChan: make(chan bool, 1),
}
sweepReq1.Notifier = notifier
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Wait for the notifier to be installed.
require.Eventually(t, func() bool {
batch := tryGetOnlyBatch(ctx, batcher)
if batch == nil {
return false
}
batch = batch.snapshot(ctx)
sweep := batch.sweeps[batch.primarySweepID]
return sweep.notifier != nil &&
sweep.notifier.SpendChan == spendChan
}, test.Timeout, eventuallyCheckFrequency)
// Create the spending tx that will trigger the spend monitor of the
// batch.
spendingTx := &wire.MsgTx{
@ -861,6 +924,7 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
},
TxOut: []*wire.TxOut{
{
Value: outputValue,
PkScript: []byte{3, 2, 1},
},
},
@ -879,6 +943,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
// We notify the spend.
lnd.SpendChannel <- spendDetail
// Make sure the notifier got a proper spending notification.
spending := <-spendChan
require.Equal(t, spendingTxHash, spending.Tx.TxHash())
require.Equal(t, btcutil.Amount(fee), spending.OnChainFeePortion)
// After receiving the spend, the batch is now monitoring for confs.
confReg := <-lnd.RegisterConfChannel
@ -889,7 +958,90 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
// The batch should eventually read the spend notification and progress
// its state to closed.
require.Eventually(t, func() bool {
batch := batch.snapshot(ctx)
batch := tryGetOnlyBatch(ctx, batcher)
if batch == nil {
return false
}
batch = batch.snapshot(ctx)
return batch.state == Closed
}, test.Timeout, eventuallyCheckFrequency)
// Emulate a confirmation error.
confReg.ErrChan <- testError
// Make sure the notifier gets the confirmation error.
confErr := <-confErrChan
require.ErrorIs(t, confErr, testError)
// Wait for the batcher to crash because of the confirmation error.
runErr = <-runErrChan
require.ErrorIs(t, runErr, testError)
// Now launch the batcher again.
batcher = NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
batcherStore, sweepStore)
go func() {
runErrChan <- batcher.Run(ctx)
}()
// When batch is successfully created it will execute it's first step,
// which leads to a spend monitor of the primary sweep.
<-lnd.RegisterSpendChannel
// Deliver sweep request to batcher.
spendChan = make(chan *SpendDetail, 1)
confChan := make(chan *ConfDetail)
notifier = &SpendNotifier{
SpendChan: spendChan,
SpendErrChan: make(chan error, 1),
ConfChan: confChan,
QuitChan: make(chan bool, 1),
}
sweepReq1.Notifier = notifier
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Wait for tx to be published. A closed batch is stored in DB as Open.
<-lnd.TxPublishChannel
// Wait for the notifier to be installed.
require.Eventually(t, func() bool {
batch := tryGetOnlyBatch(ctx, batcher)
if batch == nil {
return false
}
batch = batch.snapshot(ctx)
sweep := batch.sweeps[batch.primarySweepID]
return sweep.notifier != nil &&
sweep.notifier.SpendChan == spendChan
}, test.Timeout, eventuallyCheckFrequency)
// We notify the spend.
lnd.SpendChannel <- spendDetail
// Make sure the notifier got a proper spending notification.
spending = <-spendChan
require.Equal(t, spendingTxHash, spending.Tx.TxHash())
require.Equal(t, btcutil.Amount(fee), spending.OnChainFeePortion)
// After receiving the spend, the batch is now monitoring for confs.
confReg = <-lnd.RegisterConfChannel
// Make sure the confirmation has proper height hint. It should pass
// the swap initiation height, not the current height.
require.Equal(t, int32(initiationHeight), confReg.HeightHint)
// The batch should eventually read the spend notification and progress
// its state to closed.
require.Eventually(t, func() bool {
batch := tryGetOnlyBatch(ctx, batcher)
if batch == nil {
return false
}
batch = batch.snapshot(ctx)
return batch.state == Closed
}, test.Timeout, eventuallyCheckFrequency)
@ -899,14 +1051,125 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
// We mock the tx confirmation notification.
lnd.ConfChannel <- &chainntnfs.TxConfirmation{
Tx: spendingTx,
BlockHeight: 604,
Tx: spendingTx,
}
// Make sure the notifier gets a confirmation notification.
conf := <-confChan
require.Equal(t, uint32(604), conf.BlockHeight)
require.Equal(t, spendingTx.TxHash(), conf.Tx.TxHash())
require.Equal(t, btcutil.Amount(fee), conf.OnChainFeePortion)
// Eventually the batch receives the confirmation notification and
// confirms itself.
require.Eventually(t, func() bool {
batch := tryGetOnlyBatch(ctx, batcher)
if batch == nil {
return false
}
return batch.isComplete()
}, test.Timeout, eventuallyCheckFrequency)
// Now emulate adding the sweep again after it was fully confirmed.
// This triggers another code path (monitorSpendAndNotify).
spendChan = make(chan *SpendDetail, 1)
confChan = make(chan *ConfDetail)
notifier = &SpendNotifier{
SpendChan: spendChan,
SpendErrChan: make(chan error, 1),
ConfChan: confChan,
QuitChan: make(chan bool, 1),
}
sweepReq1.Notifier = notifier
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Expect a spending registration.
<-lnd.RegisterSpendChannel
// We notify the spend.
lnd.SpendChannel <- spendDetail
// Now expect the notifier to produce the spending details.
spending = <-spendChan
require.Equal(t, spendingTxHash, spending.Tx.TxHash())
require.Equal(t, btcutil.Amount(fee), spending.OnChainFeePortion)
// We mock the tx confirmation notification.
<-lnd.RegisterConfChannel
lnd.ConfChannel <- &chainntnfs.TxConfirmation{
BlockHeight: 604,
Tx: spendingTx,
}
// Make sure the notifier gets a confirmation notification.
conf = <-confChan
require.Equal(t, uint32(604), conf.BlockHeight)
require.Equal(t, spendingTx.TxHash(), conf.Tx.TxHash())
require.Equal(t, btcutil.Amount(fee), conf.OnChainFeePortion)
// Now check what happens in case of a spending error.
spendErrChan = make(chan error, 1)
notifier = &SpendNotifier{
SpendChan: make(chan *SpendDetail, 1),
SpendErrChan: spendErrChan,
QuitChan: make(chan bool, 1),
}
sweepReq1.Notifier = notifier
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Expect a spending registration.
spendReg = <-lnd.RegisterSpendChannel
// Emulate spend error.
spendReg.ErrChan <- testError
// Make sure the caller of AddSweep got the spending error.
notifierErr = <-spendErrChan
require.Error(t, notifierErr)
require.ErrorIs(t, notifierErr, testError)
// Wait for the batcher to crash because of the spending error.
runErr = <-runErrChan
require.ErrorIs(t, runErr, testError)
// Now launch the batcher again.
batcher = NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
batcherStore, sweepStore)
go func() {
runErrChan <- batcher.Run(ctx)
}()
// Now check what happens in case of a confirmation error.
confErrChan = make(chan error, 1)
notifier = &SpendNotifier{
SpendChan: make(chan *SpendDetail, 1),
SpendErrChan: make(chan error, 1),
ConfErrChan: confErrChan,
QuitChan: make(chan bool, 1),
}
sweepReq1.Notifier = notifier
require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
// Expect a spending registration.
<-lnd.RegisterSpendChannel
// We notify the spend.
lnd.SpendChannel <- spendDetail
// We mock the tx confirmation error notification.
confReg = <-lnd.RegisterConfChannel
confReg.ErrChan <- testError
// Make sure the notifier gets the confirmation error.
confErr = <-confErrChan
require.ErrorIs(t, confErr, testError)
// Wait for the batcher to crash because of the confirmation error.
runErr = <-runErrChan
require.ErrorIs(t, runErr, testError)
}
// wrappedLogger implements btclog.Logger, recording last debug message format.

26
test/chainnotifier_mock.go
View file

@ -36,6 +36,7 @@ type SpendRegistration struct {
Outpoint *wire.OutPoint
PkScript []byte
HeightHint int32
ErrChan chan<- error
}
// ConfRegistration contains registration details.
@ -45,18 +46,24 @@ type ConfRegistration struct {
HeightHint int32
NumConfs int32
ConfChan chan *chainntnfs.TxConfirmation
ErrChan chan<- error
}
func (c *mockChainNotifier) RegisterSpendNtfn(ctx context.Context,
outpoint *wire.OutPoint, pkScript []byte, heightHint int32) (
chan *chainntnfs.SpendDetail, chan error, error) {
c.lnd.RegisterSpendChannel <- &SpendRegistration{
spendErrChan := make(chan error, 1)
reg := &SpendRegistration{
HeightHint: heightHint,
Outpoint: outpoint,
PkScript: pkScript,
ErrChan: spendErrChan,
}
c.lnd.RegisterSpendChannel <- reg
spendChan := make(chan *chainntnfs.SpendDetail, 1)
errChan := make(chan error, 1)
@ -70,6 +77,13 @@ func (c *mockChainNotifier) RegisterSpendNtfn(ctx context.Context,
case spendChan <- m:
case <-ctx.Done():
}
case err := <-spendErrChan:
select {
case errChan <- err:
case <-ctx.Done():
}
case <-ctx.Done():
}
}()
@ -129,12 +143,15 @@ func (c *mockChainNotifier) RegisterConfirmationsNtfn(ctx context.Context,
opts ...lndclient.NotifierOption) (chan *chainntnfs.TxConfirmation,
chan error, error) {
confErrChan := make(chan error, 1)
reg := &ConfRegistration{
PkScript: pkScript,
TxID: txid,
HeightHint: heightHint,
NumConfs: numConfs,
ConfChan: make(chan *chainntnfs.TxConfirmation, 1),
ErrChan: confErrChan,
}
c.Lock()
@ -169,6 +186,13 @@ func (c *mockChainNotifier) RegisterConfirmationsNtfn(ctx context.Context,
}
}
c.Unlock()
case err := <-confErrChan:
select {
case errChan <- err:
case <-ctx.Done():
}
case <-ctx.Done():
}
}()

10
test/context.go
View file

@ -61,7 +61,7 @@ func (ctx *Context) NotifySpend(tx *wire.MsgTx, inputIndex uint32) {
SpenderInputIndex: inputIndex,
}:
case <-time.After(Timeout):
ctx.T.Fatalf("htlc spend not consumed")
ctx.T.Fatalf("spend not consumed")
}
}
@ -74,7 +74,7 @@ func (ctx *Context) NotifyConf(tx *wire.MsgTx) {
Tx: tx,
}:
case <-time.After(Timeout):
ctx.T.Fatalf("htlc spend not consumed")
ctx.T.Fatalf("confirmation not consumed")
}
}
@ -86,7 +86,7 @@ func (ctx *Context) AssertRegisterSpendNtfn(script []byte) {
case spendIntent := <-ctx.Lnd.RegisterSpendChannel:
require.Equal(
ctx.T, script, spendIntent.PkScript,
"server not listening for published htlc script",
"server not listening for published script",
)
case <-time.After(Timeout):
@ -134,7 +134,7 @@ func (ctx *Context) AssertRegisterConf(expectTxHash bool, confs int32) *ConfRegi
require.Equal(ctx.T, confs, confIntent.NumConfs)
case <-time.After(Timeout):
ctx.T.Fatalf("htlc confirmed not subscribed to")
ctx.T.Fatalf("tx confirmed not subscribed to")
}
return confIntent
@ -249,7 +249,7 @@ func (ctx *Context) GetOutputIndex(tx *wire.MsgTx,
}
}
ctx.T.Fatal("htlc not present in tx")
ctx.T.Fatal("the output not present in tx")
return 0
}