multi: updates to HasChannelPolicy

Author: ellemouton

channeldb/graph.go

@@ -1157,13 +1157,84 @@ func (c *ChannelGraph) addChannelEdge(tx kvdb.RwTx,
 	return chanIndex.Put(b.Bytes(), chanKey[:])
 }
 
-// HasChannelEdge returns true if the database knows of a channel edge with the
+func (c *ChannelGraph) HasChannelEdge(chanID uint64) (bool, bool, error) {
+	var (
+		exists   bool
+		isZombie bool
+	)
+
+	// We'll query the cache with the shared lock held to allow multiple
+	// readers to access values in the cache concurrently if they exist.
+	c.cacheMu.RLock()
+	if entry, ok := c.rejectCache.get(chanID); ok {
+		c.cacheMu.RUnlock()
+		exists, isZombie = entry.flags.unpack()
+
+		return exists, isZombie, nil
+	}
+	c.cacheMu.RUnlock()
+
+	c.cacheMu.Lock()
+	defer c.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 := c.rejectCache.get(chanID); ok && entry.times != nil {
+		exists, isZombie = entry.flags.unpack()
+
+		return exists, isZombie, nil
+	}
+
+	if err := kvdb.View(c.db, func(tx kvdb.RTx) error {
+		edges := tx.ReadBucket(edgeBucket)
+		if edges == nil {
+			return ErrGraphNoEdgesFound
+		}
+		edgeIndex := edges.NestedReadBucket(edgeIndexBucket)
+		if edgeIndex == nil {
+			return ErrGraphNoEdgesFound
+		}
+
+		var channelID [8]byte
+		byteOrder.PutUint64(channelID[:], chanID)
+
+		// If the edge doesn't exist, then we'll also check our zombie
+		// index.
+		if edgeIndex.Get(channelID[:]) == nil {
+			exists = false
+			zombieIndex := edges.NestedReadBucket(zombieBucket)
+			if zombieIndex != nil {
+				isZombie, _, _ = isZombieEdge(
+					zombieIndex, chanID,
+				)
+			}
+
+			return nil
+		}
+
+		exists = true
+		isZombie = false
+
+		return nil
+	}, func() {}); err != nil {
+		return exists, isZombie, err
+	}
+
+	c.rejectCache.insert(chanID, rejectCacheEntry{
+		flags: packRejectFlags(exists, isZombie),
+	})
+
+	return exists, isZombie, nil
+}
+
+// HasChannelEdge1 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 *ChannelGraph) HasChannelEdge(
+func (c *ChannelGraph) HasChannelEdge1(
 	chanID uint64) (time.Time, time.Time, bool, bool, error) {
 
 	var (
@@ -1279,6 +1350,122 @@ func (c *ChannelGraph) HasChannelEdge(
 	return upd1Time, upd2Time, exists, isZombie, nil
 }
 
+func (c *ChannelGraph) HasChannelEdge2(
+	chanID uint64) (uint32, uint32, bool, bool, error) {
+
+	var (
+		upd1Height uint32
+		upd2Height uint32
+		exists     bool
+		isZombie   bool
+	)
+
+	// We'll query the cache with the shared lock held to allow multiple
+	// readers to access values in the cache concurrently if they exist.
+	c.cacheMu.RLock()
+	if entry, ok := c.rejectCache.get(chanID); ok && entry.blocks != nil {
+		c.cacheMu.RUnlock()
+		exists, isZombie = entry.flags.unpack()
+
+		return entry.blocks.updBlock1, entry.blocks.updBlock2, exists,
+			isZombie, nil
+	}
+	c.cacheMu.RUnlock()
+
+	c.cacheMu.Lock()
+	defer c.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 := c.rejectCache.get(chanID); ok && entry.blocks != nil {
+		exists, isZombie = entry.flags.unpack()
+
+		return entry.blocks.updBlock1, entry.blocks.updBlock2, exists,
+			isZombie, nil
+	}
+
+	if err := kvdb.View(c.db, func(tx kvdb.RTx) error {
+		edges := tx.ReadBucket(edgeBucket)
+		if edges == nil {
+			return ErrGraphNoEdgesFound
+		}
+		edgeIndex := edges.NestedReadBucket(edgeIndexBucket)
+		if edgeIndex == nil {
+			return ErrGraphNoEdgesFound
+		}
+
+		var channelID [8]byte
+		byteOrder.PutUint64(channelID[:], chanID)
+
+		// If the edge doesn't exist, then we'll also check our zombie
+		// index.
+		if edgeIndex.Get(channelID[:]) == nil {
+			exists = false
+			zombieIndex := edges.NestedReadBucket(zombieBucket)
+			if zombieIndex != nil {
+				isZombie, _, _ = isZombieEdge(
+					zombieIndex, chanID,
+				)
+			}
+
+			return nil
+		}
+
+		exists = true
+		isZombie = false
+
+		// If the channel has been found in the graph, then retrieve
+		// the edges itself so we can return the last updated
+		// timestamps.
+		nodes := tx.ReadBucket(nodeBucket)
+		if nodes == nil {
+			return ErrGraphNodeNotFound
+		}
+
+		edge1, edge2, err := fetchChanEdgePolicies(
+			edgeIndex, edges, channelID[:],
+		)
+		if err != nil {
+			return err
+		}
+
+		// As we may have only one of the edges populated, only set the
+		// update time if the edge was found in the database.
+		if edge1 != nil {
+			e1, ok := edge1.(*models.ChannelEdgePolicy2)
+			if !ok {
+				return fmt.Errorf("expected "+
+					"*ChannelEdgePolicy2, got: %T", edge1)
+			}
+			upd1Height = e1.BlockHeight.Val
+		}
+		if edge2 != nil {
+			e2, ok := edge2.(*models.ChannelEdgePolicy2)
+			if !ok {
+				return fmt.Errorf("expected "+
+					"*ChannelEdgePolicy2, got: %T", edge2)
+			}
+
+			upd2Height = e2.BlockHeight.Val
+		}
+
+		return nil
+	}, func() {}); err != nil {
+		return 0, 0, exists, isZombie, err
+	}
+
+	c.rejectCache.insert(chanID, rejectCacheEntry{
+		blocks: &updateBlocks{
+			updBlock1: upd1Height,
+			updBlock2: upd2Height,
+		},
+		flags: packRejectFlags(exists, isZombie),
+	})
+
+	return upd1Height, upd2Height, exists, isZombie, nil
+}
+
 // UpdateChannelEdge retrieves and update edge of the graph database. Method
 // only reserved for updating an edge info after its already been created.
 // In order to maintain this constraints, we return an error in the scenario

channeldb/graph_test.go

@@ -517,15 +517,15 @@ 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(edgeInfo.ChannelID)
 	require.NoError(t, err, "unable to query for edge")
 	if has {
 		t.Fatalf("edge1 was not pruned from the graph")
 	}
 	if isZombie {
 		t.Fatal("reorged edge1 should not be marked as zombie")
 	}
-	_, _, has, isZombie, err = graph.HasChannelEdge(edgeInfo2.ChannelID)
+	has, isZombie, err = graph.HasChannelEdge(edgeInfo2.ChannelID)
 	require.NoError(t, err, "unable to query for edge")
 	if has {
 		t.Fatalf("edge2 was not pruned from the graph")
@@ -535,7 +535,7 @@ func TestDisconnectBlockAtHeight(t *testing.T) {
 	}
 
 	// Edge 3 should not be removed.
-	_, _, has, isZombie, err = graph.HasChannelEdge(edgeInfo3.ChannelID)
+	has, isZombie, err = graph.HasChannelEdge(edgeInfo3.ChannelID)
 	require.NoError(t, err, "unable to query for edge")
 	if !has {
 		t.Fatalf("edge3 was pruned from the graph")
@@ -759,7 +759,7 @@ 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(chanID)
 	require.NoError(t, err, "unable to query for edge")
 	if !found {
 		t.Fatalf("graph should have of inserted edge")
@@ -2283,7 +2283,7 @@ func TestStressTestChannelGraphAPI(t *testing.T) {
 					return nil
 				}
 
-				_, _, _, _, err := graph.HasChannelEdge(
+				_, _, err := graph.HasChannelEdge(
 					channel.id.ToUint64(),
 				)
 

graph/builder.go

@@ -1146,9 +1146,7 @@ func (b *Builder) processUpdate(msg interface{},
 
 		// 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(
-			chanID,
-		)
+		exists, isZombie, err := b.cfg.Graph.HasChannelEdge(chanID)
 		if err != nil &&
 			!errors.Is(err, channeldb.ErrGraphNoEdgesFound) {
 
@@ -1338,7 +1336,7 @@ func (b *Builder) processUpdate(msg interface{},
 		defer b.channelEdgeMtx.Unlock(msg.ChannelID)
 
 		edge1Timestamp, edge2Timestamp, exists, isZombie, err :=
-			b.cfg.Graph.HasChannelEdge(msg.ChannelID)
+			b.cfg.Graph.HasChannelEdge1(msg.ChannelID)
 		if err != nil && !errors.Is(
 			err, channeldb.ErrGraphNoEdgesFound,
 		) {
@@ -1683,9 +1681,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(
-		chanID.ToUint64(),
-	)
+	exists, isZombie, _ := b.cfg.Graph.HasChannelEdge(chanID.ToUint64())
 
 	return exists || isZombie
 }
@@ -1698,7 +1694,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.HasChannelEdge1(chanID.ToUint64())
 	if err != nil {
 		log.Debugf("Check stale edge policy got error: %v", err)
 		return false