diff --git a/staticaddr/address/manager.go b/staticaddr/address/manager.go index 42b5d639..6420db97 100644 --- a/staticaddr/address/manager.go +++ b/staticaddr/address/manager.go @@ -11,7 +11,6 @@ import ( "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/chaincfg" "github.com/lightninglabs/lndclient" - "github.com/lightninglabs/loop/staticaddr" "github.com/lightninglabs/loop/staticaddr/script" "github.com/lightninglabs/loop/staticaddr/version" "github.com/lightninglabs/loop/swap" diff --git a/staticaddr/address/manager_test.go b/staticaddr/address/manager_test.go index 8e15732d..9548640d 100644 --- a/staticaddr/address/manager_test.go +++ b/staticaddr/address/manager_test.go @@ -32,6 +32,17 @@ type mockStaticAddressClient struct { mock.Mock } +func (m *mockStaticAddressClient) ServerWithdrawDeposits(ctx context.Context, + in *swapserverrpc.ServerWithdrawRequest, + opts ...grpc.CallOption) (*swapserverrpc.ServerWithdrawResponse, + error) { + + args := m.Called(ctx, in, opts) + + return args.Get(0).(*swapserverrpc.ServerWithdrawResponse), + args.Error(1) +} + func (m *mockStaticAddressClient) ServerNewAddress(ctx context.Context, in *swapserverrpc.ServerNewAddressRequest, opts ...grpc.CallOption) ( *swapserverrpc.ServerNewAddressResponse, error) { diff --git a/staticaddr/deposit/actions.go b/staticaddr/deposit/actions.go index 91b54c55..835a9607 100644 --- a/staticaddr/deposit/actions.go +++ b/staticaddr/deposit/actions.go @@ -154,3 +154,19 @@ func (f *FSM) SweptExpiredDepositAction(ctx context.Context, 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(ctx context.Context, + _ fsm.EventContext) fsm.EventType { + + select { + case <-ctx.Done(): + return fsm.OnError + + default: + f.finalizedDepositChan <- f.deposit.OutPoint + } + + return fsm.NoOp +} diff --git a/staticaddr/deposit/deposit.go b/staticaddr/deposit/deposit.go index 8c119991..207ad440 100644 --- a/staticaddr/deposit/deposit.go +++ b/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 + // FinalizedWithdrawalTx is the coop signed withdrawal transaction. It + // is republished on new block arrivals and on client restarts. + FinalizedWithdrawalTx *wire.MsgTx + 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 { diff --git a/staticaddr/deposit/fsm.go b/staticaddr/deposit/fsm.go index 91bc951d..347ad5df 100644 --- a/staticaddr/deposit/fsm.go +++ b/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") @@ -40,11 +44,13 @@ var ( // Events. var ( - OnStart = fsm.EventType("OnStart") - OnExpiry = fsm.EventType("OnExpiry") - OnExpiryPublished = fsm.EventType("OnExpiryPublished") - OnExpirySwept = fsm.EventType("OnExpirySwept") - OnRecover = fsm.EventType("OnRecover") + OnStart = fsm.EventType("OnStart") + OnWithdrawInitiated = fsm.EventType("OnWithdrawInitiated") + OnWithdrawn = fsm.EventType("OnWithdrawn") + OnExpiry = fsm.EventType("OnExpiry") + OnExpiryPublished = fsm.EventType("OnExpiryPublished") + OnExpirySwept = fsm.EventType("OnExpirySwept") + OnRecover = fsm.EventType("OnRecover") ) // FSM is the state machine that handles the instant out. @@ -115,13 +121,9 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig, for { select { case currentHeight := <-depoFsm.blockNtfnChan: - err := depoFsm.handleBlockNotification( + depoFsm.handleBlockNotification( ctx, currentHeight, ) - if err != nil { - log.Errorf("error handling block "+ - "notification: %v", err) - } case <-ctx.Done(): return @@ -136,16 +138,11 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig, // 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(ctx context.Context, - currentHeight uint32) error { - - params, err := f.cfg.AddressManager.GetStaticAddressParameters(ctx) - if err != nil { - return err - } + 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(ctx, nil) } else { @@ -158,8 +155,6 @@ func (f *FSM) handleBlockNotification(ctx context.Context, }() } } - - return nil } // DepositStatesV0 returns the states a deposit can be in. @@ -173,8 +168,9 @@ func (f *FSM) DepositStatesV0() fsm.States { }, Deposited: fsm.State{ Transitions: fsm.Transitions{ - OnExpiry: PublishExpiredDeposit, - OnRecover: Deposited, + OnExpiry: PublishExpiredDeposit, + OnWithdrawInitiated: Withdrawing, + OnRecover: Deposited, }, Action: fsm.NoOpAction, }, @@ -209,6 +205,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, diff --git a/staticaddr/deposit/manager.go b/staticaddr/deposit/manager.go index 9a08d0b0..329ca0f0 100644 --- a/staticaddr/deposit/manager.go +++ b/staticaddr/deposit/manager.go @@ -3,6 +3,7 @@ package deposit import ( "context" "fmt" + "sort" "sync" "time" @@ -11,6 +12,7 @@ import ( "github.com/btcsuite/btcd/wire" "github.com/lightninglabs/lndclient" "github.com/lightninglabs/loop" + "github.com/lightninglabs/loop/fsm" staticaddressrpc "github.com/lightninglabs/loop/swapserverrpc" "github.com/lightningnetwork/lnd/lnrpc/walletrpc" "github.com/lightningnetwork/lnd/lnwallet" @@ -29,6 +31,10 @@ const ( // MaxConfs is unset since we don't require a max number of // confirmations for deposits. MaxConfs = 0 + + // DefaultTransitionTimeout is the default timeout for transitions in + // the deposit state machine. + DefaultTransitionTimeout = 1 * time.Minute ) // ManagerConfig holds the configuration for the address manager. @@ -128,7 +134,7 @@ func (m *Manager) Run(ctx context.Context, currentHeight uint32) error { } // Start the deposit notifier. - m.pollDeposits(ctx) + m.pollDeposits(m.runCtx) // Communicate to the caller that the address manager has completed its // initialization. @@ -420,3 +426,89 @@ 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(m.runCtx, event, nil) + if err != nil { + return err + } + err = sm.DefaultObserver.WaitForState( + m.runCtx, DefaultTransitionTimeout, expectedFinalState, + ) + if err != nil { + return err + } + } + + return nil +} + +// UpdateDeposit overrides all fields of the deposit with given ID in the store. +func (m *Manager) UpdateDeposit(d *Deposit) error { + return m.cfg.Store.UpdateDeposit(m.runCtx, d) +} diff --git a/staticaddr/deposit/manager_test.go b/staticaddr/deposit/manager_test.go index 4c9ee933..eeb875f8 100644 --- a/staticaddr/deposit/manager_test.go +++ b/staticaddr/deposit/manager_test.go @@ -51,6 +51,17 @@ type mockStaticAddressClient struct { mock.Mock } +func (m *mockStaticAddressClient) ServerWithdrawDeposits(ctx context.Context, + in *swapserverrpc.ServerWithdrawRequest, + opts ...grpc.CallOption) (*swapserverrpc.ServerWithdrawResponse, + error) { + + args := m.Called(ctx, in, opts) + + return args.Get(0).(*swapserverrpc.ServerWithdrawResponse), + args.Error(1) +} + func (m *mockStaticAddressClient) ServerNewAddress(ctx context.Context, in *swapserverrpc.ServerNewAddressRequest, opts ...grpc.CallOption) ( *swapserverrpc.ServerNewAddressResponse, error) { @@ -158,8 +169,7 @@ func (m *MockChainNotifier) RegisterSpendNtfn(ctx context.Context, // TestManager checks that the manager processes the right channel notifications // while a deposit is expiring. func TestManager(t *testing.T) { - ctxb, cancel := context.WithCancel(context.Background()) - defer cancel() + ctx := context.Background() // Create the test context with required mocks. testContext := newManagerTestContext(t) @@ -167,7 +177,7 @@ func TestManager(t *testing.T) { // Start the deposit manager. go func() { err := testContext.manager.Run( - ctxb, uint32(testContext.mockLnd.Height), + ctx, uint32(testContext.mockLnd.Height), ) require.NoError(t, err) }() @@ -200,6 +210,9 @@ func TestManager(t *testing.T) { BlockHeight: defaultDepositConfirmations + defaultExpiry + 3, Tx: expiryTx, } + + // Ensure that the deposit is finalized. + <-finalizedDepositChan } // ManagerTestContext is a helper struct that contains all the necessary diff --git a/staticaddr/deposit/sql_store.go b/staticaddr/deposit/sql_store.go index 72900775..6221210e 100644 --- a/staticaddr/deposit/sql_store.go +++ b/staticaddr/deposit/sql_store.go @@ -1,8 +1,10 @@ package deposit import ( + "bytes" "context" "database/sql" + "encoding/hex" "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -74,16 +76,35 @@ func (s *SqlStore) UpdateDeposit(ctx context.Context, deposit *Deposit) error { } confirmationHeight = sql.NullInt64{ Int64: deposit.ConfirmationHeight, - Valid: deposit.ConfirmationHeight != 0, } ) + var finalizedWithdrawalTx string + if deposit.FinalizedWithdrawalTx != nil { + var buffer bytes.Buffer + err := deposit.FinalizedWithdrawalTx.Serialize( + &buffer, + ) + if err != nil { + return err + } + + finalizedWithdrawalTx = hex.EncodeToString(buffer.Bytes()) + } + updateArgs := sqlc.UpdateDepositParams{ DepositID: deposit.ID[:], TxHash: txHash, OutIndex: outIndex.Int32, ConfirmationHeight: confirmationHeight.Int64, - ExpirySweepTxid: deposit.ExpirySweepTxid[:], + FinalizedWithdrawalTx: sql.NullString{ + String: finalizedWithdrawalTx, + Valid: finalizedWithdrawalTx != "", + }, + } + + if deposit.ExpirySweepTxid != (chainhash.Hash{}) { + updateArgs.ExpirySweepTxid = deposit.ExpirySweepTxid[:] } return s.baseDB.ExecTx(ctx, &loopdb.SqliteTxOptions{}, @@ -193,6 +214,20 @@ func (s *SqlStore) toDeposit(row sqlc.Deposit, expirySweepTxid = *hash } + var finalizedWithdrawalTx *wire.MsgTx + if row.FinalizedWithdrawalTx.Valid { + finalizedWithdrawalTx = &wire.MsgTx{} + tx, err := hex.DecodeString(row.FinalizedWithdrawalTx.String) + if err != nil { + return nil, err + } + + err = finalizedWithdrawalTx.Deserialize(bytes.NewReader(tx)) + if err != nil { + return nil, err + } + } + return &Deposit{ ID: id, state: fsm.StateType(lastUpdate.UpdateState), @@ -200,10 +235,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, + FinalizedWithdrawalTx: finalizedWithdrawalTx, }, nil }