lnd.xprv/invoices/invoiceregistry.go

1611 lines
47 KiB
Go
Raw Normal View History

package invoices
import (
"errors"
2019-09-03 13:23:39 +03:00
"fmt"
"sync"
"sync/atomic"
2019-09-03 13:23:39 +03:00
"time"
"github.com/lightningnetwork/lnd/channeldb"
"github.com/lightningnetwork/lnd/clock"
"github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/queue"
"github.com/lightningnetwork/lnd/record"
)
var (
// ErrInvoiceExpiryTooSoon is returned when an invoice is attempted to be
// accepted or settled with not enough blocks remaining.
ErrInvoiceExpiryTooSoon = errors.New("invoice expiry too soon")
// ErrInvoiceAmountTooLow is returned when an invoice is attempted to be
// accepted or settled with an amount that is too low.
ErrInvoiceAmountTooLow = errors.New("paid amount less than invoice amount")
// ErrShuttingDown is returned when an operation failed because the
// invoice registry is shutting down.
ErrShuttingDown = errors.New("invoice registry shutting down")
)
2019-09-03 13:23:39 +03:00
const (
// DefaultHtlcHoldDuration defines the default for how long mpp htlcs
// are held while waiting for the other set members to arrive.
DefaultHtlcHoldDuration = 120 * time.Second
)
2019-12-04 20:47:53 +03:00
// RegistryConfig contains the configuration parameters for invoice registry.
type RegistryConfig struct {
// FinalCltvRejectDelta defines the number of blocks before the expiry
// of the htlc where we no longer settle it as an exit hop and instead
// cancel it back. Normally this value should be lower than the cltv
// expiry of any invoice we create and the code effectuating this should
// not be hit.
FinalCltvRejectDelta int32
2019-09-03 13:23:39 +03:00
// HtlcHoldDuration defines for how long mpp htlcs are held while
// waiting for the other set members to arrive.
HtlcHoldDuration time.Duration
// Clock holds the clock implementation that is used to provide
// Now() and TickAfter() and is useful to stub out the clock functions
// during testing.
Clock clock.Clock
// AcceptKeySend indicates whether we want to accept spontaneous key
// send payments.
AcceptKeySend bool
// GcCanceledInvoicesOnStartup if set, we'll attempt to garbage collect
// all canceled invoices upon start.
GcCanceledInvoicesOnStartup bool
// GcCanceledInvoicesOnTheFly if set, we'll garbage collect all newly
// canceled invoices on the fly.
GcCanceledInvoicesOnTheFly bool
// KeysendHoldTime indicates for how long we want to accept and hold
// spontaneous keysend payments.
KeysendHoldTime time.Duration
2019-09-03 13:23:39 +03:00
}
// htlcReleaseEvent describes an htlc auto-release event. It is used to release
// mpp htlcs for which the complete set didn't arrive in time.
type htlcReleaseEvent struct {
// invoiceRef identifiers the invoice this htlc belongs to.
invoiceRef channeldb.InvoiceRef
2019-09-03 13:23:39 +03:00
// key is the circuit key of the htlc to release.
key channeldb.CircuitKey
// releaseTime is the time at which to release the htlc.
releaseTime time.Time
}
// Less is used to order PriorityQueueItem's by their release time such that
// items with the older release time are at the top of the queue.
//
// NOTE: Part of the queue.PriorityQueueItem interface.
func (r *htlcReleaseEvent) Less(other queue.PriorityQueueItem) bool {
return r.releaseTime.Before(other.(*htlcReleaseEvent).releaseTime)
2019-12-04 20:47:53 +03:00
}
// 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
2019-12-04 20:47:53 +03:00
// cfg contains the registry's configuration parameters.
cfg *RegistryConfig
2018-12-20 20:54:11 +03:00
clientMtx sync.Mutex
nextClientID uint32
notificationClients map[uint32]*InvoiceSubscription
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{}
// subscriptions is a map from a circuit key to a list of subscribers.
// It is used for efficient notification of links.
hodlSubscriptions map[channeldb.CircuitKey]map[chan<- interface{}]struct{}
// reverseSubscriptions tracks circuit keys subscribed to per
// subscriber. This is used to unsubscribe from all hashes efficiently.
hodlReverseSubscriptions map[chan<- interface{}]map[channeldb.CircuitKey]struct{}
2019-09-03 13:23:39 +03:00
// htlcAutoReleaseChan contains the new htlcs that need to be
// auto-released.
htlcAutoReleaseChan chan *htlcReleaseEvent
expiryWatcher *InvoiceExpiryWatcher
wg sync.WaitGroup
quit chan struct{}
}
// NewRegistry 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 NewRegistry(cdb *channeldb.DB, expiryWatcher *InvoiceExpiryWatcher,
cfg *RegistryConfig) *InvoiceRegistry {
return &InvoiceRegistry{
2018-12-20 20:54:11 +03:00
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),
hodlSubscriptions: make(map[channeldb.CircuitKey]map[chan<- interface{}]struct{}),
hodlReverseSubscriptions: make(map[chan<- interface{}]map[channeldb.CircuitKey]struct{}),
2019-12-04 20:47:53 +03:00
cfg: cfg,
2019-09-03 13:23:39 +03:00
htlcAutoReleaseChan: make(chan *htlcReleaseEvent),
expiryWatcher: expiryWatcher,
2018-12-20 20:54:11 +03:00
quit: make(chan struct{}),
}
}
// scanInvoicesOnStart will scan all invoices on start and add active invoices
// to the invoice expirt watcher while also attempting to delete all canceled
// invoices.
func (i *InvoiceRegistry) scanInvoicesOnStart() error {
var (
pending []*invoiceExpiry
removable []channeldb.InvoiceDeleteRef
)
reset := func() {
// Zero out our results on start and if the scan is ever run
// more than once. This latter case can happen if the kvdb
// layer needs to retry the View transaction underneath (eg.
// using the etcd driver, where all transactions are allowed
// to retry for serializability).
pending = nil
removable = make([]channeldb.InvoiceDeleteRef, 0)
}
scanFunc := func(
paymentHash lntypes.Hash, invoice *channeldb.Invoice) error {
if invoice.IsPending() {
expiryRef := makeInvoiceExpiry(paymentHash, invoice)
if expiryRef != nil {
pending = append(pending, expiryRef)
}
} else if i.cfg.GcCanceledInvoicesOnStartup &&
invoice.State == channeldb.ContractCanceled {
// Consider invoice for removal if it is already
// canceled. Invoices that are expired but not yet
// canceled, will be queued up for cancellation after
// startup and will be deleted afterwards.
ref := channeldb.InvoiceDeleteRef{
PayHash: paymentHash,
AddIndex: invoice.AddIndex,
SettleIndex: invoice.SettleIndex,
}
if invoice.Terms.PaymentAddr != channeldb.BlankPayAddr {
ref.PayAddr = &invoice.Terms.PaymentAddr
}
removable = append(removable, ref)
}
return nil
}
err := i.cdb.ScanInvoices(scanFunc, reset)
if err != nil {
return err
}
log.Debugf("Adding %d pending invoices to the expiry watcher",
len(pending))
i.expiryWatcher.AddInvoices(pending...)
if err := i.cdb.DeleteInvoice(removable); err != nil {
log.Warnf("Deleting old invoices failed: %v", err)
}
return nil
}
// Start starts the registry and all goroutines it needs to carry out its task.
func (i *InvoiceRegistry) Start() error {
// Start InvoiceExpiryWatcher and prepopulate it with existing active
// invoices.
err := i.expiryWatcher.Start(i.cancelInvoiceImpl)
if err != nil {
return err
}
i.wg.Add(1)
2019-09-03 13:23:39 +03:00
go i.invoiceEventLoop()
// Now scan all pending and removable invoices to the expiry watcher or
// delete them.
err = i.scanInvoicesOnStart()
if err != nil {
i.Stop()
return err
}
return nil
}
// Stop signals the registry for a graceful shutdown.
func (i *InvoiceRegistry) Stop() {
i.expiryWatcher.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 {
2018-12-20 20:54:11 +03:00
hash lntypes.Hash
invoice *channeldb.Invoice
}
2019-09-03 13:23:39 +03:00
// tickAt returns a channel that ticks at the specified time. If the time has
// already passed, it will tick immediately.
func (i *InvoiceRegistry) tickAt(t time.Time) <-chan time.Time {
now := i.cfg.Clock.Now()
return i.cfg.Clock.TickAfter(t.Sub(now))
2019-09-03 13:23:39 +03:00
}
// invoiceEventLoop is the dedicated goroutine responsible for accepting
// new notification subscriptions, cancelling old subscriptions, and
// dispatching new invoice events.
2019-09-03 13:23:39 +03:00
func (i *InvoiceRegistry) invoiceEventLoop() {
defer i.wg.Done()
2019-09-03 13:23:39 +03:00
// Set up a heap for htlc auto-releases.
autoReleaseHeap := &queue.PriorityQueue{}
for {
2019-09-03 13:23:39 +03:00
// If there is something to release, set up a release tick
// channel.
var nextReleaseTick <-chan time.Time
if autoReleaseHeap.Len() > 0 {
head := autoReleaseHeap.Top().(*htlcReleaseEvent)
nextReleaseTick = i.tickAt(head.releaseTime)
}
select {
2018-12-20 20:54:11 +03:00
// 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)
2018-12-20 20:54:11 +03:00
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.
case event := <-i.invoiceEvents:
switch e := event.(type) {
// 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
}
2019-09-03 13:23:39 +03:00
// A new htlc came in for auto-release.
case event := <-i.htlcAutoReleaseChan:
log.Debugf("Scheduling auto-release for htlc: "+
"ref=%v, key=%v at %v",
event.invoiceRef, event.key, event.releaseTime)
2019-09-03 13:23:39 +03:00
// We use an independent timer for every htlc rather
// than a set timer that is reset with every htlc coming
// in. Otherwise the sender could keep resetting the
// timer until the broadcast window is entered and our
// channel is force closed.
autoReleaseHeap.Push(event)
// The htlc at the top of the heap needs to be auto-released.
case <-nextReleaseTick:
event := autoReleaseHeap.Pop().(*htlcReleaseEvent)
err := i.cancelSingleHtlc(
event.invoiceRef, event.key, ResultMppTimeout,
2019-09-03 13:23:39 +03:00
)
if err != nil {
log.Errorf("HTLC timer: %v", err)
}
case <-i.quit:
return
}
}
}
2018-12-20 20:54:11 +03:00
// 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 {
payHash := client.invoiceRef.PayHash()
if payHash == nil || *payHash != event.hash {
2018-12-20 20:54:11 +03:00
continue
}
2019-08-09 12:44:27 +03:00
client.notify(event)
2018-12-20 20:54:11 +03:00
}
}
// 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.
func (i *InvoiceRegistry) dispatchToClients(event *invoiceEvent) {
invoice := event.invoice
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.
// TODO(joostjager): Refactor switches.
state := event.invoice.State
switch {
// If we've already sent this settle event to
// the client, then we can skip this.
case state == channeldb.ContractSettled &&
client.settleIndex >= invoice.SettleIndex:
continue
// Similarly, if we've already sent this add to
// the client then we can skip this one.
case state == channeldb.ContractOpen &&
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 state == channeldb.ContractOpen &&
client.addIndex+1 != invoice.AddIndex:
log.Warnf("client=%v for invoice "+
"notifications missed an update, "+
"add_index=%v, new add event index=%v",
clientID, client.addIndex,
invoice.AddIndex)
case state == channeldb.ContractSettled &&
client.settleIndex+1 != invoice.SettleIndex:
log.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{
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.
switch event.invoice.State {
case channeldb.ContractSettled:
client.settleIndex = invoice.SettleIndex
case channeldb.ContractOpen:
client.addIndex = invoice.AddIndex
default:
log.Errorf("unexpected invoice state: %v",
event.invoice.State)
}
}
}
// 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 {
addEvents, err := i.cdb.InvoicesAddedSince(client.addIndex)
if err != nil {
return err
}
2018-12-20 20:54:11 +03:00
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{
invoice: &addEvent,
}:
case <-i.quit:
return ErrShuttingDown
}
}
2018-12-20 20:54:11 +03:00
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{
invoice: &settleEvent,
}:
case <-i.quit:
return ErrShuttingDown
}
}
return nil
}
2018-12-20 20:54:11 +03:00
// deliverSingleBacklogEvents will attempt to query the invoice database to
// retrieve the current invoice state and deliver this to the subscriber. Single
// invoice subscribers will always receive the current state right after
// subscribing. Only in case the invoice does not yet exist, nothing is sent
// yet.
func (i *InvoiceRegistry) deliverSingleBacklogEvents(
client *SingleInvoiceSubscription) error {
invoice, err := i.cdb.LookupInvoice(client.invoiceRef)
2018-12-20 20:54:11 +03:00
// It is possible that the invoice does not exist yet, but the client is
// already watching it in anticipation.
if err == channeldb.ErrInvoiceNotFound ||
err == channeldb.ErrNoInvoicesCreated {
2018-12-20 20:54:11 +03:00
return nil
}
if err != nil {
return err
}
payHash := client.invoiceRef.PayHash()
if payHash == nil {
return nil
}
2018-12-20 20:54:11 +03:00
err = client.notify(&invoiceEvent{
hash: *payHash,
2018-12-20 20:54:11 +03:00
invoice: &invoice,
})
if err != nil {
return err
}
return nil
}
// 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. A side effect of this function is that it also sets
// AddIndex on the invoice argument.
2018-12-20 20:54:11 +03:00
func (i *InvoiceRegistry) AddInvoice(invoice *channeldb.Invoice,
paymentHash lntypes.Hash) (uint64, error) {
i.Lock()
ref := channeldb.InvoiceRefByHash(paymentHash)
log.Debugf("Invoice%v: added with terms %v", ref, invoice.Terms)
2018-10-05 11:14:56 +03:00
addIndex, err := i.cdb.AddInvoice(invoice, paymentHash)
if err != nil {
i.Unlock()
return 0, err
}
// Now that we've added the invoice, we'll send dispatch a message to
// notify the clients of this new invoice.
2018-12-20 20:54:11 +03:00
i.notifyClients(paymentHash, invoice, channeldb.ContractOpen)
i.Unlock()
// InvoiceExpiryWatcher.AddInvoice must not be locked by InvoiceRegistry
// to avoid deadlock when a new invoice is added while an other is being
// canceled.
invoiceExpiryRef := makeInvoiceExpiry(paymentHash, invoice)
if invoiceExpiryRef != nil {
i.expiryWatcher.AddInvoices(invoiceExpiryRef)
}
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.
//
// TODO(roasbeef): ignore if settled?
func (i *InvoiceRegistry) LookupInvoice(rHash lntypes.Hash) (channeldb.Invoice,
error) {
// We'll check the database to see if there's an existing matching
// invoice.
ref := channeldb.InvoiceRefByHash(rHash)
return i.cdb.LookupInvoice(ref)
}
2019-09-03 13:23:39 +03:00
// startHtlcTimer starts a new timer via the invoice registry main loop that
// cancels a single htlc on an invoice when the htlc hold duration has passed.
func (i *InvoiceRegistry) startHtlcTimer(invoiceRef channeldb.InvoiceRef,
2019-09-03 13:23:39 +03:00
key channeldb.CircuitKey, acceptTime time.Time) error {
releaseTime := acceptTime.Add(i.cfg.HtlcHoldDuration)
event := &htlcReleaseEvent{
invoiceRef: invoiceRef,
2019-09-03 13:23:39 +03:00
key: key,
releaseTime: releaseTime,
}
select {
case i.htlcAutoReleaseChan <- event:
return nil
case <-i.quit:
return ErrShuttingDown
}
}
// cancelSingleHtlc cancels a single accepted htlc on an invoice. It takes
// a resolution result which will be used to notify subscribed links and
// resolvers of the details of the htlc cancellation.
func (i *InvoiceRegistry) cancelSingleHtlc(invoiceRef channeldb.InvoiceRef,
key channeldb.CircuitKey, result FailResolutionResult) error {
2019-09-03 13:23:39 +03:00
i.Lock()
defer i.Unlock()
updateInvoice := func(invoice *channeldb.Invoice) (
*channeldb.InvoiceUpdateDesc, error) {
// Only allow individual htlc cancelation on open invoices.
if invoice.State != channeldb.ContractOpen {
log.Debugf("cancelSingleHtlc: invoice %v no longer "+
"open", invoiceRef)
2019-09-03 13:23:39 +03:00
return nil, nil
}
// Lookup the current status of the htlc in the database.
htlc, ok := invoice.Htlcs[key]
if !ok {
return nil, fmt.Errorf("htlc %v not found", key)
}
// Cancelation is only possible if the htlc wasn't already
// resolved.
if htlc.State != channeldb.HtlcStateAccepted {
log.Debugf("cancelSingleHtlc: htlc %v on invoice %v "+
"is already resolved", key, invoiceRef)
2019-09-03 13:23:39 +03:00
return nil, nil
}
log.Debugf("cancelSingleHtlc: cancelling htlc %v on invoice %v",
key, invoiceRef)
2019-09-03 13:23:39 +03:00
// Return an update descriptor that cancels htlc and keeps
// invoice open.
canceledHtlcs := map[channeldb.CircuitKey]struct{}{
key: {},
}
return &channeldb.InvoiceUpdateDesc{
CancelHtlcs: canceledHtlcs,
}, nil
}
// Try to mark the specified htlc as canceled in the invoice database.
// Intercept the update descriptor to set the local updated variable. If
// no invoice update is performed, we can return early.
var updated bool
invoice, err := i.cdb.UpdateInvoice(invoiceRef,
2019-09-03 13:23:39 +03:00
func(invoice *channeldb.Invoice) (
*channeldb.InvoiceUpdateDesc, error) {
updateDesc, err := updateInvoice(invoice)
if err != nil {
return nil, err
}
updated = updateDesc != nil
return updateDesc, err
},
)
if err != nil {
return err
}
if !updated {
return nil
}
// The invoice has been updated. Notify subscribers of the htlc
// resolution.
htlc, ok := invoice.Htlcs[key]
if !ok {
return fmt.Errorf("htlc %v not found", key)
}
if htlc.State == channeldb.HtlcStateCanceled {
resolution := NewFailResolution(
key, int32(htlc.AcceptHeight), result,
)
i.notifyHodlSubscribers(resolution)
2019-09-03 13:23:39 +03:00
}
return nil
}
// processKeySend just-in-time inserts an invoice if this htlc is a keysend
// htlc.
2020-02-04 17:15:48 +03:00
func (i *InvoiceRegistry) processKeySend(ctx invoiceUpdateCtx) error {
// Retrieve keysend record if present.
preimageSlice, ok := ctx.customRecords[record.KeySendType]
if !ok {
return nil
}
// Cancel htlc is preimage is invalid.
preimage, err := lntypes.MakePreimage(preimageSlice)
if err != nil {
return err
}
if preimage.Hash() != ctx.hash {
return fmt.Errorf("invalid keysend preimage %v for hash %v",
preimage, ctx.hash)
}
// Only allow keysend for non-mpp payments.
if ctx.mpp != nil {
return errors.New("no mpp keysend supported")
}
// Create an invoice for the htlc amount.
amt := ctx.amtPaid
// Set tlv optional feature vector on the invoice. Otherwise we wouldn't
// be able to pay to it with keysend.
rawFeatures := lnwire.NewRawFeatureVector(
lnwire.TLVOnionPayloadOptional,
)
features := lnwire.NewFeatureVector(rawFeatures, lnwire.Features)
// Use the minimum block delta that we require for settling htlcs.
finalCltvDelta := i.cfg.FinalCltvRejectDelta
2020-01-10 23:35:10 +03:00
// Pre-check expiry here to prevent inserting an invoice that will not
// be settled.
if ctx.expiry < uint32(ctx.currentHeight+finalCltvDelta) {
return errors.New("final expiry too soon")
}
// The invoice database indexes all invoices by payment address, however
// legacy keysend payment do not have one. In order to avoid a new
// payment type on-disk wrt. to indexing, we'll continue to insert a
// blank payment address which is special cased in the insertion logic
// to not be indexed. In the future, once AMP is merged, this should be
// replaced by generating a random payment address on the behalf of the
// sender.
payAddr := channeldb.BlankPayAddr
// Create placeholder invoice.
invoice := &channeldb.Invoice{
CreationDate: i.cfg.Clock.Now(),
Terms: channeldb.ContractTerm{
FinalCltvDelta: finalCltvDelta,
Value: amt,
PaymentPreimage: &preimage,
PaymentAddr: payAddr,
Features: features,
},
}
if i.cfg.KeysendHoldTime != 0 {
invoice.HodlInvoice = true
invoice.Terms.Expiry = i.cfg.KeysendHoldTime
}
// Insert invoice into database. Ignore duplicates, because this
// may be a replay.
2020-02-04 17:15:48 +03:00
_, err = i.AddInvoice(invoice, ctx.hash)
if err != nil && err != channeldb.ErrDuplicateInvoice {
return err
}
return nil
}
2021-03-22 22:37:23 +03:00
// processAMP just-in-time inserts an invoice if this htlc is a keysend
// htlc.
func (i *InvoiceRegistry) processAMP(ctx invoiceUpdateCtx) error {
// AMP payments MUST also include an MPP record.
if ctx.mpp == nil {
return errors.New("no MPP record for AMP")
}
// Create an invoice for the total amount expected, provided in the MPP
// record.
amt := ctx.mpp.TotalMsat()
// Set the TLV and MPP optional features on the invoice. We'll also make
// the AMP features required so that it can't be paid by legacy or MPP
// htlcs.
rawFeatures := lnwire.NewRawFeatureVector(
lnwire.TLVOnionPayloadOptional,
lnwire.PaymentAddrOptional,
lnwire.AMPRequired,
)
features := lnwire.NewFeatureVector(rawFeatures, lnwire.Features)
// Use the minimum block delta that we require for settling htlcs.
finalCltvDelta := i.cfg.FinalCltvRejectDelta
// Pre-check expiry here to prevent inserting an invoice that will not
// be settled.
if ctx.expiry < uint32(ctx.currentHeight+finalCltvDelta) {
return errors.New("final expiry too soon")
}
// We'll use the sender-generated payment address provided in the HTLC
// to create our AMP invoice.
payAddr := ctx.mpp.PaymentAddr()
// Create placeholder invoice.
invoice := &channeldb.Invoice{
CreationDate: i.cfg.Clock.Now(),
Terms: channeldb.ContractTerm{
FinalCltvDelta: finalCltvDelta,
Value: amt,
PaymentPreimage: nil,
PaymentAddr: payAddr,
Features: features,
},
}
// Insert invoice into database. Ignore duplicates payment hashes and
// payment addrs, this may be a replay or a different HTLC for the AMP
// invoice.
_, err := i.AddInvoice(invoice, ctx.hash)
switch {
case err == channeldb.ErrDuplicateInvoice:
return nil
case err == channeldb.ErrDuplicatePayAddr:
return nil
default:
return err
}
}
// NotifyExitHopHtlc attempts to mark an invoice as settled. The return value
// describes how the htlc should be resolved.
//
// When the preimage of the invoice is not yet known (hodl invoice), this
// function moves the invoice to the accepted state. When SettleHoldInvoice is
// called later, a resolution message will be send back to the caller via the
// provided hodlChan. Invoice registry sends on this channel what action needs
// to be taken on the htlc (settle or cancel). The caller needs to ensure that
// the channel is either buffered or received on from another goroutine to
// prevent deadlock.
2019-09-03 13:23:39 +03:00
//
// In the case that the htlc is part of a larger set of htlcs that pay to the
// same invoice (multi-path payment), the htlc is held until the set is
// complete. If the set doesn't fully arrive in time, a timer will cancel the
// held htlc.
func (i *InvoiceRegistry) NotifyExitHopHtlc(rHash lntypes.Hash,
amtPaid lnwire.MilliSatoshi, expiry uint32, currentHeight int32,
circuitKey channeldb.CircuitKey, hodlChan chan<- interface{},
payload Payload) (HtlcResolution, error) {
// Create the update context containing the relevant details of the
// incoming htlc.
ctx := invoiceUpdateCtx{
2020-02-04 17:15:48 +03:00
hash: rHash,
circuitKey: circuitKey,
amtPaid: amtPaid,
expiry: expiry,
currentHeight: currentHeight,
2019-12-04 20:47:53 +03:00
finalCltvRejectDelta: i.cfg.FinalCltvRejectDelta,
customRecords: payload.CustomRecords(),
mpp: payload.MultiPath(),
amp: payload.AMPRecord(),
}
invoices: replay awareness Previously the invoice registry wasn't aware of replayed htlcs. This was dealt with by keeping the invoice accept/settle logic idempotent, so that a replay wouldn't have an effect. This mechanism has two limitations: 1. No accurate tracking of the total amount paid to an invoice. The total amount couldn't just be increased with every htlc received, because it could be a replay which would lead to counting the htlc amount multiple times. Therefore the total amount was set to the amount of the first htlc that was received, even though there may have been multiple htlcs paying to the invoice. 2. Impossible to check htlc expiry consistently for hodl invoices. When an htlc is new, its expiry needs to be checked against the invoice cltv delta. But for a replay, that check must be skipped. The htlc was accepted in time, the invoice was moved to the accepted state and a replay some blocks later shouldn't lead to that htlc being cancelled. Because the invoice registry couldn't recognize replays, it stopped checking htlc expiry heights when the invoice reached the accepted state. This prevents hold htlcs from being cancelled after a restart. But unfortunately this also caused additional htlcs to be accepted on an already accepted invoice without their expiry being checked. In this commit, the invoice registry starts to persistently track htlcs so that replays can be recognized. For replays, an htlc resolution action is returned early. This fixes both limitations mentioned above.
2019-08-09 16:09:57 +03:00
// Process keysend if present. Do this outside of the lock, because
// AddInvoice obtains its own lock. This is no problem, because the
// operation is idempotent.
if i.cfg.AcceptKeySend {
2021-03-22 22:37:23 +03:00
if ctx.amp != nil {
err := i.processAMP(ctx)
if err != nil {
ctx.log(fmt.Sprintf("amp error: %v", err))
2021-03-22 22:37:23 +03:00
return NewFailResolution(
circuitKey, currentHeight, ResultAmpError,
), nil
}
} else {
err := i.processKeySend(ctx)
if err != nil {
ctx.log(fmt.Sprintf("keysend error: %v", err))
return NewFailResolution(
circuitKey, currentHeight, ResultKeySendError,
), nil
}
}
}
2020-02-04 12:07:56 +03:00
// Execute locked notify exit hop logic.
i.Lock()
resolution, err := i.notifyExitHopHtlcLocked(&ctx, hodlChan)
2020-02-04 12:07:56 +03:00
i.Unlock()
if err != nil {
return nil, err
}
switch r := resolution.(type) {
// The htlc is held. Start a timer outside the lock if the htlc should
// be auto-released, because otherwise a deadlock may happen with the
// main event loop.
case *htlcAcceptResolution:
2020-02-04 12:07:56 +03:00
if r.autoRelease {
err := i.startHtlcTimer(
ctx.invoiceRef(), circuitKey, r.acceptTime,
)
2020-02-04 12:07:56 +03:00
if err != nil {
return nil, err
}
}
// We return a nil resolution because htlc acceptances are
// represented as nil resolutions externally.
// TODO(carla) update calling code to handle accept resolutions.
2020-02-04 12:07:56 +03:00
return nil, nil
// A direct resolution was received for this htlc.
case HtlcResolution:
return r, nil
// Fail if an unknown resolution type was received.
2020-02-04 12:07:56 +03:00
default:
return nil, errors.New("invalid resolution type")
}
}
// notifyExitHopHtlcLocked is the internal implementation of NotifyExitHopHtlc
// that should be executed inside the registry lock.
func (i *InvoiceRegistry) notifyExitHopHtlcLocked(
ctx *invoiceUpdateCtx, hodlChan chan<- interface{}) (
HtlcResolution, error) {
// We'll attempt to settle an invoice matching this rHash on disk (if
// one exists). The callback will update the invoice state and/or htlcs.
var (
resolution HtlcResolution
updateSubscribers bool
)
invoice, err := i.cdb.UpdateInvoice(
ctx.invoiceRef(),
func(inv *channeldb.Invoice) (
*channeldb.InvoiceUpdateDesc, error) {
invoices: replay awareness Previously the invoice registry wasn't aware of replayed htlcs. This was dealt with by keeping the invoice accept/settle logic idempotent, so that a replay wouldn't have an effect. This mechanism has two limitations: 1. No accurate tracking of the total amount paid to an invoice. The total amount couldn't just be increased with every htlc received, because it could be a replay which would lead to counting the htlc amount multiple times. Therefore the total amount was set to the amount of the first htlc that was received, even though there may have been multiple htlcs paying to the invoice. 2. Impossible to check htlc expiry consistently for hodl invoices. When an htlc is new, its expiry needs to be checked against the invoice cltv delta. But for a replay, that check must be skipped. The htlc was accepted in time, the invoice was moved to the accepted state and a replay some blocks later shouldn't lead to that htlc being cancelled. Because the invoice registry couldn't recognize replays, it stopped checking htlc expiry heights when the invoice reached the accepted state. This prevents hold htlcs from being cancelled after a restart. But unfortunately this also caused additional htlcs to be accepted on an already accepted invoice without their expiry being checked. In this commit, the invoice registry starts to persistently track htlcs so that replays can be recognized. For replays, an htlc resolution action is returned early. This fixes both limitations mentioned above.
2019-08-09 16:09:57 +03:00
2020-02-04 12:07:56 +03:00
updateDesc, res, err := updateInvoice(ctx, inv)
if err != nil {
return nil, err
invoices: replay awareness Previously the invoice registry wasn't aware of replayed htlcs. This was dealt with by keeping the invoice accept/settle logic idempotent, so that a replay wouldn't have an effect. This mechanism has two limitations: 1. No accurate tracking of the total amount paid to an invoice. The total amount couldn't just be increased with every htlc received, because it could be a replay which would lead to counting the htlc amount multiple times. Therefore the total amount was set to the amount of the first htlc that was received, even though there may have been multiple htlcs paying to the invoice. 2. Impossible to check htlc expiry consistently for hodl invoices. When an htlc is new, its expiry needs to be checked against the invoice cltv delta. But for a replay, that check must be skipped. The htlc was accepted in time, the invoice was moved to the accepted state and a replay some blocks later shouldn't lead to that htlc being cancelled. Because the invoice registry couldn't recognize replays, it stopped checking htlc expiry heights when the invoice reached the accepted state. This prevents hold htlcs from being cancelled after a restart. But unfortunately this also caused additional htlcs to be accepted on an already accepted invoice without their expiry being checked. In this commit, the invoice registry starts to persistently track htlcs so that replays can be recognized. For replays, an htlc resolution action is returned early. This fixes both limitations mentioned above.
2019-08-09 16:09:57 +03:00
}
// Only send an update if the invoice state was changed.
updateSubscribers = updateDesc != nil &&
updateDesc.State != nil
// Assign resolution to outer scope variable.
resolution = res
invoices: replay awareness Previously the invoice registry wasn't aware of replayed htlcs. This was dealt with by keeping the invoice accept/settle logic idempotent, so that a replay wouldn't have an effect. This mechanism has two limitations: 1. No accurate tracking of the total amount paid to an invoice. The total amount couldn't just be increased with every htlc received, because it could be a replay which would lead to counting the htlc amount multiple times. Therefore the total amount was set to the amount of the first htlc that was received, even though there may have been multiple htlcs paying to the invoice. 2. Impossible to check htlc expiry consistently for hodl invoices. When an htlc is new, its expiry needs to be checked against the invoice cltv delta. But for a replay, that check must be skipped. The htlc was accepted in time, the invoice was moved to the accepted state and a replay some blocks later shouldn't lead to that htlc being cancelled. Because the invoice registry couldn't recognize replays, it stopped checking htlc expiry heights when the invoice reached the accepted state. This prevents hold htlcs from being cancelled after a restart. But unfortunately this also caused additional htlcs to be accepted on an already accepted invoice without their expiry being checked. In this commit, the invoice registry starts to persistently track htlcs so that replays can be recognized. For replays, an htlc resolution action is returned early. This fixes both limitations mentioned above.
2019-08-09 16:09:57 +03:00
return updateDesc, nil
},
)
switch err {
case channeldb.ErrInvoiceNotFound:
// If the invoice was not found, return a failure resolution
// with an invoice not found result.
return NewFailResolution(
2020-02-04 12:07:56 +03:00
ctx.circuitKey, ctx.currentHeight,
ResultInvoiceNotFound,
), nil
case nil:
default:
2020-02-04 12:07:56 +03:00
ctx.log(err.Error())
return nil, err
}
switch res := resolution.(type) {
case *HtlcFailResolution:
// Inspect latest htlc state on the invoice. If it is found,
// we will update the accept height as it was recorded in the
// invoice database (which occurs in the case where the htlc
// reached the database in a previous call). If the htlc was
// not found on the invoice, it was immediately failed so we
// send the failure resolution as is, which has the current
// height set as the accept height.
invoiceHtlc, ok := invoice.Htlcs[ctx.circuitKey]
if ok {
res.AcceptHeight = int32(invoiceHtlc.AcceptHeight)
}
ctx.log(fmt.Sprintf("failure resolution result "+
"outcome: %v, at accept height: %v",
res.Outcome, res.AcceptHeight))
// Some failures apply to the entire HTLC set. Break here if
// this isn't one of them.
if !res.Outcome.IsSetFailure() {
break
}
// Also cancel any HTLCs in the HTLC set that are also in the
// canceled state with the same failure result.
setID := ctx.setID()
for key, htlc := range invoice.Htlcs {
if htlc.State != channeldb.HtlcStateCanceled {
continue
}
if !htlc.IsInHTLCSet(setID) {
continue
}
htlcFailResolution := NewFailResolution(
key, int32(htlc.AcceptHeight), res.Outcome,
)
i.notifyHodlSubscribers(htlcFailResolution)
}
// If the htlc was settled, we will settle any previously accepted
// htlcs and notify our peer to settle them.
case *HtlcSettleResolution:
ctx.log(fmt.Sprintf("settle resolution result "+
"outcome: %v, at accept height: %v",
res.Outcome, res.AcceptHeight))
// Also settle any previously accepted htlcs. If a htlc is
// marked as settled, we should follow now and settle the htlc
// with our peer.
2019-09-03 13:23:39 +03:00
for key, htlc := range invoice.Htlcs {
if htlc.State != channeldb.HtlcStateSettled {
continue
}
preimage := res.Preimage
if htlc.AMP != nil && htlc.AMP.Preimage != nil {
preimage = *htlc.AMP.Preimage
}
// Notify subscribers that the htlcs should be settled
// with our peer. Note that the outcome of the
// resolution is set based on the outcome of the single
// htlc that we just settled, so may not be accurate
// for all htlcs.
htlcSettleResolution := NewSettleResolution(
preimage, key,
int32(htlc.AcceptHeight), res.Outcome,
)
// Notify subscribers that the htlc should be settled
// with our peer.
i.notifyHodlSubscribers(htlcSettleResolution)
2019-09-03 13:23:39 +03:00
}
// If we accepted the htlc, subscribe to the hodl invoice and return
// an accept resolution with the htlc's accept time on it.
case *htlcAcceptResolution:
invoiceHtlc, ok := invoice.Htlcs[ctx.circuitKey]
if !ok {
return nil, fmt.Errorf("accepted htlc: %v not"+
" present on invoice: %x", ctx.circuitKey,
ctx.hash[:])
}
// Determine accepted height of this htlc. If the htlc reached
// the invoice database (possibly in a previous call to the
// invoice registry), we'll take the original accepted height
// as it was recorded in the database.
acceptHeight := int32(invoiceHtlc.AcceptHeight)
ctx.log(fmt.Sprintf("accept resolution result "+
"outcome: %v, at accept height: %v",
res.outcome, acceptHeight))
2020-02-04 12:07:56 +03:00
// Auto-release the htlc if the invoice is still open. It can
2019-09-03 13:23:39 +03:00
// only happen for mpp payments that there are htlcs in state
// Accepted while the invoice is Open.
if invoice.State == channeldb.ContractOpen {
res.acceptTime = invoiceHtlc.AcceptTime
res.autoRelease = true
2020-02-04 12:07:56 +03:00
2019-09-03 13:23:39 +03:00
}
2020-02-04 12:07:56 +03:00
i.hodlSubscribe(hodlChan, ctx.circuitKey)
default:
panic("unknown action")
}
// Now that the links have been notified of any state changes to their
// HTLCs, we'll go ahead and notify any clients wiaiting on the invoice
// state changes.
if updateSubscribers {
i.notifyClients(ctx.hash, invoice, invoice.State)
}
return resolution, nil
}
// SettleHodlInvoice sets the preimage of a hodl invoice.
func (i *InvoiceRegistry) SettleHodlInvoice(preimage lntypes.Preimage) error {
i.Lock()
defer i.Unlock()
updateInvoice := func(invoice *channeldb.Invoice) (
*channeldb.InvoiceUpdateDesc, error) {
switch invoice.State {
case channeldb.ContractOpen:
return nil, channeldb.ErrInvoiceStillOpen
case channeldb.ContractCanceled:
return nil, channeldb.ErrInvoiceAlreadyCanceled
case channeldb.ContractSettled:
return nil, channeldb.ErrInvoiceAlreadySettled
}
return &channeldb.InvoiceUpdateDesc{
State: &channeldb.InvoiceStateUpdateDesc{
NewState: channeldb.ContractSettled,
Preimage: &preimage,
},
}, nil
}
hash := preimage.Hash()
invoiceRef := channeldb.InvoiceRefByHash(hash)
invoice, err := i.cdb.UpdateInvoice(invoiceRef, updateInvoice)
if err != nil {
log.Errorf("SettleHodlInvoice with preimage %v: %v",
preimage, err)
return err
}
log.Debugf("Invoice%v: settled with preimage %v", invoiceRef,
invoice.Terms.PaymentPreimage)
// In the callback, we marked the invoice as settled. UpdateInvoice will
// have seen this and should have moved all htlcs that were accepted to
// the settled state. In the loop below, we go through all of these and
// notify links and resolvers that are waiting for resolution. Any htlcs
// that were already settled before, will be notified again. This isn't
// necessary but doesn't hurt either.
for key, htlc := range invoice.Htlcs {
if htlc.State != channeldb.HtlcStateSettled {
continue
}
resolution := NewSettleResolution(
preimage, key, int32(htlc.AcceptHeight), ResultSettled,
)
i.notifyHodlSubscribers(resolution)
}
i.notifyClients(hash, invoice, invoice.State)
return nil
}
// CancelInvoice attempts to cancel the invoice corresponding to the passed
// payment hash.
func (i *InvoiceRegistry) CancelInvoice(payHash lntypes.Hash) error {
return i.cancelInvoiceImpl(payHash, true)
}
// cancelInvoice attempts to cancel the invoice corresponding to the passed
// payment hash. Accepted invoices will only be canceled if explicitly
// requested to do so. It notifies subscribing links and resolvers that
// the associated htlcs were canceled if they change state.
func (i *InvoiceRegistry) cancelInvoiceImpl(payHash lntypes.Hash,
cancelAccepted bool) error {
i.Lock()
defer i.Unlock()
ref := channeldb.InvoiceRefByHash(payHash)
log.Debugf("Invoice%v: canceling invoice", ref)
updateInvoice := func(invoice *channeldb.Invoice) (
*channeldb.InvoiceUpdateDesc, error) {
// Only cancel the invoice in ContractAccepted state if explicitly
// requested to do so.
if invoice.State == channeldb.ContractAccepted && !cancelAccepted {
return nil, nil
}
// Move invoice to the canceled state. Rely on validation in
// channeldb to return an error if the invoice is already
// settled or canceled.
return &channeldb.InvoiceUpdateDesc{
State: &channeldb.InvoiceStateUpdateDesc{
NewState: channeldb.ContractCanceled,
},
}, nil
}
invoiceRef := channeldb.InvoiceRefByHash(payHash)
invoice, err := i.cdb.UpdateInvoice(invoiceRef, updateInvoice)
// Implement idempotency by returning success if the invoice was already
// canceled.
if err == channeldb.ErrInvoiceAlreadyCanceled {
log.Debugf("Invoice%v: already canceled", ref)
return nil
}
if err != nil {
return err
}
// Return without cancellation if the invoice state is ContractAccepted.
if invoice.State == channeldb.ContractAccepted {
log.Debugf("Invoice%v: remains accepted as cancel wasn't"+
"explicitly requested.", ref)
return nil
}
log.Debugf("Invoice%v: canceled", ref)
// In the callback, some htlcs may have been moved to the canceled
// state. We now go through all of these and notify links and resolvers
// that are waiting for resolution. Any htlcs that were already canceled
// before, will be notified again. This isn't necessary but doesn't hurt
// either.
for key, htlc := range invoice.Htlcs {
2019-10-03 18:22:43 +03:00
if htlc.State != channeldb.HtlcStateCanceled {
continue
}
i.notifyHodlSubscribers(
NewFailResolution(
key, int32(htlc.AcceptHeight), ResultCanceled,
),
)
}
i.notifyClients(payHash, invoice, channeldb.ContractCanceled)
// Attempt to also delete the invoice if requested through the registry
// config.
if i.cfg.GcCanceledInvoicesOnTheFly {
// Assemble the delete reference and attempt to delete through
// the invocice from the DB.
deleteRef := channeldb.InvoiceDeleteRef{
PayHash: payHash,
AddIndex: invoice.AddIndex,
SettleIndex: invoice.SettleIndex,
}
if invoice.Terms.PaymentAddr != channeldb.BlankPayAddr {
deleteRef.PayAddr = &invoice.Terms.PaymentAddr
}
err = i.cdb.DeleteInvoice(
[]channeldb.InvoiceDeleteRef{deleteRef},
)
// If by any chance deletion failed, then log it instead of
// returning the error, as the invoice itsels has already been
// canceled.
if err != nil {
log.Warnf("Invoice%v could not be deleted: %v",
ref, err)
}
}
return nil
}
2016-12-25 03:51:25 +03:00
// notifyClients notifies all currently registered invoice notification clients
// of a newly added/settled invoice.
2018-12-20 20:54:11 +03:00
func (i *InvoiceRegistry) notifyClients(hash lntypes.Hash,
invoice *channeldb.Invoice,
state channeldb.ContractState) {
event := &invoiceEvent{
invoice: invoice,
2018-12-20 20:54:11 +03:00
hash: hash,
}
select {
case i.invoiceEvents <- event:
case <-i.quit:
}
}
2018-12-20 20:54:11 +03:00
// invoiceSubscriptionKit defines that are common to both all invoice
// subscribers and single invoice subscribers.
type invoiceSubscriptionKit struct {
id uint32
inv *InvoiceRegistry
ntfnQueue *queue.ConcurrentQueue
2019-10-03 18:22:43 +03:00
canceled uint32 // To be used atomically.
2018-12-20 20:54:11 +03:00
cancelChan chan struct{}
wg sync.WaitGroup
}
// 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 {
2018-12-20 20:54:11 +03:00
invoiceSubscriptionKit
// 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
2018-12-20 20:54:11 +03:00
}
2018-12-20 20:54:11 +03:00
// SingleInvoiceSubscription represents an intent to receive updates for a
// specific invoice.
type SingleInvoiceSubscription struct {
invoiceSubscriptionKit
invoiceRef channeldb.InvoiceRef
2018-12-20 20:54:11 +03:00
// Updates is a channel that we'll use to send all invoice events for
// the invoice that is subscribed to.
Updates chan *channeldb.Invoice
}
// Cancel unregisters the InvoiceSubscription, freeing any previously allocated
// resources.
2018-12-20 20:54:11 +03:00
func (i *invoiceSubscriptionKit) Cancel() {
2019-10-03 18:22:43 +03:00
if !atomic.CompareAndSwapUint32(&i.canceled, 0, 1) {
return
}
select {
case i.inv.subscriptionCancels <- i.id:
case <-i.inv.quit:
}
i.ntfnQueue.Stop()
close(i.cancelChan)
i.wg.Wait()
}
2018-12-20 20:54:11 +03:00
func (i *invoiceSubscriptionKit) notify(event *invoiceEvent) error {
select {
case i.ntfnQueue.ChanIn() <- event:
case <-i.inv.quit:
return ErrShuttingDown
2018-12-20 20:54:11 +03:00
}
return nil
}
// 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, error) {
client := &InvoiceSubscription{
NewInvoices: make(chan *channeldb.Invoice),
SettledInvoices: make(chan *channeldb.Invoice),
addIndex: addIndex,
settleIndex: settleIndex,
2018-12-20 20:54:11 +03:00
invoiceSubscriptionKit: invoiceSubscriptionKit{
inv: i,
ntfnQueue: queue.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)
var targetChan chan *channeldb.Invoice
state := invoiceEvent.invoice.State
switch state {
case channeldb.ContractOpen:
targetChan = client.NewInvoices
case channeldb.ContractSettled:
targetChan = client.SettledInvoices
default:
log.Errorf("unknown invoice "+
"state: %v", state)
continue
}
select {
case targetChan <- invoiceEvent.invoice:
case <-client.cancelChan:
return
case <-i.quit:
return
}
case <-client.cancelChan:
return
case <-i.quit:
return
}
}
}()
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:
return nil, ErrShuttingDown
}
return client, nil
}
2018-12-20 20:54:11 +03:00
// SubscribeSingleInvoice returns an SingleInvoiceSubscription which allows the
// caller to receive async notifications for a specific invoice.
func (i *InvoiceRegistry) SubscribeSingleInvoice(
hash lntypes.Hash) (*SingleInvoiceSubscription, error) {
2018-12-20 20:54:11 +03:00
client := &SingleInvoiceSubscription{
Updates: make(chan *channeldb.Invoice),
invoiceSubscriptionKit: invoiceSubscriptionKit{
inv: i,
ntfnQueue: queue.NewConcurrentQueue(20),
cancelChan: make(chan struct{}),
},
invoiceRef: channeldb.InvoiceRefByHash(hash),
2018-12-20 20:54:11 +03:00
}
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 will dispatch the event to the
// client.
case ntfn := <-client.ntfnQueue.ChanOut():
invoiceEvent := ntfn.(*invoiceEvent)
select {
case client.Updates <- invoiceEvent.invoice:
case <-client.cancelChan:
return
case <-i.quit:
return
}
case <-client.cancelChan:
return
case <-i.quit:
return
}
}
}()
// 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
}
2018-12-20 20:54:11 +03:00
select {
case i.invoiceEvents <- client:
2018-12-20 20:54:11 +03:00
case <-i.quit:
return nil, ErrShuttingDown
2018-12-20 20:54:11 +03:00
}
return client, nil
2018-12-20 20:54:11 +03:00
}
// notifyHodlSubscribers sends out the htlc resolution to all current
// subscribers.
func (i *InvoiceRegistry) notifyHodlSubscribers(htlcResolution HtlcResolution) {
subscribers, ok := i.hodlSubscriptions[htlcResolution.CircuitKey()]
if !ok {
return
}
// Notify all interested subscribers and remove subscription from both
// maps. The subscription can be removed as there only ever will be a
// single resolution for each hash.
for subscriber := range subscribers {
select {
case subscriber <- htlcResolution:
case <-i.quit:
return
}
delete(
i.hodlReverseSubscriptions[subscriber],
htlcResolution.CircuitKey(),
)
}
delete(i.hodlSubscriptions, htlcResolution.CircuitKey())
}
// hodlSubscribe adds a new invoice subscription.
func (i *InvoiceRegistry) hodlSubscribe(subscriber chan<- interface{},
circuitKey channeldb.CircuitKey) {
log.Debugf("Hodl subscribe for %v", circuitKey)
subscriptions, ok := i.hodlSubscriptions[circuitKey]
if !ok {
subscriptions = make(map[chan<- interface{}]struct{})
i.hodlSubscriptions[circuitKey] = subscriptions
}
subscriptions[subscriber] = struct{}{}
reverseSubscriptions, ok := i.hodlReverseSubscriptions[subscriber]
if !ok {
reverseSubscriptions = make(map[channeldb.CircuitKey]struct{})
i.hodlReverseSubscriptions[subscriber] = reverseSubscriptions
}
reverseSubscriptions[circuitKey] = struct{}{}
}
// HodlUnsubscribeAll cancels the subscription.
func (i *InvoiceRegistry) HodlUnsubscribeAll(subscriber chan<- interface{}) {
i.Lock()
defer i.Unlock()
hashes := i.hodlReverseSubscriptions[subscriber]
for hash := range hashes {
delete(i.hodlSubscriptions[hash], subscriber)
}
delete(i.hodlReverseSubscriptions, subscriber)
}