From 245dae0071dfe976231388c832dd9bdbf8097b2a Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:14 +0200 Subject: [PATCH 01/16] contractcourt/channel_arbitrator: properly commit to next state --- contractcourt/channel_arbitrator.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index 8c3689b1d..e1d2151d5 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -613,7 +613,11 @@ 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.cfg.MarkChannelResolved(); err != nil { + log.Errorf("unable to mark channel resolved: %v", err) + return StateError, closeTx, err + } } if err := c.log.CommitState(nextState); err != nil { From 6df302d7f5c91e582ff210d392d4850c4ed768f0 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:14 +0200 Subject: [PATCH 02/16] contractcourt/channel_arbitrator: remove state callback This commit removes the state callback, and instead logs the contract resolutions directly after receiving the unilateral close event. The resolutions won't change so there's not really necessary to wait to log them, and this greatly simplifies the code. --- contractcourt/channel_arbitrator.go | 78 +++++++++-------------------- 1 file changed, 24 insertions(+), 54 deletions(-) diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index e1d2151d5..6bf6c385d 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -1,7 +1,6 @@ package contractcourt import ( - "fmt" "sync" "sync/atomic" @@ -248,7 +247,7 @@ func (c *ChannelArbitrator) Start() error { // on-chain state, and our set of active contracts. startingState := c.state nextState, _, err := c.advanceState( - uint32(bestHeight), chainTrigger, nil, + uint32(bestHeight), chainTrigger, ) if err != nil { c.cfg.BlockEpochs.Cancel() @@ -638,8 +637,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 @@ -666,15 +664,6 @@ 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. @@ -1322,7 +1311,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) @@ -1400,31 +1389,21 @@ 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 - } - - err := c.log.LogContractResolutions( - contractRes, - ) - if err != nil { - return fmt.Errorf("unable to "+ - "write resolutions: %v", - err) - } - - 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 } // 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) @@ -1457,30 +1436,21 @@ 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 - } - - err := c.log.LogContractResolutions( - contractRes, - ) - if err != nil { - return fmt.Errorf("unable to write "+ - "resolutions: %v", err) - } - - 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 } // 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) @@ -1525,7 +1495,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) From 06c683d48d2ea74cd604566c70fe44f6fdc3ad0c Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:14 +0200 Subject: [PATCH 03/16] contractcourt/channel_arbitrator: move next state commit out of stateStep --- contractcourt/channel_arbitrator.go | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index 6bf6c385d..69700f030 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -619,14 +619,9 @@ func (c *ChannelArbitrator) stateStep(triggerHeight uint32, } } - if err := c.log.CommitState(nextState); err != nil { - return StateError, nil, err - } - log.Tracef("ChannelArbitrator(%v): next_state=%v", c.cfg.ChanPoint, nextState) - c.state = nextState return nextState, closeTx, nil } @@ -656,7 +651,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 } @@ -668,10 +664,21 @@ func (c *ChannelArbitrator) advanceState(triggerHeight uint32, // 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 } } From 862c7ae223ca8ba2978eb63a6601c7e2006c6a2b Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:14 +0200 Subject: [PATCH 04/16] contractcourt/channel_arbitrator test: check both respChan and errChan // Both are sent on, so in some cases we would read from errChan first, making the test fail. --- contractcourt/channel_arbitrator_test.go | 35 +++++++++++++++++++----- 1 file changed, 28 insertions(+), 7 deletions(-) diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index f17482860..8d139aa3c 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -221,10 +221,17 @@ func TestChannelArbitratorLocalForceClose(t *testing.T) { 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 @@ -308,9 +315,16 @@ func TestChannelArbitratorLocalForceCloseRemoteConfirmed(t *testing.T) { // 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") } @@ -396,9 +410,16 @@ func TestChannelArbitratorLocalForceDoubleSpend(t *testing.T) { // 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") } From 740db4716f78f71ebe9e18871a05fcc216efb815 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:14 +0200 Subject: [PATCH 05/16] contractcourt/channel_arbitrator test: test state transitions --- contractcourt/channel_arbitrator_test.go | 192 ++++++++++++++++++----- 1 file changed, 153 insertions(+), 39 deletions(-) diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index 8d139aa3c..14553398f 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -12,6 +12,65 @@ import ( "github.com/lightningnetwork/lnd/lnwire" ) +type mockArbitratorLog struct { + state ArbitratorState + newStates chan 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 { + 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 { + return nil +} + +func (b *mockArbitratorLog) FetchContractResolutions() (*ContractResolutions, error) { + 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 +90,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() {}, } @@ -81,16 +141,8 @@ func createTestChannelArbitrator() (*ChannelArbitrator, chan struct{}, func(), e 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 @@ -104,11 +156,15 @@ func assertState(t *testing.T, c *ChannelArbitrator, expected ArbitratorState) { // TestChannelArbitratorCooperativeClose tests that the ChannelArbitertor // correctly does nothing 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, _, 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) @@ -124,15 +180,37 @@ func TestChannelArbitratorCooperativeClose(t *testing.T) { assertState(t, chanArb, StateDefault) } -// 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 +231,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 +292,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,6 +306,10 @@ 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): @@ -246,28 +337,33 @@ func TestChannelArbitratorLocalForceClose(t *testing.T) { HtlcResolutions: &lnwallet.HtlcResolutions{}, }, } - // 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) @@ -301,6 +397,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 { @@ -312,6 +411,10 @@ 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: @@ -341,6 +444,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: @@ -348,10 +455,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 @@ -359,11 +462,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) @@ -396,6 +503,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 { @@ -407,6 +517,10 @@ 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: @@ -436,6 +550,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: @@ -443,8 +561,4 @@ 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) } From 3549f2bc17c36b291dfb5cfcea96a0492ebd102b Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:15 +0200 Subject: [PATCH 06/16] contractcourt/[channel|chain]_arbitrator: define MarkChannelClosed --- contractcourt/chain_arbitrator.go | 1 + contractcourt/channel_arbitrator.go | 7 +++++++ contractcourt/channel_arbitrator_test.go | 4 +++- 3 files changed, 11 insertions(+), 1 deletion(-) diff --git a/contractcourt/chain_arbitrator.go b/contractcourt/chain_arbitrator.go index 1e728d4ec..5525c1857 100644 --- a/contractcourt/chain_arbitrator.go +++ b/contractcourt/chain_arbitrator.go @@ -243,6 +243,7 @@ func newActiveChannelArbitrator(channel *channeldb.OpenChannel, return chanMachine.ForceClose() }, MarkCommitmentBroadcasted: channel.MarkCommitmentBroadcasted, + MarkChannelClosed: channel.CloseChannel, ChainArbitratorConfig: c.cfg, ChainEvents: chanEvents, } diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index 69700f030..8a2403e5a 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -94,6 +94,13 @@ 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 + // 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 diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index 14553398f..00240b5f5 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -137,7 +137,9 @@ func createTestChannelArbitrator(log ArbitratorLog) (*ChannelArbitrator, MarkCommitmentBroadcasted: func() error { return nil }, - + MarkChannelClosed: func(*channeldb.ChannelCloseSummary) error { + return nil + }, ChainArbitratorConfig: chainArbCfg, ChainEvents: chanEvents, } From ad1d78884a508072ba6acea709d23e785124dc0b Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:15 +0200 Subject: [PATCH 07/16] contractcourt: move responsibility for closing coop close channel to chanArb --- contractcourt/chain_watcher.go | 32 +++++++++++---------- contractcourt/channel_arbitrator.go | 44 +++++++++++++++++++++-------- 2 files changed, 50 insertions(+), 26 deletions(-) diff --git a/contractcourt/chain_watcher.go b/contractcourt/chain_watcher.go index f1786441c..38ed88d38 100644 --- a/contractcourt/chain_watcher.go +++ b/contractcourt/chain_watcher.go @@ -23,6 +23,12 @@ type LocalUnilateralCloseInfo struct { *lnwallet.LocalForceCloseSummary } +// 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 // for any on-chain events related to a channel. There are three types of // possible on-chain events: a cooperative channel closure, a unilateral @@ -42,9 +48,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 +236,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 +515,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") diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index 8a2403e5a..4ff382824 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -338,6 +338,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 @@ -356,6 +360,9 @@ func (t transitionTrigger) String() string { case localCloseTrigger: return "localCloseTrigger" + case coopCloseTrigger: + return "coopCloseTrigger" + default: return "unknown trigger" } @@ -421,10 +428,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", @@ -1372,16 +1385,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. @@ -1440,9 +1465,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 From c11e0940c22d43ae7d4ea5af2db6ab73e16a1f93 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:15 +0200 Subject: [PATCH 08/16] contractcourt/channel_arbitrator test: test new coop close actions --- contractcourt/channel_arbitrator_test.go | 43 ++++++++++++++++++++---- 1 file changed, 36 insertions(+), 7 deletions(-) diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index 00240b5f5..36d393d0b 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -8,6 +8,7 @@ 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" ) @@ -101,7 +102,7 @@ func createTestChannelArbitrator(log ArbitratorLog) (*ChannelArbitrator, 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), } @@ -156,14 +157,15 @@ 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) { log := &mockArbitratorLog{ state: StateDefault, newStates: make(chan ArbitratorState, 5), } - chanArb, _, err := createTestChannelArbitrator(log) + chanArb, resolved, err := createTestChannelArbitrator(log) if err != nil { t.Fatalf("unable to create ChannelArbitrator: %v", err) } @@ -176,10 +178,37 @@ 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") + } } func assertStateTransitions(t *testing.T, newStates <-chan ArbitratorState, From 75df58c68b625667946931b54d93b375164dac30 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:15 +0200 Subject: [PATCH 09/16] contractcourt: move responsibility for closing force closes to chanArb This commit moves the responsibility for closing local and remote force closes in the database from the chain watcher to the channel arbitrator. We do this because we previously would close the channel in the database, before sending the event to the channel arbitrator. This could lead to a situation where the channel was marked closed, but the channel arbitrator didn't receive the event before shutdown. As we don't listen for chain events for channels that are closed, those channels would be stuck in the pending close state forever, as the channel arbitrator state machine wouldn't progress. We fix this by letting the ChannelArbitrator close the channel in the database. After the contract resolutions are logged (in the state callback before transitioning to StateContractClosed) we mark the channel closed in the database. This way we make sure that it is marked closed only if the resolutions have been successfully persisted. --- contractcourt/chain_watcher.go | 25 ++++------------ contractcourt/channel_arbitrator.go | 37 ++++++++++++++++++++++++ contractcourt/channel_arbitrator_test.go | 1 + 3 files changed, 44 insertions(+), 19 deletions(-) diff --git a/contractcourt/chain_watcher.go b/contractcourt/chain_watcher.go index 38ed88d38..c183788fa 100644 --- a/contractcourt/chain_watcher.go +++ b/contractcourt/chain_watcher.go @@ -21,6 +21,7 @@ import ( type LocalUnilateralCloseInfo struct { *chainntnfs.SpendDetail *lnwallet.LocalForceCloseSummary + *channeldb.ChannelCloseSummary } // CooperativeCloseInfo encapsulates all the informnation we need to act @@ -560,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, @@ -589,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 { @@ -643,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: @@ -745,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 4ff382824..9cd2fee20 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -1438,6 +1438,25 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) { return } + // 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( @@ -1482,6 +1501,24 @@ func (c *ChannelArbitrator) channelAttendant(bestHeight int32) { return } + // 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( diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index 36d393d0b..39dfa3e52 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -367,6 +367,7 @@ func TestChannelArbitratorLocalForceClose(t *testing.T) { CloseTx: &wire.MsgTx{}, HtlcResolutions: &lnwallet.HtlcResolutions{}, }, + &channeldb.ChannelCloseSummary{}, } // It should transition StateContractClosed -> StateFullyResolved. From 6b24ca03871da6b0e2f7247f2e1133af9d1439c7 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:15 +0200 Subject: [PATCH 10/16] contractcourt test: add TestChannelArbitratorPeristence --- contractcourt/channel_arbitrator_test.go | 137 +++++++++++++++++++++++ 1 file changed, 137 insertions(+) diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index 39dfa3e52..6f4b7f2e4 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -16,6 +16,8 @@ import ( type mockArbitratorLog struct { state ArbitratorState newStates chan ArbitratorState + failLog bool + failFetch error } // A compile time check to ensure mockArbitratorLog meets the ArbitratorLog @@ -50,10 +52,16 @@ func (b *mockArbitratorLog) ResolveContract(res ContractResolver) error { } 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 @@ -594,3 +602,132 @@ func TestChannelArbitratorLocalForceDoubleSpend(t *testing.T) { t.Fatalf("contract was not resolved") } } + +// 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") + } +} From 1758ad829f37272904c1fcffb5c86b3f59b5a4e1 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:16 +0200 Subject: [PATCH 11/16] contractcourt/[chain|channel]_arbitrator: define IsPendingClose --- contractcourt/chain_arbitrator.go | 4 ++++ contractcourt/channel_arbitrator.go | 12 ++++++++++++ contractcourt/channel_arbitrator_test.go | 1 + 3 files changed, 17 insertions(+) diff --git a/contractcourt/chain_arbitrator.go b/contractcourt/chain_arbitrator.go index 5525c1857..b8491b069 100644 --- a/contractcourt/chain_arbitrator.go +++ b/contractcourt/chain_arbitrator.go @@ -244,6 +244,7 @@ func newActiveChannelArbitrator(channel *channeldb.OpenChannel, }, MarkCommitmentBroadcasted: channel.MarkCommitmentBroadcasted, MarkChannelClosed: channel.CloseChannel, + IsPendingClose: false, ChainArbitratorConfig: c.cfg, ChainEvents: chanEvents, } @@ -399,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/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index 9cd2fee20..f6ef30fb8 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -101,6 +101,18 @@ type ChannelArbitratorConfig struct { // 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 diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index 6f4b7f2e4..aeedf47c7 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -149,6 +149,7 @@ func createTestChannelArbitrator(log ArbitratorLog) (*ChannelArbitrator, MarkChannelClosed: func(*channeldb.ChannelCloseSummary) error { return nil }, + IsPendingClose: false, ChainArbitratorConfig: chainArbCfg, ChainEvents: chanEvents, } From f2a033e9656e8e47b0836a88f7300b8e7a4fd204 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:16 +0200 Subject: [PATCH 12/16] contractcourt/channel_arbitrator: override trigger on startup if IsClosing At ChannelArbitrator startup we now check the database close status of the channel. If we detect that the channel is closed, but our state machine hasn't advanced to reflect that (possibly because of a shutdown before the state transition was finished), we manually trigger the state transition to recover. --- contractcourt/channel_arbitrator.go | 32 ++++++++++++++++++++++++++--- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index f6ef30fb8..f803f806b 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -262,12 +262,38 @@ 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, - ) + nextState, _, err := c.advanceState(triggerHeight, trigger) if err != nil { c.cfg.BlockEpochs.Cancel() return err From 53286b8cee667a2a38b97564351d79d6a7bc42f0 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:21:16 +0200 Subject: [PATCH 13/16] contractcourt test: add TestChannelArbitratorCommitFailure --- contractcourt/channel_arbitrator_test.go | 100 ++++++++++++++++++++++- 1 file changed, 96 insertions(+), 4 deletions(-) diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index aeedf47c7..aa77cbaa3 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -14,10 +14,12 @@ import ( ) type mockArbitratorLog struct { - state ArbitratorState - newStates chan ArbitratorState - failLog bool - failFetch error + state ArbitratorState + newStates chan ArbitratorState + failLog bool + failFetch error + failCommit bool + failCommitState ArbitratorState } // A compile time check to ensure mockArbitratorLog meets the ArbitratorLog @@ -29,6 +31,10 @@ func (b *mockArbitratorLog) CurrentState() (ArbitratorState, error) { } 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 @@ -732,3 +738,89 @@ func TestChannelArbitratorPersistence(t *testing.T) { 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") + } +} From b031d87daf1d8e8306378f72b8f1db9f36ce9085 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:53:56 +0200 Subject: [PATCH 14/16] contractcourt/briefcase: test FetchContractResolutions error when no resolutions logged --- contractcourt/briefcase_test.go | 7 +++++++ 1 file changed, 7 insertions(+) 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 { From ed5db96e05f208c819f86e44ca0cd78a126d4d34 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:54:58 +0200 Subject: [PATCH 15/16] contractcourt/channel_arbittrator: ignore noResolutions error on startup --- contractcourt/channel_arbitrator.go | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index f803f806b..ba64eecc2 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -295,8 +295,23 @@ func (c *ChannelArbitrator) Start() error { startingState := c.state 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 From bb7185620475530c9701d23291feeb16df832e16 Mon Sep 17 00:00:00 2001 From: "Johan T. Halseth" Date: Tue, 21 Aug 2018 12:55:19 +0200 Subject: [PATCH 16/16] contractcourt/channel_arbitrator test: add TestChannelArbitratorEmptyResolutions --- contractcourt/channel_arbitrator_test.go | 43 ++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/contractcourt/channel_arbitrator_test.go b/contractcourt/channel_arbitrator_test.go index aa77cbaa3..b5a410734 100644 --- a/contractcourt/channel_arbitrator_test.go +++ b/contractcourt/channel_arbitrator_test.go @@ -824,3 +824,46 @@ func TestChannelArbitratorCommitFailure(t *testing.T) { 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() +}