diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 00cfeaf33..2887b7d22 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -669,7 +669,8 @@ func BenchmarkGraphReadMethods(b *testing.B) { name: "ForEachNodeCacheable", fn: func(b testing.TB, store Store) { err := store.ForEachNodeCacheable( - ctx, func(_ route.Vertex, + ctx, lnwire.GossipVersion1, + func(_ route.Vertex, _ *lnwire.FeatureVector) error { // Increment the counter to diff --git a/graph/db/graph.go b/graph/db/graph.go index baac06e8c..ea41f8eca 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -163,19 +163,23 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error { log.Info("Populating in-memory channel graph, this might take a " + "while...") - err := c.db.ForEachNodeCacheable(ctx, func(node route.Vertex, - features *lnwire.FeatureVector) error { + // TODO(elle): the cache should be populated with data from across + // protocol versions. + err := c.db.ForEachNodeCacheable( + ctx, lnwire.GossipVersion1, func(node route.Vertex, + features *lnwire.FeatureVector) error { - c.graphCache.AddNodeFeatures(node, features) + c.graphCache.AddNodeFeatures(node, features) - return nil - }, func() {}) + return nil + }, func() {}, + ) if err != nil { return err } err = c.db.ForEachChannelCacheable( - func(info *models.CachedEdgeInfo, + lnwire.GossipVersion1, func(info *models.CachedEdgeInfo, policy1, policy2 *models.CachedEdgePolicy) error { c.graphCache.AddChannel(info, policy1, policy2) @@ -606,10 +610,10 @@ func (c *ChannelGraph) ForEachNode(ctx context.Context, // ForEachNodeCacheable iterates through all stored vertices/nodes in the graph. func (c *ChannelGraph) ForEachNodeCacheable(ctx context.Context, - cb func(route.Vertex, *lnwire.FeatureVector) error, - reset func()) error { + v lnwire.GossipVersion, cb func(route.Vertex, + *lnwire.FeatureVector) error, reset func()) error { - return c.db.ForEachNodeCacheable(ctx, cb, reset) + return c.db.ForEachNodeCacheable(ctx, v, cb, reset) } // NodeUpdatesInHorizon returns all known lightning nodes with updates in the @@ -642,14 +646,6 @@ func (c *ChannelGraph) ForEachChannel(ctx context.Context, return c.db.ForEachChannel(ctx, v, cb, reset) } -// ForEachChannelCacheable iterates through all channel edges for the cache. -func (c *ChannelGraph) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, - reset func()) error { - - return c.db.ForEachChannelCacheable(cb, reset) -} - // DisabledChannelIDs returns the channel ids of disabled channels. func (c *ChannelGraph) DisabledChannelIDs() ([]uint64, error) { return c.db.DisabledChannelIDs() @@ -923,6 +919,22 @@ func (c *VersionedGraph) ForEachChannel(ctx context.Context, return c.db.ForEachChannel(ctx, c.v, cb, reset) } +// ForEachNodeCacheable iterates through all stored vertices/nodes in the graph. +func (c *VersionedGraph) ForEachNodeCacheable(ctx context.Context, + cb func(route.Vertex, *lnwire.FeatureVector) error, + reset func()) error { + + return c.db.ForEachNodeCacheable(ctx, c.v, cb, reset) +} + +// ForEachChannelCacheable iterates through all channel edges for the cache. +func (c *VersionedGraph) ForEachChannelCacheable( + cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, + *models.CachedEdgePolicy) error, reset func()) error { + + return c.db.ForEachChannelCacheable(c.v, cb, reset) +} + // IsPublicNode determines whether the node is seen as public in the graph. func (c *VersionedGraph) IsPublicNode(pubKey [33]byte) (bool, error) { return c.db.IsPublicNode(c.v, pubKey) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 4a2beeff0..792aaf85a 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1847,16 +1847,17 @@ func TestGraphTraversalCacheable(t *testing.T) { // iterating over each node, once again if the map is empty that // indicates that all edges have properly been reached. var nodes []route.Vertex - err = graph.ForEachNodeCacheable(ctx, func(node route.Vertex, - features *lnwire.FeatureVector) error { + err = graph.ForEachNodeCacheable( + ctx, lnwire.GossipVersion1, func(node route.Vertex, + features *lnwire.FeatureVector) error { - delete(nodeMap, node) - nodes = append(nodes, node) + delete(nodeMap, node) + nodes = append(nodes, node) - return nil - }, func() { - nodes = nil - }) + return nil + }, func() { + nodes = nil + }) require.NoError(t, err) require.Len(t, nodeMap, 0) @@ -5062,15 +5063,16 @@ func BenchmarkForEachChannel(b *testing.B) { ) var nodes []route.Vertex - err := graph.ForEachNodeCacheable(ctx, func(node route.Vertex, - vector *lnwire.FeatureVector) error { + err := graph.ForEachNodeCacheable( + ctx, lnwire.GossipVersion1, func(node route.Vertex, + vector *lnwire.FeatureVector) error { - nodes = append(nodes, node) + nodes = append(nodes, node) - return nil - }, func() { - nodes = nil - }) + return nil + }, func() { + nodes = nil + }) require.NoError(b, err) for _, n := range nodes { diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 9c904b569..e7ddc6588 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -103,8 +103,9 @@ type Store interface { //nolint:interfacebloat // in the graph, executing the passed callback with each node // encountered. If the callback returns an error, then the transaction // is aborted and the iteration stops early. - ForEachNodeCacheable(ctx context.Context, cb func(route.Vertex, - *lnwire.FeatureVector) error, reset func()) error + ForEachNodeCacheable(ctx context.Context, v lnwire.GossipVersion, + cb func(route.Vertex, *lnwire.FeatureVector) error, + reset func()) error // LookupAlias attempts to return the alias as advertised by the target // node. @@ -182,9 +183,9 @@ type Store interface { //nolint:interfacebloat // // NOTE: this method is like ForEachChannel but fetches only the data // required for the graph cache. - ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, - reset func()) error + ForEachChannelCacheable(v lnwire.GossipVersion, + cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, + *models.CachedEdgePolicy) error, reset func()) error // DisabledChannelIDs returns the channel ids of disabled channels. // A channel is disabled when two of the associated ChanelEdgePolicies diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 7c81d5f42..6cff4e5e9 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -496,9 +496,13 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo, // // NOTE: this method is like ForEachChannel but fetches only the data required // for the graph cache. -func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, - reset func()) error { +func (c *KVStore) ForEachChannelCacheable(v lnwire.GossipVersion, + cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, + *models.CachedEdgePolicy) error, reset func()) error { + + if v != lnwire.GossipVersion1 { + return ErrVersionNotSupportedForKVDB + } return c.db.View(func(tx kvdb.RTx) error { edges := tx.ReadBucket(edgeBucket) @@ -874,8 +878,12 @@ func forEachNode(db kvdb.Backend, // callback returns an error, then the transaction is aborted and the iteration // stops early. func (c *KVStore) ForEachNodeCacheable(_ context.Context, - cb func(route.Vertex, *lnwire.FeatureVector) error, - reset func()) error { + v lnwire.GossipVersion, cb func(route.Vertex, + *lnwire.FeatureVector) error, reset func()) error { + + if v != lnwire.GossipVersion1 { + return ErrVersionNotSupportedForKVDB + } traversal := func(tx kvdb.RTx) error { // First grab the nodes bucket which stores the mapping from diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index c219082ee..03083b909 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1003,12 +1003,12 @@ func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, // callback returns an error, then the transaction is aborted and the iteration // stops early. func (s *SQLStore) ForEachNodeCacheable(ctx context.Context, - cb func(route.Vertex, *lnwire.FeatureVector) error, - reset func()) error { + v lnwire.GossipVersion, cb func(route.Vertex, + *lnwire.FeatureVector) error, reset func()) error { err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodeCacheable( - ctx, s.cfg.QueryCfg, db, + ctx, s.cfg.QueryCfg, db, v, func(_ int64, nodePub route.Vertex, features *lnwire.FeatureVector) error { @@ -1547,12 +1547,16 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, withAddrs bool, // // NOTE: this method is like ForEachChannel but fetches only the data // required for the graph cache. -func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, - reset func()) error { +func (s *SQLStore) ForEachChannelCacheable(v lnwire.GossipVersion, + cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, + *models.CachedEdgePolicy) error, reset func()) error { ctx := context.TODO() + if !isKnownGossipVersion(v) { + return fmt.Errorf("unsupported gossip version: %d", v) + } + handleChannel := func(_ context.Context, row sqlc.ListChannelsWithPoliciesForCachePaginatedRow) error { @@ -1596,7 +1600,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, return db.ListChannelsWithPoliciesForCachePaginated( ctx, sqlc.ListChannelsWithPoliciesForCachePaginatedParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), ID: lastID, Limit: limit, }, @@ -3475,11 +3479,12 @@ func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries, return nil } -// forEachNodeCacheable fetches all V1 node IDs and pub keys from the database, +// forEachNodeCacheable fetches all node IDs and pub keys from the database, // and executes the provided callback for each node. It does so via pagination // along with batch loading of the node feature bits. func forEachNodeCacheable(ctx context.Context, cfg *sqldb.QueryConfig, - db SQLQueries, processNode func(nodeID int64, nodePub route.Vertex, + db SQLQueries, v lnwire.GossipVersion, + processNode func(nodeID int64, nodePub route.Vertex, features *lnwire.FeatureVector) error) error { handleNode := func(_ context.Context, @@ -3504,7 +3509,7 @@ func forEachNodeCacheable(ctx context.Context, cfg *sqldb.QueryConfig, return db.ListNodeIDsAndPubKeys( ctx, sqlc.ListNodeIDsAndPubKeysParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), ID: lastID, Limit: limit, }, diff --git a/server.go b/server.go index 909246343..a91d266c0 100644 --- a/server.go +++ b/server.go @@ -1128,7 +1128,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, DefaultRoutingPolicy: cc.RoutingPolicy, ForAllOutgoingChannels: func(ctx context.Context, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error, + *models.ChannelEdgePolicy) error, reset func()) error { return s.v1Graph.ForEachNodeChannel(