Merge pull request #4285 from cfromknecht/pay-addr-index

channeldb: index payments by payment addr, use payment hash as fallback

invoices/invoiceregistry.go

@@ -61,8 +61,8 @@ type RegistryConfig struct {
 // 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 {
-	// hash is the payment hash of the htlc to release.
-	hash lntypes.Hash
+	// invoiceRef identifiers the invoice this htlc belongs to.
+	invoiceRef channeldb.InvoiceRef
 
 	// key is the circuit key of the htlc to release.
 	key channeldb.CircuitKey
@@ -280,7 +280,8 @@ func (i *InvoiceRegistry) invoiceEventLoop() {
 			// the subscriber.
 			case *SingleInvoiceSubscription:
 				log.Infof("New single invoice subscription "+
-					"client: id=%v, hash=%v", e.id, e.hash)
+					"client: id=%v, ref=%v", e.id,
+					e.invoiceRef)
 
 				i.singleNotificationClients[e.id] = e
 			}
@@ -288,8 +289,8 @@ func (i *InvoiceRegistry) invoiceEventLoop() {
 		// A new htlc came in for auto-release.
 		case event := <-i.htlcAutoReleaseChan:
 			log.Debugf("Scheduling auto-release for htlc: "+
-				"hash=%v, key=%v at %v",
-				event.hash, event.key, event.releaseTime)
+				"ref=%v, key=%v at %v",
+				event.invoiceRef, event.key, event.releaseTime)
 
 			// We use an independent timer for every htlc rather
 			// than a set timer that is reset with every htlc coming
