staticaddr: address and deposit adjustments for withdrawals

staticaddr/address/manager.go

@@ -25,10 +25,6 @@ var (
 	log btclog.Logger
 )
 
-func init() {
-	log = staticaddr.GetLogger()
-}
-
 // ManagerConfig holds the configuration for the address manager.
 type ManagerConfig struct {
 	// AddressClient is the client that communicates with the loop server
@@ -60,6 +56,7 @@ type Manager struct {
 
 // NewManager creates a new address manager.
 func NewManager(cfg *ManagerConfig) *Manager {
+	log = staticaddr.GetLogger()
 	return &Manager{
 		cfg: cfg,
 	}

staticaddr/address/manager_test.go

@@ -32,6 +32,15 @@ type mockStaticAddressClient struct {
 	mock.Mock
 }
 
+func (m *mockStaticAddressClient) ServerWithdrawDeposits(ctx context.Context,
+	in *swapserverrpc.ServerWithdrawRequest,
+	opts ...grpc.CallOption) (*swapserverrpc.ServerWithdrawResponse,
+	error) {
+
+	// TODO implement me
+	panic("implement me")
+}
+
 func (m *mockStaticAddressClient) ServerNewAddress(ctx context.Context,
 	in *swapserverrpc.ServerNewAddressRequest, opts ...grpc.CallOption) (
 	*swapserverrpc.ServerNewAddressResponse, error) {

staticaddr/deposit/actions.go

@@ -146,3 +146,17 @@ func (f *FSM) SweptExpiredDepositAction(_ fsm.EventContext) fsm.EventType {
 
 	return fsm.NoOp
 }
+
+// WithdrawnDepositAction is the final action after a withdrawal. It signals to
+// the manager that the deposit has been swept and the FSM can be removed.
+func (f *FSM) WithdrawnDepositAction(_ fsm.EventContext) fsm.EventType {
+	select {
+	case <-f.ctx.Done():
+		return fsm.OnError
+
+	default:
+		f.finalizedDepositChan <- f.deposit.OutPoint
+	}
+
+	return fsm.NoOp
+}

staticaddr/deposit/deposit.go

@@ -35,7 +35,7 @@ type Deposit struct {
 	// state is the current state of the deposit.
 	state fsm.StateType
 
-	// The outpoint of the deposit.
+	// Outpoint of the deposit.
 	wire.OutPoint
 
 	// Value is the amount of the deposit.
@@ -52,6 +52,10 @@ type Deposit struct {
 	// ExpirySweepTxid is the transaction id of the expiry sweep.
 	ExpirySweepTxid chainhash.Hash
 
+	// WithdrawalSweepAddress is the address that is used to
+	// cooperatively sweep the deposit to before it is expired.
+	WithdrawalSweepAddress string
+
 	sync.Mutex
 }
 
@@ -68,7 +72,7 @@ func (d *Deposit) IsInFinalState() bool {
 	d.Lock()
 	defer d.Unlock()
 
-	return d.state == Expired || d.state == Failed
+	return d.state == Expired || d.state == Withdrawn || d.state == Failed
 }
 
 func (d *Deposit) IsExpired(currentHeight, expiry uint32) bool {

staticaddr/deposit/fsm.go

@@ -29,6 +29,10 @@ var (
 var (
 	Deposited = fsm.StateType("Deposited")
 
+	Withdrawing = fsm.StateType("Withdrawing")
+
+	Withdrawn = fsm.StateType("Withdrawn")
+
 	PublishExpiredDeposit = fsm.StateType("PublishExpiredDeposit")
 
 	WaitForExpirySweep = fsm.StateType("WaitForExpirySweep")
@@ -41,6 +45,8 @@ var (
 // Events.
 var (
 	OnStart           = fsm.EventType("OnStart")
+	OnWithdraw        = fsm.EventType("OnWithdraw")
+	OnWithdrawn       = fsm.EventType("OnWithdrawn")
 	OnExpiry          = fsm.EventType("OnExpiry")
 	OnExpiryPublished = fsm.EventType("OnExpiryPublished")
 	OnExpirySwept     = fsm.EventType("OnExpirySwept")
@@ -118,13 +124,7 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
 		for {
 			select {
 			case currentHeight := <-depoFsm.blockNtfnChan:
-				err := depoFsm.handleBlockNotification(
-					currentHeight,
-				)
-				if err != nil {
-					log.Errorf("error handling block "+
-						"notification: %v", err)
-				}
+				depoFsm.handleBlockNotification(currentHeight)
 
 			case <-ctx.Done():
 				return
@@ -138,15 +138,10 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
 // handleBlockNotification inspects the current block height and sends the
 // OnExpiry event to publish the expiry sweep transaction if the deposit timed
 // out, or it republishes the expiry sweep transaction if it was not yet swept.
-func (f *FSM) handleBlockNotification(currentHeight uint32) error {
-	params, err := f.cfg.AddressManager.GetStaticAddressParameters(f.ctx)
-	if err != nil {
-		return err
-	}
-
+func (f *FSM) handleBlockNotification(currentHeight uint32) {
 	// If the deposit is expired but not yet sufficiently confirmed, we
 	// republish the expiry sweep transaction.
-	if f.deposit.IsExpired(currentHeight, params.Expiry) {
+	if f.deposit.IsExpired(currentHeight, f.params.Expiry) {
 		if f.deposit.IsInState(WaitForExpirySweep) {
 			f.PublishDepositExpirySweepAction(nil)
 		} else {
@@ -159,8 +154,6 @@ func (f *FSM) handleBlockNotification(currentHeight uint32) error {
 			}()
 		}
 	}
-
-	return nil
 }
 
 // DepositStatesV0 returns the states a deposit can be in.
@@ -174,8 +167,9 @@ func (f *FSM) DepositStatesV0() fsm.States {
 		},
 		Deposited: fsm.State{
 			Transitions: fsm.Transitions{
-				OnExpiry:  PublishExpiredDeposit,
-				OnRecover: Deposited,
+				OnExpiry:   PublishExpiredDeposit,
+				OnWithdraw: Withdrawing,
+				OnRecover:  Deposited,
 			},
 			Action: fsm.NoOpAction,
 		},
@@ -210,6 +204,36 @@ func (f *FSM) DepositStatesV0() fsm.States {
 			},
 			Action: f.SweptExpiredDepositAction,
 		},
+		Withdrawing: fsm.State{
+			Transitions: fsm.Transitions{
+				OnWithdrawn: Withdrawn,
+				// Upon recovery, we go back to the Deposited
+				// state. The deposit by then has a withdrawal
+				// address stamped to it which will cause it to
+				// transition into the Withdrawing state again.
+				OnRecover: Deposited,
+
+				// A precondition for the Withdrawing state is
+				// that the withdrawal transaction has been
+				// broadcast. If the deposit expires while the
+				// withdrawal isn't confirmed, we can ignore the
+				// expiry.
+				OnExpiry: Withdrawing,
+
+				// If the withdrawal failed we go back to
+				// Deposited, hoping that another withdrawal
+				// attempt will be successful. Alternatively,
+				// the client can wait for the timeout sweep.
+				fsm.OnError: Deposited,
+			},
+			Action: fsm.NoOpAction,
+		},
+		Withdrawn: fsm.State{
+			Transitions: fsm.Transitions{
+				OnExpiry: Expired,
+			},
+			Action: f.WithdrawnDepositAction,
+		},
 		Failed: fsm.State{
 			Transitions: fsm.Transitions{
 				OnExpiry: Failed,

staticaddr/deposit/manager.go

@@ -3,6 +3,7 @@ package deposit
 import (
 	"context"
 	"fmt"
+	"sort"
 	"sync"
 	"time"
 
@@ -12,6 +13,7 @@ import (
 	"github.com/btcsuite/btclog"
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop"
+	"github.com/lightninglabs/loop/fsm"
 	"github.com/lightninglabs/loop/staticaddr"
 	staticaddressrpc "github.com/lightninglabs/loop/swapserverrpc"
 	"github.com/lightningnetwork/lnd/lnrpc/walletrpc"
@@ -37,10 +39,6 @@ var (
 	log btclog.Logger
 )
 
-func init() {
-	log = staticaddr.GetLogger()
-}
-
 // ManagerConfig holds the configuration for the address manager.
 type ManagerConfig struct {
 	// AddressClient is the client that communicates with the loop server
@@ -109,6 +107,7 @@ type Manager struct {
 
 // NewManager creates a new deposit manager.
 func NewManager(cfg *ManagerConfig) *Manager {
+	log = staticaddr.GetLogger()
 	return &Manager{
 		cfg:                  cfg,
 		initChan:             make(chan struct{}),
@@ -430,3 +429,88 @@ func (m *Manager) finalizeDeposit(outpoint wire.OutPoint) {
 	delete(m.deposits, outpoint)
 	m.Unlock()
 }
+
+// GetActiveDepositsInState returns all active deposits.
+func (m *Manager) GetActiveDepositsInState(stateFilter fsm.StateType) (
+	[]*Deposit, error) {
+
+	m.Lock()
+	defer m.Unlock()
+
+	var deposits []*Deposit
+	for _, fsm := range m.activeDeposits {
+		if fsm.deposit.GetState() != stateFilter {
+			continue
+		}
+		deposits = append(deposits, fsm.deposit)
+	}
+
+	sort.Slice(deposits, func(i, j int) bool {
+		return deposits[i].ConfirmationHeight <
+			deposits[j].ConfirmationHeight
+	})
+
+	return deposits, nil
+}
+
+// GetAllDeposits returns all active deposits.
+func (m *Manager) GetAllDeposits() ([]*Deposit, error) {
+	return m.cfg.Store.AllDeposits(m.runCtx)
+}
+
+// AllOutpointsActiveDeposits checks if all deposits referenced by the outpoints
+// are active and in the specified state.
+func (m *Manager) AllOutpointsActiveDeposits(outpoints []wire.OutPoint,
+	stateFilter fsm.StateType) ([]*Deposit, bool) {
+
+	m.Lock()
+	defer m.Unlock()
+
+	deposits := make([]*Deposit, 0, len(outpoints))
+	for _, o := range outpoints {
+		if _, ok := m.activeDeposits[o]; !ok {
+			return nil, false
+		}
+
+		deposit := m.deposits[o]
+		if deposit.GetState() != stateFilter {
+			return nil, false
+		}
+
+		deposits = append(deposits, m.deposits[o])
+	}
+
+	return deposits, true
+}
+
+// TransitionDeposits allows a caller to transition a set of deposits to a new
+// state.
+func (m *Manager) TransitionDeposits(deposits []*Deposit, event fsm.EventType,
+	expectedFinalState fsm.StateType) error {
+
+	for _, d := range deposits {
+		m.Lock()
+		sm, ok := m.activeDeposits[d.OutPoint]
+		m.Unlock()
+		if !ok {
+			return fmt.Errorf("deposit not found")
+		}
+
+		err := sm.SendEvent(event, nil)
+		if err != nil {
+			return err
+		}
+		err = sm.DefaultObserver.WaitForState(
+			m.runCtx, time.Minute, expectedFinalState,
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (m *Manager) UpdateDeposit(d *Deposit) error {
+	return m.cfg.Store.UpdateDeposit(m.runCtx, d)
+}

staticaddr/deposit/manager_test.go

@@ -49,6 +49,15 @@ type mockStaticAddressClient struct {
 	mock.Mock
 }
 
+func (m *mockStaticAddressClient) ServerWithdrawDeposits(ctx context.Context,
+	in *swapserverrpc.ServerWithdrawRequest,
+	opts ...grpc.CallOption) (*swapserverrpc.ServerWithdrawResponse,
+	error) {
+
+	// TODO implement me
+	panic("implement me")
+}
+
 func (m *mockStaticAddressClient) ServerNewAddress(ctx context.Context,
 	in *swapserverrpc.ServerNewAddressRequest, opts ...grpc.CallOption) (
 	*swapserverrpc.ServerNewAddressResponse, error) {

staticaddr/deposit/sql_store.go

@@ -39,6 +39,10 @@ func (s *SqlStore) CreateDeposit(ctx context.Context, deposit *Deposit) error {
 		Amount:               int64(deposit.Value),
 		ConfirmationHeight:   deposit.ConfirmationHeight,
 		TimeoutSweepPkScript: deposit.TimeOutSweepPkScript,
+		WithdrawalSweepAddress: sql.NullString{
+			String: deposit.WithdrawalSweepAddress,
+			Valid:  deposit.WithdrawalSweepAddress != "",
+		},
 	}
 
 	updateArgs := sqlc.InsertDepositUpdateParams{
@@ -74,7 +78,6 @@ func (s *SqlStore) UpdateDeposit(ctx context.Context, deposit *Deposit) error {
 		}
 		confirmationHeight = sql.NullInt64{
 			Int64: deposit.ConfirmationHeight,
-			Valid: deposit.ConfirmationHeight != 0,
 		}
 	)
 
@@ -84,6 +87,10 @@ func (s *SqlStore) UpdateDeposit(ctx context.Context, deposit *Deposit) error {
 		OutIndex:           outIndex.Int32,
 		ConfirmationHeight: confirmationHeight.Int64,
 		ExpirySweepTxid:    deposit.ExpirySweepTxid[:],
+		WithdrawalSweepAddress: sql.NullString{
+			String: deposit.WithdrawalSweepAddress,
+			Valid:  deposit.WithdrawalSweepAddress != "",
+		},
 	}
 
 	return s.baseDB.ExecTx(ctx, &loopdb.SqliteTxOptions{},
@@ -200,10 +207,11 @@ func (s *SqlStore) toDeposit(row sqlc.Deposit,
 			Hash:  *txHash,
 			Index: uint32(row.OutIndex),
 		},
-		Value:                btcutil.Amount(row.Amount),
-		ConfirmationHeight:   row.ConfirmationHeight,
-		TimeOutSweepPkScript: row.TimeoutSweepPkScript,
-		ExpirySweepTxid:      expirySweepTxid,
+		Value:                  btcutil.Amount(row.Amount),
+		ConfirmationHeight:     row.ConfirmationHeight,
+		TimeOutSweepPkScript:   row.TimeoutSweepPkScript,
+		ExpirySweepTxid:        expirySweepTxid,
+		WithdrawalSweepAddress: row.WithdrawalSweepAddress.String,
 	}, nil
 }