lnd.xprv/invoiceregistry.go

517 lines
15 KiB
Go
Raw Normal View History

package main
import (
"bytes"
"crypto/sha256"
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcutil"
"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"
)
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, 100),
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 clientID, client := range i.notificationClients {
// Before we dispatch this event, we'll check
// to ensure that this client hasn't already
// received this notification in order to
// ensure we don't duplicate any events.
invoice := event.invoice
switch {
// If we've already sent this settle event to
// the client, then we can skip this.
case event.isSettle &&
client.settleIndex >= invoice.SettleIndex:
continue
// Similarly, if we've already sent this add to
// the client then we can skip this one.
case !event.isSettle &&
client.addIndex >= invoice.AddIndex:
continue
// These two states should never happen, but we
// log them just in case so we can detect this
// instance.
case !event.isSettle &&
client.addIndex+1 != invoice.AddIndex:
ltndLog.Warnf("client=%v for invoice "+
"notifications missed an update, "+
"add_index=%v, new add event index=%v",
clientID, client.addIndex,
invoice.AddIndex)
case event.isSettle &&
client.settleIndex+1 != invoice.SettleIndex:
ltndLog.Warnf("client=%v for invoice "+
"notifications missed an update, "+
"settle_index=%v, new settle event index=%v",
clientID, client.settleIndex,
invoice.SettleIndex)
}
select {
case client.ntfnQueue.ChanIn() <- &invoiceEvent{
isSettle: event.isSettle,
invoice: invoice,
}:
case <-i.quit:
return
}
// Each time we send a notification to a
// client, we'll record the latest add/settle
// index it has. We'll use this to ensure we
// don't send a notification twice, which can
// happen if a new event is added while we're
// catching up a new client.
if event.isSettle {
client.settleIndex = invoice.SettleIndex
} else {
client.addIndex = invoice.AddIndex
}
}
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 {
// We re-bind the loop variable to ensure we don't hold onto
// the loop reference causing is to point to the same item.
addEvent := addEvent
select {
case client.ntfnQueue.ChanIn() <- &invoiceEvent{
isSettle: false,
invoice: &addEvent,
}:
case <-i.quit:
return fmt.Errorf("registry shutting down")
}
}
for _, settleEvent := range settleEvents {
// We re-bind the loop variable to ensure we don't hold onto
// the loop reference causing is to point to the same item.
settleEvent := settleEvent
select {
case client.ntfnQueue.ChanIn() <- &invoiceEvent{
isSettle: true,
invoice: &settleEvent,
}:
case <-i.quit:
return fmt.Errorf("registry shutting down")
}
}
return nil
}
2018-07-03 06:20:17 +03:00
// AddDebugInvoice adds a debug invoice for the specified amount, identified
// by the passed preimage. Once this invoice is added, subsystems within the
2018-07-03 06:20:17 +03:00
// 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. We also return the
// addIndex of the newly created invoice which monotonically increases for each
// new invoice added.
func (i *invoiceRegistry) AddInvoice(invoice *channeldb.Invoice) (uint64, error) {
i.Lock()
defer i.Unlock()
ltndLog.Debugf("Adding invoice %v", newLogClosure(func() string {
return spew.Sdump(invoice)
}))
addIndex, err := i.cdb.AddInvoice(invoice)
if err != nil {
return 0, err
}
// Now that we've added the invoice, we'll send dispatch a message to
// notify the clients of this new invoice.
i.notifyClients(invoice, false)
return addIndex, 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 {
i.Lock()
defer i.Unlock()
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.
if _, ok := i.debugInvoices[rHash]; ok {
// Debug invoices are never fully settled, so we simply return
// immediately in this case.
return nil
}
// If this isn't a debug invoice, then we'll attempt to settle an
// invoice matching this rHash on disk (if one exists).
invoice, err := i.cdb.SettleInvoice(rHash, amtPaid)
if err != nil {
return err
}
ltndLog.Infof("Payment received: %v", spew.Sdump(invoice))
i.notifyClients(invoice, true)
return nil
}
2016-12-25 03:51:25 +03:00
// 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
}