[g175] graph/db: v2 columns and v2 node CRUD

autopilot/prefattach_test.go

@@ -31,12 +31,14 @@ type testGraph interface {
 }
 
 type testDBGraph struct {
-	db *graphdb.ChannelGraph
+	db *graphdb.VersionedReader
 	databaseChannelGraph
 }
 
 func newDiskChanGraph(t *testing.T) (testGraph, error) {
-	graphDB := graphdb.MakeTestGraph(t)
+	graphDB := graphdb.NewVersionedReader(
+		graphdb.MakeTestGraph(t), lnwire.GossipVersion1,
+	)
 	require.NoError(t, graphDB.Start())
 	t.Cleanup(func() {
 		require.NoError(t, graphDB.Stop())

graph/builder.go

@@ -111,7 +111,8 @@ type Builder struct {
 
 	bestHeight atomic.Uint32
 
-	cfg *Config
+	cfg         *Config
+	graphReader *graphdb.VersionedReader
 
 	// newBlocks is a channel in which new blocks connected to the end of
 	// the main chain are sent over, and blocks updated after a call to
@@ -146,7 +147,11 @@ var _ ChannelGraphSource = (*Builder)(nil)
 // NewBuilder constructs a new Builder.
 func NewBuilder(cfg *Config) (*Builder, error) {
 	return &Builder{
-		cfg:            cfg,
+		cfg: cfg,
+		// For now, we'll just use V1 graph reader.
+		graphReader: graphdb.NewVersionedReader(
+			cfg.Graph, lnwire.GossipVersion1,
+		),
 		channelEdgeMtx: multimutex.NewMutex[uint64](),
 		statTicker:     ticker.New(defaultStatInterval),
 		stats:          new(builderStats),
@@ -1266,7 +1271,7 @@ func (b *Builder) GetChannelByID(chanID lnwire.ShortChannelID) (
 func (b *Builder) FetchNode(ctx context.Context,
 	node route.Vertex) (*models.Node, error) {
 
-	return b.cfg.Graph.FetchNode(ctx, node)
+	return b.graphReader.FetchNode(ctx, node)
 }
 
 // ForAllOutgoingChannels is used to iterate over all outgoing channels owned by

graph/builder_test.go

@@ -359,7 +359,7 @@ func TestWakeUpOnStaleBranch(t *testing.T) {
 	// Create new router with same graph database.
 	router, err := NewBuilder(&Config{
 		SelfNode:           selfNode.PubKeyBytes,
-		Graph:              ctx.graph,
+		Graph:              ctx.graph.ChannelGraph,
 		Chain:              ctx.chain,
 		ChainView:          ctx.chainView,
 		ChannelPruneExpiry: time.Hour * 24,
@@ -1595,7 +1595,9 @@ func parseTestGraph(t *testing.T, useCache bool, path string) (
 	}
 
 	return &testGraphInstance{
-		graph:      graph,
+		graph: graphdb.NewVersionedReader(
+			graph, lnwire.GossipVersion1,
+		),
 		aliasMap:   aliasMap,
 		privKeyMap: privKeyMap,
 		channelIDs: channelIDs,
@@ -1690,7 +1692,7 @@ func asymmetricTestChannel(alias1, alias2 string, capacity btcutil.Amount,
 
 // assertChannelsPruned ensures that only the given channels are pruned from the
 // graph out of the set of all channels.
-func assertChannelsPruned(t *testing.T, graph *graphdb.ChannelGraph,
+func assertChannelsPruned(t *testing.T, graph *graphdb.VersionedReader,
 	channels []*testChannel, prunedChanIDs ...uint64) {
 
 	t.Helper()
@@ -1980,7 +1982,9 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 	}
 
 	return &testGraphInstance{
-		graph:      graph,
+		graph: graphdb.NewVersionedReader(
+			graph, lnwire.GossipVersion1,
+		),
 		aliasMap:   aliasMap,
 		privKeyMap: privKeyMap,
 		links:      links,

graph/db/graph.go

@@ -293,23 +293,6 @@ func (c *ChannelGraph) AddNode(ctx context.Context,
 	return nil
 }
 
-// DeleteNode starts a new database transaction to remove a vertex/node
-// from the database according to the node's public key.
-func (c *ChannelGraph) DeleteNode(ctx context.Context,
-	nodePub route.Vertex) error {
-
-	err := c.db.DeleteNode(ctx, lnwire.GossipVersion1, nodePub)
-	if err != nil {
-		return err
-	}
-
-	if c.graphCache != nil {
-		c.graphCache.RemoveNode(nodePub)
-	}
-
-	return nil
-}
-
 // AddChannelEdge adds a new (undirected, blank) edge to the graph database. An
 // undirected edge from the two target nodes are created. The information stored
 // denotes the static attributes of the channel, such as the channelID, the keys
@@ -603,13 +586,6 @@ func (c *ChannelGraph) UpdateEdgePolicy(ctx context.Context,
 	return nil
 }
 
-// AddrsForNode returns all known addresses for the target node public key.
-func (c *ChannelGraph) AddrsForNode(ctx context.Context,
-	nodePub *btcec.PublicKey) (bool, []net.Addr, error) {
-
-	return c.db.AddrsForNode(ctx, lnwire.GossipVersion1, nodePub)
-}
-
 // ForEachSourceNodeChannel iterates through all channels of the source node.
 func (c *ChannelGraph) ForEachSourceNodeChannel(ctx context.Context,
 	cb func(chanPoint wire.OutPoint, havePolicy bool,
@@ -657,13 +633,6 @@ func (c *ChannelGraph) NodeUpdatesInHorizon(startTime, endTime time.Time,
 	return c.db.NodeUpdatesInHorizon(startTime, endTime, opts...)
 }
 
-// FetchNode attempts to look up a target node by its identity public key.
-func (c *ChannelGraph) FetchNode(ctx context.Context,
-	nodePub route.Vertex) (*models.Node, error) {
-
-	return c.db.FetchNode(ctx, lnwire.GossipVersion1, nodePub)
-}
-
 // HasV1Node determines if the graph has a vertex identified by the target node
 // in the V1 graph.
 func (c *ChannelGraph) HasV1Node(ctx context.Context,
@@ -672,14 +641,6 @@ func (c *ChannelGraph) HasV1Node(ctx context.Context,
 	return c.db.HasV1Node(ctx, nodePub)
 }
 
-// HasNode determines if the graph has a vertex identified by the target node
-// in the V1 graph.
-func (c *ChannelGraph) HasNode(ctx context.Context, nodePub [33]byte) (bool,
-	error) {
-
-	return c.db.HasNode(ctx, lnwire.GossipVersion1, nodePub)
-}
-
 // IsPublicNode determines whether the node is seen as public in the graph.
 func (c *ChannelGraph) IsPublicNode(pubKey [33]byte) (bool, error) {
 	return c.db.IsPublicNode(pubKey)
@@ -811,6 +772,62 @@ func (c *ChannelGraph) PruneTip() (*chainhash.Hash, uint32, error) {
 	return c.db.PruneTip()
 }
 
+// VersionedReader is a wrapper around ChannelGraph that will call underlying
+// Store methods with a specific gossip version.
+type VersionedReader struct {
+	*ChannelGraph
+	v lnwire.GossipVersion
+}
+
+// NewVersionedReader creates a new VersionedReader.
+func NewVersionedReader(c *ChannelGraph,
+	v lnwire.GossipVersion) *VersionedReader {
+
+	return &VersionedReader{
+		ChannelGraph: c,
+		v:            v,
+	}
+}
+
+// FetchNode attempts to look up a target node by its identity public key.
+func (c *VersionedReader) FetchNode(ctx context.Context,
+	nodePub route.Vertex) (*models.Node, error) {
+
+	return c.db.FetchNode(ctx, c.v, nodePub)
+}
+
+// AddrsForNode returns all known addresses for the target node public key.
+func (c *VersionedReader) AddrsForNode(ctx context.Context,
+	nodePub *btcec.PublicKey) (bool, []net.Addr, error) {
+
+	return c.db.AddrsForNode(ctx, c.v, nodePub)
+}
+
+// DeleteNode starts a new database transaction to remove a vertex/node
+// from the database according to the node's public key.
+func (c *VersionedReader) DeleteNode(ctx context.Context,
+	nodePub route.Vertex) error {
+
+	err := c.db.DeleteNode(ctx, c.v, nodePub)
+	if err != nil {
+		return err
+	}
+
+	if c.graphCache != nil {
+		c.graphCache.RemoveNode(nodePub)
+	}
+
+	return nil
+}
+
+// HasNode determines if the graph has a vertex identified by the target node
+// in the V1 graph.
+func (c *VersionedReader) HasNode(ctx context.Context, nodePub [33]byte) (bool,
+	error) {
+
+	return c.db.HasNode(ctx, c.v, nodePub)
+}
+
 // MakeTestGraph creates a new instance of the ChannelGraph for testing
 // purposes. The backing Store implementation depends on the version of
 // NewTestDB included in the current build.

graph/db/graph_test.go

@@ -98,7 +98,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()
 
-	graph := MakeTestGraph(t)
+	graph := NewVersionedReader(MakeTestGraph(t), lnwire.GossipVersion1)
 
 	// We'd like to test basic insertion/deletion for vertexes from the
 	// graph, so we'll create a test vertex to start with.
@@ -123,7 +123,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 	// without any errors.
 	node := nodeWithAddrs(testAddrs)
 	require.NoError(t, graph.AddNode(ctx, node))
-	assertNodeInCache(t, graph, node, testFeatures)
+	assertNodeInCache(t, graph.ChannelGraph, node, testFeatures)
 
 	// Our AddNode implementation uses the batcher meaning that it is
 	// possible that two updates for the same node announcement may be
@@ -153,7 +153,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 
 	// Check that the node's features are fetched correctly. This check
 	// will check the database directly.
-	features, err = graph.db.FetchNodeFeatures(
+	features, err = graph.ChannelGraph.db.FetchNodeFeatures(
 		lnwire.GossipVersion1, node.PubKeyBytes,
 	)
 	require.NoError(t, err)
@@ -162,7 +162,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) {
 	// Next, delete the node from the graph, this should purge all data
 	// related to the node.
 	require.NoError(t, graph.DeleteNode(ctx, testPub))
-	assertNodeNotInCache(t, graph, testPub)
+	assertNodeNotInCache(t, graph.ChannelGraph, testPub)
 
 	// Attempting to delete the node again should return an error since
 	// the node is no longer known.
@@ -287,7 +287,7 @@ func TestPartialNode(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()
 
-	graph := MakeTestGraph(t)
+	graph := NewVersionedReader(MakeTestGraph(t), lnwire.GossipVersion1)
 
 	// To insert a partial node, we need to add a channel edge that has
 	// node keys for nodes we are not yet aware
@@ -301,8 +301,8 @@ func TestPartialNode(t *testing.T) {
 
 	// Both of the nodes should now be in both the graph (as partial/shell)
 	// nodes _and_ the cache should also have an awareness of both nodes.
-	assertNodeInCache(t, graph, &node1, nil)
-	assertNodeInCache(t, graph, &node2, nil)
+	assertNodeInCache(t, graph.ChannelGraph, &node1, nil)
+	assertNodeInCache(t, graph.ChannelGraph, &node2, nil)
 
 	// Next, fetch the node2 from the database to ensure everything was
 	// serialized properly.
@@ -332,7 +332,7 @@ func TestPartialNode(t *testing.T) {
 	// Next, delete the node from the graph, this should purge all data
 	// related to the node.
 	require.NoError(t, graph.DeleteNode(ctx, pubKey1))
-	assertNodeNotInCache(t, graph, testPub)
+	assertNodeNotInCache(t, graph.ChannelGraph, testPub)
 
 	// Finally, attempt to fetch the node again. This should fail as the
 	// node should have been deleted from the database.
@@ -3691,7 +3691,7 @@ func TestPruneGraphNodes(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()
 
-	graph := MakeTestGraph(t)
+	graph := NewVersionedReader(MakeTestGraph(t), lnwire.GossipVersion1)
 
 	// We'll start off by inserting our source node, to ensure that it's
 	// the only node left after we prune the graph.
@@ -3742,7 +3742,7 @@ func TestPruneGraphNodes(t *testing.T) {
 	// source node (which can't be pruned), and node 1+2. Nodes 1 and two
 	// should still be left in the graph as there's half of an advertised
 	// edge between them.
-	assertNumNodes(t, graph, 3)
+	assertNumNodes(t, graph.ChannelGraph, 3)
 
 	// Finally, we'll ensure that node3, the only fully unconnected node as
 	// properly deleted from the graph and not another node in its place.
@@ -3757,7 +3757,7 @@ func TestAddChannelEdgeShellNodes(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()
 
-	graph := MakeTestGraph(t)
+	graph := NewVersionedReader(MakeTestGraph(t), lnwire.GossipVersion1)
 
 	// To start, we'll create two nodes, and only add one of them to the
 	// channel graph.
@@ -3796,7 +3796,7 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()
 
-	graph := MakeTestGraph(t)
+	graph := NewVersionedReader(MakeTestGraph(t), lnwire.GossipVersion1)
 
 	// We'll first populate our graph with a single node that will be
 	// removed shortly.
@@ -3842,6 +3842,7 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
 var (
 	updateTime   = prand.Int63()
 	updateTimeMu sync.Mutex
+	updateBlock  = prand.Uint32()
 )
 
 func nextUpdateTime() time.Time {
@@ -3853,6 +3854,15 @@ func nextUpdateTime() time.Time {
 	return time.Unix(updateTime, 0)
 }
 
+func nextBlockHeight() uint32 {
+	updateTimeMu.Lock()
+	defer updateTimeMu.Unlock()
+
+	updateBlock++
+
+	return updateBlock
+}
+
 // TestNodeIsPublic ensures that we properly detect nodes that are seen as
 // public within the network graph.
 func TestNodeIsPublic(t *testing.T) {
@@ -4755,7 +4765,7 @@ func TestLightningNodePersistence(t *testing.T) {
 	ctx := t.Context()
 
 	// Create a new test graph instance.
-	graph := MakeTestGraph(t)
+	graph := NewVersionedReader(MakeTestGraph(t), lnwire.GossipVersion1)
 
 	nodeAnnBytes, err := hex.DecodeString(testNodeAnn)
 	require.NoError(t, err)

graph/notifications_test.go

@@ -1042,7 +1042,7 @@ func TestEncodeHexColor(t *testing.T) {
 type testCtx struct {
 	builder *Builder
 
-	graph *graphdb.ChannelGraph
+	graph *graphdb.VersionedReader
 
 	aliases map[string]route.Vertex
 
@@ -1059,7 +1059,9 @@ type testCtx struct {
 func createTestCtxSingleNode(t *testing.T,
 	startingHeight uint32) *testCtx {
 
-	graph := graphdb.MakeTestGraph(t)
+	graph := graphdb.NewVersionedReader(
+		graphdb.MakeTestGraph(t), lnwire.GossipVersion1,
+	)
 	sourceNode := createTestNode(t)
 
 	require.NoError(t,
@@ -1086,7 +1088,7 @@ func (c *testCtx) RestartBuilder(t *testing.T) {
 	// start it.
 	builder, err := NewBuilder(&Config{
 		SelfNode:            selfNode.PubKeyBytes,
-		Graph:               c.graph,
+		Graph:               c.graph.ChannelGraph,
 		Chain:               c.chain,
 		ChainView:           c.chainView,
 		Notifier:            c.builder.cfg.Notifier,
@@ -1108,7 +1110,7 @@ func (c *testCtx) RestartBuilder(t *testing.T) {
 }
 
 type testGraphInstance struct {
-	graph *graphdb.ChannelGraph
+	graph *graphdb.VersionedReader
 
 	// aliasMap is a map from a node's alias to its public key. This type is
 	// provided in order to allow easily look up from the human memorable
@@ -1157,7 +1159,7 @@ func createTestCtxFromGraphInstanceAssumeValid(t *testing.T,
 
 	graphBuilder, err := NewBuilder(&Config{
 		SelfNode:            selfnode.PubKeyBytes,
-		Graph:               graphInstance.graph,
+		Graph:               graphInstance.graph.ChannelGraph,
 		Chain:               chain,
 		ChainView:           chainView,
 		Notifier:            notifier,