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

lightningnetwork / lnd / 14358372723

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

Pull #9696

github

web-flow
Merge e2837e400 into 867d27d68
Pull Request #9696: Add `development_guidelines.md` for both human and machine

107055 of 188823 relevant lines covered (56.7%)

22721.56 hits per line

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

88.74
/graph/db/notifications.go
1
package graphdb
2

3
import (
4
        "fmt"
5
        "image/color"
6
        "net"
7
        "sync"
8
        "sync/atomic"
9

10
        "github.com/btcsuite/btcd/btcec/v2"
11
        "github.com/btcsuite/btcd/btcutil"
12
        "github.com/btcsuite/btcd/wire"
13
        "github.com/go-errors/errors"
14
        "github.com/lightningnetwork/lnd/graph/db/models"
15
        "github.com/lightningnetwork/lnd/lnutils"
16
        "github.com/lightningnetwork/lnd/lnwire"
17
)
18

19
// topologyManager holds all the fields required to manage the network topology
20
// subscriptions and notifications.
21
type topologyManager struct {
22
        // ntfnClientCounter is an atomic counter that's used to assign unique
23
        // notification client IDs to new clients.
24
        ntfnClientCounter atomic.Uint64
25

26
        // topologyUpdate is a channel that carries new topology updates
27
        // messages from outside the ChannelGraph to be processed by the
28
        // networkHandler.
29
        topologyUpdate chan any
30

31
        // topologyClients maps a client's unique notification ID to a
32
        // topologyClient client that contains its notification dispatch
33
        // channel.
34
        topologyClients *lnutils.SyncMap[uint64, *topologyClient]
35

36
        // ntfnClientUpdates is a channel that's used to send new updates to
37
        // topology notification clients to the ChannelGraph. Updates either
38
        // add a new notification client, or cancel notifications for an
39
        // existing client.
40
        ntfnClientUpdates chan *topologyClientUpdate
41
}
42

43
// newTopologyManager creates a new instance of the topologyManager.
44
func newTopologyManager() *topologyManager {
174✔
45
        return &topologyManager{
174✔
46
                topologyUpdate:    make(chan any),
174✔
47
                topologyClients:   &lnutils.SyncMap[uint64, *topologyClient]{},
174✔
48
                ntfnClientUpdates: make(chan *topologyClientUpdate),
174✔
49
        }
174✔
50
}
174✔
51

52
// TopologyClient represents an intent to receive notifications from the
53
// channel router regarding changes to the topology of the channel graph. The
54
// TopologyChanges channel will be sent upon with new updates to the channel
55
// graph in real-time as they're encountered.
56
type TopologyClient struct {
57
        // TopologyChanges is a receive only channel that new channel graph
58
        // updates will be sent over.
59
        //
60
        // TODO(roasbeef): chan for each update type instead?
61
        TopologyChanges <-chan *TopologyChange
62

63
        // Cancel is a function closure that should be executed when the client
64
        // wishes to cancel their notification intent. Doing so allows the
65
        // ChannelRouter to free up resources.
66
        Cancel func()
67
}
68

69
// topologyClientUpdate is a message sent to the channel router to either
70
// register a new topology client or re-register an existing client.
71
type topologyClientUpdate struct {
72
        // cancel indicates if the update to the client is cancelling an
73
        // existing client's notifications. If not then this update will be to
74
        // register a new set of notifications.
75
        cancel bool
76

77
        // clientID is the unique identifier for this client. Any further
78
        // updates (deleting or adding) to this notification client will be
79
        // dispatched according to the target clientID.
80
        clientID uint64
81

82
        // ntfnChan is a *send-only* channel in which notifications should be
83
        // sent over from router -> client.
84
        ntfnChan chan<- *TopologyChange
85
}
86

