• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

lightningnetwork / lnd / 12145425118

03 Dec 2024 05:56PM UTC coverage: 58.934% (-0.04%) from 58.972%
12145425118

push

github

web-flow
Merge pull request #9257 from starius/estimatefeerate-regtest2

chainreg: use feerate estimator in regtest and simnet

5 of 5 new or added lines in 1 file covered. (100.0%)

104 existing lines in 24 files now uncovered.

133434 of 226413 relevant lines covered (58.93%)

19484.82 hits per line

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

82.51
/routing/missioncontrol_store.go
1
package routing
2

3
import (
4
        "bytes"
5
        "container/list"
6
        "encoding/binary"
7
        "fmt"
8
        "io"
9
        "sync"
10
        "time"
11

12
        "github.com/lightningnetwork/lnd/kvdb"
13
        "github.com/lightningnetwork/lnd/lnwire"
14
        "github.com/lightningnetwork/lnd/tlv"
15
)
16

17
var (
18
        // resultsKey is the fixed key under which the attempt results are
19
        // stored.
20
        resultsKey = []byte("missioncontrol-results")
21

22
        // Big endian is the preferred byte order, due to cursor scans over
23
        // integer keys iterating in order.
24
        byteOrder = binary.BigEndian
25
)
26

27
// missionControlDB is an interface that defines the database methods that a
28
// single missionControlStore has access to. It allows the missionControlStore
29
// to be unaware of the overall DB structure and restricts its access to the DB
30
// by only providing it the bucket that it needs to care about.
31
type missionControlDB interface {
32
        // update can be used to perform reads and writes on the given bucket.
33
        update(f func(bkt kvdb.RwBucket) error, reset func()) error
34

35
        // view can be used to perform reads on the given bucket.
36
        view(f func(bkt kvdb.RBucket) error, reset func()) error
37

38
        // purge will delete all the contents in this store.
39
        purge() error
40
}
41

42
// missionControlStore is a bolt db based implementation of a mission control
43
// store. It stores the raw payment attempt data from which the internal mission
44
// controls state can be rederived on startup. This allows the mission control
45
// internal data structure to be changed without requiring a database migration.
46
// Also changes to mission control parameters can be applied to historical data.
47
// Finally, it enables importing raw data from an external source.
48
type missionControlStore struct {
49
        done chan struct{}
50
        wg   sync.WaitGroup
51
        db   missionControlDB
52

53
        // TODO(yy): Remove the usage of sync.Cond - we are better off using
54
        // channes than a Cond as suggested in the official godoc.
55
        //
56
        // queueCond is signalled when items are put into the queue.
57
        queueCond *sync.Cond
58

59
        // queue stores all pending payment results not yet added to the store.
60
        // Access is protected by the queueCond.L mutex.
61
        queue *list.List
62

63
        // keys holds the stored MC store item keys in the order of storage.
64
        // We use this list when adding/deleting items from the database to
65
        // avoid cursor use which may be slow in the remote DB case.
66
        keys *list.List
67

68
        // keysMap holds the stored MC store item keys. We use this map to check
69
        // if a new payment result has already been stored.
70
        keysMap map[string]struct{}
71

72
        // maxRecords is the maximum amount of records we will store in the db.
73
        maxRecords int
74

75
        // flushInterval is the configured interval we use to store new results
76
        // and delete outdated ones from the db.
77
        flushInterval time.Duration
78
}
79

