13522725226

Committed 25 Feb 2025 01:46PM UTC coverage: 58.83% (+0.02%) from 58.815%

Build # 13522725226

Build Type

Pull #9551

github

Committed by

ellemouton

Commit Message

graph/db: move cache writes for Prune methods

This commit moves the cache writes for PruneGraphNodes and PruneGraph
from the KVStore to the ChannelGraph.

Pull Request Pull Request #9551: graph: extract cache from CRUD [4]

Run Details

100 of 116 new or added lines in 2 files covered. (86.21%)

707 existing lines in 14 files now uncovered.

136401 of 231858 relevant lines covered (58.83%)

19263.57 hits per line

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

88.71

/graph/db/graph.go

package graphdb

import (
        "sync"
        "time"

        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/routing/route"
)

// Config is a struct that holds all the necessary dependencies for a
// ChannelGraph.
type Config struct {
        // KVDB is the kvdb.Backend that will be used for initializing the
        // KVStore CRUD layer.
        KVDB kvdb.Backend

        // KVStoreOpts is a list of functional options that will be used when
        // initializing the KVStore.
        KVStoreOpts []KVStoreOptionModifier
}

// ChannelGraph is a layer above the graph's CRUD layer.
//
// NOTE: currently, this is purely a pass-through layer directly to the backing
// KVStore. Upcoming commits will move the graph cache out of the KVStore and
// into this layer so that the KVStore is only responsible for CRUD operations.
type ChannelGraph struct {
        // cacheMu guards any writes to the graphCache. It should be held
        // across the DB write call and the graphCache update to make the
        // two updates as atomic as possible.
        cacheMu    sync.Mutex
        graphCache *GraphCache

        *KVStore
}

// NewChannelGraph creates a new ChannelGraph instance with the given backend.
func NewChannelGraph(cfg *Config, options ...ChanGraphOption) (*ChannelGraph,
        error) {

        opts := defaultChanGraphOptions()
        for _, o := range options {
                o(opts)
        }

        store, err := NewKVStore(cfg.KVDB, cfg.KVStoreOpts...)
        if err != nil {
                return nil, err
        }

        if !opts.useGraphCache {
                return &ChannelGraph{
                        KVStore: store,
                }, nil
        }

        // The graph cache can be turned off (e.g. for mobile users) for a
        // speed/memory usage tradeoff.
        graphCache := NewGraphCache(opts.preAllocCacheNumNodes)
        startTime := time.Now()
        log.Debugf("Populating in-memory channel graph, this might take a " +
                "while...")

        err = store.ForEachNodeCacheable(func(node route.Vertex,
                features *lnwire.FeatureVector) error {

                graphCache.AddNodeFeatures(node, features)

                return nil
        })
        if err != nil {
                return nil, err
        }

        err = store.ForEachChannel(func(info *models.ChannelEdgeInfo,
                policy1, policy2 *models.ChannelEdgePolicy) error {

                graphCache.AddChannel(info, policy1, policy2)

                return nil
        })
        if err != nil {
                return nil, err
        }

        log.Debugf("Finished populating in-memory channel graph (took %v, %s)",
                time.Since(startTime), graphCache.Stats())

        store.setGraphCache(graphCache)

        return &ChannelGraph{
                KVStore:    store,
                graphCache: graphCache,
        }, nil
}

// DeleteChannelEdges removes edges with the given channel IDs from the
// database and marks them as zombies. This ensures that we're unable to re-add
// it to our database once again. If an edge does not exist within the
// database, then ErrEdgeNotFound will be returned. If strictZombiePruning is
// true, then when we mark these edges as zombies, we'll set up the keys such
// that we require the node that failed to send the fresh update to be the one
// that resurrects the channel from its zombie state. The markZombie bool
// denotes whether to mark the channel as a zombie.
func (c *ChannelGraph) DeleteChannelEdges(strictZombiePruning, markZombie bool,
        chanIDs ...uint64) error {

        c.cacheMu.Lock()
        defer c.cacheMu.Unlock()

        infos, err := c.KVStore.DeleteChannelEdges(
                strictZombiePruning, markZombie, chanIDs...,
        )
        if err != nil {
                return err
        }

        if c.graphCache != nil {
                for _, info := range infos {
                        c.graphCache.RemoveChannel(
                                info.NodeKey1Bytes, info.NodeKey2Bytes,
                                info.ChannelID,
                        )
                }
        }

        return err
}

