channeldb: index payments by sequence number

Add an entry to a payments index bucket which maps sequence number
to payment hash when we initiate payments. This allows for more
efficient paginated queries. We create the top level bucket in its
own migration so that we do not need to create it on the fly.

When we retry payments and provide them with a new sequence number, we
delete the index for their existing payment so that we do not have an
index that points to a non-existent payment.

If we delete a payment, we also delete its index entry. This prevents
us from looking up entries from indexes to payments that do not exist.

channeldb/migration16/log.go

@@ -0,0 +1,14 @@
+package migration16
+
+import (
+	"github.com/btcsuite/btclog"
+)
+
+// log is a logger that is initialized as disabled.  This means the package will
+// not perform any logging by default until a logger is set.
+var log = btclog.Disabled
+
+// UseLogger uses a specified Logger to output package logging info.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

channeldb/migration16/migration.go

@@ -0,0 +1,191 @@
+package migration16
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+
+	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/channeldb/kvdb"
+)
+
+var (
+	paymentsRootBucket = []byte("payments-root-bucket")
+
+	paymentSequenceKey = []byte("payment-sequence-key")
+
+	duplicatePaymentsBucket = []byte("payment-duplicate-bucket")
+
+	paymentsIndexBucket = []byte("payments-index-bucket")
+
+	byteOrder = binary.BigEndian
+)
+
+// paymentIndexType indicates the type of index we have recorded in the payment
+// indexes bucket.
+type paymentIndexType uint8
+
+// paymentIndexTypeHash is a payment index type which indicates that we have
+// created an index of payment sequence number to payment hash.
+const paymentIndexTypeHash paymentIndexType = 0
+
+// paymentIndex stores all the information we require to create an index by
+// sequence number for a payment.
+type paymentIndex struct {
+	// paymentHash is the hash of the payment, which is its key in the
+	// payment root bucket.
+	paymentHash []byte
+
+	// sequenceNumbers is the set of sequence numbers associated with this
+	// payment hash. There will be more than one sequence number in the
+	// case where duplicate payments are present.
+	sequenceNumbers [][]byte
+}
+
+// MigrateSequenceIndex migrates the payments db to contain a new bucket which
+// provides an index from sequence number to payment hash. This is required
+// for more efficient sequential lookup of payments, which are keyed by payment
+// hash before this migration.
+func MigrateSequenceIndex(tx kvdb.RwTx) error {
+	log.Infof("Migrating payments to add sequence number index")
+
+	// Get a list of indices we need to write.
+	indexList, err := getPaymentIndexList(tx)
+	if err != nil {
+		return err
+	}
+
+	// Create the top level bucket that we will use to index payments in.
+	bucket, err := tx.CreateTopLevelBucket(paymentsIndexBucket)
+	if err != nil {
+		return err
+	}
+
+	// Write an index for each of our payments.
+	for _, index := range indexList {
+		// Write indexes for each of our sequence numbers.
+		for _, seqNr := range index.sequenceNumbers {
+			err := putIndex(bucket, seqNr, index.paymentHash)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+// putIndex performs a sanity check that ensures we are not writing duplicate
+// indexes to disk then creates the index provided.
+func putIndex(bucket kvdb.RwBucket, sequenceNr, paymentHash []byte) error {
+	// Add a sanity check that we do not already have an entry with
+	// this sequence number.
+	existingEntry := bucket.Get(sequenceNr)
+	if existingEntry != nil {
+		return fmt.Errorf("sequence number: %x duplicated",
+			sequenceNr)
+	}
+
+	bytes, err := serializePaymentIndexEntry(paymentHash)
+	if err != nil {
+		return err
+	}
+
+	return bucket.Put(sequenceNr, bytes)
+}
+
+// serializePaymentIndexEntry serializes a payment hash typed index. The value
+// produced contains a payment index type (which can be used in future to
+// signal different payment index types) and the payment hash.
+func serializePaymentIndexEntry(hash []byte) ([]byte, error) {
+	var b bytes.Buffer
+
+	err := binary.Write(&b, byteOrder, paymentIndexTypeHash)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := wire.WriteVarBytes(&b, 0, hash); err != nil {
+		return nil, err
+	}
+
+	return b.Bytes(), nil
+}
+
+// getPaymentIndexList gets a list of indices we need to write for our current
+// set of payments.
+func getPaymentIndexList(tx kvdb.RTx) ([]paymentIndex, error) {
+	// Iterate over all payments and store their indexing keys. This is
+	// needed, because no modifications are allowed inside a Bucket.ForEach
+	// loop.
+	paymentsBucket := tx.ReadBucket(paymentsRootBucket)
+	if paymentsBucket == nil {
+		return nil, nil
+	}
+
+	var indexList []paymentIndex
+	err := paymentsBucket.ForEach(func(k, v []byte) error {
+		// Get the bucket which contains the payment, fail if the key
+		// does not have a bucket.
+		bucket := paymentsBucket.NestedReadBucket(k)
+		if bucket == nil {
+			return fmt.Errorf("non bucket element in " +
+				"payments bucket")
+		}
+		seqBytes := bucket.Get(paymentSequenceKey)
+		if seqBytes == nil {
+			return fmt.Errorf("nil sequence number bytes")
+		}
+
+		seqNrs, err := fetchSequenceNumbers(bucket)
+		if err != nil {
+			return err
+		}
+
+		// Create an index object with our payment hash and sequence
+		// numbers and append it to our set of indexes.
+		index := paymentIndex{
+			paymentHash:     k,
+			sequenceNumbers: seqNrs,
+		}
+
+		indexList = append(indexList, index)
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return indexList, nil
+}
+
+// fetchSequenceNumbers fetches all the sequence numbers associated with a
+// payment, including those belonging to any duplicate payments.
+func fetchSequenceNumbers(paymentBucket kvdb.RBucket) ([][]byte, error) {
+	seqNum := paymentBucket.Get(paymentSequenceKey)
+	if seqNum == nil {
+		return nil, errors.New("expected sequence number")
+	}
+
+	sequenceNumbers := [][]byte{seqNum}
+
+	// Get the duplicate payments bucket, if it has no duplicates, just
+	// return early with the payment sequence number.
+	duplicates := paymentBucket.NestedReadBucket(duplicatePaymentsBucket)
+	if duplicates == nil {
+		return sequenceNumbers, nil
+	}
+
+	// If we do have duplicated, they are keyed by sequence number, so we
+	// iterate through the duplicates bucket and add them to our set of
+	// sequence numbers.
+	if err := duplicates.ForEach(func(k, v []byte) error {
+		sequenceNumbers = append(sequenceNumbers, k)
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+
+	return sequenceNumbers, nil
+}

channeldb/migration16/migration_test.go

@@ -0,0 +1,144 @@
+package migration16
+
+import (
+	"encoding/hex"
+	"testing"
+
+	"github.com/lightningnetwork/lnd/channeldb/kvdb"
+	"github.com/lightningnetwork/lnd/channeldb/migtest"
+)
+
+var (
+	hexStr = migtest.Hex
+
+	hash1Str   = "02acee76ebd53d00824410cf6adecad4f50334dac702bd5a2d3ba01b91709f0e"
+	hash1      = hexStr(hash1Str)
+	paymentID1 = hexStr("0000000000000001")
+
+	hash2Str   = "62eb3f0a48f954e495d0c14ac63df04a67cefa59dafdbcd3d5046d1f5647840c"
+	hash2      = hexStr(hash2Str)
+	paymentID2 = hexStr("0000000000000002")
+
+	paymentID3 = hexStr("0000000000000003")
+
+	// pre is the data in the payments root bucket in database version 13 format.
+	pre = map[string]interface{}{
+		// A payment without duplicates.
+		hash1: map[string]interface{}{
+			"payment-sequence-key": paymentID1,
+		},
+
+		// A payment with a duplicate.
+		hash2: map[string]interface{}{
+			"payment-sequence-key": paymentID2,
+			"payment-duplicate-bucket": map[string]interface{}{
+				paymentID3: map[string]interface{}{
+					"payment-sequence-key": paymentID3,
+				},
+			},
+		},
+	}
+
+	preFails = map[string]interface{}{
+		// A payment without duplicates.
+		hash1: map[string]interface{}{
+			"payment-sequence-key": paymentID1,
+			"payment-duplicate-bucket": map[string]interface{}{
+				paymentID1: map[string]interface{}{
+					"payment-sequence-key": paymentID1,
+				},
+			},
+		},
+	}
+
+	// post is the expected data after migration.
+	post = map[string]interface{}{
+		paymentID1: paymentHashIndex(hash1Str),
+		paymentID2: paymentHashIndex(hash2Str),
+		paymentID3: paymentHashIndex(hash2Str),
+	}
+)
+
+// paymentHashIndex produces a string that represents the value we expect for
+// our payment indexes from a hex encoded payment hash string.
+func paymentHashIndex(hashStr string) string {
+	hash, err := hex.DecodeString(hashStr)
+	if err != nil {
+		panic(err)
+	}
+
+	bytes, err := serializePaymentIndexEntry(hash)
+	if err != nil {
+		panic(err)
+	}
+
+	return string(bytes)
+}
+
+// MigrateSequenceIndex asserts that the database is properly migrated to
+// contain a payments index.
+func TestMigrateSequenceIndex(t *testing.T) {
+	tests := []struct {
+		name       string
+		shouldFail bool
+		pre        map[string]interface{}
+		post       map[string]interface{}
+	}{
+		{
+			name:       "migration ok",
+			shouldFail: false,
+			pre:        pre,
+			post:       post,
+		},
+		{
+			name:       "duplicate sequence number",
+			shouldFail: true,
+			pre:        preFails,
+			post:       post,
+		},
+		{
+			name:       "no payments",
+			shouldFail: false,
+			pre:        nil,
+			post:       nil,
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+
+		t.Run(test.name, func(t *testing.T) {
+			// Before the migration we have a payments bucket.
+			before := func(tx kvdb.RwTx) error {
+				return migtest.RestoreDB(
+					tx, paymentsRootBucket, test.pre,
+				)
+			}
+
+			// After the migration, we should have an untouched
+			// payments bucket and a new index bucket.
+			after := func(tx kvdb.RwTx) error {
+				if err := migtest.VerifyDB(
+					tx, paymentsRootBucket, test.pre,
+				); err != nil {
+					return err
+				}
+
+				// If we expect our migration to fail, we don't
+				// expect an index bucket.
+				if test.shouldFail {
+					return nil
+				}
+
+				return migtest.VerifyDB(
+					tx, paymentsIndexBucket, test.post,
+				)
+			}
+
+			migtest.ApplyMigration(
+				t, before, after, MigrateSequenceIndex,
+				test.shouldFail,
+			)
+		})
+	}
+}