80
func newMissionControlStore(db missionControlDB, maxRecords int,
81
        flushInterval time.Duration) (*missionControlStore, error) {
41✔
82

41✔
83
        var (
41✔
84
                keys    *list.List
41✔
85
                keysMap map[string]struct{}
41✔
86
        )
41✔
87

41✔
88
        // Create buckets if not yet existing.
41✔
89
        err := db.update(func(resultsBucket kvdb.RwBucket) error {
82✔
90
                // Collect all keys to be able to quickly calculate the
41✔
91
                // difference when updating the DB state.
41✔
92
                c := resultsBucket.ReadCursor()
41✔
93
                for k, _ := c.First(); k != nil; k, _ = c.Next() {
53✔
94
                        keys.PushBack(string(k))
12✔
95
                        keysMap[string(k)] = struct{}{}
12✔
96
                }
12✔
97

98
                return nil
41✔
99
        }, func() {
41✔
100
                keys = list.New()
41✔
101
                keysMap = make(map[string]struct{})
41✔
102
        })
41✔
103
        if err != nil {
41✔
104
                return nil, err
×
105
        }
×
106

107
        log.Infof("Loaded %d mission control entries", len(keysMap))
41✔
108

41✔
109
        return &missionControlStore{
41✔
110
                done:          make(chan struct{}),
41✔
111
                db:            db,
41✔
112
                queueCond:     sync.NewCond(&sync.Mutex{}),
41✔
113
                queue:         list.New(),
41✔
114
                keys:          keys,
41✔
115
                keysMap:       keysMap,
41✔
116
                maxRecords:    maxRecords,
41✔
117
                flushInterval: flushInterval,
41✔
118
        }, nil
41✔
119
}
120

121
// clear removes all results from the db.
122
func (b *missionControlStore) clear() error {
7✔
123
        b.queueCond.L.Lock()
7✔
124
        defer b.queueCond.L.Unlock()
7✔
125

7✔
126
        if err := b.db.purge(); err != nil {
7✔
127
                return err
×
128
        }
×
129

130
        b.queue = list.New()
7✔
131

7✔
132
        return nil
7✔
133
}
134

135
// fetchAll returns all results currently stored in the database.
136
func (b *missionControlStore) fetchAll() ([]*paymentResult, error) {
51✔
137
        var results []*paymentResult
51✔
138

51✔
139
        err := b.db.view(func(resultBucket kvdb.RBucket) error {
102✔
140
                results = make([]*paymentResult, 0)
51✔
141

51✔
142
                return resultBucket.ForEach(func(k, v []byte) error {
78✔
143
                        result, err := deserializeResult(k, v)
27✔
144
                        if err != nil {
27✔
145
                                return err
×
146
                        }
×
147

148
                        results = append(results, result)
27✔
149

27✔
150
                        return nil
27✔
151
                })
152

153
        }, func() {
51✔
154
                results = nil
51✔
155
        })
51✔
156
        if err != nil {
51✔
157
                return nil, err
×
158
        }
×
159

160
        return results, nil
51✔
161
}
162

163
// serializeResult serializes a payment result and returns a key and value byte
164
// slice to insert into the bucket.
165
func serializeResult(rp *paymentResult) ([]byte, []byte, error) {
17✔
166
        recordProducers := []tlv.RecordProducer{
17✔
167
                &rp.timeFwd,
17✔
168
                &rp.timeReply,
17✔
169
                &rp.route,
17✔
170
        }
17✔
171

17✔
172
        rp.failure.WhenSome(
17✔
173
                func(failure tlv.RecordT[tlv.TlvType3, paymentFailure]) {
32✔
174
                        recordProducers = append(recordProducers, &failure)
15✔
175
                },
15✔
176
        )
177

178
        // Compose key that identifies this result.
179
        key := getResultKey(rp)
17✔
180

17✔
181
        var buff bytes.Buffer
17✔
182
        err := lnwire.EncodeRecordsTo(
17✔
183
                &buff, lnwire.ProduceRecordsSorted(recordProducers...),
17✔
184
        )
17✔
185
        if err != nil {
17✔
186
                return nil, nil, err
×
187
        }
×
188

189
        return key, buff.Bytes(), err
17✔
190
}
191

