Merge pull request #9480 from ellemouton/autopilotRefactor

graph+autopilot: remove `autopilot` access to raw `graphdb.ChannelGraph`

graph/builder.go

@@ -1553,18 +1553,6 @@ func (b *Builder) FetchLightningNode(
 	return b.cfg.Graph.FetchLightningNode(node)
 }
 
-// ForEachNode is used to iterate over every node in router topology.
-//
-// NOTE: This method is part of the ChannelGraphSource interface.
-func (b *Builder) ForEachNode(
-	cb func(*models.LightningNode) error) error {
-
-	return b.cfg.Graph.ForEachNode(
-		func(_ kvdb.RTx, n *models.LightningNode) error {
-			return cb(n)
-		})
-}
-
 // ForAllOutgoingChannels is used to iterate over all outgoing channels owned by
 // the router.
 //

graph/db/graph.go

@@ -588,9 +588,9 @@ func (c *ChannelGraph) FetchNodeFeatures(
 	}
 }
 
-// ForEachNodeCached is similar to ForEachNode, but it utilizes the channel
+// ForEachNodeCached is similar to forEachNode, but it utilizes the channel
 // graph cache instead. Note that this doesn't return all the information the
-// regular ForEachNode method does.
+// regular forEachNode method does.
 //
 // NOTE: The callback contents MUST not be modified.
 func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex,
@@ -604,7 +604,7 @@ func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex,
 	// We'll iterate over each node, then the set of channels for each
 	// node, and construct a similar callback functiopn signature as the
 	// main funcotin expects.
