lnwire+lnwallet+fundingmanager: general improvements

This commit is contained in:
Andrey Samokhvalov 2016-11-24 11:49:18 +03:00 committed by Olaoluwa Osuntokun
parent a6f7f05323
commit 5a82240c6a
13 changed files with 146 additions and 108 deletions

@ -1,19 +1,18 @@
package main
import (
"fmt"
"sync"
"sync/atomic"
"github.com/go-errors/errors"
"github.com/lightningnetwork/lnd/lnrpc"
"github.com/lightningnetwork/lnd/lnwallet"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/routing/rt/graph"
"github.com/roasbeef/btcd/btcec"
"github.com/roasbeef/btcd/txscript"
"github.com/roasbeef/btcd/wire"
"github.com/roasbeef/btcutil"
"github.com/lightningnetwork/lnd/routing/rt/graph"
"google.golang.org/grpc"
)
@ -306,7 +305,7 @@ func (f *fundingManager) processFundingRequest(msg *lnwire.SingleFundingRequest,
f.fundingMsgs <- &fundingRequestMsg{msg, peer}
}
// handleSingleFundingRequest creates an initial 'ChannelReservation' within
// handleFundingRequest creates an initial 'ChannelReservation' within
// the wallet, then responds to the source peer with a single funder response
// message progressing the funding workflow.
// TODO(roasbeef): add error chan to all, let channelManager handle
@ -413,11 +412,16 @@ func (f *fundingManager) processFundingResponse(msg *lnwire.SingleFundingRespons
// outpoint, and a commitment signature to the remote peer.
func (f *fundingManager) handleFundingResponse(fmsg *fundingResponseMsg) {
msg := fmsg.msg
peerID := fmsg.peer.id
chanID := fmsg.msg.ChannelID
sourcePeer := fmsg.peer
f.resMtx.RLock()
resCtx := f.activeReservations[fmsg.peer.id][msg.ChannelID]
f.resMtx.RUnlock()
resCtx, err := f.getReservationCtx(peerID, chanID)
if err != nil {
fndgLog.Warnf("can' find reservation (peerID:%v, chanID:%v)",
peerID, chanID)
return
}
fndgLog.Infof("Recv'd fundingResponse for pendingID(%v)", msg.ChannelID)
@ -463,14 +467,14 @@ func (f *fundingManager) handleFundingResponse(fmsg *fundingResponseMsg) {
// the peer's readHandler once the channel is open.
fmsg.peer.barrierInits <- *outPoint
fndgLog.Infof("Generated ChannelPoint(%v) for pendingID(%v)",
outPoint, msg.ChannelID)
fndgLog.Infof("Generated ChannelPoint(%v) for pendingID(%v)", outPoint,
chanID)
revocationKey := resCtx.reservation.OurContribution().RevocationKey
obsfucator := resCtx.reservation.StateNumObfuscator()
fundingComplete := lnwire.NewSingleFundingComplete(msg.ChannelID,
outPoint, commitSig, revocationKey, obsfucator)
fundingComplete := lnwire.NewSingleFundingComplete(chanID, outPoint,
commitSig, revocationKey, obsfucator)
sourcePeer.queueMsg(fundingComplete, nil)
}
@ -485,9 +489,12 @@ func (f *fundingManager) processFundingComplete(msg *lnwire.SingleFundingComplet
// processed, a signature is sent to the remote peer allowing it to broadcast
// the funding transaction, progressing the workflow into the final stage.
func (f *fundingManager) handleFundingComplete(fmsg *fundingCompleteMsg) {
f.resMtx.RLock()
resCtx := f.activeReservations[fmsg.peer.id][fmsg.msg.ChannelID]
f.resMtx.RUnlock()
resCtx, err := f.getReservationCtx(fmsg.peer.id, fmsg.msg.ChannelID)
if err != nil {
fndgLog.Warnf("can' find reservation (peerID:%v, chanID:%v)",
fmsg.peer.id, fmsg.msg.ChannelID)
return
}
// The channel initiator has responded with the funding outpoint of the
// final funding transaction, as well as a signature for our version of
@ -507,8 +514,8 @@ func (f *fundingManager) handleFundingComplete(fmsg *fundingCompleteMsg) {
// With all the necessary data available, attempt to advance the
// funding workflow to the next stage. If this succeeds then the
// funding transaction will broadcast after our next message.
err := resCtx.reservation.CompleteReservationSingle(revokeKey,
fundingOut, commitSig, obsfucator)
err = resCtx.reservation.CompleteReservationSingle(revokeKey, fundingOut,
commitSig, obsfucator)
if err != nil {
// TODO(roasbeef): better error logging: peerID, channelID, etc.
fndgLog.Errorf("unable to complete single reservation: %v", err)
@ -551,10 +558,14 @@ func (f *fundingManager) processFundingSignComplete(msg *lnwire.SingleFundingSig
// proofs of transaction inclusion.
func (f *fundingManager) handleFundingSignComplete(fmsg *fundingSignCompleteMsg) {
chanID := fmsg.msg.ChannelID
peerID := fmsg.peer.id
f.resMtx.RLock()
resCtx := f.activeReservations[fmsg.peer.id][chanID]
f.resMtx.RUnlock()
resCtx, err := f.getReservationCtx(peerID, chanID)
if err != nil {
fndgLog.Warnf("can' find reservation (peerID:%v, chanID:%v)",
peerID, chanID)
return
}
// The remote peer has responded with a signature for our commitment
// transaction. We'll verify the signature for validity, then commit
@ -594,12 +605,10 @@ func (f *fundingManager) handleFundingSignComplete(fmsg *fundingSignCompleteMsg)
case openChan := <-resCtx.reservation.DispatchChan():
// This reservation is no longer pending as the funding
// transaction has been fully confirmed.
f.resMtx.Lock()
delete(f.activeReservations[fmsg.peer.id], chanID)
f.resMtx.Unlock()
f.deleteReservationCtx(peerID, chanID)
fndgLog.Infof("ChannelPoint(%v) with peerID(%v) is now active",
fundingPoint, fmsg.peer.id)
fundingPoint, peerID)
// Now that the channel is open, we need to notify a
// number of parties of this event.
@ -669,9 +678,15 @@ func (f *fundingManager) processFundingOpenProof(msg *lnwire.SingleFundingOpenPr
// the initiating node is verified, which if correct, marks the channel as open
// to the source peer.
func (f *fundingManager) handleFundingOpen(fmsg *fundingOpenMsg) {
f.resMtx.RLock()
resCtx := f.activeReservations[fmsg.peer.id][fmsg.msg.ChannelID]
f.resMtx.RUnlock()
chanID := fmsg.msg.ChannelID
peerID := fmsg.peer.id
resCtx, err := f.getReservationCtx(peerID, chanID)
if err != nil {
fndgLog.Warnf("can' find reservation (peerID:%v, chanID:%v)",
peerID, chanID)
return
}
// The channel initiator has claimed the channel is now open, so we'll
// verify the contained SPV proof for validity.
@ -690,12 +705,10 @@ func (f *fundingManager) handleFundingOpen(fmsg *fundingOpenMsg) {
// The reservation has been completed, therefore we can stop tracking
// it within our active reservations map.
f.resMtx.Lock()
delete(f.activeReservations[fmsg.peer.id], fmsg.msg.ChannelID)
f.resMtx.Unlock()
f.deleteReservationCtx(peerID, chanID)
fndgLog.Infof("FundingOpen: ChannelPoint(%v) with peerID(%v) is now open",
resCtx.reservation.FundingOutpoint, fmsg.peer.id)
resCtx.reservation.FundingOutpoint, peerID)
// Notify the L3 routing manager of the newly active channel link.
capacity := int64(resCtx.reservation.OurContribution().FundingAmount +
@ -818,37 +831,67 @@ func (f *fundingManager) processErrorGeneric(err *lnwire.ErrorGeneric,
// depends on the type of error we should do different clean up steps and
// inform user about it.
func (f *fundingManager) handleErrorGenericMsg(fmsg *fundingErrorMsg) {
switch fmsg.err.Code {
e := fmsg.err
switch e.Code {
case lnwire.ErrorMaxPendingChannels:
peerID := fmsg.peer.id
chanID := fmsg.err.PendingChannelID
f.resMtx.RLock()
resCtx, ok := f.activeReservations[peerID][chanID]
f.resMtx.RUnlock()
if !ok {
fndgLog.Warnf("ErrorGeneric error was returned from " +
"remote peer for unknown channel (id: %v)")
if ctx, err := f.cancelReservationCtx(peerID, chanID); err != nil {
fndgLog.Warnf("unable to delete reservation: %v", err)
return
} else {
ctx.err <- grpc.Errorf(e.Code.ToGrpcCode(), e.Problem)
return
}
if err := resCtx.reservation.Cancel(); err != nil {
resCtx.err <- fmt.Errorf("max pending channels "+
"exceeded -- unable to cancel reservation: %v",
err)
} else {
resCtx.err <- grpc.Errorf(ErrorMaxPendingChannels,
"unable to create channel, max number of "+
"pending channels exceeded.")
}
// TODO(roasbeef): possibly cancel funding barrier in peer's
// channelManager?
f.resMtx.Lock()
delete(f.activeReservations[peerID], chanID)
f.resMtx.Unlock()
default:
fndgLog.Warnf("unknown funding error %v", fmsg.err)
fndgLog.Warnf("unknown funding error (%v:%v)", e.Code, e.Problem)
}
}
// cancelReservationCtx do all needed work in order to securely cancel the
// reservation.
func (f *fundingManager) cancelReservationCtx(peerID int32,
chanID uint64) (*reservationWithCtx, error) {
ctx, err := f.getReservationCtx(peerID, chanID)
if err != nil {
return nil, errors.Errorf("can't find reservation: %v",
err)
}
if err := ctx.reservation.Cancel(); err != nil {
ctx.err <- err
return nil, errors.Errorf("can't cancel reservation: %v",
err)
}
f.deleteReservationCtx(peerID, chanID)
return ctx, nil
}
// deleteReservationCtx is needed in order to securely delete the reservation.
func (f *fundingManager) deleteReservationCtx(peerID int32, chanID uint64) {
// TODO(roasbeef): possibly cancel funding barrier in peer's
// channelManager?
f.resMtx.Lock()
delete(f.activeReservations[peerID], chanID)
f.resMtx.Unlock()
}
// getReservationCtx returns the reservation context by peer id and channel id.
func (f *fundingManager) getReservationCtx(peerID int32,
chanID uint64) (*reservationWithCtx, error) {
f.resMtx.RLock()
resCtx, ok := f.activeReservations[peerID][chanID]
f.resMtx.RUnlock()
if !ok {
return nil, errors.Errorf("unknown channel (id: %v)", chanID)
}
return resCtx, nil
}

@ -16,6 +16,7 @@ import (
"github.com/davecgh/go-spew/spew"
"github.com/go-errors/errors"
"github.com/lightningnetwork/lnd/lnrpc"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/roasbeef/btcd/rpctest"
"github.com/roasbeef/btcd/wire"
"github.com/roasbeef/btcrpcclient"
@ -886,7 +887,7 @@ func testMaxPendingChannels(net *networkHarness, t *harnessTest) {
_, err = net.OpenChannel(ctx, net.Alice, carol, amount, 1)
if err == nil {
t.Fatalf("error wasn't received")
} else if grpc.Code(err) != ErrorMaxPendingChannels {
} else if grpc.Code(err) != lnwire.ErrorMaxPendingChannels.ToGrpcCode() {
t.Fatalf("not expected error was received: %v", err)
}
@ -1243,6 +1244,22 @@ func TestLightningNetworkDaemon(t *testing.T) {
OnTxAccepted: lndHarness.OnTxAccepted,
}
// Spawn a new goroutine to watch for any fatal errors that any of the
// running lnd processes encounter. If an error occurs, then the test
// fails immediately with a fatal error, as far as fatal is happening
// inside goroutine main goroutine would not be finished at the same
// time as we receive fatal error from lnd process.
testsFin := make(chan struct{})
go func() {
select {
case err := <-lndHarness.ProcessErrors():
ht.Fatalf("lnd finished with error (stderr): "+
"\n%v", err)
case <-testsFin:
return
}
}()
// First create an instance of the btcd's rpctest.Harness. This will be
// used to fund the wallets of the nodes within the test network and to
// drive blockchain related events within the network.
@ -1269,22 +1286,6 @@ func TestLightningNetworkDaemon(t *testing.T) {
ht.Fatalf("unable to set up test lightning network: %v", err)
}
// Spawn a new goroutine to watch for any fatal errors that any of the
// running lnd processes encounter. If an error occurs, then the test
// fails immediately with a fatal error, as far as fatal is happening
// inside goroutine main goroutine would not be finished at the same
// time as we receive fatal error from lnd process.
testsFin := make(chan struct{})
go func() {
select {
case err := <-lndHarness.ProcessErrors():
ht.Fatalf("lnd finished with error (stderr): "+
"\n%v", err)
case <-testsFin:
return
}
}()
t.Logf("Running %v integration tests", len(testsCases))
for _, testCase := range testsCases {
ht.RunTestCase(testCase, lndHarness)

@ -1585,7 +1585,7 @@ func (lc *LightningChannel) AddHTLC(htlc *lnwire.HTLCAddRequest) (uint32, error)
EntryType: Add,
RHash: PaymentHash(htlc.RedemptionHashes[0]),
Timeout: htlc.Expiry,
Amount: btcutil.Amount(htlc.Amount),
Amount: htlc.Amount,
Index: lc.ourLogCounter,
}
@ -1611,7 +1611,7 @@ func (lc *LightningChannel) ReceiveHTLC(htlc *lnwire.HTLCAddRequest) (uint32, er
EntryType: Add,
RHash: PaymentHash(htlc.RedemptionHashes[0]),
Timeout: htlc.Expiry,
Amount: btcutil.Amount(htlc.Amount),
Amount: htlc.Amount,
Index: lc.theirLogCounter,
}

@ -355,7 +355,7 @@ func TestSimpleAddSettleWorkflow(t *testing.T) {
htlc := &lnwire.HTLCAddRequest{
RedemptionHashes: [][32]byte{paymentHash},
// TODO(roasbeef): properly switch to credits: (1 msat)
Amount: lnwire.CreditsAmount(1e8),
Amount: btcutil.Amount(1e8),
Expiry: uint32(5),
}
@ -628,7 +628,7 @@ func TestCheckCommitTxSize(t *testing.T) {
return &lnwire.HTLCAddRequest{
RedemptionHashes: [][32]byte{paymentHash},
Amount: lnwire.CreditsAmount(1e7),
Amount: btcutil.Amount(1e7),
Expiry: uint32(5),
}, returnPreimage
}
@ -742,7 +742,7 @@ func TestCheckHTLCNumberConstraint(t *testing.T) {
paymentHash := fastsha256.Sum256(preimage)
return &lnwire.HTLCAddRequest{
RedemptionHashes: [][32]byte{paymentHash},
Amount: lnwire.CreditsAmount(1e7),
Amount: btcutil.Amount(1e7),
Expiry: uint32(5),
}
}
@ -864,7 +864,7 @@ func TestStateUpdatePersistence(t *testing.T) {
rHash := fastsha256.Sum256(alicePreimage[:])
h := &lnwire.HTLCAddRequest{
RedemptionHashes: [][32]byte{rHash},
Amount: lnwire.CreditsAmount(1000),
Amount: btcutil.Amount(1000),
Expiry: uint32(10),
}
@ -874,7 +874,7 @@ func TestStateUpdatePersistence(t *testing.T) {
rHash := fastsha256.Sum256(bobPreimage[:])
bobh := &lnwire.HTLCAddRequest{
RedemptionHashes: [][32]byte{rHash},
Amount: lnwire.CreditsAmount(1000),
Amount: btcutil.Amount(1000),
Expiry: uint32(10),
}
bobChannel.AddHTLC(bobh)

@ -908,12 +908,6 @@ func testSingleFunderReservationWorkflowResponder(miner *rpctest.Harness,
// TODO(roasbeef): bob verify alice's sig
}
func testFundingReservationInvalidCounterpartySigs(miner *rpctest.Harness, lnwallet *lnwallet.LightningWallet, t *testing.T) {
}
func testFundingTransactionTxFees(miner *rpctest.Harness, lnwallet *lnwallet.LightningWallet, t *testing.T) {
}
func testListTransactionDetails(miner *rpctest.Harness, wallet *lnwallet.LightningWallet, t *testing.T) {
t.Log("Running list transaction details test")
@ -1234,10 +1228,10 @@ var walletTests = []func(miner *rpctest.Harness, w *lnwallet.LightningWallet, te
testSingleFunderReservationWorkflowResponder,
testFundingTransactionLockedOutputs,
testFundingCancellationNotEnoughFunds,
testFundingReservationInvalidCounterpartySigs,
testTransactionSubscriptions,
testListTransactionDetails,
testSignOutputPrivateTweak,
testCancelNonExistantReservation,
}
type testLnWallet struct {
@ -1324,28 +1318,28 @@ func TestLightningWallet(t *testing.T) {
}
// Funding via 20 outputs with 4BTC each.
lnwallet, err := createTestWallet(tempTestDir, miningNode, netParams,
lnw, err := createTestWallet(tempTestDir, miningNode, netParams,
chainNotifier, wc, signer, bio)
if err != nil {
t.Fatalf("unable to create test ln wallet: %v", err)
}
// The wallet should now have 80BTC available for spending.
assertProperBalance(t, lnwallet, 1, 80)
assertProperBalance(t, lnw, 1, 80)
// Execute every test, clearing possibly mutated wallet state after
// each step.
for _, walletTest := range walletTests {
walletTest(miningNode, lnwallet, t)
walletTest(miningNode, lnw, t)
// TODO(roasbeef): possible reset mining node's chainstate to
// initial level, cleanly wipe buckets
if err := clearWalletState(lnwallet); err != nil &&
if err := clearWalletState(lnw); err != nil &&
err != bolt.ErrBucketNotFound {
t.Fatalf("unable to wipe wallet state: %v", err)
}
}
lnwallet.Shutdown()
lnw.Shutdown()
}
}

@ -12,7 +12,7 @@ import (
// ChannelContribution is the primary constituent of the funding workflow within
// lnwallet. Each side first exchanges their respective contributions along with
// channel specific paramters like the min fee/KB. Once contributions have been
// channel specific parameters like the min fee/KB. Once contributions have been
// exchanged, each side will then produce signatures for all their inputs to the
// funding transactions, and finally a signature for the other party's version
// of the commitment transaction.
@ -98,10 +98,6 @@ type ChannelReservation struct {
// fundingTx is the funding transaction for this pending channel.
fundingTx *wire.MsgTx
// For CLTV it is nLockTime, for CSV it's nSequence, for segwit it's
// not needed
fundingLockTime uint32
// In order of sorted inputs. Sorting is done in accordance
// to BIP-69: https://github.com/bitcoin/bips/blob/master/bip-0069.mediawiki.
ourFundingInputScripts []*InputScript
@ -228,7 +224,7 @@ func (r *ChannelReservation) OurContribution() *ChannelContribution {
return r.ourContribution
}
// ProcesContribution verifies the counterparty's contribution to the pending
// ProcessContribution verifies the counterparty's contribution to the pending
// payment channel. As a result of this incoming message, lnwallet is able to
// build the funding transaction, and both commitment transactions. Once this
// message has been processed, all signatures to inputs to the funding

@ -768,7 +768,6 @@ func (l *LightningWallet) handleContributionMsg(req *addContributionMsg) {
// With the funding tx complete, create both commitment transactions.
// TODO(roasbeef): much cleanup + de-duplication
pendingReservation.fundingLockTime = theirContribution.CsvDelay
ourBalance := ourContribution.FundingAmount
theirBalance := theirContribution.FundingAmount
ourCommitKey := ourContribution.CommitKey

@ -5,12 +5,21 @@ import (
"io"
"github.com/roasbeef/btcd/wire"
"google.golang.org/grpc/codes"
)
// ErrorCode represents the short error code for each of the defined errors
// within the Lightning Network protocol spec.
type ErrorCode uint16
// ToGrpcCode is used to generate gRPC specific code which will be propagated
// to the ln rpc client. This code is used to have more detailed view of what
// goes wrong and also in order to have the ability pragmatically determine
// the error and take specific actions on the client side.
func (e ErrorCode) ToGrpcCode() codes.Code {
return (codes.Code)(e) + 100
}
const (
// ErrorMaxPendingChannels is returned by remote peer when the number
// of active pending channels exceeds their maximum policy limit.

@ -5,6 +5,7 @@ import (
"io"
"github.com/roasbeef/btcd/wire"
"github.com/roasbeef/btcutil"
)
// HTLCAddRequest is the message sent by Alice to Bob when she wishes to add an
@ -28,7 +29,7 @@ type HTLCAddRequest struct {
// Difference between hop and first item in blob is the fee to complete
// Amount is the number of credits this HTLC is worth.
Amount CreditsAmount
Amount btcutil.Amount
// RefundContext is for payment cancellation
// TODO(j): not currently in use, add later

@ -4,6 +4,7 @@ import (
"bytes"
"reflect"
"testing"
"github.com/roasbeef/btcutil"
)
func TestHTLCAddRequestEncodeDecode(t *testing.T) {
@ -14,7 +15,7 @@ func TestHTLCAddRequestEncodeDecode(t *testing.T) {
addReq := &HTLCAddRequest{
ChannelPoint: outpoint1,
Expiry: uint32(144),
Amount: CreditsAmount(123456000),
Amount: btcutil.Amount(123456000),
ContractType: uint8(17),
RedemptionHashes: redemptionHashes,
OnionBlob: []byte{255, 0, 255, 0, 255, 0, 255, 0},

@ -32,7 +32,7 @@ type HTLCKey int64
// HTLC lists on either side will increment this height. As a result this value
// should always be monotonically increasing. Any CommitSignature or
// CommitRevocation messages will reference a value for the commitment height
// up to which it covers. HTLC's are only explicltly excluded by sending
// up to which it covers. HTLC's are only explicitly excluded by sending
// HTLCReject messages referencing a particular HTLCKey.
type CommitHeight uint64

@ -1532,7 +1532,7 @@ func logEntryToHtlcPkt(chanPoint wire.OutPoint,
}
msg = &lnwire.HTLCAddRequest{
Amount: lnwire.CreditsAmount(pd.Amount),
Amount: btcutil.Amount(pd.Amount),
RedemptionHashes: [][32]byte{pd.RHash},
OnionBlob: b.Bytes(),
}

@ -29,12 +29,6 @@ import (
"golang.org/x/net/context"
)
const (
// ErrorMaxPendingChannels is an additional gRPC error, which is
// returned if max pending channel restriction was violated.
ErrorMaxPendingChannels = 100
)
var (
defaultAccount uint32 = waddrmgr.DefaultAccountNum
)
@ -771,7 +765,7 @@ func (r *rpcServer) constructPaymentRoute(destPubkey []byte, amt int64,
// meta-data within this packet will be used to route the payment
// through the network.
htlcAdd := &lnwire.HTLCAddRequest{
Amount: lnwire.CreditsAmount(amt),
Amount: btcutil.Amount(amt),
RedemptionHashes: [][32]byte{rHash},
OnionBlob: sphinxPacket,
}