mirror of
https://github.com/lightninglabs/pool.git
synced 2026-08-20 13:27:51 +02:00
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.
This commit is contained in:
Oliver Gugger
parent
6923729566
commit
ae5da2c92f
3 changed files with 226 additions and 138 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
97
rpcserver.go
97
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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue