17777197509

Committed 16 Sep 2025 07:46PM UTC coverage: 57.202% (-9.5%) from 66.657%

Build # 17777197509

Build Type

Pull #9489

github

Committed by

web-flow

Commit Message

Merge bd2ae0bae into cbed86e21

Pull Request Pull Request #9489: multi: add BuildOnion, SendOnion, and TrackOnion RPCs

Run Details

329 of 564 new or added lines in 12 files covered. (58.33%)

28576 existing lines in 457 files now uncovered.

99724 of 174338 relevant lines covered (57.2%)

1.78 hits per line

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

82.17

/htlcswitch/payment_result.go

package htlcswitch

import (
        "bytes"
        "encoding/binary"
        "errors"
        "io"
        "sync"

        "github.com/lightningnetwork/lnd/channeldb"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/multimutex"
)

var (

        // networkResultStoreBucketKey is used for the root level bucket that
        // stores the network result for each payment ID.
        networkResultStoreBucketKey = []byte("network-result-store-bucket")

        // ErrPaymentIDNotFound is an error returned if the given paymentID is
        // not found.
        ErrPaymentIDNotFound = errors.New("paymentID not found")

        // ErrPaymentIDAlreadyExists is returned if we try to write a pending
        // payment whose paymentID already exists.
        ErrPaymentIDAlreadyExists = errors.New("paymentID already exists")
)

// PaymentResult wraps a decoded result received from the network after a
// payment attempt was made. This is what is eventually handed to the router
// for processing.
type PaymentResult struct {
        // Preimage is set by the switch in case a sent HTLC was settled.
        Preimage [32]byte

        // Error is non-nil in case a HTLC send failed, and the HTLC is now
        // irrevocably canceled. If the payment failed during forwarding, this
        // error will be a *ForwardingError.
        Error error

        // EncryptedError will contain the raw bytes of an encrypted error
        // in the event of a payment failure if the switch is instructed to
        // defer error processing to external sub-systems.
        EncryptedError []byte
}

// networkResult is the raw result received from the network after a payment
// attempt has been made. Since the switch doesn't always have the necessary
// data to decode the raw message, we store it together with some meta data,
// and decode it when the router query for the final result.
type networkResult struct {
        // msg is the received result. This should be of type UpdateFulfillHTLC
        // or UpdateFailHTLC.
        msg lnwire.Message

        // unencrypted indicates whether the failure encoded in the message is
        // unencrypted, and hence doesn't need to be decrypted.
        unencrypted bool

        // isResolution indicates whether this is a resolution message, in
        // which the failure reason might not be included.
        isResolution bool
}

// serializeNetworkResult serializes the networkResult.
func serializeNetworkResult(w io.Writer, n *networkResult) error {
        return channeldb.WriteElements(w, n.msg, n.unencrypted, n.isResolution)
}

// deserializeNetworkResult deserializes the networkResult.
func deserializeNetworkResult(r io.Reader) (*networkResult, error) {
        n := &networkResult{}

        if err := channeldb.ReadElements(r,
                &n.msg, &n.unencrypted, &n.isResolution,
        ); err != nil {
                return nil, err
        }

        return n, nil
}

// networkResultStore is a persistent store that stores any results of HTLCs in
// flight on the network. Since payment results are inherently asynchronous, it
// is used as a common access point for senders of HTLCs, to know when a result
// is back. The Switch will checkpoint any received result to the store, and
// the store will keep results and notify the callers about them.
type networkResultStore struct {
        backend kvdb.Backend

        // results is a map from paymentIDs to channels where subscribers to
        // payment results will be notified.
        results    map[uint64][]chan *networkResult
        resultsMtx sync.Mutex

        // attemptIDMtx is a multimutex used to make sure the database and
        // result subscribers map is consistent for each attempt ID in case of
        // concurrent callers.
        attemptIDMtx *multimutex.Mutex[uint64]
}

func newNetworkResultStore(db kvdb.Backend) *networkResultStore {
        return &networkResultStore{
                backend:      db,
                results:      make(map[uint64][]chan *networkResult),
                attemptIDMtx: multimutex.NewMutex[uint64](),
        }
}

