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)