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

lightningnetwork / lnd / 23185093231

17 Mar 2026 08:25AM UTC coverage: 62.303% (-0.03%) from 62.33%
23185093231

push

github

web-flow
Merge pull request #10582 from ellemouton/g175-db-8

[g175] graph/db: add versioned range queries and complete v2 graph query migration

210 of 471 new or added lines in 11 files covered. (44.59%)

87 existing lines in 22 files now uncovered.

140872 of 226108 relevant lines covered (62.3%)

19402.99 hits per line

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

77.99
/graph/builder.go
1
package graph
2

3
import (
4
        "context"
5
        "errors"
6
        "fmt"
7
        "sync"
8
        "sync/atomic"
9
        "time"
10

11
        "github.com/btcsuite/btcd/btcec/v2"
12
        "github.com/btcsuite/btcd/wire"
13
        "github.com/lightningnetwork/lnd/batch"
14
        "github.com/lightningnetwork/lnd/chainntnfs"
15
        graphdb "github.com/lightningnetwork/lnd/graph/db"
16
        "github.com/lightningnetwork/lnd/graph/db/models"
17
        "github.com/lightningnetwork/lnd/lnutils"
18
        "github.com/lightningnetwork/lnd/lnwallet"
19
        "github.com/lightningnetwork/lnd/lnwire"
20
        "github.com/lightningnetwork/lnd/multimutex"
21
        "github.com/lightningnetwork/lnd/netann"
22
        "github.com/lightningnetwork/lnd/routing/chainview"
23
        "github.com/lightningnetwork/lnd/routing/route"
24
        "github.com/lightningnetwork/lnd/ticker"
25
)
26

27
const (
28
        // DefaultChannelPruneExpiry is the default duration used to determine
29
        // if a channel should be pruned or not.
30
        DefaultChannelPruneExpiry = time.Hour * 24 * 14
31

32
        // avgBitcoinBlockTime is the approximate time between Bitcoin blocks,
33
        // used to convert a time-based channel prune expiry into a
34
        // block-height-based expiry for v2 gossip channels.
35
        avgBitcoinBlockTime = 10 * time.Minute
36

37
        // DefaultFirstTimePruneDelay is the time we'll wait after startup
38
        // before attempting to prune the graph for zombie channels. We don't
39
        // do it immediately after startup to allow lnd to start up without
40
        // getting blocked by this job.
41
        DefaultFirstTimePruneDelay = 30 * time.Second
42

43
        // defaultStatInterval governs how often the router will log non-empty
44
        // stats related to processing new channels, updates, or node
45
        // announcements.
46
        defaultStatInterval = time.Minute
47
)
48

49
var (
50
        // ErrGraphBuilderShuttingDown is returned if the graph builder is in
51
        // the process of shutting down.
52
        ErrGraphBuilderShuttingDown = fmt.Errorf("graph builder shutting down")
53
)
54

55
// Config holds the configuration required by the Builder.
56
type Config struct {
57
        // SelfNode is the public key of the node that this channel router
58
        // belongs to.
59
        SelfNode route.Vertex
60

61
        // Graph is the channel graph that the ChannelRouter will use to gather
62
        // metrics from and also to carry out path finding queries.
63
        Graph *graphdb.ChannelGraph
64

65
        // Chain is the router's source to the most up-to-date blockchain data.
66
        // All incoming advertised channels will be checked against the chain
67
        // to ensure that the channels advertised are still open.
68
        Chain lnwallet.BlockChainIO
69

70
        // ChainView is an instance of a FilteredChainView which is used to
71
        // watch the sub-set of the UTXO set (the set of active channels) that
72
        // we need in order to properly maintain the channel graph.
73
        ChainView chainview.FilteredChainView
74

75
        // Notifier is a reference to the ChainNotifier, used to grab
76
        // the latest blocks if the router is missing any.
77
        Notifier chainntnfs.ChainNotifier
78

79
        // ChannelPruneExpiry is the duration used to determine if a channel
80
        // should be pruned or not. If the delta between now and when the
81
        // channel was last updated is greater than ChannelPruneExpiry, then
82
        // the channel is marked as a zombie channel eligible for pruning.
83
        ChannelPruneExpiry time.Duration
84

85
        // GraphPruneInterval is used as an interval to determine how often we
86
        // should examine the channel graph to garbage collect zombie channels.
87
        GraphPruneInterval time.Duration
88

89
        // FirstTimePruneDelay is the time we'll wait after startup before
90
        // attempting to prune the graph for zombie channels. We don't do it
91
        // immediately after startup to allow lnd to start up without getting
92
        // blocked by this job.
93
        FirstTimePruneDelay time.Duration
94

95
        // AssumeChannelValid toggles whether the builder will prune channels
96
        // based on their spentness vs using the fact that they are considered
97
        // zombies.
98
        AssumeChannelValid bool
99

100
        // StrictZombiePruning determines if we attempt to prune zombie
101
        // channels according to a stricter criteria. If true, then we'll prune
102
        // a channel if only *one* of the edges is considered a zombie.
103
        // Otherwise, we'll only prune the channel when both edges have a very
104
        // dated last update.
105
        StrictZombiePruning bool
106

107
        // IsAlias returns whether a passed ShortChannelID is an alias. This is
108
        // only used for our local channels.
109
        IsAlias func(scid lnwire.ShortChannelID) bool
110
}
111

112
// Builder builds and maintains a view of the Lightning Network graph.
113
type Builder struct {
114
        started atomic.Bool
115
        stopped atomic.Bool
116

117
        bestHeight atomic.Uint32
118

119
        cfg     *Config
120
        v1Graph *graphdb.VersionedGraph
121

122
        // newBlocks is a channel in which new blocks connected to the end of
123
        // the main chain are sent over, and blocks updated after a call to
124
        // UpdateFilter.
125
        newBlocks <-chan *chainview.FilteredBlock
126

127
        // staleBlocks is a channel in which blocks disconnected from the end
128
        // of our currently known best chain are sent over.
129
        staleBlocks <-chan *chainview.FilteredBlock
130

131
        // channelEdgeMtx is a mutex we use to make sure we process only one
132
        // ChannelEdgePolicy at a time for a given channelID, to ensure
133
        // consistency between the various database accesses.
134
        channelEdgeMtx *multimutex.Mutex[uint64]
135

136
        // statTicker is a resumable ticker that logs the router's progress as
137
        // it discovers channels or receives updates.
138
        statTicker ticker.Ticker
139

140
        // stats tracks newly processed channels, updates, and node
141
        // announcements over a window of defaultStatInterval.
142
        stats *builderStats
143

144
        quit chan struct{}
145
        wg   sync.WaitGroup
146
}
147

148
// A compile time check to ensure Builder implements the
149
// ChannelGraphSource interface.
150
var _ ChannelGraphSource = (*Builder)(nil)
151

152
// NewBuilder constructs a new Builder.
153
func NewBuilder(cfg *Config) (*Builder, error) {
23✔
154
        return &Builder{
23✔
155
                cfg: cfg,
23✔
156
                // For now, we'll just use V1 graph reader.
23✔
157
                v1Graph: graphdb.NewVersionedGraph(
23✔
158
                        cfg.Graph, lnwire.GossipVersion1,
23✔
159
                ),
23✔
160
                channelEdgeMtx: multimutex.NewMutex[uint64](),
23✔
161
                statTicker:     ticker.New(defaultStatInterval),
23✔
162
                stats:          new(builderStats),
23✔
163
                quit:           make(chan struct{}),
23✔
164
        }, nil
23✔
165
}
23✔
166

167
// Start launches all the goroutines the Builder requires to carry out its
168
// duties. If the builder has already been started, then this method is a noop.
169
func (b *Builder) Start() error {
23✔
170
        if !b.started.CompareAndSwap(false, true) {
23✔
171
                return nil
×
172
        }
×
173

174
        log.Info("Builder starting")
23✔
175

23✔
176
        bestHash, bestHeight, err := b.cfg.Chain.GetBestBlock()
23✔
177
        if err != nil {
23✔
178
                return err
×
179
        }
×
180

181
        // If the graph has never been pruned, or hasn't fully been created yet,
182
        // then we don't treat this as an explicit error.
183
        if _, _, err := b.cfg.Graph.PruneTip(context.TODO()); err != nil {
44✔
184
                switch {
21✔
185
                case errors.Is(err, graphdb.ErrGraphNeverPruned):
21✔
186
                        fallthrough
21✔
187

188
                case errors.Is(err, graphdb.ErrGraphNotFound):
21✔
189
                        // If the graph has never been pruned, then we'll set
21✔
190
                        // the prune height to the current best height of the
21✔
191
                        // chain backend.
21✔
192
                        _, err = b.cfg.Graph.PruneGraph(
21✔
193
                                context.TODO(), nil, bestHash,
21✔
194
                                uint32(bestHeight),
21✔
195
                        )
21✔
196
                        if err != nil {
21✔
197
                                return err
×
198
                        }
×
199

200
                default:
×
201
                        return err
×
202
                }
203
        }
204

205
        // If AssumeChannelValid is present, then we won't rely on pruning
206
        // channels from the graph based on their spentness, but whether they
207
        // are considered zombies or not. We will start zombie pruning after a
208
        // small delay, to avoid slowing down startup of lnd.
209
        if b.cfg.AssumeChannelValid { //nolint:nestif
24✔
210
                time.AfterFunc(b.cfg.FirstTimePruneDelay, func() {
2✔
211
                        select {
1✔
212
                        case <-b.quit:
×
213
                                return
×
214
                        default:
1✔
215
                        }
216

217
                        log.Info("Initial zombie prune starting")
1✔
218
                        if err := b.pruneZombieChans(); err != nil {
1✔
219
                                log.Errorf("Unable to prune zombies: %v", err)
×
220
                        }
×
221
                })
222
        } else {
22✔
223
                // Otherwise, we'll use our filtered chain view to prune
22✔
224
                // channels as soon as they are detected as spent on-chain.
22✔
225
                if err := b.cfg.ChainView.Start(); err != nil {
22✔
226
                        return err
×
227
                }
×
228

229
                // Once the instance is active, we'll fetch the channel we'll
230
                // receive notifications over.
231
                b.newBlocks = b.cfg.ChainView.FilteredBlocks()
22✔
232
                b.staleBlocks = b.cfg.ChainView.DisconnectedBlocks()
22✔
233

22✔
234
                // Before we perform our manual block pruning, we'll construct
22✔
235
                // and apply a fresh chain filter to the active
22✔
236
                // FilteredChainView instance.  We do this before, as otherwise
22✔
237
                // we may miss on-chain events as the filter hasn't properly
22✔
238
                // been applied.
22✔
239
                channelView, err := b.v1Graph.ChannelView(context.TODO())
22✔
240
                if err != nil && !errors.Is(
22✔
241
                        err, graphdb.ErrGraphNoEdgesFound,
22✔
242
                ) {
22✔
243

×
244
                        return err
×
245
                }
×
246

247
                log.Infof("Filtering chain using %v channels active",
22✔
248
                        len(channelView))
22✔
249

22✔
250
                if len(channelView) != 0 {
32✔
251
                        err = b.cfg.ChainView.UpdateFilter(
10✔
252
                                channelView, uint32(bestHeight),
10✔
253
                        )
10✔
254
                        if err != nil {
10✔
255
                                return err
×
256
                        }
×
257
                }
258

259
                // The graph pruning might have taken a while and there could be
260
                // new blocks available.
261
                _, bestHeight, err = b.cfg.Chain.GetBestBlock()
22✔
262
                if err != nil {
22✔
263
                        return err
×
264
                }
×
265
                b.bestHeight.Store(uint32(bestHeight))
22✔
266

22✔
267
                // Before we begin normal operation of the router, we first need
22✔
268
                // to synchronize the channel graph to the latest state of the
22✔
269
                // UTXO set.
22✔
270
                if err := b.syncGraphWithChain(); err != nil {
22✔
271
                        return err
×
272
                }
×
273

274
                // Finally, before we proceed, we'll prune any unconnected nodes
275
                // from the graph in order to ensure we maintain a tight graph
276
                // of "useful" nodes.
277
                err = b.cfg.Graph.PruneGraphNodes(context.TODO())
22✔
278
                if err != nil &&
22✔
279
                        !errors.Is(err, graphdb.ErrGraphNodesNotFound) {
22✔
280

×
281
                        return err
×
282
                }
×
283
        }
284

285
        b.wg.Add(1)
23✔
286
        go b.networkHandler()
23✔
287

23✔
288
        log.Debug("Builder started")
23✔
289

23✔
290
        return nil
23✔
291
}
292

293
// Stop signals to the Builder that it should halt all routines. This method
294
// will *block* until all goroutines have excited. If the builder has already
295
// stopped then this method will return immediately.
296
func (b *Builder) Stop() error {
23✔
297
        if !b.stopped.CompareAndSwap(false, true) {
25✔
298
                return nil
2✔
299
        }
2✔
300

301
        log.Info("Builder shutting down...")
21✔
302

21✔
303
        // Our filtered chain view could've only been started if
21✔
304
        // AssumeChannelValid isn't present.
21✔
305
        if !b.cfg.AssumeChannelValid {
41✔
306
                if err := b.cfg.ChainView.Stop(); err != nil {
20✔
307
                        return err
×
308
                }
×
309
        }
310

311
        close(b.quit)
21✔
312
        b.wg.Wait()
21✔
313

21✔
314
        log.Debug("Builder shutdown complete")
21✔
315

21✔
316
        return nil
21✔
317
}
318

319
// syncGraphWithChain attempts to synchronize the current channel graph with
320
// the latest UTXO set state. This process involves pruning from the channel
321
// graph any channels which have been closed by spending their funding output
322
// since we've been down.
323
func (b *Builder) syncGraphWithChain() error {
22✔
324
        // First, we'll need to check to see if we're already in sync with the
22✔
325
        // latest state of the UTXO set.
22✔
326
        bestHash, bestHeight, err := b.cfg.Chain.GetBestBlock()
22✔
327
        if err != nil {
22✔
328
                return err
×
329
        }
×
330
        b.bestHeight.Store(uint32(bestHeight))
22✔
331

22✔
332
        pruneHash, pruneHeight, err := b.cfg.Graph.PruneTip(context.TODO())
22✔
333
        if err != nil {
22✔
334
                switch {
×
335
                // If the graph has never been pruned, or hasn't fully been
336
                // created yet, then we don't treat this as an explicit error.
337
                case errors.Is(err, graphdb.ErrGraphNeverPruned):
×
338
                case errors.Is(err, graphdb.ErrGraphNotFound):
×
339
                default:
×
340
                        return err
×
341
                }
342
        }
343

344
        log.Infof("Prune tip for Channel Graph: height=%v, hash=%v",
22✔
345
                pruneHeight, pruneHash)
22✔
346

22✔
347
        switch {
22✔
348
        // If the graph has never been pruned, then we can exit early as this
349
        // entails it's being created for the first time and hasn't seen any
350
        // block or created channels.
351
        case pruneHeight == 0 || pruneHash == nil:
3✔
352
                return nil
3✔
353

354
        // If the block hashes and heights match exactly, then we don't need to
355
        // prune the channel graph as we're already fully in sync.
356
        case bestHash.IsEqual(pruneHash) && uint32(bestHeight) == pruneHeight:
17✔
357
                return nil
17✔
358
        }
359

360
        // If the main chain blockhash at prune height is different from the
361
        // prune hash, this might indicate the database is on a stale branch.
362
        mainBlockHash, err := b.cfg.Chain.GetBlockHash(int64(pruneHeight))
5✔
363
        if err != nil {
5✔
364
                return err
×
365
        }
×
366

367
        // While we are on a stale branch of the chain, walk backwards to find
368
        // first common block.
369
        for !pruneHash.IsEqual(mainBlockHash) {
15✔
370
                log.Infof("channel graph is stale. Disconnecting block %v "+
10✔
371
                        "(hash=%v)", pruneHeight, pruneHash)
10✔
372
                // Prune the graph for every channel that was opened at height
10✔
373
                // >= pruneHeight.
10✔
374
                _, err := b.cfg.Graph.DisconnectBlockAtHeight(
10✔
375
                        context.TODO(), pruneHeight,
10✔
376
                )
10✔
377
                if err != nil {
10✔
378
                        return err
×
379
                }
×
380

381
                pruneHash, pruneHeight, err = b.cfg.Graph.PruneTip(
10✔
382
                        context.TODO(),
10✔
383
                )
10✔
384
                switch {
10✔
385
                // If at this point the graph has never been pruned, we can exit
386
                // as this entails we are back to the point where it hasn't seen
387
                // any block or created channels, alas there's nothing left to
388
                // prune.
389
                case errors.Is(err, graphdb.ErrGraphNeverPruned):
×
390
                        return nil
×
391

392
                case errors.Is(err, graphdb.ErrGraphNotFound):
×
393
                        return nil
×
394

395
                case err != nil:
×
396
                        return err
×
397

398
                default:
10✔
399
                }
400

401
                mainBlockHash, err = b.cfg.Chain.GetBlockHash(
10✔
402
                        int64(pruneHeight),
10✔
403
                )
10✔
404
                if err != nil {
10✔
405
                        return err
×
406
                }
×
407
        }
408

409
        log.Infof("Syncing channel graph from height=%v (hash=%v) to "+
5✔
410
                "height=%v (hash=%v)", pruneHeight, pruneHash, bestHeight,
5✔
411
                bestHash)
5✔
412

5✔
413
        // If we're not yet caught up, then we'll walk forward in the chain
5✔
414
        // pruning the channel graph with each new block that hasn't yet been
5✔
415
        // consumed by the channel graph.
5✔
416
        var spentOutputs []*wire.OutPoint
5✔
417
        for nextHeight := pruneHeight + 1; nextHeight <= uint32(bestHeight); nextHeight++ { //nolint:ll
30✔
418
                // Break out of the rescan early if a shutdown has been
25✔
419
                // requested, otherwise long rescans will block the daemon from
25✔
420
                // shutting down promptly.
25✔
421
                select {
25✔
422
                case <-b.quit:
×
423
                        return ErrGraphBuilderShuttingDown
×
424
                default:
25✔
425
                }
426

427
                // Using the next height, request a manual block pruning from
428
                // the chainview for the particular block hash.
429
                log.Infof("Filtering block for closed channels, at height: %v",
25✔
430
                        int64(nextHeight))
25✔
431
                nextHash, err := b.cfg.Chain.GetBlockHash(int64(nextHeight))
25✔
432
                if err != nil {
25✔
433
                        return err
×
434
                }
×
435
                log.Tracef("Running block filter on block with hash: %v",
25✔
436
                        nextHash)
25✔
437
                filterBlock, err := b.cfg.ChainView.FilterBlock(nextHash)
25✔
438
                if err != nil {
25✔
439
                        return err
×
440
                }
×
441

442
                // We're only interested in all prior outputs that have been
443
                // spent in the block, so collate all the referenced previous
444
                // outpoints within each tx and input.
445
                for _, tx := range filterBlock.Transactions {
29✔
446
                        for _, txIn := range tx.TxIn {
8✔
447
                                spentOutputs = append(spentOutputs,
4✔
448
                                        &txIn.PreviousOutPoint)
4✔
449
                        }
4✔
450
                }
451
        }
452

453
        // With the spent outputs gathered, attempt to prune the channel graph,
454
        // also passing in the best hash+height so the prune tip can be updated.
455
        closedChans, err := b.cfg.Graph.PruneGraph(
5✔
456
                context.TODO(), spentOutputs, bestHash, uint32(bestHeight),
5✔
457
        )
5✔
458
        if err != nil {
5✔
459
                return err
×
460
        }
×
461

462
        log.Infof("Graph pruning complete: %v channels were closed since "+
5✔
463
                "height %v", len(closedChans), pruneHeight)
5✔
464

5✔
465
        return nil
5✔
466
}
467

468
// isTimestampStale returns true if the given freshness timestamp is considered
469
// stale based on the gossip version. For v1, staleness is determined by
470
// wall-clock time since the unix timestamp. For v2, staleness is determined by
471
// how many blocks have elapsed since the block height timestamp.
472
func (b *Builder) isTimestampStale(v lnwire.GossipVersion,
473
        freshness lnwire.Timestamp) bool {
28✔
474

28✔
475
        chanExpiry := b.cfg.ChannelPruneExpiry
28✔
476

28✔
477
        switch v {
28✔
478
        case lnwire.GossipVersion1:
18✔
479
                ts, ok := freshness.(lnwire.UnixTimestamp)
18✔
480
                if !ok || ts.IsZero() {
27✔
481
                        return true
9✔
482
                }
9✔
483

484
                t := time.Unix(int64(ts), 0)
9✔
485

9✔
486
                return time.Since(t) >= chanExpiry
9✔
487

488
        default:
10✔
489
                h, ok := freshness.(lnwire.BlockHeightTimestamp)
10✔
490
                if !ok || uint32(h) == 0 {
10✔
NEW
491
                        return true
×
NEW
492
                }
×
493

494
                expiryBlocks := uint32(chanExpiry / avgBitcoinBlockTime)
10✔
495
                currentHeight := b.bestHeight.Load()
10✔
496
                height := uint32(h)
10✔
497

10✔
498
                if height > currentHeight {
11✔
499
                        return false
1✔
500
                }
1✔
501

502
                return currentHeight-height >= expiryBlocks
9✔
503
        }
504
}
505

506
// isPolicyZombie returns true if the given edge policy is considered stale
507
// based on version-specific freshness criteria.
508
func (b *Builder) isPolicyZombie(e *models.ChannelEdgePolicy) bool {
16✔
509
        var freshness lnwire.Timestamp
16✔
510
        if e.Version == lnwire.GossipVersion1 {
28✔
511
                freshness = lnwire.UnixTimestamp(e.LastUpdate.Unix())
12✔
512
        } else {
16✔
513
                freshness = lnwire.BlockHeightTimestamp(e.LastBlockHeight)
4✔
514
        }
4✔
515

516
        return b.isTimestampStale(e.Version, freshness)
16✔
517
}
518

519
// isZombieChannel takes two edge policy updates and determines if the
520
// corresponding channel should be considered a zombie. The first boolean is
521
// true if the policy update from node 1 is considered a zombie, the second
522
// boolean is that of node 2, and the final boolean is true if the channel
523
// is considered a zombie.
524
func (b *Builder) isZombieChannel(e1,
525
        e2 *models.ChannelEdgePolicy) (bool, bool, bool) {
6✔
526

6✔
527
        e1Zombie := e1 == nil || b.isPolicyZombie(e1)
6✔
528
        e2Zombie := e2 == nil || b.isPolicyZombie(e2)
6✔
529

6✔
530
        // If strict zombie pruning is enabled, a channel is a zombie if
6✔
531
        // either edge is stale.
6✔
532
        if b.cfg.StrictZombiePruning {
9✔
533
                return e1Zombie, e2Zombie, e1Zombie || e2Zombie
3✔
534
        }
3✔
535

536
        // Otherwise a channel is only a zombie if both edges are stale.
537
        return e1Zombie, e2Zombie, e1Zombie && e2Zombie
3✔
538
}
539

540
// IsZombieChannel returns true if the channel described by info should be
541
// considered a zombie. For v1 channels, freshness is a unix timestamp; for v2+
542
// channels it is a block height.
543
func (b *Builder) IsZombieChannel(info graphdb.ChannelUpdateInfo) bool {
6✔
544
        e1Zombie := b.isTimestampStale(info.Version, info.Node1Freshness)
6✔
545
        e2Zombie := b.isTimestampStale(info.Version, info.Node2Freshness)
6✔
546

6✔
547
        if b.cfg.StrictZombiePruning {
8✔
548
                return e1Zombie || e2Zombie
2✔
549
        }
2✔
550

551
        return e1Zombie && e2Zombie
4✔
552
}
553

554
// pruneZombieChans is a method that will be called periodically to prune out
555
// any "zombie" channels. We consider channels zombies if *both* edges haven't
556
// been updated since our zombie horizon. If AssumeChannelValid is present,
557
// we'll also consider channels zombies if *both* edges are disabled. This
558
// usually signals that a channel has been closed on-chain. We do this
559
// periodically to keep a healthy, lively routing table.
560
func (b *Builder) pruneZombieChans() error {
5✔
561
        chansToPrune := make(map[uint64]struct{})
5✔
562
        chanExpiry := b.cfg.ChannelPruneExpiry
5✔
563

5✔
564
        log.Infof("Examining channel graph for zombie channels")
5✔
565

5✔
566
        // A helper method to detect if the channel belongs to this node
5✔
567
        isSelfChannelEdge := func(info *models.ChannelEdgeInfo) bool {
16✔
568
                return info.NodeKey1Bytes == b.cfg.SelfNode ||
11✔
569
                        info.NodeKey2Bytes == b.cfg.SelfNode
11✔
570
        }
11✔
571

572
        // First, we'll collect all the channels which are eligible for garbage
573
        // collection due to being zombies.
574
        filterPruneChans := func(info *models.ChannelEdgeInfo,
5✔
575
                e1, e2 *models.ChannelEdgePolicy) error {
13✔
576

8✔
577
                // Exit early in case this channel is already marked to be
8✔
578
                // pruned
8✔
579
                _, markedToPrune := chansToPrune[info.ChannelID]
8✔
580
                if markedToPrune {
8✔
581
                        return nil
×
582
                }
×
583

584
                // We'll ensure that we don't attempt to prune our *own*
585
                // channels from the graph, as in any case this should be
586
                // re-advertised by the sub-system above us.
587
                if isSelfChannelEdge(info) {
10✔
588
                        return nil
2✔
589
                }
2✔
590

591
                e1Zombie, e2Zombie, isZombieChan := b.isZombieChannel(e1, e2)
6✔
592

6✔
593
                if e1Zombie {
10✔
594
                        log.Tracef("Node1 pubkey=%x of chan_id=%v is zombie",
4✔
595
                                info.NodeKey1Bytes, info.ChannelID)
4✔
596
                }
4✔
597

598
                if e2Zombie {
12✔
599
                        log.Tracef("Node2 pubkey=%x of chan_id=%v is zombie",
6✔
600
                                info.NodeKey2Bytes, info.ChannelID)
6✔
601
                }
6✔
602

603
                // If either edge hasn't been updated for a period of
604
                // chanExpiry, then we'll mark the channel itself as eligible
605
                // for graph pruning.
606
                if !isZombieChan {
7✔
607
                        return nil
1✔
608
                }
1✔
609

610
                log.Debugf("ChannelID(%v) is a zombie, collecting to prune",
5✔
611
                        info.ChannelID)
5✔
612

5✔
613
                // TODO(roasbeef): add ability to delete single directional edge
5✔
614
                chansToPrune[info.ChannelID] = struct{}{}
5✔
615

5✔
616
                return nil
5✔
617
        }
618

619
        // If AssumeChannelValid is present we'll look at the disabled bit for
620
        // both edges. If they're both disabled, then we can interpret this as
621
        // the channel being closed and can prune it from our graph.
622
        if b.cfg.AssumeChannelValid {
7✔
623
                disabledChanIDs, err := b.cfg.Graph.DisabledChannelIDs(
2✔
624
                        context.TODO(), lnwire.GossipVersion1,
2✔
625
                )
2✔
626
                if err != nil {
2✔
627
                        return fmt.Errorf("unable to get disabled channels "+
×
628
                                "ids chans: %v", err)
×
629
                }
×
630

631
                disabledEdges, err := b.v1Graph.FetchChanInfos(
2✔
632
                        context.TODO(), disabledChanIDs,
2✔
633
                )
2✔
634
                if err != nil {
2✔
635
                        return fmt.Errorf("unable to fetch disabled channels "+
×
636
                                "edges chans: %v", err)
×
637
                }
×
638

639
                // Ensuring we won't prune our own channel from the graph.
640
                for _, disabledEdge := range disabledEdges {
5✔
641
                        if !isSelfChannelEdge(disabledEdge.Info) {
4✔
642
                                chansToPrune[disabledEdge.Info.ChannelID] =
1✔
643
                                        struct{}{}
1✔
644
                        }
1✔
645
                }
646
        }
647

648
        startTime := time.Unix(0, 0)
5✔
649
        endTime := time.Now().Add(-1 * chanExpiry)
5✔
650
        oldEdgesIter := b.cfg.Graph.ChanUpdatesInHorizon(
5✔
651
                context.TODO(), startTime, endTime,
5✔
652
        )
5✔
653

5✔
654
        for u, err := range oldEdgesIter {
13✔
655
                if err != nil {
8✔
656
                        return fmt.Errorf("unable to fetch expired "+
×
657
                                "channel updates chans: %v", err)
×
658
                }
×
659

660
                err = filterPruneChans(u.Info, u.Policy1, u.Policy2)
8✔
661
                if err != nil {
8✔
662
                        return fmt.Errorf("error filtering channels to "+
×
663
                                "prune: %w", err)
×
664
                }
×
665
        }
666

667
        log.Infof("Pruning %v zombie channels", len(chansToPrune))
5✔
668
        if len(chansToPrune) == 0 {
7✔
669
                return nil
2✔
670
        }
2✔
671

672
        // With the set of zombie-like channels obtained, we'll do another pass
673
        // to delete them from the channel graph.
674
        toPrune := make([]uint64, 0, len(chansToPrune))
3✔
675
        for chanID := range chansToPrune {
9✔
676
                toPrune = append(toPrune, chanID)
6✔
677
                log.Tracef("Pruning zombie channel with ChannelID(%v)", chanID)
6✔
678
        }
6✔
679
        err := b.v1Graph.DeleteChannelEdges(
3✔
680
                context.TODO(), b.cfg.StrictZombiePruning, true, toPrune...,
3✔
681
        )
3✔
682
        if err != nil {
3✔
683
                return fmt.Errorf("unable to delete zombie channels: %w", err)
×
684
        }
×
685

686
        // With the channels pruned, we'll also attempt to prune any nodes that
687
        // were a part of them.
688
        err = b.cfg.Graph.PruneGraphNodes(context.TODO())
3✔
689
        if err != nil && !errors.Is(err, graphdb.ErrGraphNodesNotFound) {
3✔
690
                return fmt.Errorf("unable to prune graph nodes: %w", err)
×
691
        }
×
692

693
        return nil
3✔
694
}
695

696
// networkHandler is the primary goroutine for the Builder. The roles of
697
// this goroutine include answering queries related to the state of the
698
// network, pruning the graph on new block notification, applying network
699
// updates, and registering new topology clients.
700
//
701
// NOTE: This MUST be run as a goroutine.
702
func (b *Builder) networkHandler() {
23✔
703
        defer b.wg.Done()
23✔
704

23✔
705
        graphPruneTicker := time.NewTicker(b.cfg.GraphPruneInterval)
23✔
706
        defer graphPruneTicker.Stop()
23✔
707

23✔
708
        defer b.statTicker.Stop()
23✔
709

23✔
710
        b.stats.Reset()
23✔
711

23✔
712
        for {
122✔
713
                // If there are stats, resume the statTicker.
99✔
714
                if !b.stats.Empty() {
132✔
715
                        b.statTicker.Resume()
33✔
716
                }
33✔
717

718
                select {
99✔
719
                case chainUpdate, ok := <-b.staleBlocks:
13✔
720
                        // If the channel has been closed, then this indicates
13✔
721
                        // the daemon is shutting down, so we exit ourselves.
13✔
722
                        if !ok {
13✔
723
                                return
×
724
                        }
×
725

726
                        // Since this block is stale, we update our best height
727
                        // to the previous block.
728
                        blockHeight := chainUpdate.Height
13✔
729
                        b.bestHeight.Store(blockHeight - 1)
13✔
730

13✔
731
                        // Update the channel graph to reflect that this block
13✔
732
                        // was disconnected.
13✔
733
                        _, err := b.cfg.Graph.DisconnectBlockAtHeight(
13✔
734
                                context.TODO(), blockHeight,
13✔
735
                        )
13✔
736
                        if err != nil {
13✔
737
                                log.Errorf("unable to prune graph with stale "+
×
738
                                        "block: %v", err)
×
739
                                continue
×
740
                        }
741

742
                        // TODO(halseth): notify client about the reorg?
743

744
                // A new block has arrived, so we can prune the channel graph
745
                // of any channels which were closed in the block.
746
                case chainUpdate, ok := <-b.newBlocks:
69✔
747
                        // If the channel has been closed, then this indicates
69✔
748
                        // the daemon is shutting down, so we exit ourselves.
69✔
749
                        if !ok {
69✔
750
                                return
×
751
                        }
×
752

753
                        // We'll ensure that any new blocks received attach
754
                        // directly to the end of our main chain. If not, then
755
                        // we've somehow missed some blocks. Here we'll catch
756
                        // up the chain with the latest blocks.
757
                        currentHeight := b.bestHeight.Load()
69✔
758
                        switch {
69✔
759
                        case chainUpdate.Height == currentHeight+1:
63✔
760
                                err := b.updateGraphWithClosedChannels(
63✔
761
                                        chainUpdate,
63✔
762
                                )
63✔
763
                                if err != nil {
63✔
764
                                        log.Errorf("unable to prune graph "+
×
765
                                                "with closed channels: %v", err)
×
766
                                }
×
767

768
                        case chainUpdate.Height > currentHeight+1:
1✔
769
                                log.Errorf("out of order block: expecting "+
1✔
770
                                        "height=%v, got height=%v",
1✔
771
                                        currentHeight+1, chainUpdate.Height)
1✔
772

1✔
773
                                err := b.getMissingBlocks(
1✔
774
                                        currentHeight, chainUpdate,
1✔
775
                                )
1✔
776
                                if err != nil {
1✔
777
                                        log.Errorf("unable to retrieve missing"+
×
778
                                                "blocks: %v", err)
×
779
                                }
×
780

781
                        case chainUpdate.Height < currentHeight+1:
6✔
782
                                log.Errorf("out of order block: expecting "+
6✔
783
                                        "height=%v, got height=%v",
6✔
784
                                        currentHeight+1, chainUpdate.Height)
6✔
785

6✔
786
                                log.Infof("Skipping channel pruning since "+
6✔
787
                                        "received block height %v was already"+
6✔
788
                                        " processed.", chainUpdate.Height)
6✔
789
                        }
790

791
                // The graph prune ticker has ticked, so we'll examine the
792
                // state of the known graph to filter out any zombie channels
793
                // for pruning.
794
                case <-graphPruneTicker.C:
×
795
                        if err := b.pruneZombieChans(); err != nil {
×
796
                                log.Errorf("Unable to prune zombies: %v", err)
×
797
                        }
×
798

799
                // Log any stats if we've processed a non-empty number of
800
                // channels, updates, or nodes. We'll only pause the ticker if
801
                // the last window contained no updates to avoid resuming and
802
                // pausing while consecutive windows contain new info.
803
                case <-b.statTicker.Ticks():
3✔
804
                        if !b.stats.Empty() {
6✔
805
                                log.Infof(b.stats.String())
3✔
806
                        } else {
3✔
807
                                b.statTicker.Pause()
×
808
                        }
×
809
                        b.stats.Reset()
3✔
810

811
                // The router has been signalled to exit, to we exit our main
812
                // loop so the wait group can be decremented.
813
                case <-b.quit:
21✔
814
                        return
21✔
815
                }
816
        }
817
}
818

819
// getMissingBlocks walks through all missing blocks and updates the graph
820
// closed channels accordingly.
821
func (b *Builder) getMissingBlocks(currentHeight uint32,
822
        chainUpdate *chainview.FilteredBlock) error {
1✔
823

1✔
824
        outdatedHash, err := b.cfg.Chain.GetBlockHash(int64(currentHeight))
1✔
825
        if err != nil {
1✔
826
                return err
×
827
        }
×
828

829
        outdatedBlock := &chainntnfs.BlockEpoch{
1✔
830
                Height: int32(currentHeight),
1✔
831
                Hash:   outdatedHash,
1✔
832
        }
1✔
833

1✔
834
        epochClient, err := b.cfg.Notifier.RegisterBlockEpochNtfn(
1✔
835
                outdatedBlock,
1✔
836
        )
1✔
837
        if err != nil {
1✔
838
                return err
×
839
        }
×
840
        defer epochClient.Cancel()
1✔
841

1✔
842
        blockDifference := int(chainUpdate.Height - currentHeight)
1✔
843

1✔
844
        // We'll walk through all the outdated blocks and make sure we're able
1✔
845
        // to update the graph with any closed channels from them.
1✔
846
        for i := 0; i < blockDifference; i++ {
6✔
847
                var (
5✔
848
                        missingBlock *chainntnfs.BlockEpoch
5✔
849
                        ok           bool
5✔
850
                )
5✔
851

5✔
852
                select {
5✔
853
                case missingBlock, ok = <-epochClient.Epochs:
5✔
854
                        if !ok {
5✔
855
                                return nil
×
856
                        }
×
857

858
                case <-b.quit:
×
859
                        return nil
×
860
                }
861

862
                filteredBlock, err := b.cfg.ChainView.FilterBlock(
5✔
863
                        missingBlock.Hash,
5✔
864
                )
5✔
865
                if err != nil {
5✔
866
                        return err
×
867
                }
×
868

869
                err = b.updateGraphWithClosedChannels(
5✔
870
                        filteredBlock,
5✔
871
                )
5✔
872
                if err != nil {
5✔
873
                        return err
×
874
                }
×
875
        }
876

877
        return nil
1✔
878
}
879

880
// updateGraphWithClosedChannels prunes the channel graph of closed channels
881
// that are no longer needed.
882
func (b *Builder) updateGraphWithClosedChannels(
883
        chainUpdate *chainview.FilteredBlock) error {
68✔
884

68✔
885
        // Once a new block arrives, we update our running track of the height
68✔
886
        // of the chain tip.
68✔
887
        blockHeight := chainUpdate.Height
68✔
888

68✔
889
        b.bestHeight.Store(blockHeight)
68✔
890
        log.Infof("Pruning channel graph using block %v (height=%v)",
68✔
891
                chainUpdate.Hash, blockHeight)
68✔
892

68✔
893
        // We're only interested in all prior outputs that have been spent in
68✔
894
        // the block, so collate all the referenced previous outpoints within
68✔
895
        // each tx and input.
68✔
896
        var spentOutputs []*wire.OutPoint
68✔
897
        for _, tx := range chainUpdate.Transactions {
72✔
898
                for _, txIn := range tx.TxIn {
8✔
899
                        spentOutputs = append(spentOutputs,
4✔
900
                                &txIn.PreviousOutPoint)
4✔
901
                }
4✔
902
        }
903

904
        // With the spent outputs gathered, attempt to prune the channel graph,
905
        // also passing in the hash+height of the block being pruned so the
906
        // prune tip can be updated.
907
        chansClosed, err := b.cfg.Graph.PruneGraph(
68✔
908
                context.TODO(), spentOutputs, &chainUpdate.Hash,
68✔
909
                chainUpdate.Height,
68✔
910
        )
68✔
911
        if err != nil {
68✔
912
                log.Errorf("unable to prune routing table: %v", err)
×
913
                return err
×
914
        }
×
915

916
        log.Infof("Block %v (height=%v) closed %v channels", chainUpdate.Hash,
68✔
917
                blockHeight, len(chansClosed))
68✔
918

68✔
919
        return nil
68✔
920
}
921

922
// assertNodeAnnFreshness returns a non-nil error if we have an announcement in
923
// the database for the passed node with a timestamp newer than the passed
924
// timestamp. ErrIgnored will be returned if we already have the node, and
925
// ErrOutdated will be returned if we have a timestamp that's after the new
926
// timestamp.
927
func (b *Builder) assertNodeAnnFreshness(ctx context.Context, node route.Vertex,
928
        msgTimestamp time.Time) error {
13✔
929

13✔
930
        // If we are not already aware of this node, it means that we don't
13✔
931
        // know about any channel using this node. To avoid a DoS attack by
13✔
932
        // node announcements, we will ignore such nodes. If we do know about
13✔
933
        // this node, check that this update brings info newer than what we
13✔
934
        // already have.
13✔
935
        lastUpdate, exists, err := b.cfg.Graph.HasV1Node(ctx, node)
13✔
936
        if err != nil {
13✔
937
                return fmt.Errorf("unable to query for the "+
×
938
                        "existence of node: %w", err)
×
939
        }
×
940
        if !exists {
17✔
941
                return NewErrf(ErrIgnored, "Ignoring node announcement"+
4✔
942
                        " for node not found in channel graph (%x)",
4✔
943
                        node[:])
4✔
944
        }
4✔
945

946
        // If we've reached this point then we're aware of the vertex being
947
        // advertised. So we now check if the new message has a new time stamp,
948
        // if not then we won't accept the new data as it would override newer
949
        // data.
950
        if !lastUpdate.Before(msgTimestamp) {
16✔
951
                return NewErrf(ErrOutdated, "Ignoring outdated "+
4✔
952
                        "announcement for %x", node[:])
4✔
953
        }
4✔
954

955
        return nil
11✔
956
}
957

958
// MarkZombieEdge adds a channel that failed complete validation into the zombie
959
// index so we can avoid having to re-validate it in the future.
960
func (b *Builder) MarkZombieEdge(chanID uint64) error {
×
961
        // If the edge fails validation we'll mark the edge itself as a zombie
×
962
        // so we don't continue to request it. We use the "zero key" for both
×
963
        // node pubkeys so this edge can't be resurrected.
×
964
        var zeroKey [33]byte
×
965
        err := b.cfg.Graph.MarkEdgeZombie(
×
NEW
966
                context.TODO(), lnwire.GossipVersion1, chanID, zeroKey, zeroKey,
×
967
        )
×
968
        if err != nil {
×
969
                return fmt.Errorf("unable to mark spent chan(id=%v) as a "+
×
970
                        "zombie: %w", chanID, err)
×
971
        }
×
972

973
        return nil
×
974
}
975

976
// ApplyChannelUpdate validates a channel update and if valid, applies it to the
977
// database. It returns a bool indicating whether the updates were successful.
978
func (b *Builder) ApplyChannelUpdate(msg *lnwire.ChannelUpdate1) bool {
3✔
979
        ctx := context.TODO()
3✔
980

3✔
981
        ch, _, _, err := b.GetChannelByID(msg.ShortChannelID)
3✔
982
        if err != nil {
6✔
983
                log.Errorf("Unable to retrieve channel by id: %v", err)
3✔
984
                return false
3✔
985
        }
3✔
986

987
        var pubKey *btcec.PublicKey
3✔
988

3✔
989
        switch msg.ChannelFlags & lnwire.ChanUpdateDirection {
3✔
990
        case 0:
3✔
991
                pubKey, _ = ch.NodeKey1()
3✔
992

993
        case 1:
3✔
994
                pubKey, _ = ch.NodeKey2()
3✔
995
        }
996

997
        // Exit early if the pubkey cannot be decided.
998
        if pubKey == nil {
3✔
999
                log.Errorf("Unable to decide pubkey with ChannelFlags=%v",
×
1000
                        msg.ChannelFlags)
×
1001
                return false
×
1002
        }
×
1003

1004
        err = netann.ValidateChannelUpdateAnn(pubKey, ch.Capacity, msg)
3✔
1005
        if err != nil {
3✔
1006
                log.Errorf("Unable to validate channel update: %v", err)
×
1007
                return false
×
1008
        }
×
1009

1010
        update, err := models.ChanEdgePolicyFromWire(
3✔
1011
                msg.ShortChannelID.ToUint64(), msg,
3✔
1012
        )
3✔
1013
        if err != nil {
3✔
1014
                log.Errorf("Unable to parse channel update: %v", err)
×
1015
                return false
×
1016
        }
×
1017

1018
        err = b.UpdateEdge(ctx, update)
3✔
1019
        if err != nil && !IsError(err, ErrIgnored, ErrOutdated) {
3✔
1020
                log.Errorf("Unable to apply channel update: %v", err)
×
1021
                return false
×
1022
        }
×
1023

1024
        return true
3✔
1025
}
1026

1027
// AddNode is used to add information about a node to the router database. If
1028
// the node with this pubkey is not present in an existing channel, it will
1029
// be ignored.
1030
//
1031
// NOTE: This method is part of the ChannelGraphSource interface.
1032
func (b *Builder) AddNode(ctx context.Context, node *models.Node,
1033
        op ...batch.SchedulerOption) error {
10✔
1034

10✔
1035
        err := b.addNode(ctx, node, op...)
10✔
1036
        if err != nil {
14✔
1037
                logNetworkMsgProcessError(err)
4✔
1038

4✔
1039
                return err
4✔
1040
        }
4✔
1041

1042
        return nil
9✔
1043
}
1044

1045
// addNode does some basic checks on the given Node against what we
1046
// currently have persisted in the graph, and then adds it to the graph. If we
1047
// already know about the node, then we only update our DB if the new update
1048
// has a newer timestamp than the last one we received.
1049
func (b *Builder) addNode(ctx context.Context, node *models.Node,
1050
        op ...batch.SchedulerOption) error {
10✔
1051

10✔
1052
        // Before we add the node to the database, we'll check to see if the
10✔
1053
        // announcement is "fresh" or not. If it isn't, then we'll return an
10✔
1054
        // error.
10✔
1055
        err := b.assertNodeAnnFreshness(ctx, node.PubKeyBytes, node.LastUpdate)
10✔
1056
        if err != nil {
14✔
1057
                return err
4✔
1058
        }
4✔
1059

1060
        if err := b.cfg.Graph.AddNode(ctx, node, op...); err != nil {
9✔
1061
                return fmt.Errorf("unable to add node %x to the "+
×
1062
                        "graph: %w", node.PubKeyBytes, err)
×
1063
        }
×
1064

1065
        log.Tracef("Updated vertex data for node=%x", node.PubKeyBytes)
9✔
1066
        b.stats.incNumNodeUpdates()
9✔
1067

9✔
1068
        return nil
9✔
1069
}
1070

1071
// AddEdge is used to add edge/channel to the topology of the router, after all
1072
// information about channel will be gathered this edge/channel might be used
1073
// in construction of payment path.
1074
//
1075
// NOTE: This method is part of the ChannelGraphSource interface.
1076
func (b *Builder) AddEdge(ctx context.Context, edge *models.ChannelEdgeInfo,
1077
        op ...batch.SchedulerOption) error {
17✔
1078

17✔
1079
        err := b.addEdge(ctx, edge, op...)
17✔
1080
        if err != nil {
20✔
1081
                logNetworkMsgProcessError(err)
3✔
1082

3✔
1083
                return err
3✔
1084
        }
3✔
1085

1086
        return nil
17✔
1087
}
1088

1089
// addEdge does some validation on the new channel edge against what we
1090
// currently have persisted in the graph, and then adds it to the graph. The
1091
// Chain View is updated with the new edge if it is successfully added to the
1092
// graph. We only persist the channel if we currently dont have it at all in
1093
// our graph.
1094
func (b *Builder) addEdge(ctx context.Context, edge *models.ChannelEdgeInfo,
1095
        op ...batch.SchedulerOption) error {
17✔
1096

17✔
1097
        log.Debugf("Received ChannelEdgeInfo for channel %v", edge.ChannelID)
17✔
1098

17✔
1099
        // Prior to processing the announcement we first check if we
17✔
1100
        // already know of this channel, if so, then we can exit early.
17✔
1101
        exists, isZombie, err := b.cfg.Graph.HasChannelEdge(
17✔
1102
                ctx, edge.Version, edge.ChannelID,
17✔
1103
        )
17✔
1104
        if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) {
17✔
1105
                return fmt.Errorf("unable to check for edge existence: %w",
×
1106
                        err)
×
1107
        }
×
1108
        if isZombie {
17✔
1109
                return NewErrf(ErrIgnored, "ignoring msg for zombie chan_id=%v",
×
1110
                        edge.ChannelID)
×
1111
        }
×
1112
        if exists {
20✔
1113
                return NewErrf(ErrIgnored, "ignoring msg for known chan_id=%v",
3✔
1114
                        edge.ChannelID)
3✔
1115
        }
3✔
1116

1117
        if err := b.cfg.Graph.AddChannelEdge(ctx, edge, op...); err != nil {
17✔
1118
                return fmt.Errorf("unable to add edge: %w", err)
×
1119
        }
×
1120

1121
        b.stats.incNumEdgesDiscovered()
17✔
1122

17✔
1123
        // If AssumeChannelValid is present, of if the SCID is an alias, then
17✔
1124
        // the gossiper would not have done the expensive work of fetching
17✔
1125
        // a funding transaction and validating it. So we won't have the channel
17✔
1126
        // capacity nor the funding script. So we just log and return here.
17✔
1127
        scid := lnwire.NewShortChanIDFromInt(edge.ChannelID)
17✔
1128
        if b.cfg.AssumeChannelValid || b.cfg.IsAlias(scid) {
20✔
1129
                log.Tracef("New channel discovered! Link connects %x and %x "+
3✔
1130
                        "with ChannelID(%v)", edge.NodeKey1Bytes,
3✔
1131
                        edge.NodeKey2Bytes, edge.ChannelID)
3✔
1132

3✔
1133
                return nil
3✔
1134
        }
3✔
1135

1136
        log.Debugf("New channel discovered! Link connects %x and %x with "+
17✔
1137
                "ChannelPoint(%v): chan_id=%v, capacity=%v", edge.NodeKey1Bytes,
17✔
1138
                edge.NodeKey2Bytes, edge.ChannelPoint, edge.ChannelID,
17✔
1139
                edge.Capacity)
17✔
1140

17✔
1141
        // Otherwise, then we expect the funding script to be present on the
17✔
1142
        // edge since it would have been fetched when the gossiper validated the
17✔
1143
        // announcement.
17✔
1144
        fundingPkScript, err := edge.FundingScript.UnwrapOrErr(fmt.Errorf(
17✔
1145
                "expected the funding transaction script to be set",
17✔
1146
        ))
17✔
1147
        if err != nil {
17✔
1148
                return err
×
1149
        }
×
1150

1151
        // As a new edge has been added to the channel graph, we'll update the
1152
        // current UTXO filter within our active FilteredChainView so we are
1153
        // notified if/when this channel is closed.
1154
        filterUpdate := []graphdb.EdgePoint{
17✔
1155
                {
17✔
1156
                        FundingPkScript: fundingPkScript,
17✔
1157
                        OutPoint:        edge.ChannelPoint,
17✔
1158
                },
17✔
1159
        }
17✔
1160

17✔
1161
        err = b.cfg.ChainView.UpdateFilter(filterUpdate, b.bestHeight.Load())
17✔
1162
        if err != nil {
17✔
1163
                return fmt.Errorf("unable to update chain view: %w", err)
×
1164
        }
×
1165

1166
        return nil
17✔
1167
}
1168

1169
// UpdateEdge is used to update edge information, without this message edge
1170
// considered as not fully constructed.
1171
//
1172
// NOTE: This method is part of the ChannelGraphSource interface.
1173
func (b *Builder) UpdateEdge(ctx context.Context,
1174
        update *models.ChannelEdgePolicy, op ...batch.SchedulerOption) error {
9✔
1175

9✔
1176
        err := b.updateEdge(ctx, update, op...)
9✔
1177
        if err != nil {
13✔
1178
                logNetworkMsgProcessError(err)
4✔
1179

4✔
1180
                return err
4✔
1181
        }
4✔
1182

1183
        return nil
8✔
1184
}
1185

1186
// updateEdge validates the new edge policy against what we currently have
1187
// persisted in the graph, and then applies it to the graph if the update is
1188
// considered fresh enough and if we actually have a channel persisted for the
1189
// given update.
1190
func (b *Builder) updateEdge(ctx context.Context,
1191
        policy *models.ChannelEdgePolicy, op ...batch.SchedulerOption) error {
9✔
1192

9✔
1193
        log.Debugf("Received ChannelEdgePolicy for channel %v",
9✔
1194
                policy.ChannelID)
9✔
1195

9✔
1196
        // We make sure to hold the mutex for this channel ID, such that no
9✔
1197
        // other goroutine is concurrently doing database accesses for the same
9✔
1198
        // channel ID.
9✔
1199
        b.channelEdgeMtx.Lock(policy.ChannelID)
9✔
1200
        defer b.channelEdgeMtx.Unlock(policy.ChannelID)
9✔
1201

9✔
1202
        edge1Timestamp, edge2Timestamp, exists, isZombie, err :=
9✔
1203
                b.cfg.Graph.HasV1ChannelEdge(ctx, policy.ChannelID)
9✔
1204
        if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) {
9✔
1205
                return fmt.Errorf("unable to check for edge existence: %w", err)
×
1206
        }
×
1207

1208
        // If the channel is marked as a zombie in our database, and
1209
        // we consider this a stale update, then we should not apply the
1210
        // policy.
1211
        isStaleUpdate := time.Since(policy.LastUpdate) >
9✔
1212
                b.cfg.ChannelPruneExpiry
9✔
1213

9✔
1214
        if isZombie && isStaleUpdate {
9✔
1215
                return NewErrf(ErrIgnored, "ignoring stale update "+
×
1216
                        "(flags=%v|%v) for zombie chan_id=%v",
×
1217
                        policy.MessageFlags, policy.ChannelFlags,
×
1218
                        policy.ChannelID)
×
1219
        }
×
1220

1221
        // If the channel doesn't exist in our database, we cannot apply the
1222
        // updated policy.
1223
        if !exists {
10✔
1224
                return NewErrf(ErrIgnored, "ignoring update (flags=%v|%v) for "+
1✔
1225
                        "unknown chan_id=%v", policy.MessageFlags,
1✔
1226
                        policy.ChannelFlags, policy.ChannelID)
1✔
1227
        }
1✔
1228

1229
        log.Debugf("Found edge1Timestamp=%v, edge2Timestamp=%v",
8✔
1230
                edge1Timestamp, edge2Timestamp)
8✔
1231

8✔
1232
        // As edges are directional edge node has a unique policy for the
8✔
1233
        // direction of the edge they control. Therefore, we first check if we
8✔
1234
        // already have the most up-to-date information for that edge. If this
8✔
1235
        // message has a timestamp not strictly newer than what we already know
8✔
1236
        // of we can exit early.
8✔
1237
        switch policy.ChannelFlags & lnwire.ChanUpdateDirection {
8✔
1238
        // A flag set of 0 indicates this is an announcement for the "first"
1239
        // node in the channel.
1240
        case 0:
6✔
1241
                // Ignore outdated message.
6✔
1242
                if !edge1Timestamp.Before(policy.LastUpdate) {
9✔
1243
                        return NewErrf(ErrOutdated, "Ignoring "+
3✔
1244
                                "outdated update (flags=%v|%v) for "+
3✔
1245
                                "known chan_id=%v", policy.MessageFlags,
3✔
1246
                                policy.ChannelFlags, policy.ChannelID)
3✔
1247
                }
3✔
1248

1249
        // Similarly, a flag set of 1 indicates this is an announcement
1250
        // for the "second" node in the channel.
1251
        case 1:
5✔
1252
                // Ignore outdated message.
5✔
1253
                if !edge2Timestamp.Before(policy.LastUpdate) {
8✔
1254
                        return NewErrf(ErrOutdated, "Ignoring "+
3✔
1255
                                "outdated update (flags=%v|%v) for "+
3✔
1256
                                "known chan_id=%v", policy.MessageFlags,
3✔
1257
                                policy.ChannelFlags, policy.ChannelID)
3✔
1258
                }
3✔
1259
        }
1260

1261
        // Now that we know this isn't a stale update, we'll apply the new edge
1262
        // policy to the proper directional edge within the channel graph.
1263
        if err = b.cfg.Graph.UpdateEdgePolicy(ctx, policy, op...); err != nil {
8✔
1264
                err := fmt.Errorf("unable to add channel: %w", err)
×
1265
                log.Error(err)
×
1266
                return err
×
1267
        }
×
1268

1269
        log.Tracef("New channel update applied: %v",
8✔
1270
                lnutils.SpewLogClosure(policy))
8✔
1271
        b.stats.incNumChannelUpdates()
8✔
1272

8✔
1273
        return nil
8✔
1274
}
1275

1276
// logNetworkMsgProcessError logs the error received from processing a network
1277
// message. It logs as a debug message if the error is not critical.
1278
func logNetworkMsgProcessError(err error) {
5✔
1279
        if IsError(err, ErrIgnored, ErrOutdated) {
10✔
1280
                log.Debugf("process network updates got: %v", err)
5✔
1281

5✔
1282
                return
5✔
1283
        }
5✔
1284

1285
        log.Errorf("process network updates got: %v", err)
×
1286
}
1287

1288
// CurrentBlockHeight returns the block height from POV of the router subsystem.
1289
//
1290
// NOTE: This method is part of the ChannelGraphSource interface.
1291
func (b *Builder) CurrentBlockHeight() (uint32, error) {
3✔
1292
        _, height, err := b.cfg.Chain.GetBestBlock()
3✔
1293
        return uint32(height), err
3✔
1294
}
3✔
1295

1296
// SyncedHeight returns the block height to which the router subsystem currently
1297
// is synced to. This can differ from the above chain height if the goroutine
1298
// responsible for processing the blocks isn't yet up to speed.
1299
func (b *Builder) SyncedHeight() uint32 {
3✔
1300
        return b.bestHeight.Load()
3✔
1301
}
3✔
1302

1303
// GetChannelByID return the channel by the channel id.
1304
//
1305
// NOTE: This method is part of the ChannelGraphSource interface.
1306
func (b *Builder) GetChannelByID(chanID lnwire.ShortChannelID) (
1307
        *models.ChannelEdgeInfo,
1308
        *models.ChannelEdgePolicy,
1309
        *models.ChannelEdgePolicy, error) {
4✔
1310

4✔
1311
        return b.cfg.Graph.FetchChannelEdgesByID(
4✔
1312
                context.TODO(), chanID.ToUint64(),
4✔
1313
        )
4✔
1314
}
4✔
1315

1316
// FetchNode attempts to look up a target node by its identity public
1317
// key. graphdb.ErrGraphNodeNotFound is returned if the node doesn't exist
1318
// within the graph.
1319
//
1320
// NOTE: This method is part of the ChannelGraphSource interface.
1321
func (b *Builder) FetchNode(ctx context.Context,
1322
        node route.Vertex) (*models.Node, error) {
3✔
1323

3✔
1324
        return b.v1Graph.FetchNode(ctx, node)
3✔
1325
}
3✔
1326

1327
// ForAllOutgoingChannels is used to iterate over all outgoing channels owned by
1328
// the router.
1329
//
1330
// NOTE: This method is part of the ChannelGraphSource interface.
1331
func (b *Builder) ForAllOutgoingChannels(ctx context.Context,
1332
        cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy) error,
1333
        reset func()) error {
3✔
1334

3✔
1335
        return b.cfg.Graph.ForEachNodeChannel(
3✔
1336
                ctx, lnwire.GossipVersion1, b.cfg.SelfNode,
3✔
1337
                func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy,
3✔
1338
                        _ *models.ChannelEdgePolicy) error {
6✔
1339

3✔
1340
                        if e == nil {
3✔
1341
                                return fmt.Errorf("channel from self node " +
×
1342
                                        "has no policy")
×
1343
                        }
×
1344

1345
                        return cb(c, e)
3✔
1346
                }, reset,
1347
        )
1348
}
1349

1350
// AddProof updates the channel edge info with proof which is needed to
1351
// properly announce the edge to the rest of the network.
1352
//
1353
// NOTE: This method is part of the ChannelGraphSource interface.
1354
func (b *Builder) AddProof(chanID lnwire.ShortChannelID,
1355
        proof *models.ChannelAuthProof) error {
4✔
1356

4✔
1357
        return b.cfg.Graph.AddEdgeProof(context.TODO(), chanID, proof)
4✔
1358
}
4✔
1359

1360
// IsStaleNode returns true if the graph source has a node announcement for the
1361
// target node with a more recent timestamp.
1362
//
1363
// NOTE: This method is part of the ChannelGraphSource interface.
1364
func (b *Builder) IsStaleNode(ctx context.Context, node route.Vertex,
1365
        timestamp time.Time) bool {
6✔
1366

6✔
1367
        // If our attempt to assert that the node announcement is fresh fails,
6✔
1368
        // then we know that this is actually a stale announcement.
6✔
1369
        err := b.assertNodeAnnFreshness(ctx, node, timestamp)
6✔
1370
        if err != nil {
10✔
1371
                log.Debugf("Checking stale node %s got %v", node, err)
4✔
1372
                return true
4✔
1373
        }
4✔
1374

1375
        return false
5✔
1376
}
1377

1378
// IsPublicNode determines whether the given vertex is seen as a public node in
1379
// the graph from the graph's source node's point of view.
1380
//
1381
// NOTE: This method is part of the ChannelGraphSource interface.
1382
func (b *Builder) IsPublicNode(node route.Vertex) (bool, error) {
3✔
1383
        return b.v1Graph.IsPublicNode(context.TODO(), node)
3✔
1384
}
3✔
1385

1386
// IsKnownEdge returns true if the graph source already knows of the passed
1387
// channel ID either as a live or zombie edge.
1388
//
1389
// NOTE: This method is part of the ChannelGraphSource interface.
1390
func (b *Builder) IsKnownEdge(chanID lnwire.ShortChannelID) bool {
4✔
1391
        exists, isZombie, _ := b.cfg.Graph.HasChannelEdge(
4✔
1392
                context.TODO(), lnwire.GossipVersion1, chanID.ToUint64(),
4✔
1393
        )
4✔
1394

4✔
1395
        return exists || isZombie
4✔
1396
}
4✔
1397

1398
// IsZombieEdge returns true if the graph source has marked the given channel ID
1399
// as a zombie edge.
1400
//
1401
// NOTE: This method is part of the ChannelGraphSource interface.
1402
func (b *Builder) IsZombieEdge(chanID lnwire.ShortChannelID) (bool, error) {
×
1403
        _, isZombie, err := b.cfg.Graph.HasChannelEdge(
×
1404
                context.TODO(), lnwire.GossipVersion1, chanID.ToUint64(),
×
1405
        )
×
1406

×
1407
        return isZombie, err
×
1408
}
×
1409

1410
// IsStaleEdgePolicy returns true if the graph source has a channel edge for
1411
// the passed channel ID (and flags) that have a more recent timestamp.
1412
//
1413
// NOTE: This method is part of the ChannelGraphSource interface.
1414
func (b *Builder) IsStaleEdgePolicy(chanID lnwire.ShortChannelID,
1415
        timestamp time.Time, flags lnwire.ChanUpdateChanFlags) bool {
9✔
1416

9✔
1417
        edge1Timestamp, edge2Timestamp, exists, isZombie, err :=
9✔
1418
                b.cfg.Graph.HasV1ChannelEdge(
9✔
1419
                        context.TODO(), chanID.ToUint64(),
9✔
1420
                )
9✔
1421
        if err != nil {
9✔
1422
                log.Debugf("Check stale edge policy got error: %v", err)
×
1423
                return false
×
1424
        }
×
1425

1426
        // If we know of the edge as a zombie, then we'll make some additional
1427
        // checks to determine if the new policy is fresh.
1428
        if isZombie {
9✔
1429
                // When running with AssumeChannelValid, we also prune channels
×
1430
                // if both of their edges are disabled. We'll mark the new
×
1431
                // policy as stale if it remains disabled.
×
1432
                if b.cfg.AssumeChannelValid {
×
1433
                        isDisabled := flags&lnwire.ChanUpdateDisabled ==
×
1434
                                lnwire.ChanUpdateDisabled
×
1435
                        if isDisabled {
×
1436
                                return true
×
1437
                        }
×
1438
                }
1439

1440
                // Otherwise, we'll fall back to our usual ChannelPruneExpiry.
1441
                return time.Since(timestamp) > b.cfg.ChannelPruneExpiry
×
1442
        }
1443

1444
        // If we don't know of the edge, then it means it's fresh (thus not
1445
        // stale).
1446
        if !exists {
14✔
1447
                return false
5✔
1448
        }
5✔
1449

1450
        // As edges are directional edge node has a unique policy for the
1451
        // direction of the edge they control. Therefore, we first check if we
1452
        // already have the most up-to-date information for that edge. If so,
1453
        // then we can exit early.
1454
        switch {
7✔
1455
        // A flag set of 0 indicates this is an announcement for the "first"
1456
        // node in the channel.
1457
        case flags&lnwire.ChanUpdateDirection == 0:
5✔
1458
                return !edge1Timestamp.Before(timestamp)
5✔
1459

1460
        // Similarly, a flag set of 1 indicates this is an announcement for the
1461
        // "second" node in the channel.
1462
        case flags&lnwire.ChanUpdateDirection == 1:
5✔
1463
                return !edge2Timestamp.Before(timestamp)
5✔
1464
        }
1465

1466
        return false
×
1467
}
1468

1469
// MarkEdgeLive clears an edge from our zombie index for the given gossip
1470
// version, deeming it as live.
1471
//
1472
// NOTE: This method is part of the ChannelGraphSource interface.
1473
func (b *Builder) MarkEdgeLive(v lnwire.GossipVersion,
NEW
1474
        chanID lnwire.ShortChannelID) error {
×
NEW
1475

×
1476
        return b.cfg.Graph.MarkEdgeLive(
×
NEW
1477
                context.TODO(), v, chanID.ToUint64(),
×
1478
        )
×
1479
}
×
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

© 2026 Coveralls, Inc