highperfocused/lnd
1
0
Fork 0
mirror of https://github.com/lightningnetwork/lnd.git synced 2025-06-27 17:21:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #9476 from ellemouton/graph1

Browse Source 
graph: refactor `graph.Builder` update handling
This commit is contained in:
Oliver Gugger 2025-02-07 07:23:41 -06:00 committed by GitHub
commit 3c0350e481
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 381 additions and 398 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
discovery/gossiper.go
View File

@ -1977,7 +1977,7 @@ func (d *AuthenticatedGossiper) fetchPKScript(chanID *lnwire.ShortChannelID) (
func (d *AuthenticatedGossiper) addNode(msg *lnwire.NodeAnnouncement, func (d *AuthenticatedGossiper) addNode(msg *lnwire.NodeAnnouncement,
op ...batch.SchedulerOption) error { op ...batch.SchedulerOption) error {
if err := graph.ValidateNodeAnn(msg); err != nil { if err := netann.ValidateNodeAnn(msg); err != nil {
return fmt.Errorf("unable to validate node announcement: %w", return fmt.Errorf("unable to validate node announcement: %w",
err) err)
} }

3
docs/release-notes/release-notes-0.19.0.md
View File

@ -243,6 +243,9 @@ The underlying functionality between those two options remain the same.
* [Golang was updated to * [Golang was updated to
`v1.22.11`](https://github.com/lightningnetwork/lnd/pull/9462). `v1.22.11`](https://github.com/lightningnetwork/lnd/pull/9462).
* Various refactors to simplify the
`graph.Builder` [1](https://github.com/lightningnetwork/lnd/pull/9476).
## Breaking Changes ## Breaking Changes
## Performance Improvements ## Performance Improvements

47
graph/ann_validation.go
View File

@ -1,47 +0,0 @@
package graph
import (
"bytes"
"github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/go-errors/errors"
"github.com/lightningnetwork/lnd/lnwire"
)
// ValidateNodeAnn validates the node announcement by ensuring that the
// attached signature is needed a signature of the node announcement under the
// specified node public key.
func ValidateNodeAnn(a *lnwire.NodeAnnouncement) error {
// Reconstruct the data of announcement which should be covered by the
// signature so we can verify the signature shortly below
data, err := a.DataToSign()
if err != nil {
return err
}
nodeSig, err := a.Signature.ToSignature()
if err != nil {
return err
}
nodeKey, err := btcec.ParsePubKey(a.NodeID[:])
if err != nil {
return err
}
// Finally ensure that the passed signature is valid, if not we'll
// return an error so this node announcement can be rejected.
dataHash := chainhash.DoubleHashB(data)
if !nodeSig.Verify(dataHash, nodeKey) {
var msgBuf bytes.Buffer
if _, err := lnwire.WriteMessage(&msgBuf, a, 0); err != nil {
return err
}
return errors.Errorf("signature on NodeAnnouncement(%x) is "+
"invalid: %x", nodeKey.SerializeCompressed(),
msgBuf.Bytes())
}
return nil
}

666
graph/builder.go
View File

@ -131,11 +131,6 @@ type Builder struct {
// of our currently known best chain are sent over. // of our currently known best chain are sent over.
staleBlocks <-chan *chainview.FilteredBlock staleBlocks <-chan *chainview.FilteredBlock
// networkUpdates is a channel that carries new topology updates
// messages from outside the Builder to be processed by the
// networkHandler.
networkUpdates chan *routingMsg
// topologyClients maps a client's unique notification ID to a // topologyClients maps a client's unique notification ID to a
// topologyClient client that contains its notification dispatch // topologyClient client that contains its notification dispatch
// channel. // channel.
@ -158,7 +153,7 @@ type Builder struct {
// stats tracks newly processed channels, updates, and node // stats tracks newly processed channels, updates, and node
// announcements over a window of defaultStatInterval. // announcements over a window of defaultStatInterval.
stats *routerStats stats *builderStats
quit chan struct{} quit chan struct{}
wg sync.WaitGroup wg sync.WaitGroup
@ -172,12 +167,11 @@ var _ ChannelGraphSource = (*Builder)(nil)
func NewBuilder(cfg *Config) (*Builder, error) { func NewBuilder(cfg *Config) (*Builder, error) {
return &Builder{ return &Builder{
cfg: cfg, cfg: cfg,
networkUpdates: make(chan *routingMsg),
topologyClients: &lnutils.SyncMap[uint64, *topologyClient]{}, topologyClients: &lnutils.SyncMap[uint64, *topologyClient]{},
ntfnClientUpdates: make(chan *topologyClientUpdate), ntfnClientUpdates: make(chan *topologyClientUpdate),
channelEdgeMtx: multimutex.NewMutex[uint64](), channelEdgeMtx: multimutex.NewMutex[uint64](),
statTicker: ticker.New(defaultStatInterval), statTicker: ticker.New(defaultStatInterval),
stats: new(routerStats), stats: new(builderStats),
quit: make(chan struct{}), quit: make(chan struct{}),
}, nil }, nil
} }
@ -674,7 +668,20 @@ func (b *Builder) handleNetworkUpdate(update *routingMsg) {
// Process the routing update to determine if this is either a new // Process the routing update to determine if this is either a new
// update from our PoV or an update to a prior vertex/edge we // update from our PoV or an update to a prior vertex/edge we
// previously accepted. // previously accepted.
err := b.processUpdate(update.msg, update.op...) var err error
switch msg := update.msg.(type) {
case *models.LightningNode:
err = b.addNode(msg, update.op...)
case *models.ChannelEdgeInfo:
err = b.addEdge(msg, update.op...)
case *models.ChannelEdgePolicy:
err = b.updateEdge(msg, update.op...)
default:
err = errors.Errorf("wrong routing update message type")
}
update.err <- err update.err <- err
// If the error is not nil here, there's no need to send topology // If the error is not nil here, there's no need to send topology
@ -708,8 +715,8 @@ func (b *Builder) handleNetworkUpdate(update *routingMsg) {
// networkHandler is the primary goroutine for the Builder. The roles of // networkHandler is the primary goroutine for the Builder. The roles of
// this goroutine include answering queries related to the state of the // this goroutine include answering queries related to the state of the
// network, pruning the graph on new block notification, applying network // network, pruning the graph on new block notification and registering new
// updates, and registering new topology clients. // topology clients.
// //
// NOTE: This MUST be run as a goroutine. // NOTE: This MUST be run as a goroutine.
func (b *Builder) networkHandler() { func (b *Builder) networkHandler() {
@ -729,17 +736,6 @@ func (b *Builder) networkHandler() {
} }
select { select {
// A new fully validated network update has just arrived. As a
// result we'll modify the channel graph accordingly depending
// on the exact type of the message.
case update := <-b.networkUpdates:
b.wg.Add(1)
go b.handleNetworkUpdate(update)
// TODO(roasbeef): remove all unconnected vertexes
// after N blocks pass with no corresponding
// announcements.
case chainUpdate, ok := <-b.staleBlocks: case chainUpdate, ok := <-b.staleBlocks:
// If the channel has been closed, then this indicates // If the channel has been closed, then this indicates
// the daemon is shutting down, so we exit ourselves. // the daemon is shutting down, so we exit ourselves.
@ -1094,303 +1090,6 @@ func makeFundingScript(bitcoinKey1, bitcoinKey2 []byte, chanFeatures []byte,
return legacyFundingScript() return legacyFundingScript()
} }
// processUpdate processes a new relate authenticated channel/edge, node or
// channel/edge update network update. If the update didn't affect the internal
// state of the draft due to either being out of date, invalid, or redundant,
// then error is returned.
//
//nolint:funlen
func (b *Builder) processUpdate(msg interface{},
op ...batch.SchedulerOption) error {
switch msg := msg.(type) {
case *models.LightningNode:
// Before we add the node to the database, we'll check to see
// if the announcement is "fresh" or not. If it isn't, then
// we'll return an error.
err := b.assertNodeAnnFreshness(msg.PubKeyBytes, msg.LastUpdate)
if err != nil {
return err
}
if err := b.cfg.Graph.AddLightningNode(msg, op...); err != nil {
return errors.Errorf("unable to add node %x to the "+
"graph: %v", msg.PubKeyBytes, err)
}
log.Tracef("Updated vertex data for node=%x", msg.PubKeyBytes)
b.stats.incNumNodeUpdates()
case *models.ChannelEdgeInfo:
log.Debugf("Received ChannelEdgeInfo for channel %v",
msg.ChannelID)
// Prior to processing the announcement we first check if we
// already know of this channel, if so, then we can exit early.
_, _, exists, isZombie, err := b.cfg.Graph.HasChannelEdge(
msg.ChannelID,
)
if err != nil &&
!errors.Is(err, graphdb.ErrGraphNoEdgesFound) {
return errors.Errorf("unable to check for edge "+
"existence: %v", err)
}
if isZombie {
return NewErrf(ErrIgnored, "ignoring msg for zombie "+
"chan_id=%v", msg.ChannelID)
}
if exists {
return NewErrf(ErrIgnored, "ignoring msg for known "+
"chan_id=%v", msg.ChannelID)
}
// If AssumeChannelValid is present, then we are unable to
// perform any of the expensive checks below, so we'll
// short-circuit our path straight to adding the edge to our
// graph. If the passed ShortChannelID is an alias, then we'll
// skip validation as it will not map to a legitimate tx. This
// is not a DoS vector as only we can add an alias
// ChannelAnnouncement from the gossiper.
scid := lnwire.NewShortChanIDFromInt(msg.ChannelID)
if b.cfg.AssumeChannelValid || b.cfg.IsAlias(scid) {
err := b.cfg.Graph.AddChannelEdge(msg, op...)
if err != nil {
return fmt.Errorf("unable to add edge: %w", err)
}
log.Tracef("New channel discovered! Link "+
"connects %x and %x with ChannelID(%v)",
msg.NodeKey1Bytes, msg.NodeKey2Bytes,
msg.ChannelID)
b.stats.incNumEdgesDiscovered()
break
}
// Before we can add the channel to the channel graph, we need
// to obtain the full funding outpoint that's encoded within
// the channel ID.
channelID := lnwire.NewShortChanIDFromInt(msg.ChannelID)
fundingTx, err := lnwallet.FetchFundingTxWrapper(
b.cfg.Chain, &channelID, b.quit,
)
if err != nil {
//nolint:ll
//
// In order to ensure we don't erroneously mark a
// channel as a zombie due to an RPC failure, we'll
// attempt to string match for the relevant errors.
//
// * btcd:
// * https://github.com/btcsuite/btcd/blob/master/rpcserver.go#L1316
// * https://github.com/btcsuite/btcd/blob/master/rpcserver.go#L1086
// * bitcoind:
// * https://github.com/bitcoin/bitcoin/blob/7fcf53f7b4524572d1d0c9a5fdc388e87eb02416/src/rpc/blockchain.cpp#L770
// * https://github.com/bitcoin/bitcoin/blob/7fcf53f7b4524572d1d0c9a5fdc388e87eb02416/src/rpc/blockchain.cpp#L954
switch {
case strings.Contains(err.Error(), "not found"):
fallthrough
case strings.Contains(err.Error(), "out of range"):
// If the funding transaction isn't found at
// all, then we'll mark the edge itself as a
// zombie so we don't continue to request it.
// We use the "zero key" for both node pubkeys
// so this edge can't be resurrected.
zErr := b.addZombieEdge(msg.ChannelID)
if zErr != nil {
return zErr
}
default:
}
return NewErrf(ErrNoFundingTransaction, "unable to "+
"locate funding tx: %v", err)
}
// Recreate witness output to be sure that declared in channel
// edge bitcoin keys and channel value corresponds to the
// reality.
fundingPkScript, err := makeFundingScript(
msg.BitcoinKey1Bytes[:], msg.BitcoinKey2Bytes[:],
msg.Features, msg.TapscriptRoot,
)
if err != nil {
return err
}
// Next we'll validate that this channel is actually well
// formed. If this check fails, then this channel either
// doesn't exist, or isn't the one that was meant to be created
// according to the passed channel proofs.
fundingPoint, err := chanvalidate.Validate(
&chanvalidate.Context{
Locator: &chanvalidate.ShortChanIDChanLocator{
ID: channelID,
},
MultiSigPkScript: fundingPkScript,
FundingTx: fundingTx,
},
)
if err != nil {
// Mark the edge as a zombie so we won't try to
// re-validate it on start up.
if err := b.addZombieEdge(msg.ChannelID); err != nil {
return err
}
return NewErrf(ErrInvalidFundingOutput, "output "+
"failed validation: %w", err)
}
// Now that we have the funding outpoint of the channel, ensure
// that it hasn't yet been spent. If so, then this channel has
// been closed so we'll ignore it.
chanUtxo, err := b.cfg.Chain.GetUtxo(
fundingPoint, fundingPkScript, channelID.BlockHeight,
b.quit,
)
if err != nil {
if errors.Is(err, btcwallet.ErrOutputSpent) {
zErr := b.addZombieEdge(msg.ChannelID)
if zErr != nil {
return zErr
}
}
return NewErrf(ErrChannelSpent, "unable to fetch utxo "+
"for chan_id=%v, chan_point=%v: %v",
msg.ChannelID, fundingPoint, err)
}
// TODO(roasbeef): this is a hack, needs to be removed
// after commitment fees are dynamic.
msg.Capacity = btcutil.Amount(chanUtxo.Value)
msg.ChannelPoint = *fundingPoint
if err := b.cfg.Graph.AddChannelEdge(msg, op...); err != nil {
return errors.Errorf("unable to add edge: %v", err)
}
log.Debugf("New channel discovered! Link "+
"connects %x and %x with ChannelPoint(%v): "+
"chan_id=%v, capacity=%v",
msg.NodeKey1Bytes, msg.NodeKey2Bytes,
fundingPoint, msg.ChannelID, msg.Capacity)
b.stats.incNumEdgesDiscovered()
// As a new edge has been added to the channel graph, we'll
// update the current UTXO filter within our active
// FilteredChainView so we are notified if/when this channel is
// closed.
filterUpdate := []graphdb.EdgePoint{
{
FundingPkScript: fundingPkScript,
OutPoint: *fundingPoint,
},
}
err = b.cfg.ChainView.UpdateFilter(
filterUpdate, b.bestHeight.Load(),
)
if err != nil {
return errors.Errorf("unable to update chain "+
"view: %v", err)
}
case *models.ChannelEdgePolicy:
log.Debugf("Received ChannelEdgePolicy for channel %v",
msg.ChannelID)
// We make sure to hold the mutex for this channel ID,
// such that no other goroutine is concurrently doing
// database accesses for the same channel ID.
b.channelEdgeMtx.Lock(msg.ChannelID)
defer b.channelEdgeMtx.Unlock(msg.ChannelID)
edge1Timestamp, edge2Timestamp, exists, isZombie, err :=
b.cfg.Graph.HasChannelEdge(msg.ChannelID)
if err != nil && !errors.Is(
err, graphdb.ErrGraphNoEdgesFound,
) {
return errors.Errorf("unable to check for edge "+
"existence: %v", err)
}
// If the channel is marked as a zombie in our database, and
// we consider this a stale update, then we should not apply the
// policy.
isStaleUpdate := time.Since(msg.LastUpdate) >
b.cfg.ChannelPruneExpiry
if isZombie && isStaleUpdate {
return NewErrf(ErrIgnored, "ignoring stale update "+
"(flags=%v|%v) for zombie chan_id=%v",
msg.MessageFlags, msg.ChannelFlags,
msg.ChannelID)
}
// If the channel doesn't exist in our database, we cannot
// apply the updated policy.
if !exists {
return NewErrf(ErrIgnored, "ignoring update "+
"(flags=%v|%v) for unknown chan_id=%v",
msg.MessageFlags, msg.ChannelFlags,
msg.ChannelID)
}
log.Debugf("Found edge1Timestamp=%v, edge2Timestamp=%v",
edge1Timestamp, edge2Timestamp)
// As edges are directional edge node has a unique policy for
// the direction of the edge they control. Therefore, we first
// check if we already have the most up-to-date information for
// that edge. If this message has a timestamp not strictly
// newer than what we already know of we can exit early.
switch msg.ChannelFlags & lnwire.ChanUpdateDirection {
// A flag set of 0 indicates this is an announcement for the
// "first" node in the channel.
case 0:
// Ignore outdated message.
if !edge1Timestamp.Before(msg.LastUpdate) {
return NewErrf(ErrOutdated, "Ignoring "+
"outdated update (flags=%v|%v) for "+
"known chan_id=%v", msg.MessageFlags,
msg.ChannelFlags, msg.ChannelID)
}
// Similarly, a flag set of 1 indicates this is an announcement
// for the "second" node in the channel.
case 1:
// Ignore outdated message.
if !edge2Timestamp.Before(msg.LastUpdate) {
return NewErrf(ErrOutdated, "Ignoring "+
"outdated update (flags=%v|%v) for "+
"known chan_id=%v", msg.MessageFlags,
msg.ChannelFlags, msg.ChannelID)
}
}
// Now that we know this isn't a stale update, we'll apply the
// new edge policy to the proper directional edge within the
// channel graph.
if err = b.cfg.Graph.UpdateEdgePolicy(msg, op...); err != nil {
err := errors.Errorf("unable to add channel: %v", err)
log.Error(err)
return err
}
log.Tracef("New channel update applied: %v",
lnutils.SpewLogClosure(msg))
b.stats.incNumChannelUpdates()
default:
return errors.Errorf("wrong routing update message type")
}
return nil
}
// routingMsg couples a routing related routing topology update to the // routingMsg couples a routing related routing topology update to the
// error channel. // error channel.
type routingMsg struct { type routingMsg struct {
@ -1466,19 +1165,43 @@ func (b *Builder) AddNode(node *models.LightningNode,
err: make(chan error, 1), err: make(chan error, 1),
} }
b.wg.Add(1)
go b.handleNetworkUpdate(rMsg)
select { select {
case b.networkUpdates <- rMsg: case err := <-rMsg.err:
select { return err
case err := <-rMsg.err:
return err
case <-b.quit:
return ErrGraphBuilderShuttingDown
}
case <-b.quit: case <-b.quit:
return ErrGraphBuilderShuttingDown return ErrGraphBuilderShuttingDown
} }
} }
// addNode does some basic checks on the given LightningNode against what we
// currently have persisted in the graph, and then adds it to the graph. If we
// already know about the node, then we only update our DB if the new update
// has a newer timestamp than the last one we received.
func (b *Builder) addNode(node *models.LightningNode,
op ...batch.SchedulerOption) error {
// Before we add the node to the database, we'll check to see if the
// announcement is "fresh" or not. If it isn't, then we'll return an
// error.
err := b.assertNodeAnnFreshness(node.PubKeyBytes, node.LastUpdate)
if err != nil {
return err
}
if err := b.cfg.Graph.AddLightningNode(node, op...); err != nil {
return errors.Errorf("unable to add node %x to the "+
"graph: %v", node.PubKeyBytes, err)
}
log.Tracef("Updated vertex data for node=%x", node.PubKeyBytes)
b.stats.incNumNodeUpdates()
return nil
}
// AddEdge is used to add edge/channel to the topology of the router, after all // AddEdge is used to add edge/channel to the topology of the router, after all
// information about channel will be gathered this edge/channel might be used // information about channel will be gathered this edge/channel might be used
// in construction of payment path. // in construction of payment path.
@ -1493,19 +1216,193 @@ func (b *Builder) AddEdge(edge *models.ChannelEdgeInfo,
err: make(chan error, 1), err: make(chan error, 1),
} }
b.wg.Add(1)
go b.handleNetworkUpdate(rMsg)
select { select {
case b.networkUpdates <- rMsg: case err := <-rMsg.err:
select { return err
case err := <-rMsg.err:
return err
case <-b.quit:
return ErrGraphBuilderShuttingDown
}
case <-b.quit: case <-b.quit:
return ErrGraphBuilderShuttingDown return ErrGraphBuilderShuttingDown
} }
} }
// addEdge does some validation on the new channel edge against what we
// currently have persisted in the graph, and then adds it to the graph. The
// Chain View is updated with the new edge if it is successfully added to the
// graph. We only persist the channel if we currently dont have it at all in
// our graph.
//
// TODO(elle): this currently also does funding-transaction validation. But this
// should be moved to the gossiper instead.
func (b *Builder) addEdge(edge *models.ChannelEdgeInfo,
op ...batch.SchedulerOption) error {
log.Debugf("Received ChannelEdgeInfo for channel %v", edge.ChannelID)
// Prior to processing the announcement we first check if we
// already know of this channel, if so, then we can exit early.
_, _, exists, isZombie, err := b.cfg.Graph.HasChannelEdge(
edge.ChannelID,
)
if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) {
return errors.Errorf("unable to check for edge existence: %v",
err)
}
if isZombie {
return NewErrf(ErrIgnored, "ignoring msg for zombie chan_id=%v",
edge.ChannelID)
}
if exists {
return NewErrf(ErrIgnored, "ignoring msg for known chan_id=%v",
edge.ChannelID)
}
// If AssumeChannelValid is present, then we are unable to perform any
// of the expensive checks below, so we'll short-circuit our path
// straight to adding the edge to our graph. If the passed
// ShortChannelID is an alias, then we'll skip validation as it will
// not map to a legitimate tx. This is not a DoS vector as only we can
// add an alias ChannelAnnouncement from the gossiper.
scid := lnwire.NewShortChanIDFromInt(edge.ChannelID)
if b.cfg.AssumeChannelValid || b.cfg.IsAlias(scid) {
err := b.cfg.Graph.AddChannelEdge(edge, op...)
if err != nil {
return fmt.Errorf("unable to add edge: %w", err)
}
log.Tracef("New channel discovered! Link connects %x and %x "+
"with ChannelID(%v)", edge.NodeKey1Bytes,
edge.NodeKey2Bytes, edge.ChannelID)
b.stats.incNumEdgesDiscovered()
return nil
}
// Before we can add the channel to the channel graph, we need to obtain
// the full funding outpoint that's encoded within the channel ID.
channelID := lnwire.NewShortChanIDFromInt(edge.ChannelID)
fundingTx, err := lnwallet.FetchFundingTxWrapper(
b.cfg.Chain, &channelID, b.quit,
)
if err != nil {
//nolint:ll
//
// In order to ensure we don't erroneously mark a channel as a
// zombie due to an RPC failure, we'll attempt to string match
// for the relevant errors.
//
// * btcd:
// * https://github.com/btcsuite/btcd/blob/master/rpcserver.go#L1316
// * https://github.com/btcsuite/btcd/blob/master/rpcserver.go#L1086
// * bitcoind:
// * https://github.com/bitcoin/bitcoin/blob/7fcf53f7b4524572d1d0c9a5fdc388e87eb02416/src/rpc/blockchain.cpp#L770
// * https://github.com/bitcoin/bitcoin/blob/7fcf53f7b4524572d1d0c9a5fdc388e87eb02416/src/rpc/blockchain.cpp#L954
switch {
case strings.Contains(err.Error(), "not found"):
fallthrough
case strings.Contains(err.Error(), "out of range"):
// If the funding transaction isn't found at all, then
// we'll mark the edge itself as a zombie so we don't
// continue to request it. We use the "zero key" for
// both node pubkeys so this edge can't be resurrected.
zErr := b.addZombieEdge(edge.ChannelID)
if zErr != nil {
return zErr
}
default:
}
return NewErrf(ErrNoFundingTransaction, "unable to "+
"locate funding tx: %v", err)
}
// Recreate witness output to be sure that declared in channel edge
// bitcoin keys and channel value corresponds to the reality.
fundingPkScript, err := makeFundingScript(
edge.BitcoinKey1Bytes[:], edge.BitcoinKey2Bytes[:],
edge.Features, edge.TapscriptRoot,
)
if err != nil {
return err
}
// Next we'll validate that this channel is actually well formed. If
// this check fails, then this channel either doesn't exist, or isn't
// the one that was meant to be created according to the passed channel
// proofs.
fundingPoint, err := chanvalidate.Validate(
&chanvalidate.Context{
Locator: &chanvalidate.ShortChanIDChanLocator{
ID: channelID,
},
MultiSigPkScript: fundingPkScript,
FundingTx: fundingTx,
},
)
if err != nil {
// Mark the edge as a zombie so we won't try to re-validate it
// on start up.
if err := b.addZombieEdge(edge.ChannelID); err != nil {
return err
}
return NewErrf(ErrInvalidFundingOutput, "output failed "+
"validation: %w", err)
}
// Now that we have the funding outpoint of the channel, ensure
// that it hasn't yet been spent. If so, then this channel has
// been closed so we'll ignore it.
chanUtxo, err := b.cfg.Chain.GetUtxo(
fundingPoint, fundingPkScript, channelID.BlockHeight, b.quit,
)
if err != nil {
if errors.Is(err, btcwallet.ErrOutputSpent) {
zErr := b.addZombieEdge(edge.ChannelID)
if zErr != nil {
return zErr
}
}
return NewErrf(ErrChannelSpent, "unable to fetch utxo for "+
"chan_id=%v, chan_point=%v: %v", edge.ChannelID,
fundingPoint, err)
}
// TODO(roasbeef): this is a hack, needs to be removed after commitment
// fees are dynamic.
edge.Capacity = btcutil.Amount(chanUtxo.Value)
edge.ChannelPoint = *fundingPoint
if err := b.cfg.Graph.AddChannelEdge(edge, op...); err != nil {
return errors.Errorf("unable to add edge: %v", err)
}
log.Debugf("New channel discovered! Link connects %x and %x with "+
"ChannelPoint(%v): chan_id=%v, capacity=%v", edge.NodeKey1Bytes,
edge.NodeKey2Bytes, fundingPoint, edge.ChannelID, edge.Capacity)
b.stats.incNumEdgesDiscovered()
// As a new edge has been added to the channel graph, we'll update the
// current UTXO filter within our active FilteredChainView so we are
// notified if/when this channel is closed.
filterUpdate := []graphdb.EdgePoint{
{
FundingPkScript: fundingPkScript,
OutPoint: *fundingPoint,
},
}
err = b.cfg.ChainView.UpdateFilter(filterUpdate, b.bestHeight.Load())
if err != nil {
return errors.Errorf("unable to update chain "+
"view: %v", err)
}
return nil
}
// UpdateEdge is used to update edge information, without this message edge // UpdateEdge is used to update edge information, without this message edge
// considered as not fully constructed. // considered as not fully constructed.
// //
@ -1519,19 +1416,108 @@ func (b *Builder) UpdateEdge(update *models.ChannelEdgePolicy,
err: make(chan error, 1), err: make(chan error, 1),
} }
b.wg.Add(1)
go b.handleNetworkUpdate(rMsg)
select { select {
case b.networkUpdates <- rMsg: case err := <-rMsg.err:
select { return err
case err := <-rMsg.err:
return err
case <-b.quit:
return ErrGraphBuilderShuttingDown
}
case <-b.quit: case <-b.quit:
return ErrGraphBuilderShuttingDown return ErrGraphBuilderShuttingDown
} }
} }
// updateEdge validates the new edge policy against what we currently have
// persisted in the graph, and then applies it to the graph if the update is
// considered fresh enough and if we actually have a channel persisted for the
// given update.
func (b *Builder) updateEdge(policy *models.ChannelEdgePolicy,
op ...batch.SchedulerOption) error {
log.Debugf("Received ChannelEdgePolicy for channel %v",
policy.ChannelID)
// We make sure to hold the mutex for this channel ID, such that no
// other goroutine is concurrently doing database accesses for the same
// channel ID.
b.channelEdgeMtx.Lock(policy.ChannelID)
defer b.channelEdgeMtx.Unlock(policy.ChannelID)
edge1Timestamp, edge2Timestamp, exists, isZombie, err :=
b.cfg.Graph.HasChannelEdge(policy.ChannelID)
if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) {
return errors.Errorf("unable to check for edge existence: %v",
err)
}
// If the channel is marked as a zombie in our database, and
// we consider this a stale update, then we should not apply the
// policy.
isStaleUpdate := time.Since(policy.LastUpdate) >
b.cfg.ChannelPruneExpiry
if isZombie && isStaleUpdate {
return NewErrf(ErrIgnored, "ignoring stale update "+
"(flags=%v|%v) for zombie chan_id=%v",
policy.MessageFlags, policy.ChannelFlags,
policy.ChannelID)
}
// If the channel doesn't exist in our database, we cannot apply the
// updated policy.
if !exists {
return NewErrf(ErrIgnored, "ignoring update (flags=%v|%v) for "+
"unknown chan_id=%v", policy.MessageFlags,
policy.ChannelFlags, policy.ChannelID)
}
log.Debugf("Found edge1Timestamp=%v, edge2Timestamp=%v",
edge1Timestamp, edge2Timestamp)
// As edges are directional edge node has a unique policy for the
// direction of the edge they control. Therefore, we first check if we
// already have the most up-to-date information for that edge. If this
// message has a timestamp not strictly newer than what we already know
// of we can exit early.
switch policy.ChannelFlags & lnwire.ChanUpdateDirection {
// A flag set of 0 indicates this is an announcement for the "first"
// node in the channel.
case 0:
// Ignore outdated message.
if !edge1Timestamp.Before(policy.LastUpdate) {
return NewErrf(ErrOutdated, "Ignoring "+
"outdated update (flags=%v|%v) for "+
"known chan_id=%v", policy.MessageFlags,
policy.ChannelFlags, policy.ChannelID)
}
// Similarly, a flag set of 1 indicates this is an announcement
// for the "second" node in the channel.
case 1:
// Ignore outdated message.
if !edge2Timestamp.Before(policy.LastUpdate) {
return NewErrf(ErrOutdated, "Ignoring "+
"outdated update (flags=%v|%v) for "+
"known chan_id=%v", policy.MessageFlags,
policy.ChannelFlags, policy.ChannelID)
}
}
// Now that we know this isn't a stale update, we'll apply the new edge
// policy to the proper directional edge within the channel graph.
if err = b.cfg.Graph.UpdateEdgePolicy(policy, op...); err != nil {
err := errors.Errorf("unable to add channel: %v", err)
log.Error(err)
return err
}
log.Tracef("New channel update applied: %v",
lnutils.SpewLogClosure(policy))
b.stats.incNumChannelUpdates()
return nil
}
// CurrentBlockHeight returns the block height from POV of the router subsystem. // CurrentBlockHeight returns the block height from POV of the router subsystem.
// //
// NOTE: This method is part of the ChannelGraphSource interface. // NOTE: This method is part of the ChannelGraphSource interface.

20
graph/stats.go
View File

@ -6,9 +6,9 @@ import (
"time" "time"
) )
// routerStats is a struct that tracks various updates to the graph and // builderStats is a struct that tracks various updates to the graph and
// facilitates aggregate logging of the statistics. // facilitates aggregate logging of the statistics.
type routerStats struct { type builderStats struct {
numChannels uint32 numChannels uint32
numUpdates uint32 numUpdates uint32
numNodes uint32 numNodes uint32
@ -18,28 +18,28 @@ type routerStats struct {
} }
// incNumEdges increments the number of discovered edges. // incNumEdges increments the number of discovered edges.
func (g *routerStats) incNumEdgesDiscovered() { func (g *builderStats) incNumEdgesDiscovered() {
g.mu.Lock() g.mu.Lock()
g.numChannels++ g.numChannels++
g.mu.Unlock() g.mu.Unlock()
} }
// incNumUpdates increments the number of channel updates processed. // incNumUpdates increments the number of channel updates processed.
func (g *routerStats) incNumChannelUpdates() { func (g *builderStats) incNumChannelUpdates() {
g.mu.Lock() g.mu.Lock()
g.numUpdates++ g.numUpdates++
g.mu.Unlock() g.mu.Unlock()
} }
// incNumNodeUpdates increments the number of node updates processed. // incNumNodeUpdates increments the number of node updates processed.
func (g *routerStats) incNumNodeUpdates() { func (g *builderStats) incNumNodeUpdates() {
g.mu.Lock() g.mu.Lock()
g.numNodes++ g.numNodes++
g.mu.Unlock() g.mu.Unlock()
} }
// Empty returns true if all stats are zero. // Empty returns true if all stats are zero.
func (g *routerStats) Empty() bool { func (g *builderStats) Empty() bool {
g.mu.RLock() g.mu.RLock()
isEmpty := g.numChannels == 0 && isEmpty := g.numChannels == 0 &&
g.numUpdates == 0 && g.numUpdates == 0 &&
@ -48,8 +48,8 @@ func (g *routerStats) Empty() bool {
return isEmpty return isEmpty
} }
// Reset clears any router stats and sets the lastReset field to now. // Reset clears any stats and sets the lastReset field to now.
func (g *routerStats) Reset() { func (g *builderStats) Reset() {
g.mu.Lock() g.mu.Lock()
g.numChannels = 0 g.numChannels = 0
g.numUpdates = 0 g.numUpdates = 0
@ -58,8 +58,8 @@ func (g *routerStats) Reset() {
g.mu.Unlock() g.mu.Unlock()
} }
// String returns a human-readable description of the router stats. // String returns a human-readable description of the stats.
func (g *routerStats) String() string { func (g *builderStats) String() string {
g.mu.RLock() g.mu.RLock()
str := fmt.Sprintf("Processed channels=%d updates=%d nodes=%d in "+ str := fmt.Sprintf("Processed channels=%d updates=%d nodes=%d in "+
"last %v", g.numChannels, g.numUpdates, g.numNodes, "last %v", g.numChannels, g.numUpdates, g.numNodes,

41
netann/node_announcement.go
View File

@ -1,10 +1,14 @@
package netann package netann
import ( import (
"bytes"
"image/color" "image/color"
"net" "net"
"time" "time"
"github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/go-errors/errors"
"github.com/lightningnetwork/lnd/keychain" "github.com/lightningnetwork/lnd/keychain"
"github.com/lightningnetwork/lnd/lnwallet" "github.com/lightningnetwork/lnd/lnwallet"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
@ -76,3 +80,40 @@ func SignNodeAnnouncement(signer lnwallet.MessageSigner,
nodeAnn.Signature, err = lnwire.NewSigFromSignature(sig) nodeAnn.Signature, err = lnwire.NewSigFromSignature(sig)
return err return err
} }
// ValidateNodeAnn validates the node announcement by ensuring that the
// attached signature is needed a signature of the node announcement under the
// specified node public key.
func ValidateNodeAnn(a *lnwire.NodeAnnouncement) error {
// Reconstruct the data of announcement which should be covered by the
// signature so we can verify the signature shortly below
data, err := a.DataToSign()
if err != nil {
return err
}
nodeSig, err := a.Signature.ToSignature()
if err != nil {
return err
}
nodeKey, err := btcec.ParsePubKey(a.NodeID[:])
if err != nil {
return err
}
// Finally ensure that the passed signature is valid, if not we'll
// return an error so this node announcement can be rejected.
dataHash := chainhash.DoubleHashB(data)
if !nodeSig.Verify(dataHash, nodeKey) {
var msgBuf bytes.Buffer
if _, err := lnwire.WriteMessage(&msgBuf, a, 0); err != nil {
return err
}
return errors.Errorf("signature on NodeAnnouncement(%x) is "+
"invalid: %x", nodeKey.SerializeCompressed(),
msgBuf.Bytes())
}
return nil
}