wip: pending writes

Author: djkazic

channeldb/graph.go

@@ -22,6 +22,7 @@ import (
 	"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"
@@ -192,6 +193,9 @@ type ChannelGraph struct {
 
 	// cacheErr returns errors from the graph cache loading goroutine.
 	cacheErr chan error
+
+	// cacheUpdates stores deferred in-flight cache writes.
+	cacheUpdates *fn.ConcurrentQueue[func()]
 }
 
 // NewChannelGraph allocates a new ChannelGraph backed by a DB instance. The
@@ -217,6 +221,7 @@ func NewChannelGraph(db kvdb.Backend, rejectCacheSize, chanCacheSize int,
 	g.nodeScheduler = batch.NewTimeScheduler(
 		db, nil, batchCommitInterval,
 	)
+	g.cacheUpdates = fn.NewConcurrentQueue[func()](1000)
 
 	// The graph cache can be turned off (e.g. for mobile users) for a
 	// speed/memory usage tradeoff.
@@ -226,6 +231,7 @@ func NewChannelGraph(db kvdb.Backend, rejectCacheSize, chanCacheSize int,
 		// Start populating the cache asynchronously.
 		g.cacheReady = make(chan struct{})
 		g.cacheErr = make(chan error, 1)
+		g.cacheUpdates.Start()
 		go g.populateGraphCache()
 	}
 
@@ -266,6 +272,14 @@ func (c *ChannelGraph) populateGraphCache() {
 		return
 	}
 
+	go func() {
+		for update := range c.cacheUpdates.ChanOut() {
+			c.cacheMu.Lock()
+			update()
+			c.cacheMu.Unlock()
+		}
+	}()
+
 	if c.graphCache != nil {
 		log.Debugf("Finished populating in-memory channel graph (took "+
 			"%v, %s)", time.Since(startTime), c.graphCache.Stats())
@@ -289,16 +303,46 @@ func (c *ChannelGraph) populateGraphCache() {
 // This method is non-blocking and should be used to safely access the graph
 // cache from concurrent goroutines.
 func (c *ChannelGraph) getGraphCache() (*GraphCache, error) {
+	// If graph cache is not being used, return nil
 	if c.graphCache == nil {
 		return nil, nil
 	}
 
+	// Check if there was an error during population
+	if err := c.checkCacheErr(); err != nil {
+		return nil, fmt.Errorf("graph cache population failed: %w", err)
+	}
+
+	// At this point, we know the cache exists and there were no errors
+	// We can return it regardless of whether it's fully populated or not.
+	return c.graphCache, nil
+}
+
+func (c *ChannelGraph) waitUntilGraphReady() {
+	<-c.cacheReady
+}
+
+// checkCacheErr is a helper function to check for errors in cache population
+func (c *ChannelGraph) checkCacheErr() error {
 	select {
-	case <-c.cacheReady:
-		return c.graphCache, nil
 	case err := <-c.cacheErr:
-		return nil, fmt.Errorf("graph cache population failed: %w", err)
+		return err
+	default:
+		return nil
+	}
+}
+
+// enqueueCacheUpdate either defers a graph cache write or writes it
+// right away, depending on the cache state.
+func (c *ChannelGraph) enqueueCacheUpdate(update func()) {
+	if c.graphCache == nil {
+		// If we're not using the graph cache, execute the update
+		update()
+		return
 	}
+
+	// Otherwise, queue the update
+	c.cacheUpdates.ChanIn() <- update
 }
 
 // getChannelMap loads all channel edge policies from the database and stores
@@ -892,16 +936,15 @@ 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
+			c.enqueueCacheUpdate(func() {
+				graphCache, err := c.getGraphCache()
+				if err == nil && graphCache != nil {
+					cNode := newGraphCacheNode(
+						node.PubKeyBytes, node.Features,
+					)
+					graphCache.AddNode(tx, cNode)
 				}
-			}
+			})
 
 			return addLightningNode(tx, node)
 		},
@@ -981,10 +1024,12 @@ func (c *ChannelGraph) DeleteLightningNode(nodePub route.Vertex) error {
 			return ErrGraphNodeNotFound
 		}
 
-		graphCache, err := c.getGraphCache()
-		if err == nil && graphCache != nil {
-			graphCache.RemoveNode(nodePub)
-		}
+		c.enqueueCacheUpdate(func() {
+			graphCache, err := c.getGraphCache()
+			if err == nil && graphCache != nil {
+				graphCache.RemoveNode(nodePub)
+			}
+		})
 
 		return c.deleteLightningNode(nodes, nodePub[:])
 	}, func() {})
