refactor: move graph related DB code to graph/db from channeldb

This is a pure refactor commit. It moves over all the graph related CRUD
code from `channeldb` to `graph/db`.

graph/db/channel_cache.go

@@ -0,0 +1,50 @@
+package graphdb
+
+// channelCache is an in-memory cache used to improve the performance of
+// ChanUpdatesInHorizon. It caches the chan info and edge policies for a
+// particular channel.
+type channelCache struct {
+	n        int
+	channels map[uint64]ChannelEdge
+}
+
+// newChannelCache creates a new channelCache with maximum capacity of n
+// channels.
+func newChannelCache(n int) *channelCache {
+	return &channelCache{
+		n:        n,
+		channels: make(map[uint64]ChannelEdge),
+	}
+}
+
+// get returns the channel from the cache, if it exists.
+func (c *channelCache) get(chanid uint64) (ChannelEdge, bool) {
+	channel, ok := c.channels[chanid]
+	return channel, ok
+}
+
+// insert adds the entry to the channel cache. If an entry for chanid already
+// exists, it will be replaced with the new entry. If the entry doesn't exist,
+// it will be inserted to the cache, performing a random eviction if the cache
+// is at capacity.
+func (c *channelCache) insert(chanid uint64, channel ChannelEdge) {
+	// If entry exists, replace it.
+	if _, ok := c.channels[chanid]; ok {
+		c.channels[chanid] = channel
+		return
+	}
+
+	// Otherwise, evict an entry at random and insert.
+	if len(c.channels) == c.n {
+		for id := range c.channels {
+			delete(c.channels, id)
+			break
+		}
+	}
+	c.channels[chanid] = channel
+}
+
+// remove deletes an edge for chanid from the cache, if it exists.
+func (c *channelCache) remove(chanid uint64) {
+	delete(c.channels, chanid)
+}

graph/db/channel_cache_test.go

@@ -0,0 +1,107 @@
+package graphdb
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/lightningnetwork/lnd/graph/db/models"
+)
+
+// TestChannelCache checks the behavior of the channelCache with respect to
+// insertion, eviction, and removal of cache entries.
+func TestChannelCache(t *testing.T) {
+	const cacheSize = 100
+
+	// Create a new channel cache with the configured max size.
+	c := newChannelCache(cacheSize)
+
+	// As a sanity check, assert that querying the empty cache does not
+	// return an entry.
+	_, ok := c.get(0)
+	if ok {
+		t.Fatalf("channel cache should be empty")
+	}
+
+	// Now, fill up the cache entirely.
+	for i := uint64(0); i < cacheSize; i++ {
+		c.insert(i, channelForInt(i))
+	}
+
+	// Assert that the cache has all of the entries just inserted, since no
+	// eviction should occur until we try to surpass the max size.
+	assertHasChanEntries(t, c, 0, cacheSize)
+
+	// Now, insert a new element that causes the cache to evict an element.
+	c.insert(cacheSize, channelForInt(cacheSize))
+
+	// Assert that the cache has this last entry, as the cache should evict
+	// some prior element and not the newly inserted one.
+	assertHasChanEntries(t, c, cacheSize, cacheSize)
+
+	// Iterate over all inserted elements and construct a set of the evicted
+	// elements.
+	evicted := make(map[uint64]struct{})
+	for i := uint64(0); i < cacheSize+1; i++ {
+		_, ok := c.get(i)
+		if !ok {
+			evicted[i] = struct{}{}
+		}
+	}
+
+	// Assert that exactly one element has been evicted.
+	numEvicted := len(evicted)
+	if numEvicted != 1 {
+		t.Fatalf("expected one evicted entry, got: %d", numEvicted)
+	}
+
+	// Remove the highest item which initially caused the eviction and
+	// reinsert the element that was evicted prior.
+	c.remove(cacheSize)
+	for i := range evicted {
+		c.insert(i, channelForInt(i))
+	}
+
+	// Since the removal created an extra slot, the last insertion should
+	// not have caused an eviction and the entries for all channels in the
+	// original set that filled the cache should be present.
+	assertHasChanEntries(t, c, 0, cacheSize)
+
+	// Finally, reinsert the existing set back into the cache and test that
+	// the cache still has all the entries. If the randomized eviction were
+	// happening on inserts for existing cache items, we expect this to fail
+	// with high probability.
+	for i := uint64(0); i < cacheSize; i++ {
+		c.insert(i, channelForInt(i))
+	}
+	assertHasChanEntries(t, c, 0, cacheSize)
+
+}
+
+// assertHasEntries queries the edge cache for all channels in the range [start,
+// end), asserting that they exist and their value matches the entry produced by
+// entryForInt.
+func assertHasChanEntries(t *testing.T, c *channelCache, start, end uint64) {
+	t.Helper()
+
+	for i := start; i < end; i++ {
+		entry, ok := c.get(i)
+		if !ok {
+			t.Fatalf("channel cache should contain chan %d", i)
+		}
+
+		expEntry := channelForInt(i)
+		if !reflect.DeepEqual(entry, expEntry) {
+			t.Fatalf("entry mismatch, want: %v, got: %v",
+				expEntry, entry)
+		}
+	}
+}
+
+// channelForInt generates a unique ChannelEdge given an integer.
+func channelForInt(i uint64) ChannelEdge {
+	return ChannelEdge{
+		Info: &models.ChannelEdgeInfo{
+			ChannelID: i,
+		},
+	}
+}

graph/db/errors.go

@@ -1,9 +1,75 @@
 package graphdb
 
-import "fmt"
+import (
+	"errors"
+	"fmt"
+)
 
 var (
+	// ErrEdgePolicyOptionalFieldNotFound is an error returned if a channel
+	// policy field is not found in the db even though its message flags
+	// indicate it should be.
+	ErrEdgePolicyOptionalFieldNotFound = fmt.Errorf("optional field not " +
+		"present")
+
+	// ErrGraphNotFound is returned when at least one of the components of
+	// graph doesn't exist.
+	ErrGraphNotFound = fmt.Errorf("graph bucket not initialized")
+
+	// ErrGraphNeverPruned is returned when graph was never pruned.
+	ErrGraphNeverPruned = fmt.Errorf("graph never pruned")
+
+	// ErrSourceNodeNotSet is returned if the source node of the graph
+	// hasn't been added The source node is the center node within a
+	// star-graph.
+	ErrSourceNodeNotSet = fmt.Errorf("source node does not exist")
+
+	// ErrGraphNodesNotFound is returned in case none of the nodes has
+	// been added in graph node bucket.
+	ErrGraphNodesNotFound = fmt.Errorf("no graph nodes exist")
+
+	// ErrGraphNoEdgesFound is returned in case of none of the channel/edges
+	// has been added in graph edge bucket.
+	ErrGraphNoEdgesFound = fmt.Errorf("no graph edges exist")
+
+	// ErrGraphNodeNotFound is returned when we're unable to find the target
+	// node.
+	ErrGraphNodeNotFound = fmt.Errorf("unable to find node")
+
+	// ErrZombieEdge is an error returned when we attempt to look up an edge
+	// but it is marked as a zombie within the zombie index.
+	ErrZombieEdge = errors.New("edge marked as zombie")
+
+	// ErrEdgeNotFound is returned when an edge for the target chanID
+	// can't be found.
+	ErrEdgeNotFound = fmt.Errorf("edge not found")
+
+	// ErrEdgeAlreadyExist is returned when edge with specific
+	// channel id can't be added because it already exist.
+	ErrEdgeAlreadyExist = fmt.Errorf("edge already exist")
+
+	// ErrNodeAliasNotFound is returned when alias for node can't be found.
+	ErrNodeAliasNotFound = fmt.Errorf("alias for node not found")
+
+	// ErrClosedScidsNotFound is returned when the closed scid bucket
+	// hasn't been created.
+	ErrClosedScidsNotFound = fmt.Errorf("closed scid bucket doesn't exist")
+
+	// ErrZombieEdgeNotFound is an error returned when we attempt to find an
+	// edge in the zombie index which is not there.
+	ErrZombieEdgeNotFound = errors.New("edge not found in zombie index")
+
 	// ErrUnknownAddressType is returned when a node's addressType is not
 	// an expected value.
 	ErrUnknownAddressType = fmt.Errorf("address type cannot be resolved")
 )
+
+// ErrTooManyExtraOpaqueBytes creates an error which should be returned if the
+// caller attempts to write an announcement message which bares too many extra
+// opaque bytes. We limit this value in order to ensure that we don't waste
+// disk space due to nodes unnecessarily padding out their announcements with
+// garbage data.
+func ErrTooManyExtraOpaqueBytes(numBytes int) error {
+	return fmt.Errorf("max allowed number of opaque bytes is %v, received "+
+		"%v bytes", MaxAllowedExtraOpaqueBytes, numBytes)
+}

graph/db/graph_cache.go

@@ -0,0 +1,442 @@
+package graphdb
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/btcsuite/btcd/btcutil"
+	"github.com/lightningnetwork/lnd/graph/db/models"
+	"github.com/lightningnetwork/lnd/kvdb"
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/routing/route"
+)
+
+// GraphCacheNode is an interface for all the information the cache needs to know
+// about a lightning node.
+type GraphCacheNode interface {
+	// PubKey is the node's public identity key.
+	PubKey() route.Vertex
+
+	// Features returns the node's p2p features.
+	Features() *lnwire.FeatureVector
+
+	// ForEachChannel iterates through all channels of a given node,
+	// executing the passed callback with an edge info structure and the
+	// policies of each end of the channel. The first edge policy is the
+	// outgoing edge *to* the connecting node, while the second is the
+	// incoming edge *from* the connecting node. If the callback returns an
+	// error, then the iteration is halted with the error propagated back up
+	// to the caller.
+	ForEachChannel(kvdb.RTx,
+		func(kvdb.RTx, *models.ChannelEdgeInfo,
+			*models.ChannelEdgePolicy,
+			*models.ChannelEdgePolicy) error) error
+}
+
+// DirectedChannel is a type that stores the channel information as seen from
+// one side of the channel.
+type DirectedChannel struct {
+	// ChannelID is the unique identifier of this channel.
+	ChannelID uint64
+
+	// IsNode1 indicates if this is the node with the smaller public key.
+	IsNode1 bool
+
+	// OtherNode is the public key of the node on the other end of this
+	// channel.
+	OtherNode route.Vertex
+
+	// Capacity is the announced capacity of this channel in satoshis.
+	Capacity btcutil.Amount
+
+	// OutPolicySet is a boolean that indicates whether the node has an
+	// outgoing policy set. For pathfinding only the existence of the policy
+	// is important to know, not the actual content.
+	OutPolicySet bool
+
+	// InPolicy is the incoming policy *from* the other node to this node.
+	// In path finding, we're walking backward from the destination to the
+	// source, so we're always interested in the edge that arrives to us
+	// from the other node.
+	InPolicy *models.CachedEdgePolicy
+
+	// Inbound fees of this node.
+	InboundFee lnwire.Fee
+}
+
+// DeepCopy creates a deep copy of the channel, including the incoming policy.
+func (c *DirectedChannel) DeepCopy() *DirectedChannel {
+	channelCopy := *c
+
+	if channelCopy.InPolicy != nil {
+		inPolicyCopy := *channelCopy.InPolicy
+		channelCopy.InPolicy = &inPolicyCopy
+
+		// The fields for the ToNode can be overwritten by the path
+		// finding algorithm, which is why we need a deep copy in the
+		// first place. So we always start out with nil values, just to
+		// be sure they don't contain any old data.
+		channelCopy.InPolicy.ToNodePubKey = nil
+		channelCopy.InPolicy.ToNodeFeatures = nil
+	}
+
+	return &channelCopy
+}
+
+// GraphCache is a type that holds a minimal set of information of the public
+// channel graph that can be used for pathfinding.
+type GraphCache struct {
+	nodeChannels map[route.Vertex]map[uint64]*DirectedChannel
+	nodeFeatures map[route.Vertex]*lnwire.FeatureVector
+
+	mtx sync.RWMutex
+}
+
+// NewGraphCache creates a new graphCache.
+func NewGraphCache(preAllocNumNodes int) *GraphCache {
+	return &GraphCache{
+		nodeChannels: make(
+			map[route.Vertex]map[uint64]*DirectedChannel,
+			// A channel connects two nodes, so we can look it up
+			// from both sides, meaning we get double the number of
+			// entries.
+			preAllocNumNodes*2,
+		),
+		nodeFeatures: make(
+			map[route.Vertex]*lnwire.FeatureVector,
+			preAllocNumNodes,
+		),
+	}
+}
+
+// Stats returns statistics about the current cache size.
+func (c *GraphCache) Stats() string {
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+
+	numChannels := 0
+	for node := range c.nodeChannels {
+		numChannels += len(c.nodeChannels[node])
+	}
+	return fmt.Sprintf("num_node_features=%d, num_nodes=%d, "+
+		"num_channels=%d", len(c.nodeFeatures), len(c.nodeChannels),
+		numChannels)
+}
+
+// AddNodeFeatures adds a graph node and its features to the cache.
+func (c *GraphCache) AddNodeFeatures(node GraphCacheNode) {
+	nodePubKey := node.PubKey()
+
+	// Only hold the lock for a short time. The `ForEachChannel()` below is
+	// possibly slow as it has to go to the backend, so we can unlock
+	// between the calls. And the AddChannel() method will acquire its own
+	// lock anyway.
+	c.mtx.Lock()
+	c.nodeFeatures[nodePubKey] = node.Features()
+	c.mtx.Unlock()
+}
+
+// AddNode adds a graph node, including all the (directed) channels of that
+// node.
+func (c *GraphCache) AddNode(tx kvdb.RTx, node GraphCacheNode) error {
+	c.AddNodeFeatures(node)
+
+	return node.ForEachChannel(
+		tx, func(tx kvdb.RTx, info *models.ChannelEdgeInfo,
+			outPolicy *models.ChannelEdgePolicy,
+			inPolicy *models.ChannelEdgePolicy) error {
+
+			c.AddChannel(info, outPolicy, inPolicy)
+
+			return nil
+		},
+	)
+}
+
+// AddChannel adds a non-directed channel, meaning that the order of policy 1
+// and policy 2 does not matter, the directionality is extracted from the info
+// and policy flags automatically. The policy will be set as the outgoing policy
+// on one node and the incoming policy on the peer's side.
+func (c *GraphCache) AddChannel(info *models.ChannelEdgeInfo,
+	policy1 *models.ChannelEdgePolicy, policy2 *models.ChannelEdgePolicy) {
+
+	if info == nil {
+		return
+	}
+
+	if policy1 != nil && policy1.IsDisabled() &&
+		policy2 != nil && policy2.IsDisabled() {
+
+		return
+	}
+
+	// Create the edge entry for both nodes.
+	c.mtx.Lock()
+	c.updateOrAddEdge(info.NodeKey1Bytes, &DirectedChannel{
+		ChannelID: info.ChannelID,
+		IsNode1:   true,
+		OtherNode: info.NodeKey2Bytes,
+		Capacity:  info.Capacity,
+	})
+	c.updateOrAddEdge(info.NodeKey2Bytes, &DirectedChannel{
+		ChannelID: info.ChannelID,
+		IsNode1:   false,
+		OtherNode: info.NodeKey1Bytes,
+		Capacity:  info.Capacity,
+	})
+	c.mtx.Unlock()
+
+	// The policy's node is always the to_node. So if policy 1 has to_node
+	// of node 2 then we have the policy 1 as seen from node 1.
+	if policy1 != nil {
+		fromNode, toNode := info.NodeKey1Bytes, info.NodeKey2Bytes
+		if policy1.ToNode != info.NodeKey2Bytes {
+			fromNode, toNode = toNode, fromNode
+		}
+		isEdge1 := policy1.ChannelFlags&lnwire.ChanUpdateDirection == 0
+		c.UpdatePolicy(policy1, fromNode, toNode, isEdge1)
+	}
+	if policy2 != nil {
+		fromNode, toNode := info.NodeKey2Bytes, info.NodeKey1Bytes
+		if policy2.ToNode != info.NodeKey1Bytes {
+			fromNode, toNode = toNode, fromNode
+		}
+		isEdge1 := policy2.ChannelFlags&lnwire.ChanUpdateDirection == 0
+		c.UpdatePolicy(policy2, fromNode, toNode, isEdge1)
+	}
+}
+
+// updateOrAddEdge makes sure the edge information for a node is either updated
+// if it already exists or is added to that node's list of channels.
+func (c *GraphCache) updateOrAddEdge(node route.Vertex, edge *DirectedChannel) {
+	if len(c.nodeChannels[node]) == 0 {
+		c.nodeChannels[node] = make(map[uint64]*DirectedChannel)
+	}
+
+	c.nodeChannels[node][edge.ChannelID] = edge
+}
+
+// UpdatePolicy updates a single policy on both the from and to node. The order
+// of the from and to node is not strictly important. But we assume that a
+// channel edge was added beforehand so that the directed channel struct already
+// exists in the cache.
+func (c *GraphCache) UpdatePolicy(policy *models.ChannelEdgePolicy, fromNode,
+	toNode route.Vertex, edge1 bool) {
+
+	// Extract inbound fee if possible and available. If there is a decoding
+	// error, ignore this policy.
+	var inboundFee lnwire.Fee
+	_, err := policy.ExtraOpaqueData.ExtractRecords(&inboundFee)
+	if err != nil {
+		return
+	}
+
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	updatePolicy := func(nodeKey route.Vertex) {
+		if len(c.nodeChannels[nodeKey]) == 0 {
+			return
+		}
+
+		channel, ok := c.nodeChannels[nodeKey][policy.ChannelID]
+		if !ok {
+			return
+		}
+
+		// Edge 1 is defined as the policy for the direction of node1 to
+		// node2.
+		switch {
+		// This is node 1, and it is edge 1, so this is the outgoing
+		// policy for node 1.
+		case channel.IsNode1 && edge1:
+			channel.OutPolicySet = true
+			channel.InboundFee = inboundFee
+
+		// This is node 2, and it is edge 2, so this is the outgoing
+		// policy for node 2.
+		case !channel.IsNode1 && !edge1:
+			channel.OutPolicySet = true
+			channel.InboundFee = inboundFee
+
+		// The other two cases left mean it's the inbound policy for the
+		// node.
+		default:
+			channel.InPolicy = models.NewCachedPolicy(policy)
+		}
+	}
+
+	updatePolicy(fromNode)
+	updatePolicy(toNode)
+}
+
+// RemoveNode completely removes a node and all its channels (including the
+// peer's side).
+func (c *GraphCache) RemoveNode(node route.Vertex) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	delete(c.nodeFeatures, node)
+
+	// First remove all channels from the other nodes' lists.
+	for _, channel := range c.nodeChannels[node] {
+		c.removeChannelIfFound(channel.OtherNode, channel.ChannelID)
+	}
+
+	// Then remove our whole node completely.
+	delete(c.nodeChannels, node)
+}
+
+// RemoveChannel removes a single channel between two nodes.
+func (c *GraphCache) RemoveChannel(node1, node2 route.Vertex, chanID uint64) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	// Remove that one channel from both sides.
+	c.removeChannelIfFound(node1, chanID)
+	c.removeChannelIfFound(node2, chanID)
+}
+
+// removeChannelIfFound removes a single channel from one side.
+func (c *GraphCache) removeChannelIfFound(node route.Vertex, chanID uint64) {
+	if len(c.nodeChannels[node]) == 0 {
+		return
+	}
+
+	delete(c.nodeChannels[node], chanID)
+}
+
+// UpdateChannel updates the channel edge information for a specific edge. We
+// expect the edge to already exist and be known. If it does not yet exist, this
+// call is a no-op.
+func (c *GraphCache) UpdateChannel(info *models.ChannelEdgeInfo) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	if len(c.nodeChannels[info.NodeKey1Bytes]) == 0 ||
+		len(c.nodeChannels[info.NodeKey2Bytes]) == 0 {
+
+		return
+	}
+
+	channel, ok := c.nodeChannels[info.NodeKey1Bytes][info.ChannelID]
+	if ok {
+		// We only expect to be called when the channel is already
+		// known.
+		channel.Capacity = info.Capacity
+		channel.OtherNode = info.NodeKey2Bytes
+	}
+
+	channel, ok = c.nodeChannels[info.NodeKey2Bytes][info.ChannelID]
+	if ok {
+		channel.Capacity = info.Capacity
+		channel.OtherNode = info.NodeKey1Bytes
+	}
+}
+
+// getChannels returns a copy of the passed node's channels or nil if there
+// isn't any.
+func (c *GraphCache) getChannels(node route.Vertex) []*DirectedChannel {
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+
+	channels, ok := c.nodeChannels[node]
+	if !ok {
+		return nil
+	}
+
+	features, ok := c.nodeFeatures[node]
+	if !ok {
+		// If the features were set to nil explicitly, that's fine here.
+		// The router will overwrite the features of the destination
+		// node with those found in the invoice if necessary. But if we
+		// didn't yet get a node announcement we want to mimic the
+		// behavior of the old DB based code that would always set an
+		// empty feature vector instead of leaving it nil.
+		features = lnwire.EmptyFeatureVector()
+	}
+
+	toNodeCallback := func() route.Vertex {
+		return node
+	}
+
+	i := 0
+	channelsCopy := make([]*DirectedChannel, len(channels))
+	for _, channel := range channels {
+		// We need to copy the channel and policy to avoid it being
+		// updated in the cache if the path finding algorithm sets
+		// fields on it (currently only the ToNodeFeatures of the
+		// policy).
+		channelCopy := channel.DeepCopy()
+		if channelCopy.InPolicy != nil {
+			channelCopy.InPolicy.ToNodePubKey = toNodeCallback
+			channelCopy.InPolicy.ToNodeFeatures = features
+		}
+
+		channelsCopy[i] = channelCopy
+		i++
+	}
+
+	return channelsCopy
+}
+
+// ForEachChannel invokes the given callback for each channel of the given node.
+func (c *GraphCache) ForEachChannel(node route.Vertex,
+	cb func(channel *DirectedChannel) error) error {
+
+	// Obtain a copy of the node's channels. We need do this in order to
+	// avoid deadlocks caused by interaction with the graph cache, channel
+	// state and the graph database from multiple goroutines. This snapshot
+	// is only used for path finding where being stale is acceptable since
+	// the real world graph and our representation may always become
+	// slightly out of sync for a short time and the actual channel state
+	// is stored separately.
+	channels := c.getChannels(node)
+	for _, channel := range channels {
+		if err := cb(channel); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// ForEachNode iterates over the adjacency list of the graph, executing the
+// call back for each node and the set of channels that emanate from the given
+// node.
+//
+// NOTE: This method should be considered _read only_, the channels or nodes
+// passed in MUST NOT be modified.
+func (c *GraphCache) ForEachNode(cb func(node route.Vertex,
+	channels map[uint64]*DirectedChannel) error) error {
+
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+
+	for node, channels := range c.nodeChannels {
+		// We don't make a copy here since this is a read-only RPC
+		// call. We also don't need the node features either for this
+		// call.
+		if err := cb(node, channels); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// GetFeatures returns the features of the node with the given ID. If no
+// features are known for the node, an empty feature vector is returned.
+func (c *GraphCache) GetFeatures(node route.Vertex) *lnwire.FeatureVector {
+	c.mtx.RLock()
+	defer c.mtx.RUnlock()
+
+	features, ok := c.nodeFeatures[node]
+	if !ok || features == nil {
+		// The router expects the features to never be nil, so we return
+		// an empty feature set instead.
+		return lnwire.EmptyFeatureVector()
+	}
+
+	return features
+}

graph/db/graph_cache_test.go

@@ -0,0 +1,173 @@
+package graphdb
+
+import (
+	"encoding/hex"
+	"testing"
+
+	"github.com/lightningnetwork/lnd/graph/db/models"
+	"github.com/lightningnetwork/lnd/kvdb"
+	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/routing/route"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	pubKey1Bytes, _ = hex.DecodeString(
+		"0248f5cba4c6da2e4c9e01e81d1404dfac0cbaf3ee934a4fc117d2ea9a64" +
+			"22c91d",
+	)
+	pubKey2Bytes, _ = hex.DecodeString(
+		"038155ba86a8d3b23c806c855097ca5c9fa0f87621f1e7a7d2835ad057f6" +
+			"f4484f",
+	)
+
+	pubKey1, _ = route.NewVertexFromBytes(pubKey1Bytes)
+	pubKey2, _ = route.NewVertexFromBytes(pubKey2Bytes)
+)
+
+type node struct {
+	pubKey   route.Vertex
+	features *lnwire.FeatureVector
+
+	edgeInfos   []*models.ChannelEdgeInfo
+	outPolicies []*models.ChannelEdgePolicy
+	inPolicies  []*models.ChannelEdgePolicy
+}
+
+func (n *node) PubKey() route.Vertex {
+	return n.pubKey
+}
+func (n *node) Features() *lnwire.FeatureVector {
+	return n.features
+}
+
+func (n *node) ForEachChannel(tx kvdb.RTx,
+	cb func(kvdb.RTx, *models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+		*models.ChannelEdgePolicy) error) error {
+
+	for idx := range n.edgeInfos {
+		err := cb(
+			tx, n.edgeInfos[idx], n.outPolicies[idx],
+			n.inPolicies[idx],
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// TestGraphCacheAddNode tests that a channel going from node A to node B can be
+// cached correctly, independent of the direction we add the channel as.
+func TestGraphCacheAddNode(t *testing.T) {
+	t.Parallel()
+
+	runTest := func(nodeA, nodeB route.Vertex) {
+		t.Helper()
+
+		channelFlagA, channelFlagB := 0, 1
+		if nodeA == pubKey2 {
+			channelFlagA, channelFlagB = 1, 0
+		}
+
+		outPolicy1 := &models.ChannelEdgePolicy{
+			ChannelID:    1000,
+			ChannelFlags: lnwire.ChanUpdateChanFlags(channelFlagA),
+			ToNode:       nodeB,
+			// Define an inbound fee.
+			ExtraOpaqueData: []byte{
+				253, 217, 3, 8, 0, 0, 0, 10, 0, 0, 0, 20,
+			},
+		}
+		inPolicy1 := &models.ChannelEdgePolicy{
+			ChannelID:    1000,
+			ChannelFlags: lnwire.ChanUpdateChanFlags(channelFlagB),
+			ToNode:       nodeA,
+		}
+		node := &node{
+			pubKey:   nodeA,
+			features: lnwire.EmptyFeatureVector(),
+			edgeInfos: []*models.ChannelEdgeInfo{{
+				ChannelID: 1000,
+				// Those are direction independent!
+				NodeKey1Bytes: pubKey1,
+				NodeKey2Bytes: pubKey2,
+				Capacity:      500,
+			}},
+			outPolicies: []*models.ChannelEdgePolicy{outPolicy1},
+			inPolicies:  []*models.ChannelEdgePolicy{inPolicy1},
+		}
+		cache := NewGraphCache(10)
+		require.NoError(t, cache.AddNode(nil, node))
+
+		var fromChannels, toChannels []*DirectedChannel
+		_ = cache.ForEachChannel(nodeA, func(c *DirectedChannel) error {
+			fromChannels = append(fromChannels, c)
+			return nil
+		})
+		_ = cache.ForEachChannel(nodeB, func(c *DirectedChannel) error {
+			toChannels = append(toChannels, c)
+			return nil
+		})
+
+		require.Len(t, fromChannels, 1)
+		require.Len(t, toChannels, 1)
+
+		require.Equal(t, outPolicy1 != nil, fromChannels[0].OutPolicySet)
+		assertCachedPolicyEqual(t, inPolicy1, fromChannels[0].InPolicy)
+
+		require.Equal(t, inPolicy1 != nil, toChannels[0].OutPolicySet)
+		assertCachedPolicyEqual(t, outPolicy1, toChannels[0].InPolicy)
+
+		// Now that we've inserted two nodes into the graph, check that
+		// we'll recover the same set of channels during ForEachNode.
+		nodes := make(map[route.Vertex]struct{})
+		chans := make(map[uint64]struct{})
+		_ = cache.ForEachNode(func(node route.Vertex,
+			edges map[uint64]*DirectedChannel) error {
+
+			nodes[node] = struct{}{}
+			for chanID, directedChannel := range edges {
+				chans[chanID] = struct{}{}
+
+				if node == nodeA {
+					require.NotZero(
+						t, directedChannel.InboundFee,
+					)
+				} else {
+					require.Zero(
+						t, directedChannel.InboundFee,
+					)
+				}
+			}
+
+			return nil
+		})
+
+		require.Len(t, nodes, 2)
+		require.Len(t, chans, 1)
+	}
+
+	runTest(pubKey1, pubKey2)
+	runTest(pubKey2, pubKey1)
+}
+
+func assertCachedPolicyEqual(t *testing.T, original *models.ChannelEdgePolicy,
+	cached *models.CachedEdgePolicy) {
+
+	require.Equal(t, original.ChannelID, cached.ChannelID)
+	require.Equal(t, original.MessageFlags, cached.MessageFlags)
+	require.Equal(t, original.ChannelFlags, cached.ChannelFlags)
+	require.Equal(t, original.TimeLockDelta, cached.TimeLockDelta)
+	require.Equal(t, original.MinHTLC, cached.MinHTLC)
+	require.Equal(t, original.MaxHTLC, cached.MaxHTLC)
+	require.Equal(t, original.FeeBaseMSat, cached.FeeBaseMSat)
+	require.Equal(
+		t, original.FeeProportionalMillionths,
+		cached.FeeProportionalMillionths,
+	)
+	require.Equal(
+		t, route.Vertex(original.ToNode), cached.ToNodePubKey(),
+	)
+}

graph/db/log.go

@@ -0,0 +1,31 @@
+package graphdb
+
+import (
+	"github.com/btcsuite/btclog"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+// Subsystem defines the logging code for this subsystem.
+const Subsystem = "GRDB"
+
+// log is a logger that is initialized with no output filters.  This
+// means the package will not perform any logging by default until the caller
+// requests it.
+var log btclog.Logger
+
+func init() {
+	UseLogger(build.NewSubLogger(Subsystem, nil))
+}
+
+// DisableLog disables all library log output.  Logging output is disabled
+// by default until UseLogger is called.
+func DisableLog() {
+	UseLogger(btclog.Disabled)
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

graph/db/options.go

@@ -0,0 +1,93 @@
+package graphdb
+
+import "time"
+
+const (
+	// DefaultRejectCacheSize is the default number of rejectCacheEntries to
+	// cache for use in the rejection cache of incoming gossip traffic. This
+	// produces a cache size of around 1MB.
+	DefaultRejectCacheSize = 50000
+
+	// DefaultChannelCacheSize is the default number of ChannelEdges cached
+	// in order to reply to gossip queries. This produces a cache size of
+	// around 40MB.
+	DefaultChannelCacheSize = 20000
+
+	// DefaultPreAllocCacheNumNodes is the default number of channels we
+	// assume for mainnet for pre-allocating the graph cache. As of
+	// September 2021, there currently are 14k nodes in a strictly pruned
+	// graph, so we choose a number that is slightly higher.
+	DefaultPreAllocCacheNumNodes = 15000
+)
+
+// Options holds parameters for tuning and customizing a graph.DB.
+type Options struct {
+	// RejectCacheSize is the maximum number of rejectCacheEntries to hold
+	// in the rejection cache.
+	RejectCacheSize int
+
+	// ChannelCacheSize is the maximum number of ChannelEdges to hold in the
+	// channel cache.
+	ChannelCacheSize int
+
+	// BatchCommitInterval is the maximum duration the batch schedulers will
+	// wait before attempting to commit a pending set of updates.
+	BatchCommitInterval time.Duration
+
+	// PreAllocCacheNumNodes is the number of nodes we expect to be in the
+	// graph cache, so we can pre-allocate the map accordingly.
+	PreAllocCacheNumNodes int
+
+	// UseGraphCache denotes whether the in-memory graph cache should be
+	// used or a fallback version that uses the underlying database for
+	// path finding.
+	UseGraphCache bool
+
+	// NoMigration specifies that underlying backend was opened in read-only
+	// mode and migrations shouldn't be performed. This can be useful for
+	// applications that use the channeldb package as a library.
+	NoMigration bool
+}
+
+// DefaultOptions returns an Options populated with default values.
+func DefaultOptions() *Options {
+	return &Options{
+		RejectCacheSize:       DefaultRejectCacheSize,
+		ChannelCacheSize:      DefaultChannelCacheSize,
+		PreAllocCacheNumNodes: DefaultPreAllocCacheNumNodes,
+		UseGraphCache:         true,
+		NoMigration:           false,
+	}
+}
+
+// OptionModifier is a function signature for modifying the default Options.
+type OptionModifier func(*Options)
+
+// WithChannelCacheSize sets the ChannelCacheSize to n.
+func WithChannelCacheSize(n int) OptionModifier {
+	return func(o *Options) {
+		o.ChannelCacheSize = n
+	}
+}
+
+// WithPreAllocCacheNumNodes sets the PreAllocCacheNumNodes to n.
+func WithPreAllocCacheNumNodes(n int) OptionModifier {
+	return func(o *Options) {
+		o.PreAllocCacheNumNodes = n
+	}
+}
+
+// WithBatchCommitInterval sets the batch commit interval for the interval batch
+// schedulers.
+func WithBatchCommitInterval(interval time.Duration) OptionModifier {
+	return func(o *Options) {
+		o.BatchCommitInterval = interval
+	}
+}
+
+// WithUseGraphCache sets the UseGraphCache option to the given value.
+func WithUseGraphCache(use bool) OptionModifier {
+	return func(o *Options) {
+		o.UseGraphCache = use
+	}
+}

graph/db/reject_cache.go

@@ -0,0 +1,95 @@
+package graphdb
+
+// rejectFlags is a compact representation of various metadata stored by the
+// reject cache about a particular channel.
+type rejectFlags uint8
+
+const (
+	// rejectFlagExists is a flag indicating whether the channel exists,
+	// i.e. the channel is open and has a recent channel update. If this
+	// flag is not set, the channel is either a zombie or unknown.
+	rejectFlagExists rejectFlags = 1 << iota
+
+	// rejectFlagZombie is a flag indicating whether the channel is a
+	// zombie, i.e. the channel is open but has no recent channel updates.
+	rejectFlagZombie
+)
+
+// packRejectFlags computes the rejectFlags corresponding to the passed boolean
+// values indicating whether the edge exists or is a zombie.
+func packRejectFlags(exists, isZombie bool) rejectFlags {
+	var flags rejectFlags
+	if exists {
+		flags |= rejectFlagExists
+	}
+	if isZombie {
+		flags |= rejectFlagZombie
+	}
+
+	return flags
+}
+
+// unpack returns the booleans packed into the rejectFlags. The first indicates
+// if the edge exists in our graph, the second indicates if the edge is a
+// zombie.
+func (f rejectFlags) unpack() (bool, bool) {
+	return f&rejectFlagExists == rejectFlagExists,
+		f&rejectFlagZombie == rejectFlagZombie
+}
+
+// rejectCacheEntry caches frequently accessed information about a channel,
+// including the timestamps of its latest edge policies and whether or not the
+// channel exists in the graph.
+type rejectCacheEntry struct {
+	upd1Time int64
+	upd2Time int64
+	flags    rejectFlags
+}
+
+// rejectCache is an in-memory cache used to improve the performance of
+// HasChannelEdge. It caches information about the whether or channel exists, as
+// well as the most recent timestamps for each policy (if they exists).
+type rejectCache struct {
+	n     int
+	edges map[uint64]rejectCacheEntry
+}
+
+// newRejectCache creates a new rejectCache with maximum capacity of n entries.
+func newRejectCache(n int) *rejectCache {
+	return &rejectCache{
+		n:     n,
+		edges: make(map[uint64]rejectCacheEntry, n),
+	}
+}
+
+// get returns the entry from the cache for chanid, if it exists.
+func (c *rejectCache) get(chanid uint64) (rejectCacheEntry, bool) {
+	entry, ok := c.edges[chanid]
+	return entry, ok
+}
+
+// insert adds the entry to the reject cache. If an entry for chanid already
+// exists, it will be replaced with the new entry. If the entry doesn't exists,
+// it will be inserted to the cache, performing a random eviction if the cache
+// is at capacity.
+func (c *rejectCache) insert(chanid uint64, entry rejectCacheEntry) {
+	// If entry exists, replace it.
+	if _, ok := c.edges[chanid]; ok {
+		c.edges[chanid] = entry
+		return
+	}
+
+	// Otherwise, evict an entry at random and insert.
+	if len(c.edges) == c.n {
+		for id := range c.edges {
+			delete(c.edges, id)
+			break
+		}
+	}
+	c.edges[chanid] = entry
+}
+
+// remove deletes an entry for chanid from the cache, if it exists.
+func (c *rejectCache) remove(chanid uint64) {
+	delete(c.edges, chanid)
+}

graph/db/reject_cache_test.go

@@ -0,0 +1,107 @@
+package graphdb
+
+import (
+	"reflect"
+	"testing"
+)
+
+// TestRejectCache checks the behavior of the rejectCache with respect to insertion,
+// eviction, and removal of cache entries.
+func TestRejectCache(t *testing.T) {
+	const cacheSize = 100
+
+	// Create a new reject cache with the configured max size.
+	c := newRejectCache(cacheSize)
+
+	// As a sanity check, assert that querying the empty cache does not
+	// return an entry.
+	_, ok := c.get(0)
+	if ok {
+		t.Fatalf("reject cache should be empty")
+	}
+
+	// Now, fill up the cache entirely.
+	for i := uint64(0); i < cacheSize; i++ {
+		c.insert(i, entryForInt(i))
+	}
+
+	// Assert that the cache has all of the entries just inserted, since no
+	// eviction should occur until we try to surpass the max size.
+	assertHasEntries(t, c, 0, cacheSize)
+
+	// Now, insert a new element that causes the cache to evict an element.
+	c.insert(cacheSize, entryForInt(cacheSize))
+
+	// Assert that the cache has this last entry, as the cache should evict
+	// some prior element and not the newly inserted one.
+	assertHasEntries(t, c, cacheSize, cacheSize)
+
+	// Iterate over all inserted elements and construct a set of the evicted
+	// elements.
+	evicted := make(map[uint64]struct{})
+	for i := uint64(0); i < cacheSize+1; i++ {
+		_, ok := c.get(i)
+		if !ok {
+			evicted[i] = struct{}{}
+		}
+	}
+
+	// Assert that exactly one element has been evicted.
+	numEvicted := len(evicted)
+	if numEvicted != 1 {
+		t.Fatalf("expected one evicted entry, got: %d", numEvicted)
+	}
+
+	// Remove the highest item which initially caused the eviction and
+	// reinsert the element that was evicted prior.
+	c.remove(cacheSize)
+	for i := range evicted {
+		c.insert(i, entryForInt(i))
+	}
+
+	// Since the removal created an extra slot, the last insertion should
+	// not have caused an eviction and the entries for all channels in the
+	// original set that filled the cache should be present.
+	assertHasEntries(t, c, 0, cacheSize)
+
+	// Finally, reinsert the existing set back into the cache and test that
+	// the cache still has all the entries. If the randomized eviction were
+	// happening on inserts for existing cache items, we expect this to fail
+	// with high probability.
+	for i := uint64(0); i < cacheSize; i++ {
+		c.insert(i, entryForInt(i))
+	}
+	assertHasEntries(t, c, 0, cacheSize)
+
+}
+
+// assertHasEntries queries the reject cache for all channels in the range [start,
+// end), asserting that they exist and their value matches the entry produced by
+// entryForInt.
+func assertHasEntries(t *testing.T, c *rejectCache, start, end uint64) {
+	t.Helper()
+
+	for i := start; i < end; i++ {
+		entry, ok := c.get(i)
+		if !ok {
+			t.Fatalf("reject cache should contain chan %d", i)
+		}
+
+		expEntry := entryForInt(i)
+		if !reflect.DeepEqual(entry, expEntry) {
+			t.Fatalf("entry mismatch, want: %v, got: %v",
+				expEntry, entry)
+		}
+	}
+}
+
+// entryForInt generates a unique rejectCacheEntry given an integer.
+func entryForInt(i uint64) rejectCacheEntry {
+	exists := i%2 == 0
+	isZombie := i%3 == 0
+	return rejectCacheEntry{
+		upd1Time: int64(2 * i),
+		upd2Time: int64(2*i + 1),
+		flags:    packRejectFlags(exists, isZombie),
+	}
+}

graph/db/setup_test.go

@@ -0,0 +1,11 @@
+package graphdb
+
+import (
+	"testing"
+
+	"github.com/lightningnetwork/lnd/kvdb"
+)
+
+func TestMain(m *testing.M) {
+	kvdb.RunTests(m)
+}