192
// deserializeResult deserializes a payment result.
193
func deserializeResult(k, v []byte) (*paymentResult, error) {
27✔
194
        // Parse payment id.
27✔
195
        result := paymentResult{
27✔
196
                id: byteOrder.Uint64(k[8:]),
27✔
197
        }
27✔
198

27✔
199
        failure := tlv.ZeroRecordT[tlv.TlvType3, paymentFailure]()
27✔
200
        recordProducers := []tlv.RecordProducer{
27✔
201
                &result.timeFwd,
27✔
202
                &result.timeReply,
27✔
203
                &result.route,
27✔
204
                &failure,
27✔
205
        }
27✔
206

27✔
207
        r := bytes.NewReader(v)
27✔
208
        typeMap, err := lnwire.DecodeRecords(
27✔
209
                r, lnwire.ProduceRecordsSorted(recordProducers...)...,
27✔
210
        )
27✔
211
        if err != nil {
27✔
212
                return nil, err
×
213
        }
×
214

215
        if _, ok := typeMap[result.failure.TlvType()]; ok {
52✔
216
                result.failure = tlv.SomeRecordT(failure)
25✔
217
        }
25✔
218

219
        return &result, nil
27✔
220
}
221

222
// serializeRoute serializes a mcRoute and writes the resulting bytes to the
223
// given io.Writer.
224
func serializeRoute(w io.Writer, r *mcRoute) error {
30✔
225
        records := lnwire.ProduceRecordsSorted(
30✔
226
                &r.sourcePubKey,
30✔
227
                &r.totalAmount,
30✔
228
                &r.hops,
30✔
229
        )
30✔
230

30✔
231
        return lnwire.EncodeRecordsTo(w, records)
30✔
232
}
30✔
233

234
// deserializeRoute deserializes the mcRoute from the given io.Reader.
235
func deserializeRoute(r io.Reader) (*mcRoute, error) {
27✔
236
        var rt mcRoute
27✔
237
        records := lnwire.ProduceRecordsSorted(
27✔
238
                &rt.sourcePubKey,
27✔
239
                &rt.totalAmount,
27✔
240
                &rt.hops,
27✔
241
        )
27✔
242

27✔
243
        _, err := lnwire.DecodeRecords(r, records...)
27✔
244
        if err != nil {
27✔
245
                return nil, err
×
246
        }
×
247

248
        return &rt, nil
27✔
249
}
250

251
// deserializeHop deserializes the mcHop from the given io.Reader.
252
func deserializeHop(r io.Reader) (*mcHop, error) {
31✔
253
        var (
31✔
254
                h        mcHop
31✔
255
                blinding = tlv.ZeroRecordT[tlv.TlvType3, lnwire.TrueBoolean]()
31✔
256
                custom   = tlv.ZeroRecordT[tlv.TlvType4, lnwire.TrueBoolean]()
31✔
257
        )
31✔
258
        records := lnwire.ProduceRecordsSorted(
31✔
259
                &h.channelID,
31✔
260
                &h.pubKeyBytes,
31✔
261
                &h.amtToFwd,
31✔
262
                &blinding,
31✔
263
                &custom,
31✔
264
        )
31✔
265

31✔
266
        typeMap, err := lnwire.DecodeRecords(r, records...)
31✔
267
        if err != nil {
31✔
268
                return nil, err
×
269
        }
×
270

271
        if _, ok := typeMap[h.hasBlindingPoint.TlvType()]; ok {
35✔
272
                h.hasBlindingPoint = tlv.SomeRecordT(blinding)
4✔
273
        }
4✔
274

275
        if _, ok := typeMap[h.hasCustomRecords.TlvType()]; ok {
54✔
276
                h.hasCustomRecords = tlv.SomeRecordT(custom)
23✔
277
        }
23✔
278

279
        return &h, nil
31✔
280
}
281

282
// serializeHop serializes a mcHop and writes the resulting bytes to the given
283
// io.Writer.
284
func serializeHop(w io.Writer, h *mcHop) error {
72✔
285
        recordProducers := []tlv.RecordProducer{
72✔
286
                &h.channelID,
72✔
287
                &h.pubKeyBytes,
72✔
288
                &h.amtToFwd,
72✔
289
        }
72✔
290

72✔
291
        h.hasBlindingPoint.WhenSome(func(
72✔
292
                hasBlinding tlv.RecordT[tlv.TlvType3, lnwire.TrueBoolean]) {
76✔
293

4✔
294
                recordProducers = append(recordProducers, &hasBlinding)
4✔
295
        })
4✔
296

297
        h.hasCustomRecords.WhenSome(func(
72✔
298
                hasCustom tlv.RecordT[tlv.TlvType4, lnwire.TrueBoolean]) {
112✔
299

40✔
300
                recordProducers = append(recordProducers, &hasCustom)
40✔
301
        })
40✔
302

303
        return lnwire.EncodeRecordsTo(
72✔
304
                w, lnwire.ProduceRecordsSorted(recordProducers...),
72✔
305
        )
72✔
306
}
307

