staticaddr: listen for spents on withdrawals

Author: hieblmi

staticaddr/deposit/fsm.go

@@ -361,7 +361,8 @@ func (f *FSM) DepositStatesV0() fsm.States {
 		},
 		Withdrawn: fsm.State{
 			Transitions: fsm.Transitions{
-				OnExpiry: Expired,
+				OnExpiry:    Expired,
+				OnWithdrawn: Withdrawn,
 			},
 			Action: f.FinalizeDepositAction,
 		},

staticaddr/withdraw/manager.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"reflect"
 	"strings"
+	"sync"
 	"sync/atomic"
 
 	"github.com/btcsuite/btcd/btcec/v2/schnorr"
@@ -93,6 +94,9 @@ type newWithdrawalResponse struct {
 type Manager struct {
 	cfg *ManagerConfig
 
+	// mu protects access to finalizedWithdrawalTxns.
+	mu sync.Mutex
+
 	// initChan signals the daemon that the withdrawal manager has completed
 	// its initialization.
 	initChan chan struct{}
@@ -113,25 +117,17 @@ type Manager struct {
 	// finalizedWithdrawalTx are the finalized withdrawal transactions that
 	// are published to the network and re-published on block arrivals.
 	finalizedWithdrawalTxns map[chainhash.Hash]*wire.MsgTx
-
-	// withdrawalHandlerQuitChans is a map of quit channels for each
-	// withdrawal transaction. The quit channels are used to stop the
-	// withdrawal handler for a specific withdrawal transaction, e.g. if
-	// a new rbf'd transaction has to be monitored for confirmation in
-	// favor of the previous one.
-	withdrawalHandlerQuitChans map[chainhash.Hash]chan struct{}
 }
 
 // NewManager creates a new deposit withdrawal manager.
 func NewManager(cfg *ManagerConfig, currentHeight uint32) *Manager {
 	m := &Manager{
-		cfg:                        cfg,
-		initChan:                   make(chan struct{}),
-		finalizedWithdrawalTxns:    make(map[chainhash.Hash]*wire.MsgTx),
-		exitChan:                   make(chan struct{}),
-		newWithdrawalRequestChan:   make(chan newWithdrawalRequest),
-		errChan:                    make(chan error),
-		withdrawalHandlerQuitChans: make(map[chainhash.Hash]chan struct{}),
+		cfg:                      cfg,
+		initChan:                 make(chan struct{}),
+		finalizedWithdrawalTxns:  make(map[chainhash.Hash]*wire.MsgTx),
+		exitChan:                 make(chan struct{}),
+		newWithdrawalRequestChan: make(chan newWithdrawalRequest),
+		errChan:                  make(chan error),
 	}
 	m.initiationHeight.Store(currentHeight)
 
@@ -251,14 +247,14 @@ func (m *Manager) recoverWithdrawals(ctx context.Context) error {
 			return err
 		}
 
-		err = m.handleWithdrawal(
-			ctx, deposits, tx.TxHash(), tx.TxOut[0].PkScript,
-		)
+		err = m.handleWithdrawal(ctx, deposits, tx.TxHash())
 		if err != nil {
 			return err
 		}
 
+		m.mu.Lock()
 		m.finalizedWithdrawalTxns[tx.TxHash()] = tx
+		m.mu.Unlock()
 	}
 
 	return nil
@@ -316,6 +312,30 @@ func (m *Manager) WithdrawDeposits(ctx context.Context,
 					"previous withdrawal tx hash")
 			}
 		}
