lnd.xprv/invoiceregistry.go
Olaoluwa Osuntokun 7917fd2ab0
invoiceregistry: re-work logic to support delivering notification backlog
In this commit, we re-work the existing invoiceRegistry struct to
support delivering backlog notifications to subscription clients if
needed. Rather than using 1 goroutine per-client per-event, each client
now gains a concurrent notification queue. This queue will then be used
to ensure in-order delivery of all notifications from the
invoiceEventNotifier.

The SubscribeNotifications method now takes two params: addIndex, and
settleIndex. These should be the values of the last add index and settle
index the caller knows of. If specified (not zero), then we'll look up
all the notifications that the caller has missed, and then deliver those
before sending out any new notifications. In order to do this without
losing ordering of events, we've added a new central goroutine which
will ensure that all events are properly serialized.
2018-07-06 12:21:59 -07:00

464 lines
14 KiB
Go

package main
import (
"bytes"
"crypto/sha256"
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/davecgh/go-spew/spew"
"github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/channeldb"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/zpay32"
"github.com/roasbeef/btcd/chaincfg/chainhash"
"github.com/roasbeef/btcutil"
)
var (
// debugPre is the default debug preimage which is inserted into the
// invoice registry if the --debughtlc flag is activated on start up.
// All nodes initialized with the flag active will immediately settle
// any incoming HTLC whose rHash corresponds with the debug
// preimage.
debugPre, _ = chainhash.NewHash(bytes.Repeat([]byte{1}, 32))
debugHash = chainhash.Hash(sha256.Sum256(debugPre[:]))
)
// invoiceRegistry is a central registry of all the outstanding invoices
// created by the daemon. The registry is a thin wrapper around a map in order
// to ensure that all updates/reads are thread safe.
type invoiceRegistry struct {
sync.RWMutex
cdb *channeldb.DB
clientMtx sync.Mutex
nextClientID uint32
notificationClients map[uint32]*invoiceSubscription
newSubscriptions chan *invoiceSubscription
subscriptionCancels chan uint32
invoiceEvents chan *invoiceEvent
// debugInvoices is a map which stores special "debug" invoices which
// should be only created/used when manual tests require an invoice
// that *all* nodes are able to fully settle.
debugInvoices map[chainhash.Hash]*channeldb.Invoice
wg sync.WaitGroup
quit chan struct{}
}
// newInvoiceRegistry creates a new invoice registry. The invoice registry
// wraps the persistent on-disk invoice storage with an additional in-memory
// layer. The in-memory layer is in place such that debug invoices can be added
// which are volatile yet available system wide within the daemon.
func newInvoiceRegistry(cdb *channeldb.DB) *invoiceRegistry {
return &invoiceRegistry{
cdb: cdb,
debugInvoices: make(map[chainhash.Hash]*channeldb.Invoice),
notificationClients: make(map[uint32]*invoiceSubscription),
newSubscriptions: make(chan *invoiceSubscription),
subscriptionCancels: make(chan uint32),
invoiceEvents: make(chan *invoiceEvent),
quit: make(chan struct{}),
}
}
// Start starts the registry and all goroutines it needs to carry out its task.
func (i *invoiceRegistry) Start() error {
i.wg.Add(1)
go i.invoiceEventNotifier()
return nil
}
// Stop signals the registry for a graceful shutdown.
func (i *invoiceRegistry) Stop() {
close(i.quit)
i.wg.Wait()
}
// invoiceEvent represents a new event that has modified on invoice on disk.
// Only two event types are currently supported: newly created invoices, and
// instance where invoices are settled.
type invoiceEvent struct {
isSettle bool
invoice *channeldb.Invoice
}
// invoiceEventNotifier is the dedicated goroutine responsible for accepting
// new notification subscriptions, cancelling old subscriptions, and
// dispatching new invoice events.
func (i *invoiceRegistry) invoiceEventNotifier() {
defer i.wg.Done()
for {
select {
// A new invoice subscription has just arrived! 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 {
ltndLog.Errorf("unable to deliver backlog invoice "+
"notifications: %v", err)
}
ltndLog.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:
ltndLog.Infof("Cancelling invoice subscription for "+
"client=%v", clientID)
delete(i.notificationClients, clientID)
// A sub-systems has just modified the invoice state, so we'll
// dispatch notifications to all registered clients.
case event := <-i.invoiceEvents:
for _, client := range i.notificationClients {
select {
case client.ntfnQueue.ChanIn() <- &invoiceEvent{
isSettle: event.isSettle,
invoice: event.invoice,
}:
case <-i.quit:
return
}
}
case <-i.quit:
return
}
}
}
// 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
}
settleEvents, err := i.cdb.InvoicesSettledSince(client.settleIndex)
if err != nil {
return err
}
// If we have any to deliver, then we'll append them to the end of the
// notification queue in order to catch up the client before delivering
// any new notifications.
for _, addEvent := range addEvents {
select {
case client.ntfnQueue.ChanIn() <- &invoiceEvent{
isSettle: false,
invoice: &addEvent,
}:
case <-i.quit:
return fmt.Errorf("registry shutting down")
}
}
for _, settleEvent := range settleEvents {
select {
case client.ntfnQueue.ChanIn() <- &invoiceEvent{
isSettle: true,
invoice: &settleEvent,
}:
case <-i.quit:
return fmt.Errorf("registry shutting down")
}
}
return nil
}
// AddDebugInvoice adds a debug invoice for the specified amount, identified
// by the passed preimage. Once this invoice is added, subsystems within the
// daemon add/forward HTLCs that are able to obtain the proper preimage
// required for redemption in the case that we're the final destination.
func (i *invoiceRegistry) AddDebugInvoice(amt btcutil.Amount, preimage chainhash.Hash) {
paymentHash := chainhash.Hash(sha256.Sum256(preimage[:]))
invoice := &channeldb.Invoice{
CreationDate: time.Now(),
Terms: channeldb.ContractTerm{
Value: lnwire.NewMSatFromSatoshis(amt),
PaymentPreimage: preimage,
},
}
i.Lock()
i.debugInvoices[paymentHash] = invoice
i.Unlock()
ltndLog.Debugf("Adding debug invoice %v", newLogClosure(func() string {
return spew.Sdump(invoice)
}))
}
// AddInvoice adds a regular invoice for the specified amount, identified by
// the passed preimage. Additionally, any memo or receipt data provided will
// also be stored on-disk. Once this invoice is added, subsystems within the
// daemon add/forward HTLCs are able to obtain the proper preimage required for
// redemption in the case that we're the final destination.
func (i *invoiceRegistry) AddInvoice(invoice *channeldb.Invoice) error {
ltndLog.Debugf("Adding invoice %v", newLogClosure(func() string {
return spew.Sdump(invoice)
}))
if err := i.cdb.AddInvoice(invoice); err != nil {
return err
}
// We'll launch a new goroutine to notify all of our active listeners
// that a new invoice was just added.
i.notifyClients(invoice, false)
return nil
}
// LookupInvoice looks up an invoice by its payment hash (R-Hash), if found
// then we're able to pull the funds pending within an HTLC. We'll also return
// what the expected min final CLTV delta is, pre-parsed from the payment
// request. This may be used by callers to determine if an HTLC is well formed
// according to the cltv delta.
//
// TODO(roasbeef): ignore if settled?
func (i *invoiceRegistry) LookupInvoice(rHash chainhash.Hash) (channeldb.Invoice, uint32, error) {
// First check the in-memory debug invoice index to see if this is an
// existing invoice added for debugging.
i.RLock()
debugInv, ok := i.debugInvoices[rHash]
i.RUnlock()
// If found, then simply return the invoice directly.
if ok {
return *debugInv, 0, nil
}
// Otherwise, we'll check the database to see if there's an existing
// matching invoice.
invoice, err := i.cdb.LookupInvoice(rHash)
if err != nil {
return channeldb.Invoice{}, 0, err
}
payReq, err := zpay32.Decode(
string(invoice.PaymentRequest), activeNetParams.Params,
)
if err != nil {
return channeldb.Invoice{}, 0, err
}
return *invoice, uint32(payReq.MinFinalCLTVExpiry()), nil
}
// SettleInvoice attempts to mark an invoice as settled. If the invoice is a
// debug invoice, then this method is a noop as debug invoices are never fully
// settled.
func (i *invoiceRegistry) SettleInvoice(rHash chainhash.Hash,
amtPaid lnwire.MilliSatoshi) error {
ltndLog.Debugf("Settling invoice %x", rHash[:])
// First check the in-memory debug invoice index to see if this is an
// existing invoice added for debugging.
i.RLock()
if _, ok := i.debugInvoices[rHash]; ok {
// Debug invoices are never fully settled, so we simply return
// immediately in this case.
i.RUnlock()
return nil
}
i.RUnlock()
// If this isn't a debug invoice, then we'll attempt to settle an
// invoice matching this rHash on disk (if one exists).
if err := i.cdb.SettleInvoice(rHash, amtPaid); err != nil {
return err
}
// Launch a new goroutine to notify any/all registered invoice
// notification clients.
go func() {
invoice, err := i.cdb.LookupInvoice(rHash)
if err != nil {
ltndLog.Errorf("unable to find invoice: %v", err)
return
}
ltndLog.Infof("Payment received: %v", spew.Sdump(invoice))
i.notifyClients(&invoice, true)
}()
return nil
}
// notifyClients notifies all currently registered invoice notification clients
// of a newly added/settled invoice.
func (i *invoiceRegistry) notifyClients(invoice *channeldb.Invoice, settle bool) {
event := &invoiceEvent{
isSettle: settle,
invoice: invoice,
}
select {
case i.invoiceEvents <- event:
case <-i.quit:
}
}
// invoiceSubscription represents an intent to receive updates for newly added
// or settled invoices. For each newly added invoice, a copy of the invoice
// will be sent over the NewInvoices channel. Similarly, for each newly settled
// invoice, a copy of the invoice will be sent over the SettledInvoices
// channel.
type invoiceSubscription struct {
cancelled uint32 // To be used atomically.
// NewInvoices is a channel that we'll use to send all newly created
// invoices with an invoice index greater than the specified
// StartingInvoiceIndex field.
NewInvoices chan *channeldb.Invoice
// SettledInvoices is a channel that we'll use to send all setted
// invoices with an invoices index greater than the specified
// StartingInvoiceIndex field.
SettledInvoices chan *channeldb.Invoice
// addIndex is the highest add index the caller knows of. We'll use
// this information to send out an event backlog to the notifications
// subscriber. Any new add events with an index greater than this will
// be dispatched before any new notifications are sent out.
addIndex uint64
// settleIndex is the highest settle index the caller knows of. We'll
// use this information to send out an event backlog to the
// notifications subscriber. Any new settle events with an index
// greater than this will be dispatched before any new notifications
// are sent out.
settleIndex uint64
ntfnQueue *chainntnfs.ConcurrentQueue
id uint32
inv *invoiceRegistry
cancelChan chan struct{}
wg sync.WaitGroup
}
// Cancel unregisters the invoiceSubscription, freeing any previously allocated
// resources.
func (i *invoiceSubscription) Cancel() {
if !atomic.CompareAndSwapUint32(&i.cancelled, 0, 1) {
return
}
select {
case i.inv.subscriptionCancels <- i.id:
case <-i.inv.quit:
}
i.ntfnQueue.Stop()
close(i.cancelChan)
i.wg.Wait()
}
// SubscribeNotifications returns an invoiceSubscription which allows the
// caller to receive async notifications when any invoices are settled or
// added. The invoiceIndex parameter is a streaming "checkpoint". We'll start
// by first sending out all new events with an invoice index _greater_ than
// this value. Afterwards, we'll send out real-time notifications.
func (i *invoiceRegistry) SubscribeNotifications(addIndex, settleIndex uint64) *invoiceSubscription {
client := &invoiceSubscription{
NewInvoices: make(chan *channeldb.Invoice),
SettledInvoices: make(chan *channeldb.Invoice),
addIndex: addIndex,
settleIndex: settleIndex,
inv: i,
ntfnQueue: chainntnfs.NewConcurrentQueue(20),
cancelChan: make(chan struct{}),
}
client.ntfnQueue.Start()
i.clientMtx.Lock()
client.id = i.nextClientID
i.nextClientID++
i.clientMtx.Unlock()
// Before we register this new invoice subscription, we'll launch a new
// goroutine that will proxy all notifications appended to the end of
// the concurrent queue to the two client-side channels the caller will
// feed off of.
i.wg.Add(1)
go func() {
defer i.wg.Done()
for {
select {
// A new invoice event has been sent by the
// invoiceRegistry! We'll figure out if this is an add
// event or a settle event, then dispatch the event to
// the client.
case ntfn := <-client.ntfnQueue.ChanOut():
invoiceEvent := ntfn.(*invoiceEvent)
targetChan := client.NewInvoices
if invoiceEvent.isSettle {
targetChan = client.SettledInvoices
}
select {
case targetChan <- invoiceEvent.invoice:
case <-client.cancelChan:
return
case <-i.quit:
return
}
case <-client.cancelChan:
return
case <-i.quit:
return
}
}
}()
select {
case i.newSubscriptions <- client:
case <-i.quit:
}
return client
}