// storeResult stores the networkResult for the given attemptID, and notifies
// any subscribers.
func (store *networkResultStore) storeResult(attemptID uint64,
        result *networkResult) error {

        // We get a mutex for this attempt ID. This is needed to ensure
        // consistency between the database state and the subscribers in case
        // of concurrent calls.
        store.attemptIDMtx.Lock(attemptID)
        defer store.attemptIDMtx.Unlock(attemptID)

        log.Debugf("Storing result for attemptID=%v", attemptID)

        // Serialize the payment result.
        var b bytes.Buffer
        if err := serializeNetworkResult(&b, result); err != nil {
                return err
        }

        var attemptIDBytes [8]byte
        binary.BigEndian.PutUint64(attemptIDBytes[:], attemptID)

        err := kvdb.Batch(store.backend, func(tx kvdb.RwTx) error {
                networkResults, err := tx.CreateTopLevelBucket(
                        networkResultStoreBucketKey,
                )
                if err != nil {
                        return err
                }

                return networkResults.Put(attemptIDBytes[:], b.Bytes())
        })
        if err != nil {
                return err
        }

        // Now that the result is stored in the database, we can notify any
        // active subscribers.
        store.resultsMtx.Lock()
        for _, res := range store.results[attemptID] {
                res <- result
        }
        delete(store.results, attemptID)
        store.resultsMtx.Unlock()

        return nil
}

// subscribeResult is used to get the HTLC attempt result for the given attempt
// ID.  It returns a channel on which the result will be delivered when ready.
func (store *networkResultStore) subscribeResult(attemptID uint64) (
        <-chan *networkResult, error) {

        // We get a mutex for this payment ID. This is needed to ensure
        // consistency between the database state and the subscribers in case
        // of concurrent calls.
        store.attemptIDMtx.Lock(attemptID)
        defer store.attemptIDMtx.Unlock(attemptID)

        log.Debugf("Subscribing to result for attemptID=%v", attemptID)

        var (
                result     *networkResult
                resultChan = make(chan *networkResult, 1)
        )

        err := kvdb.View(store.backend, func(tx kvdb.RTx) error {
                var err error
                result, err = fetchResult(tx, attemptID)
                switch {

                // Result not yet available, we will notify once a result is
                // available.
                case err == ErrPaymentIDNotFound:
                        return nil

                case err != nil:
                        return err

                // The result was found, and will be returned immediately.
                default:
                        return nil
                }
        }, func() {
                result = nil
        })
        if err != nil {
                return nil, err
        }

        // If the result was found, we can send it on the result channel
        // imemdiately.
        if result != nil {
                resultChan <- result
                return resultChan, nil
        }

        // Otherwise we store the result channel for when the result is
        // available.
        store.resultsMtx.Lock()
        store.results[attemptID] = append(
                store.results[attemptID], resultChan,
        )
        store.resultsMtx.Unlock()

        return resultChan, nil
}

// getResult attempts to immediately fetch the result for the given pid from
// the store. If no result is available, ErrPaymentIDNotFound is returned.
func (store *networkResultStore) getResult(pid uint64) (
        *networkResult, error) {

        var result *networkResult
        err := kvdb.View(store.backend, func(tx kvdb.RTx) error {
                var err error
                result, err = fetchResult(tx, pid)
                return err
        }, func() {
                result = nil
        })
        if err != nil {
                return nil, err
        }

        return result, nil
}

func fetchResult(tx kvdb.RTx, pid uint64) (*networkResult, error) {
        var attemptIDBytes [8]byte
        binary.BigEndian.PutUint64(attemptIDBytes[:], pid)

        networkResults := tx.ReadBucket(networkResultStoreBucketKey)
        if networkResults == nil {
                return nil, ErrPaymentIDNotFound
        }

        // Check whether a result is already available.
        resultBytes := networkResults.Get(attemptIDBytes[:])
        if resultBytes == nil {
                return nil, ErrPaymentIDNotFound
        }

        // Decode the result we found.
        r := bytes.NewReader(resultBytes)

        return deserializeNetworkResult(r)
}

