diff --git a/channeldb/invoice_test.go b/channeldb/invoice_test.go index e4d9ffffa..262ad402c 100644 --- a/channeldb/invoice_test.go +++ b/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 -// correctly. -func TestScanInvoices(t *testing.T) { +// TestFetchPendingInvoices tests that we can fetch all pending invoices from +// the database using the FetchPendingInvoices method. +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 - testInvoices := make(map[lntypes.Hash]*invpkg.Invoice) + // Make sure that fetching pending invoices from an empty database + // 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 - require.NoError(t, db.ScanInvoices(ctxb, scanFunc, reset)) - require.Equal(t, numInvoices, callCount) - require.Equal(t, testInvoices, invoices) - require.Equal(t, 1, resetCount) + // Fetch all pending invoices. + pending, err = db.FetchPendingInvoices(ctxb) + require.NoError(t, err) + require.Equal(t, pendingInvoices, pending) } // 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) { diff --git a/channeldb/invoices.go b/channeldb/invoices.go index d7016ff33..0c7b54436 100644 --- a/channeldb/invoices.go +++ b/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 -// for each invoice with its respective payment hash. Additionally a reset() -// closure is passed which is used to reset/initialize partial results and also -// to signal if the kvdb.View transaction has been retried. -func (d *DB) ScanInvoices(_ context.Context, scanFunc invpkg.InvScanFunc, - reset func()) error { +// FetchPendingInvoices returns all invoices that have not yet been settled or +// canceled. The returned map is keyed by the payment hash of each respective +// invoice. +func (d *DB) FetchPendingInvoices(_ context.Context) ( + map[lntypes.Hash]invpkg.Invoice, 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 - copy(paymentHash[:], k) + if invoice.IsPending() { + 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. diff --git a/docs/release-notes/release-notes-0.18.0.md b/docs/release-notes/release-notes-0.18.0.md index c9052b8be..0b0f2d508 100644 --- a/docs/release-notes/release-notes-0.18.0.md +++ b/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 diff --git a/invoices/interface.go b/invoices/interface.go index 36bc0ea48..cffc8e41c 100644 --- a/invoices/interface.go +++ b/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 - // for each invoice with its respective payment hash. Additionally a - // reset() closure is passed which is used to reset/initialize partial - // results and also to signal if the kvdb.View transaction has been - // retried. - // - // TODO(positiveblue): abstract this functionality so it makes sense for - // other backends like sql. - ScanInvoices(ctx context.Context, scanFunc InvScanFunc, - reset func()) error + // FetchPendingInvoices returns all invoices that have not yet been + // settled or canceled. + FetchPendingInvoices(ctx context.Context) (map[lntypes.Hash]Invoice, + 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 diff --git a/invoices/invoiceregistry.go b/invoices/invoiceregistry.go index e58ad4cdb..806b7a5e6 100644 --- a/invoices/invoiceregistry.go +++ b/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 ( - 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) + pendingInvoices, err := i.idb.FetchPendingInvoices(ctx) 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 { - log.Infof("Attempting to delete %v canceled invoices", - len(removable)) - if err := i.idb.DeleteInvoice(ctx, removable); err != nil { + if i.cfg.GcCanceledInvoicesOnStartup { + log.Infof("Deleting canceled invoices") + err = i.idb.DeleteCanceledInvoices(ctx) + if err != nil { log.Warnf("Deleting canceled invoices failed: %v", err) - } else { - log.Infof("Deleted %v canceled invoices", - len(removable)) + return err } } diff --git a/invoices/mock.go b/invoices/mock.go index 8410208bf..25c81a35d 100644 --- a/invoices/mock.go +++ b/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, - reset func()) error { +func (m *MockInvoiceDB) FetchPendingInvoices(ctx context.Context) ( + map[lntypes.Hash]Invoice, error) { - args := m.Called(scanFunc, reset) - - return args.Error(0) + args := m.Called(ctx) + return args.Get(0).(map[lntypes.Hash]Invoice), args.Error(1) } 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) +}