diff --git a/contractcourt/briefcase_test.go b/contractcourt/briefcase_test.go index 622fba2c1..5010cde9e 100644 --- a/contractcourt/briefcase_test.go +++ b/contractcourt/briefcase_test.go @@ -532,6 +532,13 @@ func TestContractResolutionsStorage(t *testing.T) { }, } + // First make sure that fetching unlogged contract resolutions will + // fail. + _, err = testLog.FetchContractResolutions() + if err == nil { + t.Fatalf("expected reading unlogged resolution from db to fail") + } + // Insert the resolution into the database, then immediately retrieve // them so we can compare equality against the original version. if err := testLog.LogContractResolutions(&res); err != nil { diff --git a/contractcourt/chain_arbitrator.go b/contractcourt/chain_arbitrator.go index 1e728d4ec..b8491b069 100644 --- a/contractcourt/chain_arbitrator.go +++ b/contractcourt/chain_arbitrator.go @@ -243,6 +243,8 @@ func newActiveChannelArbitrator(channel *channeldb.OpenChannel, return chanMachine.ForceClose() }, MarkCommitmentBroadcasted: channel.MarkCommitmentBroadcasted, + MarkChannelClosed: channel.CloseChannel, + IsPendingClose: false, ChainArbitratorConfig: c.cfg, ChainEvents: chanEvents, } @@ -398,6 +400,9 @@ func (c *ChainArbitrator) Start() error { BlockEpochs: blockEpoch, ChainArbitratorConfig: c.cfg, ChainEvents: &ChainEventSubscription{}, + IsPendingClose: true, + ClosingHeight: closeChanInfo.CloseHeight, + CloseType: closeChanInfo.CloseType, } chanLog, err := newBoltArbitratorLog( c.chanSource.DB, arbCfg, c.cfg.ChainHash, chanPoint, diff --git a/contractcourt/chain_watcher.go b/contractcourt/chain_watcher.go index f1786441c..c183788fa 100644 --- a/contractcourt/chain_watcher.go +++ b/contractcourt/chain_watcher.go @@ -21,6 +21,13 @@ import ( type LocalUnilateralCloseInfo struct { *chainntnfs.SpendDetail *lnwallet.LocalForceCloseSummary + *channeldb.ChannelCloseSummary +} + +// CooperativeCloseInfo encapsulates all the informnation we need to act +// on a cooperative close that gets confirmed. +type CooperativeCloseInfo struct { + *channeldb.ChannelCloseSummary } // ChainEventSubscription is a struct that houses a subscription to be notified @@ -42,9 +49,7 @@ type ChainEventSubscription struct { // CooperativeClosure is a signal that will be sent upon once a // cooperative channel closure has been detected confirmed. - // - // TODO(roasbeef): or something else - CooperativeClosure chan struct{} + CooperativeClosure chan *CooperativeCloseInfo // ContractBreach is a channel that will be sent upon if we detect a // contract breach. The struct sent across the channel contains all the @@ -232,7 +237,7 @@ func (c *chainWatcher) SubscribeChannelEvents() *ChainEventSubscription { ChanPoint: c.cfg.chanState.FundingOutpoint, RemoteUnilateralClosure: make(chan *lnwallet.UnilateralCloseSummary, 1), LocalUnilateralClosure: make(chan *LocalUnilateralCloseInfo, 1), - CooperativeClosure: make(chan struct{}, 1), + CooperativeClosure: make(chan *CooperativeCloseInfo, 1), ContractBreach: make(chan *lnwallet.BreachRetribution, 1), Cancel: func() { c.Lock() @@ -511,26 +516,24 @@ func (c *chainWatcher) dispatchCooperativeClose(commitSpend *chainntnfs.SpendDet SettledBalance: localAmt, CloseType: channeldb.CooperativeClose, ShortChanID: c.cfg.chanState.ShortChanID(), - IsPending: false, + IsPending: true, RemoteCurrentRevocation: c.cfg.chanState.RemoteCurrentRevocation, RemoteNextRevocation: c.cfg.chanState.RemoteNextRevocation, LocalChanConfig: c.cfg.chanState.LocalChanCfg, } - err := c.cfg.chanState.CloseChannel(closeSummary) - if err != nil && err != channeldb.ErrNoActiveChannels && - err != channeldb.ErrNoChanDBExists { - return fmt.Errorf("unable to close chan state: %v", err) + + // Create a summary of all the information needed to handle the + // cooperative closure. + closeInfo := &CooperativeCloseInfo{ + ChannelCloseSummary: closeSummary, } - log.Infof("closeObserver: ChannelPoint(%v) is fully "+ - "closed, at height: %v", - c.cfg.chanState.FundingOutpoint, - commitSpend.SpendingHeight) - + // With the event processed, we'll now notify all subscribers of the + // event. c.Lock() for _, sub := range c.clientSubscriptions { select { - case sub.CooperativeClosure <- struct{}{}: + case sub.CooperativeClosure <- closeInfo: case <-c.quit: c.Unlock() return fmt.Errorf("exiting") @@ -558,8 +561,7 @@ func (c *chainWatcher) dispatchLocalForceClose( } // As we've detected that the channel has been closed, immediately - // delete the state from disk, creating a close summary for future - // usage by related sub-systems. + // creating a close summary for future usage by related sub-systems. chanSnapshot := forceClose.ChanSnapshot closeSummary := &channeldb.ChannelCloseSummary{ ChanPoint: chanSnapshot.ChannelPoint, @@ -587,14 +589,12 @@ func (c *chainWatcher) dispatchLocalForceClose( htlcValue := btcutil.Amount(htlc.SweepSignDesc.Output.Value) closeSummary.TimeLockedBalance += htlcValue } - err = c.cfg.chanState.CloseChannel(closeSummary) - if err != nil { - return fmt.Errorf("unable to delete channel state: %v", err) - } // With the event processed, we'll now notify all subscribers of the // event. - closeInfo := &LocalUnilateralCloseInfo{commitSpend, forceClose} + closeInfo := &LocalUnilateralCloseInfo{ + commitSpend, forceClose, closeSummary, + } c.Lock() for _, sub := range c.clientSubscriptions { select { @@ -641,22 +641,10 @@ func (c *chainWatcher) dispatchRemoteForceClose( return err } - // As we've detected that the channel has been closed, immediately - // delete the state from disk, creating a close summary for future - // usage by related sub-systems. - err = c.cfg.chanState.CloseChannel(&uniClose.ChannelCloseSummary) - if err != nil { - return fmt.Errorf("unable to delete channel state: %v", err) - } - // With the event processed, we'll now notify all subscribers of the // event. c.Lock() for _, sub := range c.clientSubscriptions { - // TODO(roasbeef): send msg before writing to disk - // * need to ensure proper fault tolerance in all cases - // * get ACK from the consumer of the ntfn before writing to disk? - // * no harm in repeated ntfns: at least once semantics select { case sub.RemoteUnilateralClosure <- uniClose: case <-c.quit: @@ -743,6 +731,7 @@ func (c *chainWatcher) dispatchContractBreach(spendEvent *chainntnfs.SpendDetail // channel as pending force closed. // // TODO(roasbeef): instead mark we got all the monies? + // TODO(halseth): move responsibility to breach arbiter? settledBalance := remoteCommit.LocalBalance.ToSatoshis() closeSummary := channeldb.ChannelCloseSummary{ ChanPoint: c.cfg.chanState.FundingOutpoint, diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index 8c3689b1d..ba64eecc2 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -1,7 +1,6 @@ package contractcourt import ( - "fmt" "sync" "sync/atomic" @@ -95,6 +94,25 @@ type ChannelArbitratorConfig struct { // being broadcast, and we are waiting for the commitment to confirm. MarkCommitmentBroadcasted func() error + // MarkChannelClosed marks the channel closed in the database, with the + // passed close summary. After this method successfully returns we can + // no longer expect to receive chain events for this channel, and must + // be able to recover from a failure without getting the close event + // again. + MarkChannelClosed func(*channeldb.ChannelCloseSummary) error + + // IsPendingClose is a boolean indicating whether the channel is marked + // as pending close in the database. + IsPendingClose bool + + // ClosingHeight is the height at which the channel was closed. Note + // that this value is only valid if IsPendingClose is true. + ClosingHeight uint32 + + // CloseType is the type of the close event in case IsPendingClose is + // true. Otherwise this value is unset. + CloseType channeldb.ClosureType + // MarkChannelResolved is a function closure that serves to mark a // channel as "fully resolved". A channel itself can be considered // fully resolved once all active contracts have individually been @@ -244,15 +262,56 @@ func (c *ChannelArbitrator) Start() error { return err } + // If the channel has been marked pending close in the database, and we + // haven't transitioned the state machine to StateContractClosed (or a + // suceeding state), then a state transition most likely failed. We'll + // try to recover from this by manually advancing the state by setting + // the corresponding close trigger. + trigger := chainTrigger + triggerHeight := uint32(bestHeight) + if c.cfg.IsPendingClose { + switch c.state { + case StateDefault: + fallthrough + case StateBroadcastCommit: + fallthrough + case StateCommitmentBroadcasted: + switch c.cfg.CloseType { + case channeldb.LocalForceClose: + trigger = localCloseTrigger + case channeldb.RemoteForceClose: + trigger = remoteCloseTrigger + } + triggerHeight = c.cfg.ClosingHeight + + log.Warnf("ChannelArbitrator(%v): detected stalled "+ + "state=%v for closed channel, using "+ + "trigger=%v", c.cfg.ChanPoint, c.state, trigger) + } + } + // We'll now attempt to advance our state forward based on the current // on-chain state, and our set of active contracts. startingState := c.state - nextState, _, err := c.advanceState( - uint32(bestHeight), chainTrigger, nil, - ) + nextState, _, err := c.advanceState(triggerHeight, trigger) if err != nil { - c.cfg.BlockEpochs.Cancel() - return err + switch err { + + // If we detect that we tried to fetch resolutions, but failed, + // this channel was marked closed in the database before + // resolutions successfully written. In this case there is not + // much we can do, so we don't return the error. + case errScopeBucketNoExist: + fallthrough + case errNoResolutions: + log.Warnf("ChannelArbitrator(%v): detected closed"+ + "channel with no contract resolutions written.", + c.cfg.ChanPoint) + + default: + c.cfg.BlockEpochs.Cancel() + return err + } } // If we start and ended at the awaiting full resolution state, then @@ -332,6 +391,10 @@ const ( // localCloseTrigger is a transition trigger driven by our commitment // being confirmed. localCloseTrigger + + // coopCloseTrigger is a transition trigger driven by a cooperative + // close transaction being confirmed. + coopCloseTrigger ) // String returns a human readable string describing the passed @@ -350,6 +413,9 @@ func (t transitionTrigger) String() string { case localCloseTrigger: return "localCloseTrigger" + case coopCloseTrigger: + return "coopCloseTrigger" + default: return "unknown trigger" } @@ -415,10 +481,16 @@ func (c *ChannelArbitrator) stateStep(triggerHeight uint32, case userTrigger: nextState = StateBroadcastCommit + // If the trigger is a cooperative close being confirmed, then + // we can go straight to StateFullyResolved, as there won't be + // any contracts to resolve. + case coopCloseTrigger: + nextState = StateFullyResolved + // Otherwise, if this state advance was triggered by a // commitment being confirmed on chain, then we'll jump // straight to the state where the contract has already been - // closed. + // closed, and we will inspect the set of unresolved contracts. case localCloseTrigger: log.Errorf("ChannelArbitrator(%v): unexpected local "+ "commitment confirmed while in StateDefault", @@ -613,17 +685,16 @@ func (c *ChannelArbitrator) stateStep(triggerHeight uint32, log.Infof("ChannelPoint(%v) has been fully resolved "+ "on-chain at height=%v", c.cfg.ChanPoint, triggerHeight) - return nextState, closeTx, c.cfg.MarkChannelResolved() - } - if err := c.log.CommitState(nextState); err != nil { - return StateError, nil, err + if err := c.cfg.MarkChannelResolved(); err != nil { + log.Errorf("unable to mark channel resolved: %v", err) + return StateError, closeTx, err + } } log.Tracef("ChannelArbitrator(%v): next_state=%v", c.cfg.ChanPoint, nextState) - c.state = nextState return nextState, closeTx, nil } @@ -634,8 +705,7 @@ func (c *ChannelArbitrator) stateStep(triggerHeight uint32, // param is a callback that allows the caller to execute an arbitrary action // after each state transition. func (c *ChannelArbitrator) advanceState(triggerHeight uint32, - trigger transitionTrigger, stateCallback func(ArbitratorState) error) ( - ArbitratorState, *wire.MsgTx, error) { + trigger transitionTrigger) (ArbitratorState, *wire.MsgTx, error) { var ( priorState ArbitratorState @@ -654,7 +724,8 @@ func (c *ChannelArbitrator) advanceState(triggerHeight uint32, triggerHeight, trigger, ) if err != nil { - log.Errorf("unable to advance state: %v", err) + log.Errorf("ChannelArbitrator(%v): unable to advance "+ + "state: %v", c.cfg.ChanPoint, err) return priorState, nil, err } @@ -662,23 +733,25 @@ func (c *ChannelArbitrator) advanceState(triggerHeight uint32, forceCloseTx = closeTx } - // If we have a state callback, then we'll attempt to execute - // it. If the callback doesn't execute successfully, then we'll - // exit early. - if stateCallback != nil { - if err := stateCallback(nextState); err != nil { - return nextState, closeTx, err - } - } - // Our termination transition is a noop transition. If we get // our prior state back as the next state, then we'll // terminate. if nextState == priorState { - log.Tracef("ChannelArbitrator(%v): terminating at state=%v", - c.cfg.ChanPoint, nextState) + log.Tracef("ChannelArbitrator(%v): terminating at "+ + "state=%v", c.cfg.ChanPoint, nextState) return nextState, forceCloseTx, nil } + + // As the prior state was successfully executed, we can now + // commit the next state. This ensures that we will re-execute + // the prior state if anything fails. + if err := c.log.CommitState(nextState); err != nil { + log.Errorf("ChannelArbitrator(%v): unable to commit "+ + "next state(%v): %v", c.cfg.ChanPoint, + nextState, err) + return priorState, nil, err + } + c.state = nextState } } @@ -1318,7 +1391,7 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) { // Now that a new block has arrived, we'll attempt to // advance our state forward. nextState, _, err := c.advanceState( - uint32(bestHeight), chainTrigger, nil, + uint32(bestHeight), chainTrigger, ) if err != nil { log.Errorf("unable to advance state: %v", err) @@ -1365,16 +1438,28 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) { // We've cooperatively closed the channel, so we're no longer // needed. We'll mark the channel as resolved and exit. - case <-c.cfg.ChainEvents.CooperativeClosure: - log.Infof("ChannelArbitrator(%v) closing due to co-op "+ - "closure", c.cfg.ChanPoint) + case closeInfo := <-c.cfg.ChainEvents.CooperativeClosure: + log.Infof("ChannelArbitrator(%v) marking channel "+ + "cooperatively closed", c.cfg.ChanPoint) - if err := c.cfg.MarkChannelResolved(); err != nil { - log.Errorf("Unable to mark contract "+ - "resolved: %v", err) + err := c.cfg.MarkChannelClosed( + closeInfo.ChannelCloseSummary, + ) + if err != nil { + log.Errorf("unable to mark channel closed: "+ + "%v", err) + return } - return + // We'll now advance our state machine until it reaches + // a terminal state, and the channel is marked resolved. + _, _, err = c.advanceState( + closeInfo.CloseHeight, coopCloseTrigger, + ) + if err != nil { + log.Errorf("unable to advance state: %v", err) + return + } // We have broadcasted our commitment, and it is now confirmed // on-chain. @@ -1396,31 +1481,40 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) { } // When processing a unilateral close event, we'll - // transition directly to the ContractClosed state. - // When the state machine reaches that state, we'll log - // out the set of resolutions. - stateCb := func(nextState ArbitratorState) error { - if nextState != StateContractClosed { - return nil - } + // transition to the ContractClosed state. We'll log + // out the set of resolutions such that they are + // available to fetch in that state. + err := c.log.LogContractResolutions(contractRes) + if err != nil { + log.Errorf("unable to write resolutions: %v", + err) + return + } - err := c.log.LogContractResolutions( - contractRes, - ) - if err != nil { - return fmt.Errorf("unable to "+ - "write resolutions: %v", - err) - } - - return nil + // After the set of resolutions are successfully + // logged, we can safely close the channel. After this + // succeeds we won't be getting chain events anymore, + // so we must make sure we can recover on restart after + // it is marked closed. If the next state transation + // fails, we'll start up in the prior state again, and + // we won't be longer getting chain events. In this + // case we must manually re-trigger the state + // transition into StateContractClosed based on the + // close status of the channel. + err = c.cfg.MarkChannelClosed( + closeInfo.ChannelCloseSummary, + ) + if err != nil { + log.Errorf("unable to mark "+ + "channel closed: %v", err) + return } // We'll now advance our state machine until it reaches // a terminal state. - _, _, err := c.advanceState( + _, _, err = c.advanceState( uint32(closeInfo.SpendingHeight), - localCloseTrigger, stateCb, + localCloseTrigger, ) if err != nil { log.Errorf("unable to advance state: %v", err) @@ -1443,9 +1537,6 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) { HtlcResolutions: *uniClosure.HtlcResolutions, } - // TODO(roasbeef): modify signal to also detect - // cooperative closures? - // As we're now acting upon an event triggered by the // broadcast of the remote commitment transaction, // we'll swap out our active HTLC set with the set @@ -1453,30 +1544,39 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) { c.activeHTLCs = newHtlcSet(uniClosure.RemoteCommit.Htlcs) // When processing a unilateral close event, we'll - // transition directly to the ContractClosed state. - // When the state machine reaches that state, we'll log - // out the set of resolutions. - stateCb := func(nextState ArbitratorState) error { - if nextState != StateContractClosed { - return nil - } + // transition to the ContractClosed state. We'll log + // out the set of resolutions such that they are + // available to fetch in that state. + err := c.log.LogContractResolutions(contractRes) + if err != nil { + log.Errorf("unable to write resolutions: %v", + err) + return + } - err := c.log.LogContractResolutions( - contractRes, - ) - if err != nil { - return fmt.Errorf("unable to write "+ - "resolutions: %v", err) - } - - return nil + // After the set of resolutions are successfully + // logged, we can safely close the channel. After this + // succeeds we won't be getting chain events anymore, + // so we must make sure we can recover on restart after + // it is marked closed. If the next state transation + // fails, we'll start up in the prior state again, and + // we won't be longer getting chain events. In this + // case we must manually re-trigger the state + // transition into StateContractClosed based on the + // close status of the channel. + closeSummary := &uniClosure.ChannelCloseSummary + err = c.cfg.MarkChannelClosed(closeSummary) + if err != nil { + log.Errorf("unable to mark channel closed: %v", + err) + return } // We'll now advance our state machine until it reaches // a terminal state. - _, _, err := c.advanceState( + _, _, err = c.advanceState( uint32(uniClosure.SpendingHeight), - remoteCloseTrigger, stateCb, + remoteCloseTrigger, ) if err != nil { log.Errorf("unable to advance state: %v", err) @@ -1521,7 +1621,7 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) { } nextState, closeTx, err := c.advanceState( - uint32(bestHeight), userTrigger, nil, + uint32(bestHeight), userTrigger, ) if err != nil { log.Errorf("unable to advance state: %v", err) diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index f17482860..b5a410734 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -8,10 +8,84 @@ import ( "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/wire" "github.com/lightningnetwork/lnd/chainntnfs" + "github.com/lightningnetwork/lnd/channeldb" "github.com/lightningnetwork/lnd/lnwallet" "github.com/lightningnetwork/lnd/lnwire" ) +type mockArbitratorLog struct { + state ArbitratorState + newStates chan ArbitratorState + failLog bool + failFetch error + failCommit bool + failCommitState ArbitratorState +} + +// A compile time check to ensure mockArbitratorLog meets the ArbitratorLog +// interface. +var _ ArbitratorLog = (*mockArbitratorLog)(nil) + +func (b *mockArbitratorLog) CurrentState() (ArbitratorState, error) { + return b.state, nil +} + +func (b *mockArbitratorLog) CommitState(s ArbitratorState) error { + if b.failCommit && s == b.failCommitState { + return fmt.Errorf("intentional commit error at state %v", + b.failCommitState) + } + b.state = s + b.newStates <- s + return nil +} + +func (b *mockArbitratorLog) FetchUnresolvedContracts() ([]ContractResolver, error) { + var contracts []ContractResolver + return contracts, nil +} + +func (b *mockArbitratorLog) InsertUnresolvedContracts(resolvers ...ContractResolver) error { + return nil +} + +func (b *mockArbitratorLog) SwapContract(oldContract, newContract ContractResolver) error { + return nil +} + +func (b *mockArbitratorLog) ResolveContract(res ContractResolver) error { + return nil +} + +func (b *mockArbitratorLog) LogContractResolutions(c *ContractResolutions) error { + if b.failLog { + return fmt.Errorf("intentional log failure") + } + return nil +} + +func (b *mockArbitratorLog) FetchContractResolutions() (*ContractResolutions, error) { + if b.failFetch != nil { + return nil, b.failFetch + } + c := &ContractResolutions{} + + return c, nil +} + +func (b *mockArbitratorLog) LogChainActions(actions ChainActionMap) error { + return nil +} + +func (b *mockArbitratorLog) FetchChainActions() (ChainActionMap, error) { + actionsMap := make(ChainActionMap) + return actionsMap, nil +} + +func (b *mockArbitratorLog) WipeHistory() error { + return nil +} + type mockChainIO struct{} func (*mockChainIO) GetBestBlock() (*chainhash.Hash, int32, error) { @@ -31,7 +105,8 @@ func (*mockChainIO) GetBlock(blockHash *chainhash.Hash) (*wire.MsgBlock, error) return nil, nil } -func createTestChannelArbitrator() (*ChannelArbitrator, chan struct{}, func(), error) { +func createTestChannelArbitrator(log ArbitratorLog) (*ChannelArbitrator, + chan struct{}, error) { blockEpoch := &chainntnfs.BlockEpochEvent{ Cancel: func() {}, } @@ -41,7 +116,7 @@ func createTestChannelArbitrator() (*ChannelArbitrator, chan struct{}, func(), e chanEvents := &ChainEventSubscription{ RemoteUnilateralClosure: make(chan *lnwallet.UnilateralCloseSummary, 1), LocalUnilateralClosure: make(chan *LocalUnilateralCloseInfo, 1), - CooperativeClosure: make(chan struct{}, 1), + CooperativeClosure: make(chan *CooperativeCloseInfo, 1), ContractBreach: make(chan *lnwallet.BreachRetribution, 1), } @@ -77,20 +152,15 @@ func createTestChannelArbitrator() (*ChannelArbitrator, chan struct{}, func(), e MarkCommitmentBroadcasted: func() error { return nil }, - + MarkChannelClosed: func(*channeldb.ChannelCloseSummary) error { + return nil + }, + IsPendingClose: false, ChainArbitratorConfig: chainArbCfg, ChainEvents: chanEvents, } - testLog, cleanUp, err := newTestBoltArbLog( - testChainHash, testChanPoint1, - ) - if err != nil { - return nil, nil, nil, fmt.Errorf("unable to create test log: %v", - err) - } - return NewChannelArbitrator(arbCfg, nil, testLog), - resolvedChan, cleanUp, nil + return NewChannelArbitrator(arbCfg, nil, log), resolvedChan, nil } // assertState checks that the ChannelArbitrator is in the state we expect it @@ -102,13 +172,18 @@ func assertState(t *testing.T, c *ChannelArbitrator, expected ArbitratorState) { } // TestChannelArbitratorCooperativeClose tests that the ChannelArbitertor -// correctly does nothing in case a cooperative close is confirmed. +// correctly marks the channel resolved in case a cooperative close is +// confirmed. func TestChannelArbitratorCooperativeClose(t *testing.T) { - chanArb, _, cleanUp, err := createTestChannelArbitrator() + log := &mockArbitratorLog{ + state: StateDefault, + newStates: make(chan ArbitratorState, 5), + } + + chanArb, resolved, err := createTestChannelArbitrator(log) if err != nil { t.Fatalf("unable to create ChannelArbitrator: %v", err) } - defer cleanUp() if err := chanArb.Start(); err != nil { t.Fatalf("unable to start ChannelArbitrator: %v", err) @@ -118,21 +193,70 @@ func TestChannelArbitratorCooperativeClose(t *testing.T) { // It should start out in the default state. assertState(t, chanArb, StateDefault) - // Cooperative close should do nothing. - // TODO: this will change? - chanArb.cfg.ChainEvents.CooperativeClosure <- struct{}{} - assertState(t, chanArb, StateDefault) + // We set up a channel to detect when MarkChannelClosed is called. + closeInfos := make(chan *channeldb.ChannelCloseSummary) + chanArb.cfg.MarkChannelClosed = func( + closeInfo *channeldb.ChannelCloseSummary) error { + closeInfos <- closeInfo + return nil + } + + // Cooperative close should do trigger a MarkChannelClosed + + // MarkChannelResolved. + closeInfo := &CooperativeCloseInfo{ + &channeldb.ChannelCloseSummary{}, + } + chanArb.cfg.ChainEvents.CooperativeClosure <- closeInfo + + select { + case c := <-closeInfos: + if c.CloseType != channeldb.CooperativeClose { + t.Fatalf("expected cooperative close, got %v", c.CloseType) + } + case <-time.After(5 * time.Second): + t.Fatalf("timeout waiting for channel close") + } + + // It should mark the channel as resolved. + select { + case <-resolved: + // Expected. + case <-time.After(5 * time.Second): + t.Fatalf("contract was not resolved") + } } -// TestChannelArbitratorRemoteForceClose checks that the ChannelArbitrotor goes +func assertStateTransitions(t *testing.T, newStates <-chan ArbitratorState, + expectedStates ...ArbitratorState) { + t.Helper() + + for _, exp := range expectedStates { + var state ArbitratorState + select { + case state = <-newStates: + case <-time.After(5 * time.Second): + t.Fatalf("new state not received") + } + + if state != exp { + t.Fatalf("expected new state %v, got %v", exp, state) + } + } +} + +// TestChannelArbitratorRemoteForceClose checks that the ChannelArbitrator goes // through the expected states if a remote force close is observed in the // chain. func TestChannelArbitratorRemoteForceClose(t *testing.T) { - chanArb, resolved, cleanUp, err := createTestChannelArbitrator() + log := &mockArbitratorLog{ + state: StateDefault, + newStates: make(chan ArbitratorState, 5), + } + + chanArb, resolved, err := createTestChannelArbitrator(log) if err != nil { t.Fatalf("unable to create ChannelArbitrator: %v", err) } - defer cleanUp() if err := chanArb.Start(); err != nil { t.Fatalf("unable to start ChannelArbitrator: %v", err) @@ -153,28 +277,34 @@ func TestChannelArbitratorRemoteForceClose(t *testing.T) { } chanArb.cfg.ChainEvents.RemoteUnilateralClosure <- uniClose - // It should mark the channel as resolved. + // It should transition StateDefault -> StateContractClosed -> + // StateFullyResolved. + assertStateTransitions( + t, log.newStates, StateContractClosed, StateFullyResolved, + ) + + // It should alos mark the channel as resolved. select { case <-resolved: // Expected. case <-time.After(5 * time.Second): t.Fatalf("contract was not resolved") } - - // TODO: intermediate states. - // We expect the ChannelArbitrator to end up in the the resolved state. - assertState(t, chanArb, StateFullyResolved) } // TestChannelArbitratorLocalForceClose tests that the ChannelArbitrator goes // through the expected states in case we request it to force close the channel, // and the local force close event is observed in chain. func TestChannelArbitratorLocalForceClose(t *testing.T) { - chanArb, resolved, cleanUp, err := createTestChannelArbitrator() + log := &mockArbitratorLog{ + state: StateDefault, + newStates: make(chan ArbitratorState, 5), + } + + chanArb, resolved, err := createTestChannelArbitrator(log) if err != nil { t.Fatalf("unable to create ChannelArbitrator: %v", err) } - defer cleanUp() if err := chanArb.Start(); err != nil { t.Fatalf("unable to start ChannelArbitrator: %v", err) @@ -208,6 +338,9 @@ func TestChannelArbitratorLocalForceClose(t *testing.T) { closeTx: respChan, } + // It should transition to StateBroadcastCommit. + assertStateTransitions(t, log.newStates, StateBroadcastCommit) + // When it is broadcasting the force close, its state should be // StateBroadcastCommit. select { @@ -219,12 +352,23 @@ func TestChannelArbitratorLocalForceClose(t *testing.T) { t.Fatalf("did not get state update") } + // After broadcasting, transition should be to + // StateCommitmentBroadcasted. + assertStateTransitions(t, log.newStates, StateCommitmentBroadcasted) + select { case <-respChan: + case <-time.After(5 * time.Second): + t.Fatalf("no response received") + } + + select { case err := <-errChan: - t.Fatalf("error force closing channel: %v", err) - case <-time.After(15 * time.Second): - t.Fatalf("did not receive reponse") + if err != nil { + t.Fatalf("error force closing channel: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatalf("no response received") } // After broadcasting the close tx, it should be in state @@ -238,29 +382,35 @@ func TestChannelArbitratorLocalForceClose(t *testing.T) { CloseTx: &wire.MsgTx{}, HtlcResolutions: &lnwallet.HtlcResolutions{}, }, + &channeldb.ChannelCloseSummary{}, } - // It should mark the channel as resolved. + + // It should transition StateContractClosed -> StateFullyResolved. + assertStateTransitions(t, log.newStates, StateContractClosed, + StateFullyResolved) + + // It should also mark the channel as resolved. select { case <-resolved: // Expected. case <-time.After(5 * time.Second): t.Fatalf("contract was not resolved") } - - // And end up in the StateFullyResolved state. - // TODO: intermediate states as well. - assertState(t, chanArb, StateFullyResolved) } // TestChannelArbitratorLocalForceCloseRemoteConfiremd tests that the // ChannelArbitrator behaves as expected in the case where we request a local // force close, but a remote commitment ends up being confirmed in chain. func TestChannelArbitratorLocalForceCloseRemoteConfirmed(t *testing.T) { - chanArb, resolved, cleanUp, err := createTestChannelArbitrator() + log := &mockArbitratorLog{ + state: StateDefault, + newStates: make(chan ArbitratorState, 5), + } + + chanArb, resolved, err := createTestChannelArbitrator(log) if err != nil { t.Fatalf("unable to create ChannelArbitrator: %v", err) } - defer cleanUp() if err := chanArb.Start(); err != nil { t.Fatalf("unable to start ChannelArbitrator: %v", err) @@ -294,6 +444,9 @@ func TestChannelArbitratorLocalForceCloseRemoteConfirmed(t *testing.T) { closeTx: respChan, } + // It should transition to StateBroadcastCommit. + assertStateTransitions(t, log.newStates, StateBroadcastCommit) + // We expect it to be in state StateBroadcastCommit when publishing // the force close. select { @@ -305,12 +458,23 @@ func TestChannelArbitratorLocalForceCloseRemoteConfirmed(t *testing.T) { t.Fatalf("no state update received") } + // After broadcasting, transition should be to + // StateCommitmentBroadcasted. + assertStateTransitions(t, log.newStates, StateCommitmentBroadcasted) + // Wait for a response to the force close. select { case <-respChan: + case <-time.After(5 * time.Second): + t.Fatalf("no response received") + } + + select { case err := <-errChan: - t.Fatalf("error force closing channel: %v", err) - case <-time.After(15 * time.Second): + if err != nil { + t.Fatalf("error force closing channel: %v", err) + } + case <-time.After(5 * time.Second): t.Fatalf("no response received") } @@ -327,6 +491,10 @@ func TestChannelArbitratorLocalForceCloseRemoteConfirmed(t *testing.T) { } chanArb.cfg.ChainEvents.RemoteUnilateralClosure <- uniClose + // It should transition StateContractClosed -> StateFullyResolved. + assertStateTransitions(t, log.newStates, StateContractClosed, + StateFullyResolved) + // It should resolve. select { case <-resolved: @@ -334,10 +502,6 @@ func TestChannelArbitratorLocalForceCloseRemoteConfirmed(t *testing.T) { case <-time.After(15 * time.Second): t.Fatalf("contract was not resolved") } - - // And we expect it to end up in StateFullyResolved. - // TODO: intermediate states as well. - assertState(t, chanArb, StateFullyResolved) } // TestChannelArbitratorLocalForceCloseDoubleSpend tests that the @@ -345,11 +509,15 @@ func TestChannelArbitratorLocalForceCloseRemoteConfirmed(t *testing.T) { // force close, but we fail broadcasting our commitment because a remote // commitment has already been published. func TestChannelArbitratorLocalForceDoubleSpend(t *testing.T) { - chanArb, resolved, cleanUp, err := createTestChannelArbitrator() + log := &mockArbitratorLog{ + state: StateDefault, + newStates: make(chan ArbitratorState, 5), + } + + chanArb, resolved, err := createTestChannelArbitrator(log) if err != nil { t.Fatalf("unable to create ChannelArbitrator: %v", err) } - defer cleanUp() if err := chanArb.Start(); err != nil { t.Fatalf("unable to start ChannelArbitrator: %v", err) @@ -382,6 +550,9 @@ func TestChannelArbitratorLocalForceDoubleSpend(t *testing.T) { closeTx: respChan, } + // It should transition to StateBroadcastCommit. + assertStateTransitions(t, log.newStates, StateBroadcastCommit) + // We expect it to be in state StateBroadcastCommit when publishing // the force close. select { @@ -393,12 +564,23 @@ func TestChannelArbitratorLocalForceDoubleSpend(t *testing.T) { t.Fatalf("no state update received") } + // After broadcasting, transition should be to + // StateCommitmentBroadcasted. + assertStateTransitions(t, log.newStates, StateCommitmentBroadcasted) + // Wait for a response to the force close. select { case <-respChan: + case <-time.After(5 * time.Second): + t.Fatalf("no response received") + } + + select { case err := <-errChan: - t.Fatalf("error force closing channel: %v", err) - case <-time.After(15 * time.Second): + if err != nil { + t.Fatalf("error force closing channel: %v", err) + } + case <-time.After(5 * time.Second): t.Fatalf("no response received") } @@ -415,6 +597,10 @@ func TestChannelArbitratorLocalForceDoubleSpend(t *testing.T) { } chanArb.cfg.ChainEvents.RemoteUnilateralClosure <- uniClose + // It should transition StateContractClosed -> StateFullyResolved. + assertStateTransitions(t, log.newStates, StateContractClosed, + StateFullyResolved) + // It should resolve. select { case <-resolved: @@ -422,8 +608,262 @@ func TestChannelArbitratorLocalForceDoubleSpend(t *testing.T) { case <-time.After(15 * time.Second): t.Fatalf("contract was not resolved") } - - // And we expect it to end up in StateFullyResolved. - // TODO: intermediate states as well. - assertState(t, chanArb, StateFullyResolved) +} + +// TestChannelArbitratorPersistence tests that the ChannelArbitrator is able to +// keep advancing the state machine from various states after restart. +func TestChannelArbitratorPersistence(t *testing.T) { + // Start out with a log that will fail writing the set of resolutions. + log := &mockArbitratorLog{ + state: StateDefault, + newStates: make(chan ArbitratorState, 5), + failLog: true, + } + + chanArb, resolved, err := createTestChannelArbitrator(log) + if err != nil { + t.Fatalf("unable to create ChannelArbitrator: %v", err) + } + + if err := chanArb.Start(); err != nil { + t.Fatalf("unable to start ChannelArbitrator: %v", err) + } + + // It should start in StateDefault. + assertState(t, chanArb, StateDefault) + + // Send a remote force close event. + commitSpend := &chainntnfs.SpendDetail{ + SpenderTxHash: &chainhash.Hash{}, + } + + uniClose := &lnwallet.UnilateralCloseSummary{ + SpendDetail: commitSpend, + HtlcResolutions: &lnwallet.HtlcResolutions{}, + } + chanArb.cfg.ChainEvents.RemoteUnilateralClosure <- uniClose + + // Since writing the resolutions fail, the arbitrator should not + // advance to the next state. + time.Sleep(100 * time.Millisecond) + if log.state != StateDefault { + t.Fatalf("expected to stay in StateDefault") + } + chanArb.Stop() + + // Create a new arbitrator with the same log. + chanArb, resolved, err = createTestChannelArbitrator(log) + if err != nil { + t.Fatalf("unable to create ChannelArbitrator: %v", err) + } + + if err := chanArb.Start(); err != nil { + t.Fatalf("unable to start ChannelArbitrator: %v", err) + } + + // Again, it should start up in the default state. + assertState(t, chanArb, StateDefault) + + // Now we make the log succeed writing the resolutions, but fail when + // attempting to close the channel. + log.failLog = false + chanArb.cfg.MarkChannelClosed = func(*channeldb.ChannelCloseSummary) error { + return fmt.Errorf("intentional close error") + } + + // Send a new remote force close event. + chanArb.cfg.ChainEvents.RemoteUnilateralClosure <- uniClose + + // Since closing the channel failed, the arbitrator should stay in the + // default state. + time.Sleep(100 * time.Millisecond) + if log.state != StateDefault { + t.Fatalf("expected to stay in StateDefault") + } + chanArb.Stop() + + // Create yet another arbitrator with the same log. + chanArb, resolved, err = createTestChannelArbitrator(log) + if err != nil { + t.Fatalf("unable to create ChannelArbitrator: %v", err) + } + + if err := chanArb.Start(); err != nil { + t.Fatalf("unable to start ChannelArbitrator: %v", err) + } + + // Starts out in StateDefault. + assertState(t, chanArb, StateDefault) + + // Now make fetching the resolutions fail. + log.failFetch = fmt.Errorf("intentional fetch failure") + chanArb.cfg.ChainEvents.RemoteUnilateralClosure <- uniClose + + // Since logging the resolutions and closing the channel now succeeds, + // it should advance to StateContractClosed. + assertStateTransitions( + t, log.newStates, StateContractClosed, + ) + + // It should not advance further, however, as fetching resolutions + // failed. + time.Sleep(100 * time.Millisecond) + if log.state != StateContractClosed { + t.Fatalf("expected to stay in StateContractClosed") + } + chanArb.Stop() + + // Create a new arbitrator, and now make fetching resolutions succeed. + log.failFetch = nil + chanArb, resolved, err = createTestChannelArbitrator(log) + if err != nil { + t.Fatalf("unable to create ChannelArbitrator: %v", err) + } + + if err := chanArb.Start(); err != nil { + t.Fatalf("unable to start ChannelArbitrator: %v", err) + } + defer chanArb.Stop() + + // Finally it should advance to StateFullyResolved. + assertStateTransitions( + t, log.newStates, StateFullyResolved, + ) + + // It should also mark the channel as resolved. + select { + case <-resolved: + // Expected. + case <-time.After(5 * time.Second): + t.Fatalf("contract was not resolved") + } +} + +// TestChannelArbitratorCommitFailure tests that the channel arbitrator is able +// to recover from a failed CommitState call at restart. +func TestChannelArbitratorCommitFailure(t *testing.T) { + // Start out with a log that will fail committing to StateContractClosed. + log := &mockArbitratorLog{ + state: StateDefault, + newStates: make(chan ArbitratorState, 5), + failCommit: true, + failCommitState: StateContractClosed, + } + + chanArb, resolved, err := createTestChannelArbitrator(log) + if err != nil { + t.Fatalf("unable to create ChannelArbitrator: %v", err) + } + + if err := chanArb.Start(); err != nil { + t.Fatalf("unable to start ChannelArbitrator: %v", err) + } + + // It should start in StateDefault. + assertState(t, chanArb, StateDefault) + + closed := make(chan struct{}) + chanArb.cfg.MarkChannelClosed = func(*channeldb.ChannelCloseSummary) error { + close(closed) + return nil + } + + // Send a remote force close event. + commitSpend := &chainntnfs.SpendDetail{ + SpenderTxHash: &chainhash.Hash{}, + } + + uniClose := &lnwallet.UnilateralCloseSummary{ + SpendDetail: commitSpend, + HtlcResolutions: &lnwallet.HtlcResolutions{}, + } + chanArb.cfg.ChainEvents.RemoteUnilateralClosure <- uniClose + + select { + case <-closed: + case <-time.After(5 * time.Second): + t.Fatalf("channel was not marked closed") + } + + // Since the channel was marked closed in the database, but the commit + // to the next state failed, the state should still be StateDefault. + time.Sleep(100 * time.Millisecond) + if log.state != StateDefault { + t.Fatalf("expected to stay in StateDefault") + } + chanArb.Stop() + + // Start the arbitrator again, with IsPendingClose reporting the + // channel closed in the database. + chanArb, resolved, err = createTestChannelArbitrator(log) + if err != nil { + t.Fatalf("unable to create ChannelArbitrator: %v", err) + } + + log.failCommit = false + + chanArb.cfg.IsPendingClose = true + chanArb.cfg.ClosingHeight = 100 + chanArb.cfg.CloseType = channeldb.RemoteForceClose + + if err := chanArb.Start(); err != nil { + t.Fatalf("unable to start ChannelArbitrator: %v", err) + } + + // Since the channel is marked closed in the database, it should + // advance to StateContractClosed and StateFullyResolved. + assertStateTransitions( + t, log.newStates, StateContractClosed, StateFullyResolved, + ) + + // It should also mark the channel as resolved. + select { + case <-resolved: + // Expected. + case <-time.After(5 * time.Second): + t.Fatalf("contract was not resolved") + } +} + +// TestChannelArbitratorEmptyResolutions makes sure that a channel that is +// pending close in the database, but haven't had any resolutions logged will +// not be marked resolved. This situation must be handled to avoid closing +// channels from earlier versions of the ChannelArbitrator, which didn't have a +// proper handoff from the ChainWatcher, and we could risk ending up in a state +// where the channel was closed in the DB, but the resolutions weren't properly +// written. +func TestChannelArbitratorEmptyResolutions(t *testing.T) { + // Start out with a log that will fail writing the set of resolutions. + log := &mockArbitratorLog{ + state: StateDefault, + newStates: make(chan ArbitratorState, 5), + failFetch: errNoResolutions, + } + + chanArb, _, err := createTestChannelArbitrator(log) + if err != nil { + t.Fatalf("unable to create ChannelArbitrator: %v", err) + } + + chanArb.cfg.IsPendingClose = true + chanArb.cfg.ClosingHeight = 100 + chanArb.cfg.CloseType = channeldb.RemoteForceClose + + if err := chanArb.Start(); err != nil { + t.Fatalf("unable to start ChannelArbitrator: %v", err) + } + + // It should not advance its state beyond StateContractClosed, since + // fetching resolutions fails. + assertStateTransitions( + t, log.newStates, StateContractClosed, + ) + + // It should not advance further, however, as fetching resolutions + // failed. + time.Sleep(100 * time.Millisecond) + if log.state != StateContractClosed { + t.Fatalf("expected to stay in StateContractClosed") + } + chanArb.Stop() }