wip: AddLightningNode cache updates

Author: djkazic

channeldb/error.go

@@ -43,6 +43,8 @@ var (
 	// created.
 	ErrMetaNotFound = fmt.Errorf("unable to locate meta information")
 
+	ErrGraphNotReady = fmt.Errorf("graph cache is not ready")
+
 	// ErrGraphNotFound is returned when at least one of the components of
 	// graph doesn't exist.
 	ErrGraphNotFound = fmt.Errorf("graph bucket not initialized")

channeldb/graph.go

@@ -20,9 +20,11 @@ import (
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/btcsuite/btcd/wire"
+	"github.com/btcsuite/btcwallet/walletdb"
 	"github.com/lightningnetwork/lnd/aliasmgr"
 	"github.com/lightningnetwork/lnd/batch"
 	"github.com/lightningnetwork/lnd/channeldb/models"
+	"github.com/lightningnetwork/lnd/fn"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/lnwire"
@@ -185,9 +187,11 @@ type ChannelGraph struct {
 	chanCache   *channelCache
 	graphCache  *GraphCache
 
-	graphCacheOnce  sync.Once
 	graphCacheReady atomic.Bool
 	graphCacheErr   error
+	updateQueue     *fn.ConcurrentQueue[func(tx walletdb.ReadWriteTx) error]
+
+	cachePopulated sync.WaitGroup
 
 	chanScheduler batch.Scheduler
 	nodeScheduler batch.Scheduler
@@ -216,11 +220,14 @@ func NewChannelGraph(db kvdb.Backend, rejectCacheSize, chanCacheSize int,
 	g.nodeScheduler = batch.NewTimeScheduler(
 		db, nil, batchCommitInterval,
 	)
+	g.updateQueue =
+		fn.NewConcurrentQueue[func(tx walletdb.ReadWriteTx) error](100)
 
 	// The graph cache can be turned off (e.g. for mobile users) for a
 	// speed/memory usage tradeoff.
 	if useGraphCache {
 		g.graphCache = NewGraphCache(preAllocCacheNumNodes)
+		g.cachePopulated.Add(1)
 		// Start populating the cache asynchronously
 		go g.populateGraphCache()
 	}
@@ -235,6 +242,8 @@ type channelMapKey struct {
 }
 
 func (g *ChannelGraph) populateGraphCache() {
+	defer g.cachePopulated.Done()
+
 	startTime := time.Now()
 	log.Debugf("Populating in-memory channel graph, this might " +
 		"take a while...")
@@ -263,27 +272,36 @@ func (g *ChannelGraph) populateGraphCache() {
 		return
 	}
 
-	log.Debugf("Finished populating in-memory channel graph (took "+
-		"%v, %s)", time.Since(startTime), g.graphCache.Stats())
+	if g.graphCache != nil {
+		log.Debugf("Finished populating in-memory channel graph (took "+
+			"%v, %s)", time.Since(startTime), g.graphCache.Stats())
+	}
 
+	// Set graphCache to be ready and process pending updates.
 	g.graphCacheReady.Store(true)
+	g.updateQueue.Start()
+	go func() {
+		for update := range g.updateQueue.ChanOut() {
+			err := g.nodeScheduler.Execute(&batch.Request{
+				Update: func(tx kvdb.RwTx) error {
+					return update(tx)
+				},
+			})
+			if err != nil {
+				fmt.Printf("Failed to execute cache update "+
+					"function: %v\n", err)
+			}
+		}
+	}()
 }
 
 func (c *ChannelGraph) getGraphCache() (*GraphCache, error) {
 	if c.graphCache == nil {
 		return nil, nil
 	}
 
-	// Wait for up to 5 seconds for the cache to be ready
-	timeout := time.After(5 * time.Second)
-	for !c.graphCacheReady.Load() {
-		select {
-		case <-timeout:
-			return nil, fmt.Errorf("timeout waiting for graph " +
-				"cache to be ready")
-		case <-time.After(100 * time.Millisecond):
-			// Check again
-		}
+	if !c.graphCacheReady.Load() {
+		return nil, ErrGraphNotReady
 	}
 
 	if c.graphCacheErr != nil {
@@ -506,6 +524,7 @@ func (c *ChannelGraph) ForEachChannel(cb func(*models.ChannelEdgeInfo,
 func (c *ChannelGraph) ForEachNodeDirectedChannel(tx kvdb.RTx,
 	node route.Vertex, cb func(channel *DirectedChannel) error) error {
 
+	// If graphCache is fully populated, use it.
 	graphCache, err := c.getGraphCache()
 	if err == nil && graphCache != nil {
 		return graphCache.ForEachChannel(node, cb)
@@ -564,6 +583,7 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(tx kvdb.RTx,
 func (c *ChannelGraph) FetchNodeFeatures(
 	node route.Vertex) (*lnwire.FeatureVector, error) {
 
+	// If graphCache is fully populated, use it.
 	graphCache, err := c.getGraphCache()
 	if err == nil && graphCache != nil {
 		return graphCache.GetFeatures(node), nil
@@ -875,17 +895,6 @@ func (c *ChannelGraph) AddLightningNode(node *LightningNode,
 
 	r := &batch.Request{
 		Update: func(tx kvdb.RwTx) error {
-			graphCache, err := c.getGraphCache()
-			if err == nil && graphCache != nil {
-				cNode := newGraphCacheNode(
-					node.PubKeyBytes, node.Features,
-				)
-				err := graphCache.AddNode(tx, cNode)
-				if err != nil {
-					return err
-				}
-			}
-
 			return addLightningNode(tx, node)
 		},
 	}
@@ -894,6 +903,31 @@ func (c *ChannelGraph) AddLightningNode(node *LightningNode,
 		f(r)
 	}
 
+	updateCache := func(tx kvdb.RwTx) error {
+		graphCache, err := c.getGraphCache()
+		if err == nil && graphCache != nil {
+			cNode := newGraphCacheNode(node.PubKeyBytes, node.Features)
+			err := graphCache.AddNode(tx, cNode)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	// If the graph cache is not ready, queue the cache update
+	if !c.graphCacheReady.Load() {
+		c.updateQueue.ChanIn() <- updateCache
+	} else {
+		// If the cache is ready, execute the cache update immediately
+		err := c.nodeScheduler.Execute(&batch.Request{
+			Update: updateCache,
+		})
+		if err != nil {
+			return err
+		}
+	}
+
 	return c.nodeScheduler.Execute(r)
 }
 

channeldb/graph_test.go

@@ -4007,14 +4007,14 @@ func TestGraphLoading(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	_, err = graph.getGraphCache()
-	require.NoError(t, err)
-
 	// Populate the graph with test data.
 	const numNodes = 100
 	const numChannels = 4
 	_, _ = fillTestGraph(t, graph, numNodes, numChannels)
 
+	// Wait for graph cache to be populated.
+	graph.cachePopulated.Wait()
+
 	// Recreate the graph. This should cause the graph cache to be
 	// populated.
 	graphReloaded, err := NewChannelGraph(
@@ -4024,8 +4024,8 @@ func TestGraphLoading(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	_, err = graphReloaded.getGraphCache()
-	require.NoError(t, err)
+	// Wait for graph cache to be populated.
+	graphReloaded.cachePopulated.Wait()
 
 	// Assert that the cache content is identical.
 	require.Equal(