htlcswitch: hodl invoice

This commit modifies the invoice registry to handle invoices for which
the preimage is not known yet (hodl invoices). In that case, the
resolution channel passed in from links and resolvers is stored until we
either learn the preimage or want to cancel the htlc.

invoices/invoiceregistry.go

@@ -62,6 +62,14 @@ type InvoiceRegistry struct {
 	// value from the payment request.
 	decodeFinalCltvExpiry func(invoice string) (uint32, error)
 
+	// subscriptions is a map from a payment hash to a list of subscribers.
+	// It is used for efficient notification of links.
+	hodlSubscriptions map[lntypes.Hash]map[chan<- interface{}]struct{}
+
+	// reverseSubscriptions tracks hashes subscribed to per subscriber. This
+	// is used to unsubscribe from all hashes efficiently.
+	hodlReverseSubscriptions map[chan<- interface{}]map[lntypes.Hash]struct{}
+
 	wg   sync.WaitGroup
 	quit chan struct{}
 }
@@ -82,6 +90,8 @@ func NewRegistry(cdb *channeldb.DB, decodeFinalCltvExpiry func(invoice string) (
 		newSingleSubscriptions:    make(chan *SingleInvoiceSubscription),
 		subscriptionCancels:       make(chan uint32),
 		invoiceEvents:             make(chan *invoiceEvent, 100),
+		hodlSubscriptions:         make(map[lntypes.Hash]map[chan<- interface{}]struct{}),
+		hodlReverseSubscriptions:  make(map[chan<- interface{}]map[lntypes.Hash]struct{}),
 		decodeFinalCltvExpiry:     decodeFinalCltvExpiry,
 		quit:                      make(chan struct{}),
 	}
@@ -171,8 +181,10 @@ func (i *InvoiceRegistry) invoiceEventNotifier() {
 		// dispatch notifications to all registered clients.
 		case event := <-i.invoiceEvents:
 			// For backwards compatibility, do not notify all
-			// invoice subscribers of cancel events.
-			if event.state != channeldb.ContractCanceled {
+			// invoice subscribers of cancel and accept events.
+			if event.state != channeldb.ContractCanceled &&
+				event.state != channeldb.ContractAccepted {
+
 				i.dispatchToClients(event)
 			}
 			i.dispatchToSingleClients(event)
@@ -449,8 +461,17 @@ func (i *InvoiceRegistry) LookupInvoice(rHash lntypes.Hash) (channeldb.Invoice,
 // NotifyExitHopHtlc 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. 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.
 func (i *InvoiceRegistry) NotifyExitHopHtlc(rHash lntypes.Hash,
-	amtPaid lnwire.MilliSatoshi) (*HodlEvent, error) {
+	amtPaid lnwire.MilliSatoshi, hodlChan chan<- interface{}) (
+	*HodlEvent, error) {
 
 	i.Lock()
 	defer i.Unlock()
@@ -474,7 +495,7 @@ func (i *InvoiceRegistry) NotifyExitHopHtlc(rHash lntypes.Hash,
 
 	// 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)
+	invoice, err := i.cdb.AcceptOrSettleInvoice(rHash, amtPaid)
 	switch err {
 
 	// If invoice is already settled, settle htlc. This means we accept more
@@ -486,14 +507,55 @@ func (i *InvoiceRegistry) NotifyExitHopHtlc(rHash lntypes.Hash,
 	case channeldb.ErrInvoiceAlreadyCanceled:
 		return createEvent(nil), nil
 
-	// If this call settled the invoice, settle the htlc.
+	// If invoice is already accepted, add this htlc to the list of
+	// subscribers.
+	case channeldb.ErrInvoiceAlreadyAccepted:
+		i.hodlSubscribe(hodlChan, rHash)
+		return nil, nil
+
+	// If this call settled the invoice, settle the htlc. Otherwise
+	// subscribe for a future hodl event.
 	case nil:
 		i.notifyClients(rHash, invoice, invoice.Terms.State)
-		return createEvent(&invoice.Terms.PaymentPreimage), nil
+		switch invoice.Terms.State {
+		case channeldb.ContractSettled:
+			return createEvent(&invoice.Terms.PaymentPreimage), nil
+		case channeldb.ContractAccepted:
+			// Subscribe to updates to this invoice.
+			i.hodlSubscribe(hodlChan, rHash)
+			return nil, nil
+		default:
+			return nil, fmt.Errorf("unexpected invoice state %v",
+				invoice.Terms.State)
+		}
+
+	default:
+		return nil, err
+	}
+}
+
+// SettleHodlInvoice sets the preimage of a hodl invoice.
+func (i *InvoiceRegistry) SettleHodlInvoice(preimage lntypes.Preimage) error {
+	i.Lock()
+	defer i.Unlock()
+
+	invoice, err := i.cdb.SettleHoldInvoice(preimage)
+	if err != nil {
+		log.Errorf("Invoice SetPreimage %v: %v", preimage, err)
+		return err
 	}
 
-	// If another error occurred, return that.
-	return nil, err
+	hash := preimage.Hash()
+	log.Infof("Notifying clients of set preimage to %v",
+		invoice.Terms.PaymentPreimage)
+
+	i.notifyHodlSubscribers(HodlEvent{
+		Hash:     hash,
+		Preimage: &preimage,
+	})
+	i.notifyClients(hash, invoice, invoice.Terms.State)
+
+	return nil
 }
 
 // CancelInvoice attempts to cancel the invoice corresponding to the passed
@@ -512,13 +574,14 @@ func (i *InvoiceRegistry) CancelInvoice(payHash lntypes.Hash) error {
 		log.Debugf("Invoice %v already canceled", payHash)
 		return nil
 	}
-
 	if err != nil {
 		return err
 	}
 
 	log.Infof("Invoice %v canceled", payHash)
-
+	i.notifyHodlSubscribers(HodlEvent{
+		Hash: payHash,
+	})
 	i.notifyClients(payHash, invoice, channeldb.ContractCanceled)
 
 	return nil
@@ -770,3 +833,60 @@ func (i *InvoiceRegistry) SubscribeSingleInvoice(
 
 	return client
 }
+
+// notifyHodlSubscribers sends out the hodl event to all current subscribers.
+func (i *InvoiceRegistry) notifyHodlSubscribers(hodlEvent HodlEvent) {
+	subscribers, ok := i.hodlSubscriptions[hodlEvent.Hash]
+	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 <- hodlEvent:
+		case <-i.quit:
+			return
+		}
+
+		delete(i.hodlReverseSubscriptions[subscriber], hodlEvent.Hash)
+	}
+
+	delete(i.hodlSubscriptions, hodlEvent.Hash)
+}
+
+// hodlSubscribe adds a new invoice subscription.
+func (i *InvoiceRegistry) hodlSubscribe(subscriber chan<- interface{},
+	hash lntypes.Hash) {
+
+	log.Debugf("Hodl subscribe for %v", hash)
+
+	subscriptions, ok := i.hodlSubscriptions[hash]
+	if !ok {
+		subscriptions = make(map[chan<- interface{}]struct{})
+		i.hodlSubscriptions[hash] = subscriptions
+	}
+	subscriptions[subscriber] = struct{}{}
+
+	reverseSubscriptions, ok := i.hodlReverseSubscriptions[subscriber]
+	if !ok {
+		reverseSubscriptions = make(map[lntypes.Hash]struct{})
+		i.hodlReverseSubscriptions[subscriber] = reverseSubscriptions
+	}
+	reverseSubscriptions[hash] = 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)
+}

invoices/invoiceregistry_test.go

@@ -117,9 +117,11 @@ func TestSettleInvoice(t *testing.T) {
 		t.Fatal("no update received")
 	}
 
+	hodlChan := make(chan interface{}, 1)
+
 	// Settle invoice with a slightly higher amount.
 	amtPaid := lnwire.MilliSatoshi(100500)
-	_, err = registry.NotifyExitHopHtlc(hash, amtPaid)
+	_, err = registry.NotifyExitHopHtlc(hash, amtPaid, hodlChan)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -151,13 +153,13 @@ func TestSettleInvoice(t *testing.T) {
 	}
 
 	// Try to settle again.
-	_, err = registry.NotifyExitHopHtlc(hash, amtPaid)
+	_, err = registry.NotifyExitHopHtlc(hash, amtPaid, hodlChan)
 	if err != nil {
 		t.Fatal("expected duplicate settle to succeed")
 	}
 
 	// Try to settle again with a different amount.
-	_, err = registry.NotifyExitHopHtlc(hash, amtPaid+600)
+	_, err = registry.NotifyExitHopHtlc(hash, amtPaid+600, hodlChan)
 	if err != nil {
 		t.Fatal("expected duplicate settle to succeed")
 	}
@@ -176,6 +178,13 @@ func TestSettleInvoice(t *testing.T) {
 	if err != channeldb.ErrInvoiceAlreadySettled {
 		t.Fatal("expected cancelation of a settled invoice to fail")
 	}
+
+	// As this is a direct sette, we expect nothing on the hodl chan.
+	select {
+	case <-hodlChan:
+		t.Fatal("unexpected event")
+	default:
+	}
 }
 
 // TestCancelInvoice tests cancelation of an invoice and related notifications.
@@ -264,7 +273,8 @@ func TestCancelInvoice(t *testing.T) {
 
 	// Notify arrival of a new htlc paying to this invoice. This should
 	// succeed.
-	event, err := registry.NotifyExitHopHtlc(hash, amt)
+	hodlChan := make(chan interface{})
+	event, err := registry.NotifyExitHopHtlc(hash, amt, hodlChan)
 	if err != nil {
 		t.Fatal("expected settlement of a canceled invoice to succeed")
 	}
@@ -274,6 +284,134 @@ func TestCancelInvoice(t *testing.T) {
 	}
 }
 
+// TestHoldInvoice tests settling of a hold invoice and related notifications.
+func TestHoldInvoice(t *testing.T) {
+	defer timeout(t)()
+
+	cdb, cleanup, err := newDB()
+	defer cleanup()
+
+	// Instantiate and start the invoice registry.
+	registry := NewRegistry(cdb, decodeExpiry)
+
+	err = registry.Start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer registry.Stop()
+
+	allSubscriptions := registry.SubscribeNotifications(0, 0)
+	defer allSubscriptions.Cancel()
+
+	// Subscribe to the not yet existing invoice.
+	subscription := registry.SubscribeSingleInvoice(hash)
+	defer subscription.Cancel()
+
+	if subscription.hash != hash {
+		t.Fatalf("expected subscription for provided hash")
+	}
+
+	// Add the invoice.
+	invoice := &channeldb.Invoice{
+		Terms: channeldb.ContractTerm{
+			PaymentPreimage: channeldb.UnknownPreimage,
+			Value:           lnwire.MilliSatoshi(100000),
+		},
+	}
+
+	_, err = registry.AddInvoice(invoice, hash)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// We expect the open state to be sent to the single invoice subscriber.
+	update := <-subscription.Updates
+	if update.Terms.State != channeldb.ContractOpen {
+		t.Fatalf("expected state ContractOpen, but got %v",
+			update.Terms.State)
+	}
+
+	// We expect a new invoice notification to be sent out.
+	newInvoice := <-allSubscriptions.NewInvoices
+	if newInvoice.Terms.State != channeldb.ContractOpen {
+		t.Fatalf("expected state ContractOpen, but got %v",
+			newInvoice.Terms.State)
+	}
+
+	// Use slightly higher amount for accept/settle.
+	amtPaid := lnwire.MilliSatoshi(100500)
+
+	hodlChan := make(chan interface{}, 1)
+
+	// NotifyExitHopHtlc without a preimage present in the invoice registry
+	// should be possible.
+	event, err := registry.NotifyExitHopHtlc(hash, amtPaid, hodlChan)
+	if err != nil {
+		t.Fatalf("expected settle to succeed but got %v", err)
+	}
+	if event != nil {
+		t.Fatalf("unexpect direct settle")
+	}
+
+	// Test idempotency.
+	event, err = registry.NotifyExitHopHtlc(hash, amtPaid, hodlChan)
+	if err != nil {
+		t.Fatalf("expected settle to succeed but got %v", err)
+	}
+	if event != nil {
+		t.Fatalf("unexpect direct settle")
+	}
+
+	// We expect the accepted state to be sent to the single invoice
+	// subscriber. For all invoice subscribers, we don't expect an update.
+	// Those only get notified on settle.
+	update = <-subscription.Updates
+	if update.Terms.State != channeldb.ContractAccepted {
+		t.Fatalf("expected state ContractAccepted, but got %v",
+			update.Terms.State)
+	}
+	if update.AmtPaid != amtPaid {
+		t.Fatal("invoice AmtPaid incorrect")
+	}
+
+	// Settling with preimage should succeed.
+	err = registry.SettleHodlInvoice(preimage)
+	if err != nil {
+		t.Fatal("expected set preimage to succeed")
+	}
+
+	hodlEvent := (<-hodlChan).(HodlEvent)
+	if *hodlEvent.Preimage != preimage {
+		t.Fatal("unexpected preimage in hodl event")
+	}
+
+	// We expect a settled notification to be sent out for both all and
+	// single invoice subscribers.
+	settledInvoice := <-allSubscriptions.SettledInvoices
+	if settledInvoice.Terms.State != channeldb.ContractSettled {
+		t.Fatalf("expected state ContractSettled, but got %v",
+			settledInvoice.Terms.State)
+	}
+
+	update = <-subscription.Updates
+	if update.Terms.State != channeldb.ContractSettled {
+		t.Fatalf("expected state ContractSettled, but got %v",
+			update.Terms.State)
+	}
+
+	// Idempotency.
+	err = registry.SettleHodlInvoice(preimage)
+	if err != channeldb.ErrInvoiceAlreadySettled {
+		t.Fatalf("expected ErrInvoiceAlreadySettled but got %v", err)
+	}
+
+	// Try to cancel.
+	err = registry.CancelInvoice(hash)
+	if err == nil {
+		t.Fatal("expected cancelation of a settled invoice to fail")
+	}
+}
+
 func newDB() (*channeldb.DB, func(), error) {
 	// First, create a temporary directory to be used for the duration of
 	// this test.

invoices/utils_test.go

@@ -0,0 +1,26 @@
+package invoices
+
+import (
+	"os"
+	"runtime/pprof"
+	"testing"
+	"time"
+)
+
+// timeout implements a test level timeout.
+func timeout(t *testing.T) func() {
+	done := make(chan struct{})
+	go func() {
+		select {
+		case <-time.After(5 * time.Second):
+			pprof.Lookup("goroutine").WriteTo(os.Stdout, 1)
+
+			panic("test timeout")
+		case <-done:
+		}
+	}()
+
+	return func() {
+		close(done)
+	}
+}