12199391122

Committed 06 Dec 2024 01:10PM UTC coverage: 49.807% (-9.1%) from 58.933%

Build # 12199391122

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #9337 from Guayaba221/patch-1

chore: fix typo in ruby.md

Run Details

100137 of 201051 relevant lines covered (49.81%)

2.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.29

/routing/control_tower.go

package routing

import (
        "sync"

        "github.com/lightningnetwork/lnd/channeldb"
        "github.com/lightningnetwork/lnd/lntypes"
        "github.com/lightningnetwork/lnd/multimutex"
        "github.com/lightningnetwork/lnd/queue"
)

// DBMPPayment is an interface derived from channeldb.MPPayment that is used by
// the payment lifecycle.
type DBMPPayment interface {
        // GetState returns the current state of the payment.
        GetState() *channeldb.MPPaymentState

        // Terminated returns true if the payment is in a final state.
        Terminated() bool

        // GetStatus returns the current status of the payment.
        GetStatus() channeldb.PaymentStatus

        // NeedWaitAttempts specifies whether the payment needs to wait for the
        // outcome of an attempt.
        NeedWaitAttempts() (bool, error)

        // GetHTLCs returns all HTLCs of this payment.
        GetHTLCs() []channeldb.HTLCAttempt

        // InFlightHTLCs returns all HTLCs that are in flight.
        InFlightHTLCs() []channeldb.HTLCAttempt

        // AllowMoreAttempts is used to decide whether we can safely attempt
        // more HTLCs for a given payment state. Return an error if the payment
        // is in an unexpected state.
        AllowMoreAttempts() (bool, error)

        // TerminalInfo returns the settled HTLC attempt or the payment's
        // failure reason.
        TerminalInfo() (*channeldb.HTLCAttempt, *channeldb.FailureReason)
}

// ControlTower tracks all outgoing payments made, whose primary purpose is to
// prevent duplicate payments to the same payment hash. In production, a
// persistent implementation is preferred so that tracking can survive across
// restarts. Payments are transitioned through various payment states, and the
// ControlTower interface provides access to driving the state transitions.
type ControlTower interface {
        // This method checks that no succeeded payment exist for this payment
        // hash.
        InitPayment(lntypes.Hash, *channeldb.PaymentCreationInfo) error

        // DeleteFailedAttempts removes all failed HTLCs from the db. It should
        // be called for a given payment whenever all inflight htlcs are
        // completed, and the payment has reached a final settled state.
        DeleteFailedAttempts(lntypes.Hash) error

        // RegisterAttempt atomically records the provided HTLCAttemptInfo.
        RegisterAttempt(lntypes.Hash, *channeldb.HTLCAttemptInfo) error

        // SettleAttempt marks the given attempt settled with the preimage. If
        // this is a multi shard payment, this might implicitly mean the the
        // full payment succeeded.
        //
        // After invoking this method, InitPayment should always return an
        // error to prevent us from making duplicate payments to the same
        // payment hash. The provided preimage is atomically saved to the DB
        // for record keeping.
        SettleAttempt(lntypes.Hash, uint64, *channeldb.HTLCSettleInfo) (
                *channeldb.HTLCAttempt, error)

        // FailAttempt marks the given payment attempt failed.
        FailAttempt(lntypes.Hash, uint64, *channeldb.HTLCFailInfo) (
                *channeldb.HTLCAttempt, error)

        // FetchPayment fetches the payment corresponding to the given payment
        // hash.
        FetchPayment(paymentHash lntypes.Hash) (DBMPPayment, error)

        // FailPayment transitions a payment into the Failed state, and records
        // the ultimate reason the payment failed. Note that this should only
        // be called when all active attempts are already failed. After
        // invoking this method, InitPayment should return nil on its next call
        // for this payment hash, allowing the user to make a subsequent
        // payment.
        FailPayment(lntypes.Hash, channeldb.FailureReason) error

        // FetchInFlightPayments returns all payments with status InFlight.
        FetchInFlightPayments() ([]*channeldb.MPPayment, error)

        // SubscribePayment subscribes to updates for the payment with the given
        // hash. A first update with the current state of the payment is always
        // sent out immediately.
        SubscribePayment(paymentHash lntypes.Hash) (ControlTowerSubscriber,
                error)

        // SubscribeAllPayments subscribes to updates for all payments. A first
        // update with the current state of every inflight payment is always
        // sent out immediately.
        SubscribeAllPayments() (ControlTowerSubscriber, error)
}