@@ -1113,10 +1158,12 @@ func (c *ChannelGraph) addChannelEdge(tx kvdb.RwTx,
 		return ErrEdgeAlreadyExist
 	}
 
-	graphCache, err := c.getGraphCache()
-	if err == nil && graphCache != nil {
-		graphCache.AddChannel(edge, nil, nil)
-	}
+	c.enqueueCacheUpdate(func() {
+		graphCache, err := c.getGraphCache()
+		if err == nil && graphCache != nil {
+			graphCache.AddChannel(edge, nil, nil)
+		}
+	})
 
 	// Before we insert the channel into the database, we'll ensure that
 	// both nodes already exist in the channel graph. If either node
@@ -1317,10 +1364,12 @@ func (c *ChannelGraph) UpdateChannelEdge(edge *models.ChannelEdgeInfo) error {
 			return ErrEdgeNotFound
 		}
 
-		graphCache, err := c.getGraphCache()
-		if err == nil && graphCache != nil {
-			graphCache.UpdateChannel(edge)
-		}
+		c.enqueueCacheUpdate(func() {
+			graphCache, err := c.getGraphCache()
+			if err == nil && graphCache != nil {
+				graphCache.UpdateChannel(edge)
+			}
+		})
 
 		return putChanEdgeInfo(edgeIndex, edge, chanKey)
 	}, func() {})
@@ -1565,10 +1614,12 @@ func (c *ChannelGraph) pruneGraphNodes(nodes kvdb.RwBucket,
 			continue
 		}
 
-		graphCache, err := c.getGraphCache()
-		if err == nil && graphCache != nil {
-			graphCache.RemoveNode(nodePubKey)
-		}
+		c.enqueueCacheUpdate(func() {
+			graphCache, err := c.getGraphCache()
+			if err == nil && graphCache != nil {
+				graphCache.RemoveNode(nodePubKey)
+			}
+		})
 
 		// If we reach this point, then there are no longer any edges
 		// that connect this node, so we can delete it.
@@ -2597,13 +2648,15 @@ func (c *ChannelGraph) delChannelEdgeUnsafe(edges, edgeIndex, chanIndex,
 		return err
 	}
 
-	graphCache, err := c.getGraphCache()
-	if err == nil && graphCache != nil {
-		graphCache.RemoveChannel(
-			edgeInfo.NodeKey1Bytes, edgeInfo.NodeKey2Bytes,
-			edgeInfo.ChannelID,
-		)
-	}
+	c.enqueueCacheUpdate(func() {
+		graphCache, err := c.getGraphCache()
+		if err == nil && graphCache != nil {
+			graphCache.RemoveChannel(
+				edgeInfo.NodeKey1Bytes, edgeInfo.NodeKey2Bytes,
+				edgeInfo.ChannelID,
+			)
+		}
+	})
 
 	// We'll also remove the entry in the edge update index bucket before
 	// we delete the edges themselves so we can access their last update
