paymentsdb: implement FetchInFlightPayments for sql backend
What changed, and why it matters
This commit adds a missing database method for the new SQL backend in LND. It lets the node efficiently find all payments that still have HTLC attempts in flight during startup. There is no security fix here; it is a feature/performance parity change to match the older key-value store backend.
No security action required. Review as normal code-quality/performance change.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The patch implements FetchInFlightPayments for the SQL payments store. It paginates through payment_htlc_attempts rows that lack a resolution record, batch-loads the related payment/intent and attempt/hop/custom-record data, and returns full MPPayment objects. The SQL query FetchAllInflightAttempts gains cursor-based pagination (attempt_index > $1 LIMIT $2), and FetchPaymentsByIDs is rewritten to return a flat row instead of an embedded Payment struct so it can be reused by the new batch loader. No access controls, cryptographic operations, or trust boundaries are changed.
Changed components
payments/db/sql_store.gosqldb/sqlc/db_custom.gosqldb/sqlc/payments.sql.gosqldb/sqlc/querier.gosqldb/sqlc/queries/payments.sqlInspect captured patch +288 / −40
diff --git a/payments/db/sql_store.go b/payments/db/sql_store.go
index 759800a..e7aab1c 100644
--- a/payments/db/sql_store.go
+++ b/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.
diff --git a/sqldb/sqlc/db_custom.go b/sqldb/sqlc/db_custom.go
index 7888f81..1b8d465 100644
--- a/sqldb/sqlc/db_custom.go
+++ b/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{}
diff --git a/sqldb/sqlc/payments.sql.go b/sqldb/sqlc/payments.sql.go
index d3a7364..b9ec314 100644
--- a/sqldb/sqlc/payments.sql.go
+++ b/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 {
diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go
index 05c9c81..3be738b 100644
--- a/sqldb/sqlc/querier.go
+++ b/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
diff --git a/sqldb/sqlc/queries/payments.sql b/sqldb/sqlc/queries/payments.sql
index 7fb979a..419f7bf 100644
--- a/sqldb/sqlc/queries/payments.sql
+++ b/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
Why this scored 12/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.