// DisconnectBlockAtHeight is used to indicate that the block specified
// by the passed height has been disconnected from the main chain. This
// will "rewind" the graph back to the height below, deleting channels
// that are no longer confirmed from the graph. The prune log will be
// set to the last prune height valid for the remaining chain.
// Channels that were removed from the graph resulting from the
// disconnected block are returned.
func (c *ChannelGraph) DisconnectBlockAtHeight(height uint32) (
        []*models.ChannelEdgeInfo, error) {

        c.cacheMu.Lock()
        defer c.cacheMu.Unlock()

        edges, err := c.KVStore.DisconnectBlockAtHeight(height)
        if err != nil {
                return nil, err
        }

        if c.graphCache != nil {
                for _, edge := range edges {
                        c.graphCache.RemoveChannel(
                                edge.NodeKey1Bytes, edge.NodeKey2Bytes,
                                edge.ChannelID,
                        )
                }
        }

        return edges, nil
}

// PruneGraph prunes newly closed channels from the channel graph in response
// to a new block being solved on the network. Any transactions which spend the
// funding output of any known channels within he graph will be deleted.
// Additionally, the "prune tip", or the last block which has been used to
// prune the graph is stored so callers can ensure the graph is fully in sync
// with the current UTXO state. A slice of channels that have been closed by
// the target block are returned if the function succeeds without error.
func (c *ChannelGraph) PruneGraph(spentOutputs []*wire.OutPoint,
        blockHash *chainhash.Hash, blockHeight uint32) (
        []*models.ChannelEdgeInfo, error) {

        c.cacheMu.Lock()
        defer c.cacheMu.Unlock()

        edges, err := c.KVStore.PruneGraph(spentOutputs, blockHash, blockHeight)
        if err != nil {
                return nil, err
        }

        // Now that the graph has been pruned, we'll also attempt to
        // prune any nodes that have had a channel closed within the
        // latest block.
        nodes, err := c.KVStore.PruneGraphNodes()
        if err != nil {
                return nil, err
        }

        if c.graphCache != nil {
                for _, edge := range edges {
                        c.graphCache.RemoveChannel(
                                edge.NodeKey1Bytes, edge.NodeKey2Bytes,
                                edge.ChannelID,
                        )
                }

                for _, node := range nodes {
                        c.graphCache.RemoveNode(node)
                }

                log.Debugf("Pruned graph, cache now has %s",
                        c.graphCache.Stats())
        }

        return edges, nil
}

// PruneGraphNodes is a garbage collection method which attempts to prune out
// any nodes from the channel graph that are currently unconnected. This ensure
// that we only maintain a graph of reachable nodes. In the event that a pruned
// node gains more channels, it will be re-added back to the graph.
func (c *ChannelGraph) PruneGraphNodes() error {
        c.cacheMu.Lock()
        defer c.cacheMu.Unlock()

        nodes, err := c.KVStore.PruneGraphNodes()
        if err != nil {
                return err
        }

        if c.graphCache != nil {
                for _, node := range nodes {
                        c.graphCache.RemoveNode(node)
                }
        }

        return nil
}