87
// SubscribeTopology returns a new topology client which can be used by the
88
// caller to receive notifications whenever a change in the channel graph
89
// topology occurs. Changes that will be sent at notifications include: new
90
// nodes appearing, node updating their attributes, new channels, channels
91
// closing, and updates in the routing policies of a channel's directed edges.
92
func (c *ChannelGraph) SubscribeTopology() (*TopologyClient, error) {
4✔
93
        // If the router is not yet started, return an error to avoid a
4✔
94
        // deadlock waiting for it to handle the subscription request.
4✔
95
        if !c.started.Load() {
4✔
96
                return nil, fmt.Errorf("router not started")
×
97
        }
×
98

99
        // We'll first atomically obtain the next ID for this client from the
100
        // incrementing client ID counter.
101
        clientID := c.ntfnClientCounter.Add(1)
4✔
102

4✔
103
        log.Debugf("New graph topology client subscription, client %v",
4✔
104
                clientID)
4✔
105

4✔
106
        ntfnChan := make(chan *TopologyChange, 10)
4✔
107

4✔
108
        select {
4✔
109
        case c.ntfnClientUpdates <- &topologyClientUpdate{
110
                cancel:   false,
111
                clientID: clientID,
112
                ntfnChan: ntfnChan,
113
        }:
4✔
114
        case <-c.quit:
×
115
                return nil, errors.New("ChannelRouter shutting down")
×
116
        }
117

118
        return &TopologyClient{
4✔
119
                TopologyChanges: ntfnChan,
4✔
120
                Cancel: func() {
5✔
121
                        select {
1✔
122
                        case c.ntfnClientUpdates <- &topologyClientUpdate{
123
                                cancel:   true,
124
                                clientID: clientID,
125
                        }:
1✔
126
                        case <-c.quit:
×
127
                                return
×
128
                        }
129
                },
130
        }, nil
131
}
132

133
// topologyClient is a data-structure use by the channel router to couple the
134
// client's notification channel along with a special "exit" channel that can
135
// be used to cancel all lingering goroutines blocked on a send to the
136
// notification channel.
137
type topologyClient struct {
138
        // ntfnChan is a send-only channel that's used to propagate
139
        // notification s from the channel router to an instance of a
140
        // topologyClient client.
141
        ntfnChan chan<- *TopologyChange
142

143
        // exit is a channel that is used internally by the channel router to
144
        // cancel any active un-consumed goroutine notifications.
145
        exit chan struct{}
146

147
        wg sync.WaitGroup
148
}
149

150
// notifyTopologyChange notifies all registered clients of a new change in
151
// graph topology in a non-blocking.
152
func (c *ChannelGraph) notifyTopologyChange(topologyDiff *TopologyChange) {
3,471✔
153
        // notifyClient is a helper closure that will send topology updates to
3,471✔
154
        // the given client.
3,471✔
155
        notifyClient := func(clientID uint64, client *topologyClient) bool {
3,477✔
156
                client.wg.Add(1)
6✔
157

6✔
158
                log.Tracef("Sending topology notification to client=%v, "+
6✔
159
                        "NodeUpdates=%v, ChannelEdgeUpdates=%v, "+
6✔
160
                        "ClosedChannels=%v", clientID,
6✔
161
                        len(topologyDiff.NodeUpdates),
6✔
162
                        len(topologyDiff.ChannelEdgeUpdates),
6✔
163
                        len(topologyDiff.ClosedChannels))
6✔
164

6✔
165
                go func(t *topologyClient) {
12✔
166
                        defer t.wg.Done()
6✔
167

6✔
168
                        select {
6✔
169

170
                        // In this case we'll try to send the notification
171
                        // directly to the upstream client consumer.
172
                        case t.ntfnChan <- topologyDiff:
6✔
173

174
                        // If the client cancels the notifications, then we'll
175
                        // exit early.
176
                        case <-t.exit:
×
177

178
                        // Similarly, if the ChannelRouter itself exists early,
179
                        // then we'll also exit ourselves.
180
                        case <-c.quit:
×
181
                        }
182
                }(client)
183

184
                // Always return true here so the following Range will iterate
185
                // all clients.
186
                return true
6✔
187
        }
188

189
        // Range over the set of active clients, and attempt to send the
190
        // topology updates.
191
        c.topologyClients.Range(notifyClient)
3,471✔
192
}
193