// ControlTowerSubscriber contains the state for a payment update subscriber.
type ControlTowerSubscriber interface {
        // Updates is the channel over which *channeldb.MPPayment updates can be
        // received.
        Updates() <-chan interface{}

        // Close signals that the subscriber is no longer interested in updates.
        Close()
}

// ControlTowerSubscriberImpl contains the state for a payment update
// subscriber.
type controlTowerSubscriberImpl struct {
        updates <-chan interface{}
        queue   *queue.ConcurrentQueue
        quit    chan struct{}
}

// newControlTowerSubscriber instantiates a new subscriber state object.
func newControlTowerSubscriber() *controlTowerSubscriberImpl {
        // Create a queue for payment updates.
        queue := queue.NewConcurrentQueue(20)
        queue.Start()

        return &controlTowerSubscriberImpl{
                updates: queue.ChanOut(),
                queue:   queue,
                quit:    make(chan struct{}),
        }
}

// Close signals that the subscriber is no longer interested in updates.
func (s *controlTowerSubscriberImpl) Close() {
        // Close quit channel so that any pending writes to the queue are
        // cancelled.
        close(s.quit)

        // Stop the queue goroutine so that it won't leak.
        s.queue.Stop()
}

// Updates is the channel over which *channeldb.MPPayment updates can be
// received.
func (s *controlTowerSubscriberImpl) Updates() <-chan interface{} {
        return s.updates
}

// controlTower is persistent implementation of ControlTower to restrict
// double payment sending.
type controlTower struct {
        db *channeldb.PaymentControl

        // subscriberIndex is used to provide a unique id for each subscriber
        // to all payments. This is used to easily remove the subscriber when
        // necessary.
        subscriberIndex        uint64
        subscribersAllPayments map[uint64]*controlTowerSubscriberImpl
        subscribers            map[lntypes.Hash][]*controlTowerSubscriberImpl
        subscribersMtx         sync.Mutex

        // paymentsMtx provides synchronization on the payment level to ensure
        // that no race conditions occur in between updating the database and
        // sending a notification.
        paymentsMtx *multimutex.Mutex[lntypes.Hash]
}

// NewControlTower creates a new instance of the controlTower.
func NewControlTower(db *channeldb.PaymentControl) ControlTower {
        return &controlTower{
                db: db,
                subscribersAllPayments: make(
                        map[uint64]*controlTowerSubscriberImpl,
                ),
                subscribers: make(map[lntypes.Hash][]*controlTowerSubscriberImpl),
                paymentsMtx: multimutex.NewMutex[lntypes.Hash](),
        }
}

// InitPayment checks or records the given PaymentCreationInfo with the DB,
// making sure it does not already exist as an in-flight payment. Then this
// method returns successfully, the payment is guaranteed to be in the
// Initiated state.
func (p *controlTower) InitPayment(paymentHash lntypes.Hash,
        info *channeldb.PaymentCreationInfo) error {

        err := p.db.InitPayment(paymentHash, info)
        if err != nil {
                return err
        }

        // Take lock before querying the db to prevent missing or duplicating
        // an update.
        p.paymentsMtx.Lock(paymentHash)
        defer p.paymentsMtx.Unlock(paymentHash)

        payment, err := p.db.FetchPayment(paymentHash)
        if err != nil {
                return err
        }

        p.notifySubscribers(paymentHash, payment)

        return nil
}

// DeleteFailedAttempts deletes all failed htlcs if the payment was
// successfully settled.
func (p *controlTower) DeleteFailedAttempts(paymentHash lntypes.Hash) error {
        return p.db.DeleteFailedAttempts(paymentHash)
}

// RegisterAttempt atomically records the provided HTLCAttemptInfo to the
// DB.
func (p *controlTower) RegisterAttempt(paymentHash lntypes.Hash,
        attempt *channeldb.HTLCAttemptInfo) error {

        p.paymentsMtx.Lock(paymentHash)
        defer p.paymentsMtx.Unlock(paymentHash)

        payment, err := p.db.RegisterAttempt(paymentHash, attempt)
        if err != nil {
                return err
        }

        // Notify subscribers of the attempt registration.
        p.notifySubscribers(paymentHash, payment)

        return nil
}

