multi: thread context through FetchInflightPayments
This commit is contained in:
7 files changed
+25
-16
No files matched your search
@@ -25,7 +25,7 @@ type PaymentReader interface {
|
|||||||
paymentHash lntypes.Hash) (*MPPayment, error)
|
paymentHash lntypes.Hash) (*MPPayment, error)
|
||||||
|
|
||||||
// FetchInFlightPayments returns all payments with status InFlight.
|
// FetchInFlightPayments returns all payments with status InFlight.
|
||||||
FetchInFlightPayments() ([]*MPPayment, error)
|
FetchInFlightPayments(ctx context.Context) ([]*MPPayment, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
// PaymentWriter represents the interface to write operations to the payments
|
// PaymentWriter represents the interface to write operations to the payments
|
||||||
|
|||||||
@@ -741,7 +741,9 @@ func fetchPaymentStatus(bucket kvdb.RBucket) (PaymentStatus, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// FetchInFlightPayments returns all payments with status InFlight.
|
// FetchInFlightPayments returns all payments with status InFlight.
|
||||||
func (p *KVStore) FetchInFlightPayments() ([]*MPPayment, error) {
|
func (p *KVStore) FetchInFlightPayments(_ context.Context) ([]*MPPayment,
|
||||||
|
error) {
|
||||||
|
|
||||||
var (
|
var (
|
||||||
inFlights []*MPPayment
|
inFlights []*MPPayment
|
||||||
start = time.Now()
|
start = time.Now()
|
||||||
|
|||||||
@@ -2740,6 +2740,8 @@ func TestQueryPayments(t *testing.T) {
|
|||||||
func TestFetchInFlightPayments(t *testing.T) {
|
func TestFetchInFlightPayments(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := t.Context()
|
||||||
|
|
||||||
paymentDB, _ := NewTestDB(t)
|
paymentDB, _ := NewTestDB(t)
|
||||||
|
|
||||||
// Register payments with different statuses:
|
// Register payments with different statuses:
|
||||||
@@ -2765,7 +2767,7 @@ func TestFetchInFlightPayments(t *testing.T) {
|
|||||||
assertDBPayments(t, paymentDB, payments)
|
assertDBPayments(t, paymentDB, payments)
|
||||||
|
|
||||||
// Fetch in-flight payments.
|
// Fetch in-flight payments.
|
||||||
inFlightPayments, err := paymentDB.FetchInFlightPayments()
|
inFlightPayments, err := paymentDB.FetchInFlightPayments(ctx)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// We should only get the two in-flight payments.
|
// We should only get the two in-flight payments.
|
||||||
@@ -2795,7 +2797,7 @@ func TestFetchInFlightPayments(t *testing.T) {
|
|||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// Fetch in-flight payments again.
|
// Fetch in-flight payments again.
|
||||||
inFlightPayments, err = paymentDB.FetchInFlightPayments()
|
inFlightPayments, err = paymentDB.FetchInFlightPayments(ctx)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// We should now only get one in-flight payment.
|
// We should now only get one in-flight payment.
|
||||||
@@ -2812,6 +2814,8 @@ func TestFetchInFlightPayments(t *testing.T) {
|
|||||||
func TestFetchInFlightPaymentsMultipleAttempts(t *testing.T) {
|
func TestFetchInFlightPaymentsMultipleAttempts(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
ctx := t.Context()
|
||||||
|
|
||||||
paymentDB, _ := NewTestDB(t)
|
paymentDB, _ := NewTestDB(t)
|
||||||
|
|
||||||
preimg, err := genPreimage(t)
|
preimg, err := genPreimage(t)
|
||||||
@@ -2843,7 +2847,7 @@ func TestFetchInFlightPaymentsMultipleAttempts(t *testing.T) {
|
|||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// Both attempts are in-flight. Fetch in-flight payments.
|
// Both attempts are in-flight. Fetch in-flight payments.
|
||||||
inFlightPayments, err := paymentDB.FetchInFlightPayments()
|
inFlightPayments, err := paymentDB.FetchInFlightPayments(ctx)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// We should only get one payment even though it has 2 in-flight
|
// We should only get one payment even though it has 2 in-flight
|
||||||
|
|||||||
@@ -972,11 +972,9 @@ func (s *SQLStore) FetchPayment(ctx context.Context,
|
|||||||
// While inflight payments are typically a small subset, this would improve
|
// While inflight payments are typically a small subset, this would improve
|
||||||
// memory efficiency for nodes with unusually high numbers of concurrent
|
// memory efficiency for nodes with unusually high numbers of concurrent
|
||||||
// payments and would better leverage the existing pagination infrastructure.
|
// payments and would better leverage the existing pagination infrastructure.
|
||||||
func (s *SQLStore) FetchInFlightPayments() ([]*MPPayment,
|
func (s *SQLStore) FetchInFlightPayments(ctx context.Context) ([]*MPPayment,
|
||||||
error) {
|
error) {
|
||||||
|
|
||||||
ctx := context.TODO()
|
|
||||||
|
|
||||||
var mpPayments []*MPPayment
|
var mpPayments []*MPPayment
|
||||||
|
|
||||||
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
||||||
|
|||||||
@@ -67,7 +67,8 @@ type ControlTower interface {
|
|||||||
FailPayment(lntypes.Hash, paymentsdb.FailureReason) error
|
FailPayment(lntypes.Hash, paymentsdb.FailureReason) error
|
||||||
|
|
||||||
// FetchInFlightPayments returns all payments with status InFlight.
|
// FetchInFlightPayments returns all payments with status InFlight.
|
||||||
FetchInFlightPayments() ([]*paymentsdb.MPPayment, error)
|
FetchInFlightPayments(ctx context.Context) ([]*paymentsdb.MPPayment,
|
||||||
|
error)
|
||||||
|
|
||||||
// SubscribePayment subscribes to updates for the payment with the given
|
// SubscribePayment subscribes to updates for the payment with the given
|
||||||
// hash. A first update with the current state of the payment is always
|
// hash. A first update with the current state of the payment is always
|
||||||
@@ -286,10 +287,10 @@ func (p *controlTower) FailPayment(paymentHash lntypes.Hash,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// FetchInFlightPayments returns all payments with status InFlight.
|
// FetchInFlightPayments returns all payments with status InFlight.
|
||||||
func (p *controlTower) FetchInFlightPayments() ([]*paymentsdb.MPPayment,
|
func (p *controlTower) FetchInFlightPayments(
|
||||||
error) {
|
ctx context.Context) ([]*paymentsdb.MPPayment, error) {
|
||||||
|
|
||||||
return p.db.FetchInFlightPayments()
|
return p.db.FetchInFlightPayments(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
// SubscribePayment subscribes to updates for the payment with the given hash. A
|
// SubscribePayment subscribes to updates for the payment with the given hash. A
|
||||||
@@ -342,6 +343,8 @@ func (p *controlTower) SubscribePayment(paymentHash lntypes.Hash) (
|
|||||||
func (p *controlTower) SubscribeAllPayments() (ControlTowerSubscriber, error) {
|
func (p *controlTower) SubscribeAllPayments() (ControlTowerSubscriber, error) {
|
||||||
subscriber := newControlTowerSubscriber()
|
subscriber := newControlTowerSubscriber()
|
||||||
|
|
||||||
|
ctx := context.TODO()
|
||||||
|
|
||||||
// Add the subscriber to the list before fetching in-flight payments, so
|
// Add the subscriber to the list before fetching in-flight payments, so
|
||||||
// no events are missed. If a payment attempt update occurs after
|
// no events are missed. If a payment attempt update occurs after
|
||||||
// appending and before fetching in-flight payments, an out-of-order
|
// appending and before fetching in-flight payments, an out-of-order
|
||||||
@@ -353,7 +356,7 @@ func (p *controlTower) SubscribeAllPayments() (ControlTowerSubscriber, error) {
|
|||||||
p.subscribersMtx.Unlock()
|
p.subscribersMtx.Unlock()
|
||||||
|
|
||||||
log.Debugf("Scanning for inflight payments")
|
log.Debugf("Scanning for inflight payments")
|
||||||
inflightPayments, err := p.db.FetchInFlightPayments()
|
inflightPayments, err := p.db.FetchInFlightPayments(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -546,7 +546,7 @@ func (m *mockControlTowerOld) fetchPayment(phash lntypes.Hash) (
|
|||||||
return mp, nil
|
return mp, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *mockControlTowerOld) FetchInFlightPayments() (
|
func (m *mockControlTowerOld) FetchInFlightPayments(_ context.Context) (
|
||||||
[]*paymentsdb.MPPayment, error) {
|
[]*paymentsdb.MPPayment, error) {
|
||||||
|
|
||||||
if m.fetchInFlight != nil {
|
if m.fetchInFlight != nil {
|
||||||
@@ -801,7 +801,7 @@ func (m *mockControlTower) FetchPayment(_ context.Context,
|
|||||||
return payment, args.Error(1)
|
return payment, args.Error(1)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *mockControlTower) FetchInFlightPayments() (
|
func (m *mockControlTower) FetchInFlightPayments(_ context.Context) (
|
||||||
[]*paymentsdb.MPPayment, error) {
|
[]*paymentsdb.MPPayment, error) {
|
||||||
|
|
||||||
args := m.Called()
|
args := m.Called()
|
||||||
|
|||||||
+3
-1
@@ -1417,9 +1417,11 @@ func (r *ChannelRouter) BuildRoute(amt fn.Option[lnwire.MilliSatoshi],
|
|||||||
// resumePayments fetches inflight payments and resumes their payment
|
// resumePayments fetches inflight payments and resumes their payment
|
||||||
// lifecycles.
|
// lifecycles.
|
||||||
func (r *ChannelRouter) resumePayments() error {
|
func (r *ChannelRouter) resumePayments() error {
|
||||||
|
ctx := context.TODO()
|
||||||
|
|
||||||
// Get all payments that are inflight.
|
// Get all payments that are inflight.
|
||||||
log.Debugf("Scanning for inflight payments")
|
log.Debugf("Scanning for inflight payments")
|
||||||
payments, err := r.cfg.Control.FetchInFlightPayments()
|
payments, err := r.cfg.Control.FetchInFlightPayments(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user