1	package graphdb
2
3	import (
4	"sync"
5	"time"
6
7	"github.com/btcsuite/btcd/chaincfg/chainhash"
8	"github.com/btcsuite/btcd/wire"
9	"github.com/lightningnetwork/lnd/graph/db/models"
10	"github.com/lightningnetwork/lnd/kvdb"
11	"github.com/lightningnetwork/lnd/lnwire"
12	"github.com/lightningnetwork/lnd/routing/route"
13	)
14
15	// Config is a struct that holds all the necessary dependencies for a
16	// ChannelGraph.
17	type Config struct {
18	// KVDB is the kvdb.Backend that will be used for initializing the
19	// KVStore CRUD layer.
20	KVDB kvdb.Backend
21
22	// KVStoreOpts is a list of functional options that will be used when
23	// initializing the KVStore.
24	KVStoreOpts []KVStoreOptionModifier
25	}
26
27	// ChannelGraph is a layer above the graph's CRUD layer.
28	//
29	// NOTE: currently, this is purely a pass-through layer directly to the backing
30	// KVStore. Upcoming commits will move the graph cache out of the KVStore and
31	// into this layer so that the KVStore is only responsible for CRUD operations.
32	type ChannelGraph struct {
33	// cacheMu guards any writes to the graphCache. It should be held
34	// across the DB write call and the graphCache update to make the
35	// two updates as atomic as possible.
36	cacheMu sync.Mutex
37	graphCache *GraphCache
38
39	*KVStore
40	}
41
42	// NewChannelGraph creates a new ChannelGraph instance with the given backend.
43	func NewChannelGraph(cfg Config, options ...ChanGraphOption) (ChannelGraph,
44	error) {	176✔
45		176✔
46	opts := defaultChanGraphOptions()	176✔
47	for _, o := range options {	281✔
48	o(opts)	105✔
49	}	105✔
50
51	store, err := NewKVStore(cfg.KVDB, cfg.KVStoreOpts...)	176✔
52	if err != nil {	176✔
53	return nil, err	×
54	}	×
55
56	if !opts.useGraphCache {	212✔
57	return &ChannelGraph{	36✔
58	KVStore: store,	36✔
59	}, nil	36✔
60	}	36✔
61
62	// The graph cache can be turned off (e.g. for mobile users) for a
63	// speed/memory usage tradeoff.
64	graphCache := NewGraphCache(opts.preAllocCacheNumNodes)	143✔
65	startTime := time.Now()	143✔
66	log.Debugf("Populating in-memory channel graph, this might take a " +	143✔
67	"while...")	143✔
68		143✔
69	err = store.ForEachNodeCacheable(func(node route.Vertex,	143✔
70	features *lnwire.FeatureVector) error {	246✔
71		103✔
72	graphCache.AddNodeFeatures(node, features)	103✔
73		103✔
74	return nil	103✔
75	})	103✔
76	if err != nil {	143✔
77	return nil, err	×
78	}	×
79
80	err = store.ForEachChannel(func(info *models.ChannelEdgeInfo,	143✔
81	policy1, policy2 *models.ChannelEdgePolicy) error {	542✔
82		399✔
83	graphCache.AddChannel(info, policy1, policy2)	399✔
84		399✔
85	return nil	399✔
86	})	399✔
87	if err != nil {	143✔
88	return nil, err	×
89	}	×
90
91	log.Debugf("Finished populating in-memory channel graph (took %v, %s)",	143✔
92	time.Since(startTime), graphCache.Stats())	143✔
93		143✔
94	store.setGraphCache(graphCache)	143✔
95		143✔
96	return &ChannelGraph{	143✔
97	KVStore: store,	143✔
98	graphCache: graphCache,	143✔
99	}, nil	143✔
100	}
101
102	// DeleteChannelEdges removes edges with the given channel IDs from the
103	// database and marks them as zombies. This ensures that we're unable to re-add
104	// it to our database once again. If an edge does not exist within the
105	// database, then ErrEdgeNotFound will be returned. If strictZombiePruning is
106	// true, then when we mark these edges as zombies, we'll set up the keys such
107	// that we require the node that failed to send the fresh update to be the one
108	// that resurrects the channel from its zombie state. The markZombie bool
109	// denotes whether to mark the channel as a zombie.
110	func (c *ChannelGraph) DeleteChannelEdges(strictZombiePruning, markZombie bool,
111	chanIDs ...uint64) error {	137✔
112		137✔
113	c.cacheMu.Lock()	137✔
114	defer c.cacheMu.Unlock()	137✔
115		137✔
116	infos, err := c.KVStore.DeleteChannelEdges(	137✔
117	strictZombiePruning, markZombie, chanIDs...,	137✔
118	)	137✔
119	if err != nil {	197✔
120	return err	60✔
121	}	60✔
122
123	if c.graphCache != nil {	154✔
124	for _, info := range infos {	104✔
125	c.graphCache.RemoveChannel(	27✔
126	info.NodeKey1Bytes, info.NodeKey2Bytes,	27✔
127	info.ChannelID,	27✔
128	)	27✔
129	}	27✔
130	}
131
132	return err	77✔
133	}
134
135	// DisconnectBlockAtHeight is used to indicate that the block specified
136	// by the passed height has been disconnected from the main chain. This
137	// will "rewind" the graph back to the height below, deleting channels
138	// that are no longer confirmed from the graph. The prune log will be
139	// set to the last prune height valid for the remaining chain.
140	// Channels that were removed from the graph resulting from the
141	// disconnected block are returned.
142	func (c *ChannelGraph) DisconnectBlockAtHeight(height uint32) (
143	[]*models.ChannelEdgeInfo, error) {	169✔
144		169✔
145	c.cacheMu.Lock()	169✔
146	defer c.cacheMu.Unlock()	169✔
147		169✔
148	edges, err := c.KVStore.DisconnectBlockAtHeight(height)	169✔
149	if err != nil {	169✔
NEW 150	return nil, err	×
NEW 151	}	×
152
153	if c.graphCache != nil {	338✔
154	for _, edge := range edges {	273✔
155	c.graphCache.RemoveChannel(	104✔
156	edge.NodeKey1Bytes, edge.NodeKey2Bytes,	104✔
157	edge.ChannelID,	104✔
158	)	104✔
159	}	104✔
160	}
161
162	return edges, nil	169✔
163	}
164
165	// PruneGraph prunes newly closed channels from the channel graph in response
166	// to a new block being solved on the network. Any transactions which spend the
167	// funding output of any known channels within he graph will be deleted.
168	// Additionally, the "prune tip", or the last block which has been used to
169	// prune the graph is stored so callers can ensure the graph is fully in sync
170	// with the current UTXO state. A slice of channels that have been closed by
171	// the target block are returned if the function succeeds without error.
172	func (c ChannelGraph) PruneGraph(spentOutputs []wire.OutPoint,
173	blockHash *chainhash.Hash, blockHeight uint32) (
174	[]*models.ChannelEdgeInfo, error) {	245✔
175		245✔
176	c.cacheMu.Lock()	245✔
177	defer c.cacheMu.Unlock()	245✔
178		245✔
179	edges, err := c.KVStore.PruneGraph(spentOutputs, blockHash, blockHeight)	245✔
180	if err != nil {	245✔
NEW 181	return nil, err	×
NEW 182	}	×
183
184	// Now that the graph has been pruned, we'll also attempt to
185	// prune any nodes that have had a channel closed within the
186	// latest block.
187	nodes, err := c.KVStore.PruneGraphNodes()	245✔
188	if err != nil {	245✔
NEW 189	return nil, err	×
NEW 190	}	×
191
192	if c.graphCache != nil {	490✔
193	for _, edge := range edges {	265✔
194	c.graphCache.RemoveChannel(	20✔
195	edge.NodeKey1Bytes, edge.NodeKey2Bytes,	20✔
196	edge.ChannelID,	20✔
197	)	20✔
198	}	20✔
199
200	for _, node := range nodes {	300✔
201	c.graphCache.RemoveNode(node)	55✔
202	}	55✔
203
204	log.Debugf("Pruned graph, cache now has %s",	245✔
205	c.graphCache.Stats())	245✔
206	}
207
208	return edges, nil	245✔
209	}
210
211	// PruneGraphNodes is a garbage collection method which attempts to prune out
212	// any nodes from the channel graph that are currently unconnected. This ensure
213	// that we only maintain a graph of reachable nodes. In the event that a pruned
214	// node gains more channels, it will be re-added back to the graph.
215	func (c *ChannelGraph) PruneGraphNodes() error {	26✔
216	c.cacheMu.Lock()	26✔
217	defer c.cacheMu.Unlock()	26✔
218		26✔
219	nodes, err := c.KVStore.PruneGraphNodes()	26✔
220	if err != nil {	26✔
NEW 221	return err	×
NEW 222	}	×
223
224	if c.graphCache != nil {	52✔
225	for _, node := range nodes {	33✔
226	c.graphCache.RemoveNode(node)	7✔
227	}	7✔
228	}
229
230	return nil	26✔
231	}

lightningnetwork / lnd / 13522725226

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