19924300449

Committed 04 Dec 2025 09:35AM UTC coverage: 53.479% (-1.9%) from 55.404%

Build # 19924300449

Build Type

Pull #10419

github

Committed by

web-flow

Commit Message

Merge f811805c6 into 20473482d

Pull Request Pull Request #10419: [docs] Document use-native-sql=true for SQL migration step 2

Run Details

110496 of 206616 relevant lines covered (53.48%)

21221.61 hits per line

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

81.76

/routing/missioncontrol_store.go

package routing

import (
        "bytes"
        "container/list"
        "encoding/binary"
        "fmt"
        "io"
        "sync"
        "time"

        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/tlv"
)

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

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

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

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

        // purge will delete all the contents in this store.
        purge() error
}

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

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

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

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

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

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

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

func newMissionControlStore(db missionControlDB, maxRecords int,
        flushInterval time.Duration) (*missionControlStore, error) {

        var (
                keys    *list.List
                keysMap map[string]struct{}
        )

        // Create buckets if not yet existing.
        err := db.update(func(resultsBucket kvdb.RwBucket) error {
                // Collect all keys to be able to quickly calculate the
                // difference when updating the DB state.
                c := resultsBucket.ReadCursor()
                for k, _ := c.First(); k != nil; k, _ = c.Next() {
                        keys.PushBack(string(k))
                        keysMap[string(k)] = struct{}{}
                }

                return nil
        }, func() {
                keys = list.New()
                keysMap = make(map[string]struct{})
        })
        if err != nil {
                return nil, err
        }

        log.Infof("Loaded %d mission control entries", len(keysMap))

        return &missionControlStore{
                done:          make(chan struct{}),
                db:            db,
                queueCond:     sync.NewCond(&sync.Mutex{}),
                queue:         list.New(),
                keys:          keys,
                keysMap:       keysMap,
                maxRecords:    maxRecords,
                flushInterval: flushInterval,
        }, nil
}

// clear removes all results from the db.
func (b *missionControlStore) clear() error {
        b.queueCond.L.Lock()
        defer b.queueCond.L.Unlock()

        if err := b.db.purge(); err != nil {
                return err
        }

        b.queue = list.New()

        return nil
}

// fetchAll returns all results currently stored in the database.
// It also removes any corrupted entries that fail to deserialize from both
// the database and the in-memory tracking structures.
func (b *missionControlStore) fetchAll() ([]*paymentResult, error) {
        var results []*paymentResult
        var corruptedKeys [][]byte

        // Read all results and identify corrupted entries.
        err := b.db.view(func(resultBucket kvdb.RBucket) error {
                results = make([]*paymentResult, 0)
                corruptedKeys = make([][]byte, 0)

                err := resultBucket.ForEach(func(k, v []byte) error {
                        result, err := deserializeResult(k, v)

                        // In case of an error, track the key for removal.
                        if err != nil {
                                log.Warnf("Failed to deserialize mission "+
                                        "control entry (key=%x): %v", k, err)

                                // Make a copy of the key since ForEach reuses
                                // the slice.
                                keyCopy := make([]byte, len(k))
                                copy(keyCopy, k)
                                corruptedKeys = append(corruptedKeys, keyCopy)

                                return nil
                        }

                        results = append(results, result)

                        return nil
                })
                if err != nil {
                        return err
                }

                return nil
        }, func() {
                results = nil
                corruptedKeys = nil
        })
        if err != nil {
                return nil, err
        }

        // Delete corrupted entries from the database which were identified
        // when loading the results from the database.
        //
        // TODO: This code part should eventually be removed once we move the
        // mission control store to a native sql database and have to do a
        // full migration of the data.
        if len(corruptedKeys) > 0 {
                err = b.db.update(func(resultBucket kvdb.RwBucket) error {
                        for _, key := range corruptedKeys {
                                if err := resultBucket.Delete(key); err != nil {
                                        return fmt.Errorf("failed to delete "+
                                                "corrupted entry: %w", err)
                                }
                        }

                        return nil
                }, func() {})
                if err != nil {
                        return nil, err
                }

                // Build a set of corrupted keys.
                corruptedSet := make(map[string]struct{}, len(corruptedKeys))
                for _, key := range corruptedKeys {
                        corruptedSet[string(key)] = struct{}{}
                }

                // Remove corrupted keys from in-memory map.
                for keyStr := range corruptedSet {
                        delete(b.keysMap, keyStr)
                }

                // Remove from the keys list in a single pass.
                for e := b.keys.Front(); e != nil; {
                        next := e.Next()
                        keyVal, ok := e.Value.(string)
                        if ok {
                                _, isCorrupted := corruptedSet[keyVal]
                                if isCorrupted {
                                        b.keys.Remove(e)
                                }
                        }
                        e = next
                }

                log.Infof("Removed %d corrupted mission control entries",
                        len(corruptedKeys))
        }

        return results, nil
}

