From 969b00e4a5497c463dcbafa71ed3312127729297 Mon Sep 17 00:00:00 2001 From: ziggie Date: Wed, 15 Oct 2025 18:23:55 +0200 Subject: [PATCH] paymentsdb: implement FetchInFlightPayments for sql backend --- payments/db/sql_store.go | 221 +++++++++++++++++++++++++++++++- sqldb/sqlc/db_custom.go | 16 +-- sqldb/sqlc/payments.sql.go | 50 +++++--- sqldb/sqlc/querier.go | 8 +- sqldb/sqlc/queries/payments.sql | 33 +++-- 5 files changed, 288 insertions(+), 40 deletions(-) diff --git a/payments/db/sql_store.go b/payments/db/sql_store.go index 759800aec..e7aab1cfb 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 7888f81f4..1b8d465e7 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 d3a7364c2..b9ec31499 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 05c9c81a5..3be738ba9 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 7fb979ac6..419f7bf1a 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