mirror of
https://github.com/lightningnetwork/lnd.git
synced 2025-06-12 18:01:23 +02:00
graph/db: use V1Store interface in ChannelGraph
Use the new `V1Store` interface as a field in the `ChannelGraph` rather than a pointer to the `KVStore` implementation in preparation for allowing a SQL implementation of `V1Store` to be used by the `ChannelGraph`. Note that two tests are adjusted in this commit to be skipped if the V1Store is not the `KVStore` implementation since the tests are very bbolt specific.
This commit is contained in:
parent
83eb7e3fed
commit
9f79322f81
@ -48,7 +48,7 @@ type ChannelGraph struct {
|
|||||||
|
|
||||||
graphCache *GraphCache
|
graphCache *GraphCache
|
||||||
|
|
||||||
*KVStore
|
V1Store
|
||||||
*topologyManager
|
*topologyManager
|
||||||
|
|
||||||
quit chan struct{}
|
quit chan struct{}
|
||||||
@ -70,7 +70,7 @@ func NewChannelGraph(cfg *Config, options ...ChanGraphOption) (*ChannelGraph,
|
|||||||
}
|
}
|
||||||
|
|
||||||
g := &ChannelGraph{
|
g := &ChannelGraph{
|
||||||
KVStore: store,
|
V1Store: store,
|
||||||
topologyManager: newTopologyManager(),
|
topologyManager: newTopologyManager(),
|
||||||
quit: make(chan struct{}),
|
quit: make(chan struct{}),
|
||||||
}
|
}
|
||||||
@ -184,7 +184,7 @@ func (c *ChannelGraph) populateCache() error {
|
|||||||
log.Info("Populating in-memory channel graph, this might take a " +
|
log.Info("Populating in-memory channel graph, this might take a " +
|
||||||
"while...")
|
"while...")
|
||||||
|
|
||||||
err := c.KVStore.ForEachNodeCacheable(func(node route.Vertex,
|
err := c.V1Store.ForEachNodeCacheable(func(node route.Vertex,
|
||||||
features *lnwire.FeatureVector) error {
|
features *lnwire.FeatureVector) error {
|
||||||
|
|
||||||
c.graphCache.AddNodeFeatures(node, features)
|
c.graphCache.AddNodeFeatures(node, features)
|
||||||
@ -195,7 +195,7 @@ func (c *ChannelGraph) populateCache() error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
err = c.KVStore.ForEachChannel(func(info *models.ChannelEdgeInfo,
|
err = c.V1Store.ForEachChannel(func(info *models.ChannelEdgeInfo,
|
||||||
policy1, policy2 *models.ChannelEdgePolicy) error {
|
policy1, policy2 *models.ChannelEdgePolicy) error {
|
||||||
|
|
||||||
c.graphCache.AddChannel(info, policy1, policy2)
|
c.graphCache.AddChannel(info, policy1, policy2)
|
||||||
@ -229,7 +229,7 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex,
|
|||||||
return c.graphCache.ForEachChannel(node, cb)
|
return c.graphCache.ForEachChannel(node, cb)
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.KVStore.ForEachNodeDirectedChannel(node, cb)
|
return c.V1Store.ForEachNodeDirectedChannel(node, cb)
|
||||||
}
|
}
|
||||||
|
|
||||||
// FetchNodeFeatures returns the features of the given node. If no features are
|
// FetchNodeFeatures returns the features of the given node. If no features are
|
||||||
@ -245,7 +245,7 @@ func (c *ChannelGraph) FetchNodeFeatures(node route.Vertex) (
|
|||||||
return c.graphCache.GetFeatures(node), nil
|
return c.graphCache.GetFeatures(node), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.KVStore.FetchNodeFeatures(node)
|
return c.V1Store.FetchNodeFeatures(node)
|
||||||
}
|
}
|
||||||
|
|
||||||
// GraphSession will provide the call-back with access to a NodeTraverser
|
// GraphSession will provide the call-back with access to a NodeTraverser
|
||||||
@ -257,7 +257,7 @@ func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error) error {
|
|||||||
return cb(c)
|
return cb(c)
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.KVStore.GraphSession(cb)
|
return c.V1Store.GraphSession(cb)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ForEachNodeCached iterates through all the stored vertices/nodes in the
|
// ForEachNodeCached iterates through all the stored vertices/nodes in the
|
||||||
@ -271,7 +271,7 @@ func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex,
|
|||||||
return c.graphCache.ForEachNode(cb)
|
return c.graphCache.ForEachNode(cb)
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.KVStore.ForEachNodeCached(cb)
|
return c.V1Store.ForEachNodeCached(cb)
|
||||||
}
|
}
|
||||||
|
|
||||||
// AddLightningNode adds a vertex/node to the graph database. If the node is not
|
// AddLightningNode adds a vertex/node to the graph database. If the node is not
|
||||||
@ -286,7 +286,7 @@ func (c *ChannelGraph) AddLightningNode(node *models.LightningNode,
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
err := c.KVStore.AddLightningNode(node, op...)
|
err := c.V1Store.AddLightningNode(node, op...)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -312,7 +312,7 @@ func (c *ChannelGraph) DeleteLightningNode(nodePub route.Vertex) error {
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
err := c.KVStore.DeleteLightningNode(nodePub)
|
err := c.V1Store.DeleteLightningNode(nodePub)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -336,7 +336,7 @@ func (c *ChannelGraph) AddChannelEdge(edge *models.ChannelEdgeInfo,
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
err := c.KVStore.AddChannelEdge(edge, op...)
|
err := c.V1Store.AddChannelEdge(edge, op...)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -361,7 +361,7 @@ func (c *ChannelGraph) MarkEdgeLive(chanID uint64) error {
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
err := c.KVStore.MarkEdgeLive(chanID)
|
err := c.V1Store.MarkEdgeLive(chanID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -369,7 +369,7 @@ func (c *ChannelGraph) MarkEdgeLive(chanID uint64) error {
|
|||||||
if c.graphCache != nil {
|
if c.graphCache != nil {
|
||||||
// We need to add the channel back into our graph cache,
|
// We need to add the channel back into our graph cache,
|
||||||
// otherwise we won't use it for path finding.
|
// otherwise we won't use it for path finding.
|
||||||
infos, err := c.KVStore.FetchChanInfos([]uint64{chanID})
|
infos, err := c.V1Store.FetchChanInfos([]uint64{chanID})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -400,7 +400,7 @@ func (c *ChannelGraph) DeleteChannelEdges(strictZombiePruning, markZombie bool,
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
infos, err := c.KVStore.DeleteChannelEdges(
|
infos, err := c.V1Store.DeleteChannelEdges(
|
||||||
strictZombiePruning, markZombie, chanIDs...,
|
strictZombiePruning, markZombie, chanIDs...,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -432,7 +432,7 @@ func (c *ChannelGraph) DisconnectBlockAtHeight(height uint32) (
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
edges, err := c.KVStore.DisconnectBlockAtHeight(height)
|
edges, err := c.V1Store.DisconnectBlockAtHeight(height)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@ -463,7 +463,7 @@ func (c *ChannelGraph) PruneGraph(spentOutputs []*wire.OutPoint,
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
edges, nodes, err := c.KVStore.PruneGraph(
|
edges, nodes, err := c.V1Store.PruneGraph(
|
||||||
spentOutputs, blockHash, blockHeight,
|
spentOutputs, blockHash, blockHeight,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -508,7 +508,7 @@ func (c *ChannelGraph) PruneGraphNodes() error {
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
nodes, err := c.KVStore.PruneGraphNodes()
|
nodes, err := c.V1Store.PruneGraphNodes()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -530,7 +530,7 @@ func (c *ChannelGraph) PruneGraphNodes() error {
|
|||||||
func (c *ChannelGraph) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo,
|
func (c *ChannelGraph) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo,
|
||||||
isZombieChan func(time.Time, time.Time) bool) ([]uint64, error) {
|
isZombieChan func(time.Time, time.Time) bool) ([]uint64, error) {
|
||||||
|
|
||||||
unknown, knownZombies, err := c.KVStore.FilterKnownChanIDs(chansInfo)
|
unknown, knownZombies, err := c.V1Store.FilterKnownChanIDs(chansInfo)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@ -559,7 +559,7 @@ func (c *ChannelGraph) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo,
|
|||||||
// timestamps could bring it back from the dead, then we mark it
|
// timestamps could bring it back from the dead, then we mark it
|
||||||
// alive, and we let it be added to the set of IDs to query our
|
// alive, and we let it be added to the set of IDs to query our
|
||||||
// peer for.
|
// peer for.
|
||||||
err := c.KVStore.MarkEdgeLive(
|
err := c.V1Store.MarkEdgeLive(
|
||||||
info.ShortChannelID.ToUint64(),
|
info.ShortChannelID.ToUint64(),
|
||||||
)
|
)
|
||||||
// Since there is a chance that the edge could have been marked
|
// Since there is a chance that the edge could have been marked
|
||||||
@ -583,7 +583,7 @@ func (c *ChannelGraph) MarkEdgeZombie(chanID uint64,
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
err := c.KVStore.MarkEdgeZombie(chanID, pubKey1, pubKey2)
|
err := c.V1Store.MarkEdgeZombie(chanID, pubKey1, pubKey2)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -608,7 +608,7 @@ func (c *ChannelGraph) UpdateEdgePolicy(edge *models.ChannelEdgePolicy,
|
|||||||
c.cacheMu.Lock()
|
c.cacheMu.Lock()
|
||||||
defer c.cacheMu.Unlock()
|
defer c.cacheMu.Unlock()
|
||||||
|
|
||||||
from, to, err := c.KVStore.UpdateEdgePolicy(edge, op...)
|
from, to, err := c.V1Store.UpdateEdgePolicy(edge, op...)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
@ -2950,6 +2950,12 @@ func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) {
|
|||||||
graph, err := MakeTestGraph(t)
|
graph, err := MakeTestGraph(t)
|
||||||
require.NoError(t, err, "unable to make test database")
|
require.NoError(t, err, "unable to make test database")
|
||||||
|
|
||||||
|
// The update index only applies to the bbolt graph.
|
||||||
|
boltStore, ok := graph.V1Store.(*KVStore)
|
||||||
|
if !ok {
|
||||||
|
t.Skipf("skipping test that is aimed at a bbolt graph DB")
|
||||||
|
}
|
||||||
|
|
||||||
sourceNode := createTestVertex(t)
|
sourceNode := createTestVertex(t)
|
||||||
if err := graph.SetSourceNode(sourceNode); err != nil {
|
if err := graph.SetSourceNode(sourceNode); err != nil {
|
||||||
t.Fatalf("unable to set source node: %v", err)
|
t.Fatalf("unable to set source node: %v", err)
|
||||||
@ -2999,7 +3005,7 @@ func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) {
|
|||||||
timestampSet[t] = struct{}{}
|
timestampSet[t] = struct{}{}
|
||||||
}
|
}
|
||||||
|
|
||||||
err := kvdb.View(graph.db, func(tx kvdb.RTx) error {
|
err := kvdb.View(boltStore.db, func(tx kvdb.RTx) error {
|
||||||
edges := tx.ReadBucket(edgeBucket)
|
edges := tx.ReadBucket(edgeBucket)
|
||||||
if edges == nil {
|
if edges == nil {
|
||||||
return ErrGraphNoEdgesFound
|
return ErrGraphNoEdgesFound
|
||||||
@ -3467,6 +3473,12 @@ func TestEdgePolicyMissingMaxHtcl(t *testing.T) {
|
|||||||
graph, err := MakeTestGraph(t)
|
graph, err := MakeTestGraph(t)
|
||||||
require.NoError(t, err, "unable to make test database")
|
require.NoError(t, err, "unable to make test database")
|
||||||
|
|
||||||
|
// This test currently directly edits the bytes stored in the bbolt DB.
|
||||||
|
boltStore, ok := graph.V1Store.(*KVStore)
|
||||||
|
if !ok {
|
||||||
|
t.Skipf("skipping test that is aimed at a bbolt graph DB")
|
||||||
|
}
|
||||||
|
|
||||||
// We'd like to test the update of edges inserted into the database, so
|
// We'd like to test the update of edges inserted into the database, so
|
||||||
// we create two vertexes to connect.
|
// we create two vertexes to connect.
|
||||||
node1 := createTestVertex(t)
|
node1 := createTestVertex(t)
|
||||||
@ -3515,25 +3527,11 @@ func TestEdgePolicyMissingMaxHtcl(t *testing.T) {
|
|||||||
|
|
||||||
// Attempting to deserialize these bytes should return an error.
|
// Attempting to deserialize these bytes should return an error.
|
||||||
r := bytes.NewReader(stripped)
|
r := bytes.NewReader(stripped)
|
||||||
err = kvdb.View(graph.db, func(tx kvdb.RTx) error {
|
_, err = deserializeChanEdgePolicy(r)
|
||||||
nodes := tx.ReadBucket(nodeBucket)
|
require.ErrorIs(t, err, ErrEdgePolicyOptionalFieldNotFound)
|
||||||
if nodes == nil {
|
|
||||||
return ErrGraphNotFound
|
|
||||||
}
|
|
||||||
|
|
||||||
_, err = deserializeChanEdgePolicy(r)
|
|
||||||
if err != ErrEdgePolicyOptionalFieldNotFound {
|
|
||||||
t.Fatalf("expected "+
|
|
||||||
"ErrEdgePolicyOptionalFieldNotFound, got %v",
|
|
||||||
err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}, func() {})
|
|
||||||
require.NoError(t, err, "error reading db")
|
|
||||||
|
|
||||||
// Put the stripped bytes in the DB.
|
// Put the stripped bytes in the DB.
|
||||||
err = kvdb.Update(graph.db, func(tx kvdb.RwTx) error {
|
err = kvdb.Update(boltStore.db, func(tx kvdb.RwTx) error {
|
||||||
edges := tx.ReadWriteBucket(edgeBucket)
|
edges := tx.ReadWriteBucket(edgeBucket)
|
||||||
if edges == nil {
|
if edges == nil {
|
||||||
return ErrEdgeNotFound
|
return ErrEdgeNotFound
|
||||||
|
Loading…
x
Reference in New Issue
Block a user