+
+		// We also avoid that the user selects a subset of previously
+		// clustered deposits for a fee bump. This would result in a
+		// different transaction shape.
+		allDeposits, err := m.cfg.DepositManager.GetActiveDepositsInState(
+			deposit.Withdrawing,
+		)
+		if err != nil {
+			return "", "", err
+		}
+
+		allDepositsWithHash := make(map[chainhash.Hash][]*deposit.Deposit)
+		for _, d := range allDeposits {
+			if d.FinalizedWithdrawalTx.TxHash() == hash {
+				allDepositsWithHash[hash] = append(
+					allDepositsWithHash[hash], d,
+				)
+			}
+		}
+
+		if len(allDepositsWithHash[hash]) != len(deposits) {
+			return "", "", fmt.Errorf("can't bump fee for subset " +
+				"of clustered deposits")
+		}
 	}
 
 	var (
@@ -358,30 +378,24 @@ func (m *Manager) WithdrawDeposits(ctx context.Context,
 		return "", "", nil
 	}
 
-	withdrawalPkScript, err := txscript.PayToAddrScript(withdrawalAddress)
-	if err != nil {
-		return "", "", err
-	}
-
-	err = m.handleWithdrawal(
-		ctx, deposits, finalizedTx.TxHash(), withdrawalPkScript,
-	)
+	err = m.handleWithdrawal(ctx, deposits, finalizedTx.TxHash())
 	if err != nil {
 		return "", "", err
 	}
 
 	// If a previous withdrawal existed across the selected deposits, and
-	// it isn't the same as the new withdrawal, we'll stop monitoring the
-	// previous withdrawal and remove it from the finalized withdrawals.
+	// it isn't the same as the new withdrawal, we remove it from the
+	// finalized withdrawals to stop republishing it, but we keep the
+	// goroutine in handleWithdrawal running to monitor the potential
+	// confirmation of the previous withdrawal.
 	deposits[0].Lock()
 	prevTx := deposits[0].FinalizedWithdrawalTx
 	deposits[0].Unlock()
 
 	if prevTx != nil && prevTx.TxHash() != finalizedTx.TxHash() {
-		quitChan := m.withdrawalHandlerQuitChans[prevTx.TxHash()]
-		close(quitChan)
-		delete(m.withdrawalHandlerQuitChans, prevTx.TxHash())
+		m.mu.Lock()
 		delete(m.finalizedWithdrawalTxns, prevTx.TxHash())
+		m.mu.Unlock()
 	}
 
 	// Attach the finalized withdrawal tx to the deposits. After a client
@@ -394,7 +408,9 @@ func (m *Manager) WithdrawDeposits(ctx context.Context,
 		d.Unlock()
 	}
 
+	m.mu.Lock()
 	m.finalizedWithdrawalTxns[finalizedTx.TxHash()] = finalizedTx
+	m.mu.Unlock()
 
 	// Transition the deposits to the withdrawing state. This updates each
 	// deposits withdrawal address. If a transition fails, we'll return an
@@ -540,27 +556,16 @@ func (m *Manager) publishFinalizedWithdrawalTx(ctx context.Context,
 }
 
 func (m *Manager) handleWithdrawal(ctx context.Context,
-	deposits []*deposit.Deposit, txHash chainhash.Hash,
-	withdrawalPkScript []byte) error {
+	deposits []*deposit.Deposit, txHash chainhash.Hash) error {
 
-	confChan, errChan, err := m.cfg.ChainNotifier.RegisterConfirmationsNtfn(
-		ctx, &txHash, withdrawalPkScript, MinConfs,
-		int32(m.initiationHeight.Load()),
+	d := deposits[0]
+	spentChan, errChan, err := m.cfg.ChainNotifier.RegisterSpendNtfn(
+		ctx, &d.OutPoint, nil, int32(d.ConfirmationHeight),
 	)
-	if err != nil {
-		return err
-	}
-
-	// Create a new quit chan for this set of deposits under the same
-	// withdrawal tx hash. If a new withdrawal is requested the quit chan
-	// is closed in favor of a new one, to start monitoring the new
-	// withdrawal transaction.
-	m.withdrawalHandlerQuitChans[txHash] = make(chan struct{})
-	quitChan := m.withdrawalHandlerQuitChans[txHash]
 
 	go func() {
 		select {
-		case <-confChan:
+		case <-spentChan:
 			err = m.cfg.DepositManager.TransitionDeposits(
 				ctx, deposits, deposit.OnWithdrawn,
 				deposit.Withdrawn,
@@ -570,16 +575,11 @@ func (m *Manager) handleWithdrawal(ctx context.Context,
 					err)
 			}
 
-			// Remove the withdrawal from the active withdrawals and
-			// remove its finalized to stop republishing it on block
-			// arrivals.
+			// Remove the withdrawal tx from the active withdrawals
+			// to stop republishing it on block arrivals.
+			m.mu.Lock()
 			delete(m.finalizedWithdrawalTxns, txHash)
-
-		case <-quitChan:
-			log.Debugf("Exiting withdrawal handler for tx %v",
-				txHash)
-
-			return
+			m.mu.Unlock()
 
 		case err := <-errChan:
 			log.Errorf("Error waiting for confirmation: %v", err)
@@ -987,7 +987,14 @@ func (m *Manager) toPrevOuts(deposits []*deposit.Deposit,
 }
 
 func (m *Manager) republishWithdrawals(ctx context.Context) error {
-	for _, finalizedTx := range m.finalizedWithdrawalTxns {
+	m.mu.Lock()
+	txns := make([]*wire.MsgTx, 0, len(m.finalizedWithdrawalTxns))
+	for _, tx := range m.finalizedWithdrawalTxns {
+		txns = append(txns, tx)
+	}
+	m.mu.Unlock()
+
+	for _, finalizedTx := range txns {
 		if finalizedTx == nil {
 			log.Warnf("Finalized withdrawal tx is nil")
 			continue