From f4f446519b8d8d1b41fcf438dadaac5d9e067d6f Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Mon, 29 Nov 2021 08:32:07 +0800 Subject: [PATCH 1/4] invoices: remove potential cyclic pointer references This commit removes the pointer reference to InvoiceRegistry when creating an InvoiceSubscriptionKit and mounts the channels needed for communication. --- invoices/invoiceregistry.go | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/invoices/invoiceregistry.go b/invoices/invoiceregistry.go index c49535330..543c41968 100644 --- a/invoices/invoiceregistry.go +++ b/invoices/invoiceregistry.go @@ -1404,8 +1404,15 @@ func (i *InvoiceRegistry) notifyClients(hash lntypes.Hash, // invoiceSubscriptionKit defines that are common to both all invoice // subscribers and single invoice subscribers. type invoiceSubscriptionKit struct { - id uint32 - inv *InvoiceRegistry + id uint32 // nolint:structcheck + + // subscriptionCancels is a chan mounted to InvoiceRegistry that + // signals the current subscription has been cancled. + subscriptionCancels chan uint32 + + // quit is a chan mouted to InvoiceRegistry that signals a shutdown. + quit chan struct{} + ntfnQueue *queue.ConcurrentQueue canceled uint32 // To be used atomically. @@ -1465,8 +1472,8 @@ func (i *invoiceSubscriptionKit) Cancel() { } select { - case i.inv.subscriptionCancels <- i.id: - case <-i.inv.quit: + case i.subscriptionCancels <- i.id: + case <-i.quit: } i.ntfnQueue.Stop() @@ -1478,7 +1485,7 @@ func (i *invoiceSubscriptionKit) Cancel() { func (i *invoiceSubscriptionKit) notify(event *invoiceEvent) error { select { case i.ntfnQueue.ChanIn() <- event: - case <-i.inv.quit: + case <-i.quit: return ErrShuttingDown } @@ -1499,9 +1506,10 @@ func (i *InvoiceRegistry) SubscribeNotifications( addIndex: addIndex, settleIndex: settleIndex, invoiceSubscriptionKit: invoiceSubscriptionKit{ - inv: i, - ntfnQueue: queue.NewConcurrentQueue(20), - cancelChan: make(chan struct{}), + subscriptionCancels: i.subscriptionCancels, + quit: i.quit, + ntfnQueue: queue.NewConcurrentQueue(20), + cancelChan: make(chan struct{}), }, } client.ntfnQueue.Start() @@ -1596,9 +1604,10 @@ func (i *InvoiceRegistry) SubscribeSingleInvoice( client := &SingleInvoiceSubscription{ Updates: make(chan *channeldb.Invoice), invoiceSubscriptionKit: invoiceSubscriptionKit{ - inv: i, - ntfnQueue: queue.NewConcurrentQueue(20), - cancelChan: make(chan struct{}), + subscriptionCancels: i.subscriptionCancels, + quit: i.quit, + ntfnQueue: queue.NewConcurrentQueue(20), + cancelChan: make(chan struct{}), }, invoiceRef: channeldb.InvoiceRefByHash(hash), } From 3230cc48223debda4cd47fdf530cd879a2572b11 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Mon, 29 Nov 2021 13:45:42 +0800 Subject: [PATCH 2/4] invoices: remove redundant mutex lock --- invoices/invoiceregistry.go | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/invoices/invoiceregistry.go b/invoices/invoiceregistry.go index 543c41968..a2ce60550 100644 --- a/invoices/invoiceregistry.go +++ b/invoices/invoiceregistry.go @@ -101,13 +101,13 @@ func (r *htlcReleaseEvent) Less(other queue.PriorityQueueItem) bool { type InvoiceRegistry struct { sync.RWMutex + nextClientID uint32 // must be used atomically + cdb *channeldb.DB // cfg contains the registry's configuration parameters. cfg *RegistryConfig - clientMtx sync.Mutex - nextClientID uint32 notificationClients map[uint32]*InvoiceSubscription singleNotificationClients map[uint32]*SingleInvoiceSubscription @@ -1514,10 +1514,9 @@ func (i *InvoiceRegistry) SubscribeNotifications( } client.ntfnQueue.Start() - i.clientMtx.Lock() - client.id = i.nextClientID - i.nextClientID++ - i.clientMtx.Unlock() + // Always increment by 1 first, and our client ID will start with 1, + // not 0. + client.id = atomic.AddUint32(&i.nextClientID, 1) // Before we register this new invoice subscription, we'll launch a new // goroutine that will proxy all notifications appended to the end of @@ -1613,10 +1612,9 @@ func (i *InvoiceRegistry) SubscribeSingleInvoice( } client.ntfnQueue.Start() - i.clientMtx.Lock() - client.id = i.nextClientID - i.nextClientID++ - i.clientMtx.Unlock() + // Always increment by 1 first, and our client ID will start with 1, + // not 0. + client.id = atomic.AddUint32(&i.nextClientID, 1) // Before we register this new invoice subscription, we'll launch a new // goroutine that will proxy all notifications appended to the end of From df2ecd6bc5535b468a600c41cd692c2e258d4137 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Mon, 29 Nov 2021 13:50:40 +0800 Subject: [PATCH 3/4] invoices: refactor invoice subscriptions to avoid blocking This commit fixes a potential blocking when notifying invoice updates. When a new subscription client is created followed by an immediate cancel, it's likely the client will be removed from the registry's map(noop) and then added to its map again. This subscription will then be kept in registry until lnd is restarted. Another more serious issue is when multiple subscriptions are made for the same invoice, when the above case happens, other subscriptions may never send invoice updates because a previous client has a stopped notification queue that blocks following notifications. --- invoices/invoiceregistry.go | 197 +++++++++++++-------------- lnrpc/invoicesrpc/invoices_server.go | 6 +- 2 files changed, 98 insertions(+), 105 deletions(-) diff --git a/invoices/invoiceregistry.go b/invoices/invoiceregistry.go index a2ce60550..a72555fcf 100644 --- a/invoices/invoiceregistry.go +++ b/invoices/invoiceregistry.go @@ -108,15 +108,15 @@ type InvoiceRegistry struct { // cfg contains the registry's configuration parameters. cfg *RegistryConfig - notificationClients map[uint32]*InvoiceSubscription + notificationClients map[uint32]*InvoiceSubscription + + // TODO(yy): use map[lntypes.Hash]*SingleInvoiceSubscription for better + // performance. singleNotificationClients map[uint32]*SingleInvoiceSubscription - newSubscriptions chan *InvoiceSubscription - subscriptionCancels chan uint32 - - // invoiceEvents is a single channel over which both invoice updates and - // new single invoice subscriptions are carried. - invoiceEvents chan interface{} + // invoiceEvents is a single channel over which invoice updates are + // carried. + invoiceEvents chan *invoiceEvent // subscriptions is a map from a circuit key to a list of subscribers. // It is used for efficient notification of links. @@ -147,9 +147,7 @@ func NewRegistry(cdb *channeldb.DB, expiryWatcher *InvoiceExpiryWatcher, cdb: cdb, notificationClients: make(map[uint32]*InvoiceSubscription), singleNotificationClients: make(map[uint32]*SingleInvoiceSubscription), - newSubscriptions: make(chan *InvoiceSubscription), - subscriptionCancels: make(chan uint32), - invoiceEvents: make(chan interface{}, 100), + invoiceEvents: make(chan *invoiceEvent, 100), hodlSubscriptions: make(map[channeldb.CircuitKey]map[chan<- interface{}]struct{}), hodlReverseSubscriptions: make(map[chan<- interface{}]map[channeldb.CircuitKey]struct{}), cfg: cfg, @@ -301,61 +299,18 @@ func (i *InvoiceRegistry) invoiceEventLoop() { } select { - // A new invoice subscription for all invoices has just arrived! - // We'll query for any backlog notifications, then add it to the - // set of clients. - case newClient := <-i.newSubscriptions: - log.Infof("New invoice subscription "+ - "client: id=%v", newClient.id) - - // With the backlog notifications delivered (if any), - // we'll add this to our active subscriptions and - // continue. - i.notificationClients[newClient.id] = newClient - - // A client no longer wishes to receive invoice notifications. - // So we'll remove them from the set of active clients. - case clientID := <-i.subscriptionCancels: - log.Infof("Cancelling invoice subscription for "+ - "client=%v", clientID) - - delete(i.notificationClients, clientID) - delete(i.singleNotificationClients, clientID) - - // An invoice event has come in. This can either be an update to - // an invoice or a new single invoice subscriber. Both type of - // events are passed in via the same channel, to make sure that - // subscribers get a consistent view of the event sequence. + // A sub-systems has just modified the invoice state, so we'll + // dispatch notifications to all registered clients. case event := <-i.invoiceEvents: - switch e := event.(type) { + // For backwards compatibility, do not notify all + // invoice subscribers of cancel and accept events. + state := event.invoice.State + if state != channeldb.ContractCanceled && + state != channeldb.ContractAccepted { - // A sub-systems has just modified the invoice state, so - // we'll dispatch notifications to all registered - // clients. - case *invoiceEvent: - // For backwards compatibility, do not notify - // all invoice subscribers of cancel and accept - // events. - state := e.invoice.State - if state != channeldb.ContractCanceled && - state != channeldb.ContractAccepted { - - i.dispatchToClients(e) - } - i.dispatchToSingleClients(e) - - // A new single invoice subscription has arrived. Add it - // to the set of clients. It is important to do this in - // sequence with any other invoice events, because an - // initial invoice update has already been sent out to - // the subscriber. - case *SingleInvoiceSubscription: - log.Infof("New single invoice subscription "+ - "client: id=%v, ref=%v", e.id, - e.invoiceRef) - - i.singleNotificationClients[e.id] = e + i.dispatchToClients(event) } + i.dispatchToSingleClients(event) // A new htlc came in for auto-release. case event := <-i.htlcAutoReleaseChan: @@ -386,27 +341,36 @@ func (i *InvoiceRegistry) invoiceEventLoop() { } } -// dispatchToSingleClients passes the supplied event to all notification clients -// that subscribed to all the invoice this event applies to. +// dispatchToSingleClients passes the supplied event to all notification +// clients that subscribed to all the invoice this event applies to. func (i *InvoiceRegistry) dispatchToSingleClients(event *invoiceEvent) { // Dispatch to single invoice subscribers. - for _, client := range i.singleNotificationClients { + clients := i.copySingleClients() + for _, client := range clients { payHash := client.invoiceRef.PayHash() + if payHash == nil || *payHash != event.hash { continue } + if atomic.LoadUint32(&client.canceled) == 1 { + log.Errorf("Client(id=%v) has stopped, skipped "+ + "notification for event(pay_hash=%v)", + client.id, payHash) + continue + } client.notify(event) } } // dispatchToClients passes the supplied event to all notification clients that -// subscribed to all invoices. Add and settle indices are used to make sure that -// clients don't receive duplicate or unwanted events. +// subscribed to all invoices. Add and settle indices are used to make sure +// that clients don't receive duplicate or unwanted events. func (i *InvoiceRegistry) dispatchToClients(event *invoiceEvent) { invoice := event.invoice - for clientID, client := range i.notificationClients { + clients := i.copyClients() + for clientID, client := range clients { // Before we dispatch this event, we'll check // to ensure that this client hasn't already // received this notification in order to @@ -568,6 +532,9 @@ func (i *InvoiceRegistry) deliverSingleBacklogEvents( return err } + log.Debugf("Client(id=%v) delivered single backlog event: payHash=%v", + client.id, payHash) + return nil } @@ -1406,10 +1373,6 @@ func (i *InvoiceRegistry) notifyClients(hash lntypes.Hash, type invoiceSubscriptionKit struct { id uint32 // nolint:structcheck - // subscriptionCancels is a chan mounted to InvoiceRegistry that - // signals the current subscription has been cancled. - subscriptionCancels chan uint32 - // quit is a chan mouted to InvoiceRegistry that signals a shutdown. quit chan struct{} @@ -1471,11 +1434,6 @@ func (i *invoiceSubscriptionKit) Cancel() { return } - select { - case i.subscriptionCancels <- i.id: - case <-i.quit: - } - i.ntfnQueue.Stop() close(i.cancelChan) @@ -1506,10 +1464,9 @@ func (i *InvoiceRegistry) SubscribeNotifications( addIndex: addIndex, settleIndex: settleIndex, invoiceSubscriptionKit: invoiceSubscriptionKit{ - subscriptionCancels: i.subscriptionCancels, - quit: i.quit, - ntfnQueue: queue.NewConcurrentQueue(20), - cancelChan: make(chan struct{}), + quit: i.quit, + ntfnQueue: queue.NewConcurrentQueue(20), + cancelChan: make(chan struct{}), }, } client.ntfnQueue.Start() @@ -1525,6 +1482,7 @@ func (i *InvoiceRegistry) SubscribeNotifications( i.wg.Add(1) go func() { defer i.wg.Done() + defer i.deleteClient(client.id) for { select { @@ -1576,9 +1534,6 @@ func (i *InvoiceRegistry) SubscribeNotifications( } }() - i.Lock() - defer i.Unlock() - // Query the database to see if based on the provided addIndex and // settledIndex we need to deliver any backlog notifications. err := i.deliverBacklogEvents(client) @@ -1586,11 +1541,13 @@ func (i *InvoiceRegistry) SubscribeNotifications( return nil, err } - select { - case i.newSubscriptions <- client: - case <-i.quit: - return nil, ErrShuttingDown - } + log.Infof("New invoice subscription client: id=%v", client.id) + + i.Lock() + // With the backlog notifications delivered (if any), we'll add this to + // our active subscriptions. + i.notificationClients[client.id] = client + i.Unlock() return client, nil } @@ -1603,10 +1560,9 @@ func (i *InvoiceRegistry) SubscribeSingleInvoice( client := &SingleInvoiceSubscription{ Updates: make(chan *channeldb.Invoice), invoiceSubscriptionKit: invoiceSubscriptionKit{ - subscriptionCancels: i.subscriptionCancels, - quit: i.quit, - ntfnQueue: queue.NewConcurrentQueue(20), - cancelChan: make(chan struct{}), + quit: i.quit, + ntfnQueue: queue.NewConcurrentQueue(20), + cancelChan: make(chan struct{}), }, invoiceRef: channeldb.InvoiceRefByHash(hash), } @@ -1623,6 +1579,7 @@ func (i *InvoiceRegistry) SubscribeSingleInvoice( i.wg.Add(1) go func() { defer i.wg.Done() + defer i.deleteClient(client.id) for { select { @@ -1651,22 +1608,17 @@ func (i *InvoiceRegistry) SubscribeSingleInvoice( } }() - // Within the lock, we both query the invoice state and pass the client - // subscription to the invoiceEvents channel. This is to make sure that - // the client receives a consistent stream of events. - i.Lock() - defer i.Unlock() - err := i.deliverSingleBacklogEvents(client) if err != nil { return nil, err } - select { - case i.invoiceEvents <- client: - case <-i.quit: - return nil, ErrShuttingDown - } + log.Infof("New single invoice subscription client: id=%v, ref=%v", + client.id, client.invoiceRef) + + i.Lock() + i.singleNotificationClients[client.id] = client + i.Unlock() return client, nil } @@ -1731,3 +1683,40 @@ func (i *InvoiceRegistry) HodlUnsubscribeAll(subscriber chan<- interface{}) { delete(i.hodlReverseSubscriptions, subscriber) } + +// copySingleClients copies i.SingleInvoiceSubscription inside a lock. This is +// useful when we need to iterate the map to send notifications. +func (i *InvoiceRegistry) copySingleClients() map[uint32]*SingleInvoiceSubscription { + i.RLock() + defer i.RUnlock() + + clients := make(map[uint32]*SingleInvoiceSubscription) + for k, v := range i.singleNotificationClients { + clients[k] = v + } + return clients +} + +// copyClients copies i.notificationClients inside a lock. This is useful when +// we need to iterate the map to send notifications. +func (i *InvoiceRegistry) copyClients() map[uint32]*InvoiceSubscription { + i.RLock() + defer i.RUnlock() + + clients := make(map[uint32]*InvoiceSubscription) + for k, v := range i.notificationClients { + clients[k] = v + } + return clients +} + +// deleteClient removes a client by its ID inside a lock. Noop if the client is +// not found. +func (i *InvoiceRegistry) deleteClient(clientID uint32) { + i.Lock() + defer i.Unlock() + + log.Infof("Cancelling invoice subscription for client=%v", clientID) + delete(i.notificationClients, clientID) + delete(i.singleNotificationClients, clientID) +} diff --git a/lnrpc/invoicesrpc/invoices_server.go b/lnrpc/invoicesrpc/invoices_server.go index 102abb0c9..8081e011b 100644 --- a/lnrpc/invoicesrpc/invoices_server.go +++ b/lnrpc/invoicesrpc/invoices_server.go @@ -244,6 +244,8 @@ func (s *Server) SubscribeSingleInvoice(req *SubscribeSingleInvoiceRequest, } defer invoiceClient.Cancel() + log.Debugf("Created new single invoice(pay_hash=%v) subscription", hash) + for { select { case newInvoice := <-invoiceClient.Updates: @@ -265,7 +267,9 @@ func (s *Server) SubscribeSingleInvoice(req *SubscribeSingleInvoiceRequest, } case <-updateStream.Context().Done(): - return updateStream.Context().Err() + return fmt.Errorf("subscription for "+ + "invoice(pay_hash=%v): %w", hash, + updateStream.Context().Err()) case <-s.quit: return nil From 48827bb30ffc459b24c3b35002d2194e42234979 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Sat, 4 Dec 2021 16:17:54 +0800 Subject: [PATCH 4/4] docs: add release note for invoice registry fix --- docs/release-notes/release-notes-0.15.0.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/docs/release-notes/release-notes-0.15.0.md b/docs/release-notes/release-notes-0.15.0.md index 5052fe1f5..9cbe3f5d0 100644 --- a/docs/release-notes/release-notes-0.15.0.md +++ b/docs/release-notes/release-notes-0.15.0.md @@ -10,6 +10,14 @@ * Add [auto-generated command-line completions](https://github.com/lightningnetwork/lnd/pull/4177) for Fish shell. +## Bug Fixes + +* [Fixed an inactive invoice subscription not removed from invoice + registry](https://github.com/lightningnetwork/lnd/pull/6053). When an invoice + subscription is created and canceled immediately, it could be left uncleaned + due to the cancel signal is processed before the creation. It is now properly + handled by moving creation before deletion. + ## Misc * [An example systemd service file](https://github.com/lightningnetwork/lnd/pull/6033) @@ -37,3 +45,4 @@ * Daniel McNally * ErikEk * Torkel Rogstad +* Yong Yu