graph/db: thread context through to SourceNode
This commit is contained in:
12 files changed
+49
-32
No files matched your search
@@ -347,7 +347,7 @@ func TestWakeUpOnStaleBranch(t *testing.T) {
|
|||||||
// Give time to process new blocks.
|
// Give time to process new blocks.
|
||||||
time.Sleep(time.Millisecond * 500)
|
time.Sleep(time.Millisecond * 500)
|
||||||
|
|
||||||
selfNode, err := ctx.graph.SourceNode()
|
selfNode, err := ctx.graph.SourceNode(context.Background())
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// Create new router with same graph database.
|
// Create new router with same graph database.
|
||||||
|
|||||||
@@ -373,7 +373,7 @@ func TestSourceNode(t *testing.T) {
|
|||||||
|
|
||||||
// Attempt to fetch the source node, this should return an error as the
|
// Attempt to fetch the source node, this should return an error as the
|
||||||
// source node hasn't yet been set.
|
// source node hasn't yet been set.
|
||||||
_, err := graph.SourceNode()
|
_, err := graph.SourceNode(ctx)
|
||||||
require.ErrorIs(t, err, ErrSourceNodeNotSet)
|
require.ErrorIs(t, err, ErrSourceNodeNotSet)
|
||||||
|
|
||||||
// Set the source node, this should insert the node into the
|
// Set the source node, this should insert the node into the
|
||||||
@@ -382,7 +382,7 @@ func TestSourceNode(t *testing.T) {
|
|||||||
|
|
||||||
// Retrieve the source node from the database, it should exactly match
|
// Retrieve the source node from the database, it should exactly match
|
||||||
// the one we set above.
|
// the one we set above.
|
||||||
sourceNode, err := graph.SourceNode()
|
sourceNode, err := graph.SourceNode(ctx)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
compareNodes(t, testNode, sourceNode)
|
compareNodes(t, testNode, sourceNode)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -339,7 +339,7 @@ type V1Store interface { //nolint:interfacebloat
|
|||||||
// treated as the center node within a star-graph. This method may be
|
// treated as the center node within a star-graph. This method may be
|
||||||
// used to kick off a path finding algorithm in order to explore the
|
// used to kick off a path finding algorithm in order to explore the
|
||||||
// reachability of another node based off the source node.
|
// reachability of another node based off the source node.
|
||||||
SourceNode() (*models.LightningNode, error)
|
SourceNode(ctx context.Context) (*models.LightningNode, error)
|
||||||
|
|
||||||
// SetSourceNode sets the source node within the graph database. The
|
// SetSourceNode sets the source node within the graph database. The
|
||||||
// source node is to be used as the center of a star-graph within path
|
// source node is to be used as the center of a star-graph within path
|
||||||
|
|||||||
@@ -874,7 +874,9 @@ func (c *KVStore) ForEachNodeCacheable(cb func(route.Vertex,
|
|||||||
// as the center node within a star-graph. This method may be used to kick off
|
// as the center node within a star-graph. This method may be used to kick off
|
||||||
// a path finding algorithm in order to explore the reachability of another
|
// a path finding algorithm in order to explore the reachability of another
|
||||||
// node based off the source node.
|
// node based off the source node.
|
||||||
func (c *KVStore) SourceNode() (*models.LightningNode, error) {
|
func (c *KVStore) SourceNode(_ context.Context) (*models.LightningNode,
|
||||||
|
error) {
|
||||||
|
|
||||||
var source *models.LightningNode
|
var source *models.LightningNode
|
||||||
err := kvdb.View(c.db, func(tx kvdb.RTx) error {
|
err := kvdb.View(c.db, func(tx kvdb.RTx) error {
|
||||||
// First grab the nodes bucket which stores the mapping from
|
// First grab the nodes bucket which stores the mapping from
|
||||||
|
|||||||
@@ -394,8 +394,8 @@ func (s *SQLStore) LookupAlias(pub *btcec.PublicKey) (string, error) {
|
|||||||
// node based off the source node.
|
// node based off the source node.
|
||||||
//
|
//
|
||||||
// NOTE: part of the V1Store interface.
|
// NOTE: part of the V1Store interface.
|
||||||
func (s *SQLStore) SourceNode() (*models.LightningNode, error) {
|
func (s *SQLStore) SourceNode(ctx context.Context) (*models.LightningNode,
|
||||||
ctx := context.TODO()
|
error) {
|
||||||
|
|
||||||
var node *models.LightningNode
|
var node *models.LightningNode
|
||||||
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
||||||
|
|||||||
+1
-1
@@ -134,7 +134,7 @@ type DB interface {
|
|||||||
// treated as the center node within a star-graph. This method may be
|
// treated as the center node within a star-graph. This method may be
|
||||||
// used to kick off a path finding algorithm in order to explore the
|
// used to kick off a path finding algorithm in order to explore the
|
||||||
// reachability of another node based off the source node.
|
// reachability of another node based off the source node.
|
||||||
SourceNode() (*models.LightningNode, error)
|
SourceNode(ctx context.Context) (*models.LightningNode, error)
|
||||||
|
|
||||||
// DisabledChannelIDs returns the channel ids of disabled channels.
|
// DisabledChannelIDs returns the channel ids of disabled channels.
|
||||||
// A channel is disabled when two of the associated ChanelEdgePolicies
|
// A channel is disabled when two of the associated ChanelEdgePolicies
|
||||||
|
|||||||
@@ -1077,7 +1077,7 @@ func createTestCtxSingleNode(t *testing.T,
|
|||||||
func (c *testCtx) RestartBuilder(t *testing.T) {
|
func (c *testCtx) RestartBuilder(t *testing.T) {
|
||||||
c.chainView.Reset()
|
c.chainView.Reset()
|
||||||
|
|
||||||
selfNode, err := c.graph.SourceNode()
|
selfNode, err := c.graph.SourceNode(context.Background())
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// With the chainView reset, we'll now re-create the builder itself, and
|
// With the chainView reset, we'll now re-create the builder itself, and
|
||||||
@@ -1150,7 +1150,7 @@ func createTestCtxFromGraphInstanceAssumeValid(t *testing.T,
|
|||||||
ConfChan: make(chan *chainntnfs.TxConfirmation),
|
ConfChan: make(chan *chainntnfs.TxConfirmation),
|
||||||
}
|
}
|
||||||
|
|
||||||
selfnode, err := graphInstance.graph.SourceNode()
|
selfnode, err := graphInstance.graph.SourceNode(context.Background())
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
graphBuilder, err := NewBuilder(&Config{
|
graphBuilder, err := NewBuilder(&Config{
|
||||||
|
|||||||
@@ -663,9 +663,9 @@ func Main(cfg *Config, lisCfg ListenerCfg, implCfg *ImplementationCfg,
|
|||||||
// Now we have created all dependencies necessary to populate and
|
// Now we have created all dependencies necessary to populate and
|
||||||
// start the RPC server.
|
// start the RPC server.
|
||||||
err = rpcServer.addDeps(
|
err = rpcServer.addDeps(
|
||||||
server, interceptorChain.MacaroonService(), cfg.SubRPCServers,
|
ctx, server, interceptorChain.MacaroonService(),
|
||||||
atplManager, server.invoices, tower, multiAcceptor,
|
cfg.SubRPCServers, atplManager, server.invoices, tower,
|
||||||
server.invoiceHtlcModifier,
|
multiAcceptor, server.invoiceHtlcModifier,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return mkErr("unable to add deps to RPC server", err)
|
return mkErr("unable to add deps to RPC server", err)
|
||||||
|
|||||||
+26
-12
@@ -1067,11 +1067,12 @@ func runBasicGraphPathFinding(t *testing.T, useCache bool) {
|
|||||||
func testBasicGraphPathFindingCase(t *testing.T, graphInstance *testGraphInstance,
|
func testBasicGraphPathFindingCase(t *testing.T, graphInstance *testGraphInstance,
|
||||||
test *basicGraphPathFindingTestCase) {
|
test *basicGraphPathFindingTestCase) {
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
aliases := graphInstance.aliasMap
|
aliases := graphInstance.aliasMap
|
||||||
expectedHops := test.expectedHops
|
expectedHops := test.expectedHops
|
||||||
expectedHopCount := len(expectedHops)
|
expectedHopCount := len(expectedHops)
|
||||||
|
|
||||||
sourceNode, err := graphInstance.graph.SourceNode()
|
sourceNode, err := graphInstance.graph.SourceNode(ctx)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
sourceVertex := route.Vertex(sourceNode.PubKeyBytes)
|
sourceVertex := route.Vertex(sourceNode.PubKeyBytes)
|
||||||
|
|
||||||
@@ -1211,7 +1212,9 @@ func runPathFindingWithAdditionalEdges(t *testing.T, useCache bool) {
|
|||||||
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
||||||
require.NoError(t, err, "unable to create graph")
|
require.NoError(t, err, "unable to create graph")
|
||||||
|
|
||||||
sourceNode, err := graph.graph.SourceNode()
|
ctx := context.Background()
|
||||||
|
|
||||||
|
sourceNode, err := graph.graph.SourceNode(ctx)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
paymentAmt := lnwire.NewMSatFromSatoshis(100)
|
paymentAmt := lnwire.NewMSatFromSatoshis(100)
|
||||||
@@ -1294,7 +1297,9 @@ func runPathFindingWithBlindedPathDuplicateHop(t *testing.T, useCache bool) {
|
|||||||
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
||||||
require.NoError(t, err, "unable to create graph")
|
require.NoError(t, err, "unable to create graph")
|
||||||
|
|
||||||
sourceNode, err := graph.graph.SourceNode()
|
ctx := context.Background()
|
||||||
|
|
||||||
|
sourceNode, err := graph.graph.SourceNode(ctx)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
paymentAmt := lnwire.NewMSatFromSatoshis(100)
|
paymentAmt := lnwire.NewMSatFromSatoshis(100)
|
||||||
@@ -1779,7 +1784,9 @@ func runPathNotAvailable(t *testing.T, useCache bool) {
|
|||||||
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
||||||
require.NoError(t, err, "unable to create graph")
|
require.NoError(t, err, "unable to create graph")
|
||||||
|
|
||||||
sourceNode, err := graph.graph.SourceNode()
|
ctx := context.Background()
|
||||||
|
|
||||||
|
sourceNode, err := graph.graph.SourceNode(ctx)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
// With the test graph loaded, we'll test that queries for target that
|
// With the test graph loaded, we'll test that queries for target that
|
||||||
@@ -1835,7 +1842,7 @@ func runDestTLVGraphFallback(t *testing.T, useCache bool) {
|
|||||||
|
|
||||||
ctx := newPathFindingTestContext(t, useCache, testChannels, "roasbeef")
|
ctx := newPathFindingTestContext(t, useCache, testChannels, "roasbeef")
|
||||||
|
|
||||||
sourceNode, err := ctx.graph.SourceNode()
|
sourceNode, err := ctx.graph.SourceNode(context.Background())
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
find := func(r *RestrictParams,
|
find := func(r *RestrictParams,
|
||||||
@@ -2053,7 +2060,8 @@ func runPathInsufficientCapacity(t *testing.T, useCache bool) {
|
|||||||
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
||||||
require.NoError(t, err, "unable to create graph")
|
require.NoError(t, err, "unable to create graph")
|
||||||
|
|
||||||
sourceNode, err := graph.graph.SourceNode()
|
ctx := context.Background()
|
||||||
|
sourceNode, err := graph.graph.SourceNode(ctx)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
// Next, test that attempting to find a path in which the current
|
// Next, test that attempting to find a path in which the current
|
||||||
@@ -2083,7 +2091,8 @@ func runRouteFailMinHTLC(t *testing.T, useCache bool) {
|
|||||||
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
||||||
require.NoError(t, err, "unable to create graph")
|
require.NoError(t, err, "unable to create graph")
|
||||||
|
|
||||||
sourceNode, err := graph.graph.SourceNode()
|
ctx := context.Background()
|
||||||
|
sourceNode, err := graph.graph.SourceNode(ctx)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
// We'll not attempt to route an HTLC of 10 SAT from roasbeef to Son
|
// We'll not attempt to route an HTLC of 10 SAT from roasbeef to Son
|
||||||
@@ -2167,7 +2176,8 @@ func runRouteFailDisabledEdge(t *testing.T, useCache bool) {
|
|||||||
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
||||||
require.NoError(t, err, "unable to create graph")
|
require.NoError(t, err, "unable to create graph")
|
||||||
|
|
||||||
sourceNode, err := graph.graph.SourceNode()
|
ctx := context.Background()
|
||||||
|
sourceNode, err := graph.graph.SourceNode(ctx)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
// First, we'll try to route from roasbeef -> sophon. This should
|
// First, we'll try to route from roasbeef -> sophon. This should
|
||||||
@@ -2232,7 +2242,8 @@ func runPathSourceEdgesBandwidth(t *testing.T, useCache bool) {
|
|||||||
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
graph, err := parseTestGraph(t, useCache, basicGraphFilePath)
|
||||||
require.NoError(t, err, "unable to create graph")
|
require.NoError(t, err, "unable to create graph")
|
||||||
|
|
||||||
sourceNode, err := graph.graph.SourceNode()
|
ctx := context.Background()
|
||||||
|
sourceNode, err := graph.graph.SourceNode(ctx)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
// First, we'll try to route from roasbeef -> sophon. This should
|
// First, we'll try to route from roasbeef -> sophon. This should
|
||||||
@@ -3162,7 +3173,9 @@ func newPathFindingTestContext(t *testing.T, useCache bool,
|
|||||||
)
|
)
|
||||||
require.NoError(t, err, "unable to create graph")
|
require.NoError(t, err, "unable to create graph")
|
||||||
|
|
||||||
sourceNode, err := testGraphInstance.graph.SourceNode()
|
sourceNode, err := testGraphInstance.graph.SourceNode(
|
||||||
|
context.Background(),
|
||||||
|
)
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
ctx := &pathFindingTestContext{
|
ctx := &pathFindingTestContext{
|
||||||
@@ -3233,7 +3246,8 @@ func dbFindPath(graph *graphdb.ChannelGraph,
|
|||||||
source, target route.Vertex, amt lnwire.MilliSatoshi, timePref float64,
|
source, target route.Vertex, amt lnwire.MilliSatoshi, timePref float64,
|
||||||
finalHtlcExpiry int32) ([]*unifiedEdge, error) {
|
finalHtlcExpiry int32) ([]*unifiedEdge, error) {
|
||||||
|
|
||||||
sourceNode, err := graph.SourceNode()
|
ctx := context.Background()
|
||||||
|
sourceNode, err := graph.SourceNode(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -3264,7 +3278,7 @@ func dbFindPath(graph *graphdb.ChannelGraph,
|
|||||||
func dbFindBlindedPaths(graph *graphdb.ChannelGraph,
|
func dbFindBlindedPaths(graph *graphdb.ChannelGraph,
|
||||||
restrictions *blindedPathRestrictions) ([][]blindedHop, error) {
|
restrictions *blindedPathRestrictions) ([][]blindedHop, error) {
|
||||||
|
|
||||||
sourceNode, err := graph.SourceNode()
|
sourceNode, err := graph.SourceNode(context.Background())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -133,7 +133,7 @@ func createTestCtxFromGraphInstanceAssumeValid(t *testing.T,
|
|||||||
)
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
sourceNode, err := graphInstance.graph.SourceNode()
|
sourceNode, err := graphInstance.graph.SourceNode(context.Background())
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
sessionSource := &SessionSource{
|
sessionSource := &SessionSource{
|
||||||
GraphSessionFactory: graphInstance.graph,
|
GraphSessionFactory: graphInstance.graph,
|
||||||
@@ -1203,7 +1203,7 @@ func TestFindPathFeeWeighting(t *testing.T) {
|
|||||||
var preImage [32]byte
|
var preImage [32]byte
|
||||||
copy(preImage[:], bytes.Repeat([]byte{9}, 32))
|
copy(preImage[:], bytes.Repeat([]byte{9}, 32))
|
||||||
|
|
||||||
sourceNode, err := ctx.graph.SourceNode()
|
sourceNode, err := ctx.graph.SourceNode(context.Background())
|
||||||
require.NoError(t, err, "unable to fetch source node")
|
require.NoError(t, err, "unable to fetch source node")
|
||||||
|
|
||||||
amt := lnwire.MilliSatoshi(100)
|
amt := lnwire.MilliSatoshi(100)
|
||||||
|
|||||||
+4
-3
@@ -681,14 +681,15 @@ func newRPCServer(cfg *Config, interceptorChain *rpcperms.InterceptorChain,
|
|||||||
// addDeps populates all dependencies needed by the RPC server, and any
|
// addDeps populates all dependencies needed by the RPC server, and any
|
||||||
// of the sub-servers that it maintains. When this is done, the RPC server can
|
// of the sub-servers that it maintains. When this is done, the RPC server can
|
||||||
// be started, and start accepting RPC calls.
|
// be started, and start accepting RPC calls.
|
||||||
func (r *rpcServer) addDeps(s *server, macService *macaroons.Service,
|
func (r *rpcServer) addDeps(ctx context.Context, s *server,
|
||||||
|
macService *macaroons.Service,
|
||||||
subServerCgs *subRPCServerConfigs, atpl *autopilot.Manager,
|
subServerCgs *subRPCServerConfigs, atpl *autopilot.Manager,
|
||||||
invoiceRegistry *invoices.InvoiceRegistry, tower *watchtower.Standalone,
|
invoiceRegistry *invoices.InvoiceRegistry, tower *watchtower.Standalone,
|
||||||
chanPredicate chanacceptor.MultiplexAcceptor,
|
chanPredicate chanacceptor.MultiplexAcceptor,
|
||||||
invoiceHtlcModifier *invoices.HtlcModificationInterceptor) error {
|
invoiceHtlcModifier *invoices.HtlcModificationInterceptor) error {
|
||||||
|
|
||||||
// Set up router rpc backend.
|
// Set up router rpc backend.
|
||||||
selfNode, err := s.graphDB.SourceNode()
|
selfNode, err := s.graphDB.SourceNode(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -7653,7 +7654,7 @@ func (r *rpcServer) FeeReport(ctx context.Context,
|
|||||||
_ *lnrpc.FeeReportRequest) (*lnrpc.FeeReportResponse, error) {
|
_ *lnrpc.FeeReportRequest) (*lnrpc.FeeReportResponse, error) {
|
||||||
|
|
||||||
channelGraph := r.server.graphDB
|
channelGraph := r.server.graphDB
|
||||||
selfNode, err := channelGraph.SourceNode()
|
selfNode, err := channelGraph.SourceNode(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1070,7 +1070,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
|
|||||||
MinProbability: routingConfig.MinRouteProbability,
|
MinProbability: routingConfig.MinRouteProbability,
|
||||||
}
|
}
|
||||||
|
|
||||||
sourceNode, err := dbs.GraphDB.SourceNode()
|
sourceNode, err := dbs.GraphDB.SourceNode(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("error getting source node: %w", err)
|
return nil, fmt.Errorf("error getting source node: %w", err)
|
||||||
}
|
}
|
||||||
@@ -3502,7 +3502,7 @@ func (s *server) updateAndBroadcastSelfNode(ctx context.Context,
|
|||||||
// Update the on-disk version of our announcement.
|
// Update the on-disk version of our announcement.
|
||||||
// Load and modify self node istead of creating anew instance so we
|
// Load and modify self node istead of creating anew instance so we
|
||||||
// don't risk overwriting any existing values.
|
// don't risk overwriting any existing values.
|
||||||
selfNode, err := s.graphDB.SourceNode()
|
selfNode, err := s.graphDB.SourceNode(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("unable to get current source node: %w", err)
|
return fmt.Errorf("unable to get current source node: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user