Merge pull request #9695 from ellemouton/graph-clean-up

graph/db: remove various `kvdb` parameters from exported methods

Author: ellemouton

autopilot/prefattach_test.go

@@ -736,7 +736,7 @@ func (t *testNodeTx) Node() *models.LightningNode {
 func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo,
 	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
 
-	return t.db.db.ForEachNodeChannel(t.node.PubKeyBytes, func(_ kvdb.RTx,
+	return t.db.db.ForEachNodeChannel(t.node.PubKeyBytes, func(
 		edge *models.ChannelEdgeInfo, policy1,
 		policy2 *models.ChannelEdgePolicy) error {
 

graph/builder.go

@@ -13,7 +13,6 @@ import (
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	graphdb "github.com/lightningnetwork/lnd/graph/db"
 	"github.com/lightningnetwork/lnd/graph/db/models"
-	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/lnutils"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
@@ -1276,8 +1275,7 @@ func (b *Builder) ForAllOutgoingChannels(cb func(*models.ChannelEdgeInfo,
 	*models.ChannelEdgePolicy) error) error {
 
 	return b.cfg.Graph.ForEachNodeChannel(b.cfg.SelfNode,
-		func(_ kvdb.RTx, c *models.ChannelEdgeInfo,
-			e *models.ChannelEdgePolicy,
+		func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy,
 			_ *models.ChannelEdgePolicy) error {
 
 			if e == nil {

graph/db/graph_test.go

@@ -618,7 +618,7 @@ func createChannelEdge(node1, node2 *models.LightningNode) (
 	chanID := uint64(prand.Int63())
 	outpoint := wire.OutPoint{
 		Hash:  rev,
-		Index: 9,
+		Index: prand.Uint32(),
 	}
 
 	// Add the new edge to the database, this should proceed without any
@@ -991,6 +991,97 @@ func newEdgePolicy(chanID uint64, updateTime int64) *models.ChannelEdgePolicy {
 	}
 }
 
+// TestForEachSourceNodeChannel tests that the ForEachSourceNodeChannel
+// correctly iterates through the channels of the set source node.
+func TestForEachSourceNodeChannel(t *testing.T) {
+	t.Parallel()
+
+	graph, err := MakeTestGraph(t)
+	require.NoError(t, err, "unable to make test database")
+
+	// Create a source node (A) and set it as such in the DB.
+	nodeA := createTestVertex(t)
+	require.NoError(t, graph.SetSourceNode(nodeA))
+
+	// Now, create a few more nodes (B, C, D) along with some channels
+	// between them. We'll create the following graph:
+	//
+	// 	A -- B -- D
+	//  	|
+	// 	C
+	//
+	// The graph includes a channel (B-D) that does not belong to the source
+	// node along with 2 channels (A-B and A-C) that do belong to the source
+	// node. For the A-B channel, we will let the source node set an
+	// outgoing policy but for the A-C channel, we will set only an incoming
+	// policy.
+
+	nodeB := createTestVertex(t)
+	nodeC := createTestVertex(t)
+	nodeD := createTestVertex(t)
+
+	abEdge, abPolicy1, abPolicy2 := createChannelEdge(nodeA, nodeB)
+	require.NoError(t, graph.AddChannelEdge(abEdge))
+	acEdge, acPolicy1, acPolicy2 := createChannelEdge(nodeA, nodeC)
+	require.NoError(t, graph.AddChannelEdge(acEdge))
+	bdEdge, _, _ := createChannelEdge(nodeB, nodeD)
+	require.NoError(t, graph.AddChannelEdge(bdEdge))
+
+	// Figure out which of the policies returned above are node A's so that
+	// we know which to persist.
+	//
+	// First, set the outgoing policy for the A-B channel.
+	abPolicyAOutgoing := abPolicy1
+	if !bytes.Equal(abPolicy1.ToNode[:], nodeB.PubKeyBytes[:]) {
+		abPolicyAOutgoing = abPolicy2
+	}
+	require.NoError(t, graph.UpdateEdgePolicy(abPolicyAOutgoing))
+
+	// Now, set the incoming policy for the A-C channel.
+	acPolicyAIncoming := acPolicy1
+	if !bytes.Equal(acPolicy1.ToNode[:], nodeA.PubKeyBytes[:]) {
+		acPolicyAIncoming = acPolicy2
+	}
+	require.NoError(t, graph.UpdateEdgePolicy(acPolicyAIncoming))
+
+	type sourceNodeChan struct {
+		otherNode  route.Vertex
+		havePolicy bool
+	}
+
+	// Put together our expected source node channels.
+	expectedSrcChans := map[wire.OutPoint]*sourceNodeChan{
+		abEdge.ChannelPoint: {
+			otherNode:  nodeB.PubKeyBytes,
+			havePolicy: true,
+		},
+		acEdge.ChannelPoint: {
+			otherNode:  nodeC.PubKeyBytes,
+			havePolicy: false,
+		},
+	}
+
+	// Now, we'll use the ForEachSourceNodeChannel and assert that it
+	// returns the expected data in the call-back.
+	err = graph.ForEachSourceNodeChannel(func(chanPoint wire.OutPoint,
+		havePolicy bool, otherNode *models.LightningNode) error {
+
+		require.Contains(t, expectedSrcChans, chanPoint)
+		expected := expectedSrcChans[chanPoint]
+
+		require.Equal(
+			t, expected.otherNode[:], otherNode.PubKeyBytes[:],
+		)
+		require.Equal(t, expected.havePolicy, havePolicy)
+
+		delete(expectedSrcChans, chanPoint)
+
+		return nil
+	})
+	require.NoError(t, err)
+	require.Empty(t, expectedSrcChans)
+}
+
 func TestGraphTraversal(t *testing.T) {
 	t.Parallel()
 
@@ -1050,7 +1141,7 @@ func TestGraphTraversal(t *testing.T) {
 	numNodeChans := 0
 	firstNode, secondNode := nodeList[0], nodeList[1]
 	err = graph.ForEachNodeChannel(firstNode.PubKeyBytes,
-		func(_ kvdb.RTx, _ *models.ChannelEdgeInfo, outEdge,
+		func(_ *models.ChannelEdgeInfo, outEdge,
 			inEdge *models.ChannelEdgePolicy) error {
 
 			// All channels between first and second node should
@@ -1126,26 +1217,15 @@ func TestGraphTraversalCacheable(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, nodeMap, 0)
 
-	err = graph.db.View(func(tx kvdb.RTx) error {
-		for _, node := range nodes {
-			err := graph.ForEachNodeChannelTx(tx, node,
-				func(tx kvdb.RTx, info *models.ChannelEdgeInfo,
-					policy *models.ChannelEdgePolicy,
-					policy2 *models.ChannelEdgePolicy) error { //nolint:ll
-
-					delete(chanIndex, info.ChannelID)
-					return nil
-				},
-			)
-			if err != nil {
-				return err
-			}
-		}
-
-		return nil
-	}, func() {})
-
-	require.NoError(t, err)
+	for _, node := range nodes {
+		err = graph.ForEachNodeDirectedChannel(
+			node, func(d *DirectedChannel) error {
+				delete(chanIndex, d.ChannelID)
+				return nil
+			},
+		)
+		require.NoError(t, err)
+	}
 	require.Len(t, chanIndex, 0)
 }
 
@@ -2802,7 +2882,7 @@ func TestIncompleteChannelPolicies(t *testing.T) {
 
 		calls := 0
 		err := graph.ForEachNodeChannel(node.PubKeyBytes,
-			func(_ kvdb.RTx, _ *models.ChannelEdgeInfo, outEdge,
+			func(_ *models.ChannelEdgeInfo, outEdge,
 				inEdge *models.ChannelEdgePolicy) error {
 
 				if !expectedOut && outEdge != nil {
@@ -3921,8 +4001,7 @@ func BenchmarkForEachChannel(b *testing.B) {
 		require.NoError(b, err)
 
 		for _, n := range nodes {
-			cb := func(tx kvdb.RTx,
-				info *models.ChannelEdgeInfo,
+			cb := func(info *models.ChannelEdgeInfo,
 				policy *models.ChannelEdgePolicy,
 				policy2 *models.ChannelEdgePolicy) error { //nolint:ll
 

graph/db/kv_store.go

@@ -299,29 +299,6 @@ var graphTopLevelBuckets = [][]byte{
 	closedScidBucket,
 }
 
-// Wipe completely deletes all saved state within all used buckets within the
-// database. The deletion is done in a single transaction, therefore this
-// operation is fully atomic.
-func (c *KVStore) Wipe() error {
-	err := kvdb.Update(c.db, func(tx kvdb.RwTx) error {
-		for _, tlb := range graphTopLevelBuckets {
-			err := tx.DeleteTopLevelBucket(tlb)
-			if err != nil &&
-				!errors.Is(err, kvdb.ErrBucketNotFound) {
-
-				return err
-			}
-		}
-
-		return nil
-	}, func() {})
-	if err != nil {
-		return err
-	}
-
-	return initKVStore(c.db)
-}
-
 // createChannelDB creates and initializes a fresh version of  In
 // the case that the target path has not yet been created or doesn't yet exist,
 // then the path is created. Additionally, all required top-level buckets used
@@ -584,7 +561,7 @@ func (c *KVStore) ForEachNodeCached(cb func(node route.Vertex,
 
 		channels := make(map[uint64]*DirectedChannel)
 
-		err := c.ForEachNodeChannelTx(tx, node.PubKeyBytes,
+		err := c.forEachNodeChannelTx(tx, node.PubKeyBytes,
 			func(tx kvdb.RTx, e *models.ChannelEdgeInfo,
 				p1 *models.ChannelEdgePolicy,
 				p2 *models.ChannelEdgePolicy) error {
@@ -2873,7 +2850,7 @@ func (c *KVStore) isPublic(tx kvdb.RTx, nodePub route.Vertex,
 	// used to terminate the check early.
 	nodeIsPublic := false
 	errDone := errors.New("done")
-	err := c.ForEachNodeChannelTx(tx, nodePub, func(tx kvdb.RTx,
+	err := c.forEachNodeChannelTx(tx, nodePub, func(tx kvdb.RTx,
 		info *models.ChannelEdgeInfo, _ *models.ChannelEdgePolicy,
 		_ *models.ChannelEdgePolicy) error {
 
@@ -3126,13 +3103,56 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend,
 //
 // Unknown policies are passed into the callback as nil values.
 func (c *KVStore) ForEachNodeChannel(nodePub route.Vertex,
-	cb func(kvdb.RTx, *models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
 		*models.ChannelEdgePolicy) error) error {
 
-	return nodeTraversal(nil, nodePub[:], c.db, cb)
+	return nodeTraversal(nil, nodePub[:], c.db, func(_ kvdb.RTx,
+		info *models.ChannelEdgeInfo, policy,
+		policy2 *models.ChannelEdgePolicy) error {
+
+		return cb(info, policy, policy2)
+	})
+}
+
+// ForEachSourceNodeChannel iterates through all channels of the source node,
+// executing the passed callback on each. The callback is provided with the
+// channel's outpoint, whether we have a policy for the channel and the channel
+// peer's node information.
+func (c *KVStore) ForEachSourceNodeChannel(cb func(chanPoint wire.OutPoint,
+	havePolicy bool, otherNode *models.LightningNode) error) error {
+
+	return kvdb.View(c.db, func(tx kvdb.RTx) error {
+		nodes := tx.ReadBucket(nodeBucket)
+		if nodes == nil {
+			return ErrGraphNotFound
+		}
+
+		node, err := c.sourceNode(nodes)
+		if err != nil {
+			return err
+		}
+
+		return nodeTraversal(
+			tx, node.PubKeyBytes[:], c.db, func(tx kvdb.RTx,
+				info *models.ChannelEdgeInfo,
+				policy, _ *models.ChannelEdgePolicy) error {
+
+				peer, err := c.fetchOtherNode(
+					tx, info, node.PubKeyBytes[:],
+				)
+				if err != nil {
+					return err
+				}
+
+				return cb(
+					info.ChannelPoint, policy != nil, peer,
+				)
+			},
+		)
+	}, func() {})
 }
 
-// ForEachNodeChannelTx iterates through all channels of the given node,
+// forEachNodeChannelTx 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 outgoing edge *to*
 // the connecting node, while the second is the incoming edge *from* the
@@ -3145,19 +3165,19 @@ func (c *KVStore) ForEachNodeChannel(nodePub route.Vertex,
 // should be passed as the first argument.  Otherwise, the first argument should
 // be nil and a fresh transaction will be created to execute the graph
 // traversal.
-func (c *KVStore) ForEachNodeChannelTx(tx kvdb.RTx,
+func (c *KVStore) forEachNodeChannelTx(tx kvdb.RTx,
 	nodePub route.Vertex, cb func(kvdb.RTx, *models.ChannelEdgeInfo,
 		*models.ChannelEdgePolicy,
 		*models.ChannelEdgePolicy) error) error {
 
 	return nodeTraversal(tx, nodePub[:], c.db, cb)
 }
 
-// FetchOtherNode attempts to fetch the full LightningNode that's opposite of
+// fetchOtherNode attempts to fetch the full LightningNode that's opposite of
 // the target node in the channel. This is useful when one knows the pubkey of
 // one of the nodes, and wishes to obtain the full LightningNode for the other
 // end of the channel.
-func (c *KVStore) FetchOtherNode(tx kvdb.RTx,
+func (c *KVStore) fetchOtherNode(tx kvdb.RTx,
 	channel *models.ChannelEdgeInfo, thisNodeKey []byte) (
 	*models.LightningNode, error) {
 
@@ -4687,7 +4707,7 @@ func (c *chanGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) {
 func (c *chanGraphNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo,
 	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
 
-	return c.db.ForEachNodeChannelTx(c.tx, c.node.PubKeyBytes,
+	return c.db.forEachNodeChannelTx(c.tx, c.node.PubKeyBytes,
 		func(_ kvdb.RTx, info *models.ChannelEdgeInfo, policy1,
 			policy2 *models.ChannelEdgePolicy) error {
 

graph/interfaces.go

@@ -8,7 +8,6 @@ import (
 	"github.com/lightningnetwork/lnd/batch"
 	graphdb "github.com/lightningnetwork/lnd/graph/db"
 	"github.com/lightningnetwork/lnd/graph/db/models"
-	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/routing/route"
 )
@@ -254,10 +253,9 @@ type DB interface {
 	// to the caller.
 	//
 	// Unknown policies are passed into the callback as nil values.
-	ForEachNodeChannel(nodePub route.Vertex, cb func(kvdb.RTx,
-		*models.ChannelEdgeInfo,
-		*models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error
+	ForEachNodeChannel(nodePub route.Vertex,
+		cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+			*models.ChannelEdgePolicy) error) error
 
 	// AddEdgeProof sets the proof of an existing edge in the graph
 	// database.

rpcserver.go

@@ -56,7 +56,6 @@ import (
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/invoices"
 	"github.com/lightningnetwork/lnd/keychain"
-	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/labels"
 	"github.com/lightningnetwork/lnd/lncfg"
 	"github.com/lightningnetwork/lnd/lnrpc"
@@ -6959,7 +6958,7 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context,
 	)
 
 	err = graph.ForEachNodeChannel(node.PubKeyBytes,
-		func(_ kvdb.RTx, edge *models.ChannelEdgeInfo,
+		func(edge *models.ChannelEdgeInfo,
 			c1, c2 *models.ChannelEdgePolicy) error {
 
 			numChannels++
@@ -7641,7 +7640,7 @@ func (r *rpcServer) FeeReport(ctx context.Context,
 
 	var feeReports []*lnrpc.ChannelFeeReport
 	err = channelGraph.ForEachNodeChannel(selfNode.PubKeyBytes,
-		func(_ kvdb.RTx, chanInfo *models.ChannelEdgeInfo,
+		func(chanInfo *models.ChannelEdgeInfo,
 			edgePolicy, _ *models.ChannelEdgePolicy) error {
 
 			// Self node should always have policies for its