Merge pull request #10610 from ellemouton/graph-cleanup-todo-contexts

graph/db+refactor: surface all `context.TODO()`s
This commit is contained in:
Elle authored and GitHub committed 2026-02-26 16:18:31 +02:00
commit 8126b424fd
31 files changed
+642 -484

No files matched your search

+10 -7
View File
@@ -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.
+9 -6
View File
@@ -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
+4 -2
View File
@@ -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)
+1 -1
View File
@@ -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)
+6 -3
View File
@@ -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()
+43 -26
View File
@@ -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(),
)
}
+11 -11
View File
@@ -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 {
+2 -2
View File
@@ -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)
+131 -95
View File
@@ -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
+132 -91
View File
@@ -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)
}
+50 -39
View File
@@ -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)
}
+54 -44
View File
@@ -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)
+8 -5
View File
@@ -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)
}
+67 -97
View File
@@ -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)
}
+27 -8
View File
@@ -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,
}
}
+6 -3
View File
@@ -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)
}
+4 -1
View File
@@ -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
}
+3 -1
View File
@@ -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()
+5 -2
View File
@@ -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)
}
+1 -1
View File
@@ -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
+3 -1
View File
@@ -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,
)
+6 -3
View File
@@ -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
}
+8 -6
View File
@@ -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)
}
+12 -6
View File
@@ -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.
+13 -5
View File
@@ -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,
+2
View File
@@ -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
},
+3 -2
View File
@@ -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)
}
+7 -5
View File
@@ -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{
+2 -1
View File
@@ -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
},
)
+10 -8
View File
@@ -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
}
+2 -2
View File
@@ -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
}