multi: remove the NodeRTx interface type
This commit is contained in:
10 files changed
+58
-155
No files matched your search
+5
-4
@@ -10,6 +10,7 @@ import (
|
|||||||
"github.com/btcsuite/btcd/btcec/v2/ecdsa"
|
"github.com/btcsuite/btcd/btcec/v2/ecdsa"
|
||||||
"github.com/btcsuite/btcd/btcutil"
|
"github.com/btcsuite/btcd/btcutil"
|
||||||
graphdb "github.com/lightningnetwork/lnd/graph/db"
|
graphdb "github.com/lightningnetwork/lnd/graph/db"
|
||||||
|
"github.com/lightningnetwork/lnd/graph/db/models"
|
||||||
"github.com/lightningnetwork/lnd/lnwire"
|
"github.com/lightningnetwork/lnd/lnwire"
|
||||||
"github.com/lightningnetwork/lnd/routing/route"
|
"github.com/lightningnetwork/lnd/routing/route"
|
||||||
)
|
)
|
||||||
@@ -83,17 +84,17 @@ func (d *dbNode) Addrs() []net.Addr {
|
|||||||
func (d *databaseChannelGraph) ForEachNode(ctx context.Context,
|
func (d *databaseChannelGraph) ForEachNode(ctx context.Context,
|
||||||
cb func(context.Context, Node) error, reset func()) error {
|
cb func(context.Context, Node) error, reset func()) error {
|
||||||
|
|
||||||
return d.db.ForEachNode(ctx, func(nodeTx graphdb.NodeRTx) error {
|
return d.db.ForEachNode(ctx, func(n *models.LightningNode) error {
|
||||||
// We'll skip over any node that doesn't have any advertised
|
// We'll skip over any node that doesn't have any advertised
|
||||||
// addresses. As we won't be able to reach them to actually
|
// addresses. As we won't be able to reach them to actually
|
||||||
// open any channels.
|
// open any channels.
|
||||||
if len(nodeTx.Node().Addresses) == 0 {
|
if len(n.Addresses) == 0 {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
node := &dbNode{
|
node := &dbNode{
|
||||||
pub: nodeTx.Node().PubKeyBytes,
|
pub: n.PubKeyBytes,
|
||||||
addrs: nodeTx.Node().Addresses,
|
addrs: n.Addresses,
|
||||||
}
|
}
|
||||||
|
|
||||||
return cb(ctx, node)
|
return cb(ctx, node)
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"github.com/btcsuite/btcd/btcutil"
|
"github.com/btcsuite/btcd/btcutil"
|
||||||
"github.com/btcsuite/btcd/wire"
|
"github.com/btcsuite/btcd/wire"
|
||||||
graphdb "github.com/lightningnetwork/lnd/graph/db"
|
graphdb "github.com/lightningnetwork/lnd/graph/db"
|
||||||
|
"github.com/lightningnetwork/lnd/graph/db/models"
|
||||||
"github.com/lightningnetwork/lnd/lnwire"
|
"github.com/lightningnetwork/lnd/lnwire"
|
||||||
"github.com/lightningnetwork/lnd/routing/route"
|
"github.com/lightningnetwork/lnd/routing/route"
|
||||||
)
|
)
|
||||||
@@ -228,9 +229,9 @@ type GraphSource interface {
|
|||||||
// ForEachNode iterates through all the stored vertices/nodes in the
|
// ForEachNode iterates through all the stored vertices/nodes in the
|
||||||
// graph, executing the passed callback with each node encountered. If
|
// graph, executing the passed callback with each node encountered. If
|
||||||
// the callback returns an error, then the transaction is aborted and
|
// the callback returns an error, then the transaction is aborted and
|
||||||
// the iteration stops early. Any operations performed on the NodeTx
|
// the iteration stops early.
|
||||||
// passed to the call-back are executed under the same read transaction.
|
ForEachNode(context.Context, func(*models.LightningNode) error,
|
||||||
ForEachNode(context.Context, func(graphdb.NodeRTx) error, func()) error
|
func()) error
|
||||||
|
|
||||||
// ForEachNodeCached is similar to ForEachNode, but it utilizes the
|
// ForEachNodeCached is similar to ForEachNode, but it utilizes the
|
||||||
// channel graph cache if one is available. It is less consistent than
|
// channel graph cache if one is available. It is less consistent than
|
||||||
|
|||||||
@@ -741,26 +741,3 @@ func (m *memChannelGraph) addRandNode() (*btcec.PublicKey, error) {
|
|||||||
|
|
||||||
return newPub, nil
|
return newPub, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
type testNodeTx struct {
|
|
||||||
db *testDBGraph
|
|
||||||
node *models.LightningNode
|
|
||||||
}
|
|
||||||
|
|
||||||
func (t *testNodeTx) Node() *models.LightningNode {
|
|
||||||
return t.node
|
|
||||||
}
|
|
||||||
|
|
||||||
func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo,
|
|
||||||
*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
|
|
||||||
|
|
||||||
return t.db.db.ForEachNodeChannel(context.Background(),
|
|
||||||
t.node.PubKeyBytes, func(edge *models.ChannelEdgeInfo, policy1,
|
|
||||||
policy2 *models.ChannelEdgePolicy) error {
|
|
||||||
|
|
||||||
return f(edge, policy1, policy2)
|
|
||||||
}, func() {},
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
var _ graphdb.NodeRTx = (*testNodeTx)(nil)
|
|
||||||
+13
-15
@@ -338,12 +338,15 @@ func TestPopulateDBs(t *testing.T) {
|
|||||||
// graph.
|
// graph.
|
||||||
countNodes := func(graph *ChannelGraph) int {
|
countNodes := func(graph *ChannelGraph) int {
|
||||||
numNodes := 0
|
numNodes := 0
|
||||||
err := graph.ForEachNode(ctx, func(tx NodeRTx) error {
|
err := graph.ForEachNode(
|
||||||
numNodes++
|
ctx, func(node *models.LightningNode) error {
|
||||||
return nil
|
numNodes++
|
||||||
}, func() {
|
|
||||||
numNodes = 0
|
return nil
|
||||||
})
|
}, func() {
|
||||||
|
numNodes = 0
|
||||||
|
},
|
||||||
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
return numNodes
|
return numNodes
|
||||||
@@ -487,18 +490,12 @@ func syncGraph(t *testing.T, src, dest *ChannelGraph) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
var wgNodes sync.WaitGroup
|
var wgNodes sync.WaitGroup
|
||||||
err := src.ForEachNode(ctx, func(tx NodeRTx) error {
|
err := src.ForEachNode(ctx, func(node *models.LightningNode) error {
|
||||||
wgNodes.Add(1)
|
wgNodes.Add(1)
|
||||||
go func() {
|
go func() {
|
||||||
defer wgNodes.Done()
|
defer wgNodes.Done()
|
||||||
|
|
||||||
// NOTE: even though the transaction (tx) may have
|
err := dest.AddLightningNode(ctx, node, batch.LazyAdd())
|
||||||
// already been aborted, it is still ok to use the
|
|
||||||
// Node() result since that is a static object that
|
|
||||||
// is not affected by the transaction state.
|
|
||||||
err := dest.AddLightningNode(
|
|
||||||
ctx, tx.Node(), batch.LazyAdd(),
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
mu.Lock()
|
mu.Lock()
|
||||||
@@ -658,7 +655,8 @@ func BenchmarkGraphReadMethods(b *testing.B) {
|
|||||||
name: "ForEachNode",
|
name: "ForEachNode",
|
||||||
fn: func(b testing.TB, store V1Store) {
|
fn: func(b testing.TB, store V1Store) {
|
||||||
err := store.ForEachNode(
|
err := store.ForEachNode(
|
||||||
ctx, func(_ NodeRTx) error {
|
ctx,
|
||||||
|
func(_ *models.LightningNode) error {
|
||||||
// Increment the counter to
|
// Increment the counter to
|
||||||
// ensure the callback is doing
|
// ensure the callback is doing
|
||||||
// something.
|
// something.
|
||||||
|
|||||||
@@ -1469,8 +1469,8 @@ func TestGraphTraversalCacheable(t *testing.T) {
|
|||||||
// Create a map of all nodes with the iteration we know works (because
|
// Create a map of all nodes with the iteration we know works (because
|
||||||
// it is tested in another test).
|
// it is tested in another test).
|
||||||
nodeMap := make(map[route.Vertex]struct{})
|
nodeMap := make(map[route.Vertex]struct{})
|
||||||
err := graph.ForEachNode(ctx, func(tx NodeRTx) error {
|
err := graph.ForEachNode(ctx, func(n *models.LightningNode) error {
|
||||||
nodeMap[tx.Node().PubKeyBytes] = struct{}{}
|
nodeMap[n.PubKeyBytes] = struct{}{}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}, func() {})
|
}, func() {})
|
||||||
@@ -1601,8 +1601,8 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes,
|
|||||||
|
|
||||||
// Iterate over each node as returned by the graph, if all nodes are
|
// Iterate over each node as returned by the graph, if all nodes are
|
||||||
// reached, then the map created above should be empty.
|
// reached, then the map created above should be empty.
|
||||||
err := graph.ForEachNode(ctx, func(tx NodeRTx) error {
|
err := graph.ForEachNode(ctx, func(n *models.LightningNode) error {
|
||||||
delete(nodeIndex, tx.Node().Alias)
|
delete(nodeIndex, n.Alias)
|
||||||
return nil
|
return nil
|
||||||
}, func() {})
|
}, func() {})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -1710,11 +1710,12 @@ func assertNumChans(t *testing.T, graph *ChannelGraph, n int) {
|
|||||||
|
|
||||||
func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) {
|
func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) {
|
||||||
numNodes := 0
|
numNodes := 0
|
||||||
err := graph.ForEachNode(context.Background(), func(tx NodeRTx) error {
|
err := graph.ForEachNode(context.Background(),
|
||||||
numNodes++
|
func(_ *models.LightningNode) error {
|
||||||
|
numNodes++
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}, func() {})
|
}, func() {})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
_, _, line, _ := runtime.Caller(1)
|
_, _, line, _ := runtime.Caller(1)
|
||||||
t.Fatalf("line %v: unable to scan nodes: %v", line, err)
|
t.Fatalf("line %v: unable to scan nodes: %v", line, err)
|
||||||
|
|||||||
+2
-13
@@ -14,14 +14,6 @@ import (
|
|||||||
"github.com/lightningnetwork/lnd/routing/route"
|
"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
|
|
||||||
}
|
|
||||||
|
|
||||||
// NodeTraverser is an abstract read only interface that provides information
|
// NodeTraverser is an abstract read only interface that provides information
|
||||||
// about nodes and their edges. The interface is about providing fast read-only
|
// about nodes and their edges. The interface is about providing fast read-only
|
||||||
// access to the graph and so if a cache is available, it should be used.
|
// access to the graph and so if a cache is available, it should be used.
|
||||||
@@ -94,11 +86,8 @@ type V1Store interface { //nolint:interfacebloat
|
|||||||
// ForEachNode iterates through all the stored vertices/nodes in the
|
// ForEachNode iterates through all the stored vertices/nodes in the
|
||||||
// graph, executing the passed callback with each node encountered. If
|
// graph, executing the passed callback with each node encountered. If
|
||||||
// the callback returns an error, then the transaction is aborted and
|
// the callback returns an error, then the transaction is aborted and
|
||||||
// the iteration stops early. Any operations performed on the NodeTx
|
// the iteration stops early.
|
||||||
// passed to the call-back are executed under the same read transaction
|
ForEachNode(ctx context.Context, cb func(*models.LightningNode) error,
|
||||||
// and so, methods on the NodeTx object _MUST_ only be called from
|
|
||||||
// within the call-back.
|
|
||||||
ForEachNode(ctx context.Context, cb func(tx NodeRTx) error,
|
|
||||||
reset func()) error
|
reset func()) error
|
||||||
|
|
||||||
// ForEachNodeCacheable iterates through all the stored vertices/nodes
|
// ForEachNodeCacheable iterates through all the stored vertices/nodes
|
||||||
|
|||||||
+5
-34
@@ -800,16 +800,16 @@ func (c *KVStore) DisabledChannelIDs() ([]uint64, error) {
|
|||||||
// ForEachNode iterates through all the stored vertices/nodes in the graph,
|
// ForEachNode iterates through all the stored vertices/nodes in the graph,
|
||||||
// executing the passed callback with each node encountered. If the callback
|
// executing the passed callback with each node encountered. If the callback
|
||||||
// returns an error, then the transaction is aborted and the iteration stops
|
// 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
|
// early.
|
||||||
// executed under the same read transaction and so, methods on the NodeTx object
|
//
|
||||||
// _MUST_ only be called from within the call-back.
|
// NOTE: this is part of the V1Store interface.
|
||||||
func (c *KVStore) ForEachNode(_ context.Context,
|
func (c *KVStore) ForEachNode(_ context.Context,
|
||||||
cb func(tx NodeRTx) error, reset func()) error {
|
cb func(*models.LightningNode) error, reset func()) error {
|
||||||
|
|
||||||
return forEachNode(c.db, func(tx kvdb.RTx,
|
return forEachNode(c.db, func(tx kvdb.RTx,
|
||||||
node *models.LightningNode) error {
|
node *models.LightningNode) error {
|
||||||
|
|
||||||
return cb(newChanGraphNodeTx(tx, c, node))
|
return cb(node)
|
||||||
}, reset)
|
}, reset)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -4877,32 +4877,3 @@ func deserializeChanEdgePolicyRaw(r io.Reader) (*models.ChannelEdgePolicy,
|
|||||||
|
|
||||||
return edge, nil
|
return edge, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// chanGraphNodeTx is an implementation of the NodeRTx interface backed by the
|
|
||||||
// KVStore and a kvdb.RTx.
|
|
||||||
type chanGraphNodeTx struct {
|
|
||||||
tx kvdb.RTx
|
|
||||||
db *KVStore
|
|
||||||
node *models.LightningNode
|
|
||||||
}
|
|
||||||
|
|
||||||
// A compile-time constraint to ensure chanGraphNodeTx implements the NodeRTx
|
|
||||||
// interface.
|
|
||||||
var _ NodeRTx = (*chanGraphNodeTx)(nil)
|
|
||||||
|
|
||||||
func newChanGraphNodeTx(tx kvdb.RTx, db *KVStore,
|
|
||||||
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
|
|
||||||
}
|
|
||||||
@@ -395,23 +395,24 @@ func assertInSync(t *testing.T, kvDB *KVStore, sqlDB *SQLStore,
|
|||||||
func fetchAllNodes(t *testing.T, store V1Store) []*models.LightningNode {
|
func fetchAllNodes(t *testing.T, store V1Store) []*models.LightningNode {
|
||||||
nodes := make([]*models.LightningNode, 0)
|
nodes := make([]*models.LightningNode, 0)
|
||||||
|
|
||||||
err := store.ForEachNode(context.Background(), func(tx NodeRTx) error {
|
err := store.ForEachNode(context.Background(),
|
||||||
node := tx.Node()
|
func(node *models.LightningNode) error {
|
||||||
|
|
||||||
// Call PubKey to ensure the objects cached pubkey is set so that
|
// Call PubKey to ensure the objects cached pubkey is set so that
|
||||||
// the objects can be compared as a whole.
|
// the objects can be compared as a whole.
|
||||||
_, err := node.PubKey()
|
_, err := node.PubKey()
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// Sort the addresses to ensure a consistent order.
|
// Sort the addresses to ensure a consistent order.
|
||||||
sortAddrs(node.Addresses)
|
sortAddrs(node.Addresses)
|
||||||
|
|
||||||
nodes = append(nodes, node)
|
nodes = append(nodes, node)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}, func() {
|
}, func() {
|
||||||
nodes = nil
|
nodes = nil
|
||||||
})
|
},
|
||||||
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// Sort the nodes by their public key to ensure a consistent order.
|
// Sort the nodes by their public key to ensure a consistent order.
|
||||||
|
|||||||
+4
-40
@@ -799,60 +799,24 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context,
|
|||||||
// ForEachNode iterates through all the stored vertices/nodes in the graph,
|
// ForEachNode iterates through all the stored vertices/nodes in the graph,
|
||||||
// executing the passed callback with each node encountered. If the callback
|
// executing the passed callback with each node encountered. If the callback
|
||||||
// returns an error, then the transaction is aborted and the iteration stops
|
// 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
|
// early.
|
||||||
// executed under the same read transaction and so, methods on the NodeTx object
|
|
||||||
// _MUST_ only be called from within the call-back.
|
|
||||||
//
|
//
|
||||||
// NOTE: part of the V1Store interface.
|
// NOTE: part of the V1Store interface.
|
||||||
func (s *SQLStore) ForEachNode(ctx context.Context,
|
func (s *SQLStore) ForEachNode(ctx context.Context,
|
||||||
cb func(tx NodeRTx) error, reset func()) error {
|
cb func(node *models.LightningNode) error, reset func()) error {
|
||||||
|
|
||||||
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
||||||
return forEachNodePaginated(
|
return forEachNodePaginated(
|
||||||
ctx, s.cfg.QueryCfg, db,
|
ctx, s.cfg.QueryCfg, db,
|
||||||
ProtocolV1,
|
ProtocolV1, func(_ context.Context, _ int64,
|
||||||
func(ctx context.Context, dbNodeID int64,
|
|
||||||
node *models.LightningNode) error {
|
node *models.LightningNode) error {
|
||||||
|
|
||||||
return cb(newSQLGraphNodeTx(
|
return cb(node)
|
||||||
db, s.cfg, dbNodeID, node,
|
|
||||||
))
|
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
}, reset)
|
}, reset)
|
||||||
}
|
}
|
||||||
|
|
||||||
// sqlGraphNodeTx is an implementation of the NodeRTx interface backed by the
|
|
||||||
// SQLStore and a SQL transaction.
|
|
||||||
type sqlGraphNodeTx struct {
|
|
||||||
db SQLQueries
|
|
||||||
id int64
|
|
||||||
node *models.LightningNode
|
|
||||||
cfg *SQLStoreConfig
|
|
||||||
}
|
|
||||||
|
|
||||||
// A compile-time constraint to ensure sqlGraphNodeTx implements the NodeRTx
|
|
||||||
// interface.
|
|
||||||
var _ NodeRTx = (*sqlGraphNodeTx)(nil)
|
|
||||||
|
|
||||||
func newSQLGraphNodeTx(db SQLQueries, cfg *SQLStoreConfig,
|
|
||||||
id int64, node *models.LightningNode) *sqlGraphNodeTx {
|
|
||||||
|
|
||||||
return &sqlGraphNodeTx{
|
|
||||||
db: db,
|
|
||||||
cfg: cfg,
|
|
||||||
id: id,
|
|
||||||
node: node,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Node returns the raw information of the node.
|
|
||||||
//
|
|
||||||
// NOTE: This is a part of the NodeRTx interface.
|
|
||||||
func (s *sqlGraphNodeTx) Node() *models.LightningNode {
|
|
||||||
return s.node
|
|
||||||
}
|
|
||||||
|
|
||||||
// ForEachNodeDirectedChannel iterates through all channels of a given node,
|
// ForEachNodeDirectedChannel iterates through all channels of a given node,
|
||||||
// executing the passed callback on the directed edge representing the channel
|
// executing the passed callback on the directed edge representing the channel
|
||||||
// and its incoming policy. If the callback returns an error, then the iteration
|
// and its incoming policy. If the callback returns an error, then the iteration
|
||||||
|
|||||||
+2
-2
@@ -6757,8 +6757,8 @@ func (r *rpcServer) DescribeGraph(ctx context.Context,
|
|||||||
// First iterate through all the known nodes (connected or unconnected
|
// First iterate through all the known nodes (connected or unconnected
|
||||||
// within the graph), collating their current state into the RPC
|
// within the graph), collating their current state into the RPC
|
||||||
// response.
|
// response.
|
||||||
err := graph.ForEachNode(ctx, func(nodeTx graphdb.NodeRTx) error {
|
err := graph.ForEachNode(ctx, func(node *models.LightningNode) error {
|
||||||
lnNode := marshalNode(nodeTx.Node())
|
lnNode := marshalNode(node)
|
||||||
|
|
||||||
resp.Nodes = append(resp.Nodes, lnNode)
|
resp.Nodes = append(resp.Nodes, lnNode)
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user