graph/db: impl GraphSession

Author: ellemouton

graph/db/sql_store.go

@@ -2674,6 +2674,63 @@ func (s *SQLStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) {
 	return isClosed, nil
 }
 
+// GraphSession will provide the call-back with access to a NodeTraverser
+// instance which can be used to perform queries against the channel graph.
+//
+// NOTE: part of the V1Store interface.
+func (s *SQLStore) GraphSession(cb func(graph NodeTraverser) error) error {
+	var ctx = context.TODO()
+
+	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+		return cb(newSQLNodeTraverser(db, s.cfg.ChainHash))
+	}, sqldb.NoOpReset)
+}
+
+// sqlNodeTraverser implements the NodeTraverser interface but with a backing
+// read only transaction for a consistent view of the graph.
+type sqlNodeTraverser struct {
+	db    SQLQueries
+	chain chainhash.Hash
+}
+
+// A compile-time assertion to ensure that sqlNodeTraverser implements the
+// NodeTraverser interface.
+var _ NodeTraverser = (*sqlNodeTraverser)(nil)
+
+// newSQLNodeTraverser creates a new instance of the sqlNodeTraverser.
+func newSQLNodeTraverser(db SQLQueries,
+	chain chainhash.Hash) *sqlNodeTraverser {
+
+	return &sqlNodeTraverser{
+		db:    db,
+		chain: chain,
+	}
+}
+
+// ForEachNodeDirectedChannel calls the callback for every channel of the given
+// node.
+//
+// NOTE: Part of the NodeTraverser interface.
+func (s *sqlNodeTraverser) ForEachNodeDirectedChannel(nodePub route.Vertex,
+	cb func(channel *DirectedChannel) error) error {
+
+	ctx := context.TODO()
+
+	return forEachNodeDirectedChannel(ctx, s.db, nodePub, cb)
+}
+
+// FetchNodeFeatures returns the features of the given node. If the node is
+// unknown, assume no additional features are supported.
+//
+// NOTE: Part of the NodeTraverser interface.
+func (s *sqlNodeTraverser) FetchNodeFeatures(nodePub route.Vertex) (
+	*lnwire.FeatureVector, error) {
+
+	ctx := context.TODO()
+
+	return fetchNodeFeatures(ctx, s.db, nodePub)
+}
+
 // forEachNodeDirectedChannel iterates through all channels of a given
 // node, executing the passed callback on the directed edge representing the
 // channel and its incoming policy. If the node is not found, no error is

routing/pathfind_test.go

@@ -160,7 +160,9 @@ func makeTestGraph(t *testing.T, useCache bool) (*graphdb.ChannelGraph,
 	kvdb.Backend, error) {
 
 	// Create channelgraph for the first time.
-	graph := graphdb.MakeTestGraph(t, graphdb.WithUseGraphCache(useCache))
+	graph := graphdb.MakeTestGraphNew(
+		t, graphdb.WithUseGraphCache(useCache),
+	)
 	require.NoError(t, graph.Start())
 	t.Cleanup(func() {
 		require.NoError(t, graph.Stop())
@@ -289,6 +291,11 @@ func parseTestGraph(t *testing.T, useCache bool, path string) (
 			}
 
 			source = dbNode
+
+			// If this is the source node, we don't have to call
+			// AddLightningNode below since we will call
+			// SetSourceNode later.
+			continue
 		}
 
 		// With the node fully parsed, add it as a vertex within the
@@ -540,8 +547,8 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 	privKeyMap := make(map[string]*btcec.PrivateKey)
 
 	nodeIndex := byte(0)
-	addNodeWithAlias := func(alias string, features *lnwire.FeatureVector) (
-		*models.LightningNode, error) {
+	addNodeWithAlias := func(alias string,
+		features *lnwire.FeatureVector) error {
 
 		keyBytes := []byte{
 			0, 0, 0, 0, 0, 0, 0, 0,
@@ -571,29 +578,29 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 
 		// With the node fully parsed, add it as a vertex within the
 		// graph.
-		if err := graph.AddLightningNode(ctx, dbNode); err != nil {
-			return nil, err
+		if alias == source {
+			err = graph.SetSourceNode(ctx, dbNode)
+			require.NoError(t, err)
+		} else {
+			err := graph.AddLightningNode(ctx, dbNode)
+			require.NoError(t, err)
 		}
 
 		aliasMap[alias] = dbNode.PubKeyBytes
 		nodeIndex++
 
-		return dbNode, nil
+		return nil
 	}
 
 	// Add the source node.
-	dbNode, err := addNodeWithAlias(
+	err = addNodeWithAlias(
 		source, lnwire.NewFeatureVector(
 			lnwire.NewRawFeatureVector(sourceFeatureBits...),
 			lnwire.Features,
 		),
 	)
 	require.NoError(t, err)
 
-	if err = graph.SetSourceNode(ctx, dbNode); err != nil {
-		return nil, err
-	}
-
 	// Initialize variable that keeps track of the next channel id to assign
 	// if none is specified.
 	nextUnassignedChannelID := uint64(100000)
@@ -611,7 +618,7 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 					features =
 						node.testChannelPolicy.Features
 				}
-				_, err := addNodeWithAlias(
+				err := addNodeWithAlias(
 					node.Alias, features,
 				)
 				if err != nil {
@@ -2157,6 +2164,7 @@ func runRouteFailMaxHTLC(t *testing.T, useCache bool) {
 	require.NoError(t, err, "unable to fetch channel edges by ID")
 	midEdge.MessageFlags = 1
 	midEdge.MaxHTLC = payAmt - 1
+	midEdge.LastUpdate = midEdge.LastUpdate.Add(time.Second)
 	err = graph.UpdateEdgePolicy(context.Background(), midEdge)
 	require.NoError(t, err)
 
@@ -2199,10 +2207,12 @@ func runRouteFailDisabledEdge(t *testing.T, useCache bool) {
 	_, e1, e2, err := graph.graph.FetchChannelEdgesByID(roasToPham)
 	require.NoError(t, err, "unable to fetch edge")
 	e1.ChannelFlags |= lnwire.ChanUpdateDisabled
+	e1.LastUpdate = e1.LastUpdate.Add(time.Second)
 	if err := graph.graph.UpdateEdgePolicy(ctx, e1); err != nil {
 		t.Fatalf("unable to update edge: %v", err)
 	}
 	e2.ChannelFlags |= lnwire.ChanUpdateDisabled
+	e2.LastUpdate = e2.LastUpdate.Add(time.Second)
 	if err := graph.graph.UpdateEdgePolicy(ctx, e2); err != nil {
 		t.Fatalf("unable to update edge: %v", err)
 	}
@@ -2220,6 +2230,7 @@ func runRouteFailDisabledEdge(t *testing.T, useCache bool) {
 	_, e, _, err := graph.graph.FetchChannelEdgesByID(phamToSophon)
 	require.NoError(t, err, "unable to fetch edge")
 	e.ChannelFlags |= lnwire.ChanUpdateDisabled
+	e.LastUpdate = e.LastUpdate.Add(time.Second)
 	if err := graph.graph.UpdateEdgePolicy(ctx, e); err != nil {
 		t.Fatalf("unable to update edge: %v", err)
 	}
@@ -2302,10 +2313,12 @@ func runPathSourceEdgesBandwidth(t *testing.T, useCache bool) {
 	_, e1, e2, err := graph.graph.FetchChannelEdgesByID(roasToSongoku)
 	require.NoError(t, err, "unable to fetch edge")
 	e1.ChannelFlags |= lnwire.ChanUpdateDisabled
+	e1.LastUpdate = e1.LastUpdate.Add(time.Second)
 	if err := graph.graph.UpdateEdgePolicy(ctx, e1); err != nil {
 		t.Fatalf("unable to update edge: %v", err)
 	}
 	e2.ChannelFlags |= lnwire.ChanUpdateDisabled
+	e2.LastUpdate = e2.LastUpdate.Add(time.Second)
 	if err := graph.graph.UpdateEdgePolicy(ctx, e2); err != nil {
 		t.Fatalf("unable to update edge: %v", err)
 	}