diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index d7cc3fbc9..00cfeaf33 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -370,8 +370,8 @@ func TestPopulateDBs(t *testing.T) { numPolicies = 0 ) err := graph.ForEachChannel( - ctx, func(info *models.ChannelEdgeInfo, - policy, + ctx, lnwire.GossipVersion1, + func(info *models.ChannelEdgeInfo, policy, policy2 *models.ChannelEdgePolicy) error { numChans++ @@ -497,48 +497,49 @@ func syncGraph(t *testing.T, src, dest *ChannelGraph) { } var wgChans sync.WaitGroup - err = src.ForEachChannel(ctx, func(info *models.ChannelEdgeInfo, - policy1, policy2 *models.ChannelEdgePolicy) error { + err = src.ForEachChannel(ctx, lnwire.GossipVersion1, + func(info *models.ChannelEdgeInfo, + policy1, policy2 *models.ChannelEdgePolicy) error { - // Add each channel & policy. We do this in a goroutine to - // take advantage of batch processing. - wgChans.Add(1) - go func() { - defer wgChans.Done() + // Add each channel & policy. We do this in a goroutine + // to take advantage of batch processing. + wgChans.Add(1) + go func() { + defer wgChans.Done() - err := dest.AddChannelEdge( - ctx, info, batch.LazyAdd(), - ) - if !errors.Is(err, ErrEdgeAlreadyExist) { - require.NoError(t, err) - } - - if policy1 != nil { - err = dest.UpdateEdgePolicy( - ctx, policy1, batch.LazyAdd(), + err := dest.AddChannelEdge( + ctx, info, batch.LazyAdd(), ) - require.NoError(t, err) - } + if !errors.Is(err, ErrEdgeAlreadyExist) { + require.NoError(t, err) + } - if policy2 != nil { - err = dest.UpdateEdgePolicy( - ctx, policy2, batch.LazyAdd(), - ) - require.NoError(t, err) - } + if policy1 != nil { + err = dest.UpdateEdgePolicy( + ctx, policy1, batch.LazyAdd(), + ) + require.NoError(t, err) + } - mu.Lock() - total++ - chunk++ - s.Do(func() { - reportChanStats() - chunk = 0 - }) - mu.Unlock() - }() + if policy2 != nil { + err = dest.UpdateEdgePolicy( + ctx, policy2, batch.LazyAdd(), + ) + require.NoError(t, err) + } - return nil - }, func() {}) + mu.Lock() + total++ + chunk++ + s.Do(func() { + reportChanStats() + chunk = 0 + }) + mu.Unlock() + }() + + return nil + }, func() {}) require.NoError(t, err) wgChans.Wait() @@ -638,7 +639,8 @@ func BenchmarkGraphReadMethods(b *testing.B) { fn: func(b testing.TB, store Store) { //nolint:ll err := store.ForEachChannel( - ctx, func(_ *models.ChannelEdgeInfo, + ctx, lnwire.GossipVersion1, + func(_ *models.ChannelEdgeInfo, _ *models.ChannelEdgePolicy, _ *models.ChannelEdgePolicy) error { @@ -821,7 +823,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) { //nolint:ll err = store.ForEachChannel( - ctx, + ctx, lnwire.GossipVersion1, func(_ *models.ChannelEdgeInfo, _, _ *models.ChannelEdgePolicy) error { diff --git a/graph/db/graph.go b/graph/db/graph.go index 3aa860cc1..d7398e045 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -642,10 +642,11 @@ func (c *ChannelGraph) IsPublicNode(pubKey [33]byte) (bool, error) { // ForEachChannel iterates through all channel edges stored within the graph. func (c *ChannelGraph) ForEachChannel(ctx context.Context, - cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error, reset func()) error { + v lnwire.GossipVersion, cb func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, + reset func()) error { - return c.db.ForEachChannel(ctx, cb, reset) + return c.db.ForEachChannel(ctx, v, cb, reset) } // ForEachChannelCacheable iterates through all channel edges for the cache. @@ -904,6 +905,14 @@ func (c *VersionedGraph) HasChannelEdge(chanID uint64) (bool, bool, error) { return c.db.HasChannelEdge(c.v, chanID) } +// ForEachChannel iterates through all channel edges stored within the graph. +func (c *VersionedGraph) ForEachChannel(ctx context.Context, + cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy) error, reset func()) error { + + return c.db.ForEachChannel(ctx, 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 d1aff61e1..6a13e3203 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1219,7 +1219,8 @@ func TestEdgePolicyCRUD(t *testing.T) { // assert that the deserialized policies match the original // ones. err := graph.ForEachChannel( - ctx, func(info *models.ChannelEdgeInfo, + ctx, lnwire.GossipVersion1, + func(info *models.ChannelEdgeInfo, policy1 *models.ChannelEdgePolicy, policy2 *models.ChannelEdgePolicy) error { @@ -1703,13 +1704,14 @@ func TestGraphTraversal(t *testing.T) { // Iterate through all the known channels within the graph DB, once // again if the map is empty that indicates that all edges have // properly been reached. - err = graph.ForEachChannel(ctx, func(ei *models.ChannelEdgeInfo, - _ *models.ChannelEdgePolicy, - _ *models.ChannelEdgePolicy) error { + err = graph.ForEachChannel(ctx, lnwire.GossipVersion1, + func(ei *models.ChannelEdgeInfo, + _ *models.ChannelEdgePolicy, + _ *models.ChannelEdgePolicy) error { - delete(chanIndex, ei.ChannelID) - return nil - }, func() {}) + delete(chanIndex, ei.ChannelID) + return nil + }, func() {}) require.NoError(t, err) require.Len(t, chanIndex, 0) @@ -2005,7 +2007,8 @@ func assertPruneTip(t *testing.T, graph *ChannelGraph, func assertNumChans(t *testing.T, graph *ChannelGraph, n int) { numChans := 0 err := graph.ForEachChannel( - t.Context(), func(*models.ChannelEdgeInfo, + t.Context(), lnwire.GossipVersion1, + func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error { diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 42f63f932..c650bd23b 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -162,9 +162,12 @@ type Store interface { //nolint:interfacebloat // NOTE: If an edge can't be found, or wasn't advertised, then a nil // pointer for that particular channel edge routing policy will be // passed into the callback. - ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, - reset func()) error + // + // TODO(elle): add a cross-version iteration API and make this iterate + // over all versions. + ForEachChannel(ctx context.Context, v lnwire.GossipVersion, + cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy) error, reset func()) error // ForEachChannelCacheable iterates through all the channel edges stored // within the graph and invokes the passed callback for each edge. The diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 58b227a22..03eaf02c7 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -411,10 +411,14 @@ func (c *KVStore) AddrsForNode(ctx context.Context, v lnwire.GossipVersion, // NOTE: If an edge can't be found, or wasn't advertised, then a nil pointer // for that particular channel edge routing policy will be passed into the // callback. -func (c *KVStore) ForEachChannel(_ context.Context, +func (c *KVStore) ForEachChannel(_ context.Context, v lnwire.GossipVersion, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, reset func()) error { + if v != lnwire.GossipVersion1 { + return ErrVersionNotSupportedForKVDB + } + return forEachChannel(c.db, cb, reset) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 00be30f38..b66e2fe49 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1610,11 +1610,16 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, // // NOTE: part of the Store interface. func (s *SQLStore) ForEachChannel(ctx context.Context, - cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error, reset func()) error { + v lnwire.GossipVersion, cb func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, + reset func()) error { + + if !isKnownGossipVersion(v) { + return fmt.Errorf("unsupported gossip version: %d", v) + } return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { - return forEachChannelWithPolicies(ctx, db, s.cfg, cb) + return forEachChannelWithPolicies(ctx, db, s.cfg, v, cb) }, reset) } @@ -5973,8 +5978,8 @@ func forEachNodePaginated(ctx context.Context, cfg *sqldb.QueryConfig, // forEachChannelWithPolicies executes a paginated query to process each channel // with policies in the graph. func forEachChannelWithPolicies(ctx context.Context, db SQLQueries, - cfg *SQLStoreConfig, processChannel func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, + cfg *SQLStoreConfig, v lnwire.GossipVersion, + processChannel func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { type channelBatchIDs struct { @@ -5988,7 +5993,7 @@ func forEachChannelWithPolicies(ctx context.Context, db SQLQueries, return db.ListChannelsWithPoliciesPaginated( ctx, sqlc.ListChannelsWithPoliciesPaginatedParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), ID: lastID, Limit: limit, }, diff --git a/rpcserver.go b/rpcserver.go index 927635a84..564d8cc32 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -6853,10 +6853,11 @@ func (r *rpcServer) DescribeGraph(ctx context.Context, } } - // Obtain the pointer to the global singleton channel graph, this will - // provide a consistent view of the graph due to bolt db's - // transactional model. - graph := r.server.graphDB + // Obtain the pointer to the V1 channel graph. This will provide a + // consistent view of the graph due to bolt db's transactional model. + // + // TODO(elle): switch to a cross-version graph view when available. + graph := r.server.v1Graph // First iterate through all the known nodes (connected or unconnected // within the graph), collating their current state into the RPC