308
// AddResult adds a new result to the db.
309
func (b *missionControlStore) AddResult(rp *paymentResult) {
82✔
310
        b.queueCond.L.Lock()
82✔
311
        b.queue.PushBack(rp)
82✔
312
        b.queueCond.L.Unlock()
82✔
313

82✔
314
        b.queueCond.Signal()
82✔
315
}
82✔
316

317
// stop stops the store ticker goroutine.
318
func (b *missionControlStore) stop() {
6✔
319
        close(b.done)
6✔
320

6✔
321
        b.queueCond.Signal()
6✔
322

6✔
323
        b.wg.Wait()
6✔
324
}
6✔
325

326
// run runs the MC store ticker goroutine.
327
func (b *missionControlStore) run() {
6✔
328
        b.wg.Add(1)
6✔
329

6✔
330
        go func() {
12✔
331
                defer b.wg.Done()
6✔
332

6✔
333
                timer := time.NewTimer(b.flushInterval)
6✔
334

6✔
335
                // Immediately stop the timer. It will be started once new
6✔
336
                // items are added to the store. As the doc for time.Timer
6✔
337
                // states, every call to Stop() done on a timer that is not
6✔
338
                // known to have been fired needs to be checked and the timer's
6✔
339
                // channel needs to be drained appropriately. This could happen
6✔
340
                // if the flushInterval is very small (e.g. 1 nanosecond).
6✔
341
                if !timer.Stop() {
6✔
342
                        select {
×
343
                        case <-timer.C:
×
344
                        case <-b.done:
×
345
                                log.Debugf("Stopping mission control store")
×
346
                        }
347
                }
348

349
                for {
15✔
350
                        // Wait for the queue to not be empty.
9✔
351
                        b.queueCond.L.Lock()
9✔
352
                        for b.queue.Front() == nil {
20✔
353
                                // To make sure we can properly stop, we must
11✔
354
                                // read the `done` channel first before
11✔
355
                                // attempting to call `Wait()`. This is due to
11✔
356
                                // the fact when `Signal` is called before the
11✔
357
                                // `Wait` call, the `Wait` call will block
11✔
358
                                // indefinitely.
11✔
359
                                //
11✔
360
                                // TODO(yy): replace this with channels.
11✔
361
                                select {
11✔
362
                                case <-b.done:
6✔
363
                                        b.queueCond.L.Unlock()
6✔
364

6✔
365
                                        return
6✔
366
                                default:
9✔
367
                                }
368

369
                                b.queueCond.Wait()
9✔
370
                        }
371
                        b.queueCond.L.Unlock()
7✔
372

7✔
373
                        // Restart the timer.
7✔
374
                        timer.Reset(b.flushInterval)
7✔
375

7✔
376
                        select {
7✔
377
                        case <-timer.C:
7✔
378
                                if err := b.storeResults(); err != nil {
7✔
379
                                        log.Errorf("Failed to update mission "+
×
380
                                                "control store: %v", err)
×
381
                                }
×
382

383
                        case <-b.done:
4✔
384
                                // Release the timer's resources.
4✔
385
                                if !timer.Stop() {
4✔
UNCOV
386
                                        select {
×
UNCOV
387
                                        case <-timer.C:
×
388
                                        case <-b.done:
×
389
                                                log.Debugf("Mission control " +
×
390
                                                        "store stopped")
×
391
                                        }
392
                                }
393
                                return
4✔
394
                        }
395
                }
396
        }()
397
}
398

