graph/db: thread context through NodeUpdatesInHorizon
This commit is contained in:
7 files changed
+20
-14
No files matched your search
@@ -181,7 +181,8 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
|
|||||||
// update within the horizon as well. We send these second to
|
// update within the horizon as well. We send these second to
|
||||||
// ensure that they follow any active channels they have.
|
// ensure that they follow any active channels they have.
|
||||||
nodeAnnsInHorizon := c.graph.NodeUpdatesInHorizon(
|
nodeAnnsInHorizon := c.graph.NodeUpdatesInHorizon(
|
||||||
startTime, endTime, graphdb.WithIterPublicNodesOnly(),
|
context.TODO(), startTime, endTime,
|
||||||
|
graphdb.WithIterPublicNodesOnly(),
|
||||||
)
|
)
|
||||||
for nodeAnn, err := range nodeAnnsInHorizon {
|
for nodeAnn, err := range nodeAnnsInHorizon {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -659,7 +659,7 @@ func BenchmarkGraphReadMethods(b *testing.B) {
|
|||||||
name: "NodeUpdatesInHorizon",
|
name: "NodeUpdatesInHorizon",
|
||||||
fn: func(b testing.TB, store Store) {
|
fn: func(b testing.TB, store Store) {
|
||||||
iter := store.NodeUpdatesInHorizon(
|
iter := store.NodeUpdatesInHorizon(
|
||||||
time.Unix(0, 0), time.Now(),
|
ctx, time.Unix(0, 0), time.Now(),
|
||||||
)
|
)
|
||||||
_, err := fn.CollectErr(iter)
|
_, err := fn.CollectErr(iter)
|
||||||
require.NoError(b, err)
|
require.NoError(b, err)
|
||||||
|
|||||||
+3
-2
@@ -633,10 +633,11 @@ func (c *ChannelGraph) ForEachNodeCacheable(ctx context.Context,
|
|||||||
|
|
||||||
// NodeUpdatesInHorizon returns all known lightning nodes with updates in the
|
// NodeUpdatesInHorizon returns all known lightning nodes with updates in the
|
||||||
// range.
|
// range.
|
||||||
func (c *ChannelGraph) NodeUpdatesInHorizon(startTime, endTime time.Time,
|
func (c *ChannelGraph) NodeUpdatesInHorizon(ctx context.Context,
|
||||||
|
startTime, endTime time.Time,
|
||||||
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
|
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
|
||||||
|
|
||||||
return c.db.NodeUpdatesInHorizon(startTime, endTime, opts...)
|
return c.db.NodeUpdatesInHorizon(ctx, startTime, endTime, opts...)
|
||||||
}
|
}
|
||||||
|
|
||||||
// HasV1Node determines if the graph has a vertex identified by the target node
|
// HasV1Node determines if the graph has a vertex identified by the target node
|
||||||
|
|||||||
+10
-6
@@ -2540,7 +2540,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
|
|||||||
// If we issue an arbitrary query before we insert any nodes into the
|
// If we issue an arbitrary query before we insert any nodes into the
|
||||||
// database, then we shouldn't get any results back.
|
// database, then we shouldn't get any results back.
|
||||||
nodeUpdatesIter := graph.NodeUpdatesInHorizon(
|
nodeUpdatesIter := graph.NodeUpdatesInHorizon(
|
||||||
time.Unix(999, 0), time.Unix(9999, 0),
|
ctx, time.Unix(999, 0), time.Unix(9999, 0),
|
||||||
)
|
)
|
||||||
nodeUpdates, err := fn.CollectErr(nodeUpdatesIter)
|
nodeUpdates, err := fn.CollectErr(nodeUpdatesIter)
|
||||||
require.NoError(t, err, "unable to query for node updates")
|
require.NoError(t, err, "unable to query for node updates")
|
||||||
@@ -2615,7 +2615,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
|
|||||||
}
|
}
|
||||||
for _, queryCase := range queryCases {
|
for _, queryCase := range queryCases {
|
||||||
iter := graph.NodeUpdatesInHorizon(
|
iter := graph.NodeUpdatesInHorizon(
|
||||||
queryCase.start, queryCase.end,
|
ctx, queryCase.start, queryCase.end,
|
||||||
)
|
)
|
||||||
|
|
||||||
resp, err := fn.CollectErr(iter)
|
resp, err := fn.CollectErr(iter)
|
||||||
@@ -2743,7 +2743,7 @@ func testNodeUpdatesWithBatchSize(t *testing.T, ctx context.Context,
|
|||||||
for _, tc := range testCases {
|
for _, tc := range testCases {
|
||||||
t.Run(tc.name, func(t *testing.T) {
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
iter := testGraph.NodeUpdatesInHorizon(
|
iter := testGraph.NodeUpdatesInHorizon(
|
||||||
tc.start, tc.end,
|
ctx, tc.start, tc.end,
|
||||||
WithNodeUpdateIterBatchSize(
|
WithNodeUpdateIterBatchSize(
|
||||||
batchSize,
|
batchSize,
|
||||||
),
|
),
|
||||||
@@ -2816,7 +2816,7 @@ func TestNodeUpdatesInHorizonEarlyTermination(t *testing.T) {
|
|||||||
for _, stopAt := range terminationPoints {
|
for _, stopAt := range terminationPoints {
|
||||||
t.Run(fmt.Sprintf("StopAt%d", stopAt), func(t *testing.T) {
|
t.Run(fmt.Sprintf("StopAt%d", stopAt), func(t *testing.T) {
|
||||||
iter := graph.NodeUpdatesInHorizon(
|
iter := graph.NodeUpdatesInHorizon(
|
||||||
startTime, startTime.Add(200*time.Hour),
|
ctx, startTime, startTime.Add(200*time.Hour),
|
||||||
WithNodeUpdateIterBatchSize(10),
|
WithNodeUpdateIterBatchSize(10),
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -4091,7 +4091,9 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
|
|||||||
// update time of our test node.
|
// update time of our test node.
|
||||||
startTime := time.Unix(9, 0)
|
startTime := time.Unix(9, 0)
|
||||||
endTime := node1.LastUpdate.Add(time.Minute)
|
endTime := node1.LastUpdate.Add(time.Minute)
|
||||||
nodesInHorizonIter := graph.NodeUpdatesInHorizon(startTime, endTime)
|
nodesInHorizonIter := graph.NodeUpdatesInHorizon(
|
||||||
|
ctx, startTime, endTime,
|
||||||
|
)
|
||||||
|
|
||||||
// We should only have a single node, and that node should exactly
|
// We should only have a single node, and that node should exactly
|
||||||
// match the node we just inserted.
|
// match the node we just inserted.
|
||||||
@@ -4107,7 +4109,9 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
|
|||||||
|
|
||||||
// Now that the node has been deleted, we'll again query the nodes in
|
// Now that the node has been deleted, we'll again query the nodes in
|
||||||
// the horizon. This time we should have no nodes at all.
|
// the horizon. This time we should have no nodes at all.
|
||||||
nodesInHorizonIter = graph.NodeUpdatesInHorizon(startTime, endTime)
|
nodesInHorizonIter = graph.NodeUpdatesInHorizon(
|
||||||
|
ctx, startTime, endTime,
|
||||||
|
)
|
||||||
nodesInHorizon, err = fn.CollectErr(nodesInHorizonIter)
|
nodesInHorizon, err = fn.CollectErr(nodesInHorizonIter)
|
||||||
require.NoError(t, err, "unable to fetch nodes in horizon")
|
require.NoError(t, err, "unable to fetch nodes in horizon")
|
||||||
require.Empty(t, nodesInHorizon)
|
require.Empty(t, nodesInHorizon)
|
||||||
|
|||||||
@@ -122,7 +122,7 @@ type Store interface { //nolint:interfacebloat
|
|||||||
// an update timestamp within the passed range. This method can be used
|
// an update timestamp within the passed range. This method can be used
|
||||||
// by two nodes to quickly determine if they have the same set of up to
|
// by two nodes to quickly determine if they have the same set of up to
|
||||||
// date node announcements.
|
// date node announcements.
|
||||||
NodeUpdatesInHorizon(startTime, endTime time.Time,
|
NodeUpdatesInHorizon(ctx context.Context, startTime, endTime time.Time,
|
||||||
opts ...IteratorOption) iter.Seq2[*models.Node, error]
|
opts ...IteratorOption) iter.Seq2[*models.Node, error]
|
||||||
|
|
||||||
// FetchNode attempts to look up a target node by its identity
|
// FetchNode attempts to look up a target node by its identity
|
||||||
|
|||||||
@@ -2627,7 +2627,7 @@ func (c *KVStore) fetchNextNodeBatch(
|
|||||||
|
|
||||||
// NodeUpdatesInHorizon returns all the known lightning node which have an
|
// NodeUpdatesInHorizon returns all the known lightning node which have an
|
||||||
// update timestamp within the passed range.
|
// update timestamp within the passed range.
|
||||||
func (c *KVStore) NodeUpdatesInHorizon(startTime,
|
func (c *KVStore) NodeUpdatesInHorizon(_ context.Context, startTime,
|
||||||
endTime time.Time,
|
endTime time.Time,
|
||||||
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
|
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
|
||||||
|
|
||||||
|
|||||||
@@ -610,7 +610,8 @@ func (s *SQLStore) SetSourceNode(ctx context.Context,
|
|||||||
// announcements.
|
// announcements.
|
||||||
//
|
//
|
||||||
// NOTE: This is part of the Store interface.
|
// NOTE: This is part of the Store interface.
|
||||||
func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time,
|
func (s *SQLStore) NodeUpdatesInHorizon(ctx context.Context,
|
||||||
|
startTime, endTime time.Time,
|
||||||
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
|
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
|
||||||
|
|
||||||
cfg := defaultIteratorConfig()
|
cfg := defaultIteratorConfig()
|
||||||
@@ -620,7 +621,6 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time,
|
|||||||
|
|
||||||
return func(yield func(*models.Node, error) bool) {
|
return func(yield func(*models.Node, error) bool) {
|
||||||
var (
|
var (
|
||||||
ctx = context.TODO()
|
|
||||||
lastUpdateTime sql.NullInt64
|
lastUpdateTime sql.NullInt64
|
||||||
lastPubKey = make([]byte, 33)
|
lastPubKey = make([]byte, 33)
|
||||||
hasMore = true
|
hasMore = true
|
||||||
|
|||||||
Reference in new issue
Block a user