Merge pull request #10196 from ziggie1984/refactor-payment-part-6

refactor payments part 6

Author: yyforyongyu

payments/db/kv_store.go

@@ -133,18 +133,14 @@ type KVStore struct {
 	keepFailedPaymentAttempts bool
 }
 
-// defaultKVStoreOptions returns the default options for the KV store.
-func defaultKVStoreOptions() *StoreOptions {
-	return &StoreOptions{
-		KeepFailedPaymentAttempts: false,
-	}
-}
+// A compile-time constraint to ensure KVStore implements DB.
+var _ DB = (*KVStore)(nil)
 
 // NewKVStore creates a new KVStore for payments.
 func NewKVStore(db kvdb.Backend,
 	options ...OptionModifier) (*KVStore, error) {
 
-	opts := defaultKVStoreOptions()
+	opts := DefaultOptions()
 	for _, applyOption := range options {
 		applyOption(opts)
 	}
@@ -393,88 +389,9 @@ func (p *KVStore) RegisterAttempt(paymentHash lntypes.Hash,
 			return err
 		}
 
-		// If the final hop has encrypted data, then we know this is a
-		// blinded payment. In blinded payments, MPP records are not set
-		// for split payments and the recipient is responsible for using
-		// a consistent PathID across the various encrypted data
-		// payloads that we received from them for this payment. All we
-		// need to check is that the total amount field for each HTLC
-		// in the split payment is correct.
-		isBlinded := len(attempt.Route.FinalHop().EncryptedData) != 0
-
-		// Make sure any existing shards match the new one with regards
-		// to MPP options.
-		mpp := attempt.Route.FinalHop().MPP
-
-		// MPP records should not be set for attempts to blinded paths.
-		if isBlinded && mpp != nil {
-			return ErrMPPRecordInBlindedPayment
-		}
-
-		for _, h := range payment.InFlightHTLCs() {
-			hMpp := h.Route.FinalHop().MPP
-
-			// If this is a blinded payment, then no existing HTLCs
-			// should have MPP records.
-			if isBlinded && hMpp != nil {
-				return ErrMPPRecordInBlindedPayment
-			}
-
-			// If this is a blinded payment, then we just need to
-			// check that the TotalAmtMsat field for this shard
-			// is equal to that of any other shard in the same
-			// payment.
-			if isBlinded {
-				if attempt.Route.FinalHop().TotalAmtMsat !=
-					h.Route.FinalHop().TotalAmtMsat {
-
-					//nolint:ll
-					return ErrBlindedPaymentTotalAmountMismatch
-				}
-
-				continue
-			}
-
-			switch {
-			// We tried to register a non-MPP attempt for a MPP
-			// payment.
-			case mpp == nil && hMpp != nil:
-				return ErrMPPayment
-
-			// We tried to register a MPP shard for a non-MPP
-			// payment.
-			case mpp != nil && hMpp == nil:
-				return ErrNonMPPayment
-
-			// Non-MPP payment, nothing more to validate.
-			case mpp == nil:
-				continue
-			}
-
-			// Check that MPP options match.
-			if mpp.PaymentAddr() != hMpp.PaymentAddr() {
-				return ErrMPPPaymentAddrMismatch
-			}
-
-			if mpp.TotalMsat() != hMpp.TotalMsat() {
-				return ErrMPPTotalAmountMismatch
-			}
-		}
-
-		// If this is a non-MPP attempt, it must match the total amount
-		// exactly. Note that a blinded payment is considered an MPP
-		// attempt.
-		amt := attempt.Route.ReceiverAmt()
-		if !isBlinded && mpp == nil && amt != payment.Info.Value {
-			return ErrValueMismatch
-		}
-
-		// Ensure we aren't sending more than the total payment amount.
-		sentAmt, _ := payment.SentAmt()
-		if sentAmt+amt > payment.Info.Value {
-			return fmt.Errorf("%w: attempted=%v, payment amount="+
-				"%v", ErrValueExceedsAmt,
-				sentAmt+amt, payment.Info.Value)
+		// Verify the attempt is compatible with the existing payment.
+		if err := verifyAttempt(payment, attempt); err != nil {
+			return err
 		}
 
 		htlcsBucket, err := bucket.CreateBucketIfNotExists(

payments/db/options.go

@@ -10,6 +10,14 @@ type StoreOptions struct {
 	KeepFailedPaymentAttempts bool
 }
 
+// DefaultOptions returns a StoreOptions populated with default values.
+func DefaultOptions() *StoreOptions {
+	return &StoreOptions{
+		KeepFailedPaymentAttempts: false,
+		NoMigration:               false,
+	}
+}
+
 // OptionModifier is a function signature for modifying the default
 // StoreOptions.
 type OptionModifier func(*StoreOptions)

payments/db/payment.go

@@ -175,6 +175,20 @@ func (h *HTLCAttemptInfo) SessionKey() *btcec.PrivateKey {
 	return h.cachedSessionKey
 }
 
+// setSessionKey sets the session key for the htlc attempt.
+//
+// NOTE: Only used for testing.
+//
+//nolint:unused
+func (h *HTLCAttemptInfo) setSessionKey(sessionKey *btcec.PrivateKey) {
+	h.cachedSessionKey = sessionKey
+
+	// Also set the session key as a raw bytes.
+	var scratch [btcec.PrivKeyBytesLen]byte
+	copy(scratch[:], sessionKey.Serialize())
+	h.sessionKey = scratch
+}
+
 // OnionBlob returns the onion blob created from the sphinx construction.
 func (h *HTLCAttemptInfo) OnionBlob() ([lnwire.OnionPacketSize]byte, error) {
 	var zeroBytes [lnwire.OnionPacketSize]byte
@@ -712,3 +726,95 @@ func generateSphinxPacket(rt *route.Route, paymentHash []byte,
 		PaymentPath: sphinxPath.NodeKeys(),
 	}, nil
 }
+
+// verifyAttempt validates that a new HTLC attempt is compatible with the
+// existing payment and its in-flight HTLCs. This function checks:
+//  1. MPP (Multi-Path Payment) compatibility between attempts
+//  2. Blinded payment consistency
+//  3. Amount validation
+//  4. Total payment amount limits
+func verifyAttempt(payment *MPPayment, attempt *HTLCAttemptInfo) error {
+	// If the final hop has encrypted data, then we know this is a
+	// blinded payment. In blinded payments, MPP records are not set
+	// for split payments and the recipient is responsible for using
+	// a consistent PathID across the various encrypted data
+	// payloads that we received from them for this payment. All we
+	// need to check is that the total amount field for each HTLC
+	// in the split payment is correct.
+	isBlinded := len(attempt.Route.FinalHop().EncryptedData) != 0
+
+	// Make sure any existing shards match the new one with regards
+	// to MPP options.
+	mpp := attempt.Route.FinalHop().MPP
+
+	// MPP records should not be set for attempts to blinded paths.
+	if isBlinded && mpp != nil {
+		return ErrMPPRecordInBlindedPayment
+	}
+
+	for _, h := range payment.InFlightHTLCs() {
+		hMpp := h.Route.FinalHop().MPP
+
+		// If this is a blinded payment, then no existing HTLCs
+		// should have MPP records.
+		if isBlinded && hMpp != nil {
+			return ErrMPPRecordInBlindedPayment
+		}
+
+		// If this is a blinded payment, then we just need to
+		// check that the TotalAmtMsat field for this shard
+		// is equal to that of any other shard in the same
+		// payment.
+		if isBlinded {
+			if attempt.Route.FinalHop().TotalAmtMsat !=
+				h.Route.FinalHop().TotalAmtMsat {
+
+				return ErrBlindedPaymentTotalAmountMismatch
+			}
+
+			continue
+		}
+
+		switch {
+		// We tried to register a non-MPP attempt for a MPP
+		// payment.
+		case mpp == nil && hMpp != nil:
+			return ErrMPPayment
+
+		// We tried to register a MPP shard for a non-MPP
+		// payment.
+		case mpp != nil && hMpp == nil:
+			return ErrNonMPPayment
+
+		// Non-MPP payment, nothing more to validate.
+		case mpp == nil:
+			continue
+		}
+
+		// Check that MPP options match.
+		if mpp.PaymentAddr() != hMpp.PaymentAddr() {
+			return ErrMPPPaymentAddrMismatch
+		}
+
+		if mpp.TotalMsat() != hMpp.TotalMsat() {
+			return ErrMPPTotalAmountMismatch
+		}
+	}
+
+	// If this is a non-MPP attempt, it must match the total amount
+	// exactly. Note that a blinded payment is considered an MPP
+	// attempt.
+	amt := attempt.Route.ReceiverAmt()
+	if !isBlinded && mpp == nil && amt != payment.Info.Value {
+		return ErrValueMismatch
+	}
+
+	// Ensure we aren't sending more than the total payment amount.
+	sentAmt, _ := payment.SentAmt()
+	if sentAmt+amt > payment.Info.Value {
+		return fmt.Errorf("%w: attempted=%v, payment amount=%v",
+			ErrValueExceedsAmt, sentAmt+amt, payment.Info.Value)
+	}
+
+	return nil
+}

payments/db/query.go

@@ -1,5 +1,11 @@
 package paymentsdb
 
+const (
+	// DefaultMaxPayments is the default maximum number of payments returned
+	// in the payments query pagination.
+	DefaultMaxPayments = 100
+)
+
 // Query represents a query to the payments database starting or ending
 // at a certain offset index. The number of retrieved records can be limited.
 type Query struct {

rpcserver.go

@@ -7523,10 +7523,15 @@ func (r *rpcServer) ListPayments(ctx context.Context,
 		CreationDateEnd:   int64(req.CreationDateEnd),
 	}
 
-	// If the maximum number of payments wasn't specified, then we'll
-	// default to return the maximal number of payments representable.
+	// If the maximum number of payments wasn't specified, we default to
+	// a reasonable number to prevent resource exhaustion. All of the
+	// payments are fetched into memory. Moreover we don't want our daemon
+	// to remain stable and do other stuff rather than serving payments.
+	//
+	// TODO(ziggie): Choose a more specific default value when results of
+	// performance testing are available.
 	if req.MaxPayments == 0 {
-		query.MaxPayments = math.MaxUint64
+		query.MaxPayments = paymentsdb.DefaultMaxPayments
 	}
 
 	paymentsQuerySlice, err := r.server.paymentsDB.QueryPayments(

sqldb/postgres.go

@@ -30,7 +30,10 @@ var (
 	postgresSchemaReplacements = map[string]string{
 		"BLOB":                "BYTEA",
 		"INTEGER PRIMARY KEY": "BIGSERIAL PRIMARY KEY",
-		"TIMESTAMP":           "TIMESTAMP WITHOUT TIME ZONE",
+		// We need this space in front of the TIMESTAMP keyword to
+		// avoid replacing words which just have the word "TIMESTAMP" in
+		// them.
+		" TIMESTAMP": " TIMESTAMP WITHOUT TIME ZONE",
 	}
 
 	// Make sure PostgresStore implements the MigrationExecutor interface.

sqldb/sqlutils.go

@@ -75,6 +75,15 @@ func SQLStrValid(s string) sql.NullString {
 	}
 }
 
+// SQLBool turns a boolean into the NullBool that sql/sqlc uses when a boolean
+// can be permitted to be NULL.
+func SQLBool(b bool) sql.NullBool {
+	return sql.NullBool{
+		Bool:  b,
+		Valid: true,
+	}
+}
+
 // SQLTime turns a time.Time into the NullTime that sql/sqlc uses when a time
 // can be permitted to be NULL.
 func SQLTime(t time.Time) sql.NullTime {