399
// storeResults stores all accumulated results.
400
func (b *missionControlStore) storeResults() error {
13✔
401
        // We copy a reference to the queue and clear the original queue to be
13✔
402
        // able to release the lock.
13✔
403
        b.queueCond.L.Lock()
13✔
404
        l := b.queue
13✔
405

13✔
406
        if l.Len() == 0 {
13✔
407
                b.queueCond.L.Unlock()
×
408

×
409
                return nil
×
410
        }
×
411
        b.queue = list.New()
13✔
412
        b.queueCond.L.Unlock()
13✔
413

13✔
414
        var (
13✔
415
                newKeys    map[string]struct{}
13✔
416
                delKeys    []string
13✔
417
                storeCount int
13✔
418
                pruneCount int
13✔
419
        )
13✔
420

13✔
421
        // Create a deduped list of new entries.
13✔
422
        newKeys = make(map[string]struct{}, l.Len())
13✔
423
        for e := l.Front(); e != nil; e = e.Next() {
30✔
424
                pr, ok := e.Value.(*paymentResult)
17✔
425
                if !ok {
17✔
426
                        return fmt.Errorf("wrong type %T (not *paymentResult)",
×
427
                                e.Value)
×
428
                }
×
429
                key := string(getResultKey(pr))
17✔
430
                if _, ok := b.keysMap[key]; ok {
17✔
431
                        l.Remove(e)
×
432
                        continue
×
433
                }
434
                if _, ok := newKeys[key]; ok {
18✔
435
                        l.Remove(e)
1✔
436
                        continue
1✔
437
                }
438
                newKeys[key] = struct{}{}
16✔
439
        }
440

441
        // Create a list of entries to delete.
442
        toDelete := b.keys.Len() + len(newKeys) - b.maxRecords
13✔
443
        if b.maxRecords > 0 && toDelete > 0 {
17✔
444
                delKeys = make([]string, 0, toDelete)
4✔
445

4✔
446
                // Delete as many as needed from old keys.
4✔
447
                for e := b.keys.Front(); len(delKeys) < toDelete && e != nil; {
9✔
448
                        key, ok := e.Value.(string)
5✔
449
                        if !ok {
5✔
450
                                return fmt.Errorf("wrong type %T (not string)",
×
451
                                        e.Value)
×
452
                        }
×
453
                        delKeys = append(delKeys, key)
5✔
454
                        e = e.Next()
5✔
455
                }
456

457
                // If more deletions are needed, simply do not add from the
458
                // list of new keys.
459
                for e := l.Front(); len(delKeys) < toDelete && e != nil; {
4✔
460
                        toDelete--
×
461
                        pr, ok := e.Value.(*paymentResult)
×
462
                        if !ok {
×
463
                                return fmt.Errorf("wrong type %T (not "+
×
464
                                        "*paymentResult )", e.Value)
×
465
                        }
×
466
                        key := string(getResultKey(pr))
×
467
                        delete(newKeys, key)
×
468
                        l.Remove(e)
×
469
                        e = l.Front()
×
470
                }
471
        }
472

473
        err := b.db.update(func(bucket kvdb.RwBucket) error {
26✔
474
                for e := l.Front(); e != nil; e = e.Next() {
30✔
475
                        pr, ok := e.Value.(*paymentResult)
17✔
476
                        if !ok {
17✔
477
                                return fmt.Errorf("wrong type %T (not "+
×
478
                                        "*paymentResult)", e.Value)
×
479
                        }
×
480

481
                        // Serialize result into key and value byte slices.
482
                        k, v, err := serializeResult(pr)
17✔
483
                        if err != nil {
17✔
484
                                return err
×
485
                        }
×
486

487
                        // Put into results bucket.
488
                        if err := bucket.Put(k, v); err != nil {
17✔
489
                                return err
×
490
                        }
×
491

492
                        storeCount++
17✔
493
                }
494

495
                // Prune oldest entries.
496
                for _, key := range delKeys {
18✔
497
                        if err := bucket.Delete([]byte(key)); err != nil {
5✔
498
                                return err
×
499
                        }
×
500
                        pruneCount++
5✔
501
                }
502

503
                return nil
13✔
504
        }, func() {
13✔
505
                storeCount, pruneCount = 0, 0
13✔
506
        })
13✔
507

508
        if err != nil {
13✔
509
                return err
×
510
        }
×
511

512
        log.Debugf("Stored mission control results: %d added, %d deleted",
13✔
513
                storeCount, pruneCount)
13✔
514

13✔
515
        // DB Update was successful, update the in-memory cache.
13✔
516
        for _, key := range delKeys {
18✔
517
                delete(b.keysMap, key)
5✔
518
                b.keys.Remove(b.keys.Front())
5✔
519
        }
5✔
520
        for e := l.Front(); e != nil; e = e.Next() {
30✔
521
                pr, ok := e.Value.(*paymentResult)
17✔
522
                if !ok {
17✔
523
                        return fmt.Errorf("wrong type %T (not *paymentResult)",
×
524
                                e.Value)
×
525
                }
×
526
                key := string(getResultKey(pr))
17✔
527
                b.keys.PushBack(key)
17✔
528
        }
529

530
        return nil
13✔
531
}
532