// serializeResult serializes a payment result and returns a key and value byte
// slice to insert into the bucket.
func serializeResult(rp *paymentResult) ([]byte, []byte, error) {
        recordProducers := []tlv.RecordProducer{
                &rp.timeFwd,
                &rp.timeReply,
                &rp.route,
        }

        rp.failure.WhenSome(
                func(failure tlv.RecordT[tlv.TlvType3, paymentFailure]) {
                        recordProducers = append(recordProducers, &failure)
                },
        )

        // Compose key that identifies this result.
        key := getResultKey(rp)

        var buff bytes.Buffer
        err := lnwire.EncodeRecordsTo(
                &buff, lnwire.ProduceRecordsSorted(recordProducers...),
        )
        if err != nil {
                return nil, nil, err
        }

        return key, buff.Bytes(), nil
}

// deserializeResult deserializes a payment result.
func deserializeResult(k, v []byte) (*paymentResult, error) {
        // Parse payment id.
        result := paymentResult{
                id: byteOrder.Uint64(k[8:]),
        }

        failure := tlv.ZeroRecordT[tlv.TlvType3, paymentFailure]()
        recordProducers := []tlv.RecordProducer{
                &result.timeFwd,
                &result.timeReply,
                &result.route,
                &failure,
        }

        r := bytes.NewReader(v)
        typeMap, err := lnwire.DecodeRecords(
                r, lnwire.ProduceRecordsSorted(recordProducers...)...,
        )
        if err != nil {
                return nil, err
        }

        if _, ok := typeMap[result.failure.TlvType()]; ok {
                result.failure = tlv.SomeRecordT(failure)
        }

        return &result, nil
}

// serializeRoute serializes a mcRoute and writes the resulting bytes to the
// given io.Writer.
func serializeRoute(w io.Writer, r *mcRoute) error {
        records := lnwire.ProduceRecordsSorted(
                &r.sourcePubKey,
                &r.totalAmount,
                &r.hops,
        )

        return lnwire.EncodeRecordsTo(w, records)
}

// deserializeRoute deserializes the mcRoute from the given io.Reader.
func deserializeRoute(r io.Reader) (*mcRoute, error) {
        var rt mcRoute
        records := lnwire.ProduceRecordsSorted(
                &rt.sourcePubKey,
                &rt.totalAmount,
                &rt.hops,
        )

        _, err := lnwire.DecodeRecords(r, records...)
        if err != nil {
                return nil, err
        }

        return &rt, nil
}

// deserializeHop deserializes the mcHop from the given io.Reader.
func deserializeHop(r io.Reader) (*mcHop, error) {
        var (
                h        mcHop
                blinding = tlv.ZeroRecordT[tlv.TlvType3, lnwire.TrueBoolean]()
                custom   = tlv.ZeroRecordT[tlv.TlvType4, lnwire.TrueBoolean]()
        )
        records := lnwire.ProduceRecordsSorted(
                &h.channelID,
                &h.pubKeyBytes,
                &h.amtToFwd,
                &blinding,
                &custom,
        )

        typeMap, err := lnwire.DecodeRecords(r, records...)
        if err != nil {
                return nil, err
        }

        if _, ok := typeMap[h.hasBlindingPoint.TlvType()]; ok {
                h.hasBlindingPoint = tlv.SomeRecordT(blinding)
        }

        if _, ok := typeMap[h.hasCustomRecords.TlvType()]; ok {
                h.hasCustomRecords = tlv.SomeRecordT(custom)
        }

        return &h, nil
}

// serializeHop serializes a mcHop and writes the resulting bytes to the given
// io.Writer.
func serializeHop(w io.Writer, h *mcHop) error {
        recordProducers := []tlv.RecordProducer{
                &h.channelID,
                &h.pubKeyBytes,
                &h.amtToFwd,
        }

        h.hasBlindingPoint.WhenSome(func(
                hasBlinding tlv.RecordT[tlv.TlvType3, lnwire.TrueBoolean]) {

                recordProducers = append(recordProducers, &hasBlinding)
        })

        h.hasCustomRecords.WhenSome(func(
                hasCustom tlv.RecordT[tlv.TlvType4, lnwire.TrueBoolean]) {

                recordProducers = append(recordProducers, &hasCustom)
        })

        return lnwire.EncodeRecordsTo(
                w, lnwire.ProduceRecordsSorted(recordProducers...),
        )
}

// AddResult adds a new result to the db.
func (b *missionControlStore) AddResult(rp *paymentResult) {
        b.queueCond.L.Lock()
        b.queue.PushBack(rp)
        b.queueCond.L.Unlock()

        b.queueCond.Signal()
}

// stop stops the store ticker goroutine.
func (b *missionControlStore) stop() {
        close(b.done)

        b.queueCond.Signal()

        b.wg.Wait()
}

// run runs the MC store ticker goroutine.
func (b *missionControlStore) run() {
        b.wg.Add(1)

        go func() {
                defer b.wg.Done()

                timer := time.NewTimer(b.flushInterval)

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

                for {
                        // Wait for the queue to not be empty.
                        b.queueCond.L.Lock()
                        for b.queue.Front() == nil {
                                // To make sure we can properly stop, we must
                                // read the `done` channel first before
                                // attempting to call `Wait()`. This is due to
                                // the fact when `Signal` is called before the
                                // `Wait` call, the `Wait` call will block
                                // indefinitely.
                                //
                                // TODO(yy): replace this with channels.
                                select {
                                case <-b.done:
                                        b.queueCond.L.Unlock()

                                        return
                                default:
                                }

                                b.queueCond.Wait()
                        }
                        b.queueCond.L.Unlock()

                        // Restart the timer.
                        timer.Reset(b.flushInterval)

                        select {
                        case <-timer.C:
                                if err := b.storeResults(); err != nil {
                                        log.Errorf("Failed to update mission "+
                                                "control store: %v", err)
                                }

                        case <-b.done:
                                // Release the timer's resources.
                                if !timer.Stop() {
                                        select {
                                        case <-timer.C:
                                        case <-b.done:
                                                log.Debugf("Mission control " +
                                                        "store stopped")
                                        }
                                }
                                return
                        }
                }
        }()
}

// storeResults stores all accumulated results.
func (b *missionControlStore) storeResults() error {
        // We copy a reference to the queue and clear the original queue to be
        // able to release the lock.
        b.queueCond.L.Lock()
        l := b.queue

        if l.Len() == 0 {
                b.queueCond.L.Unlock()

                return nil
        }
        b.queue = list.New()
        b.queueCond.L.Unlock()

        var (
                newKeys    map[string]struct{}
                delKeys    []string
                storeCount int
                pruneCount int
        )

        // Create a deduped list of new entries.
        newKeys = make(map[string]struct{}, l.Len())
        for e := l.Front(); e != nil; e = e.Next() {
                pr, ok := e.Value.(*paymentResult)
                if !ok {
                        return fmt.Errorf("wrong type %T (not *paymentResult)",
                                e.Value)
                }
                key := string(getResultKey(pr))
                if _, ok := b.keysMap[key]; ok {
                        l.Remove(e)
                        continue
                }
                if _, ok := newKeys[key]; ok {
                        l.Remove(e)
                        continue
                }
                newKeys[key] = struct{}{}
        }

        // Create a list of entries to delete.
        toDelete := b.keys.Len() + len(newKeys) - b.maxRecords
        if b.maxRecords > 0 && toDelete > 0 {
                delKeys = make([]string, 0, toDelete)

                // Delete as many as needed from old keys.
                for e := b.keys.Front(); len(delKeys) < toDelete && e != nil; {
                        key, ok := e.Value.(string)
                        if !ok {
                                return fmt.Errorf("wrong type %T (not string)",
                                        e.Value)
                        }
                        delKeys = append(delKeys, key)
                        e = e.Next()
                }

                // If more deletions are needed, simply do not add from the
                // list of new keys.
                for e := l.Front(); len(delKeys) < toDelete && e != nil; {
                        toDelete--
                        pr, ok := e.Value.(*paymentResult)
                        if !ok {
                                return fmt.Errorf("wrong type %T (not "+
                                        "*paymentResult )", e.Value)
                        }
                        key := string(getResultKey(pr))
                        delete(newKeys, key)
                        l.Remove(e)
                        e = l.Front()
                }
        }

        err := b.db.update(func(bucket kvdb.RwBucket) error {
                for e := l.Front(); e != nil; e = e.Next() {
                        pr, ok := e.Value.(*paymentResult)
                        if !ok {
                                return fmt.Errorf("wrong type %T (not "+
                                        "*paymentResult)", e.Value)
                        }

                        // Serialize result into key and value byte slices.
                        k, v, err := serializeResult(pr)
                        if err != nil {
                                return err
                        }

                        // Put into results bucket.
                        if err := bucket.Put(k, v); err != nil {
                                return err
                        }

                        storeCount++
                }

                // Prune oldest entries.
                for _, key := range delKeys {
                        if err := bucket.Delete([]byte(key)); err != nil {
                                return err
                        }
                        pruneCount++
                }

                return nil
        }, func() {
                storeCount, pruneCount = 0, 0
        })

        if err != nil {
                return err
        }

        log.Debugf("Stored mission control results: %d added, %d deleted",
                storeCount, pruneCount)

        // DB Update was successful, update the in-memory cache.
        for _, key := range delKeys {
                delete(b.keysMap, key)
                b.keys.Remove(b.keys.Front())
        }
        for e := l.Front(); e != nil; e = e.Next() {
                pr, ok := e.Value.(*paymentResult)
                if !ok {
                        return fmt.Errorf("wrong type %T (not *paymentResult)",
                                e.Value)
                }
                key := string(getResultKey(pr))
                b.keys.PushBack(key)
        }

        return nil
}

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

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

        return keyBytes[:]
}

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

// Record returns a TLV record that can be used to encode/decode a list of
// failureMessage to/from a TLV stream.
func (r *failureMessage) Record() tlv.Record {
        recordSize := func() uint64 {
                var (
                        b   bytes.Buffer
                        buf [8]byte
                )
                if err := encodeFailureMessage(&b, r, &buf); err != nil {
                        panic(err)
                }

                return uint64(len(b.Bytes()))
        }

        return tlv.MakeDynamicRecord(
                0, r, recordSize, encodeFailureMessage, decodeFailureMessage,
        )
}

func encodeFailureMessage(w io.Writer, val interface{}, _ *[8]byte) error {
        if v, ok := val.(*failureMessage); ok {
                var b bytes.Buffer
                err := lnwire.EncodeFailureMessage(&b, v.FailureMessage, 0)
                if err != nil {
                        return err
                }

                _, err = w.Write(b.Bytes())

                return err
        }

        return tlv.NewTypeForEncodingErr(val, "routing.failureMessage")
}

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

        if v, ok := val.(*failureMessage); ok {
                msg, err := lnwire.DecodeFailureMessage(r, 0)
                if err != nil {
                        return err
                }

                *v = failureMessage{
                        FailureMessage: msg,
                }

                return nil
        }

        return tlv.NewTypeForDecodingErr(val, "routing.failureMessage", l, l)
}

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) {	38✔
82		38✔
83	var (	38✔
84	keys *list.List	38✔
85	keysMap map[string]struct{}	38✔
86	)	38✔
87		38✔
88	// Create buckets if not yet existing.	38✔
89	err := db.update(func(resultsBucket kvdb.RwBucket) error {	76✔
90	// Collect all keys to be able to quickly calculate the	38✔
91	// difference when updating the DB state.	38✔
92	c := resultsBucket.ReadCursor()	38✔
93	for k, _ := c.First(); k != nil; k, _ = c.Next() {	46✔
94	keys.PushBack(string(k))	8✔
95	keysMap[string(k)] = struct{}{}	8✔
96	}	8✔
97
98	return nil	38✔
99	}, func() {	38✔
100	keys = list.New()	38✔
101	keysMap = make(map[string]struct{})	38✔
102	})	38✔
103	if err != nil {	38✔
104	return nil, err	×
105	}	×
106
107	log.Infof("Loaded %d mission control entries", len(keysMap))	38✔
108		38✔
109	return &missionControlStore{	38✔
110	done: make(chan struct{}),	38✔
111	db: db,	38✔
112	queueCond: sync.NewCond(&sync.Mutex{}),	38✔
113	queue: list.New(),	38✔
114	keys: keys,	38✔
115	keysMap: keysMap,	38✔
116	maxRecords: maxRecords,	38✔
117	flushInterval: flushInterval,	38✔
118	}, nil	38✔
119	}
120
121	// clear removes all results from the db.
122	func (b *missionControlStore) clear() error {	3✔
123	b.queueCond.L.Lock()	3✔
124	defer b.queueCond.L.Unlock()	3✔
125		3✔
126	if err := b.db.purge(); err != nil {	3✔
127	return err	×
128	}	×
129
130	b.queue = list.New()	3✔
131		3✔
132	return nil	3✔
133	}
134
135	// fetchAll returns all results currently stored in the database.
136	// It also removes any corrupted entries that fail to deserialize from both
137	// the database and the in-memory tracking structures.
138	func (b missionControlStore) fetchAll() ([]paymentResult, error) {	48✔
139	var results []*paymentResult	48✔
140	var corruptedKeys [][]byte	48✔
141		48✔
142	// Read all results and identify corrupted entries.	48✔
143	err := b.db.view(func(resultBucket kvdb.RBucket) error {	96✔
144	results = make([]*paymentResult, 0)	48✔
145	corruptedKeys = make([][]byte, 0)	48✔
146		48✔
147	err := resultBucket.ForEach(func(k, v []byte) error {	74✔
148	result, err := deserializeResult(k, v)	26✔
149		26✔
150	// In case of an error, track the key for removal.	26✔
151	if err != nil {	27✔
152	log.Warnf("Failed to deserialize mission "+	1✔
153	"control entry (key=%x): %v", k, err)	1✔
154		1✔
155	// Make a copy of the key since ForEach reuses	1✔
156	// the slice.	1✔
157	keyCopy := make([]byte, len(k))	1✔
158	copy(keyCopy, k)	1✔
159	corruptedKeys = append(corruptedKeys, keyCopy)	1✔
160		1✔
161	return nil	1✔
162	}	1✔
163
164	results = append(results, result)	25✔
165		25✔
166	return nil	25✔
167	})
168	if err != nil {	48✔
169	return err	×
170	}	×
171
172	return nil	48✔
173	}, func() {	48✔
174	results = nil	48✔
175	corruptedKeys = nil	48✔
176	})	48✔
177	if err != nil {	48✔
178	return nil, err	×
179	}	×
180
181	// Delete corrupted entries from the database which were identified
182	// when loading the results from the database.
183	//
184	// TODO: This code part should eventually be removed once we move the
185	// mission control store to a native sql database and have to do a
186	// full migration of the data.
187	if len(corruptedKeys) > 0 {	49✔
188	err = b.db.update(func(resultBucket kvdb.RwBucket) error {	2✔
189	for _, key := range corruptedKeys {	2✔
190	if err := resultBucket.Delete(key); err != nil {	1✔
191	return fmt.Errorf("failed to delete "+	×
192	"corrupted entry: %w", err)	×
193	}	×
194	}
195
196	return nil	1✔
197	}, func() {})	1✔
198	if err != nil {	1✔
199	return nil, err	×
200	}	×
201
202	// Build a set of corrupted keys.
203	corruptedSet := make(map[string]struct{}, len(corruptedKeys))	1✔
204	for _, key := range corruptedKeys {	2✔
205	corruptedSet[string(key)] = struct{}{}	1✔
206	}	1✔
207
208	// Remove corrupted keys from in-memory map.
209	for keyStr := range corruptedSet {	2✔
210	delete(b.keysMap, keyStr)	1✔
211	}	1✔
212
213	// Remove from the keys list in a single pass.
214	for e := b.keys.Front(); e != nil; {	4✔
215	next := e.Next()	3✔
216	keyVal, ok := e.Value.(string)	3✔
217	if ok {	6✔
218	_, isCorrupted := corruptedSet[keyVal]	3✔
219	if isCorrupted {	4✔
220	b.keys.Remove(e)	1✔
221	}	1✔
222	}
223	e = next	3✔
224	}
225
226	log.Infof("Removed %d corrupted mission control entries",	1✔
227	len(corruptedKeys))	1✔
228	}
229
230	return results, nil	48✔
231	}
232
233	// serializeResult serializes a payment result and returns a key and value byte
234	// slice to insert into the bucket.
235	func serializeResult(rp *paymentResult) ([]byte, []byte, error) {	15✔
236	recordProducers := []tlv.RecordProducer{	15✔
237	&rp.timeFwd,	15✔
238	&rp.timeReply,	15✔
239	&rp.route,	15✔
240	}	15✔
241		15✔
242	rp.failure.WhenSome(	15✔
243	func(failure tlv.RecordT[tlv.TlvType3, paymentFailure]) {	28✔
244	recordProducers = append(recordProducers, &failure)	13✔
245	},	13✔
246	)
247
248	// Compose key that identifies this result.
249	key := getResultKey(rp)	15✔
250		15✔
251	var buff bytes.Buffer	15✔
252	err := lnwire.EncodeRecordsTo(	15✔
253	&buff, lnwire.ProduceRecordsSorted(recordProducers...),	15✔
254	)	15✔
255	if err != nil {	15✔
256	return nil, nil, err	×
257	}	×
258
259	return key, buff.Bytes(), nil	15✔
260	}
261
262	// deserializeResult deserializes a payment result.
263	func deserializeResult(k, v []byte) (*paymentResult, error) {	26✔
264	// Parse payment id.	26✔
265	result := paymentResult{	26✔
266	id: byteOrder.Uint64(k[8:]),	26✔
267	}	26✔
268		26✔
269	failure := tlv.ZeroRecordT[tlv.TlvType3, paymentFailure]()	26✔
270	recordProducers := []tlv.RecordProducer{	26✔
271	&result.timeFwd,	26✔
272	&result.timeReply,	26✔
273	&result.route,	26✔
274	&failure,	26✔
275	}	26✔
276		26✔
277	r := bytes.NewReader(v)	26✔
278	typeMap, err := lnwire.DecodeRecords(	26✔
279	r, lnwire.ProduceRecordsSorted(recordProducers...)...,	26✔
280	)	26✔
281	if err != nil {	27✔
282	return nil, err	1✔
283	}	1✔
284
285	if _, ok := typeMap[result.failure.TlvType()]; ok {	48✔
286	result.failure = tlv.SomeRecordT(failure)	23✔
287	}	23✔
288
289	return &result, nil	25✔
290	}
291
292	// serializeRoute serializes a mcRoute and writes the resulting bytes to the
293	// given io.Writer.
294	func serializeRoute(w io.Writer, r *mcRoute) error {	30✔
295	records := lnwire.ProduceRecordsSorted(	30✔
296	&r.sourcePubKey,	30✔
297	&r.totalAmount,	30✔
298	&r.hops,	30✔
299	)	30✔
300		30✔
301	return lnwire.EncodeRecordsTo(w, records)	30✔
302	}	30✔
303
304	// deserializeRoute deserializes the mcRoute from the given io.Reader.
305	func deserializeRoute(r io.Reader) (*mcRoute, error) {	25✔
306	var rt mcRoute	25✔
307	records := lnwire.ProduceRecordsSorted(	25✔
308	&rt.sourcePubKey,	25✔
309	&rt.totalAmount,	25✔
310	&rt.hops,	25✔
311	)	25✔
312		25✔
313	_, err := lnwire.DecodeRecords(r, records...)	25✔
314	if err != nil {	25✔
315	return nil, err	×
316	}	×
317
318	return &rt, nil	25✔
319	}
320
321	// deserializeHop deserializes the mcHop from the given io.Reader.
322	func deserializeHop(r io.Reader) (*mcHop, error) {	29✔
323	var (	29✔
324	h mcHop	29✔
325	blinding = tlv.ZeroRecordT[tlv.TlvType3, lnwire.TrueBoolean]()	29✔
326	custom = tlv.ZeroRecordT[tlv.TlvType4, lnwire.TrueBoolean]()	29✔
327	)	29✔
328	records := lnwire.ProduceRecordsSorted(	29✔
329	&h.channelID,	29✔
330	&h.pubKeyBytes,	29✔
331	&h.amtToFwd,	29✔
332	&blinding,	29✔
333	&custom,	29✔
334	)	29✔
335		29✔
336	typeMap, err := lnwire.DecodeRecords(r, records...)	29✔
337	if err != nil {	29✔
338	return nil, err	×
339	}	×
340
341	if _, ok := typeMap[h.hasBlindingPoint.TlvType()]; ok {	29✔
342	h.hasBlindingPoint = tlv.SomeRecordT(blinding)	×
343	}	×
344
345	if _, ok := typeMap[h.hasCustomRecords.TlvType()]; ok {	50✔
346	h.hasCustomRecords = tlv.SomeRecordT(custom)	21✔
347	}	21✔
348
349	return &h, nil	29✔
350	}
351
352	// serializeHop serializes a mcHop and writes the resulting bytes to the given
353	// io.Writer.
354	func serializeHop(w io.Writer, h *mcHop) error {	76✔
355	recordProducers := []tlv.RecordProducer{	76✔
356	&h.channelID,	76✔
357	&h.pubKeyBytes,	76✔
358	&h.amtToFwd,	76✔
359	}	76✔
360		76✔
361	h.hasBlindingPoint.WhenSome(func(	76✔
362	hasBlinding tlv.RecordT[tlv.TlvType3, lnwire.TrueBoolean]) {	76✔
363		×
364	recordProducers = append(recordProducers, &hasBlinding)	×
365	})	×
366
367	h.hasCustomRecords.WhenSome(func(	76✔
368	hasCustom tlv.RecordT[tlv.TlvType4, lnwire.TrueBoolean]) {	120✔
369		44✔
370	recordProducers = append(recordProducers, &hasCustom)	44✔
371	})	44✔
372
373	return lnwire.EncodeRecordsTo(	76✔
374	w, lnwire.ProduceRecordsSorted(recordProducers...),	76✔
375	)	76✔
376	}
377
378	// AddResult adds a new result to the db.
379	func (b missionControlStore) AddResult(rp paymentResult) {	80✔
380	b.queueCond.L.Lock()	80✔
381	b.queue.PushBack(rp)	80✔
382	b.queueCond.L.Unlock()	80✔
383		80✔
384	b.queueCond.Signal()	80✔
385	}	80✔
386
387	// stop stops the store ticker goroutine.
388	func (b *missionControlStore) stop() {	2✔
389	close(b.done)	2✔
390		2✔
391	b.queueCond.Signal()	2✔
392		2✔
393	b.wg.Wait()	2✔
394	}	2✔
395
396	// run runs the MC store ticker goroutine.
397	func (b *missionControlStore) run() {	2✔
398	b.wg.Add(1)	2✔
399		2✔
400	go func() {	4✔
401	defer b.wg.Done()	2✔
402		2✔
403	timer := time.NewTimer(b.flushInterval)	2✔
404		2✔
405	// Immediately stop the timer. It will be started once new	2✔
406	// items are added to the store. As the doc for time.Timer	2✔
407	// states, every call to Stop() done on a timer that is not	2✔
408	// known to have been fired needs to be checked and the timer's	2✔
409	// channel needs to be drained appropriately. This could happen	2✔
410	// if the flushInterval is very small (e.g. 1 nanosecond).	2✔
411	if !timer.Stop() {	2✔
412	select {	×
413	case <-timer.C:	×
414	case <-b.done:	×
415	log.Debugf("Stopping mission control store")	×
416	}
417	}
418
419	for {	7✔
420	// Wait for the queue to not be empty.	5✔
421	b.queueCond.L.Lock()	5✔
422	for b.queue.Front() == nil {	12✔
423	// To make sure we can properly stop, we must	7✔
424	// read the `done` channel first before	7✔
425	// attempting to call `Wait()`. This is due to	7✔
426	// the fact when `Signal` is called before the	7✔
427	// `Wait` call, the `Wait` call will block	7✔
428	// indefinitely.	7✔
429	//	7✔
430	// TODO(yy): replace this with channels.	7✔
431	select {	7✔
432	case <-b.done:	2✔
433	b.queueCond.L.Unlock()	2✔
434		2✔
435	return	2✔
436	default:	5✔
437	}
438
439	b.queueCond.Wait()	5✔
440	}
441	b.queueCond.L.Unlock()	3✔
442		3✔
443	// Restart the timer.	3✔
444	timer.Reset(b.flushInterval)	3✔
445		3✔
446	select {	3✔
447	case <-timer.C:	3✔
448	if err := b.storeResults(); err != nil {	3✔
449	log.Errorf("Failed to update mission "+	×
450	"control store: %v", err)	×
451	}	×
452
453	case <-b.done:	×
454	// Release the timer's resources.	×
455	if !timer.Stop() {	×
456	select {	×
457	case <-timer.C:	×
458	case <-b.done:	×
459	log.Debugf("Mission control " +	×
460	"store stopped")	×
461	}
462	}
463	return	×
464	}
465	}
466	}()
467	}
468
469	// storeResults stores all accumulated results.
470	func (b *missionControlStore) storeResults() error {	10✔
471	// We copy a reference to the queue and clear the original queue to be	10✔
472	// able to release the lock.	10✔
473	b.queueCond.L.Lock()	10✔
474	l := b.queue	10✔
475		10✔
476	if l.Len() == 0 {	10✔
477	b.queueCond.L.Unlock()	×
478		×
479	return nil	×
480	}	×
481	b.queue = list.New()	10✔
482	b.queueCond.L.Unlock()	10✔
483		10✔
484	var (	10✔
485	newKeys map[string]struct{}	10✔
486	delKeys []string	10✔
487	storeCount int	10✔
488	pruneCount int	10✔
489	)	10✔
490		10✔
491	// Create a deduped list of new entries.	10✔
492	newKeys = make(map[string]struct{}, l.Len())	10✔
493	for e := l.Front(); e != nil; e = e.Next() {	25✔
494	pr, ok := e.Value.(*paymentResult)	15✔
495	if !ok {	15✔
496	return fmt.Errorf("wrong type %T (not *paymentResult)",	×
497	e.Value)	×
498	}	×
499	key := string(getResultKey(pr))	15✔
500	if _, ok := b.keysMap[key]; ok {	15✔
501	l.Remove(e)	×
502	continue	×
503	}
504	if _, ok := newKeys[key]; ok {	16✔
505	l.Remove(e)	1✔
506	continue	1✔
507	}
508	newKeys[key] = struct{}{}	14✔
509	}
510
511	// Create a list of entries to delete.
512	toDelete := b.keys.Len() + len(newKeys) - b.maxRecords	10✔
513	if b.maxRecords > 0 && toDelete > 0 {	14✔
514	delKeys = make([]string, 0, toDelete)	4✔
515		4✔
516	// Delete as many as needed from old keys.	4✔
517	for e := b.keys.Front(); len(delKeys) < toDelete && e != nil; {	9✔
518	key, ok := e.Value.(string)	5✔
519	if !ok {	5✔
520	return fmt.Errorf("wrong type %T (not string)",	×
521	e.Value)	×
522	}	×
523	delKeys = append(delKeys, key)	5✔
524	e = e.Next()	5✔
525	}
526
527	// If more deletions are needed, simply do not add from the
528	// list of new keys.
529	for e := l.Front(); len(delKeys) < toDelete && e != nil; {	4✔
530	toDelete--	×
531	pr, ok := e.Value.(*paymentResult)	×
532	if !ok {	×
533	return fmt.Errorf("wrong type %T (not "+	×
534	"*paymentResult )", e.Value)	×
535	}	×
536	key := string(getResultKey(pr))	×
537	delete(newKeys, key)	×
538	l.Remove(e)	×
539	e = l.Front()	×
540	}
541	}
542
543	err := b.db.update(func(bucket kvdb.RwBucket) error {	20✔
544	for e := l.Front(); e != nil; e = e.Next() {	25✔
545	pr, ok := e.Value.(*paymentResult)	15✔
546	if !ok {	15✔
547	return fmt.Errorf("wrong type %T (not "+	×
548	"*paymentResult)", e.Value)	×
549	}	×
550
551	// Serialize result into key and value byte slices.
552	k, v, err := serializeResult(pr)	15✔
553	if err != nil {	15✔
554	return err	×
555	}	×
556
557	// Put into results bucket.
558	if err := bucket.Put(k, v); err != nil {	15✔
559	return err	×
560	}	×
561
562	storeCount++	15✔
563	}
564
565	// Prune oldest entries.
566	for _, key := range delKeys {	15✔
567	if err := bucket.Delete([]byte(key)); err != nil {	5✔
568	return err	×
569	}	×
570	pruneCount++	5✔
571	}
572
573	return nil	10✔
574	}, func() {	10✔
575	storeCount, pruneCount = 0, 0	10✔
576	})	10✔
577
578	if err != nil {	10✔
579	return err	×
580	}	×
581
582	log.Debugf("Stored mission control results: %d added, %d deleted",	10✔
583	storeCount, pruneCount)	10✔
584		10✔
585	// DB Update was successful, update the in-memory cache.	10✔
586	for _, key := range delKeys {	15✔
587	delete(b.keysMap, key)	5✔
588	b.keys.Remove(b.keys.Front())	5✔
589	}	5✔
590	for e := l.Front(); e != nil; e = e.Next() {	25✔
591	pr, ok := e.Value.(*paymentResult)	15✔
592	if !ok {	15✔
593	return fmt.Errorf("wrong type %T (not *paymentResult)",	×
594	e.Value)	×
595	}	×
596	key := string(getResultKey(pr))	15✔
597	b.keys.PushBack(key)	15✔
598	}
599
600	return nil	10✔
601	}
602
603	// getResultKey returns a byte slice representing a unique key for this payment
604	// result.
605	func getResultKey(rp *paymentResult) []byte {	45✔
606	var keyBytes [8 + 8 + 33]byte	45✔
607		45✔
608	// Identify records by a combination of time, payment id and sender pub	45✔
609	// key. This allows importing mission control data from an external	45✔
610	// source without key collisions and keeps the records sorted	45✔
611	// chronologically.	45✔
612	byteOrder.PutUint64(keyBytes[:], rp.timeReply.Val)	45✔
613	byteOrder.PutUint64(keyBytes[8:], rp.id)	45✔
614	copy(keyBytes[16:], rp.route.Val.sourcePubKey.Val[:])	45✔
615		45✔
616	return keyBytes[:]	45✔
617	}	45✔
618
619	// failureMessage wraps the lnwire.FailureMessage interface such that we can
620	// apply a Record method and use the failureMessage in a TLV encoded type.
621	type failureMessage struct {
622	lnwire.FailureMessage
623	}
624
625	// Record returns a TLV record that can be used to encode/decode a list of
626	// failureMessage to/from a TLV stream.
627	func (r *failureMessage) Record() tlv.Record {	49✔
628	recordSize := func() uint64 {	75✔
629	var (	26✔
630	b bytes.Buffer	26✔
631	buf [8]byte	26✔
632	)	26✔
633	if err := encodeFailureMessage(&b, r, &buf); err != nil {	26✔
634	panic(err)	×
635	}
636
637	return uint64(len(b.Bytes()))	26✔
638	}
639
640	return tlv.MakeDynamicRecord(	49✔
641	0, r, recordSize, encodeFailureMessage, decodeFailureMessage,	49✔
642	)	49✔
643	}
644
645	func encodeFailureMessage(w io.Writer, val interface{}, _ *[8]byte) error {	52✔
646	if v, ok := val.(*failureMessage); ok {	104✔
647	var b bytes.Buffer	52✔
648	err := lnwire.EncodeFailureMessage(&b, v.FailureMessage, 0)	52✔
649	if err != nil {	52✔
650	return err	×
651	}	×
652
653	_, err = w.Write(b.Bytes())	52✔
654		52✔
655	return err	52✔
656	}
657
658	return tlv.NewTypeForEncodingErr(val, "routing.failureMessage")	×
659	}
660
661	func decodeFailureMessage(r io.Reader, val interface{}, _ *[8]byte,
662	l uint64) error {	23✔
663		23✔
664	if v, ok := val.(*failureMessage); ok {	46✔
665	msg, err := lnwire.DecodeFailureMessage(r, 0)	23✔
666	if err != nil {	23✔
667	return err	×
668	}	×
669
670	*v = failureMessage{	23✔
671	FailureMessage: msg,	23✔
672	}	23✔
673		23✔
674	return nil	23✔
675	}
676
677	return tlv.NewTypeForDecodingErr(val, "routing.failureMessage", l, l)	×
678	}

lightningnetwork / lnd / 19924300449

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