diff --git a/discovery/ban.go b/discovery/ban.go index 5229c706c..0425948cb 100644 --- a/discovery/ban.go +++ b/discovery/ban.go @@ -1,6 +1,7 @@ package discovery import ( + "context" "errors" "math" "sync" @@ -55,10 +56,10 @@ type ClosedChannelTracker interface { type GraphCloser interface { // PutClosedScid marks a channel as closed so that we won't validate // channel announcements for it again. - PutClosedScid(lnwire.ShortChannelID) error + PutClosedScid(context.Context, lnwire.ShortChannelID) error // IsClosedScid checks if a short channel id is closed. - IsClosedScid(lnwire.ShortChannelID) (bool, error) + IsClosedScid(context.Context, lnwire.ShortChannelID) (bool, error) } // NodeInfoInquirier handles queries relating to specific nodes and channels @@ -88,16 +89,18 @@ func NewScidCloserMan(graph GraphCloser, // PutClosedScid marks scid as closed so the gossiper can ignore this channel // in the future. -func (s *ScidCloserMan) PutClosedScid(scid lnwire.ShortChannelID) error { - return s.graph.PutClosedScid(scid) +func (s *ScidCloserMan) PutClosedScid(ctx context.Context, + scid lnwire.ShortChannelID) error { + + return s.graph.PutClosedScid(ctx, scid) } // IsClosedScid checks whether scid is closed so that the gossiper can ignore // it. -func (s *ScidCloserMan) IsClosedScid(scid lnwire.ShortChannelID) (bool, - error) { +func (s *ScidCloserMan) IsClosedScid(ctx context.Context, + scid lnwire.ShortChannelID) (bool, error) { - return s.graph.IsClosedScid(scid) + return s.graph.IsClosedScid(ctx, scid) } // IsChannelPeer checks whether we have a channel with the peer. diff --git a/discovery/chan_series.go b/discovery/chan_series.go index ed64453a4..7ed9bbdef 100644 --- a/discovery/chan_series.go +++ b/discovery/chan_series.go @@ -115,7 +115,7 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash, // First, we'll query for all the set of channels that have an // update that falls within the specified horizon. chansInHorizon := c.graph.ChanUpdatesInHorizon( - startTime, endTime, + context.TODO(), startTime, endTime, ) for channel, err := range chansInHorizon { @@ -181,7 +181,8 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash, // update within the horizon as well. We send these second to // ensure that they follow any active channels they have. nodeAnnsInHorizon := c.graph.NodeUpdatesInHorizon( - startTime, endTime, graphdb.WithIterPublicNodesOnly(), + context.TODO(), startTime, endTime, + graphdb.WithIterPublicNodesOnly(), ) for nodeAnn, err := range nodeAnnsInHorizon { if err != nil { @@ -214,7 +215,9 @@ func (c *ChanSeries) FilterKnownChanIDs(_ chainhash.Hash, isZombieChan func(time.Time, time.Time) bool) ( []lnwire.ShortChannelID, error) { - newChanIDs, err := c.graph.FilterKnownChanIDs(superSet, isZombieChan) + newChanIDs, err := c.graph.FilterKnownChanIDs( + context.TODO(), superSet, isZombieChan, + ) if err != nil { return nil, err } @@ -240,7 +243,7 @@ func (c *ChanSeries) FilterChannelRange(_ chainhash.Hash, startHeight, error) { return c.graph.FilterChannelRange( - startHeight, endHeight, withTimestamps, + context.TODO(), startHeight, endHeight, withTimestamps, ) } @@ -260,7 +263,7 @@ func (c *ChanSeries) FetchChanAnns(chain chainhash.Hash, chanIDs = append(chanIDs, chanID.ToUint64()) } - channels, err := c.graph.FetchChanInfos(chanIDs) + channels, err := c.graph.FetchChanInfos(context.TODO(), chanIDs) if err != nil { return nil, err } @@ -353,7 +356,7 @@ func (c *ChanSeries) FetchChanUpdates(chain chainhash.Hash, shortChanID lnwire.ShortChannelID) ([]*lnwire.ChannelUpdate1, error) { chanInfo, e1, e2, err := c.graph.FetchChannelEdgesByID( - shortChanID.ToUint64(), + context.TODO(), shortChanID.ToUint64(), ) if err != nil { return nil, err diff --git a/discovery/gossiper.go b/discovery/gossiper.go index d2375cb1d..42fe9ac72 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -2724,7 +2724,7 @@ func (d *AuthenticatedGossiper) handleChanAnnouncement(ctx context.Context, // Check if the channel is already closed in which case we can ignore // it. - closed, err := d.cfg.ScidCloser.IsClosedScid(scid) + closed, err := d.cfg.ScidCloser.IsClosedScid(ctx, scid) if err != nil { log.Errorf("failed to check if scid %v is closed: %v", scid, err) @@ -2877,7 +2877,9 @@ func (d *AuthenticatedGossiper) handleChanAnnouncement(ctx context.Context, // expensive validation checks on it again. // TODO: Populate the ScidCloser by using closed // channel notifications. - dbErr := d.cfg.ScidCloser.PutClosedScid(scid) + dbErr := d.cfg.ScidCloser.PutClosedScid( + ctx, scid, + ) if dbErr != nil { log.Errorf("failed to mark scid(%v) "+ "as closed: %v", scid, dbErr) diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index 74409196b..0ee33ec0e 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -4793,7 +4793,7 @@ func TestChanAnnBanningNonChanPeer(t *testing.T) { // Check that the announcement's scid is marked as closed. isClosed, err := tCtx.gossiper.cfg.ScidCloser.IsClosedScid( - ca.ShortChannelID, + ctx, ca.ShortChannelID, ) require.Nil(t, err) require.True(t, isClosed) diff --git a/discovery/mock_test.go b/discovery/mock_test.go index 6bd93c29b..050c57096 100644 --- a/discovery/mock_test.go +++ b/discovery/mock_test.go @@ -1,6 +1,7 @@ package discovery import ( + "context" "errors" "net" "sync" @@ -176,7 +177,9 @@ func newMockScidCloser(channelPeer bool) *mockScidCloser { } } -func (m *mockScidCloser) PutClosedScid(scid lnwire.ShortChannelID) error { +func (m *mockScidCloser) PutClosedScid(_ context.Context, + scid lnwire.ShortChannelID) error { + m.Lock() m.m[scid] = struct{}{} m.Unlock() @@ -184,8 +187,8 @@ func (m *mockScidCloser) PutClosedScid(scid lnwire.ShortChannelID) error { return nil } -func (m *mockScidCloser) IsClosedScid(scid lnwire.ShortChannelID) (bool, - error) { +func (m *mockScidCloser) IsClosedScid(_ context.Context, + scid lnwire.ShortChannelID) (bool, error) { m.Lock() defer m.Unlock() diff --git a/graph/builder.go b/graph/builder.go index 5100066a3..391195fca 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -175,7 +175,7 @@ func (b *Builder) Start() error { // If the graph has never been pruned, or hasn't fully been created yet, // then we don't treat this as an explicit error. - if _, _, err := b.cfg.Graph.PruneTip(); err != nil { + if _, _, err := b.cfg.Graph.PruneTip(context.TODO()); err != nil { switch { case errors.Is(err, graphdb.ErrGraphNeverPruned): fallthrough @@ -185,7 +185,8 @@ func (b *Builder) Start() error { // the prune height to the current best height of the // chain backend. _, err = b.cfg.Graph.PruneGraph( - nil, bestHash, uint32(bestHeight), + context.TODO(), nil, bestHash, + uint32(bestHeight), ) if err != nil { return err @@ -230,7 +231,7 @@ func (b *Builder) Start() error { // FilteredChainView instance. We do this before, as otherwise // we may miss on-chain events as the filter hasn't properly // been applied. - channelView, err := b.cfg.Graph.ChannelView() + channelView, err := b.cfg.Graph.ChannelView(context.TODO()) if err != nil && !errors.Is( err, graphdb.ErrGraphNoEdgesFound, ) { @@ -268,7 +269,7 @@ func (b *Builder) Start() error { // Finally, before we proceed, we'll prune any unconnected nodes // from the graph in order to ensure we maintain a tight graph // of "useful" nodes. - err = b.cfg.Graph.PruneGraphNodes() + err = b.cfg.Graph.PruneGraphNodes(context.TODO()) if err != nil && !errors.Is(err, graphdb.ErrGraphNodesNotFound) { @@ -323,7 +324,7 @@ func (b *Builder) syncGraphWithChain() error { } b.bestHeight.Store(uint32(bestHeight)) - pruneHash, pruneHeight, err := b.cfg.Graph.PruneTip() + pruneHash, pruneHeight, err := b.cfg.Graph.PruneTip(context.TODO()) if err != nil { switch { // If the graph has never been pruned, or hasn't fully been @@ -365,12 +366,16 @@ func (b *Builder) syncGraphWithChain() error { "(hash=%v)", pruneHeight, pruneHash) // Prune the graph for every channel that was opened at height // >= pruneHeight. - _, err := b.cfg.Graph.DisconnectBlockAtHeight(pruneHeight) + _, err := b.cfg.Graph.DisconnectBlockAtHeight( + context.TODO(), pruneHeight, + ) if err != nil { return err } - pruneHash, pruneHeight, err = b.cfg.Graph.PruneTip() + pruneHash, pruneHeight, err = b.cfg.Graph.PruneTip( + context.TODO(), + ) switch { // If at this point the graph has never been pruned, we can exit // as this entails we are back to the point where it hasn't seen @@ -443,7 +448,7 @@ func (b *Builder) syncGraphWithChain() error { // With the spent outputs gathered, attempt to prune the channel graph, // also passing in the best hash+height so the prune tip can be updated. closedChans, err := b.cfg.Graph.PruneGraph( - spentOutputs, bestHash, uint32(bestHeight), + context.TODO(), spentOutputs, bestHash, uint32(bestHeight), ) if err != nil { return err @@ -574,7 +579,7 @@ func (b *Builder) pruneZombieChans() error { // the channel being closed and can prune it from our graph. if b.cfg.AssumeChannelValid { disabledChanIDs, err := b.cfg.Graph.DisabledChannelIDs( - lnwire.GossipVersion1, + context.TODO(), lnwire.GossipVersion1, ) if err != nil { return fmt.Errorf("unable to get disabled channels "+ @@ -582,7 +587,7 @@ func (b *Builder) pruneZombieChans() error { } disabledEdges, err := b.v1Graph.FetchChanInfos( - disabledChanIDs, + context.TODO(), disabledChanIDs, ) if err != nil { return fmt.Errorf("unable to fetch disabled channels "+ @@ -600,7 +605,9 @@ func (b *Builder) pruneZombieChans() error { startTime := time.Unix(0, 0) endTime := time.Now().Add(-1 * chanExpiry) - oldEdgesIter := b.cfg.Graph.ChanUpdatesInHorizon(startTime, endTime) + oldEdgesIter := b.cfg.Graph.ChanUpdatesInHorizon( + context.TODO(), startTime, endTime, + ) for u, err := range oldEdgesIter { if err != nil { @@ -628,7 +635,7 @@ func (b *Builder) pruneZombieChans() error { log.Tracef("Pruning zombie channel with ChannelID(%v)", chanID) } err := b.v1Graph.DeleteChannelEdges( - b.cfg.StrictZombiePruning, true, toPrune..., + context.TODO(), b.cfg.StrictZombiePruning, true, toPrune..., ) if err != nil { return fmt.Errorf("unable to delete zombie channels: %w", err) @@ -636,7 +643,7 @@ func (b *Builder) pruneZombieChans() error { // With the channels pruned, we'll also attempt to prune any nodes that // were a part of them. - err = b.cfg.Graph.PruneGraphNodes() + err = b.cfg.Graph.PruneGraphNodes(context.TODO()) if err != nil && !errors.Is(err, graphdb.ErrGraphNodesNotFound) { return fmt.Errorf("unable to prune graph nodes: %w", err) } @@ -682,7 +689,7 @@ func (b *Builder) networkHandler() { // Update the channel graph to reflect that this block // was disconnected. _, err := b.cfg.Graph.DisconnectBlockAtHeight( - blockHeight, + context.TODO(), blockHeight, ) if err != nil { log.Errorf("unable to prune graph with stale "+ @@ -855,8 +862,10 @@ func (b *Builder) updateGraphWithClosedChannels( // With the spent outputs gathered, attempt to prune the channel graph, // also passing in the hash+height of the block being pruned so the // prune tip can be updated. - chansClosed, err := b.cfg.Graph.PruneGraph(spentOutputs, - &chainUpdate.Hash, chainUpdate.Height) + chansClosed, err := b.cfg.Graph.PruneGraph( + context.TODO(), spentOutputs, &chainUpdate.Hash, + chainUpdate.Height, + ) if err != nil { log.Errorf("unable to prune routing table: %v", err) return err @@ -911,7 +920,9 @@ func (b *Builder) MarkZombieEdge(chanID uint64) error { // so we don't continue to request it. We use the "zero key" for both // node pubkeys so this edge can't be resurrected. var zeroKey [33]byte - err := b.cfg.Graph.MarkEdgeZombie(chanID, zeroKey, zeroKey) + err := b.cfg.Graph.MarkEdgeZombie( + context.TODO(), chanID, zeroKey, zeroKey, + ) if err != nil { return fmt.Errorf("unable to mark spent chan(id=%v) as a "+ "zombie: %w", chanID, err) @@ -1046,7 +1057,7 @@ func (b *Builder) addEdge(ctx context.Context, edge *models.ChannelEdgeInfo, // Prior to processing the announcement we first check if we // already know of this channel, if so, then we can exit early. exists, isZombie, err := b.cfg.Graph.HasChannelEdge( - edge.Version, edge.ChannelID, + ctx, edge.Version, edge.ChannelID, ) if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) { return fmt.Errorf("unable to check for edge existence: %w", @@ -1147,7 +1158,7 @@ func (b *Builder) updateEdge(ctx context.Context, defer b.channelEdgeMtx.Unlock(policy.ChannelID) edge1Timestamp, edge2Timestamp, exists, isZombie, err := - b.cfg.Graph.HasV1ChannelEdge(policy.ChannelID) + b.cfg.Graph.HasV1ChannelEdge(ctx, policy.ChannelID) if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) { return fmt.Errorf("unable to check for edge existence: %w", err) } @@ -1255,7 +1266,9 @@ func (b *Builder) GetChannelByID(chanID lnwire.ShortChannelID) ( *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { - return b.cfg.Graph.FetchChannelEdgesByID(chanID.ToUint64()) + return b.cfg.Graph.FetchChannelEdgesByID( + context.TODO(), chanID.ToUint64(), + ) } // FetchNode attempts to look up a target node by its identity public @@ -1299,7 +1312,7 @@ func (b *Builder) ForAllOutgoingChannels(ctx context.Context, func (b *Builder) AddProof(chanID lnwire.ShortChannelID, proof *models.ChannelAuthProof) error { - return b.cfg.Graph.AddEdgeProof(chanID, proof) + return b.cfg.Graph.AddEdgeProof(context.TODO(), chanID, proof) } // IsStaleNode returns true if the graph source has a node announcement for the @@ -1325,7 +1338,7 @@ func (b *Builder) IsStaleNode(ctx context.Context, node route.Vertex, // // NOTE: This method is part of the ChannelGraphSource interface. func (b *Builder) IsPublicNode(node route.Vertex) (bool, error) { - return b.v1Graph.IsPublicNode(node) + return b.v1Graph.IsPublicNode(context.TODO(), node) } // IsKnownEdge returns true if the graph source already knows of the passed @@ -1334,7 +1347,7 @@ func (b *Builder) IsPublicNode(node route.Vertex) (bool, error) { // NOTE: This method is part of the ChannelGraphSource interface. func (b *Builder) IsKnownEdge(chanID lnwire.ShortChannelID) bool { exists, isZombie, _ := b.cfg.Graph.HasChannelEdge( - lnwire.GossipVersion1, chanID.ToUint64(), + context.TODO(), lnwire.GossipVersion1, chanID.ToUint64(), ) return exists || isZombie @@ -1346,7 +1359,7 @@ func (b *Builder) IsKnownEdge(chanID lnwire.ShortChannelID) bool { // NOTE: This method is part of the ChannelGraphSource interface. func (b *Builder) IsZombieEdge(chanID lnwire.ShortChannelID) (bool, error) { _, isZombie, err := b.cfg.Graph.HasChannelEdge( - lnwire.GossipVersion1, chanID.ToUint64(), + context.TODO(), lnwire.GossipVersion1, chanID.ToUint64(), ) return isZombie, err @@ -1360,7 +1373,9 @@ func (b *Builder) IsStaleEdgePolicy(chanID lnwire.ShortChannelID, timestamp time.Time, flags lnwire.ChanUpdateChanFlags) bool { edge1Timestamp, edge2Timestamp, exists, isZombie, err := - b.cfg.Graph.HasV1ChannelEdge(chanID.ToUint64()) + b.cfg.Graph.HasV1ChannelEdge( + context.TODO(), chanID.ToUint64(), + ) if err != nil { log.Debugf("Check stale edge policy got error: %v", err) return false @@ -1413,5 +1428,7 @@ func (b *Builder) IsStaleEdgePolicy(chanID lnwire.ShortChannelID, // // NOTE: This method is part of the ChannelGraphSource interface. func (b *Builder) MarkEdgeLive(chanID lnwire.ShortChannelID) error { - return b.cfg.Graph.MarkEdgeLive(chanID.ToUint64()) + return b.cfg.Graph.MarkEdgeLive( + context.TODO(), chanID.ToUint64(), + ) } diff --git a/graph/builder_test.go b/graph/builder_test.go index e561d25b1..2f9dc41fe 100644 --- a/graph/builder_test.go +++ b/graph/builder_test.go @@ -311,12 +311,12 @@ func TestWakeUpOnStaleBranch(t *testing.T) { require.NoError(t, ctx.builder.AddEdge(ctxb, edge2)) // Check that the fundingTxs are in the graph db. - has, isZombie, err := ctx.graph.HasChannelEdge(chanID1) + has, isZombie, err := ctx.graph.HasChannelEdge(t.Context(), chanID1) require.NoError(t, err) require.True(t, has) require.False(t, isZombie) - has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) + has, isZombie, err = ctx.graph.HasChannelEdge(t.Context(), chanID2) require.NoError(t, err) require.True(t, has) require.False(t, isZombie) @@ -363,12 +363,12 @@ func TestWakeUpOnStaleBranch(t *testing.T) { // The channel with chanID2 should not be in the database anymore, // since it is not confirmed on the longest chain. chanID1 should // still be. - has, isZombie, err = ctx.graph.HasChannelEdge(chanID1) + has, isZombie, err = ctx.graph.HasChannelEdge(t.Context(), chanID1) require.NoError(t, err) require.True(t, has) require.False(t, isZombie) - has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) + has, isZombie, err = ctx.graph.HasChannelEdge(t.Context(), chanID2) require.NoError(t, err) require.False(t, has) require.False(t, isZombie) @@ -482,12 +482,12 @@ func TestDisconnectedBlocks(t *testing.T) { require.NoError(t, ctx.builder.AddEdge(ctxb, edge2)) // Check that the fundingTxs are in the graph db. - has, isZombie, err := ctx.graph.HasChannelEdge(chanID1) + has, isZombie, err := ctx.graph.HasChannelEdge(t.Context(), chanID1) require.NoError(t, err) require.True(t, has) require.False(t, isZombie) - has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) + has, isZombie, err = ctx.graph.HasChannelEdge(t.Context(), chanID2) require.NoError(t, err) require.True(t, has) require.False(t, isZombie) @@ -523,12 +523,12 @@ func TestDisconnectedBlocks(t *testing.T) { // chanID2 should not be in the database anymore, since it is not // confirmed on the longest chain. chanID1 should still be. - has, isZombie, err = ctx.graph.HasChannelEdge(chanID1) + has, isZombie, err = ctx.graph.HasChannelEdge(t.Context(), chanID1) require.NoError(t, err) require.True(t, has) require.False(t, isZombie) - has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) + has, isZombie, err = ctx.graph.HasChannelEdge(t.Context(), chanID2) require.NoError(t, err) require.False(t, has) require.False(t, isZombie) @@ -594,7 +594,7 @@ func TestChansClosedOfflinePruneGraph(t *testing.T) { // The router should now be aware of the channel we created above. hasChan, isZombie, err := ctx.graph.HasChannelEdge( - chanID1.ToUint64(), + t.Context(), chanID1.ToUint64(), ) require.NoError(t, err) require.True(t, hasChan) @@ -662,7 +662,7 @@ func TestChansClosedOfflinePruneGraph(t *testing.T) { // At this point, the channel that was pruned should no longer be known // by the router. hasChan, isZombie, err = ctx.graph.HasChannelEdge( - chanID1.ToUint64(), + t.Context(), chanID1.ToUint64(), ) require.NoError(t, err) require.False(t, hasChan) @@ -1611,7 +1611,7 @@ func assertChannelsPruned(t *testing.T, graph *graphdb.VersionedGraph, for _, channel := range channels { _, shouldPrune := pruned[channel.ChannelID] exists, isZombie, err := graph.HasChannelEdge( - channel.ChannelID, + t.Context(), channel.ChannelID, ) require.NoError(t, err) if shouldPrune { diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 2887b7d22..229aad407 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -659,7 +659,7 @@ func BenchmarkGraphReadMethods(b *testing.B) { name: "NodeUpdatesInHorizon", fn: func(b testing.TB, store Store) { iter := store.NodeUpdatesInHorizon( - time.Unix(0, 0), time.Now(), + ctx, time.Unix(0, 0), time.Now(), ) _, err := fn.CollectErr(iter) require.NoError(b, err) @@ -709,7 +709,7 @@ func BenchmarkGraphReadMethods(b *testing.B) { name: "ChanUpdatesInHorizon", fn: func(b testing.TB, store Store) { iter := store.ChanUpdatesInHorizon( - time.Unix(0, 0), time.Now(), + ctx, time.Unix(0, 0), time.Now(), ) _, err := fn.CollectErr(iter) require.NoError(b, err) diff --git a/graph/db/graph.go b/graph/db/graph.go index 283f7d536..d60ca24cf 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -73,15 +73,17 @@ func (c *ChannelGraph) Start() error { log.Debugf("ChannelGraph starting") defer log.Debug("ChannelGraph started") + ctx := context.TODO() + if c.graphCache != nil { - if err := c.populateCache(context.TODO()); err != nil { + if err := c.populateCache(ctx); err != nil { return fmt.Errorf("could not populate the graph "+ "cache: %w", err) } } c.wg.Add(1) - go c.handleTopologySubscriptions() + go c.handleTopologySubscriptions(ctx) return nil } @@ -106,7 +108,7 @@ func (c *ChannelGraph) Stop() error { // synchronously. // // NOTE: this MUST be run in a goroutine. -func (c *ChannelGraph) handleTopologySubscriptions() { +func (c *ChannelGraph) handleTopologySubscriptions(ctx context.Context) { defer c.wg.Done() for { @@ -118,7 +120,7 @@ func (c *ChannelGraph) handleTopologySubscriptions() { // synchronously so that we can guarantee the order of // notification delivery. c.wg.Add(1) - go c.handleTopologyUpdate(update) + go c.handleTopologyUpdate(ctx, update) // TODO(roasbeef): remove all unconnected vertexes // after N blocks pass with no corresponding @@ -149,6 +151,9 @@ func (c *ChannelGraph) handleTopologySubscriptions() { exit: make(chan struct{}), }) + case <-ctx.Done(): + return + case <-c.quit: return } @@ -185,7 +190,7 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error { } err = c.db.ForEachChannelCacheable( - v, func(info *models.CachedEdgeInfo, + ctx, v, func(info *models.CachedEdgeInfo, policy1, policy2 *models.CachedEdgePolicy) error { @@ -217,8 +222,9 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error { // Unknown policies are passed into the callback as nil values. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex, - cb func(channel *DirectedChannel) error, reset func()) error { +func (c *ChannelGraph) ForEachNodeDirectedChannel(ctx context.Context, + node route.Vertex, cb func(channel *DirectedChannel) error, + reset func()) error { if c.graphCache != nil { return c.graphCache.ForEachChannel(node, cb) @@ -227,7 +233,9 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex, // TODO(elle): once the no-cache path needs to support // pathfinding across gossip versions, this should iterate // across all versions rather than defaulting to v1. - return c.db.ForEachNodeDirectedChannel(gossipV1, node, cb, reset) + return c.db.ForEachNodeDirectedChannel( + ctx, gossipV1, node, cb, reset, + ) } // FetchNodeFeatures returns the features of the given node. If no features are @@ -236,28 +244,28 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex, // features instead of the database. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (c *ChannelGraph) FetchNodeFeatures(node route.Vertex) ( - *lnwire.FeatureVector, error) { +func (c *ChannelGraph) FetchNodeFeatures(ctx context.Context, + node route.Vertex) (*lnwire.FeatureVector, error) { if c.graphCache != nil { return c.graphCache.GetFeatures(node), nil } - return c.db.FetchNodeFeatures(lnwire.GossipVersion1, node) + return c.db.FetchNodeFeatures(ctx, lnwire.GossipVersion1, node) } // GraphSession will provide the call-back with access to a NodeTraverser // instance which can be used to perform queries against the channel graph. If // the graph cache is not enabled, then the call-back will be provided with // access to the graph via a consistent read-only transaction. -func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error, - reset func()) error { +func (c *ChannelGraph) GraphSession(ctx context.Context, + cb func(graph NodeTraverser) error, reset func()) error { if c.graphCache != nil { return cb(c) } - return c.db.GraphSession(cb, reset) + return c.db.GraphSession(ctx, cb, reset) } // ForEachNodeCached iterates through all the stored vertices/nodes in the @@ -340,8 +348,8 @@ func (c *ChannelGraph) AddChannelEdge(ctx context.Context, // MarkEdgeLive clears an edge from our zombie index, deeming it as live. // If the cache is enabled, the edge will be added back to the graph cache if // we still have a record of this channel in the DB. -func (c *ChannelGraph) MarkEdgeLive(chanID uint64) error { - err := c.db.MarkEdgeLive(chanID) +func (c *ChannelGraph) MarkEdgeLive(ctx context.Context, chanID uint64) error { + err := c.db.MarkEdgeLive(ctx, chanID) if err != nil { return err } @@ -350,7 +358,7 @@ func (c *ChannelGraph) MarkEdgeLive(chanID uint64) error { // We need to add the channel back into our graph cache, // otherwise we won't use it for path finding. infos, err := c.db.FetchChanInfos( - lnwire.GossipVersion1, []uint64{chanID}, + ctx, lnwire.GossipVersion1, []uint64{chanID}, ) if err != nil { return err @@ -386,11 +394,11 @@ func (c *ChannelGraph) MarkEdgeLive(chanID uint64) error { // that we require the node that failed to send the fresh update to be the one // that resurrects the channel from its zombie state. The markZombie bool // denotes whether to mark the channel as a zombie. -func (c *ChannelGraph) DeleteChannelEdges(strictZombiePruning, markZombie bool, - chanIDs ...uint64) error { +func (c *ChannelGraph) DeleteChannelEdges(ctx context.Context, + strictZombiePruning, markZombie bool, chanIDs ...uint64) error { infos, err := c.db.DeleteChannelEdges( - lnwire.GossipVersion1, strictZombiePruning, markZombie, + ctx, lnwire.GossipVersion1, strictZombiePruning, markZombie, chanIDs..., ) if err != nil { @@ -416,10 +424,10 @@ func (c *ChannelGraph) DeleteChannelEdges(strictZombiePruning, markZombie bool, // set to the last prune height valid for the remaining chain. // Channels that were removed from the graph resulting from the // disconnected block are returned. -func (c *ChannelGraph) DisconnectBlockAtHeight(height uint32) ( - []*models.ChannelEdgeInfo, error) { +func (c *ChannelGraph) DisconnectBlockAtHeight(ctx context.Context, + height uint32) ([]*models.ChannelEdgeInfo, error) { - edges, err := c.db.DisconnectBlockAtHeight(height) + edges, err := c.db.DisconnectBlockAtHeight(ctx, height) if err != nil { return nil, err } @@ -443,12 +451,13 @@ func (c *ChannelGraph) DisconnectBlockAtHeight(height uint32) ( // prune the graph is stored so callers can ensure the graph is fully in sync // with the current UTXO state. A slice of channels that have been closed by // the target block are returned if the function succeeds without error. -func (c *ChannelGraph) PruneGraph(spentOutputs []*wire.OutPoint, +func (c *ChannelGraph) PruneGraph(ctx context.Context, + spentOutputs []*wire.OutPoint, blockHash *chainhash.Hash, blockHeight uint32) ( []*models.ChannelEdgeInfo, error) { edges, nodes, err := c.db.PruneGraph( - spentOutputs, blockHash, blockHeight, + ctx, spentOutputs, blockHash, blockHeight, ) if err != nil { return nil, err @@ -491,8 +500,8 @@ func (c *ChannelGraph) PruneGraph(spentOutputs []*wire.OutPoint, // any nodes from the channel graph that are currently unconnected. This ensure // that we only maintain a graph of reachable nodes. In the event that a pruned // node gains more channels, it will be re-added back to the graph. -func (c *ChannelGraph) PruneGraphNodes() error { - nodes, err := c.db.PruneGraphNodes() +func (c *ChannelGraph) PruneGraphNodes(ctx context.Context) error { + nodes, err := c.db.PruneGraphNodes(ctx) if err != nil { return err } @@ -511,10 +520,11 @@ func (c *ChannelGraph) PruneGraphNodes() error { // words, we perform a set difference of our set of chan ID's and the ones // passed in. This method can be used by callers to determine the set of // channels another peer knows of that we don't. -func (c *ChannelGraph) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo, +func (c *ChannelGraph) FilterKnownChanIDs(ctx context.Context, + chansInfo []ChannelUpdateInfo, isZombieChan func(time.Time, time.Time) bool) ([]uint64, error) { - unknown, knownZombies, err := c.db.FilterKnownChanIDs(chansInfo) + unknown, knownZombies, err := c.db.FilterKnownChanIDs(ctx, chansInfo) if err != nil { return nil, err } @@ -544,7 +554,7 @@ func (c *ChannelGraph) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo, // alive, and we let it be added to the set of IDs to query our // peer for. err := c.db.MarkEdgeLive( - info.ShortChannelID.ToUint64(), + ctx, info.ShortChannelID.ToUint64(), ) // Since there is a chance that the edge could have been marked // as "live" between the FilterKnownChanIDs call and the @@ -561,10 +571,10 @@ func (c *ChannelGraph) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo, // MarkEdgeZombie attempts to mark a channel identified by its channel ID as a // zombie. This method is used on an ad-hoc basis, when channels need to be // marked as zombies outside the normal pruning cycle. -func (c *ChannelGraph) MarkEdgeZombie(chanID uint64, +func (c *ChannelGraph) MarkEdgeZombie(ctx context.Context, chanID uint64, pubKey1, pubKey2 [33]byte) error { - err := c.db.MarkEdgeZombie(chanID, pubKey1, pubKey2) + err := c.db.MarkEdgeZombie(ctx, chanID, pubKey1, pubKey2) if err != nil { return err } @@ -633,10 +643,11 @@ func (c *ChannelGraph) ForEachNodeCacheable(ctx context.Context, // NodeUpdatesInHorizon returns all known lightning nodes with updates in the // 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] { - 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 @@ -648,8 +659,10 @@ func (c *ChannelGraph) HasV1Node(ctx context.Context, } // IsPublicNode determines whether the node is seen as public in the graph. -func (c *ChannelGraph) IsPublicNode(pubKey [33]byte) (bool, error) { - return c.db.IsPublicNode(lnwire.GossipVersion1, pubKey) +func (c *ChannelGraph) IsPublicNode(ctx context.Context, + pubKey [33]byte) (bool, error) { + + return c.db.IsPublicNode(ctx, lnwire.GossipVersion1, pubKey) } // ForEachChannel iterates through all channel edges stored within the graph. @@ -662,31 +675,32 @@ func (c *ChannelGraph) ForEachChannel(ctx context.Context, } // DisabledChannelIDs returns the channel ids of disabled channels. -func (c *ChannelGraph) DisabledChannelIDs(v lnwire.GossipVersion) ( +func (c *ChannelGraph) DisabledChannelIDs(ctx context.Context, + v lnwire.GossipVersion) ( []uint64, error) { - return c.db.DisabledChannelIDs(v) + return c.db.DisabledChannelIDs(ctx, v) } // HasV1ChannelEdge returns true if the database knows of a channel edge. -func (c *ChannelGraph) HasV1ChannelEdge(chanID uint64) (time.Time, - time.Time, bool, bool, error) { +func (c *ChannelGraph) HasV1ChannelEdge(ctx context.Context, + chanID uint64) (time.Time, time.Time, bool, bool, error) { - return c.db.HasV1ChannelEdge(chanID) + return c.db.HasV1ChannelEdge(ctx, chanID) } // HasChannelEdge returns true if the database knows of a channel edge. -func (c *ChannelGraph) HasChannelEdge(v lnwire.GossipVersion, - chanID uint64) (bool, bool, error) { +func (c *ChannelGraph) HasChannelEdge(ctx context.Context, + v lnwire.GossipVersion, chanID uint64) (bool, bool, error) { - return c.db.HasChannelEdge(v, chanID) + return c.db.HasChannelEdge(ctx, v, chanID) } // AddEdgeProof sets the proof of an existing edge in the graph database. -func (c *ChannelGraph) AddEdgeProof(chanID lnwire.ShortChannelID, - proof *models.ChannelAuthProof) error { +func (c *ChannelGraph) AddEdgeProof(ctx context.Context, + chanID lnwire.ShortChannelID, proof *models.ChannelAuthProof) error { - return c.db.AddEdgeProof(chanID, proof) + return c.db.AddEdgeProof(ctx, chanID, proof) } // HighestChanID returns the "highest" known channel ID in the channel graph. @@ -698,72 +712,82 @@ func (c *ChannelGraph) HighestChanID(ctx context.Context, // ChanUpdatesInHorizon returns all known channel edges with updates in the // horizon. -func (c *ChannelGraph) ChanUpdatesInHorizon(startTime, endTime time.Time, +func (c *ChannelGraph) ChanUpdatesInHorizon(ctx context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[ChannelEdge, error] { - return c.db.ChanUpdatesInHorizon(startTime, endTime, opts...) + return c.db.ChanUpdatesInHorizon(ctx, startTime, endTime, opts...) } // FilterChannelRange returns channel IDs within the passed block height range. -func (c *ChannelGraph) FilterChannelRange(startHeight, endHeight uint32, - withTimestamps bool) ([]BlockChannelRange, error) { +func (c *ChannelGraph) FilterChannelRange(ctx context.Context, + startHeight, endHeight uint32, withTimestamps bool) ( + []BlockChannelRange, error) { - return c.db.FilterChannelRange(startHeight, endHeight, withTimestamps) + return c.db.FilterChannelRange( + ctx, startHeight, endHeight, withTimestamps, + ) } // FetchChanInfos returns the set of channel edges for the passed channel IDs. -func (c *ChannelGraph) FetchChanInfos(v lnwire.GossipVersion, - chanIDs []uint64) ([]ChannelEdge, error) { +func (c *ChannelGraph) FetchChanInfos(ctx context.Context, + v lnwire.GossipVersion, chanIDs []uint64) ([]ChannelEdge, error) { - return c.db.FetchChanInfos(v, chanIDs) + return c.db.FetchChanInfos(ctx, v, chanIDs) } // FetchChannelEdgesByOutpoint attempts to lookup directed edges by funding // outpoint. -func (c *ChannelGraph) FetchChannelEdgesByOutpoint(op *wire.OutPoint) ( +func (c *ChannelGraph) FetchChannelEdgesByOutpoint(ctx context.Context, + op *wire.OutPoint) ( *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { return c.db.FetchChannelEdgesByOutpoint( - lnwire.GossipVersion1, op, + ctx, lnwire.GossipVersion1, op, ) } // FetchChannelEdgesByID attempts to lookup directed edges by channel ID. -func (c *ChannelGraph) FetchChannelEdgesByID(chanID uint64) ( +func (c *ChannelGraph) FetchChannelEdgesByID(ctx context.Context, + chanID uint64) ( *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { return c.db.FetchChannelEdgesByID( - lnwire.GossipVersion1, chanID, + ctx, lnwire.GossipVersion1, chanID, ) } // ChannelView returns the verifiable edge information for each active channel. -func (c *ChannelGraph) ChannelView() ([]EdgePoint, error) { - return c.db.ChannelView() +func (c *ChannelGraph) ChannelView(ctx context.Context) ([]EdgePoint, error) { + return c.db.ChannelView(ctx) } // IsZombieEdge returns whether the edge is considered zombie. -func (c *ChannelGraph) IsZombieEdge(chanID uint64) (bool, [33]byte, [33]byte, - error) { +func (c *ChannelGraph) IsZombieEdge(ctx context.Context, + chanID uint64) (bool, [33]byte, [33]byte, error) { - return c.db.IsZombieEdge(lnwire.GossipVersion1, chanID) + return c.db.IsZombieEdge(ctx, lnwire.GossipVersion1, chanID) } // NumZombies returns the current number of zombie channels in the graph. -func (c *ChannelGraph) NumZombies() (uint64, error) { - return c.db.NumZombies() +func (c *ChannelGraph) NumZombies(ctx context.Context) (uint64, error) { + return c.db.NumZombies(ctx) } // PutClosedScid stores a SCID for a closed channel in the database. -func (c *ChannelGraph) PutClosedScid(scid lnwire.ShortChannelID) error { - return c.db.PutClosedScid(scid) +func (c *ChannelGraph) PutClosedScid(ctx context.Context, + scid lnwire.ShortChannelID) error { + + return c.db.PutClosedScid(ctx, scid) } // IsClosedScid checks whether a channel identified by the scid is closed. -func (c *ChannelGraph) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) { - return c.db.IsClosedScid(scid) +func (c *ChannelGraph) IsClosedScid(ctx context.Context, + scid lnwire.ShortChannelID) (bool, error) { + + return c.db.IsClosedScid(ctx, scid) } // SetSourceNode sets the source node within the graph database. @@ -774,8 +798,10 @@ func (c *ChannelGraph) SetSourceNode(ctx context.Context, } // PruneTip returns the block height and hash of the latest pruning block. -func (c *ChannelGraph) PruneTip() (*chainhash.Hash, uint32, error) { - return c.db.PruneTip() +func (c *ChannelGraph) PruneTip(ctx context.Context) (*chainhash.Hash, + uint32, error) { + + return c.db.PruneTip(ctx) } // VersionedGraph is a wrapper around ChannelGraph that will call underlying @@ -803,27 +829,29 @@ func (c *VersionedGraph) FetchNode(ctx context.Context, } // FetchChannelEdgesByID attempts to lookup directed edges by channel ID. -func (c *VersionedGraph) FetchChannelEdgesByID(chanID uint64) ( +func (c *VersionedGraph) FetchChannelEdgesByID(ctx context.Context, + chanID uint64) ( *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { - return c.db.FetchChannelEdgesByID(c.v, chanID) + return c.db.FetchChannelEdgesByID(ctx, c.v, chanID) } // FetchChannelEdgesByOutpoint attempts to lookup directed edges by funding // outpoint. -func (c *VersionedGraph) FetchChannelEdgesByOutpoint(op *wire.OutPoint) ( +func (c *VersionedGraph) FetchChannelEdgesByOutpoint(ctx context.Context, + op *wire.OutPoint) ( *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { - return c.db.FetchChannelEdgesByOutpoint(c.v, op) + return c.db.FetchChannelEdgesByOutpoint(ctx, c.v, op) } // IsZombieEdge returns whether the edge is considered zombie for this version. -func (c *VersionedGraph) IsZombieEdge(chanID uint64) (bool, [33]byte, - [33]byte, error) { +func (c *VersionedGraph) IsZombieEdge(ctx context.Context, + chanID uint64) (bool, [33]byte, [33]byte, error) { - return c.db.IsZombieEdge(c.v, chanID) + return c.db.IsZombieEdge(ctx, c.v, chanID) } // AddrsForNode returns all known addresses for the target node public key. @@ -880,11 +908,11 @@ func (c *VersionedGraph) SourceNode(ctx context.Context) (*models.Node, // that we require the node that failed to send the fresh update to be the one // that resurrects the channel from its zombie state. The markZombie bool // denotes whether to mark the channel as a zombie. -func (c *VersionedGraph) DeleteChannelEdges(strictZombiePruning, - markZombie bool, chanIDs ...uint64) error { +func (c *VersionedGraph) DeleteChannelEdges(ctx context.Context, + strictZombiePruning, markZombie bool, chanIDs ...uint64) error { infos, err := c.db.DeleteChannelEdges( - c.v, strictZombiePruning, markZombie, chanIDs..., + ctx, c.v, strictZombiePruning, markZombie, chanIDs..., ) if err != nil { return err @@ -906,8 +934,10 @@ func (c *VersionedGraph) DeleteChannelEdges(strictZombiePruning, // passed channel ID and this graph's gossip version, and false otherwise. If it // is not found, then the zombie index is checked and its result is returned as // the second boolean. -func (c *VersionedGraph) HasChannelEdge(chanID uint64) (bool, bool, error) { - return c.db.HasChannelEdge(c.v, chanID) +func (c *VersionedGraph) HasChannelEdge(ctx context.Context, + chanID uint64) (bool, bool, error) { + + return c.db.HasChannelEdge(ctx, c.v, chanID) } // ForEachSourceNodeChannel iterates through all channels of the source node. @@ -944,23 +974,25 @@ func (c *VersionedGraph) ForEachNodeCacheable(ctx context.Context, } // ForEachChannelCacheable iterates through all channel edges for the cache. -func (c *VersionedGraph) ForEachChannelCacheable( +func (c *VersionedGraph) ForEachChannelCacheable(ctx context.Context, cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, reset func()) error { - return c.db.ForEachChannelCacheable(c.v, cb, reset) + return c.db.ForEachChannelCacheable(ctx, c.v, cb, reset) } // DisabledChannelIDs returns the channel ids of disabled channels. -func (c *VersionedGraph) DisabledChannelIDs() ([]uint64, error) { - return c.db.DisabledChannelIDs(c.v) +func (c *VersionedGraph) DisabledChannelIDs( + ctx context.Context) ([]uint64, error) { + + return c.db.DisabledChannelIDs(ctx, c.v) } // FetchChanInfos returns the set of channel edges for the passed channel IDs. -func (c *VersionedGraph) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, - error) { +func (c *VersionedGraph) FetchChanInfos(ctx context.Context, + chanIDs []uint64) ([]ChannelEdge, error) { - return c.db.FetchChanInfos(c.v, chanIDs) + return c.db.FetchChanInfos(ctx, c.v, chanIDs) } // HighestChanID returns the "highest" known channel ID in the channel graph. @@ -969,13 +1001,17 @@ func (c *VersionedGraph) HighestChanID(ctx context.Context) (uint64, error) { } // ChannelID attempts to lookup the 8-byte compact channel ID. -func (c *VersionedGraph) ChannelID(chanPoint *wire.OutPoint) (uint64, error) { - return c.db.ChannelID(c.v, chanPoint) +func (c *VersionedGraph) ChannelID(ctx context.Context, + chanPoint *wire.OutPoint) (uint64, error) { + + return c.db.ChannelID(ctx, c.v, chanPoint) } // IsPublicNode determines whether the node is seen as public in the graph. -func (c *VersionedGraph) IsPublicNode(pubKey [33]byte) (bool, error) { - return c.db.IsPublicNode(c.v, pubKey) +func (c *VersionedGraph) IsPublicNode(ctx context.Context, + pubKey [33]byte) (bool, error) { + + return c.db.IsPublicNode(ctx, c.v, pubKey) } // MakeTestGraph creates a new instance of the ChannelGraph for testing diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index aa0e02b3e..ebe958bc2 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -298,13 +298,13 @@ func testNodeInsertionAndDeletion(t *testing.T, v lnwire.GossipVersion) { // Check that the node's features are fetched correctly. This check // will use the graph cache to fetch the features. - features, err := graph.FetchNodeFeatures(node.PubKeyBytes) + features, err := graph.FetchNodeFeatures(ctx, node.PubKeyBytes) require.NoError(t, err) require.Equal(t, testFeatures, features) // Check that the node's features are fetched correctly. This check // will check the database directly. - features, err = graph.FetchNodeFeatures(node.PubKeyBytes) + features, err = graph.FetchNodeFeatures(ctx, node.PubKeyBytes) require.NoError(t, err) require.Equal(t, testFeatures, features) @@ -643,7 +643,7 @@ func testEdgeInsertionDeletion(t *testing.T, v lnwire.GossipVersion) { // Ensure that both policies are returned as unknown (nil) and that // the edge info round-trips correctly. - dbEdge, e1, e2, err := graph.FetchChannelEdgesByID(chanID) + dbEdge, e1, e2, err := graph.FetchChannelEdgesByID(ctx, chanID) require.NoError(t, err) require.Nil(t, e1) require.Nil(t, e2) @@ -693,34 +693,38 @@ func testEdgeInsertionDeletion(t *testing.T, v lnwire.GossipVersion) { } // Also verify fetching by outpoint returns the same data. - dbEdge2, _, _, err := graph.FetchChannelEdgesByOutpoint(&outpoint) + dbEdge2, _, _, err := graph.FetchChannelEdgesByOutpoint( + ctx, &outpoint, + ) require.NoError(t, err) require.Equal(t, dbEdge.ChannelID, dbEdge2.ChannelID) // Next, attempt to delete the edge from the database, again this // should proceed without any issues. - require.NoError(t, graph.DeleteChannelEdges(false, true, chanID)) + require.NoError(t, graph.DeleteChannelEdges( + ctx, false, true, chanID, + )) assertNoEdge(t, graph.ChannelGraph, chanID) // Ensure that any query attempts to lookup the delete channel edge are // properly deleted. - _, _, _, err = graph.FetchChannelEdgesByOutpoint(&outpoint) + _, _, _, err = graph.FetchChannelEdgesByOutpoint(ctx, &outpoint) require.ErrorIs(t, err, ErrEdgeNotFound) // Assert that if the edge is a zombie, then FetchChannelEdgesByID // still returns a populated models.ChannelEdgeInfo as its comment // description promises. - edge, _, _, err := graph.FetchChannelEdgesByID(chanID) + edge, _, _, err := graph.FetchChannelEdgesByID(ctx, chanID) require.ErrorIs(t, err, ErrZombieEdge) require.NotNil(t, edge) - isZombie, _, _, err := graph.IsZombieEdge(chanID) + isZombie, _, _, err := graph.IsZombieEdge(ctx, chanID) require.NoError(t, err) require.True(t, isZombie) // Finally, attempt to delete a (now) non-existent edge within the // database, this should result in an error. - err = graph.DeleteChannelEdges(false, true, chanID) + err = graph.DeleteChannelEdges(ctx, false, true, chanID) require.ErrorIs(t, err, ErrEdgeNotFound) } @@ -859,12 +863,12 @@ func TestDisconnectBlockAtHeight(t *testing.T) { // Prune the graph a few times to make sure we have entries in the // prune log. - _, err := graph.PruneGraph(spendOutputs, &blockHash, 155) + _, err := graph.PruneGraph(ctx, spendOutputs, &blockHash, 155) require.NoError(t, err, "unable to prune graph") var blockHash2 chainhash.Hash copy(blockHash2[:], bytes.Repeat([]byte{2}, 32)) - _, err = graph.PruneGraph(spendOutputs, &blockHash2, 156) + _, err = graph.PruneGraph(ctx, spendOutputs, &blockHash2, 156) require.NoError(t, err, "unable to prune graph") // We'll create 3 almost identical edges, so first create a helper @@ -901,7 +905,7 @@ func TestDisconnectBlockAtHeight(t *testing.T) { // Call DisconnectBlockAtHeight, which should prune every channel // that has a funding height of 'height' or greater. - removed, err := graph.DisconnectBlockAtHeight(uint32(height)) + removed, err := graph.DisconnectBlockAtHeight(ctx, height) require.NoError(t, err) assertNoEdge(t, graph, edgeInfo.ChannelID) assertNoEdge(t, graph, edgeInfo2.ChannelID) @@ -914,13 +918,13 @@ func TestDisconnectBlockAtHeight(t *testing.T) { // The two first edges should be removed from the db. has, isZombie, err := graph.HasChannelEdge( - lnwire.GossipVersion1, edgeInfo.ChannelID, + ctx, lnwire.GossipVersion1, edgeInfo.ChannelID, ) require.NoError(t, err, "unable to query for edge") require.False(t, has) require.False(t, isZombie) has, isZombie, err = graph.HasChannelEdge( - lnwire.GossipVersion1, edgeInfo2.ChannelID, + ctx, lnwire.GossipVersion1, edgeInfo2.ChannelID, ) require.NoError(t, err, "unable to query for edge") require.False(t, has) @@ -928,7 +932,7 @@ func TestDisconnectBlockAtHeight(t *testing.T) { // Edge 3 should not be removed. has, isZombie, err = graph.HasChannelEdge( - lnwire.GossipVersion1, edgeInfo3.ChannelID, + ctx, lnwire.GossipVersion1, edgeInfo3.ChannelID, ) require.NoError(t, err, "unable to query for edge") require.True(t, has) @@ -936,7 +940,7 @@ func TestDisconnectBlockAtHeight(t *testing.T) { // PruneTip should be set to the blockHash we specified for the block // at height 155. - hash, h, err := graph.PruneTip() + hash, h, err := graph.PruneTip(ctx) require.NoError(t, err, "unable to get prune tip") require.True(t, blockHash.IsEqual(hash)) require.Equal(t, h, height-1) @@ -1180,20 +1184,22 @@ func testEdgeInfoUpdates(t *testing.T, v lnwire.GossipVersion) { // Check for existence of the edge within the database, it should be // found. - found, isZombie, err := graph.HasChannelEdge(chanID) + found, isZombie, err := graph.HasChannelEdge(ctx, chanID) require.NoError(t, err, "unable to query for edge") require.True(t, found) require.False(t, isZombie) // We should also be able to retrieve the channelID only knowing the // channel point of the channel. - dbChanID, err := graph.ChannelID(&outpoint) + dbChanID, err := graph.ChannelID(ctx, &outpoint) require.NoError(t, err, "unable to retrieve channel ID") require.Equal(t, chanID, dbChanID) // With the edges inserted, perform some queries to ensure that they've // been inserted properly. - dbEdgeInfo, dbEdge1, dbEdge2, err := graph.FetchChannelEdgesByID(chanID) + dbEdgeInfo, dbEdge1, dbEdge2, err := graph.FetchChannelEdgesByID( + ctx, chanID, + ) require.NoError(t, err, "unable to fetch channel by ID") compareEdgePolicies(t, dbEdge1, edge1) compareEdgePolicies(t, dbEdge2, edge2) @@ -1202,7 +1208,7 @@ func testEdgeInfoUpdates(t *testing.T, v lnwire.GossipVersion) { // Next, attempt to query the channel edges according to the outpoint // of the channel. dbEdgeInfo, dbEdge1, dbEdge2, err = graph.FetchChannelEdgesByOutpoint( - &outpoint, + ctx, &outpoint, ) require.NoError(t, err, "unable to fetch channel by ID") compareEdgePolicies(t, dbEdge1, edge1) @@ -1576,7 +1582,9 @@ func testAddEdgeProof(t *testing.T, v lnwire.GossipVersion) { require.NoError(t, graph.AddChannelEdge(ctx, edge1)) // Fetch the edge and assert that the proof is nil. - dbEdge, _, _, err := graph.FetchChannelEdgesByID(edge1.ChannelID) + dbEdge, _, _, err := graph.FetchChannelEdgesByID( + ctx, edge1.ChannelID, + ) require.NoError(t, err) require.Nil(t, dbEdge.AuthProof) @@ -1603,10 +1611,12 @@ func testAddEdgeProof(t *testing.T, v lnwire.GossipVersion) { // Now add just the proof via AddEdgeProof. scid1 := lnwire.NewShortChanIDFromInt(edge1.ChannelID) - require.NoError(t, graph.AddEdgeProof(scid1, proof)) + require.NoError(t, graph.AddEdgeProof(ctx, scid1, proof)) // Fetch the edge again and assert that the proof is now set. - dbEdge, _, _, err = graph.FetchChannelEdgesByID(edge1.ChannelID) + dbEdge, _, _, err = graph.FetchChannelEdgesByID( + ctx, edge1.ChannelID, + ) require.NoError(t, err) require.NotNil(t, dbEdge.AuthProof) @@ -1616,7 +1626,9 @@ func testAddEdgeProof(t *testing.T, v lnwire.GossipVersion) { require.NoError(t, graph.AddChannelEdge(ctx, edge2)) // Fetch the edge and assert that the proof is set. - dbEdge2, _, _, err := graph.FetchChannelEdgesByID(edge2.ChannelID) + dbEdge2, _, _, err := graph.FetchChannelEdgesByID( + ctx, edge2.ChannelID, + ) require.NoError(t, err) require.NotNil(t, dbEdge2.AuthProof) } @@ -1875,7 +1887,7 @@ func testGraphTraversalCacheable(t *testing.T, v lnwire.GossipVersion) { // Query the ChannelGraph which uses the cache to iterate // through the channels for each node. err = graph.ChannelGraph.ForEachNodeDirectedChannel( - node, func(d *DirectedChannel) error { + ctx, node, func(d *DirectedChannel) error { delete(chanIndex, d.ChannelID) return nil }, func() {}, @@ -1884,7 +1896,7 @@ func testGraphTraversalCacheable(t *testing.T, v lnwire.GossipVersion) { // Now skip the cache and query the DB directly. err = graph.db.ForEachNodeDirectedChannel( - v, node, func(d *DirectedChannel) error { + ctx, v, node, func(d *DirectedChannel) error { delete(chanIndex2, d.ChannelID) return nil }, func() {}, @@ -2110,7 +2122,7 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, func assertPruneTip(t *testing.T, graph *ChannelGraph, blockHash *chainhash.Hash, blockHeight uint32) { - pruneHash, pruneHeight, err := graph.PruneTip() + pruneHash, pruneHeight, err := graph.PruneTip(t.Context()) require.NoError(t, err) require.Equal(t, blockHash[:], pruneHash[:]) require.Equal(t, blockHeight, pruneHeight) @@ -2262,7 +2274,7 @@ func TestGraphPruning(t *testing.T) { // With all the channel points added, we'll consult the graph to ensure // it has the same channel view as the one we just constructed. - channelView, err := graph.ChannelView() + channelView, err := graph.ChannelView(ctx) require.NoError(t, err, "unable to get graph channel view") assertChanViewEqual(t, channelView, edgePoints) @@ -2275,7 +2287,9 @@ func TestGraphPruning(t *testing.T) { copy(blockHash[:], bytes.Repeat([]byte{1}, 32)) blockHeight := uint32(1) block := channelPoints[:2] - prunedChans, err := graph.PruneGraph(block, &blockHash, blockHeight) + prunedChans, err := graph.PruneGraph( + ctx, block, &blockHash, blockHeight, + ) require.NoError(t, err, "unable to prune graph") require.Len(t, prunedChans, 2) @@ -2287,7 +2301,7 @@ func TestGraphPruning(t *testing.T) { assertNumChans(t, graph, 2) // Those channels should also be missing from the channel view. - channelView, err = graph.ChannelView() + channelView, err = graph.ChannelView(ctx) require.NoError(t, err, "unable to get graph channel view") assertChanViewEqualChanPoints(t, channelView, channelPoints[2:]) @@ -2301,7 +2315,7 @@ func TestGraphPruning(t *testing.T) { blockHash = sha256.Sum256(blockHash[:]) blockHeight = 2 prunedChans, err = graph.PruneGraph( - []*wire.OutPoint{nonChannel}, &blockHash, blockHeight, + ctx, []*wire.OutPoint{nonChannel}, &blockHash, blockHeight, ) require.NoError(t, err, "unable to prune graph") @@ -2319,7 +2333,7 @@ func TestGraphPruning(t *testing.T) { blockHash = sha256.Sum256(blockHash[:]) blockHeight = 3 prunedChans, err = graph.PruneGraph( - channelPoints[2:], &blockHash, blockHeight, + ctx, channelPoints[2:], &blockHash, blockHeight, ) require.NoError(t, err, "unable to prune graph") @@ -2336,7 +2350,7 @@ func TestGraphPruning(t *testing.T) { // Finally, the channel view at this point in the graph should now be // completely empty. Those channels should also be missing from the // channel view. - channelView, err = graph.ChannelView() + channelView, err = graph.ChannelView(ctx) require.NoError(t, err, "unable to get graph channel view") require.Empty(t, channelView) } @@ -2396,7 +2410,7 @@ func TestChanUpdatesInHorizon(t *testing.T) { // If we issue an arbitrary query before any channel updates are // inserted in the database, we should get zero results. chanIter := graph.ChanUpdatesInHorizon( - time.Unix(999, 0), time.Unix(9999, 0), + ctx, time.Unix(999, 0), time.Unix(9999, 0), ) chanUpdates, err := fn.CollectErr(chanIter) @@ -2503,7 +2517,7 @@ func TestChanUpdatesInHorizon(t *testing.T) { } for _, queryCase := range queryCases { respIter := graph.ChanUpdatesInHorizon( - queryCase.start, queryCase.end, + ctx, queryCase.start, queryCase.end, ) resp, err := fn.CollectErr(respIter) @@ -2540,7 +2554,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) { // If we issue an arbitrary query before we insert any nodes into the // database, then we shouldn't get any results back. 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) require.NoError(t, err, "unable to query for node updates") @@ -2615,7 +2629,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) { } for _, queryCase := range queryCases { iter := graph.NodeUpdatesInHorizon( - queryCase.start, queryCase.end, + ctx, queryCase.start, queryCase.end, ) resp, err := fn.CollectErr(iter) @@ -2743,7 +2757,7 @@ func testNodeUpdatesWithBatchSize(t *testing.T, ctx context.Context, for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { iter := testGraph.NodeUpdatesInHorizon( - tc.start, tc.end, + ctx, tc.start, tc.end, WithNodeUpdateIterBatchSize( batchSize, ), @@ -2816,7 +2830,7 @@ func TestNodeUpdatesInHorizonEarlyTermination(t *testing.T) { for _, stopAt := range terminationPoints { t.Run(fmt.Sprintf("StopAt%d", stopAt), func(t *testing.T) { iter := graph.NodeUpdatesInHorizon( - startTime, startTime.Add(200*time.Hour), + ctx, startTime, startTime.Add(200*time.Hour), WithNodeUpdateIterBatchSize(10), ) @@ -2905,7 +2919,7 @@ func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) { // Now we'll run the main query, and verify that we get // back the expected number of channels. iter := graph.ChanUpdatesInHorizon( - startTime, startTime.Add(26*time.Hour), + ctx, startTime, startTime.Add(26*time.Hour), WithChanUpdateIterBatchSize(batchSize), ) @@ -2929,6 +2943,7 @@ func TestChanUpdatesInHorizonBoundaryConditions(t *testing.T) { // FilterKnownChanIDs is tested in TestFilterKnownChanIDs. func TestFilterKnownChanIDsZombieRevival(t *testing.T) { t.Parallel() + ctx := t.Context() graph := MakeTestGraph(t) @@ -2939,16 +2954,22 @@ func TestFilterKnownChanIDsZombieRevival(t *testing.T) { ) isZombie := func(scid lnwire.ShortChannelID) bool { - zombie, _, _, err := graph.IsZombieEdge(scid.ToUint64()) + zombie, _, _, err := graph.IsZombieEdge( + ctx, scid.ToUint64(), + ) require.NoError(t, err) return zombie } // Mark channel 1 and 2 as zombies. - err := graph.MarkEdgeZombie(scid1.ToUint64(), [33]byte{}, [33]byte{}) + err := graph.MarkEdgeZombie( + ctx, scid1.ToUint64(), [33]byte{}, [33]byte{}, + ) require.NoError(t, err) - err = graph.MarkEdgeZombie(scid2.ToUint64(), [33]byte{}, [33]byte{}) + err = graph.MarkEdgeZombie( + ctx, scid2.ToUint64(), [33]byte{}, [33]byte{}, + ) require.NoError(t, err) require.True(t, isZombie(scid1)) @@ -2957,7 +2978,7 @@ func TestFilterKnownChanIDsZombieRevival(t *testing.T) { // Call FilterKnownChanIDs with an isStillZombie call-back that would // result in the current zombies still be considered as zombies. - _, err = graph.FilterKnownChanIDs([]ChannelUpdateInfo{ + _, err = graph.FilterKnownChanIDs(ctx, []ChannelUpdateInfo{ {ShortChannelID: scid1}, {ShortChannelID: scid2}, {ShortChannelID: scid3}, @@ -2973,7 +2994,7 @@ func TestFilterKnownChanIDsZombieRevival(t *testing.T) { // Now call it again but this time with a isStillZombie call-back that // would result in channel with SCID 2 no longer being considered a // zombie. - _, err = graph.FilterKnownChanIDs([]ChannelUpdateInfo{ + _, err = graph.FilterKnownChanIDs(ctx, []ChannelUpdateInfo{ {ShortChannelID: scid1}, { ShortChannelID: scid2, @@ -3019,7 +3040,9 @@ func TestFilterKnownChanIDs(t *testing.T) { {ShortChannelID: scid2}, {ShortChannelID: scid3}, } - filteredIDs, err := graph.FilterKnownChanIDs(preChanIDs, isZombieUpdate) + filteredIDs, err := graph.FilterKnownChanIDs( + ctx, preChanIDs, isZombieUpdate, + ) require.NoError(t, err, "unable to filter chan IDs") require.EqualValues(t, []uint64{ scid1.ToUint64(), @@ -3057,7 +3080,9 @@ func TestFilterKnownChanIDs(t *testing.T) { node1, node2, ) require.NoError(t, graph.AddChannelEdge(ctx, channel)) - err := graph.DeleteChannelEdges(false, true, channel.ChannelID) + err := graph.DeleteChannelEdges( + ctx, false, true, channel.ChannelID, + ) require.NoError(t, err) zombieIDs = append( @@ -3142,7 +3167,7 @@ func TestFilterKnownChanIDs(t *testing.T) { for _, queryCase := range queryCases { resp, err := graph.FilterKnownChanIDs( - queryCase.queryIDs, isZombieUpdate, + ctx, queryCase.queryIDs, isZombieUpdate, ) require.NoError(t, err) @@ -3299,7 +3324,7 @@ func TestStressTestChannelGraphAPI(t *testing.T) { } return graph.MarkEdgeZombie( - channel.id.ToUint64(), + ctx, channel.id.ToUint64(), node1.PubKeyBytes, node2.PubKeyBytes, ) @@ -3321,7 +3346,7 @@ func TestStressTestChannelGraphAPI(t *testing.T) { } _, err := graph.FilterKnownChanIDs( - chanIDs, + ctx, chanIDs, func(t time.Time, t2 time.Time) bool { return rand.Intn(2) == 0 }, @@ -3339,7 +3364,7 @@ func TestStressTestChannelGraphAPI(t *testing.T) { } _, _, err := graph.HasChannelEdge( - lnwire.GossipVersion1, + ctx, lnwire.GossipVersion1, channel.id.ToUint64(), ) @@ -3360,7 +3385,7 @@ func TestStressTestChannelGraphAPI(t *testing.T) { } _, err := graph.PruneGraph( - spentOutpoints, &blockHash, 100, + ctx, spentOutpoints, &blockHash, 100, ) return err @@ -3370,7 +3395,8 @@ func TestStressTestChannelGraphAPI(t *testing.T) { name: "ChanUpdateInHorizon", fn: func() error { iter := graph.ChanUpdatesInHorizon( - time.Now().Add(-time.Hour), time.Now(), + ctx, time.Now().Add(-time.Hour), + time.Now(), ) _, err := fn.CollectErr(iter) @@ -3395,7 +3421,8 @@ func TestStressTestChannelGraphAPI(t *testing.T) { } err := graph.DeleteChannelEdges( - strictPruning, markZombie, chanIDs..., + ctx, strictPruning, markZombie, + chanIDs..., ) if err != nil && !errors.Is(err, ErrEdgeNotFound) { @@ -3410,7 +3437,7 @@ func TestStressTestChannelGraphAPI(t *testing.T) { name: "DisconnectBlockAtHeight", fn: func() error { _, err := graph.DisconnectBlockAtHeight( - newBlockHeight(), + ctx, newBlockHeight(), ) return err @@ -3476,7 +3503,7 @@ func TestFilterChannelRange(t *testing.T) { // If we try to filter a channel range before we have any channels // inserted, we should get an empty slice of results. - resp, err := graph.FilterChannelRange(10, 100, false) + resp, err := graph.FilterChannelRange(ctx, 10, 100, false) require.NoError(t, err) require.Empty(t, resp) @@ -3651,7 +3678,7 @@ func TestFilterChannelRange(t *testing.T) { // First, do the query without requesting timestamps. resp, err := graph.FilterChannelRange( - test.startHeight, test.endHeight, false, + ctx, test.startHeight, test.endHeight, false, ) require.NoError(t, err) @@ -3665,7 +3692,7 @@ func TestFilterChannelRange(t *testing.T) { // Now, query the timestamps as well. resp, err = graph.FilterChannelRange( - test.startHeight, test.endHeight, true, + ctx, test.startHeight, test.endHeight, true, ) require.NoError(t, err) @@ -3753,14 +3780,16 @@ func testFetchChanInfos(t *testing.T, v lnwire.GossipVersion) { v, 666, 0, 0, 0, node1, node2, ) require.NoError(t, graph.AddChannelEdge(ctx, zombieChan)) - err := graph.DeleteChannelEdges(false, true, zombieChan.ChannelID) + err := graph.DeleteChannelEdges( + ctx, false, true, zombieChan.ChannelID, + ) require.NoError(t, err, "unable to delete and mark edge zombie") edgeQuery = append(edgeQuery, zombieChanID.ToUint64()) // We'll now attempt to query for the range of channel ID's we just // inserted into the database. We should get the exact same set of // edges back. - resp, err := graph.FetchChanInfos(edgeQuery) + resp, err := graph.FetchChanInfos(ctx, edgeQuery) require.NoError(t, err, "unable to fetch chan edges") require.Len(t, resp, len(edges)) @@ -3968,7 +3997,8 @@ func TestChannelEdgePruningUpdateIndexDeletion(t *testing.T) { var blockHash chainhash.Hash copy(blockHash[:], bytes.Repeat([]byte{2}, 32)) _, err := graph.PruneGraph( - []*wire.OutPoint{&edgeInfo.ChannelPoint}, &blockHash, 101, + ctx, []*wire.OutPoint{&edgeInfo.ChannelPoint}, &blockHash, + 101, ) require.NoError(t, err, "unable to prune graph") @@ -4017,7 +4047,7 @@ func TestPruneGraphNodes(t *testing.T) { require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1)) // We'll now initiate a around of graph pruning. - require.NoError(t, graph.PruneGraphNodes()) + require.NoError(t, graph.PruneGraphNodes(ctx)) // At this point, there should be 3 nodes left in the graph still: the // source node (which can't be pruned), and node 1+2. Nodes 1 and two @@ -4091,7 +4121,9 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) { // update time of our test node. startTime := time.Unix(9, 0) 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 // match the node we just inserted. @@ -4107,7 +4139,9 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) { // 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. - nodesInHorizonIter = graph.NodeUpdatesInHorizon(startTime, endTime) + nodesInHorizonIter = graph.NodeUpdatesInHorizon( + ctx, startTime, endTime, + ) nodesInHorizon, err = fn.CollectErr(nodesInHorizonIter) require.NoError(t, err, "unable to fetch nodes in horizon") require.Empty(t, nodesInHorizon) @@ -4196,7 +4230,7 @@ func testNodeIsPublic(t *testing.T, v lnwire.GossipVersion) { for _, node := range nodes { for _, graph := range graphs { isPublic, err := graph.IsPublicNode( - node.PubKeyBytes, + ctx, node.PubKeyBytes, ) require.NoError(t, err) @@ -4214,7 +4248,7 @@ func testNodeIsPublic(t *testing.T, v lnwire.GossipVersion) { // has any advertised edges. for _, graph := range graphs { err := graph.DeleteChannelEdges( - false, true, aliceBobEdge.ChannelID, + ctx, false, true, aliceBobEdge.ChannelID, ) require.NoError(t, err, "unable to remove edge") } @@ -4231,7 +4265,7 @@ func testNodeIsPublic(t *testing.T, v lnwire.GossipVersion) { // it without it being advertised. for _, graph := range graphs { err := graph.DeleteChannelEdges( - false, true, bobCarolEdge.ChannelID, + ctx, false, true, bobCarolEdge.ChannelID, ) require.NoError(t, err, "unable to remove edge") @@ -4301,7 +4335,7 @@ func testIsPublicNodeEmptyChannelSignature(t *testing.T, // node1 should NOT be considered public because the // channel announcement has empty signatures. - isPublic, err := graph.IsPublicNode(node1.PubKeyBytes) + isPublic, err := graph.IsPublicNode(ctx, node1.PubKeyBytes) require.NoError(t, err) require.False(t, isPublic) } @@ -4325,7 +4359,7 @@ func BenchmarkIsPublicNode(b *testing.B) { // Query random nodes to avoid query caching and better // represent real-world query patterns. nodePub := nodes[rng.Intn(len(nodes))].PubKeyBytes - _, err := graph.IsPublicNode(nodePub) + _, err := graph.IsPublicNode(b.Context(), nodePub) require.NoError(b, err) } } @@ -4359,7 +4393,7 @@ func testDisabledChannelIDs(t *testing.T, v lnwire.GossipVersion) { require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo)) // Ensure no disabled channels exist in the bucket on start. - disabledChanIds, err := graph.DisabledChannelIDs() + disabledChanIds, err := graph.DisabledChannelIDs(ctx) require.NoError(t, err, "unable to get disabled channel ids") require.Empty(t, disabledChanIds) @@ -4372,7 +4406,7 @@ func testDisabledChannelIDs(t *testing.T, v lnwire.GossipVersion) { edge1.DisableFlags |= lnwire.ChanUpdateDisableIncoming } require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1)) - disabledChanIds, err = graph.DisabledChannelIDs() + disabledChanIds, err = graph.DisabledChannelIDs(ctx) require.NoError(t, err, "unable to get disabled channel ids") require.Empty(t, disabledChanIds) @@ -4385,16 +4419,16 @@ func testDisabledChannelIDs(t *testing.T, v lnwire.GossipVersion) { edge2.DisableFlags |= lnwire.ChanUpdateDisableIncoming } require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2)) - disabledChanIds, err = graph.DisabledChannelIDs() + disabledChanIds, err = graph.DisabledChannelIDs(ctx) require.NoError(t, err, "unable to get disabled channel ids") require.Equal(t, []uint64{edgeInfo.ChannelID}, disabledChanIds) // Delete the channel edge and ensure it is removed from the disabled // list. require.NoError(t, graph.DeleteChannelEdges( - false, true, edgeInfo.ChannelID, + ctx, false, true, edgeInfo.ChannelID, )) - disabledChanIds, err = graph.DisabledChannelIDs() + disabledChanIds, err = graph.DisabledChannelIDs(ctx) require.NoError(t, err, "unable to get disabled channel ids") require.Empty(t, disabledChanIds) } @@ -4467,7 +4501,9 @@ func TestEdgePolicyMissingMaxHTLC(t *testing.T) { // we added is invalid according to the new format, it should be as we // are not aware of the policy (indicated by the policy returned being // nil) - dbEdgeInfo, dbEdge1, dbEdge2, err := graph.FetchChannelEdgesByID(chanID) + dbEdgeInfo, dbEdge1, dbEdge2, err := graph.FetchChannelEdgesByID( + ctx, chanID, + ) require.NoError(t, err, "unable to fetch channel by ID") // The first edge should have a nil-policy returned @@ -4479,7 +4515,9 @@ func TestEdgePolicyMissingMaxHTLC(t *testing.T) { // policies then become fully populated. require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1)) - dbEdgeInfo, dbEdge1, dbEdge2, err = graph.FetchChannelEdgesByID(chanID) + dbEdgeInfo, dbEdge1, dbEdge2, err = graph.FetchChannelEdgesByID( + ctx, chanID, + ) require.NoError(t, err, "unable to fetch channel by ID") compareEdgePolicies(t, dbEdge1, edge1) compareEdgePolicies(t, dbEdge2, edge2) @@ -4523,7 +4561,7 @@ func putSerializedPolicy(t *testing.T, db kvdb.Backend, from []byte, func assertNumZombies(t *testing.T, graph *ChannelGraph, expZombies uint64) { t.Helper() - numZombies, err := graph.NumZombies() + numZombies, err := graph.NumZombies(t.Context()) require.NoError(t, err, "unable to query number of zombies") require.Equal(t, expZombies, numZombies) } @@ -4552,16 +4590,18 @@ func TestGraphZombieIndex(t *testing.T) { // Since the edge is known the graph and it isn't a zombie, IsZombieEdge // should not report the channel as a zombie. - isZombie, _, _, err := graph.IsZombieEdge(edge.ChannelID) + isZombie, _, _, err := graph.IsZombieEdge(ctx, edge.ChannelID) require.NoError(t, err) require.False(t, isZombie) assertNumZombies(t, graph, 0) // If we delete the edge and mark it as a zombie, then we should expect // to see it within the index. - err = graph.DeleteChannelEdges(false, true, edge.ChannelID) + err = graph.DeleteChannelEdges(ctx, false, true, edge.ChannelID) require.NoError(t, err, "unable to mark edge as zombie") - isZombie, pubKey1, pubKey2, err := graph.IsZombieEdge(edge.ChannelID) + isZombie, pubKey1, pubKey2, err := graph.IsZombieEdge( + ctx, edge.ChannelID, + ) require.NoError(t, err) require.True(t, isZombie) require.Equal(t, node1.PubKeyBytes, pubKey1) @@ -4570,15 +4610,16 @@ func TestGraphZombieIndex(t *testing.T) { // Similarly, if we mark the same edge as live, we should no longer see // it within the index. - require.NoError(t, graph.MarkEdgeLive(edge.ChannelID)) + require.NoError(t, graph.MarkEdgeLive(ctx, edge.ChannelID)) // Attempting to mark the edge as live again now that it is no longer // in the zombie index should fail. require.ErrorIs( - t, graph.MarkEdgeLive(edge.ChannelID), ErrZombieEdgeNotFound, + t, graph.MarkEdgeLive(ctx, edge.ChannelID), + ErrZombieEdgeNotFound, ) - isZombie, _, _, err = graph.IsZombieEdge(edge.ChannelID) + isZombie, _, _, err = graph.IsZombieEdge(ctx, edge.ChannelID) require.NoError(t, err) require.False(t, isZombie) @@ -4587,11 +4628,11 @@ func TestGraphZombieIndex(t *testing.T) { // If we mark the edge as a zombie manually, then it should show up as // being a zombie once again. err = graph.MarkEdgeZombie( - edge.ChannelID, node1.PubKeyBytes, node2.PubKeyBytes, + ctx, edge.ChannelID, node1.PubKeyBytes, node2.PubKeyBytes, ) require.NoError(t, err, "unable to mark edge as zombie") - isZombie, _, _, err = graph.IsZombieEdge(edge.ChannelID) + isZombie, _, _, err = graph.IsZombieEdge(ctx, edge.ChannelID) require.NoError(t, err) require.True(t, isZombie) assertNumZombies(t, graph, 1) @@ -4724,12 +4765,12 @@ func testBatchedAddChannelEdge(t *testing.T, v lnwire.GossipVersion) { // Prune the graph a few times to make sure we have entries in the // prune log. - _, err := graph.PruneGraph(spendOutputs, &blockHash, 155) + _, err := graph.PruneGraph(ctx, spendOutputs, &blockHash, 155) require.Nil(t, err) var blockHash2 chainhash.Hash copy(blockHash2[:], bytes.Repeat([]byte{2}, 32)) - _, err = graph.PruneGraph(spendOutputs, &blockHash2, 156) + _, err = graph.PruneGraph(ctx, spendOutputs, &blockHash2, 156) require.Nil(t, err) // We'll create 3 almost identical edges, so first create a helper @@ -4928,7 +4969,7 @@ func testGraphCacheForEachNodeChannel(t *testing.T, getSingleChannel := func() *DirectedChannel { var ch *DirectedChannel err := graph.db.ForEachNodeDirectedChannel( - v, node1.PubKeyBytes, + ctx, v, node1.PubKeyBytes, func(c *DirectedChannel) error { require.Nil(t, ch) ch = c @@ -5045,16 +5086,16 @@ func TestClosedScid(t *testing.T) { scid := lnwire.ShortChannelID{} // The scid should not exist in the closedScidBucket. - exists, err := graph.IsClosedScid(scid) + exists, err := graph.IsClosedScid(t.Context(), scid) require.Nil(t, err) require.False(t, exists) // After we call PutClosedScid, the call to IsClosedScid should return // true. - err = graph.PutClosedScid(scid) + err = graph.PutClosedScid(t.Context(), scid) require.Nil(t, err) - exists, err = graph.IsClosedScid(scid) + exists, err = graph.IsClosedScid(t.Context(), scid) require.Nil(t, err) require.True(t, exists) } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 446644077..f50716239 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -21,11 +21,12 @@ import ( type NodeTraverser interface { // ForEachNodeDirectedChannel calls the callback for every channel of // the given node. - ForEachNodeDirectedChannel(nodePub route.Vertex, + ForEachNodeDirectedChannel(ctx context.Context, nodePub route.Vertex, cb func(channel *DirectedChannel) error, reset func()) error // FetchNodeFeatures returns the features of the given node. - FetchNodeFeatures(nodePub route.Vertex) (*lnwire.FeatureVector, error) + FetchNodeFeatures(ctx context.Context, + nodePub route.Vertex) (*lnwire.FeatureVector, error) } // Store represents the main interface for the channel graph database for all @@ -33,11 +34,12 @@ type NodeTraverser interface { type Store interface { //nolint:interfacebloat // ForEachNodeDirectedChannel calls the callback for every channel of // the given node. - ForEachNodeDirectedChannel(v lnwire.GossipVersion, nodePub route.Vertex, - cb func(channel *DirectedChannel) error, reset func()) error + ForEachNodeDirectedChannel(ctx context.Context, v lnwire.GossipVersion, + nodePub route.Vertex, cb func(channel *DirectedChannel) error, + reset func()) error // FetchNodeFeatures returns the features of the given node. - FetchNodeFeatures(v lnwire.GossipVersion, + FetchNodeFeatures(ctx context.Context, v lnwire.GossipVersion, nodePub route.Vertex) (*lnwire.FeatureVector, error) // AddNode adds a vertex/node to the graph database. If the @@ -121,7 +123,7 @@ type Store interface { //nolint:interfacebloat // 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 // date node announcements. - NodeUpdatesInHorizon(startTime, endTime time.Time, + NodeUpdatesInHorizon(ctx context.Context, startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[*models.Node, error] // FetchNode attempts to look up a target node by its identity @@ -148,12 +150,14 @@ type Store interface { //nolint:interfacebloat // IsPublicNode is a helper method that determines whether the node with // the given public key is seen as a public node in the graph from the // graph's source node's point of view. - IsPublicNode(v lnwire.GossipVersion, pubKey [33]byte) (bool, error) + IsPublicNode(ctx context.Context, v lnwire.GossipVersion, + pubKey [33]byte) (bool, error) // GraphSession will provide the call-back with access to a // NodeTraverser instance which can be used to perform queries against // the channel graph. - GraphSession(cb func(graph NodeTraverser) error, reset func()) error + GraphSession(ctx context.Context, + cb func(graph NodeTraverser) error, reset func()) error // ForEachChannel iterates through all the channel edges stored within // the graph and invokes the passed callback for each edge. The callback @@ -183,14 +187,15 @@ type Store interface { //nolint:interfacebloat // // NOTE: this method is like ForEachChannel but fetches only the data // required for the graph cache. - ForEachChannelCacheable(v lnwire.GossipVersion, + ForEachChannelCacheable(ctx context.Context, v lnwire.GossipVersion, cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, reset func()) error // DisabledChannelIDs returns the channel ids of disabled channels. // A channel is disabled when two of the associated ChanelEdgePolicies // have their disabled bit on. - DisabledChannelIDs(v lnwire.GossipVersion) ([]uint64, error) + DisabledChannelIDs(ctx context.Context, + v lnwire.GossipVersion) ([]uint64, error) // AddChannelEdge adds a new (undirected, blank) edge to the graph // database. An undirected edge from the two target nodes are created. @@ -208,15 +213,15 @@ type Store interface { //nolint:interfacebloat // last time the edge was updated for both directed edges are returned // along with the boolean. If it is not found, then the zombie index is // checked and its result is returned as the second boolean. - HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, bool, - error) + HasV1ChannelEdge(ctx context.Context, chanID uint64) ( + time.Time, time.Time, bool, bool, error) // HasChannelEdge returns true if the database knows of a channel edge // with the passed channel ID and gossip version, and false otherwise. // If it is not found, then the zombie index is checked and its result // is returned as the second boolean. - HasChannelEdge(v lnwire.GossipVersion, chanID uint64) (bool, bool, - error) + HasChannelEdge(ctx context.Context, v lnwire.GossipVersion, + chanID uint64) (bool, bool, error) // DeleteChannelEdges removes edges with the given channel IDs from the // database and marks them as zombies. This ensures that we're unable to @@ -227,19 +232,19 @@ type Store interface { //nolint:interfacebloat // failed to send the fresh update to be the one that resurrects the // channel from its zombie state. The markZombie bool denotes whether // to mark the channel as a zombie. - DeleteChannelEdges(v lnwire.GossipVersion, strictZombiePruning, - markZombie bool, chanIDs ...uint64) ( + DeleteChannelEdges(ctx context.Context, v lnwire.GossipVersion, + strictZombiePruning, markZombie bool, chanIDs ...uint64) ( []*models.ChannelEdgeInfo, error) // AddEdgeProof sets the proof of an existing edge in the graph // database. - AddEdgeProof(chanID lnwire.ShortChannelID, + AddEdgeProof(ctx context.Context, chanID lnwire.ShortChannelID, proof *models.ChannelAuthProof) error // ChannelID attempt to lookup the 8-byte compact channel ID which maps // to the passed channel point (outpoint). If the passed channel doesn't // exist within the database, then ErrEdgeNotFound is returned. - ChannelID(v lnwire.GossipVersion, + ChannelID(ctx context.Context, v lnwire.GossipVersion, chanPoint *wire.OutPoint) (uint64, error) // HighestChanID returns the "highest" known channel ID in the channel @@ -252,7 +257,8 @@ type Store interface { //nolint:interfacebloat // ChanUpdatesInHorizon returns all the known channel edges which have // at least one edge that has an update timestamp within the specified // horizon. - ChanUpdatesInHorizon(startTime, endTime time.Time, + ChanUpdatesInHorizon(ctx context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[ChannelEdge, error] // FilterKnownChanIDs takes a set of channel IDs and return the subset @@ -262,8 +268,9 @@ type Store interface { //nolint:interfacebloat // callers to determine the set of channels another peer knows of that // we don't. The ChannelUpdateInfos for the known zombies is also // returned. - FilterKnownChanIDs(chansInfo []ChannelUpdateInfo) ([]uint64, - []ChannelUpdateInfo, error) + FilterKnownChanIDs(ctx context.Context, + chansInfo []ChannelUpdateInfo) ([]uint64, []ChannelUpdateInfo, + error) // FilterChannelRange returns the channel ID's of all known channels // which were mined in a block height within the passed range. The @@ -273,8 +280,9 @@ type Store interface { //nolint:interfacebloat // offline. If withTimestamps is true then the timestamp info of the // latest received channel update messages of the channel will be // included in the response. - FilterChannelRange(startHeight, endHeight uint32, withTimestamps bool) ( - []BlockChannelRange, error) + FilterChannelRange(ctx context.Context, startHeight, + endHeight uint32, + withTimestamps bool) ([]BlockChannelRange, error) // FetchChanInfos returns the set of channel edges that correspond to // the passed channel ID's. If an edge is the query is unknown to the @@ -282,7 +290,7 @@ type Store interface { //nolint:interfacebloat // edges that exist at the time of the query. This can be used to // respond to peer queries that are seeking to fill in gaps in their // view of the channel graph. - FetchChanInfos(v lnwire.GossipVersion, + FetchChanInfos(ctx context.Context, v lnwire.GossipVersion, chanIDs []uint64) ([]ChannelEdge, error) // FetchChannelEdgesByOutpoint attempts to lookup the two directed edges @@ -291,7 +299,8 @@ type Store interface { //nolint:interfacebloat // houses the general information for the channel itself is returned as // well as two structs that contain the routing policies for the channel // in either direction. - FetchChannelEdgesByOutpoint(v lnwire.GossipVersion, op *wire.OutPoint) ( + FetchChannelEdgesByOutpoint(ctx context.Context, + v lnwire.GossipVersion, op *wire.OutPoint) ( *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) @@ -306,7 +315,8 @@ type Store interface { //nolint:interfacebloat // zombie within the database. In this case, the ChannelEdgePolicy's // will be nil, and the ChannelEdgeInfo will only include the public // keys of each node. - FetchChannelEdgesByID(v lnwire.GossipVersion, chanID uint64) ( + FetchChannelEdgesByID(ctx context.Context, v lnwire.GossipVersion, + chanID uint64) ( *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) @@ -314,37 +324,38 @@ type Store interface { //nolint:interfacebloat // channel within the known channel graph. The set of UTXO's (along with // their scripts) returned are the ones that need to be watched on chain // to detect channel closes on the resident blockchain. - ChannelView() ([]EdgePoint, error) + ChannelView(ctx context.Context) ([]EdgePoint, error) // MarkEdgeZombie attempts to mark a channel identified by its channel // ID as a zombie. This method is used on an ad-hoc basis, when channels // need to be marked as zombies outside the normal pruning cycle. - MarkEdgeZombie(chanID uint64, + MarkEdgeZombie(ctx context.Context, chanID uint64, pubKey1, pubKey2 [33]byte) error // MarkEdgeLive clears an edge from our zombie index, deeming it as // live. - MarkEdgeLive(chanID uint64) error + MarkEdgeLive(ctx context.Context, chanID uint64) error // IsZombieEdge returns whether the edge is considered zombie. If it is // a zombie, then the two node public keys corresponding to this edge // are also returned. - IsZombieEdge(v lnwire.GossipVersion, chanID uint64) (bool, [33]byte, - [33]byte, error) + IsZombieEdge(ctx context.Context, v lnwire.GossipVersion, + chanID uint64) (bool, [33]byte, [33]byte, error) // NumZombies returns the current number of zombie channels in the // graph. - NumZombies() (uint64, error) + NumZombies(ctx context.Context) (uint64, error) // PutClosedScid stores a SCID for a closed channel in the database. // This is so that we can ignore channel announcements that we know to // be closed without having to validate them and fetch a block. - PutClosedScid(scid lnwire.ShortChannelID) error + PutClosedScid(ctx context.Context, scid lnwire.ShortChannelID) error // IsClosedScid checks whether a channel identified by the passed in // scid is closed. This helps avoid having to perform expensive // validation checks. - IsClosedScid(scid lnwire.ShortChannelID) (bool, error) + IsClosedScid(ctx context.Context, + scid lnwire.ShortChannelID) (bool, error) // UpdateEdgePolicy updates the edge routing policy for a single // directed edge within the database for the referenced channel. The @@ -374,14 +385,14 @@ type Store interface { //nolint:interfacebloat // has been used to prune channels in the graph. Knowing the "prune tip" // allows callers to tell if the graph is currently in sync with the // current best known UTXO state. - PruneTip() (*chainhash.Hash, uint32, error) + PruneTip(ctx context.Context) (*chainhash.Hash, uint32, error) // PruneGraphNodes is a garbage collection method which attempts to // prune out any nodes from the channel graph that are currently // unconnected. This ensures that we only maintain a graph of reachable // nodes. In the event that a pruned node gains more channels, it will // be re-added back to the graph. - PruneGraphNodes() ([]route.Vertex, error) + PruneGraphNodes(ctx context.Context) ([]route.Vertex, error) // PruneGraph prunes newly closed channels from the channel graph in // response to a new block being solved on the network. Any transactions @@ -392,7 +403,7 @@ type Store interface { //nolint:interfacebloat // slice of channels that have been closed by the target block along // with any pruned nodes are returned if the function succeeds without // error. - PruneGraph(spentOutputs []*wire.OutPoint, + PruneGraph(ctx context.Context, spentOutputs []*wire.OutPoint, blockHash *chainhash.Hash, blockHeight uint32) ( []*models.ChannelEdgeInfo, []route.Vertex, error) @@ -403,6 +414,6 @@ type Store interface { //nolint:interfacebloat // set to the last prune height valid for the remaining chain. // Channels that were removed from the graph resulting from the // disconnected block are returned. - DisconnectBlockAtHeight(height uint32) ([]*models.ChannelEdgeInfo, - error) + DisconnectBlockAtHeight(ctx context.Context, + height uint32) ([]*models.ChannelEdgeInfo, error) } diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index de9ff401e..c202d76f5 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -496,9 +496,10 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo, // // NOTE: this method is like ForEachChannel but fetches only the data required // for the graph cache. -func (c *KVStore) ForEachChannelCacheable(v lnwire.GossipVersion, - cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, - *models.CachedEdgePolicy) error, reset func()) error { +func (c *KVStore) ForEachChannelCacheable(_ context.Context, + v lnwire.GossipVersion, cb func(*models.CachedEdgeInfo, + *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, + reset func()) error { if v != lnwire.GossipVersion1 { return ErrVersionNotSupportedForKVDB @@ -669,9 +670,9 @@ func (c *KVStore) fetchNodeFeatures(tx kvdb.RTx, // Unknown policies are passed into the callback as nil values. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (c *KVStore) ForEachNodeDirectedChannel(v lnwire.GossipVersion, - nodePub route.Vertex, cb func(channel *DirectedChannel) error, - reset func()) error { +func (c *KVStore) ForEachNodeDirectedChannel(_ context.Context, + v lnwire.GossipVersion, nodePub route.Vertex, + cb func(channel *DirectedChannel) error, reset func()) error { if v != lnwire.GossipVersion1 { return ErrVersionNotSupportedForKVDB @@ -684,7 +685,7 @@ func (c *KVStore) ForEachNodeDirectedChannel(v lnwire.GossipVersion, // known for the node, an empty feature vector is returned. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (c *KVStore) FetchNodeFeatures(v lnwire.GossipVersion, +func (c *KVStore) FetchNodeFeatures(_ context.Context, v lnwire.GossipVersion, nodePub route.Vertex) (*lnwire.FeatureVector, error) { if v != lnwire.GossipVersion1 { @@ -773,7 +774,7 @@ func (c *KVStore) ForEachNodeCached(ctx context.Context, withAddrs bool, // A channel is disabled when two of the associated ChanelEdgePolicies // have their disabled bit on. func (c *KVStore) DisabledChannelIDs( - v lnwire.GossipVersion) ([]uint64, error) { + _ context.Context, v lnwire.GossipVersion) ([]uint64, error) { if v != lnwire.GossipVersion1 { return nil, ErrVersionNotSupportedForKVDB @@ -1317,7 +1318,7 @@ func (c *KVStore) addChannelEdge(tx kvdb.RwTx, // the edge was updated for both directed edges are returned along with the // boolean. If it is not found, then the zombie index is checked and its // result is returned as the second boolean. -func (c *KVStore) HasV1ChannelEdge( +func (c *KVStore) HasV1ChannelEdge(_ context.Context, chanID uint64) (time.Time, time.Time, bool, bool, error) { var ( @@ -1426,20 +1427,20 @@ func (c *KVStore) HasV1ChannelEdge( // passed channel ID and gossip version, and false otherwise. If it is not // found, then the zombie index is checked and its result is returned as the // second boolean. -func (c *KVStore) HasChannelEdge(v lnwire.GossipVersion, +func (c *KVStore) HasChannelEdge(ctx context.Context, v lnwire.GossipVersion, chanID uint64) (bool, bool, error) { if v != lnwire.GossipVersion1 { return false, false, ErrVersionNotSupportedForKVDB } - _, _, exists, isZombie, err := c.HasV1ChannelEdge(chanID) + _, _, exists, isZombie, err := c.HasV1ChannelEdge(ctx, chanID) return exists, isZombie, err } // AddEdgeProof sets the proof of an existing edge in the graph database. -func (c *KVStore) AddEdgeProof(chanID lnwire.ShortChannelID, +func (c *KVStore) AddEdgeProof(_ context.Context, chanID lnwire.ShortChannelID, proof *models.ChannelAuthProof) error { // We only support v1 channel proofs in the KVStore. @@ -1491,7 +1492,7 @@ const ( // with the current UTXO state. A slice of channels that have been closed by // the target block along with any pruned nodes are returned if the function // succeeds without error. -func (c *KVStore) PruneGraph(spentOutputs []*wire.OutPoint, +func (c *KVStore) PruneGraph(_ context.Context, spentOutputs []*wire.OutPoint, blockHash *chainhash.Hash, blockHeight uint32) ( []*models.ChannelEdgeInfo, []route.Vertex, error) { @@ -1619,7 +1620,7 @@ func (c *KVStore) PruneGraph(spentOutputs []*wire.OutPoint, // any nodes from the channel graph that are currently unconnected. This ensure // that we only maintain a graph of reachable nodes. In the event that a pruned // node gains more channels, it will be re-added back to the graph. -func (c *KVStore) PruneGraphNodes() ([]route.Vertex, error) { +func (c *KVStore) PruneGraphNodes(_ context.Context) ([]route.Vertex, error) { var prunedNodes []route.Vertex err := kvdb.Update(c.db, func(tx kvdb.RwTx) error { nodes := tx.ReadWriteBucket(nodeBucket) @@ -1759,8 +1760,8 @@ func (c *KVStore) pruneGraphNodes(nodes kvdb.RwBucket, // set to the last prune height valid for the remaining chain. // Channels that were removed from the graph resulting from the // disconnected block are returned. -func (c *KVStore) DisconnectBlockAtHeight(height uint32) ( - []*models.ChannelEdgeInfo, error) { +func (c *KVStore) DisconnectBlockAtHeight(_ context.Context, + height uint32) ([]*models.ChannelEdgeInfo, error) { // Every channel having a ShortChannelID starting at 'height' // will no longer be confirmed. @@ -1886,7 +1887,7 @@ func (c *KVStore) DisconnectBlockAtHeight(height uint32) ( // used to prune channels in the graph. Knowing the "prune tip" allows callers // to tell if the graph is currently in sync with the current best known UTXO // state. -func (c *KVStore) PruneTip() (*chainhash.Hash, uint32, error) { +func (c *KVStore) PruneTip(_ context.Context) (*chainhash.Hash, uint32, error) { var ( tipHash chainhash.Hash tipHeight uint32 @@ -1933,8 +1934,9 @@ func (c *KVStore) PruneTip() (*chainhash.Hash, uint32, error) { // that we require the node that failed to send the fresh update to be the one // that resurrects the channel from its zombie state. The markZombie bool // denotes whether or not to mark the channel as a zombie. -func (c *KVStore) DeleteChannelEdges(v lnwire.GossipVersion, - strictZombiePruning, markZombie bool, chanIDs ...uint64) ( +func (c *KVStore) DeleteChannelEdges(_ context.Context, + v lnwire.GossipVersion, strictZombiePruning, markZombie bool, + chanIDs ...uint64) ( []*models.ChannelEdgeInfo, error) { if v != lnwire.GossipVersion1 { @@ -2004,7 +2006,7 @@ func (c *KVStore) DeleteChannelEdges(v lnwire.GossipVersion, // ChannelID attempt to lookup the 8-byte compact channel ID which maps to the // passed channel point (outpoint). If the passed channel doesn't exist within // the database, then ErrEdgeNotFound is returned. -func (c *KVStore) ChannelID(v lnwire.GossipVersion, +func (c *KVStore) ChannelID(_ context.Context, v lnwire.GossipVersion, chanPoint *wire.OutPoint) (uint64, error) { if v != lnwire.GossipVersion1 { @@ -2377,7 +2379,8 @@ func (c *KVStore) fetchNextChanUpdateBatch( // ChanUpdatesInHorizon returns all the known channel edges which have at least // one edge that has an update timestamp within the specified horizon. -func (c *KVStore) ChanUpdatesInHorizon(startTime, endTime time.Time, +func (c *KVStore) ChanUpdatesInHorizon(_ context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[ChannelEdge, error] { cfg := defaultIteratorConfig() @@ -2627,7 +2630,7 @@ func (c *KVStore) fetchNextNodeBatch( // NodeUpdatesInHorizon returns all the known lightning node which have an // update timestamp within the passed range. -func (c *KVStore) NodeUpdatesInHorizon(startTime, +func (c *KVStore) NodeUpdatesInHorizon(_ context.Context, startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[*models.Node, error] { @@ -2676,8 +2679,8 @@ func (c *KVStore) NodeUpdatesInHorizon(startTime, // passed in. This method can be used by callers to determine the set of // channels another peer knows of that we don't. The ChannelUpdateInfos for the // known zombies is also returned. -func (c *KVStore) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo) ([]uint64, - []ChannelUpdateInfo, error) { +func (c *KVStore) FilterKnownChanIDs(_ context.Context, + chansInfo []ChannelUpdateInfo) ([]uint64, []ChannelUpdateInfo, error) { var ( newChanIDs []uint64 @@ -2815,7 +2818,7 @@ type BlockChannelRange struct { // up after a period of time offline. If withTimestamps is true then the // timestamp info of the latest received channel update messages of the channel // will be included in the response. -func (c *KVStore) FilterChannelRange(startHeight, +func (c *KVStore) FilterChannelRange(_ context.Context, startHeight, endHeight uint32, withTimestamps bool) ([]BlockChannelRange, error) { startChanID := &lnwire.ShortChannelID{ @@ -2960,7 +2963,7 @@ func (c *KVStore) FilterChannelRange(startHeight, // skipped and the result will contain only those edges that exist at the time // of the query. This can be used to respond to peer queries that are seeking to // fill in gaps in their view of the channel graph. -func (c *KVStore) FetchChanInfos(v lnwire.GossipVersion, +func (c *KVStore) FetchChanInfos(_ context.Context, v lnwire.GossipVersion, chanIDs []uint64) ([]ChannelEdge, error) { if v != lnwire.GossipVersion1 { @@ -3896,8 +3899,9 @@ func computeEdgePolicyKeys(info *models.ChannelEdgeInfo) ([]byte, []byte) { // found, then ErrEdgeNotFound is returned. A struct which houses the general // information for the channel itself is returned as well as two structs that // contain the routing policies for the channel in either direction. -func (c *KVStore) FetchChannelEdgesByOutpoint(v lnwire.GossipVersion, - op *wire.OutPoint) (*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, +func (c *KVStore) FetchChannelEdgesByOutpoint(_ context.Context, + v lnwire.GossipVersion, op *wire.OutPoint) ( + *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { var ( @@ -3986,8 +3990,9 @@ func (c *KVStore) FetchChannelEdgesByOutpoint(v lnwire.GossipVersion, // ErrZombieEdge an be returned if the edge is currently marked as a zombie // within the database. In this case, the ChannelEdgePolicy's will be nil, and // the ChannelEdgeInfo will only include the public keys of each node. -func (c *KVStore) FetchChannelEdgesByID(v lnwire.GossipVersion, - chanID uint64) (*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, +func (c *KVStore) FetchChannelEdgesByID(_ context.Context, + v lnwire.GossipVersion, chanID uint64) ( + *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { if v != lnwire.GossipVersion1 { @@ -4098,8 +4103,8 @@ func (c *KVStore) FetchChannelEdgesByID(v lnwire.GossipVersion, // IsPublicNode is a helper method that determines whether the node with the // given public key is seen as a public node in the graph from the graph's // source node's point of view. -func (c *KVStore) IsPublicNode(v lnwire.GossipVersion, pubKey [33]byte) (bool, - error) { +func (c *KVStore) IsPublicNode(_ context.Context, v lnwire.GossipVersion, + pubKey [33]byte) (bool, error) { if v != lnwire.GossipVersion1 { return false, ErrVersionNotSupportedForKVDB @@ -4175,7 +4180,7 @@ func (e *EdgePoint) String() string { // within the known channel graph. The set of UTXO's (along with their scripts) // returned are the ones that need to be watched on chain to detect channel // closes on the resident blockchain. -func (c *KVStore) ChannelView() ([]EdgePoint, error) { +func (c *KVStore) ChannelView(_ context.Context) ([]EdgePoint, error) { var edgePoints []EdgePoint if err := kvdb.View(c.db, func(tx kvdb.RTx) error { // We're going to iterate over the entire channel index, so @@ -4241,7 +4246,7 @@ func (c *KVStore) ChannelView() ([]EdgePoint, error) { // MarkEdgeZombie attempts to mark a channel identified by its channel ID as a // zombie. This method is used on an ad-hoc basis, when channels need to be // marked as zombies outside the normal pruning cycle. -func (c *KVStore) MarkEdgeZombie(chanID uint64, +func (c *KVStore) MarkEdgeZombie(_ context.Context, chanID uint64, pubKey1, pubKey2 [33]byte) error { c.cacheMu.Lock() @@ -4287,7 +4292,7 @@ func markEdgeZombie(zombieIndex kvdb.RwBucket, chanID uint64, pubKey1, } // MarkEdgeLive clears an edge from our zombie index, deeming it as live. -func (c *KVStore) MarkEdgeLive(chanID uint64) error { +func (c *KVStore) MarkEdgeLive(_ context.Context, chanID uint64) error { c.cacheMu.Lock() defer c.cacheMu.Unlock() @@ -4342,7 +4347,7 @@ func (c *KVStore) markEdgeLiveUnsafe(tx kvdb.RwTx, chanID uint64) error { // IsZombieEdge returns whether the edge is considered zombie. If it is a // zombie, then the two node public keys corresponding to this edge are also // returned. -func (c *KVStore) IsZombieEdge(v lnwire.GossipVersion, +func (c *KVStore) IsZombieEdge(_ context.Context, v lnwire.GossipVersion, chanID uint64) (bool, [33]byte, [33]byte, error) { var ( @@ -4403,7 +4408,7 @@ func isZombieEdge(zombieIndex kvdb.RBucket, } // NumZombies returns the current number of zombie channels in the graph. -func (c *KVStore) NumZombies() (uint64, error) { +func (c *KVStore) NumZombies(_ context.Context) (uint64, error) { var numZombies uint64 err := kvdb.View(c.db, func(tx kvdb.RTx) error { edges := tx.ReadBucket(edgeBucket) @@ -4432,7 +4437,9 @@ func (c *KVStore) NumZombies() (uint64, error) { // PutClosedScid stores a SCID for a closed channel in the database. This is so // that we can ignore channel announcements that we know to be closed without // having to validate them and fetch a block. -func (c *KVStore) PutClosedScid(scid lnwire.ShortChannelID) error { +func (c *KVStore) PutClosedScid(_ context.Context, + scid lnwire.ShortChannelID) error { + return kvdb.Update(c.db, func(tx kvdb.RwTx) error { closedScids, err := tx.CreateTopLevelBucket(closedScidBucket) if err != nil { @@ -4449,7 +4456,9 @@ func (c *KVStore) PutClosedScid(scid lnwire.ShortChannelID) error { // IsClosedScid checks whether a channel identified by the passed in scid is // closed. This helps avoid having to perform expensive validation checks. // TODO: Add an LRU cache to cut down on disc reads. -func (c *KVStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) { +func (c *KVStore) IsClosedScid(_ context.Context, + scid lnwire.ShortChannelID) (bool, error) { + var isClosed bool err := kvdb.View(c.db, func(tx kvdb.RTx) error { closedScids := tx.ReadBucket(closedScidBucket) @@ -4478,8 +4487,8 @@ func (c *KVStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) { // GraphSession will provide the call-back with access to a NodeTraverser // instance which can be used to perform queries against the channel graph. -func (c *KVStore) GraphSession(cb func(graph NodeTraverser) error, - reset func()) error { +func (c *KVStore) GraphSession(_ context.Context, + cb func(graph NodeTraverser) error, reset func()) error { return c.db.View(func(tx walletdb.ReadTx) error { return cb(&nodeTraverserSession{ @@ -4501,8 +4510,8 @@ type nodeTraverserSession struct { // // NOTE: Part of the NodeTraverser interface. func (c *nodeTraverserSession) ForEachNodeDirectedChannel( - nodePub route.Vertex, cb func(channel *DirectedChannel) error, - _ func()) error { + _ context.Context, nodePub route.Vertex, + cb func(channel *DirectedChannel) error, _ func()) error { return c.db.forEachNodeDirectedChannel(c.tx, nodePub, cb, func() {}) } @@ -4511,7 +4520,8 @@ func (c *nodeTraverserSession) ForEachNodeDirectedChannel( // unknown, assume no additional features are supported. // // NOTE: Part of the NodeTraverser interface. -func (c *nodeTraverserSession) FetchNodeFeatures(nodePub route.Vertex) ( +func (c *nodeTraverserSession) FetchNodeFeatures(_ context.Context, + nodePub route.Vertex) ( *lnwire.FeatureVector, error) { return c.db.fetchNodeFeatures(c.tx, nodePub) diff --git a/graph/db/notifications.go b/graph/db/notifications.go index 54a748c66..1c8a88986 100644 --- a/graph/db/notifications.go +++ b/graph/db/notifications.go @@ -1,6 +1,7 @@ package graphdb import ( + "context" "errors" "fmt" "image/color" @@ -201,11 +202,11 @@ func (c *ChannelGraph) notifyTopologyChange(topologyDiff *TopologyChange) { // // NOTE: must be run inside goroutine and must only ever be called from within // handleTopologySubscriptions. -func (c *ChannelGraph) handleTopologyUpdate(update any) { +func (c *ChannelGraph) handleTopologyUpdate(ctx context.Context, update any) { defer c.wg.Done() topChange := &TopologyChange{} - err := c.addToTopologyChange(topChange, update) + err := c.addToTopologyChange(ctx, topChange, update) if err != nil { log.Errorf("unable to update topology change notification: %v", err) @@ -375,8 +376,8 @@ type ChannelEdgeUpdate struct { // constitutes. This function will also fetch any required auxiliary // information required to create the topology change update from the graph // database. -func (c *ChannelGraph) addToTopologyChange(update *TopologyChange, - msg any) error { +func (c *ChannelGraph) addToTopologyChange(ctx context.Context, + update *TopologyChange, msg any) error { switch m := msg.(type) { @@ -412,7 +413,9 @@ func (c *ChannelGraph) addToTopologyChange(update *TopologyChange, // We'll need to fetch the edge's information from the database // in order to get the information concerning which nodes are // being connected. - edgeInfo, _, _, err := c.FetchChannelEdgesByID(m.ChannelID) + edgeInfo, _, _, err := c.FetchChannelEdgesByID( + ctx, m.ChannelID, + ) if err != nil { return fmt.Errorf("unable fetch channel edge: %w", err) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index a04a70aba..fea367a94 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -447,10 +447,9 @@ func (s *SQLStore) DeleteNode(ctx context.Context, v lnwire.GossipVersion, // known for the node, an empty feature vector is returned. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (s *SQLStore) FetchNodeFeatures(v lnwire.GossipVersion, - nodePub route.Vertex) (*lnwire.FeatureVector, error) { - - ctx := context.TODO() +func (s *SQLStore) FetchNodeFeatures(ctx context.Context, + v lnwire.GossipVersion, nodePub route.Vertex) (*lnwire.FeatureVector, + error) { return fetchNodeFeatures(ctx, s.db, v, nodePub) } @@ -461,10 +460,9 @@ func (s *SQLStore) FetchNodeFeatures(v lnwire.GossipVersion, // // NOTE: part of the Store interface. func (s *SQLStore) DisabledChannelIDs( - v lnwire.GossipVersion) ([]uint64, error) { + ctx context.Context, v lnwire.GossipVersion) ([]uint64, error) { var ( - ctx = context.TODO() chanIDs []uint64 ) err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { @@ -612,7 +610,8 @@ func (s *SQLStore) SetSourceNode(ctx context.Context, // announcements. // // 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] { cfg := defaultIteratorConfig() @@ -622,7 +621,6 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime, endTime time.Time, return func(yield func(*models.Node, error) bool) { var ( - ctx = context.TODO() lastUpdateTime sql.NullInt64 lastPubKey = make([]byte, 33) hasMore = true @@ -1008,11 +1006,9 @@ func (s *SQLStore) ForEachNode(ctx context.Context, // Unknown policies are passed into the callback as nil values. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (s *SQLStore) ForEachNodeDirectedChannel(v lnwire.GossipVersion, - nodePub route.Vertex, cb func(channel *DirectedChannel) error, - reset func()) error { - - var ctx = context.TODO() +func (s *SQLStore) ForEachNodeDirectedChannel(ctx context.Context, + v lnwire.GossipVersion, nodePub route.Vertex, + cb func(channel *DirectedChannel) error, reset func()) error { return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodeDirectedChannel(ctx, db, v, nodePub, cb) @@ -1174,7 +1170,8 @@ func (s *SQLStore) updateChanCacheBatch(v lnwire.GossipVersion, // 6. Repeat with updated pagination cursor until no more results // // NOTE: This is part of the Store interface. -func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time, +func (s *SQLStore) ChanUpdatesInHorizon(ctx context.Context, + startTime, endTime time.Time, opts ...IteratorOption) iter.Seq2[ChannelEdge, error] { // Apply options. @@ -1185,7 +1182,6 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, endTime time.Time, return func(yield func(ChannelEdge, error) bool) { var ( - ctx = context.TODO() edgesSeen = make(map[uint64]struct{}) edgesToCache = make(map[uint64]ChannelEdge) hits int @@ -1568,12 +1564,11 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, withAddrs bool, // // NOTE: this method is like ForEachChannel but fetches only the data // required for the graph cache. -func (s *SQLStore) ForEachChannelCacheable(v lnwire.GossipVersion, +func (s *SQLStore) ForEachChannelCacheable(ctx context.Context, + v lnwire.GossipVersion, cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, reset func()) error { - ctx := context.TODO() - if !isKnownGossipVersion(v) { return fmt.Errorf("unsupported gossip version: %d", v) } @@ -1669,11 +1664,10 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, // will be included in the response. // // NOTE: This is part of the Store interface. -func (s *SQLStore) FilterChannelRange(startHeight, endHeight uint32, - withTimestamps bool) ([]BlockChannelRange, error) { +func (s *SQLStore) FilterChannelRange(ctx context.Context, startHeight, + endHeight uint32, withTimestamps bool) ([]BlockChannelRange, error) { var ( - ctx = context.TODO() startSCID = &lnwire.ShortChannelID{ BlockHeight: startHeight, } @@ -1789,11 +1783,9 @@ func (s *SQLStore) FilterChannelRange(startHeight, endHeight uint32, // marked as zombies outside the normal pruning cycle. // // NOTE: part of the Store interface. -func (s *SQLStore) MarkEdgeZombie(chanID uint64, +func (s *SQLStore) MarkEdgeZombie(ctx context.Context, chanID uint64, pubKey1, pubKey2 [33]byte) error { - ctx := context.TODO() - s.cacheMu.Lock() defer s.cacheMu.Unlock() @@ -1823,12 +1815,11 @@ func (s *SQLStore) MarkEdgeZombie(chanID uint64, // MarkEdgeLive clears an edge from our zombie index, deeming it as live. // // NOTE: part of the Store interface. -func (s *SQLStore) MarkEdgeLive(chanID uint64) error { +func (s *SQLStore) MarkEdgeLive(ctx context.Context, chanID uint64) error { s.cacheMu.Lock() defer s.cacheMu.Unlock() var ( - ctx = context.TODO() chanIDB = channelIDToBytes(chanID) ) @@ -1874,11 +1865,10 @@ func (s *SQLStore) MarkEdgeLive(chanID uint64) error { // returned. // // NOTE: part of the Store interface. -func (s *SQLStore) IsZombieEdge(v lnwire.GossipVersion, +func (s *SQLStore) IsZombieEdge(ctx context.Context, v lnwire.GossipVersion, chanID uint64) (bool, [33]byte, [33]byte, error) { var ( - ctx = context.TODO() isZombie bool pubKey1, pubKey2 route.Vertex chanIDB = channelIDToBytes(chanID) @@ -1922,9 +1912,8 @@ func (s *SQLStore) IsZombieEdge(v lnwire.GossipVersion, // NumZombies returns the current number of zombie channels in the graph. // // NOTE: part of the Store interface. -func (s *SQLStore) NumZombies() (uint64, error) { +func (s *SQLStore) NumZombies(ctx context.Context) (uint64, error) { var ( - ctx = context.TODO() numZombies uint64 ) err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { @@ -1957,8 +1946,9 @@ func (s *SQLStore) NumZombies() (uint64, error) { // denotes whether to mark the channel as a zombie. // // NOTE: part of the Store interface. -func (s *SQLStore) DeleteChannelEdges(v lnwire.GossipVersion, - strictZombiePruning, markZombie bool, chanIDs ...uint64) ( +func (s *SQLStore) DeleteChannelEdges(ctx context.Context, + v lnwire.GossipVersion, strictZombiePruning, markZombie bool, + chanIDs ...uint64) ( []*models.ChannelEdgeInfo, error) { s.cacheMu.Lock() @@ -1971,10 +1961,7 @@ func (s *SQLStore) DeleteChannelEdges(v lnwire.GossipVersion, chanLookup[chanID] = struct{}{} } - var ( - ctx = context.TODO() - edges []*models.ChannelEdgeInfo - ) + var edges []*models.ChannelEdgeInfo err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { // First, collect all channel rows. var channelRows []sqlc.GetChannelsBySCIDWithPoliciesRow @@ -2063,12 +2050,12 @@ func (s *SQLStore) DeleteChannelEdges(v lnwire.GossipVersion, // the ChannelEdgeInfo will only include the public keys of each node. // // NOTE: part of the Store interface. -func (s *SQLStore) FetchChannelEdgesByID(v lnwire.GossipVersion, - chanID uint64) (*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, +func (s *SQLStore) FetchChannelEdgesByID(ctx context.Context, + v lnwire.GossipVersion, chanID uint64) ( + *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { var ( - ctx = context.TODO() edge *models.ChannelEdgeInfo policy1, policy2 *models.ChannelEdgePolicy chanIDB = channelIDToBytes(chanID) @@ -2180,12 +2167,12 @@ func (s *SQLStore) FetchChannelEdgesByID(v lnwire.GossipVersion, // contain the routing policies for the channel in either direction. // // NOTE: part of the Store interface. -func (s *SQLStore) FetchChannelEdgesByOutpoint(v lnwire.GossipVersion, - op *wire.OutPoint) (*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, +func (s *SQLStore) FetchChannelEdgesByOutpoint(ctx context.Context, + v lnwire.GossipVersion, op *wire.OutPoint) ( + *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { var ( - ctx = context.TODO() edge *models.ChannelEdgeInfo policy1, policy2 *models.ChannelEdgePolicy ) @@ -2257,10 +2244,8 @@ func (s *SQLStore) FetchChannelEdgesByOutpoint(v lnwire.GossipVersion, // result is returned as the second boolean. // // NOTE: part of the Store interface. -func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, - bool, error) { - - ctx := context.TODO() +func (s *SQLStore) HasV1ChannelEdge(ctx context.Context, + chanID uint64) (time.Time, time.Time, bool, bool, error) { var ( exists bool @@ -2376,8 +2361,8 @@ func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, // result is returned as the second boolean. // // NOTE: part of the Store interface. -func (s *SQLStore) HasChannelEdge(v lnwire.GossipVersion, - chanID uint64) (bool, bool, error) { +func (s *SQLStore) HasChannelEdge(ctx context.Context, + v lnwire.GossipVersion, chanID uint64) (bool, bool, error) { if !isKnownGossipVersion(v) { return false, false, fmt.Errorf( @@ -2385,8 +2370,6 @@ func (s *SQLStore) HasChannelEdge(v lnwire.GossipVersion, ) } - ctx := context.TODO() - var ( exists bool isZombie bool @@ -2527,11 +2510,10 @@ func (s *SQLStore) HasChannelEdge(v lnwire.GossipVersion, // the database, then ErrEdgeNotFound is returned. // // NOTE: part of the Store interface. -func (s *SQLStore) ChannelID(v lnwire.GossipVersion, +func (s *SQLStore) ChannelID(ctx context.Context, v lnwire.GossipVersion, chanPoint *wire.OutPoint) (uint64, error) { var ( - ctx = context.TODO() channelID uint64 ) err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { @@ -2564,10 +2546,9 @@ func (s *SQLStore) ChannelID(v lnwire.GossipVersion, // source node's point of view. // // NOTE: part of the Store interface. -func (s *SQLStore) IsPublicNode(v lnwire.GossipVersion, pubKey [33]byte) (bool, - error) { +func (s *SQLStore) IsPublicNode(ctx context.Context, v lnwire.GossipVersion, + pubKey [33]byte) (bool, error) { - ctx := context.TODO() if !isKnownGossipVersion(v) { return false, fmt.Errorf("unsupported gossip version: %d", v) } @@ -2599,11 +2580,10 @@ func (s *SQLStore) IsPublicNode(v lnwire.GossipVersion, pubKey [33]byte) (bool, // fill in gaps in their view of the channel graph. // // NOTE: part of the Store interface. -func (s *SQLStore) FetchChanInfos(v lnwire.GossipVersion, - chanIDs []uint64) ([]ChannelEdge, error) { +func (s *SQLStore) FetchChanInfos(ctx context.Context, + v lnwire.GossipVersion, chanIDs []uint64) ([]ChannelEdge, error) { var ( - ctx = context.TODO() edges = make(map[uint64]ChannelEdge) ) err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { @@ -2699,11 +2679,10 @@ func (s *SQLStore) forEachChanWithPoliciesInSCIDList(ctx context.Context, // known zombies is also returned. // // NOTE: part of the Store interface. -func (s *SQLStore) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo) ([]uint64, - []ChannelUpdateInfo, error) { +func (s *SQLStore) FilterKnownChanIDs(ctx context.Context, + chansInfo []ChannelUpdateInfo) ([]uint64, []ChannelUpdateInfo, error) { var ( - ctx = context.TODO() newChanIDs []uint64 knownZombies []ChannelUpdateInfo infoLookup = make( @@ -2823,8 +2802,8 @@ func (s *SQLStore) forEachChanInSCIDList(ctx context.Context, db SQLQueries, // source nodes. // // NOTE: part of the Store interface. -func (s *SQLStore) PruneGraphNodes() ([]route.Vertex, error) { - var ctx = context.TODO() +func (s *SQLStore) PruneGraphNodes(ctx context.Context) ( + []route.Vertex, error) { var prunedNodes []route.Vertex err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { @@ -2852,11 +2831,10 @@ func (s *SQLStore) PruneGraphNodes() ([]route.Vertex, error) { // succeeds without error. // // NOTE: part of the Store interface. -func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint, - blockHash *chainhash.Hash, blockHeight uint32) ( - []*models.ChannelEdgeInfo, []route.Vertex, error) { - - ctx := context.TODO() +func (s *SQLStore) PruneGraph(ctx context.Context, + spentOutputs []*wire.OutPoint, blockHash *chainhash.Hash, + blockHeight uint32) ([]*models.ChannelEdgeInfo, []route.Vertex, + error) { s.cacheMu.Lock() defer s.cacheMu.Unlock() @@ -3007,11 +2985,8 @@ func (s *SQLStore) deleteChannels(ctx context.Context, db SQLQueries, // closes on the resident blockchain. // // NOTE: part of the Store interface. -func (s *SQLStore) ChannelView() ([]EdgePoint, error) { - var ( - ctx = context.TODO() - edgePoints []EdgePoint - ) +func (s *SQLStore) ChannelView(ctx context.Context) ([]EdgePoint, error) { + var edgePoints []EdgePoint err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { handleChannel := func(_ context.Context, @@ -3074,9 +3049,10 @@ func (s *SQLStore) ChannelView() ([]EdgePoint, error) { // state. // // NOTE: part of the Store interface. -func (s *SQLStore) PruneTip() (*chainhash.Hash, uint32, error) { +func (s *SQLStore) PruneTip(ctx context.Context) (*chainhash.Hash, uint32, + error) { + var ( - ctx = context.TODO() tipHash chainhash.Hash tipHeight uint32 ) @@ -3136,10 +3112,8 @@ func (s *SQLStore) pruneGraphNodes(ctx context.Context, // disconnected block are returned. // // NOTE: part of the Store interface. -func (s *SQLStore) DisconnectBlockAtHeight(height uint32) ( - []*models.ChannelEdgeInfo, error) { - - ctx := context.TODO() +func (s *SQLStore) DisconnectBlockAtHeight(ctx context.Context, + height uint32) ([]*models.ChannelEdgeInfo, error) { var ( // Every channel having a ShortChannelID starting at 'height' @@ -3224,8 +3198,8 @@ func (s *SQLStore) DisconnectBlockAtHeight(height uint32) ( // AddEdgeProof sets the proof of an existing edge in the graph database. // // NOTE: part of the Store interface. -func (s *SQLStore) AddEdgeProof(scid lnwire.ShortChannelID, - proof *models.ChannelAuthProof) error { +func (s *SQLStore) AddEdgeProof(ctx context.Context, + scid lnwire.ShortChannelID, proof *models.ChannelAuthProof) error { if !isKnownGossipVersion(proof.Version) { return fmt.Errorf("unsupported gossip version: %d", @@ -3233,7 +3207,6 @@ func (s *SQLStore) AddEdgeProof(scid lnwire.ShortChannelID, } var ( - ctx = context.TODO() scidBytes = channelIDToBytes(scid.ToUint64()) ) @@ -3298,9 +3271,10 @@ func (s *SQLStore) AddEdgeProof(scid lnwire.ShortChannelID, // having to validate them and fetch a block. // // NOTE: part of the Store interface. -func (s *SQLStore) PutClosedScid(scid lnwire.ShortChannelID) error { +func (s *SQLStore) PutClosedScid(ctx context.Context, + scid lnwire.ShortChannelID) error { + var ( - ctx = context.TODO() chanIDB = channelIDToBytes(scid.ToUint64()) ) @@ -3313,9 +3287,10 @@ func (s *SQLStore) PutClosedScid(scid lnwire.ShortChannelID) error { // closed. This helps avoid having to perform expensive validation checks. // // NOTE: part of the Store interface. -func (s *SQLStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) { +func (s *SQLStore) IsClosedScid(ctx context.Context, + scid lnwire.ShortChannelID) (bool, error) { + var ( - ctx = context.TODO() isClosed bool chanIDB = channelIDToBytes(scid.ToUint64()) ) @@ -3341,10 +3316,8 @@ func (s *SQLStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) { // instance which can be used to perform queries against the channel graph. // // NOTE: part of the Store interface. -func (s *SQLStore) GraphSession(cb func(graph NodeTraverser) error, - reset func()) error { - - var ctx = context.TODO() +func (s *SQLStore) GraphSession(ctx context.Context, + cb func(graph NodeTraverser) error, reset func()) error { return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return cb(newSQLNodeTraverser(db, s.cfg.ChainHash)) @@ -3377,10 +3350,8 @@ func newSQLNodeTraverser(db SQLQueries, // // NOTE: Part of the NodeTraverser interface. func (s *sqlNodeTraverser) ForEachNodeDirectedChannel( - nodePub route.Vertex, cb func(channel *DirectedChannel) error, - _ func()) error { - - ctx := context.TODO() + ctx context.Context, nodePub route.Vertex, + cb func(channel *DirectedChannel) error, _ func()) error { return forEachNodeDirectedChannel( ctx, s.db, lnwire.GossipVersion1, nodePub, cb, @@ -3391,11 +3362,10 @@ func (s *sqlNodeTraverser) ForEachNodeDirectedChannel( // unknown, assume no additional features are supported. // // NOTE: Part of the NodeTraverser interface. -func (s *sqlNodeTraverser) FetchNodeFeatures(nodePub route.Vertex) ( +func (s *sqlNodeTraverser) FetchNodeFeatures(ctx context.Context, + nodePub route.Vertex) ( *lnwire.FeatureVector, error) { - ctx := context.TODO() - return fetchNodeFeatures(ctx, s.db, lnwire.GossipVersion1, nodePub) } diff --git a/lnrpc/invoicesrpc/addinvoice.go b/lnrpc/invoicesrpc/addinvoice.go index aba5da4df..9afba7678 100644 --- a/lnrpc/invoicesrpc/addinvoice.go +++ b/lnrpc/invoicesrpc/addinvoice.go @@ -522,8 +522,16 @@ func AddInvoice(ctx context.Context, cfg *AddInvoiceConfig, //nolint:ll paths, err := blindedpath.BuildBlindedPaymentPaths( &blindedpath.BuildBlindedPathCfg{ - FindRoutes: cfg.QueryBlindedRoutes, - FetchChannelEdgesByID: cfg.Graph.FetchChannelEdgesByID, + FindRoutes: cfg.QueryBlindedRoutes, + FetchChannelEdgesByID: func(chanID uint64) ( + *models.ChannelEdgeInfo, + *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy, error) { + + return cfg.Graph.FetchChannelEdgesByID( + context.TODO(), chanID, + ) + }, FetchOurOpenChannels: cfg.ChanDB.FetchAllOpenChannels, PathID: paymentAddr[:], ValueMsat: invoice.Value, @@ -787,12 +795,23 @@ func newSelectHopHintsCfg(invoicesCfg *AddInvoiceConfig, maxHopHints int) *SelectHopHintsCfg { return &SelectHopHintsCfg{ - FetchAllChannels: invoicesCfg.ChanDB.FetchAllChannels, - IsChannelActive: invoicesCfg.IsChannelActive, - IsPublicNode: invoicesCfg.Graph.IsPublicNode, - FetchChannelEdgesByID: invoicesCfg.Graph.FetchChannelEdgesByID, - GetAlias: invoicesCfg.GetAlias, - MaxHopHints: maxHopHints, + FetchAllChannels: invoicesCfg.ChanDB.FetchAllChannels, + IsChannelActive: invoicesCfg.IsChannelActive, + IsPublicNode: func(pubKey [33]byte) (bool, error) { + return invoicesCfg.Graph.IsPublicNode( + context.TODO(), pubKey, + ) + }, + FetchChannelEdgesByID: func(chanID uint64) ( + *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy, error) { + + return invoicesCfg.Graph.FetchChannelEdgesByID( + context.TODO(), chanID, + ) + }, + GetAlias: invoicesCfg.GetAlias, + MaxHopHints: maxHopHints, } } diff --git a/lnrpc/invoicesrpc/interfaces.go b/lnrpc/invoicesrpc/interfaces.go index df47d2f24..99a8bef23 100644 --- a/lnrpc/invoicesrpc/interfaces.go +++ b/lnrpc/invoicesrpc/interfaces.go @@ -1,6 +1,8 @@ package invoicesrpc import ( + "context" + "github.com/lightningnetwork/lnd/graph/db/models" ) @@ -9,11 +11,12 @@ type GraphSource interface { // FetchChannelEdgesByID attempts to look up the two directed edges for // the channel identified by the channel ID. If the channel can't be // found, then graphdb.ErrEdgeNotFound is returned. - FetchChannelEdgesByID(chanID uint64) (*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) + FetchChannelEdgesByID(ctx context.Context, chanID uint64) ( + *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy, error) // IsPublicNode is a helper method that determines whether the node with // the given public key is seen as a public node in the graph from the // graph's source node's point of view. - IsPublicNode(pubKey [33]byte) (bool, error) + IsPublicNode(ctx context.Context, pubKey [33]byte) (bool, error) } diff --git a/netann/chan_status_manager.go b/netann/chan_status_manager.go index feb3a5dd1..b21aeb18a 100644 --- a/netann/chan_status_manager.go +++ b/netann/chan_status_manager.go @@ -1,6 +1,7 @@ package netann import ( + "context" "errors" "sync" "time" @@ -654,7 +655,9 @@ func (m *ChanStatusManager) fetchLastChanUpdateByOutPoint(op wire.OutPoint) ( *lnwire.ChannelUpdate1, bool, error) { // Get the edge info and policies for this channel from the graph. - info, edge1, edge2, err := m.cfg.Graph.FetchChannelEdgesByOutpoint(&op) + info, edge1, edge2, err := m.cfg.Graph.FetchChannelEdgesByOutpoint( + context.TODO(), &op, + ) if err != nil { return nil, false, err } diff --git a/netann/chan_status_manager_test.go b/netann/chan_status_manager_test.go index 59ce324c6..02669ce4a 100644 --- a/netann/chan_status_manager_test.go +++ b/netann/chan_status_manager_test.go @@ -2,6 +2,7 @@ package netann_test import ( "bytes" + "context" "crypto/rand" "encoding/binary" "fmt" @@ -173,7 +174,8 @@ func (g *mockGraph) FetchAllOpenChannels() ([]*channeldb.OpenChannel, error) { } func (g *mockGraph) FetchChannelEdgesByOutpoint( - op *wire.OutPoint) (*models.ChannelEdgeInfo, + _ context.Context, op *wire.OutPoint) ( + *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { g.mu.Lock() diff --git a/netann/interface.go b/netann/interface.go index aa559435d..78acc24cd 100644 --- a/netann/interface.go +++ b/netann/interface.go @@ -1,6 +1,8 @@ package netann import ( + "context" + "github.com/btcsuite/btcd/wire" "github.com/lightningnetwork/lnd/channeldb" "github.com/lightningnetwork/lnd/graph/db/models" @@ -19,6 +21,7 @@ type DB interface { type ChannelGraph interface { // FetchChannelEdgesByOutpoint returns the channel edge info and most // recent channel edge policies for a given outpoint. - FetchChannelEdgesByOutpoint(*wire.OutPoint) (*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) + FetchChannelEdgesByOutpoint(context.Context, *wire.OutPoint) ( + *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy, error) } diff --git a/peer/brontide.go b/peer/brontide.go index d92f9c459..0f4d6e349 100644 --- a/peer/brontide.go +++ b/peer/brontide.go @@ -1217,7 +1217,7 @@ func (p *Brontide) loadActiveChannels(chans []*channeldb.OpenChannel) ( // the database. graph := p.cfg.ChannelGraph info, p1, p2, err := graph.FetchChannelEdgesByOutpoint( - &chanPoint, + context.TODO(), &chanPoint, ) if err != nil && !errors.Is(err, graphdb.ErrEdgeNotFound) { return nil, err diff --git a/routing/bandwidth.go b/routing/bandwidth.go index df68cea4f..32bc45e2e 100644 --- a/routing/bandwidth.go +++ b/routing/bandwidth.go @@ -1,6 +1,7 @@ package routing import ( + "context" "fmt" "github.com/lightningnetwork/lnd/fn/v2" @@ -64,7 +65,8 @@ func newBandwidthManager(graph Graph, sourceNode route.Vertex, // First, we'll collect the set of outbound edges from the target // source node and add them to our bandwidth manager's map of channels. err := graph.ForEachNodeDirectedChannel( - sourceNode, func(channel *graphdb.DirectedChannel) error { + context.TODO(), sourceNode, + func(channel *graphdb.DirectedChannel) error { shortID := lnwire.NewShortChanIDFromInt( channel.ChannelID, ) diff --git a/routing/graph.go b/routing/graph.go index 4be34def9..b245d404e 100644 --- a/routing/graph.go +++ b/routing/graph.go @@ -1,6 +1,7 @@ package routing import ( + "context" "fmt" "github.com/btcsuite/btcd/btcutil" @@ -14,12 +15,13 @@ import ( type Graph interface { // ForEachNodeDirectedChannel calls the callback for every channel of // the given node. - ForEachNodeDirectedChannel(nodePub route.Vertex, + ForEachNodeDirectedChannel(ctx context.Context, nodePub route.Vertex, cb func(channel *graphdb.DirectedChannel) error, reset func()) error // FetchNodeFeatures returns the features of the given node. - FetchNodeFeatures(nodePub route.Vertex) (*lnwire.FeatureVector, error) + FetchNodeFeatures(ctx context.Context, + nodePub route.Vertex) (*lnwire.FeatureVector, error) } // GraphSessionFactory can be used to gain access to a graphdb.NodeTraverser @@ -30,7 +32,8 @@ type GraphSessionFactory interface { // GraphSession will provide the call-back with access to a // graphdb.NodeTraverser instance which can be used to perform queries // against the channel graph. - GraphSession(cb func(graph graphdb.NodeTraverser) error, + GraphSession(ctx context.Context, + cb func(graph graphdb.NodeTraverser) error, reset func()) error } diff --git a/routing/mock_graph_test.go b/routing/mock_graph_test.go index d9b16f646..210551471 100644 --- a/routing/mock_graph_test.go +++ b/routing/mock_graph_test.go @@ -2,6 +2,7 @@ package routing import ( "bytes" + "context" "fmt" "testing" @@ -165,8 +166,9 @@ func (m *mockGraph) addChannel(id uint64, node1id, node2id byte, // forEachNodeChannel calls the callback for every channel of the given node. // // NOTE: Part of the Graph interface. -func (m *mockGraph) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *graphdb.DirectedChannel) error, _ func()) error { +func (m *mockGraph) ForEachNodeDirectedChannel(_ context.Context, + nodePub route.Vertex, cb func(channel *graphdb.DirectedChannel) error, + _ func()) error { // Look up the mock node. node, ok := m.nodes[nodePub] @@ -221,8 +223,8 @@ func (m *mockGraph) sourceNode() route.Vertex { // fetchNodeFeatures returns the features of the given node. // // NOTE: Part of the Graph interface. -func (m *mockGraph) FetchNodeFeatures(nodePub route.Vertex) ( - *lnwire.FeatureVector, error) { +func (m *mockGraph) FetchNodeFeatures(_ context.Context, + _ route.Vertex) (*lnwire.FeatureVector, error) { return lnwire.EmptyFeatureVector(), nil } @@ -232,8 +234,8 @@ func (m *mockGraph) FetchNodeFeatures(nodePub route.Vertex) ( // the channel graph. // // NOTE: Part of the GraphSessionFactory interface. -func (m *mockGraph) GraphSession(cb func(graph graphdb.NodeTraverser) error, - _ func()) error { +func (m *mockGraph) GraphSession(_ context.Context, + cb func(graph graphdb.NodeTraverser) error, _ func()) error { return cb(m) } diff --git a/routing/pathfind.go b/routing/pathfind.go index ddaade60c..97ce80997 100644 --- a/routing/pathfind.go +++ b/routing/pathfind.go @@ -3,6 +3,7 @@ package routing import ( "bytes" "container/heap" + "context" "errors" "fmt" "math" @@ -577,7 +578,7 @@ func getOutgoingBalance(node route.Vertex, outgoingChans map[uint64]struct{}, // Iterate over all channels of the to node. err := g.ForEachNodeDirectedChannel( - node, cb, func() { + context.TODO(), node, cb, func() { max = 0 total = 0 }, @@ -621,7 +622,9 @@ func findPath(g *graphParams, r *RestrictParams, cfg *PathFindingConfig, features := r.DestFeatures if features == nil { var err error - features, err = g.graph.FetchNodeFeatures(target) + features, err = g.graph.FetchNodeFeatures( + context.TODO(), target, + ) if err != nil { return nil, 0, err } @@ -1019,7 +1022,9 @@ func findPath(g *graphParams, r *RestrictParams, cfg *PathFindingConfig, } // Fetch node features fresh from the graph. - fromFeatures, err := g.graph.FetchNodeFeatures(node) + fromFeatures, err := g.graph.FetchNodeFeatures( + context.TODO(), node, + ) if err != nil { return nil, err } @@ -1286,7 +1291,7 @@ func findBlindedPaths(g Graph, target route.Vertex, nextTargetReset = nextTarget ) err := g.ForEachNodeDirectedChannel( - nextTarget, + context.TODO(), nextTarget, func(channel *graphdb.DirectedChannel) error { // This is not the right channel, continue to // the node's other channels. @@ -1349,7 +1354,7 @@ func findBlindedPaths(g Graph, target route.Vertex, return true, nil } - features, err := g.FetchNodeFeatures(node) + features, err := g.FetchNodeFeatures(context.TODO(), node) if err != nil { return false, err } @@ -1461,7 +1466,8 @@ func processNodeForBlindedPath(g Graph, node route.Vertex, // Now, iterate over the node's channels in search for paths to this // node that can be used for blinded paths err = g.ForEachNodeDirectedChannel( - node, func(channel *graphdb.DirectedChannel) error { + context.TODO(), node, + func(channel *graphdb.DirectedChannel) error { // Keep track of how many incoming channels this node // has. We only use a node as an introduction node if it // has channels other than the one that lead us to it. diff --git a/routing/pathfind_test.go b/routing/pathfind_test.go index 01f5d9849..85689ef9e 100644 --- a/routing/pathfind_test.go +++ b/routing/pathfind_test.go @@ -2199,7 +2199,9 @@ func runRouteFailMaxHTLC(t *testing.T, useCache bool) { // Next, update the middle edge policy to only allow payments up to 100k // msat. graph := ctx.testGraphInstance.graph - _, midEdge, _, err := graph.FetchChannelEdgesByID(firstToSecondID) + _, midEdge, _, err := graph.FetchChannelEdgesByID( + t.Context(), firstToSecondID, + ) require.NoError(t, err, "unable to fetch channel edges by ID") midEdge.MessageFlags = 1 midEdge.MaxHTLC = payAmt - 1 @@ -2243,7 +2245,9 @@ func runRouteFailDisabledEdge(t *testing.T, useCache bool) { // path finding, as we don't consider the disable flag for local // channels (and roasbeef is the source). roasToPham := uint64(999991) - _, e1, e2, err := graph.graph.FetchChannelEdgesByID(roasToPham) + _, e1, e2, err := graph.graph.FetchChannelEdgesByID( + t.Context(), roasToPham, + ) require.NoError(t, err, "unable to fetch edge") e1.ChannelFlags |= lnwire.ChanUpdateDisabled e1.LastUpdate = e1.LastUpdate.Add(time.Second) @@ -2266,7 +2270,9 @@ func runRouteFailDisabledEdge(t *testing.T, useCache bool) { // Now, we'll modify the edge from phamnuwen -> sophon, to read that // it's disabled. phamToSophon := uint64(99999) - _, e, _, err := graph.graph.FetchChannelEdgesByID(phamToSophon) + _, e, _, err := graph.graph.FetchChannelEdgesByID( + t.Context(), phamToSophon, + ) require.NoError(t, err, "unable to fetch edge") e.ChannelFlags |= lnwire.ChanUpdateDisabled e.LastUpdate = e.LastUpdate.Add(time.Second) @@ -2349,7 +2355,9 @@ func runPathSourceEdgesBandwidth(t *testing.T, useCache bool) { // Finally, set the roasbeef->songoku bandwidth, but also set its // disable flag. bandwidths.hints[roasToSongoku] = 2 * payAmt - _, e1, e2, err := graph.graph.FetchChannelEdgesByID(roasToSongoku) + _, e1, e2, err := graph.graph.FetchChannelEdgesByID( + t.Context(), roasToSongoku, + ) require.NoError(t, err, "unable to fetch edge") e1.ChannelFlags |= lnwire.ChanUpdateDisabled e1.LastUpdate = e1.LastUpdate.Add(time.Second) @@ -3318,7 +3326,7 @@ func dbFindPath(graph *graphdb.VersionedGraph, } var route []*unifiedEdge - err = graph.GraphSession(func(graph graphdb.NodeTraverser) error { + err = graph.GraphSession(ctx, func(graph graphdb.NodeTraverser) error { route, _, err = findPath( &graphParams{ additionalEdges: additionalEdges, diff --git a/routing/payment_session.go b/routing/payment_session.go index bb795211c..4cddfa2ea 100644 --- a/routing/payment_session.go +++ b/routing/payment_session.go @@ -1,6 +1,7 @@ package routing import ( + "context" "fmt" "github.com/btcsuite/btcd/btcec/v2" @@ -344,6 +345,7 @@ func (p *paymentSession) RequestRoute(maxAmt, feeLimit lnwire.MilliSatoshi, for { err := p.graphSessFactory.GraphSession( + context.TODO(), findPath, func() { path = nil }, diff --git a/routing/payment_session_test.go b/routing/payment_session_test.go index 547fe0e2d..0bc0b6dcb 100644 --- a/routing/payment_session_test.go +++ b/routing/payment_session_test.go @@ -1,6 +1,7 @@ package routing import ( + "context" "testing" "time" @@ -260,8 +261,8 @@ func (g *sessionGraph) sourceNode() route.Vertex { return route.Vertex{} } -func (g *sessionGraph) GraphSession(cb func(graph graphdb.NodeTraverser) error, - _ func()) error { +func (g *sessionGraph) GraphSession(_ context.Context, + cb func(graph graphdb.NodeTraverser) error, _ func()) error { return cb(g) } diff --git a/routing/router_test.go b/routing/router_test.go index f1c6c3832..e9c9ed945 100644 --- a/routing/router_test.go +++ b/routing/router_test.go @@ -458,7 +458,7 @@ func TestChannelUpdateValidation(t *testing.T) { // Assert that the initially configured fee is retrieved correctly. _, e1, e2, err := ctx.graph.FetchChannelEdgesByID( - lnwire.NewShortChanIDFromInt(1).ToUint64(), + t.Context(), lnwire.NewShortChanIDFromInt(1).ToUint64(), ) require.NoError(t, err, "cannot retrieve channel") @@ -530,7 +530,7 @@ func TestChannelUpdateValidation(t *testing.T) { require.Error(t, err, "expected route to fail with channel update") _, e1, e2, err = ctx.graph.FetchChannelEdgesByID( - lnwire.NewShortChanIDFromInt(1).ToUint64(), + t.Context(), lnwire.NewShortChanIDFromInt(1).ToUint64(), ) require.NoError(t, err, "cannot retrieve channel") @@ -552,7 +552,7 @@ func TestChannelUpdateValidation(t *testing.T) { // This time a valid signature was supplied and the policy change should // have been applied to the graph. _, e1, e2, err = ctx.graph.FetchChannelEdgesByID( - lnwire.NewShortChanIDFromInt(1).ToUint64(), + t.Context(), lnwire.NewShortChanIDFromInt(1).ToUint64(), ) require.NoError(t, err, "cannot retrieve channel") @@ -593,7 +593,7 @@ func TestSendPaymentErrorRepeatedFeeInsufficient(t *testing.T) { // to sophon. We'll obtain this as we'll need to to generate the // FeeInsufficient error that we'll send back. _, _, edgeUpdateToFail, err := ctx.graph.FetchChannelEdgesByID( - songokuSophonChanID, + t.Context(), songokuSophonChanID, ) require.NoError(t, err, "unable to fetch chan id") @@ -946,7 +946,9 @@ func TestSendPaymentErrorNonFinalTimeLockErrors(t *testing.T) { chanID := ctx.getChannelIDFromAlias(t, "roasbeef", "songoku") roasbeefSongoku := lnwire.NewShortChanIDFromInt(chanID) - _, _, edgeUpdateToFail, err := ctx.graph.FetchChannelEdgesByID(chanID) + _, _, edgeUpdateToFail, err := ctx.graph.FetchChannelEdgesByID( + t.Context(), chanID, + ) require.NoError(t, err, "unable to fetch chan id") errChanUpdate := lnwire.ChannelUpdate1{ diff --git a/routing/unified_edges.go b/routing/unified_edges.go index fda06aabf..7966f6734 100644 --- a/routing/unified_edges.go +++ b/routing/unified_edges.go @@ -1,6 +1,7 @@ package routing import ( + "context" "math" "github.com/btcsuite/btcd/btcutil" @@ -116,7 +117,7 @@ func (u *nodeEdgeUnifier) addGraphPolicies(g Graph) error { // Iterate over all channels of the to node. err := g.ForEachNodeDirectedChannel( - u.toNode, cb, func() { + context.TODO(), u.toNode, cb, func() { channels = nil }, ) diff --git a/rpcserver.go b/rpcserver.go index c40d973c6..31804cd4c 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -716,7 +716,9 @@ func (r *rpcServer) addDeps(ctx context.Context, s *server, FetchChannelCapacity: func(chanID uint64) (btcutil.Amount, error) { - info, _, _, err := graph.FetchChannelEdgesByID(chanID) + info, _, _, err := graph.FetchChannelEdgesByID( + ctx, chanID, + ) if err != nil { return 0, err } @@ -734,7 +736,7 @@ func (r *rpcServer) addDeps(ctx context.Context, s *server, route.Vertex, error) { info, _, _, err := graph.FetchChannelEdgesByID( - chanID, + ctx, chanID, ) if err != nil { return route.Vertex{}, route.Vertex{}, @@ -3202,7 +3204,7 @@ func abandonChanFromGraph(chanGraph *graphdb.VersionedGraph, // First, we'll obtain the channel ID. If we can't locate this, then // it's the case that the channel may have already been removed from // the graph, so we'll return a nil error. - chanID, err := chanGraph.ChannelID(chanPoint) + chanID, err := chanGraph.ChannelID(context.TODO(), chanPoint) switch { case errors.Is(err, graphdb.ErrEdgeNotFound): return nil @@ -3212,7 +3214,7 @@ func abandonChanFromGraph(chanGraph *graphdb.VersionedGraph, // If the channel ID is still in the graph, then that means the channel // is still open, so we'll now move to purge it from the graph. - return chanGraph.DeleteChannelEdges(false, true, chanID) + return chanGraph.DeleteChannelEdges(context.TODO(), false, true, chanID) } // abandonChan removes a channel from the database, graph and contract court. @@ -7117,7 +7119,7 @@ func (r *rpcServer) GetNodeMetrics(ctx context.Context, // uniquely identify the location of transaction's funding output within the // blockchain. The former is an 8-byte integer, while the latter is a string // formatted as funding_txid:output_index. -func (r *rpcServer) GetChanInfo(_ context.Context, +func (r *rpcServer) GetChanInfo(ctx context.Context, in *lnrpc.ChanInfoRequest) (*lnrpc.ChannelEdge, error) { graph := r.server.graphDB @@ -7131,7 +7133,7 @@ func (r *rpcServer) GetChanInfo(_ context.Context, switch { case in.ChanId != 0: edgeInfo, edge1, edge2, err = graph.FetchChannelEdgesByID( - in.ChanId, + ctx, in.ChanId, ) case in.ChanPoint != "": @@ -7141,7 +7143,7 @@ func (r *rpcServer) GetChanInfo(_ context.Context, return nil, err } edgeInfo, edge1, edge2, err = graph.FetchChannelEdgesByOutpoint( - chanPoint, + ctx, chanPoint, ) default: @@ -7387,7 +7389,7 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context, } // Query the graph for the current number of zombie channels. - numZombies, err := graph.NumZombies() + numZombies, err := graph.NumZombies(ctx) if err != nil { return nil, err } diff --git a/server.go b/server.go index eaa5a528c..ff9130675 100644 --- a/server.go +++ b/server.go @@ -1398,7 +1398,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, *models.ChannelEdgePolicy, error) { info, e1, e2, err := s.graphDB.FetchChannelEdgesByID( - scid.ToUint64(), + context.TODO(), scid.ToUint64(), ) if errors.Is(err, graphdb.ErrEdgeNotFound) { // This is unlikely but there is a slim chance of this @@ -1427,7 +1427,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, } err = s.v1Graph.DeleteChannelEdges( - false, false, scid.ToUint64(), + context.TODO(), false, false, scid.ToUint64(), ) return ourPolicy, err }