-	return c.ForEachNode(func(tx kvdb.RTx,
+	return c.forEachNode(func(tx kvdb.RTx,
 		node *models.LightningNode) error {
 
 		channels := make(map[uint64]*DirectedChannel)
@@ -716,11 +716,25 @@ func (c *ChannelGraph) DisabledChannelIDs() ([]uint64, error) {
 // ForEachNode iterates through all the stored vertices/nodes 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. Any operations performed on the NodeTx passed to the call-back are
+// executed under the same read transaction and so, methods on the NodeTx object
+// _MUST_ only be called from within the call-back.
+func (c *ChannelGraph) ForEachNode(cb func(tx NodeRTx) error) error {
+	return c.forEachNode(func(tx kvdb.RTx,
+		node *models.LightningNode) error {
+
+		return cb(newChanGraphNodeTx(tx, c, node))
+	})
+}
+
+// forEachNode iterates through all the stored vertices/nodes 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.
 //
 // TODO(roasbeef): add iterator interface to allow for memory efficient graph
 // traversal when graph gets mega
-func (c *ChannelGraph) ForEachNode(
+func (c *ChannelGraph) forEachNode(
 	cb func(kvdb.RTx, *models.LightningNode) error) error {
 
 	traversal := func(tx kvdb.RTx) error {
@@ -4717,6 +4731,65 @@ func deserializeChanEdgePolicyRaw(r io.Reader) (*models.ChannelEdgePolicy,
 	return edge, nil
 }
 
+// chanGraphNodeTx is an implementation of the NodeRTx interface backed by the
+// ChannelGraph and a kvdb.RTx.
+type chanGraphNodeTx struct {
+	tx   kvdb.RTx
+	db   *ChannelGraph
+	node *models.LightningNode
+}
+
+// A compile-time constraint to ensure chanGraphNodeTx implements the NodeRTx
+// interface.
+var _ NodeRTx = (*chanGraphNodeTx)(nil)
+
+func newChanGraphNodeTx(tx kvdb.RTx, db *ChannelGraph,
+	node *models.LightningNode) *chanGraphNodeTx {
+
+	return &chanGraphNodeTx{
+		tx:   tx,
+		db:   db,
+		node: node,
+	}
+}
+
+// Node returns the raw information of the node.
+//
+// NOTE: This is a part of the NodeRTx interface.
+func (c *chanGraphNodeTx) Node() *models.LightningNode {
+	return c.node
+}
+
+// FetchNode fetches the node with the given pub key under the same transaction
+// used to fetch the current node. The returned node is also a NodeRTx and any
+// operations on that NodeRTx will also be done under the same transaction.
+//
+// NOTE: This is a part of the NodeRTx interface.
+func (c *chanGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) {
+	node, err := c.db.FetchLightningNodeTx(c.tx, nodePub)
+	if err != nil {
+		return nil, err
+	}
+
+	return newChanGraphNodeTx(c.tx, c.db, node), nil
+}
+
+// ForEachChannel can be used to iterate over the node's channels under
+// the same transaction used to fetch the node.
+//
+// NOTE: This is a part of the NodeRTx interface.
+func (c *chanGraphNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo,
+	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
+
+	return c.db.ForEachNodeChannelTx(c.tx, c.node.PubKeyBytes,
+		func(_ kvdb.RTx, info *models.ChannelEdgeInfo, policy1,
+			policy2 *models.ChannelEdgePolicy) error {
+
+			return f(info, policy1, policy2)
+		},
+	)
+}
+
 // MakeTestGraph creates a new instance of the ChannelGraph for testing
 // purposes.
 func MakeTestGraph(t testing.TB, modifiers ...OptionModifier) (*ChannelGraph,

graph/db/graph_cache_test.go

@@ -121,7 +121,7 @@ func TestGraphCacheAddNode(t *testing.T) {
 		assertCachedPolicyEqual(t, outPolicy1, toChannels[0].InPolicy)
 
 		// Now that we've inserted two nodes into the graph, check that
-		// we'll recover the same set of channels during ForEachNode.
+		// we'll recover the same set of channels during forEachNode.
 		nodes := make(map[route.Vertex]struct{})
 		chans := make(map[uint64]struct{})
 		_ = cache.ForEachNode(func(node route.Vertex,

graph/db/graph_test.go

@@ -1092,7 +1092,7 @@ func TestGraphTraversalCacheable(t *testing.T) {
 	// Create a map of all nodes with the iteration we know works (because
 	// it is tested in another test).
 	nodeMap := make(map[route.Vertex]struct{})
-	err = graph.ForEachNode(
+	err = graph.forEachNode(
 		func(tx kvdb.RTx, n *models.LightningNode) error {
 			nodeMap[n.PubKeyBytes] = struct{}{}
 
@@ -1217,7 +1217,7 @@ func fillTestGraph(t require.TestingT, graph *ChannelGraph, numNodes,
 
 	// Iterate over each node as returned by the graph, if all nodes are
 	// reached, then the map created above should be empty.
-	err := graph.ForEachNode(
+	err := graph.forEachNode(
 		func(_ kvdb.RTx, node *models.LightningNode) error {
 			delete(nodeIndex, node.Alias)
 			return nil
@@ -1329,7 +1329,7 @@ func assertNumChans(t *testing.T, graph *ChannelGraph, n int) {
 
 func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) {
 	numNodes := 0
-	err := graph.ForEachNode(
+	err := graph.forEachNode(
 		func(_ kvdb.RTx, _ *models.LightningNode) error {
 			numNodes++
 			return nil

graph/db/interfaces.go

@@ -0,0 +1,25 @@
+package graphdb
+
+import (
+	"github.com/lightningnetwork/lnd/graph/db/models"
+	"github.com/lightningnetwork/lnd/routing/route"
+)
+
+// NodeRTx represents transaction object with an underlying node associated that
+// can be used to make further queries to the graph under the same transaction.
+// This is useful for consistency during graph traversal and queries.
+type NodeRTx interface {
+	// Node returns the raw information of the node.
+	Node() *models.LightningNode
+
+	// ForEachChannel can be used to iterate over the node's channels under
+	// the same transaction used to fetch the node.
+	ForEachChannel(func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+		*models.ChannelEdgePolicy) error) error
+
+	// FetchNode fetches the node with the given pub key under the same
+	// transaction used to fetch the current node. The returned node is also
+	// a NodeRTx and any operations on that NodeRTx will also be done under
+	// the same transaction.
+	FetchNode(node route.Vertex) (NodeRTx, error)
+}

graph/interfaces.go

@@ -85,9 +85,6 @@ type ChannelGraphSource interface {
 	// public key. channeldb.ErrGraphNodeNotFound is returned if the node
 	// doesn't exist within the graph.
 	FetchLightningNode(route.Vertex) (*models.LightningNode, error)
-
-	// ForEachNode is used to iterate over every node in the known graph.
-	ForEachNode(func(node *models.LightningNode) error) error
 }
 
 // DB is an interface describing a persisted Lightning Network graph.
@@ -241,12 +238,6 @@ type DB interface {
 	FetchLightningNode(nodePub route.Vertex) (*models.LightningNode,
 		error)
 
-	// ForEachNode iterates through all the stored vertices/nodes 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.
-	ForEachNode(cb func(kvdb.RTx, *models.LightningNode) error) error
-
 	// ForEachNodeChannel iterates through all channels of the given node,
 	// executing the passed callback with an edge info structure and the
 	// policies of each end of the channel. The first edge policy is the