533
// getResultKey returns a byte slice representing a unique key for this payment
534
// result.
535
func getResultKey(rp *paymentResult) []byte {
43✔
536
        var keyBytes [8 + 8 + 33]byte
43✔
537

43✔
538
        // Identify records by a combination of time, payment id and sender pub
43✔
539
        // key. This allows importing mission control data from an external
43✔
540
        // source without key collisions and keeps the records sorted
43✔
541
        // chronologically.
43✔
542
        byteOrder.PutUint64(keyBytes[:], rp.timeReply.Val)
43✔
543
        byteOrder.PutUint64(keyBytes[8:], rp.id)
43✔
544
        copy(keyBytes[16:], rp.route.Val.sourcePubKey.Val[:])
43✔
545

43✔
546
        return keyBytes[:]
43✔
547
}
43✔
548

549
// failureMessage wraps the lnwire.FailureMessage interface such that we can
550
// apply a Record method and use the failureMessage in a TLV encoded type.
551
type failureMessage struct {
552
        lnwire.FailureMessage
553
}
554

555
// Record returns a TLV record that can be used to encode/decode a list of
556
// failureMessage to/from a TLV stream.
557
func (r *failureMessage) Record() tlv.Record {
69✔
558
        recordSize := func() uint64 {
117✔
559
                var (
48✔
560
                        b   bytes.Buffer
48✔
561
                        buf [8]byte
48✔
562
                )
48✔
563
                if err := encodeFailureMessage(&b, r, &buf); err != nil {
48✔
564
                        panic(err)
×
565
                }
566

567
                return uint64(len(b.Bytes()))
48✔
568
        }
569

570
        return tlv.MakeDynamicRecord(
69✔
571
                0, r, recordSize, encodeFailureMessage, decodeFailureMessage,
69✔
572
        )
69✔
573
}
574

575
func encodeFailureMessage(w io.Writer, val interface{}, _ *[8]byte) error {
92✔
576
        if v, ok := val.(*failureMessage); ok {
184✔
577
                var b bytes.Buffer
92✔
578
                err := lnwire.EncodeFailureMessage(&b, v.FailureMessage, 0)
92✔
579
                if err != nil {
92✔
580
                        return err
×
581
                }
×
582

583
                _, err = w.Write(b.Bytes())
92✔
584

92✔
585
                return err
92✔
586
        }
587

588
        return tlv.NewTypeForEncodingErr(val, "routing.failureMessage")
×
589
}
590

591
func decodeFailureMessage(r io.Reader, val interface{}, _ *[8]byte,
592
        l uint64) error {
25✔
593

25✔
594
        if v, ok := val.(*failureMessage); ok {
50✔
595
                msg, err := lnwire.DecodeFailureMessage(r, 0)
25✔
596
                if err != nil {
25✔
597
                        return err
×
598
                }
×
599

600
                *v = failureMessage{
25✔
601
                        FailureMessage: msg,
25✔
602
                }
25✔
603

25✔
604
                return nil
25✔
605
        }
606

607
        return tlv.NewTypeForDecodingErr(val, "routing.failureMessage", l, l)
×
608
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc