14358372723

Committed 09 Apr 2025 01:26PM UTC coverage: 56.696% (-12.3%) from 69.037%

Build # 14358372723

Build Type

Pull #9696

github

Committed by

web-flow

Commit Message

Merge e2837e400 into 867d27d68

Pull Request Pull Request #9696: Add `development_guidelines.md` for both human and machine

Run Details

107055 of 188823 relevant lines covered (56.7%)

22721.56 hits per line

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

87.24

/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.
//
// NOTE: This method will overwrite the failure reason if the payment is already
// failed.
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 {	15✔
124	// Create a queue for payment updates.	15✔
125	queue := queue.NewConcurrentQueue(20)	15✔
126	queue.Start()	15✔
127		15✔
128	return &controlTowerSubscriberImpl{	15✔
129	updates: queue.ChanOut(),	15✔
130	queue: queue,	15✔
131	quit: make(chan struct{}),	15✔
132	}	15✔
133	}	15✔
134
135	// Close signals that the subscriber is no longer interested in updates.
136	func (s *controlTowerSubscriberImpl) Close() {	2✔
137	// Close quit channel so that any pending writes to the queue are	2✔
138	// cancelled.	2✔
139	close(s.quit)	2✔
140		2✔
141	// Stop the queue goroutine so that it won't leak.	2✔
142	s.queue.Stop()	2✔
143	}	2✔
144
145	// Updates is the channel over which *channeldb.MPPayment updates can be
146	// received.
147	func (s *controlTowerSubscriberImpl) Updates() <-chan interface{} {	47✔
148	return s.updates	47✔
149	}	47✔
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 {	9✔
172	return &controlTower{	9✔
173	db: db,	9✔
174	subscribersAllPayments: make(	9✔
175	map[uint64]*controlTowerSubscriberImpl,	9✔
176	),	9✔
177	subscribers: make(map[lntypes.Hash][]*controlTowerSubscriberImpl),	9✔
178	paymentsMtx: multimutex.NewMutex[lntypes.Hash](),	9✔
179	}	9✔
180	}	9✔
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 {	9✔
188		9✔
189	err := p.db.InitPayment(paymentHash, info)	9✔
190	if err != nil {	9✔
191	return err	×
192	}	×
193
194	// Take lock before querying the db to prevent missing or duplicating
195	// an update.
196	p.paymentsMtx.Lock(paymentHash)	9✔
197	defer p.paymentsMtx.Unlock(paymentHash)	9✔
198		9✔
199	payment, err := p.db.FetchPayment(paymentHash)	9✔
200	if err != nil {	9✔
201	return err	×
202	}	×
203
204	p.notifySubscribers(paymentHash, payment)	9✔
205		9✔
206	return nil	9✔
207	}
208
209	// DeleteFailedAttempts deletes all failed htlcs if the payment was
210	// successfully settled.
211	func (p *controlTower) DeleteFailedAttempts(paymentHash lntypes.Hash) error {	×
212	return p.db.DeleteFailedAttempts(paymentHash)	×
213	}	×
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 {	7✔
219		7✔
220	p.paymentsMtx.Lock(paymentHash)	7✔
221	defer p.paymentsMtx.Unlock(paymentHash)	7✔
222		7✔
223	payment, err := p.db.RegisterAttempt(paymentHash, attempt)	7✔
224	if err != nil {	7✔
225	return err	×
226	}	×
227
228	// Notify subscribers of the attempt registration.
229	p.notifySubscribers(paymentHash, payment)	7✔
230		7✔
231	return nil	7✔
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) {	3✔
240		3✔
241	p.paymentsMtx.Lock(paymentHash)	3✔
242	defer p.paymentsMtx.Unlock(paymentHash)	3✔
243		3✔
244	payment, err := p.db.SettleAttempt(paymentHash, attemptID, settleInfo)	3✔
245	if err != nil {	3✔
246	return nil, err	×
247	}	×
248
249	// Notify subscribers of success event.
250	p.notifySubscribers(paymentHash, payment)	3✔
251		3✔
252	return payment.GetAttempt(attemptID)	3✔
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) {	3✔
259		3✔
260	p.paymentsMtx.Lock(paymentHash)	3✔
261	defer p.paymentsMtx.Unlock(paymentHash)	3✔
262		3✔
263	payment, err := p.db.FailAttempt(paymentHash, attemptID, failInfo)	3✔
264	if err != nil {	3✔
265	return nil, err	×
266	}	×
267
268	// Notify subscribers of failed attempt.
269	p.notifySubscribers(paymentHash, payment)	3✔
270		3✔
271	return payment.GetAttempt(attemptID)	3✔
272	}
273
274	// FetchPayment fetches the payment corresponding to the given payment hash.
275	func (p *controlTower) FetchPayment(paymentHash lntypes.Hash) (
276	DBMPPayment, error) {	×
277		×
278	return p.db.FetchPayment(paymentHash)	×
279	}	×
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	//
286	// NOTE: This method will overwrite the failure reason if the payment is already
287	// failed.
288	func (p *controlTower) FailPayment(paymentHash lntypes.Hash,
289	reason channeldb.FailureReason) error {	4✔
290		4✔
291	p.paymentsMtx.Lock(paymentHash)	4✔
292	defer p.paymentsMtx.Unlock(paymentHash)	4✔
293		4✔
294	payment, err := p.db.Fail(paymentHash, reason)	4✔
295	if err != nil {	4✔
296	return err	×
297	}	×
298
299	// Notify subscribers of fail event.
300	p.notifySubscribers(paymentHash, payment)	4✔
301		4✔
302	return nil	4✔
303	}
304
305	// FetchInFlightPayments returns all payments with status InFlight.
306	func (p controlTower) FetchInFlightPayments() ([]channeldb.MPPayment, error) {	×
307	return p.db.FetchInFlightPayments()	×
308	}	×
309
310	// SubscribePayment subscribes to updates for the payment with the given hash. A
311	// first update with the current state of the payment is always sent out
312	// immediately.
313	func (p *controlTower) SubscribePayment(paymentHash lntypes.Hash) (
314	ControlTowerSubscriber, error) {	12✔
315		12✔
316	// Take lock before querying the db to prevent missing or duplicating an	12✔
317	// update.	12✔
318	p.paymentsMtx.Lock(paymentHash)	12✔
319	defer p.paymentsMtx.Unlock(paymentHash)	12✔
320		12✔
321	payment, err := p.db.FetchPayment(paymentHash)	12✔
322	if err != nil {	13✔
323	return nil, err	1✔
324	}	1✔
325
326	subscriber := newControlTowerSubscriber()	11✔
327		11✔
328	// Always write current payment state to the channel.	11✔
329	subscriber.queue.ChanIn() <- payment	11✔
330		11✔
331	// Payment is currently in flight. Register this subscriber for further	11✔
332	// updates. Otherwise this update is the final update and the incoming	11✔
333	// channel can be closed. This will close the queue's outgoing channel	11✔
334	// when all updates have been written.	11✔
335	if !payment.Terminated() {	17✔
336	p.subscribersMtx.Lock()	6✔
337	p.subscribers[paymentHash] = append(	6✔
338	p.subscribers[paymentHash], subscriber,	6✔
339	)	6✔
340	p.subscribersMtx.Unlock()	6✔
341	} else {	11✔
342	close(subscriber.queue.ChanIn())	5✔
343	}	5✔
344
345	return subscriber, nil	11✔
346	}
347
348	// SubscribeAllPayments subscribes to updates for all inflight payments. A first
349	// update with the current state of every inflight payment is always sent out
350	// immediately.
351	// Note: If payments are in-flight while starting a new subscription, the start
352	// of the payment stream could produce out-of-order and/or duplicate events. In
353	// order to get updates for every in-flight payment attempt make sure to
354	// subscribe to this method before initiating any payments.
355	func (p *controlTower) SubscribeAllPayments() (ControlTowerSubscriber, error) {	4✔
356	subscriber := newControlTowerSubscriber()	4✔
357		4✔
358	// Add the subscriber to the list before fetching in-flight payments, so	4✔
359	// no events are missed. If a payment attempt update occurs after	4✔
360	// appending and before fetching in-flight payments, an out-of-order	4✔
361	// duplicate may be produced, because it is then fetched in below call	4✔
362	// and notified through the subscription.	4✔
363	p.subscribersMtx.Lock()	4✔
364	p.subscribersAllPayments[p.subscriberIndex] = subscriber	4✔
365	p.subscriberIndex++	4✔
366	p.subscribersMtx.Unlock()	4✔
367		4✔
368	inflightPayments, err := p.db.FetchInFlightPayments()	4✔
369	if err != nil {	4✔
370	return nil, err	×
371	}	×
372
373	for index := range inflightPayments {	6✔
374	// Always write current payment state to the channel.	2✔
375	subscriber.queue.ChanIn() <- inflightPayments[index]	2✔
376	}	2✔
377
378	return subscriber, nil	4✔
379	}
380
381	// notifySubscribers sends a final payment event to all subscribers of this
382	// payment. The channel will be closed after this. Note that this function must
383	// be executed atomically (by means of a lock) with the database update to
384	// guarantee consistency of the notifications.
385	func (p *controlTower) notifySubscribers(paymentHash lntypes.Hash,
386	event *channeldb.MPPayment) {	26✔
387		26✔
388	// Get all subscribers for this payment.	26✔
389	p.subscribersMtx.Lock()	26✔
390		26✔
391	subscribersPaymentHash, ok := p.subscribers[paymentHash]	26✔
392	if !ok && len(p.subscribersAllPayments) == 0 {	34✔
393	p.subscribersMtx.Unlock()	8✔
394	return	8✔
395	}	8✔
396
397	// If the payment reached a terminal state, the subscriber list can be
398	// cleared. There won't be any more updates.
399	terminal := event.Terminated()	18✔
400	if terminal {	25✔
401	delete(p.subscribers, paymentHash)	7✔
402	}	7✔
403
404	// Copy subscribers to all payments locally while holding the lock in
405	// order to avoid concurrency issues while reading/writing the map.
406	subscribersAllPayments := make(map[uint64]*controlTowerSubscriberImpl)	18✔
407	for k, v := range p.subscribersAllPayments {	28✔
408	subscribersAllPayments[k] = v	10✔
409	}	10✔
410	p.subscribersMtx.Unlock()	18✔
411		18✔
412	// Notify all subscribers that subscribed to the current payment hash.	18✔
413	for _, subscriber := range subscribersPaymentHash {	29✔
414	select {	11✔
415	case subscriber.queue.ChanIn() <- event:	11✔
416	// If this event is the last, close the incoming channel	11✔
417	// of the queue. This will signal the subscriber that	11✔
418	// there won't be any more updates.	11✔
419	if terminal {	17✔
420	close(subscriber.queue.ChanIn())	6✔
421	}	6✔
422
423	// If subscriber disappeared, skip notification. For further
424	// notifications, we'll keep skipping over this subscriber.
425	case <-subscriber.quit:	×
426	}
427	}
428
429	// Notify all subscribers that subscribed to all payments.
430	for key, subscriber := range subscribersAllPayments {	28✔
431	select {	10✔
432	case subscriber.queue.ChanIn() <- event:	8✔
433
434	// If subscriber disappeared, remove it from the subscribers
435	// list.
436	case <-subscriber.quit:	2✔
437	p.subscribersMtx.Lock()	2✔
438	delete(p.subscribersAllPayments, key)	2✔
439	p.subscribersMtx.Unlock()	2✔
440	}
441	}
442	}

lightningnetwork / lnd / 14358372723

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