paymentsdb: implement FetchInFlightPayments for sql backend

payments/db/sql_store.go

@@ -53,7 +53,7 @@ type SQLQueries interface {
 
 	FetchHtlcAttemptsForPayments(ctx context.Context, paymentIDs []int64) ([]sqlc.FetchHtlcAttemptsForPaymentsRow, error)
 	FetchHtlcAttemptResolutionsForPayments(ctx context.Context, paymentIDs []int64) ([]sqlc.FetchHtlcAttemptResolutionsForPaymentsRow, error)
-	FetchAllInflightAttempts(ctx context.Context) ([]sqlc.PaymentHtlcAttempt, error)
+	FetchAllInflightAttempts(ctx context.Context, arg sqlc.FetchAllInflightAttemptsParams) ([]sqlc.PaymentHtlcAttempt, error)
 	FetchHopsForAttempts(ctx context.Context, htlcAttemptIndices []int64) ([]sqlc.FetchHopsForAttemptsRow, error)
 
 	FetchPaymentLevelFirstHopCustomRecords(ctx context.Context, paymentIDs []int64) ([]sqlc.PaymentFirstHopCustomRecord, error)
@@ -165,8 +165,88 @@ func fetchPaymentWithCompleteData(ctx context.Context,
 	return buildPaymentFromBatchData(dbPayment, batchData)
 }
 
-// paymentsDetailsData holds all the batch-loaded data for multiple payments.
-// This does not include the core payment and intent data which is fetched
+// paymentsCompleteData holds the full payment data when batch loading base
+// payment data and all the related data for a payment.
+type paymentsCompleteData struct {
+	*paymentsBaseData
+	*paymentsDetailsData
+}
+
+// batchLoadPayments loads the full payment data for a batch of payment IDs.
+func batchLoadPayments(ctx context.Context, cfg *sqldb.QueryConfig,
+	db SQLQueries, paymentIDs []int64) (*paymentsCompleteData, error) {
+
+	baseData, err := batchLoadpaymentsBaseData(ctx, cfg, db, paymentIDs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load payment base data: %w",
+			err)
+	}
+
+	batchData, err := batchLoadPaymentDetailsData(ctx, cfg, db, paymentIDs)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load payment batch data: %w",
+			err)
+	}
+
+	return &paymentsCompleteData{
+		paymentsBaseData:    baseData,
+		paymentsDetailsData: batchData,
+	}, nil
+}
+
+// paymentsBaseData holds the base payment and intent data for a batch of
+// payments.
+type paymentsBaseData struct {
+	// paymentsAndIntents maps payment ID to its payment and intent data.
+	paymentsAndIntents map[int64]sqlc.PaymentAndIntent
+}
+
+// batchLoadpaymentsBaseData loads the base payment and payment intent data for
+// a batch of payment IDs. This complements loadPaymentsBatchData which loads
+// related data (attempts, hops, custom records) but not the payment table
+// and payment intent table data.
+func batchLoadpaymentsBaseData(ctx context.Context,
+	cfg *sqldb.QueryConfig, db SQLQueries,
+	paymentIDs []int64) (*paymentsBaseData, error) {
+
+	baseData := &paymentsBaseData{
+		paymentsAndIntents: make(map[int64]sqlc.PaymentAndIntent),
+	}
+
+	if len(paymentIDs) == 0 {
+		return baseData, nil
+	}
+
+	err := sqldb.ExecuteBatchQuery(
+		ctx, cfg, paymentIDs,
+		func(id int64) int64 { return id },
+		func(ctx context.Context, ids []int64) (
+			[]sqlc.FetchPaymentsByIDsRow, error) {
+
+			records, err := db.FetchPaymentsByIDs(
+				ctx, ids,
+			)
+
+			return records, err
+		},
+		func(ctx context.Context,
+			payment sqlc.FetchPaymentsByIDsRow) error {
+
+			baseData.paymentsAndIntents[payment.ID] = payment
+
+			return nil
+		},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch payment base "+
+			"data: %w", err)
+	}
+
+	return baseData, nil
+}
+
+// paymentsRelatedData holds all the batch-loaded data for multiple payments.
+// This does not include the base payment and intent data which is fetched
 // separately. It includes the additional data like attempts, hops, hop custom
 // records, and route custom records.
 type paymentsDetailsData struct {
@@ -874,6 +954,141 @@ func (s *SQLStore) FetchPayment(paymentHash lntypes.Hash) (*MPPayment, error) {
 	return mpPayment, nil
 }
 
+// FetchInFlightPayments retrieves all payments that have HTLC attempts
+// currently in flight (not yet settled or failed). These are payments with at
+// least one HTLC attempt that has been registered but has no resolution record.
+//
+// The SQLStore implementation provides a significant performance improvement
+// over the KVStore implementation by using targeted SQL queries instead of
+// scanning all payments.
+//
+// This method is part of the PaymentReader interface, which is embedded in the
+// DB interface. It's typically called during node startup to resume monitoring
+// of pending payments and ensure HTLCs are properly tracked.
+//
+// TODO(ziggie): Consider changing the interface to use a callback or iterator
+// pattern instead of returning all payments at once. This would allow
+// processing payments one at a time without holding them all in memory
+// simultaneously:
+//   - Callback: func FetchInFlightPayments(ctx, func(*MPPayment) error) error
+//   - Iterator: func FetchInFlightPayments(ctx) (PaymentIterator, error)
+//
+// While inflight payments are typically a small subset, this would improve
+// memory efficiency for nodes with unusually high numbers of concurrent
+// payments and would better leverage the existing pagination infrastructure.
+func (s *SQLStore) FetchInFlightPayments() ([]*MPPayment,
+	error) {
+
+	ctx := context.TODO()
+
+	var mpPayments []*MPPayment
+
+	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+		// Track which payment IDs we've already processed across all
+		// pages to avoid loading the same payment multiple times when
+		// multiple inflight attempts belong to the same payment.
+		processedPayments := make(map[int64]*MPPayment)
+
+		extractCursor := func(row sqlc.PaymentHtlcAttempt) int64 {
+			return row.AttemptIndex
+		}
+
+		// collectFunc extracts the payment ID from each attempt row.
+		collectFunc := func(row sqlc.PaymentHtlcAttempt) (
+			int64, error) {
+
+			return row.PaymentID, nil
+		}
+
+		// batchDataFunc loads payment data for a batch of payment IDs,
+		// but only for IDs we haven't processed yet.
+		batchDataFunc := func(ctx context.Context,
+			paymentIDs []int64) (*paymentsCompleteData, error) {
+
+			// Filter out already-processed payment IDs.
+			uniqueIDs := make([]int64, 0, len(paymentIDs))
+			for _, id := range paymentIDs {
+				_, processed := processedPayments[id]
+				if !processed {
+					uniqueIDs = append(uniqueIDs, id)
+				}
+			}
+
+			// If uniqueIDs is empty, the batch load will return
+			// empty batch data.
+			return batchLoadPayments(
+				ctx, s.cfg.QueryCfg, db, uniqueIDs,
+			)
+		}
+
+		// processAttempt processes each attempt. We only build and
+		// store the payment once per unique payment ID.
+		processAttempt := func(ctx context.Context,
+			row sqlc.PaymentHtlcAttempt,
+			batchData *paymentsCompleteData) error {
+
+			// Skip if we've already processed this payment.
+			_, processed := processedPayments[row.PaymentID]
+			if processed {
+				return nil
+			}
+
+			dbPayment := batchData.paymentsAndIntents[row.PaymentID]
+
+			// Build the payment from batch data.
+			mpPayment, err := buildPaymentFromBatchData(
+				dbPayment, batchData.paymentsDetailsData,
+			)
+			if err != nil {
+				return fmt.Errorf("failed to build payment: %w",
+					err)
+			}
+
+			// Store in our processed map.
+			processedPayments[row.PaymentID] = mpPayment
+
+			return nil
+		}
+
+		queryFunc := func(ctx context.Context, lastAttemptIndex int64,
+			limit int32) ([]sqlc.PaymentHtlcAttempt,
+			error) {
+
+			return db.FetchAllInflightAttempts(ctx,
+				sqlc.FetchAllInflightAttemptsParams{
+					AttemptIndex: lastAttemptIndex,
+					Limit:        limit,
+				},
+			)
+		}
+
+		err := sqldb.ExecuteCollectAndBatchWithSharedDataQuery(
+			ctx, s.cfg.QueryCfg, int64(-1), queryFunc,
+			extractCursor, collectFunc, batchDataFunc,
+			processAttempt,
+		)
+		if err != nil {
+			return err
+		}
+
+		// Convert map to slice.
+		mpPayments = make([]*MPPayment, 0, len(processedPayments))
+		for _, payment := range processedPayments {
+			mpPayments = append(mpPayments, payment)
+		}
+
+		return nil
+	}, func() {
+		mpPayments = nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch inflight "+
+			"payments: %w", err)
+	}
+
+	return mpPayments, nil
+}
+
 // DeleteFailedAttempts removes all failed HTLC attempts from the database for
 // the specified payment, while preserving the payment record itself and any
 // successful or in-flight attempts.

