13035292482

Committed 29 Jan 2025 03:59PM UTC coverage: 49.3% (-9.5%) from 58.777%

Build # 13035292482

Build Type

Pull #9456

github

Committed by

mohamedawnallah

Commit Message

docs: update release-notes-0.19.0.md

In this commit, we warn users about the removal
of RPCs `SendToRoute`, `SendToRouteSync`, `SendPayment`,
and `SendPaymentSync` in the next release 0.20.

Pull Request Pull Request #9456: lnrpc+docs: deprecate warning `SendToRoute`, `SendToRouteSync`, `SendPayment`, and `SendPaymentSync` in Release 0.19

Run Details

100634 of 204126 relevant lines covered (49.3%)

1.54 hits per line

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

59.32

/discovery/message_store.go

package discovery

import (
        "bytes"
        "encoding/binary"
        "errors"
        "fmt"

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

var (
        // messageStoreBucket is a key used to create a top level bucket in the
        // gossiper database, used for storing messages that are to be sent to
        // peers. Upon restarts, these messages will be read and resent to their
        // respective peers.
        //
        // maps:
        //   pubKey (33 bytes) + msgShortChanID (8 bytes) + msgType (2 bytes) -> msg
        messageStoreBucket = []byte("message-store")

        // ErrUnsupportedMessage is an error returned when we attempt to add a
        // message to the store that is not supported.
        ErrUnsupportedMessage = errors.New("unsupported message type")

        // ErrCorruptedMessageStore indicates that the on-disk bucketing
        // structure has altered since the gossip message store instance was
        // initialized.
        ErrCorruptedMessageStore = errors.New("gossip message store has been " +
                "corrupted")
)

// GossipMessageStore is a store responsible for storing gossip messages which
// we should reliably send to our peers.
type GossipMessageStore interface {
        // AddMessage adds a message to the store for this peer.
        AddMessage(lnwire.Message, [33]byte) error

        // DeleteMessage deletes a message from the store for this peer.
        DeleteMessage(lnwire.Message, [33]byte) error

        // Messages returns the total set of messages that exist within the
        // store for all peers.
        Messages() (map[[33]byte][]lnwire.Message, error)

        // Peers returns the public key of all peers with messages within the
        // store.
        Peers() (map[[33]byte]struct{}, error)

        // MessagesForPeer returns the set of messages that exists within the
        // store for the given peer.
        MessagesForPeer([33]byte) ([]lnwire.Message, error)
}

// MessageStore is an implementation of the GossipMessageStore interface backed
// by a channeldb instance. By design, this store will only keep the latest
// version of a message (like in the case of multiple ChannelUpdate's) for a
// channel with a peer.
type MessageStore struct {
        db kvdb.Backend
}

// A compile-time assertion to ensure messageStore implements the
// GossipMessageStore interface.
var _ GossipMessageStore = (*MessageStore)(nil)

// NewMessageStore creates a new message store backed by a channeldb instance.
func NewMessageStore(db kvdb.Backend) (*MessageStore, error) {
        err := kvdb.Batch(db, func(tx kvdb.RwTx) error {
                _, err := tx.CreateTopLevelBucket(messageStoreBucket)
                return err
        })
        if err != nil {
                return nil, fmt.Errorf("unable to create required buckets: %w",
                        err)
        }

        return &MessageStore{db}, nil
}

// msgShortChanID retrieves the short channel ID of the message.
func msgShortChanID(msg lnwire.Message) (lnwire.ShortChannelID, error) {
        var shortChanID lnwire.ShortChannelID
        switch msg := msg.(type) {
        case *lnwire.AnnounceSignatures1:
                shortChanID = msg.ShortChannelID
        case *lnwire.ChannelUpdate1:
                shortChanID = msg.ShortChannelID
        default:
                return shortChanID, ErrUnsupportedMessage
        }

        return shortChanID, nil
}

// messageStoreKey constructs the database key for the message to be stored.
func messageStoreKey(msg lnwire.Message, peerPubKey [33]byte) ([]byte, error) {
        shortChanID, err := msgShortChanID(msg)
        if err != nil {
                return nil, err
        }

        var k [33 + 8 + 2]byte
        copy(k[:33], peerPubKey[:])
        binary.BigEndian.PutUint64(k[33:41], shortChanID.ToUint64())
        binary.BigEndian.PutUint16(k[41:43], uint16(msg.MsgType()))

        return k[:], nil
}

// AddMessage adds a message to the store for this peer.
func (s *MessageStore) AddMessage(msg lnwire.Message, peerPubKey [33]byte) error {
        log.Tracef("Adding message of type %v to store for peer %x",
                msg.MsgType(), peerPubKey)

        // Construct the key for which we'll find this message with in the
        // store.
        msgKey, err := messageStoreKey(msg, peerPubKey)
        if err != nil {
                return err
        }

        // Serialize the message with its wire encoding.
        var b bytes.Buffer
        if _, err := lnwire.WriteMessage(&b, msg, 0); err != nil {
                return err
        }

        return kvdb.Batch(s.db, func(tx kvdb.RwTx) error {
                messageStore := tx.ReadWriteBucket(messageStoreBucket)
                if messageStore == nil {
                        return ErrCorruptedMessageStore
                }

                return messageStore.Put(msgKey, b.Bytes())
        })
}

// DeleteMessage deletes a message from the store for this peer.
func (s *MessageStore) DeleteMessage(msg lnwire.Message,
        peerPubKey [33]byte) error {

        log.Tracef("Deleting message of type %v from store for peer %x",
                msg.MsgType(), peerPubKey)

        // Construct the key for which we'll find this message with in the
        // store.
        msgKey, err := messageStoreKey(msg, peerPubKey)
        if err != nil {
                return err
        }

        return kvdb.Batch(s.db, func(tx kvdb.RwTx) error {
                messageStore := tx.ReadWriteBucket(messageStoreBucket)
                if messageStore == nil {
                        return ErrCorruptedMessageStore
                }

                // In the event that we're attempting to delete a ChannelUpdate
                // from the store, we'll make sure that we're actually deleting
                // the correct one as it can be overwritten.
                if msg, ok := msg.(*lnwire.ChannelUpdate1); ok {
                        // Deleting a value from a bucket that doesn't exist
                        // acts as a NOP, so we'll return if a message doesn't
                        // exist under this key.
                        v := messageStore.Get(msgKey)
                        if v == nil {
                                return nil
                        }

                        dbMsg, err := lnwire.ReadMessage(bytes.NewReader(v), 0)
                        if err != nil {
                                return err
                        }

                        // If the timestamps don't match, then the update stored
                        // should be the latest one, so we'll avoid deleting it.
                        m, ok := dbMsg.(*lnwire.ChannelUpdate1)
                        if !ok {
                                return fmt.Errorf("expected "+
                                        "*lnwire.ChannelUpdate1, got: %T",
                                        dbMsg)
                        }
                        if msg.Timestamp != m.Timestamp {
                                return nil
                        }
                }

                return messageStore.Delete(msgKey)
        })
}

// readMessage reads a message from its serialized form and ensures its
// supported by the current version of the message store.
func readMessage(msgBytes []byte) (lnwire.Message, error) {
        msg, err := lnwire.ReadMessage(bytes.NewReader(msgBytes), 0)
        if err != nil {
                return nil, err
        }

        // Check if the message is supported by the store. We can reuse the
        // check for ShortChannelID as its a dependency on messages stored.
        if _, err := msgShortChanID(msg); err != nil {
                return nil, err
        }

        return msg, nil
}

// Messages returns the total set of messages that exist within the store for
// all peers.
func (s *MessageStore) Messages() (map[[33]byte][]lnwire.Message, error) {
        var msgs map[[33]byte][]lnwire.Message
        err := kvdb.View(s.db, func(tx kvdb.RTx) error {
                messageStore := tx.ReadBucket(messageStoreBucket)
                if messageStore == nil {
                        return ErrCorruptedMessageStore
                }

                return messageStore.ForEach(func(k, v []byte) error {
                        var pubKey [33]byte
                        copy(pubKey[:], k[:33])

                        // Deserialize the message from its raw bytes and filter
                        // out any which are not currently supported by the
                        // store.
                        msg, err := readMessage(v)
                        if err == ErrUnsupportedMessage {
                                return nil
                        }
                        if err != nil {
                                return err
                        }

                        msgs[pubKey] = append(msgs[pubKey], msg)
                        return nil
                })
        }, func() {
                msgs = make(map[[33]byte][]lnwire.Message)
        })
        if err != nil {
                return nil, err
        }

        return msgs, nil
}

// MessagesForPeer returns the set of messages that exists within the store for
// the given peer.
func (s *MessageStore) MessagesForPeer(
        peerPubKey [33]byte) ([]lnwire.Message, error) {

        var msgs []lnwire.Message
        err := kvdb.View(s.db, func(tx kvdb.RTx) error {
                messageStore := tx.ReadBucket(messageStoreBucket)
                if messageStore == nil {
                        return ErrCorruptedMessageStore
                }

                c := messageStore.ReadCursor()
                k, v := c.Seek(peerPubKey[:])
                for ; bytes.HasPrefix(k, peerPubKey[:]); k, v = c.Next() {
                        // Deserialize the message from its raw bytes and filter
                        // out any which are not currently supported by the
                        // store.
                        msg, err := readMessage(v)
                        if err == ErrUnsupportedMessage {
                                continue
                        }
                        if err != nil {
                                return err
                        }

                        msgs = append(msgs, msg)
                }

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

        return msgs, nil
}

// Peers returns the public key of all peers with messages within the store.
func (s *MessageStore) Peers() (map[[33]byte]struct{}, error) {
        var peers map[[33]byte]struct{}
        err := kvdb.View(s.db, func(tx kvdb.RTx) error {
                messageStore := tx.ReadBucket(messageStoreBucket)
                if messageStore == nil {
                        return ErrCorruptedMessageStore
                }

                return messageStore.ForEach(func(k, _ []byte) error {
                        var pubKey [33]byte
                        copy(pubKey[:], k[:33])
                        peers[pubKey] = struct{}{}
                        return nil
                })
        }, func() {
                peers = make(map[[33]byte]struct{})
        })
        if err != nil {
                return nil, err
        }

        return peers, nil
}

1	package discovery
2
3	import (
4	"bytes"
5	"encoding/binary"
6	"errors"
7	"fmt"
8
9	"github.com/lightningnetwork/lnd/kvdb"
10	"github.com/lightningnetwork/lnd/lnwire"
11	)
12
13	var (
14	// messageStoreBucket is a key used to create a top level bucket in the
15	// gossiper database, used for storing messages that are to be sent to
16	// peers. Upon restarts, these messages will be read and resent to their
17	// respective peers.
18	//
19	// maps:
20	// pubKey (33 bytes) + msgShortChanID (8 bytes) + msgType (2 bytes) -> msg
21	messageStoreBucket = []byte("message-store")
22
23	// ErrUnsupportedMessage is an error returned when we attempt to add a
24	// message to the store that is not supported.
25	ErrUnsupportedMessage = errors.New("unsupported message type")
26
27	// ErrCorruptedMessageStore indicates that the on-disk bucketing
28	// structure has altered since the gossip message store instance was
29	// initialized.
30	ErrCorruptedMessageStore = errors.New("gossip message store has been " +
31	"corrupted")
32	)
33
34	// GossipMessageStore is a store responsible for storing gossip messages which
35	// we should reliably send to our peers.
36	type GossipMessageStore interface {
37	// AddMessage adds a message to the store for this peer.
38	AddMessage(lnwire.Message, [33]byte) error
39
40	// DeleteMessage deletes a message from the store for this peer.
41	DeleteMessage(lnwire.Message, [33]byte) error
42
43	// Messages returns the total set of messages that exist within the
44	// store for all peers.
45	Messages() (map[[33]byte][]lnwire.Message, error)
46
47	// Peers returns the public key of all peers with messages within the
48	// store.
49	Peers() (map[[33]byte]struct{}, error)
50
51	// MessagesForPeer returns the set of messages that exists within the
52	// store for the given peer.
53	MessagesForPeer([33]byte) ([]lnwire.Message, error)
54	}
55
56	// MessageStore is an implementation of the GossipMessageStore interface backed
57	// by a channeldb instance. By design, this store will only keep the latest
58	// version of a message (like in the case of multiple ChannelUpdate's) for a
59	// channel with a peer.
60	type MessageStore struct {
61	db kvdb.Backend
62	}
63
64	// A compile-time assertion to ensure messageStore implements the
65	// GossipMessageStore interface.
66	var _ GossipMessageStore = (*MessageStore)(nil)
67
68	// NewMessageStore creates a new message store backed by a channeldb instance.
69	func NewMessageStore(db kvdb.Backend) (*MessageStore, error) {	3✔
70	err := kvdb.Batch(db, func(tx kvdb.RwTx) error {	6✔
71	_, err := tx.CreateTopLevelBucket(messageStoreBucket)	3✔
72	return err	3✔
73	})	3✔
74	if err != nil {	3✔
75	return nil, fmt.Errorf("unable to create required buckets: %w",	×
76	err)	×
77	}	×
78
79	return &MessageStore{db}, nil	3✔
80	}
81
82	// msgShortChanID retrieves the short channel ID of the message.
83	func msgShortChanID(msg lnwire.Message) (lnwire.ShortChannelID, error) {	3✔
84	var shortChanID lnwire.ShortChannelID	3✔
85	switch msg := msg.(type) {	3✔
86	case *lnwire.AnnounceSignatures1:	3✔
87	shortChanID = msg.ShortChannelID	3✔
88	case *lnwire.ChannelUpdate1:	3✔
89	shortChanID = msg.ShortChannelID	3✔
90	default:	×
91	return shortChanID, ErrUnsupportedMessage	×
92	}
93
94	return shortChanID, nil	3✔
95	}
96
97	// messageStoreKey constructs the database key for the message to be stored.
98	func messageStoreKey(msg lnwire.Message, peerPubKey [33]byte) ([]byte, error) {	3✔
99	shortChanID, err := msgShortChanID(msg)	3✔
100	if err != nil {	3✔
101	return nil, err	×
102	}	×
103
104	var k [33 + 8 + 2]byte	3✔
105	copy(k[:33], peerPubKey[:])	3✔
106	binary.BigEndian.PutUint64(k[33:41], shortChanID.ToUint64())	3✔
107	binary.BigEndian.PutUint16(k[41:43], uint16(msg.MsgType()))	3✔
108		3✔
109	return k[:], nil	3✔
110	}
111
112	// AddMessage adds a message to the store for this peer.
113	func (s *MessageStore) AddMessage(msg lnwire.Message, peerPubKey [33]byte) error {	3✔
114	log.Tracef("Adding message of type %v to store for peer %x",	3✔
115	msg.MsgType(), peerPubKey)	3✔
116		3✔
117	// Construct the key for which we'll find this message with in the	3✔
118	// store.	3✔
119	msgKey, err := messageStoreKey(msg, peerPubKey)	3✔
120	if err != nil {	3✔
121	return err	×
122	}	×
123
124	// Serialize the message with its wire encoding.
125	var b bytes.Buffer	3✔
126	if _, err := lnwire.WriteMessage(&b, msg, 0); err != nil {	3✔
127	return err	×
128	}	×
129
130	return kvdb.Batch(s.db, func(tx kvdb.RwTx) error {	6✔
131	messageStore := tx.ReadWriteBucket(messageStoreBucket)	3✔
132	if messageStore == nil {	3✔
133	return ErrCorruptedMessageStore	×
134	}	×
135
136	return messageStore.Put(msgKey, b.Bytes())	3✔
137	})
138	}
139
140	// DeleteMessage deletes a message from the store for this peer.
141	func (s *MessageStore) DeleteMessage(msg lnwire.Message,
142	peerPubKey [33]byte) error {	3✔
143		3✔
144	log.Tracef("Deleting message of type %v from store for peer %x",	3✔
145	msg.MsgType(), peerPubKey)	3✔
146		3✔
147	// Construct the key for which we'll find this message with in the	3✔
148	// store.	3✔
149	msgKey, err := messageStoreKey(msg, peerPubKey)	3✔
150	if err != nil {	3✔
151	return err	×
152	}	×
153
154	return kvdb.Batch(s.db, func(tx kvdb.RwTx) error {	6✔
155	messageStore := tx.ReadWriteBucket(messageStoreBucket)	3✔
156	if messageStore == nil {	3✔
157	return ErrCorruptedMessageStore	×
158	}	×
159
160	// In the event that we're attempting to delete a ChannelUpdate
161	// from the store, we'll make sure that we're actually deleting
162	// the correct one as it can be overwritten.
163	if msg, ok := msg.(*lnwire.ChannelUpdate1); ok {	6✔
164	// Deleting a value from a bucket that doesn't exist	3✔
165	// acts as a NOP, so we'll return if a message doesn't	3✔
166	// exist under this key.	3✔
167	v := messageStore.Get(msgKey)	3✔
168	if v == nil {	3✔
169	return nil	×
170	}	×
171
172	dbMsg, err := lnwire.ReadMessage(bytes.NewReader(v), 0)	3✔
173	if err != nil {	3✔
174	return err	×
175	}	×
176
177	// If the timestamps don't match, then the update stored
178	// should be the latest one, so we'll avoid deleting it.
179	m, ok := dbMsg.(*lnwire.ChannelUpdate1)	3✔
180	if !ok {	3✔
181	return fmt.Errorf("expected "+	×
182	"*lnwire.ChannelUpdate1, got: %T",	×
183	dbMsg)	×
184	}	×
185	if msg.Timestamp != m.Timestamp {	3✔
186	return nil	×
187	}	×
188	}
189
190	return messageStore.Delete(msgKey)	3✔
191	})
192	}
193
194	// readMessage reads a message from its serialized form and ensures its
195	// supported by the current version of the message store.
196	func readMessage(msgBytes []byte) (lnwire.Message, error) {	3✔
197	msg, err := lnwire.ReadMessage(bytes.NewReader(msgBytes), 0)	3✔
198	if err != nil {	3✔
199	return nil, err	×
200	}	×
201
202	// Check if the message is supported by the store. We can reuse the
203	// check for ShortChannelID as its a dependency on messages stored.
204	if _, err := msgShortChanID(msg); err != nil {	3✔
205	return nil, err	×
206	}	×
207
208	return msg, nil	3✔
209	}
210
211	// Messages returns the total set of messages that exist within the store for
212	// all peers.
213	func (s *MessageStore) Messages() (map[[33]byte][]lnwire.Message, error) {	×
214	var msgs map[[33]byte][]lnwire.Message	×
215	err := kvdb.View(s.db, func(tx kvdb.RTx) error {	×
216	messageStore := tx.ReadBucket(messageStoreBucket)	×
217	if messageStore == nil {	×
218	return ErrCorruptedMessageStore	×
219	}	×
220
221	return messageStore.ForEach(func(k, v []byte) error {	×
222	var pubKey [33]byte	×
223	copy(pubKey[:], k[:33])	×
224		×
225	// Deserialize the message from its raw bytes and filter	×
226	// out any which are not currently supported by the	×
227	// store.	×
228	msg, err := readMessage(v)	×
229	if err == ErrUnsupportedMessage {	×
230	return nil	×
231	}	×
232	if err != nil {	×
233	return err	×
234	}	×
235
236	msgs[pubKey] = append(msgs[pubKey], msg)	×
237	return nil	×
238	})
239	}, func() {	×
240	msgs = make(map[[33]byte][]lnwire.Message)	×
241	})	×
242	if err != nil {	×
243	return nil, err	×
244	}	×
245
246	return msgs, nil	×
247	}
248
249	// MessagesForPeer returns the set of messages that exists within the store for
250	// the given peer.
251	func (s *MessageStore) MessagesForPeer(
252	peerPubKey [33]byte) ([]lnwire.Message, error) {	3✔
253		3✔
254	var msgs []lnwire.Message	3✔
255	err := kvdb.View(s.db, func(tx kvdb.RTx) error {	6✔
256	messageStore := tx.ReadBucket(messageStoreBucket)	3✔
257	if messageStore == nil {	3✔
258	return ErrCorruptedMessageStore	×
259	}	×
260
261	c := messageStore.ReadCursor()	3✔
262	k, v := c.Seek(peerPubKey[:])	3✔
263	for ; bytes.HasPrefix(k, peerPubKey[:]); k, v = c.Next() {	6✔
264	// Deserialize the message from its raw bytes and filter	3✔
265	// out any which are not currently supported by the	3✔
266	// store.	3✔
267	msg, err := readMessage(v)	3✔
268	if err == ErrUnsupportedMessage {	3✔
269	continue	×
270	}
271	if err != nil {	3✔
272	return err	×
273	}	×
274
275	msgs = append(msgs, msg)	3✔
276	}
277
278	return nil	3✔
279	}, func() {	3✔
280	msgs = nil	3✔
281	})	3✔
282	if err != nil {	3✔
283	return nil, err	×
284	}	×
285
286	return msgs, nil	3✔
287	}
288
289	// Peers returns the public key of all peers with messages within the store.
290	func (s *MessageStore) Peers() (map[[33]byte]struct{}, error) {	3✔
291	var peers map[[33]byte]struct{}	3✔
292	err := kvdb.View(s.db, func(tx kvdb.RTx) error {	6✔
293	messageStore := tx.ReadBucket(messageStoreBucket)	3✔
294	if messageStore == nil {	3✔
295	return ErrCorruptedMessageStore	×
296	}	×
297
298	return messageStore.ForEach(func(k, _ []byte) error {	6✔
299	var pubKey [33]byte	3✔
300	copy(pubKey[:], k[:33])	3✔
301	peers[pubKey] = struct{}{}	3✔
302	return nil	3✔
303	})	3✔
304	}, func() {	3✔
305	peers = make(map[[33]byte]struct{})	3✔
306	})	3✔
307	if err != nil {	3✔
308	return nil, err	×
309	}	×
310
311	return peers, nil	3✔
312	}

lightningnetwork / lnd / 13035292482

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