// SettleAttempt marks the given attempt settled with the preimage. If
// this is a multi shard payment, this might implicitly mean the the
// full payment succeeded.
func (p *controlTower) SettleAttempt(paymentHash lntypes.Hash,
        attemptID uint64, settleInfo *channeldb.HTLCSettleInfo) (
        *channeldb.HTLCAttempt, error) {

        p.paymentsMtx.Lock(paymentHash)
        defer p.paymentsMtx.Unlock(paymentHash)

        payment, err := p.db.SettleAttempt(paymentHash, attemptID, settleInfo)
        if err != nil {
                return nil, err
        }

        // Notify subscribers of success event.
        p.notifySubscribers(paymentHash, payment)

        return payment.GetAttempt(attemptID)
}

// FailAttempt marks the given payment attempt failed.
func (p *controlTower) FailAttempt(paymentHash lntypes.Hash,
        attemptID uint64, failInfo *channeldb.HTLCFailInfo) (
        *channeldb.HTLCAttempt, error) {

        p.paymentsMtx.Lock(paymentHash)
        defer p.paymentsMtx.Unlock(paymentHash)

        payment, err := p.db.FailAttempt(paymentHash, attemptID, failInfo)
        if err != nil {
                return nil, err
        }

        // Notify subscribers of failed attempt.
        p.notifySubscribers(paymentHash, payment)

        return payment.GetAttempt(attemptID)
}

// FetchPayment fetches the payment corresponding to the given payment hash.
func (p *controlTower) FetchPayment(paymentHash lntypes.Hash) (
        DBMPPayment, error) {

        return p.db.FetchPayment(paymentHash)
}

// FailPayment transitions a payment into the Failed state, and records the
// reason the payment failed. After invoking this method, InitPayment should
// return nil on its next call for this payment hash, allowing the switch to
// make a subsequent payment.
func (p *controlTower) FailPayment(paymentHash lntypes.Hash,
        reason channeldb.FailureReason) error {

        p.paymentsMtx.Lock(paymentHash)
        defer p.paymentsMtx.Unlock(paymentHash)

        payment, err := p.db.Fail(paymentHash, reason)
        if err != nil {
                return err
        }

        // Notify subscribers of fail event.
        p.notifySubscribers(paymentHash, payment)

        return nil
}

// FetchInFlightPayments returns all payments with status InFlight.
func (p *controlTower) FetchInFlightPayments() ([]*channeldb.MPPayment, error) {
        return p.db.FetchInFlightPayments()
}

// SubscribePayment subscribes to updates for the payment with the given hash. A
// first update with the current state of the payment is always sent out
// immediately.
func (p *controlTower) SubscribePayment(paymentHash lntypes.Hash) (
        ControlTowerSubscriber, error) {

        // Take lock before querying the db to prevent missing or duplicating an
        // update.
        p.paymentsMtx.Lock(paymentHash)
        defer p.paymentsMtx.Unlock(paymentHash)

        payment, err := p.db.FetchPayment(paymentHash)
        if err != nil {
                return nil, err
        }

        subscriber := newControlTowerSubscriber()

        // Always write current payment state to the channel.
        subscriber.queue.ChanIn() <- payment

        // Payment is currently in flight. Register this subscriber for further
        // updates. Otherwise this update is the final update and the incoming
        // channel can be closed. This will close the queue's outgoing channel
        // when all updates have been written.
        if !payment.Terminated() {
                p.subscribersMtx.Lock()
                p.subscribers[paymentHash] = append(
                        p.subscribers[paymentHash], subscriber,
                )
                p.subscribersMtx.Unlock()
        } else {
                close(subscriber.queue.ChanIn())
        }

        return subscriber, nil
}

