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

@@ -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.

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.

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.

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.
 //

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.
 //

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
 		}
 

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,