sqldb/sqlc/db_custom.go

@@ -222,18 +222,16 @@ func (r FetchPaymentRow) GetPaymentIntent() PaymentIntent {
 	}
 }
 
-// GetPayment returns the Payment associated with this interface.
-//
-// NOTE: This method is part of the PaymentAndIntent interface.
 func (r FetchPaymentsByIDsRow) GetPayment() Payment {
-	return r.Payment
+	return Payment{
+		ID:                r.ID,
+		AmountMsat:        r.AmountMsat,
+		CreatedAt:         r.CreatedAt,
+		PaymentIdentifier: r.PaymentIdentifier,
+		FailReason:        r.FailReason,
+	}
 }
 
-// GetPaymentIntent returns the PaymentIntent associated with this payment.
-// If the payment has no intent (IntentType is NULL), this returns a zero-value
-// PaymentIntent.
-//
-// NOTE: This method is part of the PaymentAndIntent interface.
 func (r FetchPaymentsByIDsRow) GetPaymentIntent() PaymentIntent {
 	if !r.IntentType.Valid {
 		return PaymentIntent{}

sqldb/sqlc/payments.sql.go

@@ -115,12 +115,20 @@ WHERE NOT EXISTS (
     SELECT 1 FROM payment_htlc_attempt_resolutions hr
     WHERE hr.attempt_index = ha.attempt_index
 )
+AND ha.attempt_index > $1
 ORDER BY ha.attempt_index ASC
+LIMIT $2
 `
 
-// Fetch all inflight attempts across all payments
-func (q *Queries) FetchAllInflightAttempts(ctx context.Context) ([]PaymentHtlcAttempt, error) {
-	rows, err := q.db.QueryContext(ctx, fetchAllInflightAttempts)
+type FetchAllInflightAttemptsParams struct {
+	AttemptIndex int64
+	Limit        int32
+}
+
+// Fetch all inflight attempts with their payment data using pagination.
+// Returns attempt data joined with payment and intent data to avoid separate queries.
+func (q *Queries) FetchAllInflightAttempts(ctx context.Context, arg FetchAllInflightAttemptsParams) ([]PaymentHtlcAttempt, error) {
+	rows, err := q.db.QueryContext(ctx, fetchAllInflightAttempts, arg.AttemptIndex, arg.Limit)
 	if err != nil {
 		return nil, err
 	}
@@ -522,20 +530,32 @@ func (q *Queries) FetchPaymentLevelFirstHopCustomRecords(ctx context.Context, pa
 
 const fetchPaymentsByIDs = `-- name: FetchPaymentsByIDs :many
 SELECT
-    p.id, p.amount_msat, p.created_at, p.payment_identifier, p.fail_reason,
-    i.intent_type AS "intent_type",
-    i.intent_payload AS "intent_payload"
+    p.id,
+    p.amount_msat,
+    p.created_at,
+    p.payment_identifier,
+    p.fail_reason,
+    pi.intent_type,
+    pi.intent_payload
 FROM payments p
-LEFT JOIN payment_intents i ON i.payment_id = p.id
+LEFT JOIN payment_intents pi ON pi.payment_id = p.id
 WHERE p.id IN (/*SLICE:payment_ids*/?)
+ORDER BY p.id ASC
 `
 
 type FetchPaymentsByIDsRow struct {
-	Payment       Payment
-	IntentType    sql.NullInt16
-	IntentPayload []byte
+	ID                int64
+	AmountMsat        int64
+	CreatedAt         time.Time
+	PaymentIdentifier []byte
+	FailReason        sql.NullInt32
+	IntentType        sql.NullInt16
+	IntentPayload     []byte
 }
 
+// Batch fetch payment and intent data for a set of payment IDs.
+// Used to avoid fetching redundant payment data when processing multiple
+// attempts for the same payment.
 func (q *Queries) FetchPaymentsByIDs(ctx context.Context, paymentIds []int64) ([]FetchPaymentsByIDsRow, error) {
 	query := fetchPaymentsByIDs
 	var queryParams []interface{}
@@ -556,11 +576,11 @@ func (q *Queries) FetchPaymentsByIDs(ctx context.Context, paymentIds []int64) ([
 	for rows.Next() {
 		var i FetchPaymentsByIDsRow
 		if err := rows.Scan(
-			&i.Payment.ID,
-			&i.Payment.AmountMsat,
-			&i.Payment.CreatedAt,
-			&i.Payment.PaymentIdentifier,
-			&i.Payment.FailReason,
+			&i.ID,
+			&i.AmountMsat,
+			&i.CreatedAt,
+			&i.PaymentIdentifier,
+			&i.FailReason,
 			&i.IntentType,
 			&i.IntentPayload,
 		); err != nil {

sqldb/sqlc/querier.go

@@ -38,8 +38,9 @@ type Querier interface {
 	FailPayment(ctx context.Context, arg FailPaymentParams) (sql.Result, error)
 	FetchAMPSubInvoiceHTLCs(ctx context.Context, arg FetchAMPSubInvoiceHTLCsParams) ([]FetchAMPSubInvoiceHTLCsRow, error)
 	FetchAMPSubInvoices(ctx context.Context, arg FetchAMPSubInvoicesParams) ([]AmpSubInvoice, error)
-	// Fetch all inflight attempts across all payments
-	FetchAllInflightAttempts(ctx context.Context) ([]PaymentHtlcAttempt, error)
+	// Fetch all inflight attempts with their payment data using pagination.
+	// Returns attempt data joined with payment and intent data to avoid separate queries.
+	FetchAllInflightAttempts(ctx context.Context, arg FetchAllInflightAttemptsParams) ([]PaymentHtlcAttempt, error)
 	FetchHopLevelCustomRecords(ctx context.Context, hopIds []int64) ([]PaymentHopCustomRecord, error)
 	FetchHopsForAttempts(ctx context.Context, htlcAttemptIndices []int64) ([]FetchHopsForAttemptsRow, error)
 	// Batch query to fetch only HTLC resolution status for multiple payments.
@@ -49,6 +50,9 @@ type Querier interface {
 	FetchHtlcAttemptsForPayments(ctx context.Context, paymentIds []int64) ([]FetchHtlcAttemptsForPaymentsRow, error)
 	FetchPayment(ctx context.Context, paymentIdentifier []byte) (FetchPaymentRow, error)
 	FetchPaymentLevelFirstHopCustomRecords(ctx context.Context, paymentIds []int64) ([]PaymentFirstHopCustomRecord, error)
+	// Batch fetch payment and intent data for a set of payment IDs.
+	// Used to avoid fetching redundant payment data when processing multiple
+	// attempts for the same payment.
 	FetchPaymentsByIDs(ctx context.Context, paymentIds []int64) ([]FetchPaymentsByIDsRow, error)
 	// FetchPendingInvoices returns all invoices in a pending state (open or
 	// accepted). The invoices_state_idx index on the state column makes this a

sqldb/sqlc/queries/payments.sql

@@ -40,15 +40,6 @@ FROM payments p
 LEFT JOIN payment_intents i ON i.payment_id = p.id
 WHERE p.payment_identifier = $1;
 
--- name: FetchPaymentsByIDs :many
-SELECT
-    sqlc.embed(p),
-    i.intent_type AS "intent_type",
-    i.intent_payload AS "intent_payload"
-FROM payments p
-LEFT JOIN payment_intents i ON i.payment_id = p.id
-WHERE p.id IN (sqlc.slice('payment_ids')/*SLICE:payment_ids*/);
-
 -- name: CountPayments :one
 SELECT COUNT(*) FROM payments;
 
@@ -86,8 +77,26 @@ FROM payment_htlc_attempts ha
 LEFT JOIN payment_htlc_attempt_resolutions hr ON hr.attempt_index = ha.attempt_index
 WHERE ha.payment_id IN (sqlc.slice('payment_ids')/*SLICE:payment_ids*/);
 
+-- name: FetchPaymentsByIDs :many
+-- Batch fetch payment and intent data for a set of payment IDs.
+-- Used to avoid fetching redundant payment data when processing multiple
+-- attempts for the same payment.
+SELECT
+    p.id,
+    p.amount_msat,
+    p.created_at,
+    p.payment_identifier,
+    p.fail_reason,
+    pi.intent_type,
+    pi.intent_payload
+FROM payments p
+LEFT JOIN payment_intents pi ON pi.payment_id = p.id
+WHERE p.id IN (sqlc.slice('payment_ids')/*SLICE:payment_ids*/)
+ORDER BY p.id ASC;
+
 -- name: FetchAllInflightAttempts :many
--- Fetch all inflight attempts across all payments
+-- Fetch all inflight attempts with their payment data using pagination.
+-- Returns attempt data joined with payment and intent data to avoid separate queries.
 SELECT
     ha.id,
     ha.attempt_index,
@@ -104,7 +113,9 @@ WHERE NOT EXISTS (
     SELECT 1 FROM payment_htlc_attempt_resolutions hr
     WHERE hr.attempt_index = ha.attempt_index
 )
-ORDER BY ha.attempt_index ASC;
+AND ha.attempt_index > $1
+ORDER BY ha.attempt_index ASC
+LIMIT $2;
 
 -- name: FetchHopsForAttempts :many
 SELECT