// SubscribeAllPayments subscribes to updates for all inflight payments. A first
// update with the current state of every inflight payment is always sent out
// immediately.
// Note: If payments are in-flight while starting a new subscription, the start
// of the payment stream could produce out-of-order and/or duplicate events. In
// order to get updates for every in-flight payment attempt make sure to
// subscribe to this method before initiating any payments.
func (p *controlTower) SubscribeAllPayments() (ControlTowerSubscriber, error) {
        subscriber := newControlTowerSubscriber()

        // Add the subscriber to the list before fetching in-flight payments, so
        // no events are missed. If a payment attempt update occurs after
        // appending and before fetching in-flight payments, an out-of-order
        // duplicate may be produced, because it is then fetched in below call
        // and notified through the subscription.
        p.subscribersMtx.Lock()
        p.subscribersAllPayments[p.subscriberIndex] = subscriber
        p.subscriberIndex++
        p.subscribersMtx.Unlock()

        inflightPayments, err := p.db.FetchInFlightPayments()
        if err != nil {
                return nil, err
        }

        for index := range inflightPayments {
                // Always write current payment state to the channel.
                subscriber.queue.ChanIn() <- inflightPayments[index]
        }

        return subscriber, nil
}

// notifySubscribers sends a final payment event to all subscribers of this
// payment. The channel will be closed after this. Note that this function must
// be executed atomically (by means of a lock) with the database update to
// guarantee consistency of the notifications.
func (p *controlTower) notifySubscribers(paymentHash lntypes.Hash,
        event *channeldb.MPPayment) {

        // Get all subscribers for this payment.
        p.subscribersMtx.Lock()

        subscribersPaymentHash, ok := p.subscribers[paymentHash]
        if !ok && len(p.subscribersAllPayments) == 0 {
                p.subscribersMtx.Unlock()
                return
        }

        // If the payment reached a terminal state, the subscriber list can be
        // cleared. There won't be any more updates.
        terminal := event.Terminated()
        if terminal {
                delete(p.subscribers, paymentHash)
        }

        // Copy subscribers to all payments locally while holding the lock in
        // order to avoid concurrency issues while reading/writing the map.
        subscribersAllPayments := make(map[uint64]*controlTowerSubscriberImpl)
        for k, v := range p.subscribersAllPayments {
                subscribersAllPayments[k] = v
        }
        p.subscribersMtx.Unlock()

        // Notify all subscribers that subscribed to the current payment hash.
        for _, subscriber := range subscribersPaymentHash {
                select {
                case subscriber.queue.ChanIn() <- event:
                        // If this event is the last, close the incoming channel
                        // of the queue. This will signal the subscriber that
                        // there won't be any more updates.
                        if terminal {
                                close(subscriber.queue.ChanIn())
                        }

                // If subscriber disappeared, skip notification. For further
                // notifications, we'll keep skipping over this subscriber.
                case <-subscriber.quit:
                }
        }

        // Notify all subscribers that subscribed to all payments.
        for key, subscriber := range subscribersAllPayments {
                select {
                case subscriber.queue.ChanIn() <- event:

                // If subscriber disappeared, remove it from the subscribers
                // list.
                case <-subscriber.quit:
                        p.subscribersMtx.Lock()
                        delete(p.subscribersAllPayments, key)
                        p.subscribersMtx.Unlock()
                }
        }
}

