From b81cd319b2198864602e3857e0287ffe5edaba31 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 19 Mar 2021 11:50:00 +0100 Subject: [PATCH 01/13] docs: describe sidecar channels With this commit we add a first description and HOWTO about what sidecar channels are and how they can be created. We add this commit first to give any reviewer the full context about what we're trying to achieve first. --- docs/SUMMARY.md | 1 + docs/sidecar_channels.md | 94 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+) create mode 100644 docs/sidecar_channels.md diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index 452d7cf..973d5e4 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -8,5 +8,6 @@ * [Orders](orders.md) * [Channel Leases](channel_leases.md) * [Batch Execution](batch_execution.md) +* [Sidecar Channels](sidecar_channels.md) * [FAQs](faq.md) diff --git a/docs/sidecar_channels.md b/docs/sidecar_channels.md new file mode 100644 index 0000000..995d3bd --- /dev/null +++ b/docs/sidecar_channels.md @@ -0,0 +1,94 @@ +# Sidecar Channels + +## Goal + +Alice is a new user of LN and has no UTXOs but wants to use Lightning. She’s +willing to pay an onboarding fee to a wallet provider to get set up with both an +initial balance (=outbound capacity) and the ability to receive payments +(=inbound capacity). + +The example in this document assumes that Alice is aware of the costs of +onboarding and is paying for them herself (which happens out of band from the +perspective of the Pool server and is only mentioned explicitly for the +completeness of the example). + +With the APIs proposed to be implemented in the Pool server and client, it +should also be possible for a wallet provider to implement the flow in a way +that they cover the full costs of the onboarding and therefore “gift” the +channel and the balance to Alice. That would hide some of the parameters the API +needs to work, therefore we show the more involved example here. + +## Participants + +* Alice: New user, possibly using a mobile wallet, has no UTXOs but a wallet app + that has lnd and Pool integrated as part of a unified (mobile) binary (only + runs the Pool daemon, possibly even a "Pool light" daemon, does not need her + own Pool account). +* Charlie: A wallet/service provider that offers the sidecar bootstrap service + for a fee. Offers to convert fiat or BTC into an initial LN balance for new + users, minus said fee. Has a well funded Pool account that can be used to pay + for leases. +* Bob: An independent liquidity provider that is willing to open channels to new + users. Has a well funded Pool account that can be used to open channels. + +## Requirements + +* All three participants need to run the Pool client that implements the sidecar + functionality: + * Alice: Must be able to register with the auctioneer by providing the + multisig pubkey and be able to sign with it (explained later on). + * Charlie: Must be able to submit an order that refers to a multisig pubkey + of the receiver node. + * Bob: Must be able to accept orders that require to set a push amount when + opening the channel. +* No new functionality in lnd is required (though can be optimized later by + replacing the channel acceptor with a new push amount parameter in the funding + shim). + +## Example flow (command line only) + +1. Precondition: An agreement between Alice and Charlie is reached out-of-band. + For this example, we assume Charlie is going to buy **10 units** (1m satoshi) + of capacity from Pool for Alice. Of those 10 units, **2 units will be + pushed** to Alice for her to gain outbound liquidity as well. + The cost for the order submission fees, chain fees, the premium and the push + amount are all paid from Charlie's Pool account. Whether Alice reimburses + Charlie for those costs or not is not part of the protocol and irrelevant for + this example. +1. ```shell + charlie$ pool sidecar offer --capacity 1000000 --self_chan_balance 200000 + { + "sidecar_ticket": "sidecarAAQgHBgUEAwIBAAMBYwUBAgdxAAAAAAAAAwkAAAAAAA...." + } + ``` + The sidecar ticket now contains the offer from Charlie to buy a 1m satoshi + channel with a 200k initial self channel balance (=push amount) for Alice. +1. Alice now needs to add her node's information: + ```shell + alice$ pool sidecar register sidecarAAQgHBgUEAwIBAAMBYwUBAgdxAAAAA.... + { + "sidecar_ticket": "sidecarAAQgHBgUEAwIBAAMBYwUBAgdxAAAAAAAAAwkAAAAAAA...." + } + ``` + The sidecar ticket now contains the `Recipient` field which encodes Alice's + node pubkey and pubkey that'll be used for the channel funding. + + Rationale: Alice will be counter-signing the commitment transactions for the + new channel so she needs to have the private key for her 2-of-2 multisig + channel funding key. We derive that from her keychain. +1. Charlie can now create the bid order with the updated ticket he got from + Alice: + ```shell + charlie$ pool orders submit bid --acct_key \ + --interest_rate_percent xxx --sidecar_ticket sidecarAAQgHBgUEAwIBAAMB... + { + "order_nonce": "0011223344...", + "sidecar_ticket": "sidecarAAQgHBgUEAwIBAAMBYwUBAgdxAAAAAAAAAwkAAAAAAA..." + } + ``` + The sidecar ticket now contains the order's nonce in the `Order` part. + Charlie can give the final version of the ticket back to Alice. +1. Alice instructs her node to start expecting a channel. + ```shell + alice$ pool sidecar expect-channel sidecarAAQgHBgUEAwIBAAMBYwUBAg... + ``` From 363ddd5a8ebef1db717f69bc4fef6c92e281ffa1 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 19 Mar 2021 11:50:01 +0100 Subject: [PATCH 02/13] sidecar: add multisig key index and lease duration For proper and safe channel acceptance the recipient node also needs to know which key index it used for a ticket's multisig key so it can sign the authentication flow with the auctioneer correctly (which takes a key locator and not key descriptor). To make the offer and sidecar contract more explicit, we also include the lease duration in blocks in the offer part. This field will overwrite any value set when using the CLI to submit the bid order. --- sidecar/codec_test.go | 25 +++++++++++++------------ sidecar/interface.go | 16 ++++++++++++---- sidecar/tlv.go | 25 ++++++++++++++++++++----- 3 files changed, 45 insertions(+), 21 deletions(-) diff --git a/sidecar/codec_test.go b/sidecar/codec_test.go index fc27076..5074976 100644 --- a/sidecar/codec_test.go +++ b/sidecar/codec_test.go @@ -9,15 +9,15 @@ import ( ) var ( - hardcodedTicket = "sidecarAAQgHBgUEAwIBAAIBYwMBAgp5CwgAAAAAAAADCQwIAA" + - "AAAAAAA3gNIQLVLm5gAB7Vh7eEvz8Y3CkF2DsvNuSWJj8oOxp1iSh5xw5AAA" + - "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACwAAAAAAAAAAAAAAAAAAA" + - "AAAAAAAAAAAAAAAAAAAAAAFhRGFSEC1S5uYAAe1Ye3hL8_GNwpBdg7Lzbkli" + - "Y_KDsadYkoeccWIQMZ1hb2o3OyT-XUX7KWS-pAVBloIPQe8aCTqcjiNOV89x" + - "5kHyALFiEsAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACBAAAAAAAAAAA" + - "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAGMAAAAAAAAAAAAAAAAAAAAAAAAAAA" + - "AAAAAAAAAAAAAAISgiKSBjWE0AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + - "AAAA3VE-c=" + hardcodedTicket = "sidecarAAQgHBgUEAwIBAAIBYwMBAgp_CwgAAAAAAAADCQwIAA" + + "AAAAAAA3gNBAAAB-AOIQLVLm5gAB7Vh7eEvz8Y3CkF2DsvNuSWJj8oOxp1iS" + + "h5xw9AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACwAAAAAAAAAAA" + + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAFhRMFSEC1S5uYAAe1Ye3hL8_GNwpBd" + + "g7LzbkliY_KDsadYkoeccWIQMZ1hb2o3OyT-XUX7KWS-pAVBloIPQe8aCTqc" + + "jiNOV89xcEAAAAAB5kHyALFiEsAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" + + "AAACBAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAGMAAAAAAAAAAA" + + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAISgiKSBjWE0AAAAAAAAAAAAAAAAAAA" + + "AAAAAAAAAAAAAAAAAAAHcU6Y4=" ) // TestEncodeDecode tests that a ticket can be encoded and decoded from/to a @@ -29,9 +29,10 @@ func TestEncodeDecode(t *testing.T) { Version: Version(99), State: StateRegistered, Offer: Offer{ - Capacity: 777, - PushAmt: 888, - SignPubKey: testPubKey, + Capacity: 777, + PushAmt: 888, + LeaseDurationBlocks: 2016, + SignPubKey: testPubKey, SigOfferDigest: &btcec.Signature{ R: new(big.Int).SetInt64(44), S: new(big.Int).SetInt64(22), diff --git a/sidecar/interface.go b/sidecar/interface.go index 9e49ad6..e0a7f85 100644 --- a/sidecar/interface.go +++ b/sidecar/interface.go @@ -102,6 +102,10 @@ type Offer struct { // reduce the matching chances somewhat. PushAmt btcutil.Amount + // LeaseDurationBlocks is the number of blocks the offered channel in + // this offer would be leased for. + LeaseDurationBlocks uint32 + // SignPubKey is the public key for corresponding to the private key // that signed the SigOfferDigest below and, in a later state, the // SigOrderDigest of the Order struct. @@ -124,6 +128,9 @@ type Recipient struct { // multisig keys of the channel funding transaction output and is // advertised in the bid order. MultiSigPubKey *btcec.PublicKey + + // MultiSigKeyIndex is the derivation index of the MultiSigPubKey. + MultiSigKeyIndex uint32 } // Order is a struct holding the information about the sidecar bid order after @@ -189,15 +196,16 @@ type Ticket struct { // NewTicket creates a new sidecar ticket with the given version and offer // information. func NewTicket(version Version, capacity, pushAmt btcutil.Amount, - offerPubKey *btcec.PublicKey) (*Ticket, error) { + duration uint32, offerPubKey *btcec.PublicKey) (*Ticket, error) { t := &Ticket{ Version: version, State: StateOffered, Offer: Offer{ - Capacity: capacity, - PushAmt: pushAmt, - SignPubKey: offerPubKey, + Capacity: capacity, + PushAmt: pushAmt, + LeaseDurationBlocks: duration, + SignPubKey: offerPubKey, }, } diff --git a/sidecar/tlv.go b/sidecar/tlv.go index 21304ea..2724e79 100644 --- a/sidecar/tlv.go +++ b/sidecar/tlv.go @@ -19,12 +19,14 @@ const ( offerType tlv.Type = 10 capacityType tlv.Type = 11 pushAmtType tlv.Type = 12 - signPubKeyType tlv.Type = 13 - sigOfferDigestType tlv.Type = 14 + leaseDurationType tlv.Type = 13 + signPubKeyType tlv.Type = 14 + sigOfferDigestType tlv.Type = 15 - recipientType tlv.Type = 20 - nodePubKeyType tlv.Type = 21 - multiSigPubKeyType tlv.Type = 22 + recipientType tlv.Type = 20 + nodePubKeyType tlv.Type = 21 + multiSigPubKeyType tlv.Type = 22 + multiSigKeyIndexType tlv.Type = 23 orderType tlv.Type = 30 bidNonceType tlv.Type = 31 @@ -177,6 +179,9 @@ func serializeOffer(o Offer) ([]byte, error) { tlvRecords := []tlv.Record{ tlv.MakePrimitiveRecord(capacityType, &capacity), tlv.MakePrimitiveRecord(pushAmtType, &pushAmt), + tlv.MakePrimitiveRecord( + leaseDurationType, &o.LeaseDurationBlocks, + ), } if o.SignPubKey != nil { @@ -206,6 +211,9 @@ func deserializeOffer(offerBytes []byte) (Offer, error) { offerBytes, tlv.MakePrimitiveRecord(capacityType, &capacity), tlv.MakePrimitiveRecord(pushAmtType, &pushAmt), + tlv.MakePrimitiveRecord( + leaseDurationType, &o.LeaseDurationBlocks, + ), tlv.MakePrimitiveRecord(signPubKeyType, &o.SignPubKey), tlv.MakeStaticRecord( sigOfferDigestType, &o.SigOfferDigest, 64, ESig, DSig, @@ -236,6 +244,10 @@ func serializeRecipient(r Recipient) ([]byte, error) { )) } + tlvRecords = append(tlvRecords, tlv.MakePrimitiveRecord( + multiSigKeyIndexType, &r.MultiSigKeyIndex, + )) + return encodeBytes(tlvRecords...) } @@ -247,6 +259,9 @@ func deserializeRecipient(recipientBytes []byte) (Recipient, error) { recipientBytes, tlv.MakePrimitiveRecord(nodePubKeyType, &r.NodePubKey), tlv.MakePrimitiveRecord(multiSigPubKeyType, &r.MultiSigPubKey), + tlv.MakePrimitiveRecord( + multiSigKeyIndexType, &r.MultiSigKeyIndex, + ), ) } From f84debaeff734f989e48fe5db1bfb6d27df86a62 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 19 Mar 2021 11:50:02 +0100 Subject: [PATCH 03/13] sidecar+clientdb: add sidecar storage Both the trader offering to buy a sidecar channel for another node as well as the recipient node will need to store the involved sidecar tickets in their local databases. We add a simple store interface and its implementation that will be used in future commits. --- clientdb/db.go | 4 + clientdb/sidecar.go | 173 +++++++++++++++++++++++++++++++++++++++ clientdb/sidecar_test.go | 72 ++++++++++++++++ sidecar/interface.go | 18 ++++ 4 files changed, 267 insertions(+) create mode 100644 clientdb/sidecar.go create mode 100644 clientdb/sidecar_test.go diff --git a/clientdb/db.go b/clientdb/db.go index 8d01651..fd4e4cb 100644 --- a/clientdb/db.go +++ b/clientdb/db.go @@ -97,6 +97,10 @@ func initDB(filepath string, firstInit bool) (*bbolt.DB, error) { if err != nil { return err } + _, err = tx.CreateBucketIfNotExists(sidecarsBucketKey) + if err != nil { + return err + } _, err = tx.CreateBucketIfNotExists(batchBucketKey) if err != nil { return err diff --git a/clientdb/sidecar.go b/clientdb/sidecar.go new file mode 100644 index 0000000..b8ff7b4 --- /dev/null +++ b/clientdb/sidecar.go @@ -0,0 +1,173 @@ +package clientdb + +import ( + "bytes" + "errors" + "fmt" + + "github.com/btcsuite/btcd/btcec" + "github.com/lightninglabs/pool/sidecar" + "go.etcd.io/bbolt" +) + +var ( + // ErrNoSidecar is the error returned if no sidecar with the given + // multisig pubkey exists in the store. + ErrNoSidecar = errors.New("no sidecar found") + + // sidecarsBucketKey is a bucket that contains all sidecars that are + // currently pending or completed. This bucket is keyed by the ticket ID + // and offer signing pubkey of a sidecar. + sidecarsBucketKey = []byte("sidecars") +) + +const ( + // sidecarKeyLen is the length of a sidecar ticket's key. It is the + // length of the sidecar ID (8 bytes) plus the length of a compressed + // public key (33 bytes). + sidecarKeyLen = 8 + 33 +) + +// A compile time check to make sure we satisfy the sidecar.Store interface. +var _ sidecar.Store = (*DB)(nil) + +// getSidecarKey returns the key for a sidecar. +func getSidecarKey(id [8]byte, offerSignPubKey *btcec.PublicKey) ([]byte, + error) { + + if offerSignPubKey == nil { + return nil, fmt.Errorf("offer signing pubkey cannot be nil") + } + + var result [sidecarKeyLen]byte + + copy(result[:], id[:]) + copy(result[8:], offerSignPubKey.SerializeCompressed()) + + return result[:], nil +} + +// AddSidecar adds a record for the sidecar to the database. +func (db *DB) AddSidecar(ticket *sidecar.Ticket) error { + sidecarKey, err := getSidecarKey(ticket.ID, ticket.Offer.SignPubKey) + if err != nil { + return err + } + + return db.Update(func(tx *bbolt.Tx) error { + sidecarBucket, err := getBucket(tx, sidecarsBucketKey) + if err != nil { + return err + } + + sidecarValue := sidecarBucket.Get(sidecarKey) + if len(sidecarValue) != 0 { + return fmt.Errorf("sidecar for key %x already exists", + sidecarKey) + } + + return storeSidecar(sidecarBucket, sidecarKey, ticket) + }) +} + +// UpdateSidecar updates a sidecar in the database. +func (db *DB) UpdateSidecar(ticket *sidecar.Ticket) error { + sidecarKey, err := getSidecarKey(ticket.ID, ticket.Offer.SignPubKey) + if err != nil { + return err + } + + return db.Update(func(tx *bbolt.Tx) error { + sidecarBucket, err := getBucket(tx, sidecarsBucketKey) + if err != nil { + return err + } + + sidecarValue := sidecarBucket.Get(sidecarKey) + if len(sidecarValue) == 0 { + return ErrNoSidecar + } + + return storeSidecar(sidecarBucket, sidecarKey, ticket) + }) +} + +// Sidecar retrieves a specific sidecar by its ID and provider signing key +// (offer signature pubkey) or returns ErrNoSidecar if it's not found. +func (db *DB) Sidecar(id [8]byte, + offerSignPubKey *btcec.PublicKey) (*sidecar.Ticket, error) { + + sidecarKey, err := getSidecarKey(id, offerSignPubKey) + if err != nil { + return nil, err + } + + var s *sidecar.Ticket + err = db.View(func(tx *bbolt.Tx) error { + sidecarBucket, err := getBucket(tx, sidecarsBucketKey) + if err != nil { + return err + } + + s, err = readSidecar(sidecarBucket, sidecarKey) + return err + }) + if err != nil { + return nil, err + } + + return s, nil +} + +// Sidecars retrieves all known sidecars from the database. +func (db *DB) Sidecars() ([]*sidecar.Ticket, error) { + var res []*sidecar.Ticket + err := db.View(func(tx *bbolt.Tx) error { + sidecarBucket, err := getBucket(tx, sidecarsBucketKey) + if err != nil { + return err + } + + return sidecarBucket.ForEach(func(k, v []byte) error { + // We don't expect any sub-buckets with sidecars. + if v == nil { + return fmt.Errorf("nil value for key %x", k) + } + + s, err := readSidecar(sidecarBucket, k) + if err != nil { + return err + } + res = append(res, s) + + return nil + }) + }) + if err != nil { + return nil, err + } + + return res, nil +} + +func storeSidecar(targetBucket *bbolt.Bucket, key []byte, + ticket *sidecar.Ticket) error { + + var sidecarBuf bytes.Buffer + if err := sidecar.SerializeTicket(&sidecarBuf, ticket); err != nil { + return err + } + + return targetBucket.Put(key, sidecarBuf.Bytes()) +} + +func readSidecar(sourceBucket *bbolt.Bucket, id []byte) (*sidecar.Ticket, + error) { + + sidecarBytes := sourceBucket.Get(id) + if sidecarBytes == nil { + return nil, ErrNoSidecar + } + + return sidecar.DeserializeTicket(bytes.NewReader(sidecarBytes)) +} diff --git a/clientdb/sidecar_test.go b/clientdb/sidecar_test.go new file mode 100644 index 0000000..a9b7c0b --- /dev/null +++ b/clientdb/sidecar_test.go @@ -0,0 +1,72 @@ +package clientdb + +import ( + "testing" + + "github.com/lightninglabs/pool/order" + "github.com/lightninglabs/pool/sidecar" + "github.com/stretchr/testify/require" +) + +func assertSidecarExists(t *testing.T, db *DB, expected *sidecar.Ticket) { + t.Helper() + + found, err := db.Sidecar(expected.ID, expected.Offer.SignPubKey) + require.NoError(t, err) + + require.Equal(t, expected, found) +} + +// TestSidecars ensures that all database operations involving sidecars run as +// expected. +func TestSidecars(t *testing.T) { + t.Parallel() + + db, cleanup := newTestDB(t) + defer cleanup() + + // Create a test sidecar we'll use to interact with the database. + s := &sidecar.Ticket{ + ID: [8]byte{12, 34, 56}, + State: sidecar.StateRegistered, + Offer: sidecar.Offer{ + Capacity: 1000000, + PushAmt: 200000, + SignPubKey: testTraderKey, + LeaseDurationBlocks: 2016, + }, + Recipient: &sidecar.Recipient{ + MultiSigPubKey: testTraderKey, + MultiSigKeyIndex: 7, + }, + } + + // First, we'll add it to the database. We should be able to retrieve + // after. + err := db.AddSidecar(s) + require.NoError(t, err) + assertSidecarExists(t, db, s) + + // Transition the sidecar state from SidecarInitialized to + // SidecarExpectingChannel and add the required information for that + // state. + s.State = sidecar.StateExpectingChannel + s.Order = &sidecar.Order{ + BidNonce: order.Nonce{1, 2, 3}, + } + err = db.UpdateSidecar(s) + require.NoError(t, err) + assertSidecarExists(t, db, s) + + // Retrieving all sidecars should show that we only have one sidecar, + // the same one. + sidecars, err := db.Sidecars() + require.NoError(t, err) + require.Len(t, sidecars, 1) + require.Contains(t, sidecars, s) + + // Make sure we can query a sidecar ticket by its ID and offer pubkey. + updatedTicket, err := db.Sidecar([8]byte{12, 34, 56}, testTraderKey) + require.NoError(t, err) + require.Equal(t, s, updatedTicket) +} diff --git a/sidecar/interface.go b/sidecar/interface.go index e0a7f85..07ed241 100644 --- a/sidecar/interface.go +++ b/sidecar/interface.go @@ -264,3 +264,21 @@ func (t *Ticket) OrderDigest() ([32]byte, error) { } return sha256.Sum256(msg.Bytes()), nil } + +// Store is the interface a persistent storage must implement for storing and +// retrieving sidecar tickets. +type Store interface { + // AddSidecar adds a record for the sidecar order to the database. + AddSidecar(sidecar *Ticket) error + + // UpdateSidecar updates a sidecar order in the database. + UpdateSidecar(sidecar *Ticket) error + + // Sidecar retrieves a specific sidecar by its ID and provider signing + // key (offer signature pubkey) or returns ErrNoSidecar if it's not + // found. + Sidecar(id [8]byte, offerSignPubKey *btcec.PublicKey) (*Ticket, error) + + // Sidecars retrieves all known sidecar orders from the database. + Sidecars() ([]*Ticket, error) +} From a6f5c600203211eafdd84f1753e0b700fb466470 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 19 Mar 2021 11:50:03 +0100 Subject: [PATCH 04/13] sidecar: add signing and verification functions For maximum re-usability of the signing/verifying code, we add them as helper methods to the sidecar package itself. --- sidecar/verification.go | 128 ++++++++++++++++++ sidecar/verification_test.go | 246 +++++++++++++++++++++++++++++++++++ 2 files changed, 374 insertions(+) create mode 100644 sidecar/verification_test.go diff --git a/sidecar/verification.go b/sidecar/verification.go index 1904bac..81499a8 100644 --- a/sidecar/verification.go +++ b/sidecar/verification.go @@ -1,11 +1,139 @@ package sidecar import ( + "context" "fmt" "github.com/btcsuite/btcutil" + "github.com/lightninglabs/lndclient" + "github.com/lightningnetwork/lnd/keychain" + "github.com/lightningnetwork/lnd/lnwire" ) +// SignOffer adds a signature over the offer digest to the given ticket. +func SignOffer(ctx context.Context, ticket *Ticket, + signingKeyLoc keychain.KeyLocator, signer lndclient.SignerClient) error { + + // The ticket needs to be in the correct state for us to sign it. + if ticket == nil || ticket.State < StateOffered { + return fmt.Errorf("ticket is in invalid state") + } + + // The ticket also needs to have a signed offer. + offer := ticket.Offer + if offer.SignPubKey == nil || offer.SigOfferDigest != nil { + return fmt.Errorf("offer in ticket is not in expected state " + + "to be signed") + } + + // Let's sign the offer part of the ticket with our node's identity key + // now. + offerDigest, err := ticket.OfferDigest() + if err != nil { + return fmt.Errorf("error digesting offer: %v", err) + } + rawSig, err := signer.SignMessage(ctx, offerDigest[:], signingKeyLoc) + if err != nil { + return fmt.Errorf("error signing offer: %v", err) + } + wireSig, err := lnwire.NewSigFromRawSignature(rawSig) + if err != nil { + return fmt.Errorf("error parsing raw signature: %v", err) + } + ecSig, err := wireSig.ToSignature() + if err != nil { + return fmt.Errorf("error parsing EC signature: %v", err) + } + ticket.Offer.SigOfferDigest = ecSig + + return nil +} + +// VerifyOffer verifies the state of a ticket to be in the offered state and +// also makes sure the offer signature is valid. +func VerifyOffer(ctx context.Context, ticket *Ticket, + signer lndclient.SignerClient) error { + + // The ticket needs to be in the correct state for us to verify it. + if ticket == nil || ticket.State < StateOffered { + return fmt.Errorf("ticket is in invalid state") + } + + // The ticket also needs to have a signed offer. + offer := ticket.Offer + if offer.SignPubKey == nil || offer.SigOfferDigest == nil { + return fmt.Errorf("offer in ticket is not signed") + } + + var offerPubKeyRaw [33]byte + copy(offerPubKeyRaw[:], ticket.Offer.SignPubKey.SerializeCompressed()) + + // Make sure the provider's signature over the offer is valid. + offerDigest, err := ticket.OfferDigest() + if err != nil { + return fmt.Errorf("error calculating offer digest: %v", err) + } + sigValid, err := signer.VerifyMessage( + ctx, offerDigest[:], offer.SigOfferDigest.Serialize(), + offerPubKeyRaw, + ) + if err != nil { + return fmt.Errorf("unable to verify offer signature: %v", err) + } + if !sigValid { + return fmt.Errorf("signature not valid for public key %x", + offerPubKeyRaw[:]) + } + + return nil +} + +// SignOrder adds the order part to a ticket and signs it, adding the signature +// as well. +func SignOrder(ctx context.Context, ticket *Ticket, bidNonce [32]byte, + signingKeyLoc keychain.KeyLocator, signer lndclient.SignerClient) error { + + // The ticket needs to be in the correct state for us to sign it. + if ticket == nil || ticket.State < StateRegistered { + return fmt.Errorf("ticket is in invalid state") + } + + // The ticket also needs to have a signed offer. + offer := ticket.Offer + if offer.SignPubKey == nil || offer.SigOfferDigest == nil { + return fmt.Errorf("offer in ticket is not signed") + } + + // Add the bid order's nonce to the order part of the ticket now. + if ticket.Order == nil { + ticket.Order = &Order{} + } + ticket.Order.BidNonce = bidNonce + ticket.State = StateOrdered + + // Let's sign the order part of the ticket with our node's identity key + // now. + orderDigest, err := ticket.OrderDigest() + if err != nil { + return fmt.Errorf("error digesting order: %v", err) + } + rawSig, err := signer.SignMessage(ctx, orderDigest[:], signingKeyLoc) + if err != nil { + return fmt.Errorf("error signing order: %v", err) + } + wireSig, err := lnwire.NewSigFromRawSignature(rawSig) + if err != nil { + return fmt.Errorf("error parsing raw signature: %v", err) + } + ecSig, err := wireSig.ToSignature() + if err != nil { + return fmt.Errorf("error parsing EC signature: %v", err) + } + ticket.Order.SigOrderDigest = ecSig + + return nil +} + // CheckOfferParams makes sure the offer parameters of a sidecar ticket are // valid and sane. func CheckOfferParams(capacity, pushAmt, baseSupplyUnit btcutil.Amount) error { diff --git a/sidecar/verification_test.go b/sidecar/verification_test.go new file mode 100644 index 0000000..d45b6f2 --- /dev/null +++ b/sidecar/verification_test.go @@ -0,0 +1,246 @@ +package sidecar + +import ( + "context" + "math/big" + "testing" + + "github.com/btcsuite/btcd/btcec" + "github.com/lightninglabs/pool/internal/test" + "github.com/lightningnetwork/lnd/keychain" + "github.com/stretchr/testify/require" +) + +var ( + _, providerPubKey = btcec.PrivKeyFromBytes(btcec.S256(), []byte{0x02}) + testOfferSig = &btcec.Signature{ + R: new(big.Int).SetInt64(44), + S: new(big.Int).SetInt64(22), + } +) + +// TestSignOffer makes sure that a sidecar ticket's offer part can be signed +// correctly. +func TestSignOffer(t *testing.T) { + t.Parallel() + + mockSigner := test.NewMockSigner() + mockSigner.Signature = testOfferSig.Serialize() + + testCases := []struct { + name string + ticket *Ticket + expectedErr string + }{{ + name: "no ticket", + expectedErr: "ticket is in invalid state", + }, { + name: "non offered ticket", + ticket: &Ticket{ + State: StateCreated, + }, + expectedErr: "ticket is in invalid state", + }, { + name: "offered ticket with no sign pubkey", + ticket: &Ticket{ + State: StateOffered, + Offer: Offer{}, + }, + expectedErr: "offer in ticket is not in expected state to be", + }, { + name: "offered ticket has signature", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOffered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + }, + expectedErr: "offer in ticket is not in expected state to be", + }, { + name: "all valid", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOffered, + Offer: Offer{ + SignPubKey: providerPubKey, + }, + }, + expectedErr: "", + }} + + for _, tc := range testCases { + tc := tc + + if tc.ticket != nil { + digest, err := tc.ticket.OfferDigest() + require.NoError(t, err) + mockSigner.SignatureMsg = string(digest[:]) + } + + err := SignOffer( + context.Background(), tc.ticket, keychain.KeyLocator{}, + mockSigner, + ) + + if tc.expectedErr == "" { + require.NoError(t, err) + require.Equal( + t, testOfferSig, tc.ticket.Offer.SigOfferDigest, + ) + } else { + require.Error(t, err) + require.Contains(t, err.Error(), tc.expectedErr) + } + } +} + +// TestVerifyOffer makes sure that a sidecar ticket's offer part can be verified +// correctly. +func TestVerifyOffer(t *testing.T) { + t.Parallel() + + mockSigner := test.NewMockSigner() + mockSigner.Signature = testOfferSig.Serialize() + + testCases := []struct { + name string + ticket *Ticket + expectedErr string + }{{ + name: "no ticket", + expectedErr: "ticket is in invalid state", + }, { + name: "non offered ticket", + ticket: &Ticket{ + State: StateCreated, + }, + expectedErr: "ticket is in invalid state", + }, { + name: "offered ticket with no sign pubkey", + ticket: &Ticket{ + State: StateOffered, + Offer: Offer{}, + }, + expectedErr: "offer in ticket is not signed", + }, { + name: "invalid sig", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOffered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: &btcec.Signature{ + R: new(big.Int).SetInt64(33), + S: new(big.Int).SetInt64(33), + }, + }, + }, + expectedErr: "signature not valid for public key", + }, { + name: "all valid", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOffered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + }, + expectedErr: "", + }} + + for _, tc := range testCases { + tc := tc + + if tc.ticket != nil { + digest, err := tc.ticket.OfferDigest() + require.NoError(t, err) + mockSigner.SignatureMsg = string(digest[:]) + } + + err := VerifyOffer( + context.Background(), tc.ticket, mockSigner, + ) + + if tc.expectedErr == "" { + require.NoError(t, err) + } else { + require.Error(t, err) + require.Contains(t, err.Error(), tc.expectedErr) + } + } +} + +// TestSignOrder makes sure that a sidecar ticket's order part can be signed +// correctly. +func TestSignOrder(t *testing.T) { + t.Parallel() + + mockSigner := test.NewMockSigner() + mockSigner.Signature = testOfferSig.Serialize() + + testCases := []struct { + name string + ticket *Ticket + expectedErr string + }{{ + name: "no ticket", + expectedErr: "ticket is in invalid state", + }, { + name: "non registered ticket", + ticket: &Ticket{ + State: StateCreated, + }, + expectedErr: "ticket is in invalid state", + }, { + name: "non signed offer", + ticket: &Ticket{ + State: StateRegistered, + Offer: Offer{ + SignPubKey: providerPubKey, + }, + }, + expectedErr: "offer in ticket is not signed", + }, { + name: "all valid", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateRegistered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + }, + expectedErr: "", + }} + + for _, tc := range testCases { + tc := tc + + if tc.ticket != nil { + digest, err := tc.ticket.OfferDigest() + require.NoError(t, err) + mockSigner.SignatureMsg = string(digest[:]) + } + + err := SignOrder( + context.Background(), tc.ticket, [32]byte{}, + keychain.KeyLocator{}, mockSigner, + ) + + if tc.expectedErr == "" { + require.NoError(t, err) + + require.NotNil(t, tc.ticket.Order) + require.Equal( + t, testOfferSig, tc.ticket.Order.SigOrderDigest, + ) + require.Equal(t, StateOrdered, tc.ticket.State) + } else { + require.Error(t, err) + require.Contains(t, err.Error(), tc.expectedErr) + } + } +} From 2f5637889815776ea776e3ef456de2e55a317658 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 19 Mar 2021 11:50:10 +0100 Subject: [PATCH 05/13] order+clientdb: add sidecar ticket to bid order With this commit we add the optional sidecar ticket to bid orders. Because the node submitting the bid order doesn't necessarily need to be the node creating the offer, we just store the ticket as is and don't require it to be in the database already. --- clientdb/order.go | 38 +++++++++++++++++++++++++++++++++++--- clientdb/order_test.go | 14 ++++++++++++++ order/interface.go | 37 ++++++++++++++++++++++++++++++++++++- 3 files changed, 85 insertions(+), 4 deletions(-) diff --git a/clientdb/order.go b/clientdb/order.go index 3e00507..115b503 100644 --- a/clientdb/order.go +++ b/clientdb/order.go @@ -9,6 +9,7 @@ import ( "github.com/btcsuite/btcutil" "github.com/lightninglabs/pool/event" "github.com/lightninglabs/pool/order" + "github.com/lightninglabs/pool/sidecar" "github.com/lightningnetwork/lnd/tlv" "go.etcd.io/bbolt" ) @@ -17,6 +18,10 @@ const ( // bidSelfChanBalanceType is the tlv type we use to store the self // channel balance on bid orders. bidSelfChanBalanceType tlv.Type = 1 + + // bidSidecarTicketType is the tlv type we use to store the sidecar + // ticket on bid orders. + bidSidecarTicketType tlv.Type = 2 ) var ( @@ -631,14 +636,18 @@ func DeserializeOrder(nonce order.Nonce, r io.Reader) ( func deserializeOrderTlvData(r io.Reader, o order.Order) error { var ( selfChanBalance uint64 + sidecarTicket []byte ) // We'll add records for all possible additional order data fields here // but will check below which of them were actually set, depending on // the order type as well. - tlvStream, err := tlv.NewStream(tlv.MakePrimitiveRecord( - bidSelfChanBalanceType, &selfChanBalance, - )) + tlvStream, err := tlv.NewStream( + tlv.MakePrimitiveRecord( + bidSelfChanBalanceType, &selfChanBalance, + ), + tlv.MakePrimitiveRecord(bidSidecarTicketType, &sidecarTicket), + ) if err != nil { return err } @@ -659,6 +668,15 @@ func deserializeOrderTlvData(r io.Reader, o order.Order) error { selfChanBalance, ) } + + if t, ok := parsedTypes[bidSidecarTicketType]; ok && t == nil { + castOrder.SidecarTicket, err = sidecar.DeserializeTicket( + bytes.NewReader(sidecarTicket), + ) + if err != nil { + return err + } + } } return nil @@ -679,6 +697,20 @@ func serializeOrderTlvData(w io.Writer, o order.Order) error { bidSelfChanBalanceType, &selfChanBalance, )) } + + if castOrder.SidecarTicket != nil { + var buf bytes.Buffer + if err := sidecar.SerializeTicket( + &buf, castOrder.SidecarTicket, + ); err != nil { + return err + } + + sidecarBytes := buf.Bytes() + tlvRecords = append(tlvRecords, tlv.MakePrimitiveRecord( + bidSidecarTicketType, &sidecarBytes, + )) + } } tlvStream, err := tlv.NewStream(tlvRecords...) diff --git a/clientdb/order_test.go b/clientdb/order_test.go index 48b5386..1e551e2 100644 --- a/clientdb/order_test.go +++ b/clientdb/order_test.go @@ -9,6 +9,7 @@ import ( "github.com/btcsuite/btcutil" "github.com/davecgh/go-spew/spew" "github.com/lightninglabs/pool/order" + "github.com/lightninglabs/pool/sidecar" "github.com/lightningnetwork/lnd/keychain" "github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lnwallet/chainfee" @@ -26,6 +27,19 @@ func TestSubmitOrder(t *testing.T) { Kit: *dummyOrder(500000, 1337), MinNodeTier: 2, SelfChanBalance: 123, + SidecarTicket: &sidecar.Ticket{ + ID: [8]byte{11, 22, 33, 44, 55, 66, 77}, + State: sidecar.StateRegistered, + Offer: sidecar.Offer{ + Capacity: 1000000, + PushAmt: 200000, + LeaseDurationBlocks: 2016, + }, + Recipient: &sidecar.Recipient{ + MultiSigPubKey: testTraderKey, + MultiSigKeyIndex: 7, + }, + }, } o.Details().MinUnitsMatch = 10 err := store.SubmitOrder(o) diff --git a/order/interface.go b/order/interface.go index 158f2ff..fbd7a82 100644 --- a/order/interface.go +++ b/order/interface.go @@ -49,6 +49,16 @@ const ( // self channel balance field. Only orders with this version are allowed // to use the self channel balance field. VersionSelfChanBalance Version = 3 + + // VersionSidecarChannel is the order version that added sidecar + // channels for bid orders. Only orders with this version are allowed + // to set the sidecar ticket field on bid orders. Since sidecar orders + // also add the feature of push amounts on the leased channels, this + // affects makers as well. Makers that don't want to support leasing out + // channels with a push amount (because it might screw up their + // accounting or whatever) can opt out by explicitly submitting their + // ask orders with a version previous to this one. + VersionSidecarChannel Version = 4 ) // Type is the type of an order. We don't use iota for the constants due to the @@ -370,7 +380,7 @@ func (a *Ask) Digest() ([sha256.Size]byte, error) { } case VersionNodeTierMinMatch, VersionLeaseDurationBuckets, - VersionSelfChanBalance: + VersionSelfChanBalance, VersionSidecarChannel: err := lnwire.WriteElements( &msg, a.nonce[:], uint32(a.Version), a.FixedRate, @@ -520,6 +530,15 @@ type Bid struct { // to the channel resulting from matching this bid by moving additional // funds from the taker's account into the channel. SelfChanBalance btcutil.Amount + + // SidecarTicket indicates, if non-nil, that the channel being purchased + // with this bid should be opened to a node other than the caller's + // node. The lease recipient is another Pool (light) node that + // authenticates itself to the auctioneer using the information in this + // ticket (the information exchange between bidder and lease recipient + // happens out of band). This will only be used if the order version is + // VersionSidecarChannel or greater. + SidecarTicket *sidecar.Ticket } // Type returns the order type. @@ -570,6 +589,22 @@ func (b *Bid) Digest() ([sha256.Size]byte, error) { return result, err } + case VersionSidecarChannel: + var isSidecar uint8 + if b.SidecarTicket != nil { + isSidecar = 1 + } + + err := lnwire.WriteElements( + &msg, b.nonce[:], uint32(b.Version), b.FixedRate, + b.Amt, b.LeaseDuration, uint64(b.MaxBatchFeeRate), + uint32(b.MinNodeTier), uint32(b.MinUnitsMatch), + uint64(b.SelfChanBalance), isSidecar, + ) + if err != nil { + return result, err + } + default: return result, fmt.Errorf("unknown version %d", b.Kit.Version) } From f81a1c3cfa5abfdf17850b2f3050b17437858ef8 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 19 Mar 2021 16:48:53 +0100 Subject: [PATCH 06/13] sidecar: add order verification --- sidecar/verification.go | 51 +++++++++++++++++ sidecar/verification_test.go | 106 +++++++++++++++++++++++++++++++++++ 2 files changed, 157 insertions(+) diff --git a/sidecar/verification.go b/sidecar/verification.go index 81499a8..78d79a5 100644 --- a/sidecar/verification.go +++ b/sidecar/verification.go @@ -134,6 +134,57 @@ func SignOrder(ctx context.Context, ticket *Ticket, bidNonce [32]byte, return nil } +// VerifyOrder verifies the state of a ticket to be in the ordered state and +// also makes sure the order signature is valid. +func VerifyOrder(ctx context.Context, ticket *Ticket, + signer lndclient.SignerClient) error { + + // The ticket needs to be in the correct state for us to verify it. + if ticket == nil || ticket.State < StateOrdered { + return fmt.Errorf("ticket is in invalid state") + } + + // The ticket also needs to have a pubkey in the offer and needs to be + // signed. We don't need to verify the signature again, the provider + // wouldn't sign the order if the offer itself isn't valid. + offer := ticket.Offer + if offer.SignPubKey == nil || offer.SigOfferDigest == nil { + return fmt.Errorf("offer in ticket is not signed") + } + + order := ticket.Order + if order == nil || order.SigOrderDigest == nil { + return fmt.Errorf("order in ticket is not signed") + } + + // The nonce shouldn't be empty either. + if order.BidNonce == [32]byte{} { + return fmt.Errorf("nonce in order part of ticket is empty") + } + + var orderPubKeyRaw [33]byte + copy(orderPubKeyRaw[:], ticket.Offer.SignPubKey.SerializeCompressed()) + + // Make sure the provider's signature over the order is valid. + orderDigest, err := ticket.OrderDigest() + if err != nil { + return fmt.Errorf("error calculating order digest: %v", err) + } + sigValid, err := signer.VerifyMessage( + ctx, orderDigest[:], order.SigOrderDigest.Serialize(), + orderPubKeyRaw, + ) + if err != nil { + return fmt.Errorf("unable to verify order signature: %v", err) + } + if !sigValid { + return fmt.Errorf("signature not valid for public key %x", + orderPubKeyRaw[:]) + } + + return nil +} + // CheckOfferParams makes sure the offer parameters of a sidecar ticket are // valid and sane. func CheckOfferParams(capacity, pushAmt, baseSupplyUnit btcutil.Amount) error { diff --git a/sidecar/verification_test.go b/sidecar/verification_test.go index d45b6f2..70bd036 100644 --- a/sidecar/verification_test.go +++ b/sidecar/verification_test.go @@ -244,3 +244,109 @@ func TestSignOrder(t *testing.T) { } } } + +// TestVerifyOrder makes sure that a sidecar ticket's order part can be verified +// correctly. +func TestVerifyOrder(t *testing.T) { + t.Parallel() + + mockSigner := test.NewMockSigner() + mockSigner.Signature = testOfferSig.Serialize() + + testCases := []struct { + name string + ticket *Ticket + expectedErr string + }{{ + name: "no ticket", + expectedErr: "ticket is in invalid state", + }, { + name: "non ordered ticket", + ticket: &Ticket{ + State: StateOffered, + }, + expectedErr: "ticket is in invalid state", + }, { + name: "invalid sig", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOrdered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + Order: &Order{ + SigOrderDigest: &btcec.Signature{ + R: new(big.Int).SetInt64(33), + S: new(big.Int).SetInt64(33), + }, + BidNonce: [32]byte{1, 2, 3}, + }, + }, + expectedErr: "signature not valid for public key", + }, { + name: "ordered ticket with no signature", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOrdered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + Order: &Order{ + BidNonce: [32]byte{1, 2, 3}, + }, + }, + expectedErr: "order in ticket is not signed", + }, { + name: "empty nonce", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOrdered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + Order: &Order{ + SigOrderDigest: testOfferSig, + }, + }, + expectedErr: "nonce in order part of ticket is empty", + }, { + name: "all valid", + ticket: &Ticket{ + ID: [8]byte{1, 2, 3, 4}, + State: StateOrdered, + Offer: Offer{ + SignPubKey: providerPubKey, + SigOfferDigest: testOfferSig, + }, + Order: &Order{ + SigOrderDigest: testOfferSig, + BidNonce: [32]byte{1, 2, 3}, + }, + }, + expectedErr: "", + }} + + for _, tc := range testCases { + tc := tc + + if tc.ticket != nil && tc.ticket.Order != nil { + digest, err := tc.ticket.OrderDigest() + require.NoError(t, err) + mockSigner.SignatureMsg = string(digest[:]) + } + + err := VerifyOrder( + context.Background(), tc.ticket, mockSigner, + ) + + if tc.expectedErr == "" { + require.NoError(t, err, tc.name) + } else { + require.Error(t, err, tc.name) + require.Contains(t, err.Error(), tc.expectedErr, tc.name) + } + } +} From f56eb50981cda6c6a2996ed8f9970ee183bd7114 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Thu, 1 Apr 2021 17:44:54 +0200 Subject: [PATCH 07/13] auctioneer+account: rename SubscribeAccountUpdates The name SubscribeAccountUpdates could indicate that this is a long-running operation. The method returns after successful authentication of the stream though. That's why we rename it to StartAccountSubscription and update the comment to make it more clear that this should return after a few seconds at the most. --- account/interface.go | 8 +++++--- account/manager.go | 4 ++-- account/mock_test.go | 2 +- auctioneer/client.go | 20 +++++++++++--------- 4 files changed, 19 insertions(+), 15 deletions(-) diff --git a/account/interface.go b/account/interface.go index 4db4968..61a5bdb 100644 --- a/account/interface.go +++ b/account/interface.go @@ -380,14 +380,16 @@ type Auctioneer interface { ModifyAccount(context.Context, *Account, []*wire.TxIn, []*wire.TxOut, []Modifier) ([]byte, error) - // SubscribeAccountUpdates opens a stream to the server and subscribes + // StartAccountSubscription opens a stream to the server and subscribes // to all updates that concern the given account, including all orders // that spend from that account. Only a single stream is ever open to // the server, so a second call to this method will send a second // subscription over the same stream, multiplexing all messages into the // same connection. A stream can be long-lived, so this can be called - // for every account as soon as it's confirmed open. - SubscribeAccountUpdates(context.Context, *keychain.KeyDescriptor) error + // for every account as soon as it's confirmed open. This method will + // return as soon as the authentication was successful. Messages sent + // from the server can then be received on the FromServerChan channel. + StartAccountSubscription(context.Context, *keychain.KeyDescriptor) error // Terms returns the current dynamic auctioneer terms like max account // size, max order duration in blocks and the auction fee schedule. diff --git a/account/manager.go b/account/manager.go index 9d0dbdd..9c6ca8d 100644 --- a/account/manager.go +++ b/account/manager.go @@ -647,7 +647,7 @@ func (m *Manager) resumeAccount(ctx context.Context, account *Account, // nolint // update state, as that state is ineligible for batch // execution. if account.State == StatePendingBatch { - err = m.cfg.Auctioneer.SubscribeAccountUpdates( + err = m.cfg.Auctioneer.StartAccountSubscription( ctx, account.TraderKey, ) if err != nil { @@ -832,7 +832,7 @@ func (m *Manager) handleStateOpen(ctx context.Context, account *Account) error { // level anyway. And we might end up executing multiple orders for the // same account in one batch. The messages from the server are received // and dispatched to the correct manager by the rpcServer. - err = m.cfg.Auctioneer.SubscribeAccountUpdates(ctx, account.TraderKey) + err = m.cfg.Auctioneer.StartAccountSubscription(ctx, account.TraderKey) if err != nil { return fmt.Errorf("unable to subscribe for account updates: %v", err) diff --git a/account/mock_test.go b/account/mock_test.go index dfb7be8..17a39c0 100644 --- a/account/mock_test.go +++ b/account/mock_test.go @@ -195,7 +195,7 @@ func (a *mockAuctioneer) ModifyAccount(_ context.Context, _ *Account, return []byte("auctioneer sig"), nil } -func (a *mockAuctioneer) SubscribeAccountUpdates(_ context.Context, +func (a *mockAuctioneer) StartAccountSubscription(_ context.Context, accountKey *keychain.KeyDescriptor) error { var traderKey [33]byte diff --git a/auctioneer/client.go b/auctioneer/client.go index 43adf09..05a5620 100644 --- a/auctioneer/client.go +++ b/auctioneer/client.go @@ -481,14 +481,16 @@ func (c *Client) OrderState(ctx context.Context, nonce order.Nonce) ( }) } -// SubscribeAccountUpdates opens a stream to the server and subscribes -// to all updates that concern the given account, including all orders -// that spend from that account. Only a single stream is ever open to -// the server, so a second call to this method will send a second -// subscription over the same stream, multiplexing all messages into the -// same connection. A stream can be long-lived, so this can be called -// for every account as soon as it's confirmed open. -func (c *Client) SubscribeAccountUpdates(ctx context.Context, +// StartAccountSubscription opens a stream to the server and subscribes to all +// updates that concern the given account, including all orders that spend from +// that account. Only a single stream is ever open to the server, so a second +// call to this method will send a second subscription over the same stream, +// multiplexing all messages into the same connection. A stream can be +// long-lived, so this can be called for every account as soon as it's confirmed +// open. This method will return as soon as the authentication was successful. +// Messages sent from the server can then be received on the FromServerChan +// channel. +func (c *Client) StartAccountSubscription(ctx context.Context, acctKey *keychain.KeyDescriptor) error { _, _, err := c.connectAndAuthenticate(ctx, acctKey, false) @@ -1135,7 +1137,7 @@ func (c *Client) HandleServerShutdown(err error) error { } c.subscribedAcctsMtx.Unlock() for _, acctKey := range acctKeys { - err := c.SubscribeAccountUpdates(context.Background(), acctKey) + err := c.StartAccountSubscription(context.Background(), acctKey) if err != nil { return err } From 44fadace7a6c262333fe3e9ea62426a08ae2c4cf Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 19 Mar 2021 16:48:56 +0100 Subject: [PATCH 08/13] funding: extract waitForChannelOpen method The channel funding process for a sidecar channel will look different from the normal process. To avoid too much code duplication, we first extract the parts that remain the same into its own method. We no longer need to hold a mutex since when we get to the waitForChannelOpen method all previously spawned goroutines must have completed. --- funding/manager.go | 33 ++++++++++++++++++++++++++++----- 1 file changed, 28 insertions(+), 5 deletions(-) diff --git a/funding/manager.go b/funding/manager.go index d7f2170..978b238 100644 --- a/funding/manager.go +++ b/funding/manager.go @@ -27,6 +27,12 @@ import ( "google.golang.org/grpc/status" ) +var ( + // rpcCodeFundingFailed is the error code we send if the channel funding + // fails because of a timeout or another problem. + rpcCodeFundingFailed = auctioneerrpc.OrderReject_CHANNEL_FUNDING_FAILED +) + // MatchRejectErr is an error type that is returned from the funding manager if // the trader rejects certain orders instead of the whole batch. type MatchRejectErr struct { @@ -396,7 +402,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, defer fundingRejectsMtx.Unlock() fundingRejects[nonce] = &auctioneerrpc.OrderReject{ - ReasonCode: auctioneerrpc.OrderReject_CHANNEL_FUNDING_FAILED, + ReasonCode: rpcCodeFundingFailed, Reason: reason, } @@ -555,6 +561,20 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, } } + // We've kicked off all channel open streams. We'll now need to wait for + // either completion of the funding process or a timeout. + return m.waitForChannelOpen(setupCtx, quit, chanPoints, fundingRejects) +} + +// waitForChannelOpen waits until we get a pending open channel message for each +// of the channel outpoints provided. If we don't get all of them in time, we'll +// instead return a reject error with all the orders for which the funding +// failed. +func (m *Manager) waitForChannelOpen(ctx context.Context, quit <-chan struct{}, + chanPoints map[wire.OutPoint]order.Nonce, + fundingRejects map[order.Nonce]*auctioneerrpc.OrderReject) ( + map[wire.OutPoint]*chaninfo.ChannelInfo, error) { + // Once we've waited for the operations to complete, we'll wait to // receive each channel's pending open notification in order to retrieve // some keys from their SCB we'll need to submit to the auctioneer in @@ -579,15 +599,18 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, Index: channel.PendingOpenChannel.OutputIndex, } - case <-setupCtx.Done(): + case <-ctx.Done(): // One or more of our peers timed out. At this point the // chanPoints map should only contain channels that // failed/timed out. So we can partially reject all of // them. const timeoutErrStr = "timed out waiting for pending " + "open channel notification" - for chanPoint, nonce := range chanPoints { - partialReject(nonce, timeoutErrStr, chanPoint) + for _, nonce := range chanPoints { + fundingRejects[nonce] = &auctioneerrpc.OrderReject{ + ReasonCode: rpcCodeFundingFailed, + Reason: timeoutErrStr, + } } return nil, &MatchRejectErr{ RejectedOrders: fundingRejects, @@ -607,7 +630,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, log.Debugf("Retrieving info for channel %v", chanPoint) chanInfo, err := chaninfo.GatherChannelInfo( - setupCtx, m.LightningClient, m.WalletKit, chanPoint, + ctx, m.LightningClient, m.WalletKit, chanPoint, ) if err != nil { return nil, err From 66fd38a46aa6cbd98a80632f93778457b9ca0774 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 19 Mar 2021 16:48:57 +0100 Subject: [PATCH 09/13] funding+rpcserver: use order fetcher instead of DB directly To prepare for accepting sidecar channels where we don't have a real order item in our local database, we use the order fetcher function in the funding manager instead of the DB directly. That way we can pass in fake bid orders during the sidecar acceptance. --- funding/manager.go | 4 ++-- funding/manager_test.go | 6 +++--- rpcserver.go | 4 +++- 3 files changed, 8 insertions(+), 6 deletions(-) diff --git a/funding/manager.go b/funding/manager.go index 978b238..c245a82 100644 --- a/funding/manager.go +++ b/funding/manager.go @@ -252,7 +252,7 @@ func (m *Manager) registerFundingShim(ourBid *order.Bid, // PrepChannelFunding preps the backing node to either receive or initiate a // channel funding based on the items in the order batch. -func (m *Manager) PrepChannelFunding(batch *order.Batch, +func (m *Manager) PrepChannelFunding(batch *order.Batch, getOrder order.Fetcher, quit <-chan struct{}) error { log.Infof("Batch(%x): preparing channel funding for %v orders", @@ -297,7 +297,7 @@ func (m *Manager) PrepChannelFunding(batch *order.Batch, // all the funding shims we need to be able to respond connsInitiated := make(map[route.Vertex]struct{}) for ourOrderNonce, matchedOrders := range batch.MatchedOrders { - ourOrder, err := m.DB.GetOrder(ourOrderNonce) + ourOrder, err := getOrder(ourOrderNonce) if err != nil { return err } diff --git a/funding/manager_test.go b/funding/manager_test.go index 98a439c..2e924dd 100644 --- a/funding/manager_test.go +++ b/funding/manager_test.go @@ -335,7 +335,7 @@ func TestFundingManager(t *testing.T) { h.baseClientMock.peerList = map[route.Vertex]string{ node1Key: "1.1.1.1", } - err = h.mgr.PrepChannelFunding(batch, h.quit) + err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder, h.quit) require.NoError(t, err) // Verify we have the expected connections and funding shims registered. @@ -381,7 +381,7 @@ func TestFundingManager(t *testing.T) { h.lnMock.Channels = append(h.lnMock.Channels, lndclient.ChannelInfo{ PubKeyBytes: node1Key, }) - err = h.mgr.PrepChannelFunding(batch, h.quit) + err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder, h.quit) require.Error(t, err) expectedErr := &MatchRejectErr{ @@ -399,7 +399,7 @@ func TestFundingManager(t *testing.T) { // error if the connections to the remote peers couldn't be established. h.mgr.NewNodesOnly = false h.baseClientMock.peerList = make(map[route.Vertex]string) - err = h.mgr.PrepChannelFunding(batch, h.quit) + err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder, h.quit) require.Error(t, err) expectedErr = &MatchRejectErr{ diff --git a/rpcserver.go b/rpcserver.go index 2ccefa0..a87a9a3 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -416,7 +416,9 @@ func (s *rpcServer) handleServerMessage( // Before we accept the batch, we'll finish preparations on our // end which include applying any order match predicates, // connecting out to peers, and registering funding shim. - err = s.server.fundingManager.PrepChannelFunding(batch, s.quit) + err = s.server.fundingManager.PrepChannelFunding( + batch, s.server.db.GetOrder, s.quit, + ) if err != nil { rpcLog.Warnf("Error preparing channel funding: %v", err) From 2a10b52f73912daba58825d408446ccbe67c3996 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 19 Mar 2021 16:48:58 +0100 Subject: [PATCH 10/13] rpcserver: extract marshallChannelInfos As a preparation to reuse that code in the sidecar acceptor, we extract it into its own function. --- rpcserver.go | 79 +++++++++++++++++++++++++++++++--------------------- 1 file changed, 47 insertions(+), 32 deletions(-) diff --git a/rpcserver.go b/rpcserver.go index a87a9a3..51de212 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -1587,38 +1587,9 @@ func (s *rpcServer) sendSignBatch(batch *order.Batch, sigs order.BatchSignature, rpcSigs[key] = sig.Serialize() } - rpcChannelInfos := make(map[string]*auctioneerrpc.ChannelInfo, len(chanInfos)) - for chanPoint, chanInfo := range chanInfos { - var channelType auctioneerrpc.ChannelType - switch chanInfo.Version { - case chanbackup.TweaklessCommitVersion: - channelType = auctioneerrpc.ChannelType_TWEAKLESS - - // The AnchorsCommitVersion was never widely deployed (at least - // in mainnet) because the lnd version that included it guarded - // the anchor channels behind a config flag. Also, the two - // anchor versions only differ in the fee negotiation and not - // the commitment TX format, so we don't need to distinguish - // between them for our purpose. - case chanbackup.AnchorsCommitVersion, - chanbackup.AnchorsZeroFeeHtlcTxCommitVersion: - - channelType = auctioneerrpc.ChannelType_ANCHORS - default: - return fmt.Errorf("unknown channel type: %v", - chanInfo.Version) - } - rpcChannelInfos[chanPoint.String()] = &auctioneerrpc.ChannelInfo{ - Type: channelType, - LocalNodeKey: chanInfo.LocalNodeKey. - SerializeCompressed(), - RemoteNodeKey: chanInfo.RemoteNodeKey. - SerializeCompressed(), - LocalPaymentBasePoint: chanInfo.LocalPaymentBasePoint. - SerializeCompressed(), - RemotePaymentBasePoint: chanInfo.RemotePaymentBasePoint. - SerializeCompressed(), - } + rpcChannelInfos, err := marshallChannelInfo(chanInfos) + if err != nil { + return fmt.Errorf("error marshalling channel info: %v", err) } rpcLog.Infof("Sending OrderMatchSign for batch %x", batch.ID[:]) @@ -2343,3 +2314,47 @@ func unmarshallNodeTier(nodeTier auctioneerrpc.NodeTier) (order.NodeTier, return 0, fmt.Errorf("unknown node tier: %v", nodeTier) } } + +// marshallChannelInfo turns the given channel information map into its RPC +// counterpart. +func marshallChannelInfo(chanInfos map[wire.OutPoint]*chaninfo.ChannelInfo) ( + map[string]*auctioneerrpc.ChannelInfo, error) { + + rpcChannelInfos := make( + map[string]*auctioneerrpc.ChannelInfo, len(chanInfos), + ) + for chanPoint, chanInfo := range chanInfos { + var channelType auctioneerrpc.ChannelType + switch chanInfo.Version { + case chanbackup.TweaklessCommitVersion: + channelType = auctioneerrpc.ChannelType_TWEAKLESS + + // The AnchorsCommitVersion was never widely deployed (at least + // in mainnet) because the lnd version that included it guarded + // the anchor channels behind a config flag. Also, the two + // anchor versions only differ in the fee negotiation and not + // the commitment TX format, so we don't need to distinguish + // between them for our purpose. + case chanbackup.AnchorsCommitVersion, + chanbackup.AnchorsZeroFeeHtlcTxCommitVersion: + + channelType = auctioneerrpc.ChannelType_ANCHORS + default: + return nil, fmt.Errorf("unknown channel type: %v", + chanInfo.Version) + } + rpcChannelInfos[chanPoint.String()] = &auctioneerrpc.ChannelInfo{ + Type: channelType, + LocalNodeKey: chanInfo.LocalNodeKey. + SerializeCompressed(), + RemoteNodeKey: chanInfo.RemoteNodeKey. + SerializeCompressed(), + LocalPaymentBasePoint: chanInfo.LocalPaymentBasePoint. + SerializeCompressed(), + RemotePaymentBasePoint: chanInfo.RemotePaymentBasePoint. + SerializeCompressed(), + } + } + + return rpcChannelInfos, nil +} From 692372956691384a963dfb76761f9e8580893aa4 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 2 Apr 2021 13:13:46 +0200 Subject: [PATCH 11/13] funding+server+rpcserver: use funding.ManagerConfig As a preparation to add non-exported fields to the funding manager, we want to create a NewManager function that takes a config instead of instantiating the manager directly. --- funding/manager.go | 58 +++++++++++++++++++++++++---------------- funding/manager_test.go | 8 +++--- rpcserver.go | 13 ++++----- server.go | 4 +-- 4 files changed, 49 insertions(+), 34 deletions(-) diff --git a/funding/manager.go b/funding/manager.go index c245a82..f49b6d6 100644 --- a/funding/manager.go +++ b/funding/manager.go @@ -84,7 +84,8 @@ type BaseClient interface { opts ...grpc.CallOption) (*lnrpc.AbandonChannelResponse, error) } -type Manager struct { +// ManagerConfig holds all the items passed into the funding manager externally. +type ManagerConfig struct { // DB is the client database. DB *clientdb.DB @@ -118,6 +119,19 @@ type Manager struct { NotifyShimCreated func(ourBid *order.Bid, pendingChanID [32]byte) } +// Manager is responsible for everything channel funding related during the +// match making process. +type Manager struct { + cfg *ManagerConfig +} + +// NewManager creates a new funding manager from the given config. +func NewManager(cfg *ManagerConfig) *Manager { + return &Manager{ + cfg: cfg, + } +} + // deriveFundingShim generates the proper funding shim that should be used by // the maker or taker to properly make a channel that stems off the main batch // funding transaction. @@ -161,7 +175,7 @@ func (m *Manager) deriveFundingShim(ourOrder order.Order, // scratch. ctxb := context.Background() ourKeyLocator := ourOrder.Details().MultiSigKeyLocator - ourMultiSigKey, err := m.WalletKit.DeriveKey( + ourMultiSigKey, err := m.cfg.WalletKit.DeriveKey( ctxb, &ourKeyLocator, ) if err != nil { @@ -229,7 +243,7 @@ func (m *Manager) registerFundingShim(ourBid *order.Bid, if err != nil { return err } - _, err = m.BaseClient.FundingStateStep( + _, err = m.cfg.BaseClient.FundingStateStep( ctxb, &lnrpc.FundingTransitionMsg{ Trigger: &lnrpc.FundingTransitionMsg_ShimRegister{ ShimRegister: fundingShim, @@ -243,8 +257,8 @@ func (m *Manager) registerFundingShim(ourBid *order.Bid, // In case there is a self chan balance involved, we need our channel // acceptor to be aware of the incoming order so it can verify the push // amount accordingly. - if m.NotifyShimCreated != nil { - m.NotifyShimCreated(ourBid, pendingChanID) + if m.cfg.NotifyShimCreated != nil { + m.cfg.NotifyShimCreated(ourBid, pendingChanID) } return nil @@ -260,7 +274,7 @@ func (m *Manager) PrepChannelFunding(batch *order.Batch, getOrder order.Fetcher, // As we need to change our behavior if the node has any Tor addresses, // we'll fetch the current state of our advertised addrs now. - nodeInfo, err := m.LightningClient.GetInfo(context.Background()) + nodeInfo, err := m.cfg.LightningClient.GetInfo(context.Background()) if err != nil { log.Errorf("error in GetInfo: %v", err) return err @@ -271,14 +285,14 @@ func (m *Manager) PrepChannelFunding(batch *order.Batch, getOrder order.Fetcher, // so we create a context that is valid for the whole funding step and // use that everywhere. setupCtx, cancel := context.WithTimeout( - context.Background(), m.BatchStepTimeout, + context.Background(), m.cfg.BatchStepTimeout, ) defer cancel() // Before we connect out to peers, we check that we don't get any new // channels from peers we already have channels with, in case this is // requested by the trader. - if m.NewNodesOnly { + if m.cfg.NewNodesOnly { fundingRejects, err := m.rejectDuplicateChannels(batch) if err != nil { return err @@ -416,7 +430,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, // so we create a context that is valid for the whole funding step and // use that everywhere. setupCtx, cancel := context.WithTimeout( - context.Background(), m.BatchStepTimeout, + context.Background(), m.cfg.BatchStepTimeout, ) defer cancel() @@ -427,7 +441,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, // flow, blocking until they all progress to the final state. batchTxHash := batch.BatchTX.TxHash() for ourOrderNonce, matchedOrders := range batch.MatchedOrders { - ourOrder, err := m.DB.GetOrder(ourOrderNonce) + ourOrder, err := m.cfg.DB.GetOrder(ourOrderNonce) if err != nil { return nil, err } @@ -486,7 +500,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, matchedOrderBid.SelfChanBalance, ), } - chanStream, err := m.BaseClient.OpenChannel( + chanStream, err := m.cfg.BaseClient.OpenChannel( setupCtx, fundingReq, ) @@ -591,7 +605,7 @@ func (m *Manager) waitForChannelOpen(ctx context.Context, quit <-chan struct{}, for { var chanPoint wire.OutPoint select { - case channel := <-m.PendingOpenChannels: + case channel := <-m.cfg.PendingOpenChannels: var hash chainhash.Hash copy(hash[:], channel.PendingOpenChannel.Txid) chanPoint = wire.OutPoint{ @@ -630,7 +644,7 @@ func (m *Manager) waitForChannelOpen(ctx context.Context, quit <-chan struct{}, log.Debugf("Retrieving info for channel %v", chanPoint) chanInfo, err := chaninfo.GatherChannelInfo( - ctx, m.LightningClient, m.WalletKit, chanPoint, + ctx, m.cfg.LightningClient, m.cfg.WalletKit, chanPoint, ) if err != nil { return nil, err @@ -662,7 +676,7 @@ func (m *Manager) waitForChannelOpen(ctx context.Context, quit <-chan struct{}, // // NOTE: This is part of the auctioneer.BatchCleaner interface. func (m *Manager) DeletePendingBatch() error { - return m.DB.DeletePendingBatch() + return m.cfg.DB.DeletePendingBatch() } // RemovePendingBatchArtifacts removes any funding shims or pending channels @@ -676,7 +690,7 @@ func (m *Manager) RemovePendingBatchArtifacts( batchTx *wire.MsgTx) error { err := CancelPendingFundingShims( - matchedOrders, m.BaseClient, m.DB.GetOrder, + matchedOrders, m.cfg.BaseClient, m.cfg.DB.GetOrder, ) if err != nil { // CancelPendingFundingShims only returns hard errors that @@ -689,8 +703,8 @@ func (m *Manager) RemovePendingBatchArtifacts( // from a previous round of the same batch or a previous // batch that we didn't make it into the final round. err = AbandonCanceledChannels( - matchedOrders, batchTx, m.WalletKit, m.BaseClient, - m.DB.GetOrder, + matchedOrders, batchTx, m.cfg.WalletKit, m.cfg.BaseClient, + m.cfg.DB.GetOrder, ) if err != nil { // AbandonCanceledChannels also only returns hard errors that @@ -708,7 +722,7 @@ func (m *Manager) connectToMatchedTrader(ctx context.Context, nodeKey [33]byte, addrs []net.Addr) { for _, addr := range addrs { - err := m.LightningClient.Connect( + err := m.cfg.LightningClient.Connect( ctx, nodeKey, addr.String(), false, ) if err != nil { @@ -744,7 +758,7 @@ func (m *Manager) waitForPeerConnections(ctx context.Context, // an update while we look for the already connected peers. ctxc, cancel := context.WithCancel(ctx) defer cancel() - subscription, err := m.BaseClient.SubscribePeerEvents( + subscription, err := m.cfg.BaseClient.SubscribePeerEvents( ctxc, &lnrpc.PeerEventSubscription{}, ) if err != nil { @@ -754,7 +768,7 @@ func (m *Manager) waitForPeerConnections(ctx context.Context, // Query all connected peers. This only returns active peers so once a // node key appears in this list, we can be reasonably sure the // connection is established (flapping peers notwithstanding). - resp, err := m.BaseClient.ListPeers( + resp, err := m.cfg.BaseClient.ListPeers( ctx, &lnrpc.ListPeersRequest{}, ) if err != nil { @@ -851,11 +865,11 @@ func (m *Manager) rejectDuplicateChannels( // We gather all peers from the open and pending channels. ctxb := context.Background() peers := make(map[route.Vertex]struct{}) - openChans, err := m.LightningClient.ListChannels(ctxb) + openChans, err := m.cfg.LightningClient.ListChannels(ctxb) if err != nil { return nil, fmt.Errorf("error listing open channels: %v", err) } - pendingChans, err := m.LightningClient.PendingChannels(ctxb) + pendingChans, err := m.cfg.LightningClient.PendingChannels(ctxb) if err != nil { return nil, fmt.Errorf("error listing pending channels: %v", err) diff --git a/funding/manager_test.go b/funding/manager_test.go index 2e924dd..c90298d 100644 --- a/funding/manager_test.go +++ b/funding/manager_test.go @@ -232,7 +232,7 @@ func newManagerHarness(t *testing.T) *managerHarness { msgChan: msgChan, lnMock: lightningClient, baseClientMock: baseClientMock, - mgr: &Manager{ + mgr: NewManager(&ManagerConfig{ DB: db, WalletKit: walletKitClient, LightningClient: lightningClient, @@ -240,7 +240,7 @@ func newManagerHarness(t *testing.T) *managerHarness { NewNodesOnly: true, PendingOpenChannels: msgChan, BatchStepTimeout: 400 * time.Millisecond, - }, + }), } } @@ -377,7 +377,7 @@ func TestFundingManager(t *testing.T) { // Next, make sure we get a partial reject error if we enable the "new // nodes only" flag and already have a channel with the matched node. - h.mgr.NewNodesOnly = true + h.mgr.cfg.NewNodesOnly = true h.lnMock.Channels = append(h.lnMock.Channels, lndclient.ChannelInfo{ PubKeyBytes: node1Key, }) @@ -397,7 +397,7 @@ func TestFundingManager(t *testing.T) { // As a last check of the funding preparation, make sure we get a reject // error if the connections to the remote peers couldn't be established. - h.mgr.NewNodesOnly = false + h.mgr.cfg.NewNodesOnly = false h.baseClientMock.peerList = make(map[route.Vertex]string) err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder, h.quit) require.Error(t, err) diff --git a/rpcserver.go b/rpcserver.go index 51de212..5cdca83 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -238,7 +238,7 @@ func (s *rpcServer) consumePendingOpenChannels( } // Skip any events other than the pending open channel one. - channel, ok := msg.Channel.(*lnrpc.ChannelEventUpdate_PendingOpenChannel) + _, ok := msg.Channel.(*lnrpc.ChannelEventUpdate_PendingOpenChannel) if !ok { continue } @@ -249,11 +249,12 @@ func (s *rpcServer) consumePendingOpenChannels( continue } - select { - case s.server.fundingManager.PendingOpenChannels <- channel: - case <-s.quit: - return - } + // TODO(guggero): Move this in next commit + //select { + //case s.server.fundingManager.PendingOpenChannels <- channel: + //case <-s.quit: + // return + //} } } diff --git a/server.go b/server.go index 901f4c5..d662796 100644 --- a/server.go +++ b/server.go @@ -461,7 +461,7 @@ func (s *Server) setupClient() error { // starting/stopping it though as all that logic is currently there for // the other managers as well. s.channelAcceptor = NewChannelAcceptor(s.lndServices.Client) - s.fundingManager = &funding.Manager{ + s.fundingManager = funding.NewManager(&funding.ManagerConfig{ DB: s.db, WalletKit: s.lndServices.WalletKit, LightningClient: s.lndServices.Client, @@ -472,7 +472,7 @@ func (s *Server) setupClient() error { chan *lnrpc.ChannelEventUpdate_PendingOpenChannel, ), NotifyShimCreated: s.channelAcceptor.ShimRegistered, - } + }) // Create an instance of the auctioneer client library. clientCfg := &auctioneer.Config{ From ae5da2c92f309c63323d99d00ae875baf3ca9dd2 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 2 Apr 2021 13:05:20 +0200 Subject: [PATCH 12/13] funding+rpcserver: move pending chan sub to funding mgr As a preparation to be able to serve the pending channel updates from lnd to multiple internal subscribers, we move the handling of those events to the funding manager itself. --- funding/manager.go | 142 ++++++++++++++++++++++++++++++++++++---- funding/manager_test.go | 125 +++++++++++++++++++++++------------ rpcserver.go | 97 ++++----------------------- 3 files changed, 226 insertions(+), 138 deletions(-) diff --git a/funding/manager.go b/funding/manager.go index f49b6d6..295d1ed 100644 --- a/funding/manager.go +++ b/funding/manager.go @@ -4,9 +4,11 @@ import ( "context" "encoding/hex" "fmt" + "io" "net" "strings" "sync" + "sync/atomic" "time" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -82,6 +84,14 @@ type BaseClient interface { // non-externally funded channels in dev build. AbandonChannel(ctx context.Context, in *lnrpc.AbandonChannelRequest, opts ...grpc.CallOption) (*lnrpc.AbandonChannelResponse, error) + + // SubscribeChannelEvents creates a uni-directional stream from the + // server to the client in which any updates relevant to the state of + // the channels are sent over. Events include new active channels, + // inactive channels, and closed channels. + SubscribeChannelEvents(ctx context.Context, + in *lnrpc.ChannelEventSubscription, opts ...grpc.CallOption) ( + lnrpc.Lightning_SubscribeChannelEventsClient, error) } // ManagerConfig holds all the items passed into the funding manager externally. @@ -122,13 +132,118 @@ type ManagerConfig struct { // Manager is responsible for everything channel funding related during the // match making process. type Manager struct { + started uint32 // To be used atomically. + stopped uint32 // To be used atomically. + cfg *ManagerConfig + + wg sync.WaitGroup + quit chan struct{} + + pendingOpenChanCancel func() } // NewManager creates a new funding manager from the given config. func NewManager(cfg *ManagerConfig) *Manager { return &Manager{ - cfg: cfg, + cfg: cfg, + quit: make(chan struct{}), + } +} + +// Start starts the rpcServer, making it ready to accept incoming requests. +func (m *Manager) Start() error { + if !atomic.CompareAndSwapUint32(&m.started, 0, 1) { + return nil + } + + log.Infof("Starting funding manager") + + // Subscribe to pending open channel notifications. This will be useful + // when we're creating channels with a matched order as part of a batch. + streamCtx, streamCancel := context.WithCancel(context.Background()) + m.pendingOpenChanCancel = streamCancel + subStream, err := m.cfg.BaseClient.SubscribeChannelEvents( + streamCtx, &lnrpc.ChannelEventSubscription{}, + ) + if err != nil { + return err + } + + m.wg.Add(1) + go m.consumePendingOpenChannels(subStream) + + log.Infof("Funding manager is now active") + + return nil +} + +// Stop stops the server. +func (m *Manager) Stop() error { + if !atomic.CompareAndSwapUint32(&m.stopped, 0, 1) { + return nil + } + + log.Info("Funding manager stopping") + + close(m.quit) + + // We call this before Wait to ensure the goroutine is stopped by this + // call. + m.pendingOpenChanCancel() + + m.wg.Wait() + + log.Info("Stopped funding manager") + + return nil +} + +// consumePendingOpenChannels consumes pending open channel events from the +// stream and notifies them if the trader currently has an ongoing batch. +func (m *Manager) consumePendingOpenChannels( + subStream lnrpc.Lightning_SubscribeChannelEventsClient) { + + defer m.wg.Done() + + for { + select { + case <-m.quit: + return + default: + } + + msg, err := subStream.Recv() + if err != nil { + select { + case <-m.quit: + return + default: + } + + log.Errorf("Unable to read channel event: %v", err) + + // If the lnd node shut down, there's no use continuing. + if err == io.EOF || err == io.ErrUnexpectedEOF || + status.Code(err) == codes.Unavailable { + + return + } + + continue + } + + // Skip any events other than the pending open channel one. + channel, ok := msg.Channel.(*lnrpc.ChannelEventUpdate_PendingOpenChannel) + if !ok { + continue + } + + select { + case m.cfg.PendingOpenChannels <- channel: + case <-m.quit: + return + } } } @@ -266,8 +381,8 @@ func (m *Manager) registerFundingShim(ourBid *order.Bid, // PrepChannelFunding preps the backing node to either receive or initiate a // channel funding based on the items in the order batch. -func (m *Manager) PrepChannelFunding(batch *order.Batch, getOrder order.Fetcher, - quit <-chan struct{}) error { +func (m *Manager) PrepChannelFunding(batch *order.Batch, + getOrder order.Fetcher) error { log.Infof("Batch(%x): preparing channel funding for %v orders", batch.ID[:], len(batch.MatchedOrders)) @@ -392,7 +507,7 @@ func (m *Manager) PrepChannelFunding(batch *order.Batch, getOrder order.Fetcher, // We need to wait for all connections to be established now. Otherwise // the asker won't be able to open the channel as it doesn't know the // connection details of the bidder. - return m.waitForPeerConnections(setupCtx, connsInitiated, batch, quit) + return m.waitForPeerConnections(setupCtx, connsInitiated, batch) } // BatchChannelSetup will attempt to establish new funding flows with all @@ -400,8 +515,8 @@ func (m *Manager) PrepChannelFunding(batch *order.Batch, getOrder order.Fetcher, // will block until the channel is considered pending. Once this phase is // complete, and the batch execution transaction broadcast, the channel will be // finalized and locked in. -func (m *Manager) BatchChannelSetup(batch *order.Batch, - quit <-chan struct{}) (map[wire.OutPoint]*chaninfo.ChannelInfo, error) { +func (m *Manager) BatchChannelSetup( + batch *order.Batch) (map[wire.OutPoint]*chaninfo.ChannelInfo, error) { var ( eg errgroup.Group @@ -530,7 +645,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, for { select { - case <-quit: + case <-m.quit: return fmt.Errorf("server " + "shutting down") default: @@ -569,7 +684,7 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, // we can report that together with the other errors. if err := eg.Wait(); err != nil { select { - case <-quit: + case <-m.quit: return nil, err default: } @@ -577,14 +692,14 @@ func (m *Manager) BatchChannelSetup(batch *order.Batch, // We've kicked off all channel open streams. We'll now need to wait for // either completion of the funding process or a timeout. - return m.waitForChannelOpen(setupCtx, quit, chanPoints, fundingRejects) + return m.waitForChannelOpen(setupCtx, chanPoints, fundingRejects) } // waitForChannelOpen waits until we get a pending open channel message for each // of the channel outpoints provided. If we don't get all of them in time, we'll // instead return a reject error with all the orders for which the funding // failed. -func (m *Manager) waitForChannelOpen(ctx context.Context, quit <-chan struct{}, +func (m *Manager) waitForChannelOpen(ctx context.Context, chanPoints map[wire.OutPoint]order.Nonce, fundingRejects map[order.Nonce]*auctioneerrpc.OrderReject) ( map[wire.OutPoint]*chaninfo.ChannelInfo, error) { @@ -630,7 +745,7 @@ func (m *Manager) waitForChannelOpen(ctx context.Context, quit <-chan struct{}, RejectedOrders: fundingRejects, } - case <-quit: + case <-m.quit: return nil, fmt.Errorf("server shutting down") } @@ -751,8 +866,7 @@ func (m *Manager) connectToMatchedTrader(ctx context.Context, // NOTE: The passed context MUST have a timeout applied to it, otherwise this // method will block forever in case a connection doesn't succeed. func (m *Manager) waitForPeerConnections(ctx context.Context, - peers map[route.Vertex]struct{}, batch *order.Batch, - quit <-chan struct{}) error { + peers map[route.Vertex]struct{}, batch *order.Batch) error { // First of all, subscribe to new peer events so we certainly don't miss // an update while we look for the already connected peers. @@ -808,7 +922,7 @@ func (m *Manager) waitForPeerConnections(ctx context.Context, } // We're shutting down, nothing more to do here. - case <-quit: + case <-m.quit: return nil default: diff --git a/funding/manager_test.go b/funding/manager_test.go index c90298d..4b52247 100644 --- a/funding/manager_test.go +++ b/funding/manager_test.go @@ -85,13 +85,44 @@ func (i *peerEventStream) Recv() (*lnrpc.PeerEvent, error) { } } +type channelEventStream struct { + lnrpc.Lightning_SubscribeChannelEventsClient + + updateChan chan *lnrpc.ChannelEventUpdate + ctx context.Context + cancelSubscription chan struct{} + quit chan struct{} +} + +func (c *channelEventStream) Recv() (*lnrpc.ChannelEventUpdate, error) { + select { + case msg := <-c.updateChan: + return msg, nil + + // To mimic the real GRPC client, we'll return an error status. + case <-c.ctx.Done(): + return nil, status.Error( + codes.DeadlineExceeded, c.ctx.Err().Error(), + ) + + case <-c.cancelSubscription: + return nil, status.Error( + codes.Canceled, "subscription canceled", + ) + + case <-c.quit: + return nil, context.Canceled + } +} + type fundingBaseClientMock struct { lightningClient *test.MockLightning fundingShims map[[32]byte]*lnrpc.ChanPointShim - peerList map[route.Vertex]string - peerEvents chan *lnrpc.PeerEvent - cancelSub chan struct{} + peerList map[route.Vertex]string + peerEvents chan *lnrpc.PeerEvent + channelEvents chan *lnrpc.ChannelEventUpdate + cancelSub chan struct{} quit chan struct{} } @@ -157,6 +188,17 @@ func (m *fundingBaseClientMock) OpenChannel(ctx context.Context, return stream, nil } +func (m *fundingBaseClientMock) SubscribeChannelEvents(ctx context.Context, + _ *lnrpc.ChannelEventSubscription, _ ...grpc.CallOption) ( + lnrpc.Lightning_SubscribeChannelEventsClient, error) { + + return &channelEventStream{ + quit: m.quit, + ctx: ctx, + updateChan: m.channelEvents, + }, nil +} + func (m *fundingBaseClientMock) ListPeers(_ context.Context, _ *lnrpc.ListPeersRequest, _ ...grpc.CallOption) (*lnrpc.ListPeersResponse, error) { @@ -196,7 +238,6 @@ type managerHarness struct { tempDir string db *clientdb.DB quit chan struct{} - msgChan chan *lnrpc.ChannelEventUpdate_PendingOpenChannel lnMock *test.MockLightning baseClientMock *fundingBaseClientMock mgr *Manager @@ -221,26 +262,31 @@ func newManagerHarness(t *testing.T) *managerHarness { fundingShims: make(map[[32]byte]*lnrpc.ChanPointShim), peerList: make(map[route.Vertex]string), peerEvents: make(chan *lnrpc.PeerEvent), + channelEvents: make(chan *lnrpc.ChannelEventUpdate), cancelSub: make(chan struct{}), quit: quit, } + mgr := NewManager(&ManagerConfig{ + DB: db, + WalletKit: walletKitClient, + LightningClient: lightningClient, + BaseClient: baseClientMock, + NewNodesOnly: true, + PendingOpenChannels: msgChan, + BatchStepTimeout: 400 * time.Millisecond, + }) + + err = mgr.Start() + require.NoError(t, err) + return &managerHarness{ t: t, tempDir: tempDir, db: db, quit: quit, - msgChan: msgChan, lnMock: lightningClient, baseClientMock: baseClientMock, - mgr: NewManager(&ManagerConfig{ - DB: db, - WalletKit: walletKitClient, - LightningClient: lightningClient, - BaseClient: baseClientMock, - NewNodesOnly: true, - PendingOpenChannels: msgChan, - BatchStepTimeout: 400 * time.Millisecond, - }), + mgr: mgr, } } @@ -248,6 +294,7 @@ func (m *managerHarness) stop() { close(m.quit) require.NoError(m.t, m.db.Close()) require.NoError(m.t, os.RemoveAll(m.tempDir)) + require.NoError(m.t, m.mgr.Stop()) } // TestFundingManager tests that the two main steps of the funding manager (the @@ -335,7 +382,7 @@ func TestFundingManager(t *testing.T) { h.baseClientMock.peerList = map[route.Vertex]string{ node1Key: "1.1.1.1", } - err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder, h.quit) + err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder) require.NoError(t, err) // Verify we have the expected connections and funding shims registered. @@ -381,7 +428,7 @@ func TestFundingManager(t *testing.T) { h.lnMock.Channels = append(h.lnMock.Channels, lndclient.ChannelInfo{ PubKeyBytes: node1Key, }) - err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder, h.quit) + err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder) require.Error(t, err) expectedErr := &MatchRejectErr{ @@ -399,7 +446,7 @@ func TestFundingManager(t *testing.T) { // error if the connections to the remote peers couldn't be established. h.mgr.cfg.NewNodesOnly = false h.baseClientMock.peerList = make(map[route.Vertex]string) - err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder, h.quit) + err = h.mgr.PrepChannelFunding(batch, h.db.GetOrder) require.Error(t, err) expectedErr = &MatchRejectErr{ @@ -418,27 +465,31 @@ func TestFundingManager(t *testing.T) { // message for the one where we are the asker so we simulate two msgs. go func() { timeout := time.After(time.Second) - msg := &lnrpc.ChannelEventUpdate_PendingOpenChannel{ - PendingOpenChannel: &lnrpc.PendingUpdate{ - Txid: txidHash[:], - OutputIndex: 0, + msg := &lnrpc.ChannelEventUpdate{ + Channel: &lnrpc.ChannelEventUpdate_PendingOpenChannel{ + PendingOpenChannel: &lnrpc.PendingUpdate{ + Txid: txidHash[:], + OutputIndex: 0, + }, }, } // Send the message for the first channel. select { - case h.msgChan <- msg: + case h.baseClientMock.channelEvents <- msg: case <-timeout: } // And again for the second channel. - msg2 := &lnrpc.ChannelEventUpdate_PendingOpenChannel{ - PendingOpenChannel: &lnrpc.PendingUpdate{ - Txid: txidHash[:], - OutputIndex: 1, + msg2 := &lnrpc.ChannelEventUpdate{ + Channel: &lnrpc.ChannelEventUpdate_PendingOpenChannel{ + PendingOpenChannel: &lnrpc.PendingUpdate{ + Txid: txidHash[:], + OutputIndex: 1, + }, }, } select { - case h.msgChan <- msg2: + case h.baseClientMock.channelEvents <- msg2: case <-timeout: } }() @@ -452,13 +503,13 @@ func TestFundingManager(t *testing.T) { ChannelPoint: fmt.Sprintf("%s:1", txidHash.String()), }) h.lnMock.ScbKeyRing.EncryptionKey.PubKey = pubKeyAsk - chanInfo, err := h.mgr.BatchChannelSetup(batch, h.quit) + chanInfo, err := h.mgr.BatchChannelSetup(batch) require.NoError(t, err) require.Equal(t, 2, len(chanInfo)) // Finally, make sure we get a timeout error if no channel open messages // are received. - _, err = h.mgr.BatchChannelSetup(batch, h.quit) + _, err = h.mgr.BatchChannelSetup(batch) require.Error(t, err) code := &auctioneerrpc.OrderReject{ @@ -493,9 +544,7 @@ func TestWaitForPeerConnections(t *testing.T) { node1Key: {}, node2Key: {}, } - err := h.mgr.waitForPeerConnections( - ctxt, expectedConnections, nil, h.quit, - ) + err := h.mgr.waitForPeerConnections(ctxt, expectedConnections, nil) require.NoError(t, err) // Next, make sure that connections established while waiting are @@ -524,9 +573,7 @@ func TestWaitForPeerConnections(t *testing.T) { node1Key: {}, node2Key: {}, } - err = h.mgr.waitForPeerConnections( - ctxt, expectedConnections, nil, h.quit, - ) + err = h.mgr.waitForPeerConnections(ctxt, expectedConnections, nil) require.NoError(t, err) // Final test, make sure we get the correct error message back if we @@ -561,9 +608,7 @@ func TestWaitForPeerConnections(t *testing.T) { node1Key: {}, node2Key: {}, } - err = h.mgr.waitForPeerConnections( - ctxt, expectedConnections, fakeBatch, h.quit, - ) + err = h.mgr.waitForPeerConnections(ctxt, expectedConnections, fakeBatch) require.Error(t, err) code := &auctioneerrpc.OrderReject{ @@ -588,9 +633,7 @@ func TestWaitForPeerConnections(t *testing.T) { close(h.baseClientMock.cancelSub) }() - err = h.mgr.waitForPeerConnections( - ctxt, expectedConnections, fakeBatch, h.quit, - ) + err = h.mgr.waitForPeerConnections(ctxt, expectedConnections, fakeBatch) require.Error(t, err) require.Equal(t, expectedErr, err) } diff --git a/rpcserver.go b/rpcserver.go index 5cdca83..cfec16f 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -6,7 +6,6 @@ import ( "encoding/hex" "errors" "fmt" - "io" "sync" "sync/atomic" "time" @@ -36,8 +35,6 @@ import ( "github.com/lightningnetwork/lnd/lnwallet/chainfee" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/signal" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" ) const ( @@ -63,12 +60,11 @@ type rpcServer struct { accountManager *account.Manager orderManager *order.Manager - quit chan struct{} - wg sync.WaitGroup - blockNtfnCancel func() - pendingOpenChannelStreamCancel func() - recoveryMutex sync.Mutex - recoveryPending bool + quit chan struct{} + wg sync.WaitGroup + blockNtfnCancel func() + recoveryMutex sync.Mutex + recoveryPending bool // wumboSupported is true if the backing lnd node supports wumbo // channels. @@ -167,17 +163,6 @@ func (s *rpcServer) Start() error { s.updateHeight(height) - // Subscribe to pending open channel notifications. This will be useful - // when we're creating channels with a matched order as part of a batch. - streamCtx, streamCancel := context.WithCancel(ctx) - s.pendingOpenChannelStreamCancel = streamCancel - subStream, err := s.lndClient.SubscribeChannelEvents( - streamCtx, &lnrpc.ChannelEventSubscription{}, - ) - if err != nil { - return err - } - // Start the auctioneer client first to establish a connection. if err := s.auctioneer.Start(); err != nil { return fmt.Errorf("unable to start auctioneer client: %v", err) @@ -190,74 +175,21 @@ func (s *rpcServer) Start() error { if err := s.orderManager.Start(); err != nil { return fmt.Errorf("unable to start order manager: %v", err) } + if err := s.server.fundingManager.Start(); err != nil { + return fmt.Errorf("unable to start funding manager: %v", err) + } if err := s.server.channelAcceptor.Start(blockErrChan); err != nil { return fmt.Errorf("unable to start channel acceptor: %v", err) } - s.wg.Add(2) + s.wg.Add(1) go s.serverHandler(blockChan, blockErrChan) - go s.consumePendingOpenChannels(subStream) rpcLog.Infof("Trader server is now active") return nil } -// consumePendingOpenChannels consumes pending open channel events from the -// stream and notifies them if the trader currently has an ongoing batch. -func (s *rpcServer) consumePendingOpenChannels( - subStream lnrpc.Lightning_SubscribeChannelEventsClient) { - - defer s.wg.Done() - - for { - select { - case <-s.quit: - return - default: - } - - msg, err := subStream.Recv() - if err != nil { - select { - case <-s.quit: - return - default: - } - - rpcLog.Errorf("Unable to read channel event: %v", err) - - // If the lnd node shut down, there's no use continuing. - if err == io.EOF || err == io.ErrUnexpectedEOF || - status.Code(err) == codes.Unavailable { - - return - } - - continue - } - - // Skip any events other than the pending open channel one. - _, ok := msg.Channel.(*lnrpc.ChannelEventUpdate_PendingOpenChannel) - if !ok { - continue - } - - // If we don't have a pending batch, then there's no need to - // notify any pending open channels.. - if !s.orderManager.HasPendingBatch() { - continue - } - - // TODO(guggero): Move this in next commit - //select { - //case s.server.fundingManager.PendingOpenChannels <- channel: - //case <-s.quit: - // return - //} - } -} - // Stop stops the server. func (s *rpcServer) Stop() error { if !atomic.CompareAndSwapUint32(&s.stopped, 0, 1) { @@ -265,6 +197,9 @@ func (s *rpcServer) Stop() error { } rpcLog.Info("Trader server stopping") + if err := s.server.fundingManager.Stop(); err != nil { + rpcLog.Errorf("Error stopping funding manager: %v", err) + } s.server.channelAcceptor.Stop() s.accountManager.Stop() s.orderManager.Stop() @@ -274,10 +209,6 @@ func (s *rpcServer) Stop() error { close(s.quit) - // We call this before Wait to ensure the goroutine is stopped by this - // call. - s.pendingOpenChannelStreamCancel() - s.wg.Wait() s.blockNtfnCancel() @@ -418,7 +349,7 @@ func (s *rpcServer) handleServerMessage( // end which include applying any order match predicates, // connecting out to peers, and registering funding shim. err = s.server.fundingManager.PrepChannelFunding( - batch, s.server.db.GetOrder, s.quit, + batch, s.server.db.GetOrder, ) if err != nil { rpcLog.Warnf("Error preparing channel funding: %v", @@ -439,7 +370,7 @@ func (s *rpcServer) handleServerMessage( // once all channel partners have responded. batch := s.orderManager.PendingBatch() channelKeys, err := s.server.fundingManager.BatchChannelSetup( - batch, s.quit, + batch, ) if err != nil { rpcLog.Errorf("Error setting up channels: %v", err) From 2730d69c2fa11e3d53d525480a9f715f29d348e5 Mon Sep 17 00:00:00 2001 From: Oliver Gugger Date: Fri, 2 Apr 2021 14:08:14 +0200 Subject: [PATCH 13/13] funding+server: turn pending chan sub into subscription server As a preparation for allowing multiple managers to consume the pending channel updates, we move those events from a single channel to a subscribe.Server construct. We also pass in the subscription client to use into waitForChannelOpen so we can use a different one in different flows. --- funding/manager.go | 65 +++++++++++++++++++++++++++++++---------- funding/manager_test.go | 14 ++++----- server.go | 15 ++++------ 3 files changed, 61 insertions(+), 33 deletions(-) diff --git a/funding/manager.go b/funding/manager.go index 295d1ed..302fd49 100644 --- a/funding/manager.go +++ b/funding/manager.go @@ -22,6 +22,7 @@ import ( "github.com/lightningnetwork/lnd/input" "github.com/lightningnetwork/lnd/lnrpc" "github.com/lightningnetwork/lnd/routing/route" + "github.com/lightningnetwork/lnd/subscribe" "github.com/lightningnetwork/lnd/tor" "golang.org/x/sync/errgroup" "google.golang.org/grpc" @@ -114,11 +115,6 @@ type ManagerConfig struct { // node doesn't already have channels with. NewNodesOnly bool - // PendingOpenChannels is a channel through which we'll receive - // notifications for pending open channels resulting from a successful - // batch. - PendingOpenChannels chan *lnrpc.ChannelEventUpdate_PendingOpenChannel - // BatchStepTimeout is the timeout the manager uses when executing a // single batch step. BatchStepTimeout time.Duration @@ -141,13 +137,16 @@ type Manager struct { quit chan struct{} pendingOpenChanCancel func() + pendingOpenChanServer *subscribe.Server + pendingOpenChanClient *subscribe.Client } // NewManager creates a new funding manager from the given config. func NewManager(cfg *ManagerConfig) *Manager { return &Manager{ - cfg: cfg, - quit: make(chan struct{}), + cfg: cfg, + quit: make(chan struct{}), + pendingOpenChanServer: subscribe.NewServer(), } } @@ -170,6 +169,20 @@ func (m *Manager) Start() error { return err } + if err := m.pendingOpenChanServer.Start(); err != nil { + return fmt.Errorf("error starting pending chan subscription "+ + "server: %v", err) + } + + // We want to make sure we don't miss any channel updates as long as we + // are running. But we might not be the only manager interested in the + // updates, that's why we are a client to our own server. + m.pendingOpenChanClient, err = m.SubscribePendingOpenChan() + if err != nil { + return fmt.Errorf("error subscribing to pending open "+ + "channel events: %v", err) + } + m.wg.Add(1) go m.consumePendingOpenChannels(subStream) @@ -190,6 +203,11 @@ func (m *Manager) Stop() error { // We call this before Wait to ensure the goroutine is stopped by this // call. + m.pendingOpenChanClient.Cancel() + if err := m.pendingOpenChanServer.Stop(); err != nil { + return fmt.Errorf("error stopping pending chan subscription "+ + "server: %v", err) + } m.pendingOpenChanCancel() m.wg.Wait() @@ -239,14 +257,19 @@ func (m *Manager) consumePendingOpenChannels( continue } - select { - case m.cfg.PendingOpenChannels <- channel: - case <-m.quit: - return + update := channel.PendingOpenChannel + if err := m.pendingOpenChanServer.SendUpdate(update); err != nil { + log.Errorf("Error sending open channel update: %v", err) } } } +// SubscribePendingOpenChan creates a new subscription client to receive events +// for pending open channels from lnd. +func (m *Manager) SubscribePendingOpenChan() (*subscribe.Client, error) { + return m.pendingOpenChanServer.Subscribe() +} + // deriveFundingShim generates the proper funding shim that should be used by // the maker or taker to properly make a channel that stems off the main batch // funding transaction. @@ -692,7 +715,9 @@ func (m *Manager) BatchChannelSetup( // We've kicked off all channel open streams. We'll now need to wait for // either completion of the funding process or a timeout. - return m.waitForChannelOpen(setupCtx, chanPoints, fundingRejects) + return m.waitForChannelOpen( + setupCtx, chanPoints, m.pendingOpenChanClient, fundingRejects, + ) } // waitForChannelOpen waits until we get a pending open channel message for each @@ -700,7 +725,7 @@ func (m *Manager) BatchChannelSetup( // instead return a reject error with all the orders for which the funding // failed. func (m *Manager) waitForChannelOpen(ctx context.Context, - chanPoints map[wire.OutPoint]order.Nonce, + chanPoints map[wire.OutPoint]order.Nonce, chanUpdates *subscribe.Client, fundingRejects map[order.Nonce]*auctioneerrpc.OrderReject) ( map[wire.OutPoint]*chaninfo.ChannelInfo, error) { @@ -720,12 +745,17 @@ func (m *Manager) waitForChannelOpen(ctx context.Context, for { var chanPoint wire.OutPoint select { - case channel := <-m.cfg.PendingOpenChannels: + case channel := <-chanUpdates.Updates(): + pendingChannel, ok := channel.(*lnrpc.PendingUpdate) + if !ok || pendingChannel == nil { + continue + } + var hash chainhash.Hash - copy(hash[:], channel.PendingOpenChannel.Txid) + copy(hash[:], pendingChannel.Txid) chanPoint = wire.OutPoint{ Hash: hash, - Index: channel.PendingOpenChannel.OutputIndex, + Index: pendingChannel.OutputIndex, } case <-ctx.Done(): @@ -745,6 +775,9 @@ func (m *Manager) waitForChannelOpen(ctx context.Context, RejectedOrders: fundingRejects, } + case <-m.pendingOpenChanClient.Quit(): + return nil, fmt.Errorf("server shutting down") + case <-m.quit: return nil, fmt.Errorf("server shutting down") } diff --git a/funding/manager_test.go b/funding/manager_test.go index 4b52247..1111c44 100644 --- a/funding/manager_test.go +++ b/funding/manager_test.go @@ -254,7 +254,6 @@ func newManagerHarness(t *testing.T) *managerHarness { } quit := make(chan struct{}) - msgChan := make(chan *lnrpc.ChannelEventUpdate_PendingOpenChannel) lightningClient := test.NewMockLightning() walletKitClient := test.NewMockWalletKit() baseClientMock := &fundingBaseClientMock{ @@ -267,13 +266,12 @@ func newManagerHarness(t *testing.T) *managerHarness { quit: quit, } mgr := NewManager(&ManagerConfig{ - DB: db, - WalletKit: walletKitClient, - LightningClient: lightningClient, - BaseClient: baseClientMock, - NewNodesOnly: true, - PendingOpenChannels: msgChan, - BatchStepTimeout: 400 * time.Millisecond, + DB: db, + WalletKit: walletKitClient, + LightningClient: lightningClient, + BaseClient: baseClientMock, + NewNodesOnly: true, + BatchStepTimeout: 400 * time.Millisecond, }) err = mgr.Start() diff --git a/server.go b/server.go index d662796..d14484c 100644 --- a/server.go +++ b/server.go @@ -462,15 +462,12 @@ func (s *Server) setupClient() error { // the other managers as well. s.channelAcceptor = NewChannelAcceptor(s.lndServices.Client) s.fundingManager = funding.NewManager(&funding.ManagerConfig{ - DB: s.db, - WalletKit: s.lndServices.WalletKit, - LightningClient: s.lndServices.Client, - BaseClient: s.lndClient, - BatchStepTimeout: order.DefaultBatchStepTimeout, - NewNodesOnly: s.cfg.NewNodesOnly, - PendingOpenChannels: make( - chan *lnrpc.ChannelEventUpdate_PendingOpenChannel, - ), + DB: s.db, + WalletKit: s.lndServices.WalletKit, + LightningClient: s.lndServices.Client, + BaseClient: s.lndClient, + BatchStepTimeout: order.DefaultBatchStepTimeout, + NewNodesOnly: s.cfg.NewNodesOnly, NotifyShimCreated: s.channelAcceptor.ShimRegistered, })