From 1f8065de3593635a5bbfa40b2e2e0b87c1f5104e Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Tue, 10 Oct 2023 18:41:06 +0200 Subject: [PATCH 1/5] channeldb: add k/v implementation for InvoiceDB.FetchPendingInvoices --- channeldb/invoice_test.go | 54 +++++++++++++++++++++++++++++++++++ channeldb/invoices.go | 59 +++++++++++++++++++++++++++++++++++++++ invoices/interface.go | 5 ++++ invoices/mock.go | 9 ++++++ 4 files changed, 127 insertions(+) diff --git a/channeldb/invoice_test.go b/channeldb/invoice_test.go index e4d9ffffa..f0d667332 100644 --- a/channeldb/invoice_test.go +++ b/channeldb/invoice_test.go @@ -1013,6 +1013,60 @@ func TestSettleIndexAmpPayments(t *testing.T) { require.Nil(t, err) } +// 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, OptionClock(testClock)) + require.NoError(t, err, "unable to make test db") + + ctxb := context.Background() + + // 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) + + for i := 1; i <= numInvoices; 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() + + _, 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++ + } + + // Fetch all pending invoices. + pending, err = db.FetchPendingInvoices(ctxb) + require.NoError(t, err) + require.Equal(t, pendingInvoices, pending) +} + // TestScanInvoices tests that ScanInvoices scans through all stored invoices // correctly. func TestScanInvoices(t *testing.T) { diff --git a/channeldb/invoices.go b/channeldb/invoices.go index d7016ff33..cc665d097 100644 --- a/channeldb/invoices.go +++ b/channeldb/invoices.go @@ -433,6 +433,65 @@ func fetchInvoiceNumByRef(invoiceIndex, payAddrIndex, setIDIndex kvdb.RBucket, } } +// 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) { + + result := make(map[lntypes.Hash]invpkg.Invoice) + + err := kvdb.View(d, func(tx kvdb.RTx) error { + invoices := tx.ReadBucket(invoiceBucket) + if invoices == nil { + return nil + } + + invoiceIndex := invoices.NestedReadBucket(invoiceIndexBucket) + if invoiceIndex == nil { + // Mask the error if there's no invoice + // index as that simply means there are no + // invoices added yet to the DB. In this case + // we simply return an empty list. + return nil + } + + 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.IsPending() { + var paymentHash lntypes.Hash + copy(paymentHash[:], k) + result[paymentHash] = invoice + } + + return nil + }) + }, func() { + result = make(map[lntypes.Hash]invpkg.Invoice) + }) + + if err != nil { + return nil, err + } + + return result, nil +} + // 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 diff --git a/invoices/interface.go b/invoices/interface.go index 36bc0ea48..d88a96753 100644 --- a/invoices/interface.go +++ b/invoices/interface.go @@ -56,6 +56,11 @@ type InvoiceDB interface { 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. QueryInvoices(ctx context.Context, q InvoiceQuery) (InvoiceSlice, error) diff --git a/invoices/mock.go b/invoices/mock.go index 8410208bf..5c419d0b1 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" ) @@ -55,6 +57,13 @@ func (m *MockInvoiceDB) ScanInvoices(scanFunc InvScanFunc, return args.Error(0) } +func (m *MockInvoiceDB) FetchPendingInvoices(ctx context.Context) ( + map[lntypes.Hash]Invoice, error) { + + args := m.Called(ctx) + return args.Get(0).(map[lntypes.Hash]Invoice), args.Error(1) +} + func (m *MockInvoiceDB) QueryInvoices(q InvoiceQuery) (InvoiceSlice, error) { args := m.Called(q) invoiceSlice, _ := args.Get(0).(InvoiceSlice) From 8ab267ad90012d0971681547d0f0c4adba8cd904 Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Tue, 10 Oct 2023 22:23:50 +0200 Subject: [PATCH 2/5] channeldb: add k/v implementation for InvoiceDB.DeleteCanceledInvoices method --- channeldb/invoice_test.go | 59 ++++++++++++++++++++++++ channeldb/invoices.go | 96 +++++++++++++++++++++++++++++++++++++++ invoices/interface.go | 4 ++ invoices/mock.go | 6 +++ 4 files changed, 165 insertions(+) diff --git a/channeldb/invoice_test.go b/channeldb/invoice_test.go index f0d667332..fcc959516 100644 --- a/channeldb/invoice_test.go +++ b/channeldb/invoice_test.go @@ -3146,6 +3146,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 cc665d097..608491e69 100644 --- a/channeldb/invoices.go +++ b/channeldb/invoices.go @@ -2761,6 +2761,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/invoices/interface.go b/invoices/interface.go index d88a96753..7bb39f846 100644 --- a/invoices/interface.go +++ b/invoices/interface.go @@ -95,6 +95,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/mock.go b/invoices/mock.go index 5c419d0b1..f7f6a0bc2 100644 --- a/invoices/mock.go +++ b/invoices/mock.go @@ -85,3 +85,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) +} From d7d385f78f749ce6eb18f6e56ce0af7579c80322 Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Tue, 10 Oct 2023 22:30:38 +0200 Subject: [PATCH 3/5] invoices: refactor initial invoice scan when the registry starts In this commit we change how we select invoices to follow or delete when starting the InvoiceRegistry to instead of using the deperecated scan func from channeldb, use specific functions to gather pending and delete canceled invoices. --- invoices/invoiceregistry.go | 68 ++++++++----------------------------- 1 file changed, 15 insertions(+), 53 deletions(-) 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 } } From 8bf7503aa40fe297b52a581890fe92ac41575c99 Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Tue, 10 Oct 2023 22:38:08 +0200 Subject: [PATCH 4/5] invoices: remove the now unused ScanInvoices method --- channeldb/invoice_test.go | 60 --------------------------------------- channeldb/invoices.go | 47 ------------------------------ invoices/interface.go | 15 ---------- invoices/mock.go | 8 ------ 4 files changed, 130 deletions(-) diff --git a/channeldb/invoice_test.go b/channeldb/invoice_test.go index fcc959516..262ad402c 100644 --- a/channeldb/invoice_test.go +++ b/channeldb/invoice_test.go @@ -1067,66 +1067,6 @@ func TestFetchPendingInvoices(t *testing.T) { require.Equal(t, pendingInvoices, pending) } -// TestScanInvoices tests that ScanInvoices scans through all stored invoices -// correctly. -func TestScanInvoices(t *testing.T) { - t.Parallel() - - db, err := MakeTestInvoiceDB(t) - 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) - - // 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)) - require.NoError(t, err) - - paymentHash := invoice.Terms.PaymentPreimage.Hash() - testInvoices[paymentHash] = invoice - - _, err = db.AddInvoice(ctxb, invoice, paymentHash) - require.NoError(t, err) - } - - 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) -} - // TestDuplicateSettleInvoice tests that if we add a new invoice and settle it // twice, then the second time we also receive the invoice that we settled as a // return argument. diff --git a/channeldb/invoices.go b/channeldb/invoices.go index 608491e69..0c7b54436 100644 --- a/channeldb/invoices.go +++ b/channeldb/invoices.go @@ -492,53 +492,6 @@ func (d *DB) FetchPendingInvoices(_ context.Context) ( return result, nil } -// 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 { - - return kvdb.View(d, func(tx kvdb.RTx) error { - invoices := tx.ReadBucket(invoiceBucket) - if invoices == nil { - return invpkg.ErrNoInvoicesCreated - } - - invoiceIndex := invoices.NestedReadBucket(invoiceIndexBucket) - if invoiceIndex == nil { - // Mask the error if there's no invoice - // index as that simply means there are no - // invoices added yet to the DB. In this case - // we simply return an empty list. - return nil - } - - 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 - } - - var paymentHash lntypes.Hash - copy(paymentHash[:], k) - - return scanFunc(paymentHash, &invoice) - }) - }, reset) -} - // QueryInvoices allows a caller to query the invoice database for invoices // within the specified add index range. func (d *DB) QueryInvoices(_ context.Context, q invpkg.InvoiceQuery) ( diff --git a/invoices/interface.go b/invoices/interface.go index 7bb39f846..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,17 +41,6 @@ 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, diff --git a/invoices/mock.go b/invoices/mock.go index f7f6a0bc2..25c81a35d 100644 --- a/invoices/mock.go +++ b/invoices/mock.go @@ -49,14 +49,6 @@ func (m *MockInvoiceDB) LookupInvoice(ref InvoiceRef) (Invoice, error) { return invoice, args.Error(1) } -func (m *MockInvoiceDB) ScanInvoices(scanFunc InvScanFunc, - reset func()) error { - - args := m.Called(scanFunc, reset) - - return args.Error(0) -} - func (m *MockInvoiceDB) FetchPendingInvoices(ctx context.Context) ( map[lntypes.Hash]Invoice, error) { From bedafca49dc4d82bbff687dc8a598f86f0f8d5c9 Mon Sep 17 00:00:00 2001 From: Andras Banki-Horvath Date: Wed, 11 Oct 2023 14:42:15 +0200 Subject: [PATCH 5/5] docs: update release notes for 0.18.0 --- docs/release-notes/release-notes-0.18.0.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/docs/release-notes/release-notes-0.18.0.md b/docs/release-notes/release-notes-0.18.0.md index 2bf472de9..1f29edba9 100644 --- a/docs/release-notes/release-notes-0.18.0.md +++ b/docs/release-notes/release-notes-0.18.0.md @@ -70,6 +70,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