diff --git a/graph/builder.go b/graph/builder.go index fa8b0b4a9..2cf4d1f34 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -1043,8 +1043,8 @@ 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.ChannelID, + exists, isZombie, err := b.cfg.Graph.HasChannelEdge( + edge.Version, edge.ChannelID, ) if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) { return fmt.Errorf("unable to check for edge existence: %w", @@ -1145,7 +1145,7 @@ func (b *Builder) updateEdge(ctx context.Context, defer b.channelEdgeMtx.Unlock(policy.ChannelID) edge1Timestamp, edge2Timestamp, exists, isZombie, err := - b.cfg.Graph.HasChannelEdge(policy.ChannelID) + b.cfg.Graph.HasV1ChannelEdge(policy.ChannelID) if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) { return fmt.Errorf("unable to check for edge existence: %w", err) } @@ -1331,8 +1331,8 @@ 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( - chanID.ToUint64(), + exists, isZombie, _ := b.cfg.Graph.HasChannelEdge( + lnwire.GossipVersion1, chanID.ToUint64(), ) return exists || isZombie @@ -1343,7 +1343,9 @@ 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(chanID.ToUint64()) + _, isZombie, err := b.cfg.Graph.HasChannelEdge( + lnwire.GossipVersion1, chanID.ToUint64(), + ) return isZombie, err } @@ -1356,7 +1358,7 @@ func (b *Builder) IsStaleEdgePolicy(chanID lnwire.ShortChannelID, timestamp time.Time, flags lnwire.ChanUpdateChanFlags) bool { edge1Timestamp, edge2Timestamp, exists, isZombie, err := - b.cfg.Graph.HasChannelEdge(chanID.ToUint64()) + b.cfg.Graph.HasV1ChannelEdge(chanID.ToUint64()) if err != nil { log.Debugf("Check stale edge policy got error: %v", err) return false diff --git a/graph/builder_test.go b/graph/builder_test.go index 6929a4d49..acab91ccf 100644 --- a/graph/builder_test.go +++ b/graph/builder_test.go @@ -319,7 +319,7 @@ func TestWakeUpOnStaleBranch(t *testing.T) { } // Check that the fundingTxs are in the graph db. - _, _, has, isZombie, err := ctx.graph.HasChannelEdge(chanID1) + has, isZombie, err := ctx.graph.HasChannelEdge(chanID1) if err != nil { t.Fatalf("error looking for edge: %v", chanID1) } @@ -330,7 +330,7 @@ func TestWakeUpOnStaleBranch(t *testing.T) { t.Fatal("edge was marked as zombie") } - _, _, has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) + has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) if err != nil { t.Fatalf("error looking for edge: %v", chanID2) } @@ -387,7 +387,7 @@ 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(chanID1) require.NoError(t, err) if !has { @@ -397,7 +397,7 @@ func TestWakeUpOnStaleBranch(t *testing.T) { t.Fatal("edge was marked as zombie") } - _, _, has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) + has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) if err != nil { t.Fatalf("error looking for edge: %v", chanID2) } @@ -527,7 +527,7 @@ func TestDisconnectedBlocks(t *testing.T) { } // Check that the fundingTxs are in the graph db. - _, _, has, isZombie, err := ctx.graph.HasChannelEdge(chanID1) + has, isZombie, err := ctx.graph.HasChannelEdge(chanID1) if err != nil { t.Fatalf("error looking for edge: %v", chanID1) } @@ -538,7 +538,7 @@ func TestDisconnectedBlocks(t *testing.T) { t.Fatal("edge was marked as zombie") } - _, _, has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) + has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) if err != nil { t.Fatalf("error looking for edge: %v", chanID2) } @@ -580,7 +580,7 @@ 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(chanID1) if err != nil { t.Fatalf("error looking for edge: %v", chanID1) } @@ -591,7 +591,7 @@ func TestDisconnectedBlocks(t *testing.T) { t.Fatal("edge was marked as zombie") } - _, _, has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) + has, isZombie, err = ctx.graph.HasChannelEdge(chanID2) if err != nil { t.Fatalf("error looking for edge: %v", chanID2) } @@ -665,7 +665,7 @@ func TestChansClosedOfflinePruneGraph(t *testing.T) { } // The router should now be aware of the channel we created above. - _, _, hasChan, isZombie, err := ctx.graph.HasChannelEdge( + hasChan, isZombie, err := ctx.graph.HasChannelEdge( chanID1.ToUint64(), ) if err != nil { @@ -747,7 +747,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( + hasChan, isZombie, err = ctx.graph.HasChannelEdge( chanID1.ToUint64(), ) if err != nil { @@ -1719,7 +1719,7 @@ func assertChannelsPruned(t *testing.T, graph *graphdb.VersionedGraph, for _, channel := range channels { _, shouldPrune := pruned[channel.ChannelID] - _, _, exists, isZombie, err := graph.HasChannelEdge( + exists, isZombie, err := graph.HasChannelEdge( channel.ChannelID, ) if err != nil { diff --git a/graph/db/graph.go b/graph/db/graph.go index 41396aceb..3aa860cc1 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -661,11 +661,18 @@ func (c *ChannelGraph) DisabledChannelIDs() ([]uint64, error) { return c.db.DisabledChannelIDs() } -// HasChannelEdge returns true if the database knows of a channel edge. -func (c *ChannelGraph) HasChannelEdge(chanID uint64) (time.Time, time.Time, - bool, bool, error) { +// HasV1ChannelEdge returns true if the database knows of a channel edge. +func (c *ChannelGraph) HasV1ChannelEdge(chanID uint64) (time.Time, + time.Time, bool, bool, error) { - return c.db.HasChannelEdge(chanID) + return c.db.HasV1ChannelEdge(chanID) +} + +// HasChannelEdge returns true if the database knows of a channel edge. +func (c *ChannelGraph) HasChannelEdge(v lnwire.GossipVersion, + chanID uint64) (bool, bool, error) { + + return c.db.HasChannelEdge(v, chanID) } // AddEdgeProof sets the proof of an existing edge in the graph database. @@ -889,6 +896,14 @@ func (c *VersionedGraph) DeleteChannelEdges(strictZombiePruning, return err } +// HasChannelEdge returns true if the database knows of a channel edge with the +// 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) +} + // 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) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 9b923f662..d1aff61e1 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -886,7 +886,9 @@ func TestDisconnectBlockAtHeight(t *testing.T) { } // The two first edges should be removed from the db. - _, _, has, isZombie, err := graph.HasChannelEdge(edgeInfo.ChannelID) + has, isZombie, err := graph.HasChannelEdge( + lnwire.GossipVersion1, edgeInfo.ChannelID, + ) require.NoError(t, err, "unable to query for edge") if has { t.Fatalf("edge1 was not pruned from the graph") @@ -894,7 +896,9 @@ func TestDisconnectBlockAtHeight(t *testing.T) { if isZombie { t.Fatal("reorged edge1 should not be marked as zombie") } - _, _, has, isZombie, err = graph.HasChannelEdge(edgeInfo2.ChannelID) + has, isZombie, err = graph.HasChannelEdge( + lnwire.GossipVersion1, edgeInfo2.ChannelID, + ) require.NoError(t, err, "unable to query for edge") if has { t.Fatalf("edge2 was not pruned from the graph") @@ -904,7 +908,9 @@ func TestDisconnectBlockAtHeight(t *testing.T) { } // Edge 3 should not be removed. - _, _, has, isZombie, err = graph.HasChannelEdge(edgeInfo3.ChannelID) + has, isZombie, err = graph.HasChannelEdge( + lnwire.GossipVersion1, edgeInfo3.ChannelID, + ) require.NoError(t, err, "unable to query for edge") if !has { t.Fatalf("edge3 was pruned from the graph") @@ -1129,7 +1135,9 @@ func TestEdgeInfoUpdates(t *testing.T) { // Check for existence of the edge within the database, it should be // found. - _, _, found, isZombie, err := graph.HasChannelEdge(chanID) + found, isZombie, err := graph.HasChannelEdge( + lnwire.GossipVersion1, chanID, + ) require.NoError(t, err, "unable to query for edge") if !found { t.Fatalf("graph should have of inserted edge") @@ -3317,7 +3325,8 @@ func TestStressTestChannelGraphAPI(t *testing.T) { return nil } - _, _, _, _, err := graph.HasChannelEdge( + _, _, err := graph.HasChannelEdge( + lnwire.GossipVersion1, channel.id.ToUint64(), ) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 24a48e4c7..42f63f932 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -197,13 +197,20 @@ type Store interface { //nolint:interfacebloat AddChannelEdge(ctx context.Context, edge *models.ChannelEdgeInfo, op ...batch.SchedulerOption) error - // HasChannelEdge returns true if the database knows of a channel edge + // HasV1ChannelEdge returns true if the database knows of a channel edge // with the passed channel ID, and false otherwise. If an edge with that // ID is found within the graph, then two time stamps representing the // 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. - HasChannelEdge(chanID uint64) (time.Time, time.Time, bool, bool, + HasV1ChannelEdge(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) // DeleteChannelEdges removes edges with the given channel IDs from the diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 5e71ad700..58b227a22 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -1179,8 +1179,11 @@ func (c *KVStore) AddChannelEdge(ctx context.Context, case alreadyExists: return ErrEdgeAlreadyExist default: - c.rejectCache.remove(lnwire.GossipVersion1, edge.ChannelID) + c.rejectCache.remove( + lnwire.GossipVersion1, edge.ChannelID, + ) c.chanCache.remove(edge.ChannelID) + return nil } }, @@ -1283,13 +1286,13 @@ func (c *KVStore) addChannelEdge(tx kvdb.RwTx, return chanIndex.Put(b.Bytes(), chanKey[:]) } -// HasChannelEdge returns true if the database knows of a channel edge with the -// passed channel ID, and false otherwise. If an edge with that ID is found -// within the graph, then two time stamps representing the 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. -func (c *KVStore) HasChannelEdge( +// HasV1ChannelEdge returns true if the database knows of a channel edge +// with the passed channel ID, and false otherwise. If an edge with that ID +// is found within the graph, then two time stamps representing the 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. +func (c *KVStore) HasV1ChannelEdge( chanID uint64) (time.Time, time.Time, bool, bool, error) { var ( @@ -1394,6 +1397,22 @@ func (c *KVStore) HasChannelEdge( return upd1Time, upd2Time, exists, isZombie, nil } +// 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. +func (c *KVStore) HasChannelEdge(v lnwire.GossipVersion, + chanID uint64) (bool, bool, error) { + + if v != lnwire.GossipVersion1 { + return false, false, ErrVersionNotSupportedForKVDB + } + + _, _, exists, isZombie, err := c.HasV1ChannelEdge(chanID) + + return exists, isZombie, err +} + // AddEdgeProof sets the proof of an existing edge in the graph database. func (c *KVStore) AddEdgeProof(chanID lnwire.ShortChannelID, proof *models.ChannelAuthProof) error { @@ -3265,7 +3284,8 @@ func (c *KVStore) updateEdgeCache(e *models.ChannelEdgePolicy, // the entry with the updated timestamp for the direction that was just // written. If the edge doesn't exist, we'll load the cache entry lazily // during the next query for this edge. - if entry, ok := c.rejectCache.get(lnwire.GossipVersion1, e.ChannelID); ok { + entry, ok := c.rejectCache.get(lnwire.GossipVersion1, e.ChannelID) + if ok { if isUpdate1 { entry.upd1Time = e.LastUpdate.Unix() } else { diff --git a/graph/db/reject_cache.go b/graph/db/reject_cache.go index bfe3c47ff..f54463224 100644 --- a/graph/db/reject_cache.go +++ b/graph/db/reject_cache.go @@ -59,7 +59,6 @@ type rejectCacheEntry struct { flags rejectFlags } -// newRejectCacheEntryV1 constructs a reject cache entry for v1 policies. func newRejectCacheEntryV1(upd1, upd2 time.Time, exists, isZombie bool) rejectCacheEntry { @@ -70,7 +69,6 @@ func newRejectCacheEntryV1(upd1, upd2 time.Time, exists, } } -// newRejectCacheEntryV2 constructs a reject cache entry for v2 policies. func newRejectCacheEntryV2(upd1, upd2 uint32, exists, isZombie bool) rejectCacheEntry { @@ -81,7 +79,6 @@ func newRejectCacheEntryV2(upd1, upd2 uint32, exists, } } -// updateRejectCacheEntryV1 updates the cached v1 timestamps. func updateRejectCacheEntryV1(entry *rejectCacheEntry, isUpdate1 bool, lastUpdate time.Time) { @@ -92,7 +89,6 @@ func updateRejectCacheEntryV1(entry *rejectCacheEntry, isUpdate1 bool, } } -// updateRejectCacheEntryV2 updates the cached v2 block heights. func updateRejectCacheEntryV2(entry *rejectCacheEntry, isUpdate1 bool, blockHeight uint32) { @@ -104,8 +100,7 @@ func updateRejectCacheEntryV2(entry *rejectCacheEntry, isUpdate1 bool, } // rejectCacheKey uniquely identifies a channel entry in the reject cache by -// gossip version and channel ID. This allows v1 and v2 policy state for the -// same channel ID to be cached independently. +// gossip version and channel ID. type rejectCacheKey struct { version lnwire.GossipVersion chanID uint64 diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index d72c8537f..00be30f38 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -747,8 +747,11 @@ func (s *SQLStore) AddChannelEdge(ctx context.Context, case alreadyExists: return ErrEdgeAlreadyExist default: - s.rejectCache.remove(edge.Version, edge.ChannelID) + s.rejectCache.remove( + edge.Version, edge.ChannelID, + ) s.chanCache.remove(edge.ChannelID) + return nil } }, @@ -2204,15 +2207,15 @@ func (s *SQLStore) FetchChannelEdgesByOutpoint(v lnwire.GossipVersion, return edge, policy1, policy2, nil } -// HasChannelEdge returns true if the database knows of a channel edge with the -// passed channel ID, and false otherwise. If an edge with that ID is found -// within the graph, then two time stamps representing the 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 returns true if the database knows of a channel edge +// with the passed channel ID, and false otherwise. If an edge with that ID +// is found within the graph, then two time stamps representing the 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. // // NOTE: part of the Store interface. -func (s *SQLStore) HasChannelEdge(chanID uint64) (time.Time, time.Time, bool, +func (s *SQLStore) HasV1ChannelEdge(chanID uint64) (time.Time, time.Time, bool, bool, error) { ctx := context.TODO() @@ -2325,6 +2328,158 @@ func (s *SQLStore) HasChannelEdge(chanID uint64) (time.Time, time.Time, bool, return node1LastUpdate, node2LastUpdate, exists, isZombie, nil } +// HasChannelEdge returns true if the database knows of a channel edge with the +// passed channel ID and gossip version, and false otherwise. If an edge with +// that ID is found within the graph, then the zombie index is checked and its +// 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) { + + if !isKnownGossipVersion(v) { + return false, false, fmt.Errorf( + "unsupported gossip version: %d", v, + ) + } + + ctx := context.TODO() + + var ( + exists bool + isZombie bool + node1LastUpdate time.Time + node2LastUpdate time.Time + node1Block uint32 + node2Block uint32 + ) + + // We'll query the cache with the shared lock held to allow multiple + // readers to access values in the cache concurrently if they exist. + s.cacheMu.RLock() + if entry, ok := s.rejectCache.get(v, chanID); ok { + s.cacheMu.RUnlock() + exists, isZombie = entry.flags.unpack() + return exists, isZombie, nil + } + s.cacheMu.RUnlock() + + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + + // The item was not found with the shared lock, so we'll acquire the + // exclusive lock and check the cache again in case another method added + // the entry to the cache while no lock was held. + if entry, ok := s.rejectCache.get(v, chanID); ok { + exists, isZombie = entry.flags.unpack() + return exists, isZombie, nil + } + + chanIDB := channelIDToBytes(chanID) + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + channel, err := db.GetChannelBySCID( + ctx, sqlc.GetChannelBySCIDParams{ + Scid: chanIDB, + Version: int16(v), + }, + ) + if errors.Is(err, sql.ErrNoRows) { + // Check if it is a zombie channel. + isZombie, err = db.IsZombieChannel( + ctx, sqlc.IsZombieChannelParams{ + Scid: chanIDB, + Version: int16(v), + }, + ) + if err != nil { + return fmt.Errorf("could not check if channel "+ + "is zombie: %w", err) + } + + return nil + } else if err != nil { + return fmt.Errorf("unable to fetch channel: %w", err) + } + + exists = true + + policy1, err := db.GetChannelPolicyByChannelAndNode( + ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{ + Version: int16(v), + ChannelID: channel.ID, + NodeID: channel.NodeID1, + }, + ) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("unable to fetch channel policy: %w", + err) + } else if err == nil { + switch v { + case lnwire.GossipVersion1: + if policy1.LastUpdate.Valid { + node1LastUpdate = time.Unix( + policy1.LastUpdate.Int64, 0, + ) + } + case lnwire.GossipVersion2: + if policy1.BlockHeight.Valid { + node1Block = uint32( + policy1.BlockHeight.Int64, + ) + } + } + } + + policy2, err := db.GetChannelPolicyByChannelAndNode( + ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{ + Version: int16(v), + ChannelID: channel.ID, + NodeID: channel.NodeID2, + }, + ) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("unable to fetch channel policy: %w", + err) + } else if err == nil { + switch v { + case lnwire.GossipVersion1: + if policy2.LastUpdate.Valid { + node2LastUpdate = time.Unix( + policy2.LastUpdate.Int64, 0, + ) + } + case lnwire.GossipVersion2: + if policy2.BlockHeight.Valid { + node2Block = uint32( + policy2.BlockHeight.Int64, + ) + } + } + } + + return nil + }, sqldb.NoOpReset) + if err != nil { + return false, false, + fmt.Errorf("unable to fetch channel: %w", err) + } + + var entry rejectCacheEntry + switch v { + case lnwire.GossipVersion1: + entry = newRejectCacheEntryV1( + node1LastUpdate, node2LastUpdate, exists, isZombie, + ) + case lnwire.GossipVersion2: + entry = newRejectCacheEntryV2( + node1Block, node2Block, exists, isZombie, + ) + } + s.rejectCache.insert(v, chanID, entry) + + return exists, isZombie, nil +} + // 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. @@ -3553,8 +3708,9 @@ func updateChanEdgePolicy(ctx context.Context, tx SQLQueries, if version == lnwire.GossipVersion1 { extra, err = marshalExtraOpaqueData(edge.ExtraOpaqueData) if err != nil { - return node1Pub, node2Pub, false, fmt.Errorf("unable to "+ - "marshal extra opaque data: %w", err) + return node1Pub, node2Pub, false, fmt.Errorf( + "unable to marshal extra opaque data: %w", err, + ) } }