@@ -302,7 +303,7 @@ func (i *InvoiceRegistry) invoiceEventLoop() {
 		case <-nextReleaseTick:
 			event := autoReleaseHeap.Pop().(*htlcReleaseEvent)
 			err := i.cancelSingleHtlc(
-				event.hash, event.key, ResultMppTimeout,
+				event.invoiceRef, event.key, ResultMppTimeout,
 			)
 			if err != nil {
 				log.Errorf("HTLC timer: %v", err)
@@ -319,7 +320,7 @@ func (i *InvoiceRegistry) invoiceEventLoop() {
 func (i *InvoiceRegistry) dispatchToSingleClients(event *invoiceEvent) {
 	// Dispatch to single invoice subscribers.
 	for _, client := range i.singleNotificationClients {
-		if client.hash != event.hash {
+		if client.invoiceRef.PayHash() != event.hash {
 			continue
 		}
 
@@ -453,7 +454,7 @@ func (i *InvoiceRegistry) deliverBacklogEvents(client *InvoiceSubscription) erro
 func (i *InvoiceRegistry) deliverSingleBacklogEvents(
 	client *SingleInvoiceSubscription) error {
 
-	invoice, err := i.cdb.LookupInvoice(client.hash)
+	invoice, err := i.cdb.LookupInvoice(client.invoiceRef)
 
 	// It is possible that the invoice does not exist yet, but the client is
 	// already watching it in anticipation.
@@ -467,7 +468,7 @@ func (i *InvoiceRegistry) deliverSingleBacklogEvents(
 	}
 
 	err = client.notify(&invoiceEvent{
-		hash:    client.hash,
+		hash:    client.invoiceRef.PayHash(),
 		invoice: &invoice,
 	})
 	if err != nil {
@@ -490,8 +491,8 @@ func (i *InvoiceRegistry) AddInvoice(invoice *channeldb.Invoice,
 
 	i.Lock()
 
-	log.Debugf("Invoice(%v): added with terms %v", paymentHash,
-		invoice.Terms)
+	ref := channeldb.InvoiceRefByHash(paymentHash)
+	log.Debugf("Invoice%v: added with terms %v", ref, invoice.Terms)
 
 	addIndex, err := i.cdb.AddInvoice(invoice, paymentHash)
 	if err != nil {
@@ -521,17 +522,18 @@ func (i *InvoiceRegistry) LookupInvoice(rHash lntypes.Hash) (channeldb.Invoice,
 
 	// We'll check the database to see if there's an existing matching
 	// invoice.
-	return i.cdb.LookupInvoice(rHash)
+	ref := channeldb.InvoiceRefByHash(rHash)
+	return i.cdb.LookupInvoice(ref)
 }
 
 // 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(hash lntypes.Hash,
+func (i *InvoiceRegistry) startHtlcTimer(invoiceRef channeldb.InvoiceRef,
 	key channeldb.CircuitKey, acceptTime time.Time) error {
 
 	releaseTime := acceptTime.Add(i.cfg.HtlcHoldDuration)
 	event := &htlcReleaseEvent{
-		hash:        hash,
+		invoiceRef:  invoiceRef,
 		key:         key,
 		releaseTime: releaseTime,
 	}
@@ -548,7 +550,7 @@ func (i *InvoiceRegistry) startHtlcTimer(hash lntypes.Hash,
 // 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(hash lntypes.Hash,
+func (i *InvoiceRegistry) cancelSingleHtlc(invoiceRef channeldb.InvoiceRef,
 	key channeldb.CircuitKey, result FailResolutionResult) error {
 
 	i.Lock()
@@ -560,7 +562,7 @@ func (i *InvoiceRegistry) cancelSingleHtlc(hash lntypes.Hash,
 		// Only allow individual htlc cancelation on open invoices.
 		if invoice.State != channeldb.ContractOpen {
 			log.Debugf("cancelSingleHtlc: invoice %v no longer "+
-				"open", hash)
+				"open", invoiceRef)
 
 			return nil, nil
 		}
@@ -575,13 +577,13 @@ func (i *InvoiceRegistry) cancelSingleHtlc(hash lntypes.Hash,
 		// resolved.
 		if htlc.State != channeldb.HtlcStateAccepted {
 			log.Debugf("cancelSingleHtlc: htlc %v on invoice %v "+
-				"is already resolved", key, hash)
+				"is already resolved", key, invoiceRef)
 
 			return nil, nil
 		}
 
 		log.Debugf("cancelSingleHtlc: cancelling htlc %v on invoice %v",
-			key, hash)
+			key, invoiceRef)
 
 		// Return an update descriptor that cancels htlc and keeps
 		// invoice open.
@@ -598,7 +600,7 @@ func (i *InvoiceRegistry) cancelSingleHtlc(hash lntypes.Hash,
 	// 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(hash,
+	invoice, err := i.cdb.UpdateInvoice(invoiceRef,
 		func(invoice *channeldb.Invoice) (
 			*channeldb.InvoiceUpdateDesc, error) {
 
@@ -721,11 +723,9 @@ func (i *InvoiceRegistry) NotifyExitHopHtlc(rHash lntypes.Hash,
 	circuitKey channeldb.CircuitKey, hodlChan chan<- interface{},
 	payload Payload) (HtlcResolution, error) {
 
-	mpp := payload.MultiPath()
-
 	// Create the update context containing the relevant details of the
 	// incoming htlc.
-	updateCtx := invoiceUpdateCtx{
+	ctx := invoiceUpdateCtx{
 		hash:                 rHash,
 		circuitKey:           circuitKey,
 		amtPaid:              amtPaid,
@@ -733,16 +733,16 @@ func (i *InvoiceRegistry) NotifyExitHopHtlc(rHash lntypes.Hash,
 		currentHeight:        currentHeight,
 		finalCltvRejectDelta: i.cfg.FinalCltvRejectDelta,
 		customRecords:        payload.CustomRecords(),
-		mpp:                  mpp,
+		mpp:                  payload.MultiPath(),
 	}
 
 	// 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 {
-		err := i.processKeySend(updateCtx)
+		err := i.processKeySend(ctx)
 		if err != nil {
-			updateCtx.log(fmt.Sprintf("keysend error: %v", err))
+			ctx.log(fmt.Sprintf("keysend error: %v", err))
 
 			return NewFailResolution(
 				circuitKey, currentHeight, ResultKeySendError,
@@ -752,7 +752,7 @@ func (i *InvoiceRegistry) NotifyExitHopHtlc(rHash lntypes.Hash,
 
 	// Execute locked notify exit hop logic.
 	i.Lock()
-	resolution, err := i.notifyExitHopHtlcLocked(&updateCtx, hodlChan)
+	resolution, err := i.notifyExitHopHtlcLocked(&ctx, hodlChan)
 	i.Unlock()
 	if err != nil {
 		return nil, err
@@ -764,7 +764,9 @@ func (i *InvoiceRegistry) NotifyExitHopHtlc(rHash lntypes.Hash,
 	// main event loop.
 	case *htlcAcceptResolution:
 		if r.autoRelease {
-			err := i.startHtlcTimer(rHash, circuitKey, r.acceptTime)
+			err := i.startHtlcTimer(
+				ctx.invoiceRef(), circuitKey, r.acceptTime,
+			)
 			if err != nil {
 				return nil, err
 			}
@@ -798,7 +800,7 @@ func (i *InvoiceRegistry) notifyExitHopHtlcLocked(
 		updateSubscribers bool
 	)
 	invoice, err := i.cdb.UpdateInvoice(
-		ctx.hash,
+		ctx.invoiceRef(),
 		func(inv *channeldb.Invoice) (
 			*channeldb.InvoiceUpdateDesc, error) {
 
@@ -952,7 +954,8 @@ func (i *InvoiceRegistry) SettleHodlInvoice(preimage lntypes.Preimage) error {
 	}
 
 	hash := preimage.Hash()
-	invoice, err := i.cdb.UpdateInvoice(hash, updateInvoice)
+	invoiceRef := channeldb.InvoiceRefByHash(hash)
+	invoice, err := i.cdb.UpdateInvoice(invoiceRef, updateInvoice)
 	if err != nil {
 		log.Errorf("SettleHodlInvoice with preimage %v: %v",
 			preimage, err)
@@ -960,7 +963,7 @@ func (i *InvoiceRegistry) SettleHodlInvoice(preimage lntypes.Preimage) error {
 		return err
 	}
 
-	log.Debugf("Invoice(%v): settled with preimage %v", hash,
+	log.Debugf("Invoice%v: settled with preimage %v", invoiceRef,
 		invoice.Terms.PaymentPreimage)
 
 	// In the callback, we marked the invoice as settled. UpdateInvoice will
@@ -1001,7 +1004,8 @@ func (i *InvoiceRegistry) cancelInvoiceImpl(payHash lntypes.Hash,
 	i.Lock()
 	defer i.Unlock()
 
-	log.Debugf("Invoice(%v): canceling invoice", payHash)
+	ref := channeldb.InvoiceRefByHash(payHash)
+	log.Debugf("Invoice%v: canceling invoice", ref)
 
 	updateInvoice := func(invoice *channeldb.Invoice) (
 		*channeldb.InvoiceUpdateDesc, error) {
@@ -1022,12 +1026,13 @@ func (i *InvoiceRegistry) cancelInvoiceImpl(payHash lntypes.Hash,
 		}, nil
 	}
 
-	invoice, err := i.cdb.UpdateInvoice(payHash, updateInvoice)
+	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", payHash)
+		log.Debugf("Invoice%v: already canceled", ref)
 		return nil
 	}
 	if err != nil {
@@ -1036,12 +1041,12 @@ func (i *InvoiceRegistry) cancelInvoiceImpl(payHash lntypes.Hash,
 
 	// 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.", payHash)
+		log.Debugf("Invoice%v: remains accepted as cancel wasn't"+
+			"explicitly requested.", ref)
 		return nil
 	}
 
-	log.Debugf("Invoice(%v): canceled", payHash)
+	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
@@ -1130,7 +1135,7 @@ type InvoiceSubscription struct {
 type SingleInvoiceSubscription struct {
 	invoiceSubscriptionKit
 
-	hash lntypes.Hash
+	invoiceRef channeldb.InvoiceRef
 
 	// Updates is a channel that we'll use to send all invoice events for
 	// the invoice that is subscribed to.
@@ -1272,7 +1277,7 @@ func (i *InvoiceRegistry) SubscribeSingleInvoice(
 			ntfnQueue:  queue.NewConcurrentQueue(20),
 			cancelChan: make(chan struct{}),
 		},
-		hash: hash,
+		invoiceRef: channeldb.InvoiceRefByHash(hash),
 	}
 	client.ntfnQueue.Start()
 

invoices/invoiceregistry_test.go

@@ -28,7 +28,7 @@ func TestSettleInvoice(t *testing.T) {
 	}
 	defer subscription.Cancel()
 
-	if subscription.hash != testInvoicePaymentHash {
+	if subscription.invoiceRef.PayHash() != testInvoicePaymentHash {
 		t.Fatalf("expected subscription for provided hash")
 	}
 
@@ -240,7 +240,7 @@ func TestCancelInvoice(t *testing.T) {
 	}
 	defer subscription.Cancel()
 
-	if subscription.hash != testInvoicePaymentHash {
+	if subscription.invoiceRef.PayHash() != testInvoicePaymentHash {
 		t.Fatalf("expected subscription for provided hash")
 	}
 
@@ -366,7 +366,7 @@ func TestSettleHoldInvoice(t *testing.T) {
 	}
 	defer subscription.Cancel()
 
-	if subscription.hash != testInvoicePaymentHash {
+	if subscription.invoiceRef.PayHash() != testInvoicePaymentHash {
 		t.Fatalf("expected subscription for provided hash")
 	}
 

invoices/test_utils_test.go

@@ -1,6 +1,7 @@
 package invoices
 
 import (
+	"crypto/rand"
 	"encoding/binary"
 	"encoding/hex"
 	"fmt"
@@ -198,14 +199,20 @@ func newTestInvoice(t *testing.T, preimage lntypes.Preimage,
 		expiry = time.Hour
 	}
 
+	var payAddr [32]byte
+	if _, err := rand.Read(payAddr[:]); err != nil {
+		t.Fatalf("unable to generate payment addr: %v", err)
+	}
+
 	rawInvoice, err := zpay32.NewInvoice(
 		testNetParams,
 		preimage.Hash(),
 		timestamp,
 		zpay32.Amount(testInvoiceAmount),
 		zpay32.Description(testInvoiceDescription),
-		zpay32.Expiry(expiry))
-
+		zpay32.Expiry(expiry),
+		zpay32.PaymentAddr(payAddr),
+	)
 	if err != nil {
 		t.Fatalf("Error while creating new invoice: %v", err)
 	}
@@ -219,6 +226,7 @@ func newTestInvoice(t *testing.T, preimage lntypes.Preimage,
 	return &channeldb.Invoice{
 		Terms: channeldb.ContractTerm{
 			PaymentPreimage: preimage,
+			PaymentAddr:     payAddr,
 			Value:           testInvoiceAmount,
 			Expiry:          expiry,
 			Features:        testFeatures,

invoices/update.go

@@ -22,10 +22,20 @@ type invoiceUpdateCtx struct {
 	mpp                  *record.MPP
 }
 
+// invoiceRef returns an identifier that can be used to lookup or update the
+// invoice this HTLC is targeting.
+func (i *invoiceUpdateCtx) invoiceRef() channeldb.InvoiceRef {
+	if i.mpp != nil {
+		payAddr := i.mpp.PaymentAddr()
+		return channeldb.InvoiceRefByHashAndAddr(i.hash, payAddr)
+	}
+	return channeldb.InvoiceRefByHash(i.hash)
+}
+
 // log logs a message specific to this update context.
 func (i *invoiceUpdateCtx) log(s string) {
-	log.Debugf("Invoice(%x): %v, amt=%v, expiry=%v, circuit=%v, mpp=%v",
-		i.hash[:], s, i.amtPaid, i.expiry, i.circuitKey, i.mpp)
+	log.Debugf("Invoice%v: %v, amt=%v, expiry=%v, circuit=%v, mpp=%v",
+		i.invoiceRef, s, i.amtPaid, i.expiry, i.circuitKey, i.mpp)
 }
 
 // failRes is a helper function which creates a failure resolution with