akovalenko/lnd.xprv
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

invoices/invoiceregistry: properly synchronize backlog

Browse Source 
This commit moves the db calls for retrieving add and settle backlogs
outide of the main event loop. All other db operations are performed
outside of the event loop and synchronized via the invoice registry's
mutex, which also synchronizes the order in which events submitted to be
processed.

This resolves various concurrency issues where notifications can be
missed of inconsistent reads against the databse. This is especially
important in this case because we are actually making two separate
database calls.
This commit is contained in:
Conner Fromknecht 2020-05-19 20:31:52 -07:00
parent 5b747715fc
commit 6a02fa1107
No known key found for this signature in database
GPG Key ID: E7D737B67FA592C7
1 changed files with 10 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
invoices/invoiceregistry.go
View File

@ -234,15 +234,6 @@ func (i *InvoiceRegistry) invoiceEventLoop() {
// We'll query for any backlog notifications, then add it to the
// set of clients.
case newClient := <-i.newSubscriptions:
// Before we add the client to our set of active
// clients, we'll first attempt to deliver any backlog
// invoice events.
err := i.deliverBacklogEvents(newClient)
if err != nil {
log.Errorf("unable to deliver backlog invoice "+
"notifications: %v", err)
}
log.Infof("New invoice subscription "+
"client: id=%v", newClient.id)
@ -410,9 +401,6 @@ func (i *InvoiceRegistry) dispatchToClients(event *invoiceEvent) {
// deliverBacklogEvents will attempts to query the invoice database for any
// notifications that the client has missed since it reconnected last.
func (i *InvoiceRegistry) deliverBacklogEvents(client *InvoiceSubscription) error {
// First, we'll query the database to see if based on the provided
// addIndex and settledIndex we need to deliver any backlog
// notifications.
addEvents, err := i.cdb.InvoicesAddedSince(client.addIndex)
if err != nil {
return err
@ -1253,6 +1241,16 @@ 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)
if err != nil {
return nil, err
}
select {
case i.newSubscriptions <- client:
case <-i.quit: