16005100421

Committed 01 Jul 2025 04:35PM UTC coverage: 57.772% (+0.6%) from 57.216%

Build # 16005100421

Build Type

Pull #10018

github

Committed by

web-flow

Commit Message

Merge 0b73fe73c into d8a12a5e5

Pull Request Pull Request #10018: Refactor link's long methods

Run Details

390 of 746 new or added lines in 1 file covered. (52.28%)

41 existing lines in 11 files now uncovered.

98433 of 170383 relevant lines covered (57.77%)

1.79 hits per line

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

73.18

/graph/db/notifications.go

package graphdb

import (
        "errors"
        "fmt"
        "image/color"
        "net"
        "strconv"
        "sync"
        "sync/atomic"

        "github.com/btcsuite/btcd/btcec/v2"
        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/fn/v2"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "github.com/lightningnetwork/lnd/lnutils"
        "github.com/lightningnetwork/lnd/lnwire"
)

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

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

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

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

// newTopologyManager creates a new instance of the topologyManager.
func newTopologyManager() *topologyManager {
        return &topologyManager{
                topologyUpdate:    make(chan any),
                topologyClients:   &lnutils.SyncMap[uint64, *topologyClient]{},
                ntfnClientUpdates: make(chan *topologyClientUpdate),
        }
}

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

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

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

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

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

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

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

        log.Debugf("New graph topology client subscription, client %v",
                clientID)

        ntfnChan := make(chan *TopologyChange, 10)

        select {
        case c.ntfnClientUpdates <- &topologyClientUpdate{
                cancel:   false,
                clientID: clientID,
                ntfnChan: ntfnChan,
        }:
        case <-c.quit:
                return nil, errors.New("ChannelRouter shutting down")
        }

        return &TopologyClient{
                TopologyChanges: ntfnChan,
                Cancel: func() {
                        select {
                        case c.ntfnClientUpdates <- &topologyClientUpdate{
                                cancel:   true,
                                clientID: clientID,
                        }:
                        case <-c.quit:
                                return
                        }
                },
        }, nil
}

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

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

        wg sync.WaitGroup
}

// notifyTopologyChange notifies all registered clients of a new change in
// graph topology in a non-blocking.
//
// NOTE: this should only ever be called from a call-stack originating from the
// handleTopologySubscriptions handler.
func (c *ChannelGraph) notifyTopologyChange(topologyDiff *TopologyChange) {
        // notifyClient is a helper closure that will send topology updates to
        // the given client.
        notifyClient := func(clientID uint64, client *topologyClient) bool {
                client.wg.Add(1)

                log.Tracef("Sending topology notification to client=%v, "+
                        "NodeUpdates=%v, ChannelEdgeUpdates=%v, "+
                        "ClosedChannels=%v", clientID,
                        len(topologyDiff.NodeUpdates),
                        len(topologyDiff.ChannelEdgeUpdates),
                        len(topologyDiff.ClosedChannels))

                go func(t *topologyClient) {
                        defer t.wg.Done()

                        select {

                        // In this case we'll try to send the notification
                        // directly to the upstream client consumer.
                        case t.ntfnChan <- topologyDiff:

                        // If the client cancels the notifications, then we'll
                        // exit early.
                        case <-t.exit:

                        // Similarly, if the ChannelRouter itself exists early,
                        // then we'll also exit ourselves.
                        case <-c.quit:
                        }
                }(client)

                // Always return true here so the following Range will iterate
                // all clients.
                return true
        }

        // Range over the set of active clients, and attempt to send the
        // topology updates.
        c.topologyClients.Range(notifyClient)
}

// handleTopologyUpdate is responsible for sending any topology changes
// notifications to registered clients.
//
// NOTE: must be run inside goroutine and must only ever be called from within
// handleTopologySubscriptions.
func (c *ChannelGraph) handleTopologyUpdate(update any) {
        defer c.wg.Done()

        topChange := &TopologyChange{}
        err := c.addToTopologyChange(topChange, update)
        if err != nil {
                log.Errorf("unable to update topology change notification: %v",
                        err)
                return
        }

        if topChange.isEmpty() {
                return
        }

        c.notifyTopologyChange(topChange)
}

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

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

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

// isEmpty returns true if the TopologyChange is empty. A TopologyChange is
// considered empty, if it contains no *new* updates of any type.
func (t *TopologyChange) isEmpty() bool {
        return len(t.NodeUpdates) == 0 && len(t.ChannelEdgeUpdates) == 0 &&
                len(t.ClosedChannels) == 0
}

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

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

        // ClosedHeight is the height in the chain that the channel was closed
        // at.
        ClosedHeight uint32

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

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

        closeSummaries := make([]*ClosedChanSummary, len(closedChans))
        for i, closedChan := range closedChans {
                closeSummaries[i] = &ClosedChanSummary{
                        ChanID:       closedChan.ChannelID,
                        Capacity:     closedChan.Capacity,
                        ClosedHeight: blockHeight,
                        ChanPoint:    closedChan.ChannelPoint,
                }
        }

        return closeSummaries
}

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

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

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

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

        // Features holds the set of features the node supports.
        Features *lnwire.FeatureVector
}

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

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

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

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

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

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

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

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

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

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

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

        // InboundFee is the fee that must be paid for incoming HTLCs.
        InboundFee fn.Option[lnwire.Fee]

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

// appendTopologyChange appends the passed update message to the passed
// TopologyChange, properly identifying which type of update the message
// constitutes. This function will also fetch any required auxiliary
// information required to create the topology change update from the graph
// database.
func (c *ChannelGraph) addToTopologyChange(update *TopologyChange,
        msg any) error {

        switch m := msg.(type) {

        // Any node announcement maps directly to a NetworkNodeUpdate struct.
        // No further data munging or db queries are required.
        case *models.LightningNode:
                pubKey, err := m.PubKey()
                if err != nil {
                        return err
                }

                nodeUpdate := &NetworkNodeUpdate{
                        Addresses:   m.Addresses,
                        IdentityKey: pubKey,
                        Alias:       m.Alias,
                        Color:       EncodeHexColor(m.Color),
                        Features:    m.Features.Clone(),
                }

                update.NodeUpdates = append(update.NodeUpdates, nodeUpdate)
                return nil

        // We ignore initial channel announcements as we'll only send out
        // updates once the individual edges themselves have been updated.
        case *models.ChannelEdgeInfo:
                return nil

        // Any new ChannelUpdateAnnouncements will generate a corresponding
        // ChannelEdgeUpdate notification.
        case *models.ChannelEdgePolicy:
                // We'll need to fetch the edge's information from the database
                // in order to get the information concerning which nodes are
                // being connected.
                edgeInfo, _, _, err := c.FetchChannelEdgesByID(m.ChannelID)
                if err != nil {
                        return fmt.Errorf("unable fetch channel edge: %w", err)
                }

                // If the flag is one, then the advertising node is actually
                // the second node.
                sourceNode := edgeInfo.NodeKey1
                connectingNode := edgeInfo.NodeKey2
                if m.ChannelFlags&lnwire.ChanUpdateDirection == 1 {
                        sourceNode = edgeInfo.NodeKey2
                        connectingNode = edgeInfo.NodeKey1
                }

                aNode, err := sourceNode()
                if err != nil {
                        return err
                }
                cNode, err := connectingNode()
                if err != nil {
                        return err
                }

                edgeUpdate := &ChannelEdgeUpdate{
                        ChanID:          m.ChannelID,
                        ChanPoint:       edgeInfo.ChannelPoint,
                        TimeLockDelta:   m.TimeLockDelta,
                        Capacity:        edgeInfo.Capacity,
                        MinHTLC:         m.MinHTLC,
                        MaxHTLC:         m.MaxHTLC,
                        BaseFee:         m.FeeBaseMSat,
                        FeeRate:         m.FeeProportionalMillionths,
                        AdvertisingNode: aNode,
                        ConnectingNode:  cNode,
                        Disabled:        m.ChannelFlags&lnwire.ChanUpdateDisabled != 0,
                        InboundFee:      m.InboundFee,
                        ExtraOpaqueData: m.ExtraOpaqueData,
                }

                // TODO(roasbeef): add bit to toggle
                update.ChannelEdgeUpdates = append(update.ChannelEdgeUpdates,
                        edgeUpdate)
                return nil

        case []*ClosedChanSummary:
                update.ClosedChannels = append(update.ClosedChannels, m...)
                return nil

        default:
                return fmt.Errorf("unable to add to topology change, "+
                        "unknown message type %T", msg)
        }
}

