Merge pull request #3960 from bitromortac/listpayments-pagination

channeldb+lnrpc+lncli: listpayments pagination support

channeldb/duplicate_payments.go

@@ -162,7 +162,7 @@ func fetchDuplicatePayment(bucket kvdb.ReadBucket) (*MPPayment, error) {
 	}
 
 	payment := &MPPayment{
-		sequenceNum:   sequenceNum,
+		SequenceNum:   sequenceNum,
 		Info:          creationInfo,
 		FailureReason: failureReason,
 		Status:        paymentStatus,

channeldb/mp_payment.go

@@ -108,9 +108,9 @@ type HTLCFailInfo struct {
 // have the associated Settle or Fail struct populated if the HTLC is no longer
 // in-flight.
 type MPPayment struct {
-	// sequenceNum is a unique identifier used to sort the payments in
+	// SequenceNum is a unique identifier used to sort the payments in
 	// order of creation.
-	sequenceNum uint64
+	SequenceNum uint64
 
 	// Info holds all static information about this payment, and is
 	// populated when the payment is initiated.

channeldb/payments.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"math"
 	"sort"
 	"time"
 
@@ -246,7 +247,7 @@ func (db *DB) FetchPayments() ([]*MPPayment, error) {
 
 	// Before returning, sort the payments by their sequence number.
 	sort.Slice(payments, func(i, j int) bool {
-		return payments[i].sequenceNum < payments[j].sequenceNum
+		return payments[i].SequenceNum < payments[j].SequenceNum
 	})
 
 	return payments, nil
@@ -334,7 +335,7 @@ func fetchPayment(bucket kvdb.ReadBucket) (*MPPayment, error) {
 	}
 
 	return &MPPayment{
-		sequenceNum:   sequenceNum,
+		SequenceNum:   sequenceNum,
 		Info:          creationInfo,
 		HTLCs:         htlcs,
 		FailureReason: failureReason,
@@ -423,6 +424,140 @@ func fetchHtlcFailInfo(bucket kvdb.ReadBucket) (*HTLCFailInfo, error) {
 	return deserializeHTLCFailInfo(r)
 }
 
+// PaymentsQuery represents a query to the payments database starting or ending
+// at a certain offset index. The number of retrieved records can be limited.
+type PaymentsQuery struct {
+	// IndexOffset determines the starting point of the payments query and
+	// is always exclusive. In normal order, the query starts at the next
+	// higher (available) index compared to IndexOffset. In reversed order,
+	// the query ends at the next lower (available) index compared to the
+	// IndexOffset. In the case of a zero index_offset, the query will start
+	// with the oldest payment when paginating forwards, or will end with
+	// the most recent payment when paginating backwards.
+	IndexOffset uint64
+
+	// MaxPayments is the maximal number of payments returned in the
+	// payments query.
+	MaxPayments uint64
+
+	// Reversed gives a meaning to the IndexOffset. If reversed is set to
+	// true, the query will fetch payments with indices lower than the
+	// IndexOffset, otherwise, it will return payments with indices greater
+	// than the IndexOffset.
+	Reversed bool
+
+	// If IncludeIncomplete is true, then return payments that have not yet
+	// fully completed. This means that pending payments, as well as failed
+	// payments will show up if this field is set to true.
+	IncludeIncomplete bool
+}
+
+// PaymentsResponse contains the result of a query to the payments database.
+// It includes the set of payments that match the query and integers which
+// represent the index of the first and last item returned in the series of
+// payments. These integers allow callers to resume their query in the event
+// that the query's response exceeds the max number of returnable events.
+type PaymentsResponse struct {
+	// Payments is the set of payments returned from the database for the
+	// PaymentsQuery.
+	Payments []MPPayment
+
+	// FirstIndexOffset is the index of the first element in the set of
+	// returned MPPayments. Callers can use this to resume their query
+	// in the event that the slice has too many events to fit into a single
+	// response. The offset can be used to continue reverse pagination.
+	FirstIndexOffset uint64
+
+	// LastIndexOffset is the index of the last element in the set of
+	// returned MPPayments. Callers can use this to resume their query
+	// in the event that the slice has too many events to fit into a single
+	// response. The offset can be used to continue forward pagination.
+	LastIndexOffset uint64
+}
+
+// QueryPayments is a query to the payments database which is restricted
+// to a subset of payments by the payments query, containing an offset
+// index and a maximum number of returned payments.
+func (db *DB) QueryPayments(query PaymentsQuery) (PaymentsResponse, error) {
+	var resp PaymentsResponse
+
+	allPayments, err := db.FetchPayments()
+	if err != nil {
+		return resp, err
+	}
+
+	if len(allPayments) == 0 {
+		return resp, nil
+	}
+
+	indexExclusiveLimit := query.IndexOffset
+	// In backward pagination, if the index limit is the default 0 value,
+	// we set our limit to maxint to include all payments from the highest
+	// sequence number on.
+	if query.Reversed && indexExclusiveLimit == 0 {
+		indexExclusiveLimit = math.MaxInt64
+	}
+
+	for i := range allPayments {
+		var payment *MPPayment
+
+		// If we have the max number of payments we want, exit.
+		if uint64(len(resp.Payments)) == query.MaxPayments {
+			break
+		}
+
+		if query.Reversed {
+			payment = allPayments[len(allPayments)-1-i]
+
+			// In the reversed direction, skip over all payments
+			// that have sequence numbers greater than or equal to
+			// the index offset. We skip payments with equal index
+			// because the offset is exclusive.
+			if payment.SequenceNum >= indexExclusiveLimit {
+				continue
+			}
+		} else {
+			payment = allPayments[i]
+
+			// In the forward direction, skip over all payments that
+			// have sequence numbers less than or equal to the index
+			// offset. We skip payments with equal indexes because
+			// the index offset is exclusive.
+			if payment.SequenceNum <= indexExclusiveLimit {
+				continue
+			}
+		}
+
+		// To keep compatibility with the old API, we only return
+		// non-succeeded payments if requested.
+		if payment.Status != StatusSucceeded &&
+			!query.IncludeIncomplete {
+
+			continue
+		}
+
+		resp.Payments = append(resp.Payments, *payment)
+	}
+
+	// Need to swap the payments slice order if reversed order.
+	if query.Reversed {
+		for l, r := 0, len(resp.Payments)-1; l < r; l, r = l+1, r-1 {
+			resp.Payments[l], resp.Payments[r] =
+				resp.Payments[r], resp.Payments[l]
+		}
+	}
+
+	// Set the first and last index of the returned payments so that the
+	// caller can resume from this point later on.
+	if len(resp.Payments) > 0 {
+		resp.FirstIndexOffset = resp.Payments[0].SequenceNum
+		resp.LastIndexOffset =
+			resp.Payments[len(resp.Payments)-1].SequenceNum
+	}
+
+	return resp, err
+}
+
 // DeletePayments deletes all completed and failed payments from the DB.
 func (db *DB) DeletePayments() error {
 	return kvdb.Update(db, func(tx kvdb.RwTx) error {

channeldb/payments_test.go

@@ -3,6 +3,7 @@ package channeldb
 import (
 	"bytes"
 	"fmt"
+	"math"
 	"math/rand"
 	"reflect"
 	"testing"
@@ -10,6 +11,7 @@ import (
 
 	"github.com/btcsuite/btcd/btcec"
 	"github.com/davecgh/go-spew/spew"
+	"github.com/lightningnetwork/lnd/channeldb/kvdb"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/record"
 	"github.com/lightningnetwork/lnd/routing/route"
@@ -175,3 +177,264 @@ func TestRouteSerialization(t *testing.T) {
 			spew.Sdump(testRoute), spew.Sdump(route2))
 	}
 }
+
+// deletePayment removes a payment with paymentHash from the payments database.
+func deletePayment(t *testing.T, db *DB, paymentHash lntypes.Hash) {
+	t.Helper()
+
+	err := kvdb.Update(db, func(tx kvdb.RwTx) error {
+		payments := tx.ReadWriteBucket(paymentsRootBucket)
+
+		err := payments.DeleteNestedBucket(paymentHash[:])
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		t.Fatalf("could not delete "+
+			"payment: %v", err)
+	}
+}
+
+// TestQueryPayments tests retrieval of payments with forwards and reversed
+// queries.
+func TestQueryPayments(t *testing.T) {
+	// Define table driven test for QueryPayments.
+	// Test payments have sequence indices [1, 3, 4, 5, 6, 7].
+	tests := []struct {
+		name       string
+		query      PaymentsQuery
+		firstIndex uint64
+		lastIndex  uint64
+
+		// expectedSeqNrs contains the set of sequence numbers we expect
+		// our query to return.
+		expectedSeqNrs []uint64
+	}{
+		{
+			name: "IndexOffset at the end of the payments range",
+			query: PaymentsQuery{
+				IndexOffset:       7,
+				MaxPayments:       7,
+				Reversed:          false,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     0,
+			lastIndex:      0,
+			expectedSeqNrs: nil,
+		},
+		{
+			name: "query in forwards order, start at beginning",
+			query: PaymentsQuery{
+				IndexOffset:       0,
+				MaxPayments:       2,
+				Reversed:          false,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     1,
+			lastIndex:      3,
+			expectedSeqNrs: []uint64{1, 3},
+		},
+		{
+			name: "query in forwards order, start at end, overflow",
+			query: PaymentsQuery{
+				IndexOffset:       6,
+				MaxPayments:       2,
+				Reversed:          false,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     7,
+			lastIndex:      7,
+			expectedSeqNrs: []uint64{7},
+		},
+		{
+			name: "start at offset index outside of payments",
+			query: PaymentsQuery{
+				IndexOffset:       20,
+				MaxPayments:       2,
+				Reversed:          false,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     0,
+			lastIndex:      0,
+			expectedSeqNrs: nil,
+		},
+		{
+			name: "overflow in forwards order",
+			query: PaymentsQuery{
+				IndexOffset:       4,
+				MaxPayments:       math.MaxUint64,
+				Reversed:          false,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     5,
+			lastIndex:      7,
+			expectedSeqNrs: []uint64{5, 6, 7},
+		},
+		{
+			name: "start at offset index outside of payments, " +
+				"reversed order",
+			query: PaymentsQuery{
+				IndexOffset:       9,
+				MaxPayments:       2,
+				Reversed:          true,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     6,
+			lastIndex:      7,
+			expectedSeqNrs: []uint64{6, 7},
+		},
+		{
+			name: "query in reverse order, start at end",
+			query: PaymentsQuery{
+				IndexOffset:       0,
+				MaxPayments:       2,
+				Reversed:          true,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     6,
+			lastIndex:      7,
+			expectedSeqNrs: []uint64{6, 7},
+		},
+		{
+			name: "query in reverse order, starting in middle",
+			query: PaymentsQuery{
+				IndexOffset:       4,
+				MaxPayments:       2,
+				Reversed:          true,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     1,
+			lastIndex:      3,
+			expectedSeqNrs: []uint64{1, 3},
+		},
+		{
+			name: "query in reverse order, starting in middle, " +
+				"with underflow",
+			query: PaymentsQuery{
+				IndexOffset:       4,
+				MaxPayments:       5,
+				Reversed:          true,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     1,
+			lastIndex:      3,
+			expectedSeqNrs: []uint64{1, 3},
+		},
+		{
+			name: "all payments in reverse, order maintained",
+			query: PaymentsQuery{
+				IndexOffset:       0,
+				MaxPayments:       7,
+				Reversed:          true,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     1,
+			lastIndex:      7,
+			expectedSeqNrs: []uint64{1, 3, 4, 5, 6, 7},
+		},
+		{
+			name: "exclude incomplete payments",
+			query: PaymentsQuery{
+				IndexOffset:       0,
+				MaxPayments:       7,
+				Reversed:          false,
+				IncludeIncomplete: false,
+			},
+			firstIndex:     0,
+			lastIndex:      0,
+			expectedSeqNrs: nil,
+		},
+		{
+			name: "query payments at index gap",
+			query: PaymentsQuery{
+				IndexOffset:       1,
+				MaxPayments:       7,
+				Reversed:          false,
+				IncludeIncomplete: true,
+			},
+			firstIndex:     3,
+			lastIndex:      7,
+			expectedSeqNrs: []uint64{3, 4, 5, 6, 7},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			db, err := initDB()
+			if err != nil {
+				t.Fatalf("unable to init db: %v", err)
+			}
+
+			// Populate the database with a set of test payments.
+			numberOfPayments := 7
+			pControl := NewPaymentControl(db)
+
+			for i := 0; i < numberOfPayments; i++ {
+				// Generate a test payment.
+				info, _, _, err := genInfo()
+				if err != nil {
+					t.Fatalf("unable to create test "+
+						"payment: %v", err)
+				}
+
+				// Create a new payment entry in the database.
+				err = pControl.InitPayment(info.PaymentHash, info)
+				if err != nil {
+					t.Fatalf("unable to initialize "+
+						"payment in database: %v", err)
+				}
+
+				// Immediately delete the payment with index 2.
+				if i == 1 {
+					deletePayment(t, db, info.PaymentHash)
+				}
+			}
+
+			// Fetch all payments in the database.
+			allPayments, err := db.FetchPayments()
+			if err != nil {
+				t.Fatalf("payments could not be fetched from "+
+					"database: %v", err)
+			}
+
+			if len(allPayments) != 6 {
+				t.Fatalf("Number of payments received does not "+
+					"match expected one. Got %v, want %v.",
+					len(allPayments), 6)
+			}
+
+			querySlice, err := db.QueryPayments(tt.query)
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if tt.firstIndex != querySlice.FirstIndexOffset ||
+				tt.lastIndex != querySlice.LastIndexOffset {
+				t.Errorf("First or last index does not match "+
+					"expected index. Want (%d, %d), got (%d, %d).",
+					tt.firstIndex, tt.lastIndex,
+					querySlice.FirstIndexOffset,
+					querySlice.LastIndexOffset)
+			}
+
+			if len(querySlice.Payments) != len(tt.expectedSeqNrs) {
+				t.Errorf("expected: %v payments, got: %v",
+					len(allPayments), len(querySlice.Payments))
+			}
+
+			for i, seqNr := range tt.expectedSeqNrs {
+				q := querySlice.Payments[i]
+				if seqNr != q.SequenceNum {
+					t.Errorf("sequence numbers do not match, "+
+						"got %v, want %v", q.SequenceNum, seqNr)
+				}
+			}
+		})
+	}
+}