@@ -2736,10 +2789,9 @@ func (c *ChannelGraph) UpdateEdgePolicy(edge *models.ChannelEdgePolicy,
 		},
 		Update: func(tx kvdb.RwTx) error {
 			var err error
-			graphCache, _ := c.getGraphCache()
 
 			isUpdate1, err = updateEdgePolicy(
-				tx, edge, graphCache,
+				tx, edge, c,
 			)
 
 			// Silence ErrEdgeNotFound so that the batch can
@@ -2806,7 +2858,7 @@ func (c *ChannelGraph) updateEdgeCache(e *models.ChannelEdgePolicy,
 // true if the updated policy belongs to node1, and false if the policy belonged
 // to node2.
 func updateEdgePolicy(tx kvdb.RwTx, edge *models.ChannelEdgePolicy,
-	graphCache *GraphCache) (bool, error) {
+	c *ChannelGraph) (bool, error) {
 
 	edges := tx.ReadWriteBucket(edgeBucket)
 	if edges == nil {
@@ -2857,11 +2909,14 @@ func updateEdgePolicy(tx kvdb.RwTx, edge *models.ChannelEdgePolicy,
 	copy(fromNodePubKey[:], fromNode)
 	copy(toNodePubKey[:], toNode)
 
-	if graphCache != nil {
-		graphCache.UpdatePolicy(
-			edge, fromNodePubKey, toNodePubKey, isUpdate1,
-		)
-	}
+	c.enqueueCacheUpdate(func() {
+		graphCache, _ := c.getGraphCache()
+		if graphCache != nil {
+			graphCache.UpdatePolicy(
+				edge, fromNodePubKey, toNodePubKey, isUpdate1,
+			)
+		}
+	})
 
 	return isUpdate1, nil
 }
@@ -3732,10 +3787,12 @@ func (c *ChannelGraph) MarkEdgeZombie(chanID uint64,
 				"bucket: %w", err)
 		}
 
-		graphCache, cacheErr := c.getGraphCache()
-		if cacheErr == nil && graphCache != nil {
-			graphCache.RemoveChannel(pubKey1, pubKey2, chanID)
-		}
+		c.enqueueCacheUpdate(func() {
+			graphCache, cacheErr := c.getGraphCache()
+			if cacheErr == nil && graphCache != nil {
+				graphCache.RemoveChannel(pubKey1, pubKey2, chanID)
+			}
+		})
 
 		return markEdgeZombie(zombieIndex, chanID, pubKey1, pubKey2)
 	})
@@ -3825,10 +3882,12 @@ func (c *ChannelGraph) markEdgeLiveUnsafe(tx kvdb.RwTx, chanID uint64) error {
 		}
 
 		for _, edgeInfo := range edgeInfos {
-			graphCache.AddChannel(
-				edgeInfo.Info, edgeInfo.Policy1,
-				edgeInfo.Policy2,
-			)
+			c.enqueueCacheUpdate(func() {
+				graphCache.AddChannel(
+					edgeInfo.Info, edgeInfo.Policy1,
+					edgeInfo.Policy2,
+				)
+			})
 		}
 	}
 

channeldb/graph_test.go

@@ -4007,14 +4007,14 @@ func TestGraphLoading(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	// Block until graph cache is loaded.
-	graph.getGraphCache()
-
 	// Populate the graph with test data.
 	const numNodes = 100
 	const numChannels = 4
 	_, _ = fillTestGraph(t, graph, numNodes, numChannels)
 
+	// Block until graph cache is loaded and pending updates are applied.
+	graph.waitUntilGraphReady()
+
 	// Recreate the graph. This should cause the graph cache to be
 	// populated.
 	graphReloaded, err := NewChannelGraph(
@@ -4024,17 +4024,23 @@ func TestGraphLoading(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	// Block until graph cache is loaded.
-	graphReloaded.getGraphCache()
+	// Block until graph cache is ready.
+	graphReloaded.waitUntilGraphReady()
 
 	// Assert that the cache content is identical.
-	require.Equal(
-		t, graph.graphCache.nodeChannels,
-		graphReloaded.graphCache.nodeChannels,
+	require.True(
+		t,
+		reflect.DeepEqual(
+			graph.graphCache.nodeChannels,
+			graphReloaded.graphCache.nodeChannels,
+		),
 	)
 
-	require.Equal(
-		t, graph.graphCache.nodeFeatures,
-		graphReloaded.graphCache.nodeFeatures,
+	require.True(
+		t,
+		reflect.DeepEqual(
+			graph.graphCache.nodeFeatures,
+			graphReloaded.graphCache.nodeFeatures,
+		),
 	)
 }