// cleanStore removes all entries from the store, except the payment IDs given.
// NOTE: Since every result not listed in the keep map will be deleted, care
// should be taken to ensure no new payment attempts are being made
// concurrently while this process is ongoing, as its result might end up being
// deleted.
func (store *networkResultStore) cleanStore(keep map[uint64]struct{}) error {
        return kvdb.Update(store.backend, func(tx kvdb.RwTx) error {
                networkResults, err := tx.CreateTopLevelBucket(
                        networkResultStoreBucketKey,
                )
                if err != nil {
                        return err
                }

                // Iterate through the bucket, deleting all items not in the
                // keep map.
                var toClean [][]byte
                if err := networkResults.ForEach(func(k, _ []byte) error {
                        pid := binary.BigEndian.Uint64(k)
                        if _, ok := keep[pid]; ok {
                                return nil
                        }

                        toClean = append(toClean, k)
                        return nil
                }); err != nil {
                        return err
                }

                for _, k := range toClean {
                        err := networkResults.Delete(k)
                        if err != nil {
                                return err
                        }
                }

                if len(toClean) > 0 {
                        log.Infof("Removed %d stale entries from network "+
                                "result store", len(toClean))
                }

                return nil
        }, func() {})
}

1	package htlcswitch
2
3	import (
4	"bytes"
5	"encoding/binary"
6	"errors"
7	"io"
8	"sync"
9
10	"github.com/lightningnetwork/lnd/channeldb"
11	"github.com/lightningnetwork/lnd/kvdb"
12	"github.com/lightningnetwork/lnd/lnwire"
13	"github.com/lightningnetwork/lnd/multimutex"
14	)
15
16	var (
17
18	// networkResultStoreBucketKey is used for the root level bucket that
19	// stores the network result for each payment ID.
20	networkResultStoreBucketKey = []byte("network-result-store-bucket")
21
22	// ErrPaymentIDNotFound is an error returned if the given paymentID is
23	// not found.
24	ErrPaymentIDNotFound = errors.New("paymentID not found")
25
26	// ErrPaymentIDAlreadyExists is returned if we try to write a pending
27	// payment whose paymentID already exists.
28	ErrPaymentIDAlreadyExists = errors.New("paymentID already exists")
29	)
30
31	// PaymentResult wraps a decoded result received from the network after a
32	// payment attempt was made. This is what is eventually handed to the router
33	// for processing.
34	type PaymentResult struct {
35	// Preimage is set by the switch in case a sent HTLC was settled.
36	Preimage [32]byte
37
38	// Error is non-nil in case a HTLC send failed, and the HTLC is now
39	// irrevocably canceled. If the payment failed during forwarding, this
40	// error will be a *ForwardingError.
41	Error error
42
43	// EncryptedError will contain the raw bytes of an encrypted error
44	// in the event of a payment failure if the switch is instructed to
45	// defer error processing to external sub-systems.
46	EncryptedError []byte
47	}
48
49	// networkResult is the raw result received from the network after a payment
50	// attempt has been made. Since the switch doesn't always have the necessary
51	// data to decode the raw message, we store it together with some meta data,
52	// and decode it when the router query for the final result.
53	type networkResult struct {
54	// msg is the received result. This should be of type UpdateFulfillHTLC
55	// or UpdateFailHTLC.
56	msg lnwire.Message
57
58	// unencrypted indicates whether the failure encoded in the message is
59	// unencrypted, and hence doesn't need to be decrypted.
60	unencrypted bool
61
62	// isResolution indicates whether this is a resolution message, in
63	// which the failure reason might not be included.
64	isResolution bool
65	}
66
67	// serializeNetworkResult serializes the networkResult.
68	func serializeNetworkResult(w io.Writer, n *networkResult) error {	3✔
69	return channeldb.WriteElements(w, n.msg, n.unencrypted, n.isResolution)	3✔
70	}	3✔
71
72	// deserializeNetworkResult deserializes the networkResult.
73	func deserializeNetworkResult(r io.Reader) (*networkResult, error) {	3✔
74	n := &networkResult{}	3✔
75		3✔
76	if err := channeldb.ReadElements(r,	3✔
77	&n.msg, &n.unencrypted, &n.isResolution,	3✔
78	); err != nil {	3✔
79	return nil, err	×
80	}	×
81
82	return n, nil	3✔
83	}
84
85	// networkResultStore is a persistent store that stores any results of HTLCs in
86	// flight on the network. Since payment results are inherently asynchronous, it
87	// is used as a common access point for senders of HTLCs, to know when a result
88	// is back. The Switch will checkpoint any received result to the store, and
89	// the store will keep results and notify the callers about them.
90	type networkResultStore struct {
91	backend kvdb.Backend
92
93	// results is a map from paymentIDs to channels where subscribers to
94	// payment results will be notified.
95	results map[uint64][]chan *networkResult
96	resultsMtx sync.Mutex
97
98	// attemptIDMtx is a multimutex used to make sure the database and
99	// result subscribers map is consistent for each attempt ID in case of
100	// concurrent callers.
101	attemptIDMtx *multimutex.Mutex[uint64]
102	}
103
104	func newNetworkResultStore(db kvdb.Backend) *networkResultStore {	3✔
105	return &networkResultStore{	3✔
106	backend: db,	3✔
107	results: make(map[uint64][]chan *networkResult),	3✔
108	attemptIDMtx: multimutex.NewMutex[uint64](),	3✔
109	}	3✔
110	}	3✔
111
112	// storeResult stores the networkResult for the given attemptID, and notifies
113	// any subscribers.
114	func (store *networkResultStore) storeResult(attemptID uint64,
115	result *networkResult) error {	3✔
116		3✔
117	// We get a mutex for this attempt ID. This is needed to ensure	3✔
118	// consistency between the database state and the subscribers in case	3✔
119	// of concurrent calls.	3✔
120	store.attemptIDMtx.Lock(attemptID)	3✔
121	defer store.attemptIDMtx.Unlock(attemptID)	3✔
122		3✔
123	log.Debugf("Storing result for attemptID=%v", attemptID)	3✔
124		3✔
125	// Serialize the payment result.	3✔
126	var b bytes.Buffer	3✔
127	if err := serializeNetworkResult(&b, result); err != nil {	3✔
128	return err	×
129	}	×
130
131	var attemptIDBytes [8]byte	3✔
132	binary.BigEndian.PutUint64(attemptIDBytes[:], attemptID)	3✔
133		3✔
134	err := kvdb.Batch(store.backend, func(tx kvdb.RwTx) error {	6✔
135	networkResults, err := tx.CreateTopLevelBucket(	3✔
136	networkResultStoreBucketKey,	3✔
137	)	3✔
138	if err != nil {	3✔
139	return err	×
140	}	×
141
142	return networkResults.Put(attemptIDBytes[:], b.Bytes())	3✔
143	})
144	if err != nil {	3✔
145	return err	×
146	}	×
147
148	// Now that the result is stored in the database, we can notify any
149	// active subscribers.
150	store.resultsMtx.Lock()	3✔
151	for _, res := range store.results[attemptID] {	6✔
152	res <- result	3✔
153	}	3✔
154	delete(store.results, attemptID)	3✔
155	store.resultsMtx.Unlock()	3✔
156		3✔
157	return nil	3✔
158	}
159
160	// subscribeResult is used to get the HTLC attempt result for the given attempt
161	// ID. It returns a channel on which the result will be delivered when ready.
162	func (store *networkResultStore) subscribeResult(attemptID uint64) (
163	<-chan *networkResult, error) {	3✔
164		3✔
165	// We get a mutex for this payment ID. This is needed to ensure	3✔
166	// consistency between the database state and the subscribers in case	3✔
167	// of concurrent calls.	3✔
168	store.attemptIDMtx.Lock(attemptID)	3✔
169	defer store.attemptIDMtx.Unlock(attemptID)	3✔
170		3✔
171	log.Debugf("Subscribing to result for attemptID=%v", attemptID)	3✔
172		3✔
173	var (	3✔
174	result *networkResult	3✔
175	resultChan = make(chan *networkResult, 1)	3✔
176	)	3✔
177		3✔
178	err := kvdb.View(store.backend, func(tx kvdb.RTx) error {	6✔
179	var err error	3✔
180	result, err = fetchResult(tx, attemptID)	3✔
181	switch {	3✔
182
183	// Result not yet available, we will notify once a result is
184	// available.
185	case err == ErrPaymentIDNotFound:	3✔
186	return nil	3✔
187
188	case err != nil:	×
189	return err	×
190
191	// The result was found, and will be returned immediately.
UNCOV 192	default:	×
UNCOV 193	return nil	×
194	}
195	}, func() {	3✔
196	result = nil	3✔
197	})	3✔
198	if err != nil {	3✔
199	return nil, err	×
200	}	×
201
202	// If the result was found, we can send it on the result channel
203	// imemdiately.
204	if result != nil {	3✔
UNCOV 205	resultChan <- result	×
UNCOV 206	return resultChan, nil	×
UNCOV 207	}	×
208
209	// Otherwise we store the result channel for when the result is
210	// available.
211	store.resultsMtx.Lock()	3✔
212	store.results[attemptID] = append(	3✔
213	store.results[attemptID], resultChan,	3✔
214	)	3✔
215	store.resultsMtx.Unlock()	3✔
216		3✔
217	return resultChan, nil	3✔
218	}
219
220	// getResult attempts to immediately fetch the result for the given pid from
221	// the store. If no result is available, ErrPaymentIDNotFound is returned.
222	func (store *networkResultStore) getResult(pid uint64) (
223	*networkResult, error) {	3✔
224		3✔
225	var result *networkResult	3✔
226	err := kvdb.View(store.backend, func(tx kvdb.RTx) error {	6✔
227	var err error	3✔
228	result, err = fetchResult(tx, pid)	3✔
229	return err	3✔
230	}, func() {	6✔
231	result = nil	3✔
232	})	3✔
233	if err != nil {	6✔
234	return nil, err	3✔
235	}	3✔
236
237	return result, nil	3✔
238	}
239
240	func fetchResult(tx kvdb.RTx, pid uint64) (*networkResult, error) {	3✔
241	var attemptIDBytes [8]byte	3✔
242	binary.BigEndian.PutUint64(attemptIDBytes[:], pid)	3✔
243		3✔
244	networkResults := tx.ReadBucket(networkResultStoreBucketKey)	3✔
245	if networkResults == nil {	3✔
UNCOV 246	return nil, ErrPaymentIDNotFound	×
UNCOV 247	}	×
248
249	// Check whether a result is already available.
250	resultBytes := networkResults.Get(attemptIDBytes[:])	3✔
251	if resultBytes == nil {	6✔
252	return nil, ErrPaymentIDNotFound	3✔
253	}	3✔
254
255	// Decode the result we found.
256	r := bytes.NewReader(resultBytes)	3✔
257		3✔
258	return deserializeNetworkResult(r)	3✔
259	}
260
261	// cleanStore removes all entries from the store, except the payment IDs given.
262	// NOTE: Since every result not listed in the keep map will be deleted, care
263	// should be taken to ensure no new payment attempts are being made
264	// concurrently while this process is ongoing, as its result might end up being
265	// deleted.
266	func (store *networkResultStore) cleanStore(keep map[uint64]struct{}) error {	3✔
267	return kvdb.Update(store.backend, func(tx kvdb.RwTx) error {	6✔
268	networkResults, err := tx.CreateTopLevelBucket(	3✔
269	networkResultStoreBucketKey,	3✔
270	)	3✔
271	if err != nil {	3✔
272	return err	×
273	}	×
274
275	// Iterate through the bucket, deleting all items not in the
276	// keep map.
277	var toClean [][]byte	3✔
278	if err := networkResults.ForEach(func(k, _ []byte) error {	6✔
279	pid := binary.BigEndian.Uint64(k)	3✔
280	if _, ok := keep[pid]; ok {	3✔
UNCOV 281	return nil	×
UNCOV 282	}	×
283
284	toClean = append(toClean, k)	3✔
285	return nil	3✔
286	}); err != nil {	×
287	return err	×
288	}	×
289
290	for _, k := range toClean {	6✔
291	err := networkResults.Delete(k)	3✔
292	if err != nil {	3✔
293	return err	×
294	}	×
295	}
296
297	if len(toClean) > 0 {	6✔
298	log.Infof("Removed %d stale entries from network "+	3✔
299	"result store", len(toClean))	3✔
300	}	3✔
301
302	return nil	3✔
303	}, func() {})	3✔
304	}

lightningnetwork / lnd / 17777197509

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