// EncodeHexColor takes a color and returns it in hex code format.
func EncodeHexColor(color color.RGBA) string {
        return fmt.Sprintf("#%02x%02x%02x", color.R, color.G, color.B)
}

// DecodeHexColor takes a hex color string like "#rrggbb" and returns a
// color.RGBA.
func DecodeHexColor(hex string) (color.RGBA, error) {
        if len(hex) != 7 || hex[0] != '#' {
                return color.RGBA{}, fmt.Errorf("invalid hex color string: %s",
                        hex)
        }

        r, err := strconv.ParseUint(hex[1:3], 16, 8)
        if err != nil {
                return color.RGBA{}, fmt.Errorf("invalid red component: %w",
                        err)
        }

        g, err := strconv.ParseUint(hex[3:5], 16, 8)
        if err != nil {
                return color.RGBA{}, fmt.Errorf("invalid green component: %w",
                        err)
        }

        b, err := strconv.ParseUint(hex[5:7], 16, 8)
        if err != nil {
                return color.RGBA{}, fmt.Errorf("invalid blue component: %w",
                        err)
        }

        return color.RGBA{
                R: uint8(r),
                G: uint8(g),
                B: uint8(b),
        }, nil
}

1	package graphdb
2
3	import (
4	"errors"
5	"fmt"
6	"image/color"
7	"net"
8	"strconv"
9	"sync"
10	"sync/atomic"
11
12	"github.com/btcsuite/btcd/btcec/v2"
13	"github.com/btcsuite/btcd/btcutil"
14	"github.com/btcsuite/btcd/wire"
15	"github.com/lightningnetwork/lnd/fn/v2"
16	"github.com/lightningnetwork/lnd/graph/db/models"
17	"github.com/lightningnetwork/lnd/lnutils"
18	"github.com/lightningnetwork/lnd/lnwire"
19	)
20
21	// topologyManager holds all the fields required to manage the network topology
22	// subscriptions and notifications.
23	type topologyManager struct {
24	// ntfnClientCounter is an atomic counter that's used to assign unique
25	// notification client IDs to new clients.
26	ntfnClientCounter atomic.Uint64
27
28	// topologyUpdate is a channel that carries new topology updates
29	// messages from outside the ChannelGraph to be processed by the
30	// networkHandler.
31	topologyUpdate chan any
32
33	// topologyClients maps a client's unique notification ID to a
34	// topologyClient client that contains its notification dispatch
35	// channel.
36	topologyClients lnutils.SyncMap[uint64, topologyClient]
37
38	// ntfnClientUpdates is a channel that's used to send new updates to
39	// topology notification clients to the ChannelGraph. Updates either
40	// add a new notification client, or cancel notifications for an
41	// existing client.
42	ntfnClientUpdates chan *topologyClientUpdate
43	}
44
45	// newTopologyManager creates a new instance of the topologyManager.
46	func newTopologyManager() *topologyManager {	3✔
47	return &topologyManager{	3✔
48	topologyUpdate: make(chan any),	3✔
49	topologyClients: &lnutils.SyncMap[uint64, *topologyClient]{},	3✔
50	ntfnClientUpdates: make(chan *topologyClientUpdate),	3✔
51	}	3✔
52	}	3✔
53
54	// TopologyClient represents an intent to receive notifications from the
55	// channel router regarding changes to the topology of the channel graph. The
56	// TopologyChanges channel will be sent upon with new updates to the channel
57	// graph in real-time as they're encountered.
58	type TopologyClient struct {
59	// TopologyChanges is a receive only channel that new channel graph
60	// updates will be sent over.
61	//
62	// TODO(roasbeef): chan for each update type instead?
63	TopologyChanges <-chan *TopologyChange
64
65	// Cancel is a function closure that should be executed when the client
66	// wishes to cancel their notification intent. Doing so allows the
67	// ChannelRouter to free up resources.
68	Cancel func()
69	}
70
71	// topologyClientUpdate is a message sent to the channel router to either
72	// register a new topology client or re-register an existing client.
73	type topologyClientUpdate struct {
74	// cancel indicates if the update to the client is cancelling an
75	// existing client's notifications. If not then this update will be to
76	// register a new set of notifications.
77	cancel bool
78
79	// clientID is the unique identifier for this client. Any further
80	// updates (deleting or adding) to this notification client will be
81	// dispatched according to the target clientID.
82	clientID uint64
83
84	// ntfnChan is a send-only channel in which notifications should be
85	// sent over from router -> client.
86	ntfnChan chan<- *TopologyChange
87	}
88
89	// SubscribeTopology returns a new topology client which can be used by the
90	// caller to receive notifications whenever a change in the channel graph
91	// topology occurs. Changes that will be sent at notifications include: new
92	// nodes appearing, node updating their attributes, new channels, channels
93	// closing, and updates in the routing policies of a channel's directed edges.
94	func (c ChannelGraph) SubscribeTopology() (TopologyClient, error) {	3✔
95	// If the router is not yet started, return an error to avoid a	3✔
96	// deadlock waiting for it to handle the subscription request.	3✔
97	if !c.started.Load() {	3✔
98	return nil, fmt.Errorf("router not started")	×
99	}	×
100
101	// We'll first atomically obtain the next ID for this client from the
102	// incrementing client ID counter.
103	clientID := c.ntfnClientCounter.Add(1)	3✔
104		3✔
105	log.Debugf("New graph topology client subscription, client %v",	3✔
106	clientID)	3✔
107		3✔
108	ntfnChan := make(chan *TopologyChange, 10)	3✔
109		3✔
110	select {	3✔
111	case c.ntfnClientUpdates <- &topologyClientUpdate{
112	cancel: false,
113	clientID: clientID,
114	ntfnChan: ntfnChan,
115	}:	3✔
116	case <-c.quit:	×
117	return nil, errors.New("ChannelRouter shutting down")	×
118	}
119
120	return &TopologyClient{	3✔
121	TopologyChanges: ntfnChan,	3✔
122	Cancel: func() {	6✔
123	select {	3✔
124	case c.ntfnClientUpdates <- &topologyClientUpdate{
125	cancel: true,
126	clientID: clientID,
127	}:	3✔
UNCOV 128	case <-c.quit:	×
UNCOV 129	return	×
130	}
131	},
132	}, nil
133	}
134
135	// topologyClient is a data-structure use by the channel router to couple the
136	// client's notification channel along with a special "exit" channel that can
137	// be used to cancel all lingering goroutines blocked on a send to the
138	// notification channel.
139	type topologyClient struct {
140	// ntfnChan is a send-only channel that's used to propagate
141	// notification s from the channel router to an instance of a
142	// topologyClient client.
143	ntfnChan chan<- *TopologyChange
144
145	// exit is a channel that is used internally by the channel router to
146	// cancel any active un-consumed goroutine notifications.
147	exit chan struct{}
148
149	wg sync.WaitGroup
150	}
151
152	// notifyTopologyChange notifies all registered clients of a new change in
153	// graph topology in a non-blocking.
154	//
155	// NOTE: this should only ever be called from a call-stack originating from the
156	// handleTopologySubscriptions handler.
157	func (c ChannelGraph) notifyTopologyChange(topologyDiff TopologyChange) {	3✔
158	// notifyClient is a helper closure that will send topology updates to	3✔
159	// the given client.	3✔
160	notifyClient := func(clientID uint64, client *topologyClient) bool {	6✔
161	client.wg.Add(1)	3✔
162		3✔
163	log.Tracef("Sending topology notification to client=%v, "+	3✔
164	"NodeUpdates=%v, ChannelEdgeUpdates=%v, "+	3✔
165	"ClosedChannels=%v", clientID,	3✔
166	len(topologyDiff.NodeUpdates),	3✔
167	len(topologyDiff.ChannelEdgeUpdates),	3✔
168	len(topologyDiff.ClosedChannels))	3✔
169		3✔
170	go func(t *topologyClient) {	6✔
171	defer t.wg.Done()	3✔
172		3✔
173	select {	3✔
174
175	// In this case we'll try to send the notification
176	// directly to the upstream client consumer.
177	case t.ntfnChan <- topologyDiff:	3✔
178
179	// If the client cancels the notifications, then we'll
180	// exit early.
181	case <-t.exit:	×
182
183	// Similarly, if the ChannelRouter itself exists early,
184	// then we'll also exit ourselves.
185	case <-c.quit:	×
186	}
187	}(client)
188
189	// Always return true here so the following Range will iterate
190	// all clients.
191	return true	3✔
192	}
193
194	// Range over the set of active clients, and attempt to send the
195	// topology updates.
196	c.topologyClients.Range(notifyClient)	3✔
197	}
198
199	// handleTopologyUpdate is responsible for sending any topology changes
200	// notifications to registered clients.
201	//
202	// NOTE: must be run inside goroutine and must only ever be called from within
203	// handleTopologySubscriptions.
204	func (c *ChannelGraph) handleTopologyUpdate(update any) {	3✔
205	defer c.wg.Done()	3✔
206		3✔
207	topChange := &TopologyChange{}	3✔
208	err := c.addToTopologyChange(topChange, update)	3✔
209	if err != nil {	3✔
210	log.Errorf("unable to update topology change notification: %v",	×
211	err)	×
212	return	×
213	}	×
214
215	if topChange.isEmpty() {	6✔
216	return	3✔
217	}	3✔
218
219	c.notifyTopologyChange(topChange)	3✔
220	}
221
222	// TopologyChange represents a new set of modifications to the channel graph.
223	// Topology changes will be dispatched in real-time as the ChannelGraph
224	// validates and process modifications to the authenticated channel graph.
225	type TopologyChange struct {
226	// NodeUpdates is a slice of nodes which are either new to the channel
227	// graph, or have had their attributes updated in an authenticated
228	// manner.
229	NodeUpdates []*NetworkNodeUpdate
230
231	// ChanelEdgeUpdates is a slice of channel edges which are either newly
232	// opened and authenticated, or have had their routing policies
233	// updated.
234	ChannelEdgeUpdates []*ChannelEdgeUpdate
235
236	// ClosedChannels contains a slice of close channel summaries which
237	// described which block a channel was closed at, and also carry
238	// supplemental information such as the capacity of the former channel.
239	ClosedChannels []*ClosedChanSummary
240	}
241
242	// isEmpty returns true if the TopologyChange is empty. A TopologyChange is
243	// considered empty, if it contains no new updates of any type.
244	func (t *TopologyChange) isEmpty() bool {	3✔
245	return len(t.NodeUpdates) == 0 && len(t.ChannelEdgeUpdates) == 0 &&	3✔
246	len(t.ClosedChannels) == 0	3✔
247	}	3✔
248
249	// ClosedChanSummary is a summary of a channel that was detected as being
250	// closed by monitoring the blockchain. Once a channel's funding point has been
251	// spent, the channel will automatically be marked as closed by the
252	// ChainNotifier.
253	//
254	// TODO(roasbeef): add nodes involved?
255	type ClosedChanSummary struct {
256	// ChanID is the short-channel ID which uniquely identifies the
257	// channel.
258	ChanID uint64
259
260	// Capacity was the total capacity of the channel before it was closed.
261	Capacity btcutil.Amount
262
263	// ClosedHeight is the height in the chain that the channel was closed
264	// at.
265	ClosedHeight uint32
266
267	// ChanPoint is the funding point, or the multi-sig utxo which
268	// previously represented the channel.
269	ChanPoint wire.OutPoint
270	}
271
272	// createCloseSummaries takes in a slice of channels closed at the target block
273	// height and creates a slice of summaries which of each channel closure.
274	func createCloseSummaries(blockHeight uint32,
275	closedChans ...models.ChannelEdgeInfo) []ClosedChanSummary {	3✔
276		3✔
277	closeSummaries := make([]*ClosedChanSummary, len(closedChans))	3✔
278	for i, closedChan := range closedChans {	6✔
279	closeSummaries[i] = &ClosedChanSummary{	3✔
280	ChanID: closedChan.ChannelID,	3✔
281	Capacity: closedChan.Capacity,	3✔
282	ClosedHeight: blockHeight,	3✔
283	ChanPoint: closedChan.ChannelPoint,	3✔
284	}	3✔
285	}	3✔
286
287	return closeSummaries	3✔
288	}
289
290	// NetworkNodeUpdate is an update for a node within the Lightning Network. A
291	// NetworkNodeUpdate is sent out either when a new node joins the network, or a
292	// node broadcasts a new update with a newer time stamp that supersedes its
293	// old update. All updates are properly authenticated.
294	type NetworkNodeUpdate struct {
295	// Addresses is a slice of all the node's known addresses.
296	Addresses []net.Addr
297
298	// IdentityKey is the identity public key of the target node. This is
299	// used to encrypt onion blobs as well as to authenticate any new
300	// updates.
301	IdentityKey *btcec.PublicKey
302
303	// Alias is the alias or nick name of the node.
304	Alias string
305
306	// Color is the node's color in hex code format.
307	Color string
308
309	// Features holds the set of features the node supports.
310	Features *lnwire.FeatureVector
311	}
312
313	// ChannelEdgeUpdate is an update for a new channel within the ChannelGraph.
314	// This update is sent out once a new authenticated channel edge is discovered
315	// within the network. These updates are directional, so if a channel is fully
316	// public, then there will be two updates sent out: one for each direction
317	// within the channel. Each update will carry that particular routing edge
318	// policy for the channel direction.
319	//
320	// An edge is a channel in the direction of AdvertisingNode -> ConnectingNode.
321	type ChannelEdgeUpdate struct {
322	// ChanID is the unique short channel ID for the channel. This encodes
323	// where in the blockchain the channel's funding transaction was
324	// originally confirmed.
325	ChanID uint64
326
327	// ChanPoint is the outpoint which represents the multi-sig funding
328	// output for the channel.
329	ChanPoint wire.OutPoint
330
331	// Capacity is the capacity of the newly created channel.
332	Capacity btcutil.Amount
333
334	// MinHTLC is the minimum HTLC amount that this channel will forward.
335	MinHTLC lnwire.MilliSatoshi
336
337	// MaxHTLC is the maximum HTLC amount that this channel will forward.
338	MaxHTLC lnwire.MilliSatoshi
339
340	// BaseFee is the base fee that will charged for all HTLC's forwarded
341	// across the this channel direction.
342	BaseFee lnwire.MilliSatoshi
343
344	// FeeRate is the fee rate that will be shared for all HTLC's forwarded
345	// across this channel direction.
346	FeeRate lnwire.MilliSatoshi
347
348	// TimeLockDelta is the time-lock expressed in blocks that will be
349	// added to outgoing HTLC's from incoming HTLC's. This value is the
350	// difference of the incoming and outgoing HTLC's time-locks routed
351	// through this hop.
352	TimeLockDelta uint16
353
354	// AdvertisingNode is the node that's advertising this edge.
355	AdvertisingNode *btcec.PublicKey
356
357	// ConnectingNode is the node that the advertising node connects to.
358	ConnectingNode *btcec.PublicKey
359
360	// Disabled, if true, signals that the channel is unavailable to relay
361	// payments.
362	Disabled bool
363
364	// InboundFee is the fee that must be paid for incoming HTLCs.
365	InboundFee fn.Option[lnwire.Fee]
366
367	// ExtraOpaqueData is the set of data that was appended to this message
368	// to fill out the full maximum transport message size. These fields can
369	// be used to specify optional data such as custom TLV fields.
370	ExtraOpaqueData lnwire.ExtraOpaqueData
371	}
372
373	// appendTopologyChange appends the passed update message to the passed
374	// TopologyChange, properly identifying which type of update the message
375	// constitutes. This function will also fetch any required auxiliary
376	// information required to create the topology change update from the graph
377	// database.
378	func (c ChannelGraph) addToTopologyChange(update TopologyChange,
379	msg any) error {	3✔
380		3✔
381	switch m := msg.(type) {	3✔
382
383	// Any node announcement maps directly to a NetworkNodeUpdate struct.
384	// No further data munging or db queries are required.
385	case *models.LightningNode:	3✔
386	pubKey, err := m.PubKey()	3✔
387	if err != nil {	3✔
388	return err	×
389	}	×
390
391	nodeUpdate := &NetworkNodeUpdate{	3✔
392	Addresses: m.Addresses,	3✔
393	IdentityKey: pubKey,	3✔
394	Alias: m.Alias,	3✔
395	Color: EncodeHexColor(m.Color),	3✔
396	Features: m.Features.Clone(),	3✔
397	}	3✔
398		3✔
399	update.NodeUpdates = append(update.NodeUpdates, nodeUpdate)	3✔
400	return nil	3✔
401
402	// We ignore initial channel announcements as we'll only send out
403	// updates once the individual edges themselves have been updated.
404	case *models.ChannelEdgeInfo:	3✔
405	return nil	3✔
406
407	// Any new ChannelUpdateAnnouncements will generate a corresponding
408	// ChannelEdgeUpdate notification.
409	case *models.ChannelEdgePolicy:	3✔
410	// We'll need to fetch the edge's information from the database	3✔
411	// in order to get the information concerning which nodes are	3✔
412	// being connected.	3✔
413	edgeInfo, _, _, err := c.FetchChannelEdgesByID(m.ChannelID)	3✔
414	if err != nil {	3✔
415	return fmt.Errorf("unable fetch channel edge: %w", err)	×
416	}	×
417
418	// If the flag is one, then the advertising node is actually
419	// the second node.
420	sourceNode := edgeInfo.NodeKey1	3✔
421	connectingNode := edgeInfo.NodeKey2	3✔
422	if m.ChannelFlags&lnwire.ChanUpdateDirection == 1 {	6✔
423	sourceNode = edgeInfo.NodeKey2	3✔
424	connectingNode = edgeInfo.NodeKey1	3✔
425	}	3✔
426
427	aNode, err := sourceNode()	3✔
428	if err != nil {	3✔
429	return err	×
430	}	×
431	cNode, err := connectingNode()	3✔
432	if err != nil {	3✔
433	return err	×
434	}	×
435
436	edgeUpdate := &ChannelEdgeUpdate{	3✔
437	ChanID: m.ChannelID,	3✔
438	ChanPoint: edgeInfo.ChannelPoint,	3✔
439	TimeLockDelta: m.TimeLockDelta,	3✔
440	Capacity: edgeInfo.Capacity,	3✔
441	MinHTLC: m.MinHTLC,	3✔
442	MaxHTLC: m.MaxHTLC,	3✔
443	BaseFee: m.FeeBaseMSat,	3✔
444	FeeRate: m.FeeProportionalMillionths,	3✔
445	AdvertisingNode: aNode,	3✔
446	ConnectingNode: cNode,	3✔
447	Disabled: m.ChannelFlags&lnwire.ChanUpdateDisabled != 0,	3✔
448	InboundFee: m.InboundFee,	3✔
449	ExtraOpaqueData: m.ExtraOpaqueData,	3✔
450	}	3✔
451		3✔
452	// TODO(roasbeef): add bit to toggle	3✔
453	update.ChannelEdgeUpdates = append(update.ChannelEdgeUpdates,	3✔
454	edgeUpdate)	3✔
455	return nil	3✔
456
457	case []*ClosedChanSummary:	3✔
458	update.ClosedChannels = append(update.ClosedChannels, m...)	3✔
459	return nil	3✔
460
461	default:	×
462	return fmt.Errorf("unable to add to topology change, "+	×
463	"unknown message type %T", msg)	×
464	}
465	}
466
467	// EncodeHexColor takes a color and returns it in hex code format.
468	func EncodeHexColor(color color.RGBA) string {	3✔
469	return fmt.Sprintf("#%02x%02x%02x", color.R, color.G, color.B)	3✔
470	}	3✔
471
472	// DecodeHexColor takes a hex color string like "#rrggbb" and returns a
473	// color.RGBA.
474	func DecodeHexColor(hex string) (color.RGBA, error) {	×
475	if len(hex) != 7 \|\| hex[0] != '#' {	×
476	return color.RGBA{}, fmt.Errorf("invalid hex color string: %s",	×
477	hex)	×
478	}	×
479
480	r, err := strconv.ParseUint(hex[1:3], 16, 8)	×
481	if err != nil {	×
482	return color.RGBA{}, fmt.Errorf("invalid red component: %w",	×
483	err)	×
484	}	×
485
486	g, err := strconv.ParseUint(hex[3:5], 16, 8)	×
487	if err != nil {	×
488	return color.RGBA{}, fmt.Errorf("invalid green component: %w",	×
489	err)	×
490	}	×
491
492	b, err := strconv.ParseUint(hex[5:7], 16, 8)	×
493	if err != nil {	×
494	return color.RGBA{}, fmt.Errorf("invalid blue component: %w",	×
495	err)	×
496	}	×
497
498	return color.RGBA{	×
499	R: uint8(r),	×
500	G: uint8(g),	×
501	B: uint8(b),	×
502	}, nil	×
503	}

lightningnetwork / lnd / 16005100421

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

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