Merge pull request #8081 from bhandras/invoicedb-refactor

invoices: refactor `InvoiceDB` to eliminate `ScanInvoices`

channeldb/invoice_test.go

@@ -1013,64 +1013,58 @@ func TestSettleIndexAmpPayments(t *testing.T) {
 	require.Nil(t, err)
 }
 
-// TestScanInvoices tests that ScanInvoices scans through all stored invoices
+// TestFetchPendingInvoices tests that we can fetch all pending invoices from
-// correctly.
+// the database using the FetchPendingInvoices method.
-func TestScanInvoices(t *testing.T) {
+func TestFetchPendingInvoices(t *testing.T) {
 	t.Parallel()
 
-	db, err := MakeTestInvoiceDB(t)
+	db, err := MakeTestInvoiceDB(t, OptionClock(testClock))
 	require.NoError(t, err, "unable to make test db")
 
-	var invoices map[lntypes.Hash]*invpkg.Invoice
-	callCount := 0
-	resetCount := 0
-
-	// reset is used to reset/initialize results and is called once
-	// upon calling ScanInvoices and when the underlying transaction is
-	// retried.
-	reset := func() {
-		invoices = make(map[lntypes.Hash]*invpkg.Invoice)
-		callCount = 0
-		resetCount++
-	}
-
-	scanFunc := func(paymentHash lntypes.Hash,
-		invoice *invpkg.Invoice) error {
-
-		invoices[paymentHash] = invoice
-		callCount++
-
-		return nil
-	}
-
 	ctxb := context.Background()
-	// With an empty DB we expect to not scan any invoices.
-	require.NoError(t, db.ScanInvoices(ctxb, scanFunc, reset))
-	require.Equal(t, 0, len(invoices))
-	require.Equal(t, 0, callCount)
-	require.Equal(t, 1, resetCount)
 
-	numInvoices := 5
+	// Make sure that fetching pending invoices from an empty database
-	testInvoices := make(map[lntypes.Hash]*invpkg.Invoice)
+	// returns an empty result and no errors.
+	pending, err := db.FetchPendingInvoices(ctxb)
+	require.NoError(t, err)
+	require.Empty(t, pending)
+
+	const numInvoices = 20
+	var settleIndex uint64 = 1
+	pendingInvoices := make(map[lntypes.Hash]invpkg.Invoice)
 
-	// Now populate the DB and check if we can get all invoices with their
-	// payment hashes as expected.
 	for i := 1; i <= numInvoices; i++ {
-		invoice, err := randInvoice(lnwire.MilliSatoshi(i))
+		amt := lnwire.MilliSatoshi(i * 1000)
+		invoice, err := randInvoice(amt)
 		require.NoError(t, err)
 
+		invoice.CreationDate = invoice.CreationDate.Add(
+			time.Duration(i-1) * time.Second,
+		)
+
 		paymentHash := invoice.Terms.PaymentPreimage.Hash()
-		testInvoices[paymentHash] = invoice
 
 		_, err = db.AddInvoice(ctxb, invoice, paymentHash)
 		require.NoError(t, err)
+
+		// Settle every second invoice.
+		if i%2 == 0 {
+			pendingInvoices[paymentHash] = *invoice
+			continue
+		}
+
+		ref := invpkg.InvoiceRefByHash(paymentHash)
+		_, err = db.UpdateInvoice(ctxb, ref, nil, getUpdateInvoice(amt))
+		require.NoError(t, err)
+
+		settleTestInvoice(invoice, settleIndex)
+		settleIndex++
 	}
 
-	resetCount = 0
+	// Fetch all pending invoices.
-	require.NoError(t, db.ScanInvoices(ctxb, scanFunc, reset))
+	pending, err = db.FetchPendingInvoices(ctxb)
-	require.Equal(t, numInvoices, callCount)
+	require.NoError(t, err)
-	require.Equal(t, testInvoices, invoices)
+	require.Equal(t, pendingInvoices, pending)
-	require.Equal(t, 1, resetCount)
 }
 
 // TestDuplicateSettleInvoice tests that if we add a new invoice and settle it
@@ -3092,6 +3086,65 @@ func TestDeleteInvoices(t *testing.T) {
 	assertInvoiceCount(0)
 }
 
+// TestDeleteCanceledInvoices tests that deleting canceled invoices with the
+// specific DeleteCanceledInvoices method works correctly.
+func TestDeleteCanceledInvoices(t *testing.T) {
+	t.Parallel()
+
+	db, err := MakeTestInvoiceDB(t)
+	require.NoError(t, err, "unable to make test db")
+
+	// Updatefunc is used to cancel an invoice.
+	updateFunc := func(invoice *invpkg.Invoice) (
+		*invpkg.InvoiceUpdateDesc, error) {
+
+		return &invpkg.InvoiceUpdateDesc{
+			UpdateType: invpkg.CancelInvoiceUpdate,
+			State: &invpkg.InvoiceStateUpdateDesc{
+				NewState: invpkg.ContractCanceled,
+			},
+		}, nil
+	}
+
+	// Add some invoices to the test db.
+	ctxb := context.Background()
+	var invoices []invpkg.Invoice
+	for i := 0; i < 10; i++ {
+		invoice, err := randInvoice(lnwire.MilliSatoshi(i + 1))
+		require.NoError(t, err)
+
+		paymentHash := invoice.Terms.PaymentPreimage.Hash()
+		_, err = db.AddInvoice(ctxb, invoice, paymentHash)
+		require.NoError(t, err)
+
+		// Cancel every second invoice.
+		if i%2 == 0 {
+			invoice, err = db.UpdateInvoice(
+				ctxb, invpkg.InvoiceRefByHash(paymentHash), nil,
+				updateFunc,
+			)
+			require.NoError(t, err)
+		} else {
+			invoices = append(invoices, *invoice)
+		}
+	}
+
+	// Delete canceled invoices.
+	require.NoError(t, db.DeleteCanceledInvoices(ctxb))
+
+	// Query to collect all invoices.
+	query := invpkg.InvoiceQuery{
+		IndexOffset:    0,
+		NumMaxInvoices: math.MaxUint64,
+	}
+
+	dbInvoices, err := db.QueryInvoices(ctxb, query)
+	require.NoError(t, err)
+
+	// Check that we really have the expected invoices.
+	require.Equal(t, invoices, dbInvoices.Invoices)
+}
+
 // TestAddInvoiceInvalidFeatureDeps asserts that inserting an invoice with
 // invalid transitive feature dependencies fails with the appropriate error.
 func TestAddInvoiceInvalidFeatureDeps(t *testing.T) {

channeldb/invoices.go

@@ -433,17 +433,18 @@ func fetchInvoiceNumByRef(invoiceIndex, payAddrIndex, setIDIndex kvdb.RBucket,
 	}
 }
 
-// ScanInvoices scans through all invoices and calls the passed scanFunc for
+// FetchPendingInvoices returns all invoices that have not yet been settled or
-// for each invoice with its respective payment hash. Additionally a reset()
+// canceled. The returned map is keyed by the payment hash of each respective
-// closure is passed which is used to reset/initialize partial results and also
+// invoice.
-// to signal if the kvdb.View transaction has been retried.
+func (d *DB) FetchPendingInvoices(_ context.Context) (
-func (d *DB) ScanInvoices(_ context.Context, scanFunc invpkg.InvScanFunc,
+	map[lntypes.Hash]invpkg.Invoice, error) {
-	reset func()) error {
 
-	return kvdb.View(d, func(tx kvdb.RTx) error {
+	result := make(map[lntypes.Hash]invpkg.Invoice)
+
+	err := kvdb.View(d, func(tx kvdb.RTx) error {
 		invoices := tx.ReadBucket(invoiceBucket)
 		if invoices == nil {
-			return invpkg.ErrNoInvoicesCreated
+			return nil
 		}
 
 		invoiceIndex := invoices.NestedReadBucket(invoiceIndexBucket)
@@ -472,12 +473,23 @@ func (d *DB) ScanInvoices(_ context.Context, scanFunc invpkg.InvScanFunc,
 				return err
 			}
 
-			var paymentHash lntypes.Hash
+			if invoice.IsPending() {
-			copy(paymentHash[:], k)
+				var paymentHash lntypes.Hash
+				copy(paymentHash[:], k)
+				result[paymentHash] = invoice
+			}
 
-			return scanFunc(paymentHash, &invoice)
+			return nil
 		})
-	}, reset)
+	}, func() {
+		result = make(map[lntypes.Hash]invpkg.Invoice)
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	return result, nil
 }
 
 // QueryInvoices allows a caller to query the invoice database for invoices
@@ -2702,6 +2714,102 @@ func delAMPSettleIndex(invoiceNum []byte, invoices,
 	return nil
 }
 
+// DeleteCanceledInvoices deletes all canceled invoices from the database.
+func (d *DB) DeleteCanceledInvoices(_ context.Context) error {
+	return kvdb.Update(d, func(tx kvdb.RwTx) error {
+		invoices := tx.ReadWriteBucket(invoiceBucket)
+		if invoices == nil {
+			return nil
+		}
+
+		invoiceIndex := invoices.NestedReadWriteBucket(
+			invoiceIndexBucket,
+		)
+		if invoiceIndex == nil {
+			return invpkg.ErrNoInvoicesCreated
+		}
+
+		invoiceAddIndex := invoices.NestedReadWriteBucket(
+			addIndexBucket,
+		)
+		if invoiceAddIndex == nil {
+			return invpkg.ErrNoInvoicesCreated
+		}
+
+		payAddrIndex := tx.ReadWriteBucket(payAddrIndexBucket)
+
+		return invoiceIndex.ForEach(func(k, v []byte) error {
+			// Skip the special numInvoicesKey as that does not
+			// point to a valid invoice.
+			if bytes.Equal(k, numInvoicesKey) {
+				return nil
+			}
+
+			// Skip sub-buckets.
+			if v == nil {
+				return nil
+			}
+
+			invoice, err := fetchInvoice(v, invoices)
+			if err != nil {
+				return err
+			}
+
+			if invoice.State != invpkg.ContractCanceled {
+				return nil
+			}
+
+			// Delete the payment hash from the invoice index.
+			err = invoiceIndex.Delete(k)
+			if err != nil {
+				return err
+			}
+
+			// Delete payment address index reference if there's a
+			// valid payment address.
+			if invoice.Terms.PaymentAddr != invpkg.BlankPayAddr {
+				// To ensure consistency check that the already
+				// fetched invoice key matches the one in the
+				// payment address index.
+				key := payAddrIndex.Get(
+					invoice.Terms.PaymentAddr[:],
+				)
+				if bytes.Equal(key, k) {
+					// Delete from the payment address
+					// index.
+					if err := payAddrIndex.Delete(
+						invoice.Terms.PaymentAddr[:],
+					); err != nil {
+						return err
+					}
+				}
+			}
+
+			// Remove from the add index.
+			var addIndexKey [8]byte
+			byteOrder.PutUint64(addIndexKey[:], invoice.AddIndex)
+			err = invoiceAddIndex.Delete(addIndexKey[:])
+			if err != nil {
+				return err
+			}
+
+			// Note that we don't need to delete the invoice from
+			// the settle index as it is not added until the
+			// invoice is settled.
+
+			// Now remove all sub invoices.
+			err = delAMPInvoices(k, invoices)
+			if err != nil {
+				return err
+			}
+
+			// Finally remove the serialized invoice from the
+			// invoice bucket.
+			return invoices.Delete(k)
+		})
+	}, func() {})
+}
+
 // DeleteInvoice attempts to delete the passed invoices from the database in
 // one transaction. The passed delete references hold all keys required to
 // delete the invoices without also needing to deserialze them.

docs/release-notes/release-notes-0.18.0.md

@@ -82,6 +82,9 @@
   a context parameter to all `InvoiceDB` methods which is a pre-requisite for
   the SQL implementation.
 
+* [Refactor InvoiceDB](https://github.com/lightningnetwork/lnd/pull/8081) to
+  eliminate the use of `ScanInvoices`.
+
 ## Code Health
 ## Tooling and Documentation
 

invoices/interface.go

@@ -9,10 +9,6 @@ import (
 	"github.com/lightningnetwork/lnd/record"
 )
 
-// InvScanFunc is a helper type used to specify the type used in the
-// ScanInvoices methods (part of the InvoiceDB interface).
-type InvScanFunc func(lntypes.Hash, *Invoice) error
-
 // InvoiceDB is the database that stores the information about invoices.
 type InvoiceDB interface {
 	// AddInvoice inserts the targeted invoice into the database.
@@ -45,16 +41,10 @@ type InvoiceDB interface {
 	// payment.
 	LookupInvoice(ctx context.Context, ref InvoiceRef) (Invoice, error)
 
-	// ScanInvoices scans through all invoices and calls the passed scanFunc
+	// FetchPendingInvoices returns all invoices that have not yet been
-	// for each invoice with its respective payment hash. Additionally a
+	// settled or canceled.
-	// reset() closure is passed which is used to reset/initialize partial
+	FetchPendingInvoices(ctx context.Context) (map[lntypes.Hash]Invoice,
-	// results and also to signal if the kvdb.View transaction has been
+		error)
-	// retried.
-	//
-	// TODO(positiveblue): abstract this functionality so it makes sense for
-	// other backends like sql.
-	ScanInvoices(ctx context.Context, scanFunc InvScanFunc,
-		reset func()) error
 
 	// QueryInvoices allows a caller to query the invoice database for
 	// invoices within the specified add index range.
@@ -90,6 +80,10 @@ type InvoiceDB interface {
 	// deserialze them.
 	DeleteInvoice(ctx context.Context,
 		invoicesToDelete []InvoiceDeleteRef) error
+
+	// DeleteCanceledInvoices removes all canceled invoices from the
+	// database.
+	DeleteCanceledInvoices(ctx context.Context) error
 }
 
 // Payload abstracts access to any additional fields provided in the final hop's

invoices/invoiceregistry.go

@@ -181,68 +181,30 @@ func NewRegistry(idb InvoiceDB, expiryWatcher *InvoiceExpiryWatcher,
 // to the invoice expiry watcher while also attempting to delete all canceled
 // invoices.
 func (i *InvoiceRegistry) scanInvoicesOnStart(ctx context.Context) error {
-	var (
+	pendingInvoices, err := i.idb.FetchPendingInvoices(ctx)
-		pending   []invoiceExpiry
-		removable []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([]InvoiceDeleteRef, 0)
-	}
-
-	scanFunc := func(paymentHash lntypes.Hash,
-		invoice *Invoice) error {
-
-		if invoice.IsPending() {
-			expiryRef := makeInvoiceExpiry(paymentHash, invoice)
-			if expiryRef != nil {
-				pending = append(pending, expiryRef)
-			}
-		} else if i.cfg.GcCanceledInvoicesOnStartup &&
-			invoice.State == 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 := InvoiceDeleteRef{
-				PayHash:     paymentHash,
-				AddIndex:    invoice.AddIndex,
-				SettleIndex: invoice.SettleIndex,
-			}
-
-			if invoice.Terms.PaymentAddr != BlankPayAddr {
-				ref.PayAddr = &invoice.Terms.PaymentAddr
-			}
-
-			removable = append(removable, ref)
-		}
-		return nil
-	}
-
-	err := i.idb.ScanInvoices(ctx, scanFunc, reset)
 	if err != nil {
 		return err
 	}
 
+	var pending []invoiceExpiry
+	for paymentHash, invoice := range pendingInvoices {
+		invoice := invoice
+		expiryRef := makeInvoiceExpiry(paymentHash, &invoice)
+		if expiryRef != nil {
+			pending = append(pending, expiryRef)
+		}
+	}
+
 	log.Debugf("Adding %d pending invoices to the expiry watcher",
 		len(pending))
 	i.expiryWatcher.AddInvoices(pending...)
 
-	if len(removable) > 0 {
+	if i.cfg.GcCanceledInvoicesOnStartup {
-		log.Infof("Attempting to delete %v canceled invoices",
+		log.Infof("Deleting canceled invoices")
-			len(removable))
+		err = i.idb.DeleteCanceledInvoices(ctx)
-		if err := i.idb.DeleteInvoice(ctx, removable); err != nil {
+		if err != nil {
 			log.Warnf("Deleting canceled invoices failed: %v", err)
-		} else {
+			return err
-			log.Infof("Deleted %v canceled invoices",
-				len(removable))
 		}
 	}
 

invoices/mock.go

@@ -1,6 +1,8 @@
 package invoices
 
 import (
+	"context"
+
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/stretchr/testify/mock"
 )
@@ -47,12 +49,11 @@ func (m *MockInvoiceDB) LookupInvoice(ref InvoiceRef) (Invoice, error) {
 	return invoice, args.Error(1)
 }
 
-func (m *MockInvoiceDB) ScanInvoices(scanFunc InvScanFunc,
+func (m *MockInvoiceDB) FetchPendingInvoices(ctx context.Context) (
-	reset func()) error {
+	map[lntypes.Hash]Invoice, error) {
 
-	args := m.Called(scanFunc, reset)
+	args := m.Called(ctx)
-
+	return args.Get(0).(map[lntypes.Hash]Invoice), args.Error(1)
-	return args.Error(0)
 }
 
 func (m *MockInvoiceDB) QueryInvoices(q InvoiceQuery) (InvoiceSlice, error) {
@@ -76,3 +77,9 @@ func (m *MockInvoiceDB) DeleteInvoice(invoices []InvoiceDeleteRef) error {
 
 	return args.Error(0)
 }
+
+func (m *MockInvoiceDB) DeleteCanceledInvoices(ctx context.Context) error {
+	args := m.Called(ctx)
+
+	return args.Error(0)
+}