1	package routing
2
3	import (
4	"sync"
5
6	"github.com/lightningnetwork/lnd/channeldb"
7	"github.com/lightningnetwork/lnd/lntypes"
8	"github.com/lightningnetwork/lnd/multimutex"
9	"github.com/lightningnetwork/lnd/queue"
10	)
11
12	// DBMPPayment is an interface derived from channeldb.MPPayment that is used by
13	// the payment lifecycle.
14	type DBMPPayment interface {
15	// GetState returns the current state of the payment.
16	GetState() *channeldb.MPPaymentState
17
18	// Terminated returns true if the payment is in a final state.
19	Terminated() bool
20
21	// GetStatus returns the current status of the payment.
22	GetStatus() channeldb.PaymentStatus
23
24	// NeedWaitAttempts specifies whether the payment needs to wait for the
25	// outcome of an attempt.
26	NeedWaitAttempts() (bool, error)
27
28	// GetHTLCs returns all HTLCs of this payment.
29	GetHTLCs() []channeldb.HTLCAttempt
30
31	// InFlightHTLCs returns all HTLCs that are in flight.
32	InFlightHTLCs() []channeldb.HTLCAttempt
33
34	// AllowMoreAttempts is used to decide whether we can safely attempt
35	// more HTLCs for a given payment state. Return an error if the payment
36	// is in an unexpected state.
37	AllowMoreAttempts() (bool, error)
38
39	// TerminalInfo returns the settled HTLC attempt or the payment's
40	// failure reason.
41	TerminalInfo() (channeldb.HTLCAttempt, channeldb.FailureReason)
42	}
43
44	// ControlTower tracks all outgoing payments made, whose primary purpose is to
45	// prevent duplicate payments to the same payment hash. In production, a
46	// persistent implementation is preferred so that tracking can survive across
47	// restarts. Payments are transitioned through various payment states, and the
48	// ControlTower interface provides access to driving the state transitions.
49	type ControlTower interface {
50	// This method checks that no succeeded payment exist for this payment
51	// hash.
52	InitPayment(lntypes.Hash, *channeldb.PaymentCreationInfo) error
53
54	// DeleteFailedAttempts removes all failed HTLCs from the db. It should
55	// be called for a given payment whenever all inflight htlcs are
56	// completed, and the payment has reached a final settled state.
57	DeleteFailedAttempts(lntypes.Hash) error
58
59	// RegisterAttempt atomically records the provided HTLCAttemptInfo.
60	RegisterAttempt(lntypes.Hash, *channeldb.HTLCAttemptInfo) error
61
62	// SettleAttempt marks the given attempt settled with the preimage. If
63	// this is a multi shard payment, this might implicitly mean the the
64	// full payment succeeded.
65	//
66	// After invoking this method, InitPayment should always return an
67	// error to prevent us from making duplicate payments to the same
68	// payment hash. The provided preimage is atomically saved to the DB
69	// for record keeping.
70	SettleAttempt(lntypes.Hash, uint64, *channeldb.HTLCSettleInfo) (
71	*channeldb.HTLCAttempt, error)
72
73	// FailAttempt marks the given payment attempt failed.
74	FailAttempt(lntypes.Hash, uint64, *channeldb.HTLCFailInfo) (
75	*channeldb.HTLCAttempt, error)
76
77	// FetchPayment fetches the payment corresponding to the given payment
78	// hash.
79	FetchPayment(paymentHash lntypes.Hash) (DBMPPayment, error)
80
81	// FailPayment transitions a payment into the Failed state, and records
82	// the ultimate reason the payment failed. Note that this should only
83	// be called when all active attempts are already failed. After
84	// invoking this method, InitPayment should return nil on its next call
85	// for this payment hash, allowing the user to make a subsequent
86	// payment.
87	FailPayment(lntypes.Hash, channeldb.FailureReason) error
88
89	// FetchInFlightPayments returns all payments with status InFlight.
90	FetchInFlightPayments() ([]*channeldb.MPPayment, error)
91
92	// SubscribePayment subscribes to updates for the payment with the given
93	// hash. A first update with the current state of the payment is always
94	// sent out immediately.
95	SubscribePayment(paymentHash lntypes.Hash) (ControlTowerSubscriber,
96	error)
97
98	// SubscribeAllPayments subscribes to updates for all payments. A first
99	// update with the current state of every inflight payment is always
100	// sent out immediately.
101	SubscribeAllPayments() (ControlTowerSubscriber, error)
102	}
103
104	// ControlTowerSubscriber contains the state for a payment update subscriber.
105	type ControlTowerSubscriber interface {
106	// Updates is the channel over which *channeldb.MPPayment updates can be
107	// received.
108	Updates() <-chan interface{}
109
110	// Close signals that the subscriber is no longer interested in updates.
111	Close()
112	}
113
114	// ControlTowerSubscriberImpl contains the state for a payment update
115	// subscriber.
116	type controlTowerSubscriberImpl struct {
117	updates <-chan interface{}
118	queue *queue.ConcurrentQueue
119	quit chan struct{}
120	}
121
122	// newControlTowerSubscriber instantiates a new subscriber state object.
123	func newControlTowerSubscriber() *controlTowerSubscriberImpl {	4✔
124	// Create a queue for payment updates.	4✔
125	queue := queue.NewConcurrentQueue(20)	4✔
126	queue.Start()	4✔
127		4✔
128	return &controlTowerSubscriberImpl{	4✔
129	updates: queue.ChanOut(),	4✔
130	queue: queue,	4✔
131	quit: make(chan struct{}),	4✔
132	}	4✔
133	}	4✔
134
135	// Close signals that the subscriber is no longer interested in updates.
136	func (s *controlTowerSubscriberImpl) Close() {	4✔
137	// Close quit channel so that any pending writes to the queue are	4✔
138	// cancelled.	4✔
139	close(s.quit)	4✔
140		4✔
141	// Stop the queue goroutine so that it won't leak.	4✔
142	s.queue.Stop()	4✔
143	}	4✔
144
145	// Updates is the channel over which *channeldb.MPPayment updates can be
146	// received.
147	func (s *controlTowerSubscriberImpl) Updates() <-chan interface{} {	4✔
148	return s.updates	4✔
149	}	4✔
150
151	// controlTower is persistent implementation of ControlTower to restrict
152	// double payment sending.
153	type controlTower struct {
154	db *channeldb.PaymentControl
155
156	// subscriberIndex is used to provide a unique id for each subscriber
157	// to all payments. This is used to easily remove the subscriber when
158	// necessary.
159	subscriberIndex uint64
160	subscribersAllPayments map[uint64]*controlTowerSubscriberImpl
161	subscribers map[lntypes.Hash][]*controlTowerSubscriberImpl
162	subscribersMtx sync.Mutex
163
164	// paymentsMtx provides synchronization on the payment level to ensure
165	// that no race conditions occur in between updating the database and
166	// sending a notification.
167	paymentsMtx *multimutex.Mutex[lntypes.Hash]
168	}
169
170	// NewControlTower creates a new instance of the controlTower.
171	func NewControlTower(db *channeldb.PaymentControl) ControlTower {	4✔
172	return &controlTower{	4✔
173	db: db,	4✔
174	subscribersAllPayments: make(	4✔
175	map[uint64]*controlTowerSubscriberImpl,	4✔
176	),	4✔
177	subscribers: make(map[lntypes.Hash][]*controlTowerSubscriberImpl),	4✔
178	paymentsMtx: multimutex.NewMutex[lntypes.Hash](),	4✔
179	}	4✔
180	}	4✔
181
182	// InitPayment checks or records the given PaymentCreationInfo with the DB,
183	// making sure it does not already exist as an in-flight payment. Then this
184	// method returns successfully, the payment is guaranteed to be in the
185	// Initiated state.
186	func (p *controlTower) InitPayment(paymentHash lntypes.Hash,
187	info *channeldb.PaymentCreationInfo) error {	4✔
188		4✔
189	err := p.db.InitPayment(paymentHash, info)	4✔
190	if err != nil {	8✔
191	return err	4✔
192	}	4✔
193
194	// Take lock before querying the db to prevent missing or duplicating
195	// an update.
196	p.paymentsMtx.Lock(paymentHash)	4✔
197	defer p.paymentsMtx.Unlock(paymentHash)	4✔
198		4✔
199	payment, err := p.db.FetchPayment(paymentHash)	4✔
200	if err != nil {	4✔
201	return err	×
202	}	×
203
204	p.notifySubscribers(paymentHash, payment)	4✔
205		4✔
206	return nil	4✔
207	}
208
209	// DeleteFailedAttempts deletes all failed htlcs if the payment was
210	// successfully settled.
211	func (p *controlTower) DeleteFailedAttempts(paymentHash lntypes.Hash) error {	4✔
212	return p.db.DeleteFailedAttempts(paymentHash)	4✔
213	}	4✔
214
215	// RegisterAttempt atomically records the provided HTLCAttemptInfo to the
216	// DB.
217	func (p *controlTower) RegisterAttempt(paymentHash lntypes.Hash,
218	attempt *channeldb.HTLCAttemptInfo) error {	4✔
219		4✔
220	p.paymentsMtx.Lock(paymentHash)	4✔
221	defer p.paymentsMtx.Unlock(paymentHash)	4✔
222		4✔
223	payment, err := p.db.RegisterAttempt(paymentHash, attempt)	4✔
224	if err != nil {	4✔
225	return err	×
226	}	×
227
228	// Notify subscribers of the attempt registration.
229	p.notifySubscribers(paymentHash, payment)	4✔
230		4✔
231	return nil	4✔
232	}
233
234	// SettleAttempt marks the given attempt settled with the preimage. If
235	// this is a multi shard payment, this might implicitly mean the the
236	// full payment succeeded.
237	func (p *controlTower) SettleAttempt(paymentHash lntypes.Hash,
238	attemptID uint64, settleInfo *channeldb.HTLCSettleInfo) (
239	*channeldb.HTLCAttempt, error) {	4✔
240		4✔
241	p.paymentsMtx.Lock(paymentHash)	4✔
242	defer p.paymentsMtx.Unlock(paymentHash)	4✔
243		4✔
244	payment, err := p.db.SettleAttempt(paymentHash, attemptID, settleInfo)	4✔
245	if err != nil {	4✔
246	return nil, err	×
247	}	×
248
249	// Notify subscribers of success event.
250	p.notifySubscribers(paymentHash, payment)	4✔
251		4✔
252	return payment.GetAttempt(attemptID)	4✔
253	}
254
255	// FailAttempt marks the given payment attempt failed.
256	func (p *controlTower) FailAttempt(paymentHash lntypes.Hash,
257	attemptID uint64, failInfo *channeldb.HTLCFailInfo) (
258	*channeldb.HTLCAttempt, error) {	4✔
259		4✔
260	p.paymentsMtx.Lock(paymentHash)	4✔
261	defer p.paymentsMtx.Unlock(paymentHash)	4✔
262		4✔
263	payment, err := p.db.FailAttempt(paymentHash, attemptID, failInfo)	4✔
264	if err != nil {	4✔
265	return nil, err	×
266	}	×
267
268	// Notify subscribers of failed attempt.
269	p.notifySubscribers(paymentHash, payment)	4✔
270		4✔
271	return payment.GetAttempt(attemptID)	4✔
272	}
273
274	// FetchPayment fetches the payment corresponding to the given payment hash.
275	func (p *controlTower) FetchPayment(paymentHash lntypes.Hash) (
276	DBMPPayment, error) {	4✔
277		4✔
278	return p.db.FetchPayment(paymentHash)	4✔
279	}	4✔
280
281	// FailPayment transitions a payment into the Failed state, and records the
282	// reason the payment failed. After invoking this method, InitPayment should
283	// return nil on its next call for this payment hash, allowing the switch to
284	// make a subsequent payment.
285	func (p *controlTower) FailPayment(paymentHash lntypes.Hash,
286	reason channeldb.FailureReason) error {	4✔
287		4✔
288	p.paymentsMtx.Lock(paymentHash)	4✔
289	defer p.paymentsMtx.Unlock(paymentHash)	4✔
290		4✔
291	payment, err := p.db.Fail(paymentHash, reason)	4✔
292	if err != nil {	4✔
293	return err	×
294	}	×
295
296	// Notify subscribers of fail event.
297	p.notifySubscribers(paymentHash, payment)	4✔
298		4✔
299	return nil	4✔
300	}
301
302	// FetchInFlightPayments returns all payments with status InFlight.
303	func (p controlTower) FetchInFlightPayments() ([]channeldb.MPPayment, error) {	4✔
304	return p.db.FetchInFlightPayments()	4✔
305	}	4✔
306
307	// SubscribePayment subscribes to updates for the payment with the given hash. A
308	// first update with the current state of the payment is always sent out
309	// immediately.
310	func (p *controlTower) SubscribePayment(paymentHash lntypes.Hash) (
311	ControlTowerSubscriber, error) {	4✔
312		4✔
313	// Take lock before querying the db to prevent missing or duplicating an	4✔
314	// update.	4✔
315	p.paymentsMtx.Lock(paymentHash)	4✔
316	defer p.paymentsMtx.Unlock(paymentHash)	4✔
317		4✔
318	payment, err := p.db.FetchPayment(paymentHash)	4✔
319	if err != nil {	4✔
320	return nil, err	×
321	}	×
322
323	subscriber := newControlTowerSubscriber()	4✔
324		4✔
325	// Always write current payment state to the channel.	4✔
326	subscriber.queue.ChanIn() <- payment	4✔
327		4✔
328	// Payment is currently in flight. Register this subscriber for further	4✔
329	// updates. Otherwise this update is the final update and the incoming	4✔
330	// channel can be closed. This will close the queue's outgoing channel	4✔
331	// when all updates have been written.	4✔
332	if !payment.Terminated() {	8✔
333	p.subscribersMtx.Lock()	4✔
334	p.subscribers[paymentHash] = append(	4✔
335	p.subscribers[paymentHash], subscriber,	4✔
336	)	4✔
337	p.subscribersMtx.Unlock()	4✔
338	} else {	8✔
339	close(subscriber.queue.ChanIn())	4✔
340	}	4✔
341
342	return subscriber, nil	4✔
343	}
344
345	// SubscribeAllPayments subscribes to updates for all inflight payments. A first
346	// update with the current state of every inflight payment is always sent out
347	// immediately.
348	// Note: If payments are in-flight while starting a new subscription, the start
349	// of the payment stream could produce out-of-order and/or duplicate events. In
350	// order to get updates for every in-flight payment attempt make sure to
351	// subscribe to this method before initiating any payments.
352	func (p *controlTower) SubscribeAllPayments() (ControlTowerSubscriber, error) {	4✔
353	subscriber := newControlTowerSubscriber()	4✔
354		4✔
355	// Add the subscriber to the list before fetching in-flight payments, so	4✔
356	// no events are missed. If a payment attempt update occurs after	4✔
357	// appending and before fetching in-flight payments, an out-of-order	4✔
358	// duplicate may be produced, because it is then fetched in below call	4✔
359	// and notified through the subscription.	4✔
360	p.subscribersMtx.Lock()	4✔
361	p.subscribersAllPayments[p.subscriberIndex] = subscriber	4✔
362	p.subscriberIndex++	4✔
363	p.subscribersMtx.Unlock()	4✔
364		4✔
365	inflightPayments, err := p.db.FetchInFlightPayments()	4✔
366	if err != nil {	4✔
367	return nil, err	×
368	}	×
369
370	for index := range inflightPayments {	4✔
371	// Always write current payment state to the channel.	×
372	subscriber.queue.ChanIn() <- inflightPayments[index]	×
373	}	×
374
375	return subscriber, nil	4✔
376	}
377
378	// notifySubscribers sends a final payment event to all subscribers of this
379	// payment. The channel will be closed after this. Note that this function must
380	// be executed atomically (by means of a lock) with the database update to
381	// guarantee consistency of the notifications.
382	func (p *controlTower) notifySubscribers(paymentHash lntypes.Hash,
383	event *channeldb.MPPayment) {	4✔
384		4✔
385	// Get all subscribers for this payment.	4✔
386	p.subscribersMtx.Lock()	4✔
387		4✔
388	subscribersPaymentHash, ok := p.subscribers[paymentHash]	4✔
389	if !ok && len(p.subscribersAllPayments) == 0 {	8✔
390	p.subscribersMtx.Unlock()	4✔
391	return	4✔
392	}	4✔
393
394	// If the payment reached a terminal state, the subscriber list can be
395	// cleared. There won't be any more updates.
396	terminal := event.Terminated()	4✔
397	if terminal {	8✔
398	delete(p.subscribers, paymentHash)	4✔
399	}	4✔
400
401	// Copy subscribers to all payments locally while holding the lock in
402	// order to avoid concurrency issues while reading/writing the map.
403	subscribersAllPayments := make(map[uint64]*controlTowerSubscriberImpl)	4✔
404	for k, v := range p.subscribersAllPayments {	8✔
405	subscribersAllPayments[k] = v	4✔
406	}	4✔
407	p.subscribersMtx.Unlock()	4✔
408		4✔
409	// Notify all subscribers that subscribed to the current payment hash.	4✔
410	for _, subscriber := range subscribersPaymentHash {	8✔
411	select {	4✔
412	case subscriber.queue.ChanIn() <- event:	4✔
413	// If this event is the last, close the incoming channel	4✔
414	// of the queue. This will signal the subscriber that	4✔
415	// there won't be any more updates.	4✔
416	if terminal {	8✔
417	close(subscriber.queue.ChanIn())	4✔
418	}	4✔
419
420	// If subscriber disappeared, skip notification. For further
421	// notifications, we'll keep skipping over this subscriber.
422	case <-subscriber.quit:	4✔
423	}
424	}
425
426	// Notify all subscribers that subscribed to all payments.
427	for key, subscriber := range subscribersAllPayments {	8✔
428	select {	4✔
429	case subscriber.queue.ChanIn() <- event:	4✔
430
431	// If subscriber disappeared, remove it from the subscribers
432	// list.
433	case <-subscriber.quit:	×
434	p.subscribersMtx.Lock()	×
435	delete(p.subscribersAllPayments, key)	×
436	p.subscribersMtx.Unlock()	×
437	}
438	}
439	}

lightningnetwork / lnd / 12199391122

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous