From 55aee9c7034aca652b7c29fc8237e677b5007510 Mon Sep 17 00:00:00 2001 From: Joost Jager Date: Fri, 7 Sep 2018 16:05:57 +0200 Subject: [PATCH] cnct: expose non-incubating htlcs after channel force close In this commit we fix a reporting gap that previously existed for htlcs that were still contested. --- contractcourt/chain_arbitrator.go | 15 +++ contractcourt/channel_arbitrator.go | 99 +++++++++++++++++++ contractcourt/contract_resolvers.go | 8 ++ .../htlc_incoming_contest_resolver.go | 23 +++++ .../htlc_outgoing_contest_resolver.go | 25 +++++ lnd_test.go | 24 +++-- rpcserver.go | 48 +++++++++ 7 files changed, 235 insertions(+), 7 deletions(-) diff --git a/contractcourt/chain_arbitrator.go b/contractcourt/chain_arbitrator.go index 556fba2d4..23871fb78 100644 --- a/contractcourt/chain_arbitrator.go +++ b/contractcourt/chain_arbitrator.go @@ -566,6 +566,21 @@ func (c *ChainArbitrator) UpdateContractSignals(chanPoint wire.OutPoint, return nil } +// GetChannelArbitrator safely returns the channel arbitrator for a given +// channel outpoint. +func (c *ChainArbitrator) GetChannelArbitrator(chanPoint wire.OutPoint) ( + *ChannelArbitrator, error) { + + c.Lock() + arbitrator, ok := c.activeChannels[chanPoint] + c.Unlock() + if !ok { + return nil, fmt.Errorf("unable to find arbitrator") + } + + return arbitrator, nil +} + // forceCloseReq is a request sent from an outside sub-system to the arbitrator // that watches a particular channel to broadcast the commitment transaction, // and enter the resolution phase of the channel. diff --git a/contractcourt/channel_arbitrator.go b/contractcourt/channel_arbitrator.go index 4eb499470..bc47ebe59 100644 --- a/contractcourt/channel_arbitrator.go +++ b/contractcourt/channel_arbitrator.go @@ -1,11 +1,13 @@ package contractcourt import ( + "bytes" "errors" "sync" "sync/atomic" "github.com/btcsuite/btcd/wire" + "github.com/btcsuite/btcutil" "github.com/davecgh/go-spew/spew" "github.com/lightningnetwork/lnd/chainntnfs" "github.com/lightningnetwork/lnd/channeldb" @@ -132,6 +134,36 @@ type ChannelArbitratorConfig struct { ChainArbitratorConfig } +// ContractReport provides a summary of a commitment tx output. +type ContractReport struct { + // Outpoint is the final output that will be swept back to the wallet. + Outpoint wire.OutPoint + + // Incoming indicates whether the htlc was incoming to this channel. + Incoming bool + + // Amount is the final value that will be swept in back to the wallet. + Amount btcutil.Amount + + // MaturityHeight is the absolute block height that this output will + // mature at. + MaturityHeight uint32 + + // Stage indicates whether the htlc is in the CLTV-timeout stage (1) or + // the CSV-delay stage (2). A stage 1 htlc's maturity height will be set + // to its expiry height, while a stage 2 htlc's maturity height will be + // set to its confirmation height plus the maturity requirement. + Stage uint32 + + // LimboBalance is the total number of frozen coins within this + // contract. + LimboBalance btcutil.Amount + + // RecoveredBalance is the total value that has been successfully swept + // back to the user's wallet. + RecoveredBalance btcutil.Amount +} + // htlcSet represents the set of active HTLCs on a given commitment // transaction. type htlcSet struct { @@ -202,6 +234,10 @@ type ChannelArbitrator struct { // be able to signal them for shutdown in the case that we shutdown. activeResolvers []ContractResolver + // activeResolversLock prevents simultaneous read and write to the + // resolvers slice. + activeResolversLock sync.RWMutex + // resolutionSignal is a channel that will be sent upon by contract // resolvers once their contract has been fully resolved. With each // send, we'll check to see if the contract is fully resolved. @@ -461,6 +497,33 @@ func supplementTimeoutResolver(r *htlcTimeoutResolver, return nil } +// Report returns htlc reports for the active resolvers. +func (c *ChannelArbitrator) Report() []*ContractReport { + c.activeResolversLock.RLock() + defer c.activeResolversLock.RUnlock() + + var reports []*ContractReport + for _, resolver := range c.activeResolvers { + r, ok := resolver.(reportingContractResolver) + if !ok { + continue + } + + if r.IsResolved() { + continue + } + + report := r.report() + if report == nil { + continue + } + + reports = append(reports, report) + } + + return reports +} + // Stop signals the ChannelArbitrator for a graceful shutdown. func (c *ChannelArbitrator) Stop() error { if !atomic.CompareAndSwapInt32(&c.stopped, 0, 1) { @@ -473,9 +536,11 @@ func (c *ChannelArbitrator) Stop() error { go c.cfg.ChainEvents.Cancel() } + c.activeResolversLock.RLock() for _, activeResolver := range c.activeResolvers { activeResolver.Stop() } + c.activeResolversLock.RUnlock() close(c.quit) c.wg.Wait() @@ -854,6 +919,9 @@ func (c *ChannelArbitrator) stateStep(triggerHeight uint32, // launchResolvers updates the activeResolvers list and starts the resolvers. func (c *ChannelArbitrator) launchResolvers(resolvers []ContractResolver) { + c.activeResolversLock.Lock() + defer c.activeResolversLock.Unlock() + c.activeResolvers = resolvers for _, contract := range resolvers { c.wg.Add(1) @@ -1373,6 +1441,25 @@ func (c *ChannelArbitrator) prepContractResolutions(htlcActions ChainActionMap, return htlcResolvers, msgsToSend, nil } +// replaceResolver replaces a in the list of active resolvers. If the resolver +// to be replaced is not found, it returns an error. +func (c *ChannelArbitrator) replaceResolver(oldResolver, + newResolver ContractResolver) error { + + c.activeResolversLock.Lock() + defer c.activeResolversLock.Unlock() + + oldKey := oldResolver.ResolverKey() + for i, r := range c.activeResolvers { + if bytes.Equal(r.ResolverKey(), oldKey) { + c.activeResolvers[i] = newResolver + return nil + } + } + + return errors.New("resolver to be replaced not found") +} + // resolveContract is a goroutine tasked with fully resolving an unresolved // contract. Either the initial contract will be resolved after a single step, // or the contract will itself create another contract to be resolved. In @@ -1422,6 +1509,7 @@ func (c *ChannelArbitrator) resolveContract(currentContract ContractResolver) { c.cfg.ChanPoint, currentContract, nextContract) + // Swap contract in log. err := c.log.SwapContract( currentContract, nextContract, ) @@ -1430,6 +1518,17 @@ func (c *ChannelArbitrator) resolveContract(currentContract ContractResolver) { "contract: %v", err) } + // Swap contract in resolvers list. This is to + // make sure that reports are queried from the + // new resolver. + err = c.replaceResolver( + currentContract, nextContract, + ) + if err != nil { + log.Errorf("unable to replace "+ + "contract: %v", err) + } + // As this contract produced another, we'll // re-assign, so we can continue our resolution // loop. diff --git a/contractcourt/contract_resolvers.go b/contractcourt/contract_resolvers.go index 5b836ad2d..1a19e166a 100644 --- a/contractcourt/contract_resolvers.go +++ b/contractcourt/contract_resolvers.go @@ -60,6 +60,14 @@ type ContractResolver interface { Stop() } +// reportingContractResolver is a ContractResolver that also exposes a report on +// the resolution state of the contract. +type reportingContractResolver interface { + ContractResolver + + report() *ContractReport +} + // ResolverKit is meant to be used as a mix-in struct to be embedded within a // given ContractResolver implementation. It contains all the items that a // resolver requires to carry out its duties. diff --git a/contractcourt/htlc_incoming_contest_resolver.go b/contractcourt/htlc_incoming_contest_resolver.go index 56d25327b..ec6762289 100644 --- a/contractcourt/htlc_incoming_contest_resolver.go +++ b/contractcourt/htlc_incoming_contest_resolver.go @@ -6,6 +6,8 @@ import ( "encoding/binary" "fmt" "io" + + "github.com/btcsuite/btcutil" ) // htlcIncomingContestResolver is a ContractResolver that's able to resolve an @@ -166,6 +168,27 @@ func (h *htlcIncomingContestResolver) Resolve() (ContractResolver, error) { } } +// report returns a report on the resolution state of the contract. +func (h *htlcIncomingContestResolver) report() *ContractReport { + // No locking needed as these values are read-only. + + finalAmt := h.htlcAmt.ToSatoshis() + if h.htlcResolution.SignedSuccessTx != nil { + finalAmt = btcutil.Amount( + h.htlcResolution.SignedSuccessTx.TxOut[0].Value, + ) + } + + return &ContractReport{ + Outpoint: h.htlcResolution.ClaimOutpoint, + Incoming: true, + Amount: finalAmt, + MaturityHeight: h.htlcExpiry, + LimboBalance: finalAmt, + Stage: 1, + } +} + // Stop signals the resolver to cancel any current resolution processes, and // suspend. // diff --git a/contractcourt/htlc_outgoing_contest_resolver.go b/contractcourt/htlc_outgoing_contest_resolver.go index 826631ddb..6075b3799 100644 --- a/contractcourt/htlc_outgoing_contest_resolver.go +++ b/contractcourt/htlc_outgoing_contest_resolver.go @@ -6,6 +6,7 @@ import ( "io" "github.com/btcsuite/btcd/wire" + "github.com/btcsuite/btcutil" "github.com/davecgh/go-spew/spew" "github.com/lightningnetwork/lnd/chainntnfs" ) @@ -108,6 +109,9 @@ func (h *htlcOutgoingContestResolver) Resolve() (ContractResolver, error) { scriptToWatch []byte err error ) + + // TODO(joostjager): output already set properly in + // lnwallet.newOutgoingHtlcResolution? And script too? if h.htlcResolution.SignedTimeoutTx == nil { outPointToWatch = h.htlcResolution.ClaimOutpoint scriptToWatch = h.htlcResolution.SweepSignDesc.Output.PkScript @@ -235,6 +239,27 @@ func (h *htlcOutgoingContestResolver) Resolve() (ContractResolver, error) { } } +// report returns a report on the resolution state of the contract. +func (h *htlcOutgoingContestResolver) report() *ContractReport { + // No locking needed as these values are read-only. + + finalAmt := h.htlcAmt.ToSatoshis() + if h.htlcResolution.SignedTimeoutTx != nil { + finalAmt = btcutil.Amount( + h.htlcResolution.SignedTimeoutTx.TxOut[0].Value, + ) + } + + return &ContractReport{ + Outpoint: h.htlcResolution.ClaimOutpoint, + Incoming: false, + Amount: finalAmt, + MaturityHeight: h.htlcResolution.Expiry, + LimboBalance: finalAmt, + Stage: 1, + } +} + // Stop signals the resolver to cancel any current resolution processes, and // suspend. // diff --git a/lnd_test.go b/lnd_test.go index 65de40be9..07025a611 100644 --- a/lnd_test.go +++ b/lnd_test.go @@ -2803,6 +2803,12 @@ func testChannelForceClosure(net *lntest.NetworkHarness, t *harnessTest) { assertTxInBlock(t, block, sweepTx.Hash()) + // Update current height + _, curHeight, err = net.Miner.Node.GetBestBlock() + if err != nil { + t.Fatalf("unable to get best block height") + } + err = lntest.WaitPredicate(func() bool { // Now that the commit output has been fully swept, check to see // that the channel remains open for the pending htlc outputs. @@ -2824,21 +2830,25 @@ func testChannelForceClosure(net *lntest.NetworkHarness, t *harnessTest) { // The commitment funds will have been recovered after the // commit txn was included in the last block. The htlc funds - // will not be shown in limbo, since they are still in their - // first stage and the nursery hasn't received them from the - // contract court. + // will be shown in limbo. forceClose, err := findForceClosedChannel(pendingChanResp, &op) if err != nil { predErr = err return false } - predErr = checkPendingChannelNumHtlcs(forceClose, 0) + predErr = checkPendingChannelNumHtlcs(forceClose, numInvoices) if predErr != nil { return false } - if forceClose.LimboBalance != 0 { - predErr = fmt.Errorf("expected 0 funds in limbo, "+ - "found %d", forceClose.LimboBalance) + predErr = checkPendingHtlcStageAndMaturity( + forceClose, 1, htlcExpiryHeight, + int32(htlcExpiryHeight)-curHeight, + ) + if predErr != nil { + return false + } + if forceClose.LimboBalance == 0 { + predErr = fmt.Errorf("expected funds in limbo, found 0") return false } diff --git a/rpcserver.go b/rpcserver.go index e5f028a26..8f9e347a6 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -2063,6 +2063,11 @@ func (r *rpcServer) PendingChannels(ctx context.Context, ClosingTxid: closeTXID, } + // Fetch reports from both nursery and resolvers. At the + // moment this is not an atomic snapshot. This is + // planned to be resolved when the nursery is removed + // and channel arbitrator will be the single source for + // these kind of reports. err := r.nurseryPopulateForceCloseResp( &chanPoint, currentHeight, forceClose, ) @@ -2070,6 +2075,13 @@ func (r *rpcServer) PendingChannels(ctx context.Context, return nil, err } + err = r.arbitratorPopulateForceCloseResp( + &chanPoint, currentHeight, forceClose, + ) + if err != nil { + return nil, err + } + resp.TotalLimboBalance += int64(forceClose.LimboBalance) resp.PendingForceClosingChannels = append( @@ -2115,6 +2127,42 @@ func (r *rpcServer) PendingChannels(ctx context.Context, return resp, nil } +// arbitratorPopulateForceCloseResp populates the pending channels response +// message with channel resolution information from the contract resolvers. +func (r *rpcServer) arbitratorPopulateForceCloseResp(chanPoint *wire.OutPoint, + currentHeight int32, + forceClose *lnrpc.PendingChannelsResponse_ForceClosedChannel) error { + + // Query for contract resolvers state. + arbitrator, err := r.server.chainArb.GetChannelArbitrator(*chanPoint) + if err != nil { + return err + } + reports := arbitrator.Report() + + for _, report := range reports { + htlc := &lnrpc.PendingHTLC{ + Incoming: report.Incoming, + Amount: int64(report.Amount), + Outpoint: report.Outpoint.String(), + MaturityHeight: report.MaturityHeight, + Stage: report.Stage, + } + + if htlc.MaturityHeight != 0 { + htlc.BlocksTilMaturity = + int32(htlc.MaturityHeight) - currentHeight + } + + forceClose.LimboBalance += int64(report.LimboBalance) + forceClose.RecoveredBalance += int64(report.RecoveredBalance) + + forceClose.PendingHtlcs = append(forceClose.PendingHtlcs, htlc) + } + + return nil +} + // nurseryPopulateForceCloseResp populates the pending channels response // message with contract resolution information from utxonursery. func (r *rpcServer) nurseryPopulateForceCloseResp(chanPoint *wire.OutPoint,