194
// handleTopologyUpdate is responsible for sending any topology changes
195
// notifications to registered clients.
196
//
197
// NOTE: must be run inside goroutine.
198
func (c *ChannelGraph) handleTopologyUpdate(update any) {
4,938✔
199
        defer c.wg.Done()
4,938✔
200

4,938✔
201
        topChange := &TopologyChange{}
4,938✔
202
        err := c.addToTopologyChange(topChange, update)
4,938✔
203
        if err != nil {
4,945✔
204
                log.Errorf("unable to update topology change notification: %v",
7✔
205
                        err)
7✔
206
                return
7✔
207
        }
7✔
208

209
        if topChange.isEmpty() {
6,410✔
210
                return
1,479✔
211
        }
1,479✔
212

213
        c.notifyTopologyChange(topChange)
3,452✔
214
}
215

216
// TopologyChange represents a new set of modifications to the channel graph.
217
// Topology changes will be dispatched in real-time as the ChannelGraph
218
// validates and process modifications to the authenticated channel graph.
219
type TopologyChange struct {
220
        // NodeUpdates is a slice of nodes which are either new to the channel
221
        // graph, or have had their attributes updated in an authenticated
222
        // manner.
223
        NodeUpdates []*NetworkNodeUpdate
224

225
        // ChanelEdgeUpdates is a slice of channel edges which are either newly
226
        // opened and authenticated, or have had their routing policies
227
        // updated.
228
        ChannelEdgeUpdates []*ChannelEdgeUpdate
229

230
        // ClosedChannels contains a slice of close channel summaries which
231
        // described which block a channel was closed at, and also carry
232
        // supplemental information such as the capacity of the former channel.
233
        ClosedChannels []*ClosedChanSummary
234
}
235

236
// isEmpty returns true if the TopologyChange is empty. A TopologyChange is
237
// considered empty, if it contains no *new* updates of any type.
238
func (t *TopologyChange) isEmpty() bool {
4,931✔
239
        return len(t.NodeUpdates) == 0 && len(t.ChannelEdgeUpdates) == 0 &&
4,931✔
240
                len(t.ClosedChannels) == 0
4,931✔
241
}
4,931✔
242

243
// ClosedChanSummary is a summary of a channel that was detected as being
244
// closed by monitoring the blockchain. Once a channel's funding point has been
245
// spent, the channel will automatically be marked as closed by the
246
// ChainNotifier.
247
//
248
// TODO(roasbeef): add nodes involved?
249
type ClosedChanSummary struct {
250
        // ChanID is the short-channel ID which uniquely identifies the
251
        // channel.
252
        ChanID uint64
253

254
        // Capacity was the total capacity of the channel before it was closed.
255
        Capacity btcutil.Amount
256

257
        // ClosedHeight is the height in the chain that the channel was closed
258
        // at.
259
        ClosedHeight uint32
260

261
        // ChanPoint is the funding point, or the multi-sig utxo which
262
        // previously represented the channel.
263
        ChanPoint wire.OutPoint
264
}
265

266
// createCloseSummaries takes in a slice of channels closed at the target block
267
// height and creates a slice of summaries which of each channel closure.
268
func createCloseSummaries(blockHeight uint32,
269
        closedChans ...*models.ChannelEdgeInfo) []*ClosedChanSummary {
19✔
270

19✔
271
        closeSummaries := make([]*ClosedChanSummary, len(closedChans))
19✔
272
        for i, closedChan := range closedChans {
42✔
273
                closeSummaries[i] = &ClosedChanSummary{
23✔
274
                        ChanID:       closedChan.ChannelID,
23✔
275
                        Capacity:     closedChan.Capacity,
23✔
276
                        ClosedHeight: blockHeight,
23✔
277
                        ChanPoint:    closedChan.ChannelPoint,
23✔
278
                }
23✔
279
        }
23✔
280

281
        return closeSummaries
19✔
282
}
283

284
// NetworkNodeUpdate is an update for a  node within the Lightning Network. A
285
// NetworkNodeUpdate is sent out either when a new node joins the network, or a
286
// node broadcasts a new update with a newer time stamp that supersedes its
287
// old update. All updates are properly authenticated.
288
type NetworkNodeUpdate struct {
289
        // Addresses is a slice of all the node's known addresses.
290
        Addresses []net.Addr
291

292
        // IdentityKey is the identity public key of the target node. This is
293
        // used to encrypt onion blobs as well as to authenticate any new
294
        // updates.
295
        IdentityKey *btcec.PublicKey
296

297
        // Alias is the alias or nick name of the node.
298
        Alias string
299

300
        // Color is the node's color in hex code format.
301
        Color string
302

303
        // Features holds the set of features the node supports.
304
        Features *lnwire.FeatureVector
305
}
306

307
// ChannelEdgeUpdate is an update for a new channel within the ChannelGraph.
308
// This update is sent out once a new authenticated channel edge is discovered
309
// within the network. These updates are directional, so if a channel is fully
310
// public, then there will be two updates sent out: one for each direction
311
// within the channel. Each update will carry that particular routing edge
312
// policy for the channel direction.
313
//
314
// An edge is a channel in the direction of AdvertisingNode -> ConnectingNode.
315
type ChannelEdgeUpdate struct {
316
        // ChanID is the unique short channel ID for the channel. This encodes
317
        // where in the blockchain the channel's funding transaction was
318
        // originally confirmed.
319
        ChanID uint64
320

321
        // ChanPoint is the outpoint which represents the multi-sig funding
322
        // output for the channel.
323
        ChanPoint wire.OutPoint
324

325
        // Capacity is the capacity of the newly created channel.
326
        Capacity btcutil.Amount
327

328
        // MinHTLC is the minimum HTLC amount that this channel will forward.
329
        MinHTLC lnwire.MilliSatoshi
330

331
        // MaxHTLC is the maximum HTLC amount that this channel will forward.
332
        MaxHTLC lnwire.MilliSatoshi
333

334
        // BaseFee is the base fee that will charged for all HTLC's forwarded
335
        // across the this channel direction.
336
        BaseFee lnwire.MilliSatoshi
337

338
        // FeeRate is the fee rate that will be shared for all HTLC's forwarded
339
        // across this channel direction.
340
        FeeRate lnwire.MilliSatoshi
341

342
        // TimeLockDelta is the time-lock expressed in blocks that will be
343
        // added to outgoing HTLC's from incoming HTLC's. This value is the
344
        // difference of the incoming and outgoing HTLC's time-locks routed
345
        // through this hop.
346
        TimeLockDelta uint16
347

348
        // AdvertisingNode is the node that's advertising this edge.
349
        AdvertisingNode *btcec.PublicKey
350

351
        // ConnectingNode is the node that the advertising node connects to.
352
        ConnectingNode *btcec.PublicKey
353

354
        // Disabled, if true, signals that the channel is unavailable to relay
355
        // payments.
356
        Disabled bool
357

358
        // ExtraOpaqueData is the set of data that was appended to this message
359
        // to fill out the full maximum transport message size. These fields can
360
        // be used to specify optional data such as custom TLV fields.
361
        ExtraOpaqueData lnwire.ExtraOpaqueData
362
}
363

364
// appendTopologyChange appends the passed update message to the passed
365
// TopologyChange, properly identifying which type of update the message
366
// constitutes. This function will also fetch any required auxiliary
367
// information required to create the topology change update from the graph
368
// database.
369
func (c *ChannelGraph) addToTopologyChange(update *TopologyChange,
370
        msg any) error {
4,938✔
371

4,938✔
372
        switch m := msg.(type) {
4,938✔
373

374
        // Any node announcement maps directly to a NetworkNodeUpdate struct.
375
        // No further data munging or db queries are required.
376
        case *models.LightningNode:
799✔
377
                pubKey, err := m.PubKey()
799✔
378
                if err != nil {
799✔
379
                        return err
×
380
                }
×
381

382
                nodeUpdate := &NetworkNodeUpdate{
799✔
383
                        Addresses:   m.Addresses,
799✔
384
                        IdentityKey: pubKey,
799✔
385
                        Alias:       m.Alias,
799✔
386
                        Color:       EncodeHexColor(m.Color),
799✔
387
                        Features:    m.Features.Clone(),
799✔
388
                }
799✔
389

799✔
390
                update.NodeUpdates = append(update.NodeUpdates, nodeUpdate)
799✔
391
                return nil
799✔
392

393
        // We ignore initial channel announcements as we'll only send out
394
        // updates once the individual edges themselves have been updated.
395
        case *models.ChannelEdgeInfo:
1,479✔
396
                return nil
1,479✔
397

398
        // Any new ChannelUpdateAnnouncements will generate a corresponding
399
        // ChannelEdgeUpdate notification.
400
        case *models.ChannelEdgePolicy:
2,660✔
401
                // We'll need to fetch the edge's information from the database
2,660✔
402
                // in order to get the information concerning which nodes are
2,660✔
403
                // being connected.
2,660✔
404
                edgeInfo, _, _, err := c.FetchChannelEdgesByID(m.ChannelID)
2,660✔
405
                if err != nil {
2,667✔
406
                        return errors.Errorf("unable fetch channel edge: %v",
7✔
407
                                err)
7✔
408
                }
7✔
409

410
                // If the flag is one, then the advertising node is actually
411
                // the second node.
412
                sourceNode := edgeInfo.NodeKey1
2,653✔
413
                connectingNode := edgeInfo.NodeKey2
2,653✔
414
                if m.ChannelFlags&lnwire.ChanUpdateDirection == 1 {
3,975✔
415
                        sourceNode = edgeInfo.NodeKey2
1,322✔
416
                        connectingNode = edgeInfo.NodeKey1
1,322✔
417
                }
1,322✔
418

419
                aNode, err := sourceNode()
2,653✔
420
                if err != nil {
2,653✔
421
                        return err
×
422
                }
×
423
                cNode, err := connectingNode()
2,653✔
424
                if err != nil {
2,653✔
425
                        return err
×
426
                }
×
427

428
                edgeUpdate := &ChannelEdgeUpdate{
2,653✔
429
                        ChanID:          m.ChannelID,
2,653✔
430
                        ChanPoint:       edgeInfo.ChannelPoint,
2,653✔
431
                        TimeLockDelta:   m.TimeLockDelta,
2,653✔
432
                        Capacity:        edgeInfo.Capacity,
2,653✔
433
                        MinHTLC:         m.MinHTLC,
2,653✔
434
                        MaxHTLC:         m.MaxHTLC,
2,653✔
435
                        BaseFee:         m.FeeBaseMSat,
2,653✔
436
                        FeeRate:         m.FeeProportionalMillionths,
2,653✔
437
                        AdvertisingNode: aNode,
2,653✔
438
                        ConnectingNode:  cNode,
2,653✔
439
                        Disabled:        m.ChannelFlags&lnwire.ChanUpdateDisabled != 0,
2,653✔
440
                        ExtraOpaqueData: m.ExtraOpaqueData,
2,653✔
441
                }
2,653✔
442

2,653✔
443
                // TODO(roasbeef): add bit to toggle
2,653✔
444
                update.ChannelEdgeUpdates = append(update.ChannelEdgeUpdates,
2,653✔
445
                        edgeUpdate)
2,653✔
446
                return nil
2,653✔
447

448
        default:
×
449
                return fmt.Errorf("unable to add to topology change, "+
×
450
                        "unknown message type %T", msg)
×
451
        }
452
}
453

454
// EncodeHexColor takes a color and returns it in hex code format.
455
func EncodeHexColor(color color.RGBA) string {
808✔
456
        return fmt.Sprintf("#%02x%02x%02x", color.R, color.G, color.B)
808✔
457
}
808✔
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