Merge pull request #10011 from ellemouton/graphRefactor

refactor+graph/db: refactor preparations required for incoming SQL migration code

graph/db/kv_store.go

@@ -246,7 +246,7 @@ func (c channelMapKey) String() string {
 
 // getChannelMap loads all channel edge policies from the database and stores
 // them in a map.
-func (c *KVStore) getChannelMap(edges kvdb.RBucket) (
+func getChannelMap(edges kvdb.RBucket) (
 	map[channelMapKey]*models.ChannelEdgePolicy, error) {
 
 	// Create a map to store all channel edge policies.
@@ -407,7 +407,22 @@ func (c *KVStore) AddrsForNode(ctx context.Context,
 func (c *KVStore) ForEachChannel(cb func(*models.ChannelEdgeInfo,
 	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
 
-	return c.db.View(func(tx kvdb.RTx) error {
+	return forEachChannel(c.db, cb)
+}
+
+// forEachChannel iterates through all the channel edges stored within the
+// graph and invokes the passed callback for each edge. The callback takes two
+// edges as since this is a directed graph, both the in/out edges are visited.
+// If the callback returns an error, then the transaction is aborted and the
+// iteration stops early.
+//
+// NOTE: If an edge can't be found, or wasn't advertised, then a nil pointer
+// for that particular channel edge routing policy will be passed into the
+// callback.
+func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo,
+	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
+
+	return db.View(func(tx kvdb.RTx) error {
 		edges := tx.ReadBucket(edgeBucket)
 		if edges == nil {
 			return ErrGraphNoEdgesFound
@@ -415,7 +430,7 @@ func (c *KVStore) ForEachChannel(cb func(*models.ChannelEdgeInfo,
 
 		// First, load all edges in memory indexed by node and channel
 		// id.
-		channelMap, err := c.getChannelMap(edges)
+		channelMap, err := getChannelMap(edges)
 		if err != nil {
 			return err
 		}
@@ -479,7 +494,7 @@ func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
 
 		// First, load all edges in memory indexed by node and channel
 		// id.
-		channelMap, err := c.getChannelMap(edges)
+		channelMap, err := getChannelMap(edges)
 		if err != nil {
 			return err
 		}
@@ -658,7 +673,7 @@ func (c *KVStore) ForEachNodeCached(cb func(node route.Vertex,
 	// We'll iterate over each node, then the set of channels for each
 	// node, and construct a similar callback functiopn signature as the
 	// main funcotin expects.
-	return c.forEachNode(func(tx kvdb.RTx,
+	return forEachNode(c.db, func(tx kvdb.RTx,
 		node *models.LightningNode) error {
 
 		channels := make(map[uint64]*DirectedChannel)
@@ -774,7 +789,7 @@ func (c *KVStore) DisabledChannelIDs() ([]uint64, error) {
 // executed under the same read transaction and so, methods on the NodeTx object
 // _MUST_ only be called from within the call-back.
 func (c *KVStore) ForEachNode(cb func(tx NodeRTx) error) error {
-	return c.forEachNode(func(tx kvdb.RTx,
+	return forEachNode(c.db, func(tx kvdb.RTx,
 		node *models.LightningNode) error {
 
 		return cb(newChanGraphNodeTx(tx, c, node))
@@ -788,7 +803,7 @@ func (c *KVStore) ForEachNode(cb func(tx NodeRTx) error) error {
 //
 // TODO(roasbeef): add iterator interface to allow for memory efficient graph
 // traversal when graph gets mega.
-func (c *KVStore) forEachNode(
+func forEachNode(db kvdb.Backend,
 	cb func(kvdb.RTx, *models.LightningNode) error) error {
 
 	traversal := func(tx kvdb.RTx) error {
@@ -819,7 +834,7 @@ func (c *KVStore) forEachNode(
 		})
 	}
 
-	return kvdb.View(c.db, traversal, func() {})
+	return kvdb.View(db, traversal, func() {})
 }
 
 // ForEachNodeCacheable iterates through all the stored vertices/nodes in the
@@ -866,11 +881,15 @@ func (c *KVStore) ForEachNodeCacheable(cb func(route.Vertex,
 // as the center node within a star-graph. This method may be used to kick off
 // a path finding algorithm in order to explore the reachability of another
 // node based off the source node.
-func (c *KVStore) SourceNode(_ context.Context) (*models.LightningNode,
+func (c *KVStore) SourceNode(_ context.Context) (*models.LightningNode, error) {
-	error) {
+	return sourceNode(c.db)
+}
 
+// sourceNode fetches the source node of the graph. The source node is treated
+// as the center node within a star-graph.
+func sourceNode(db kvdb.Backend) (*models.LightningNode, error) {
 	var source *models.LightningNode
-	err := kvdb.View(c.db, func(tx kvdb.RTx) error {
+	err := kvdb.View(db, func(tx kvdb.RTx) error {
 		// First grab the nodes bucket which stores the mapping from
 		// pubKey to node information.
 		nodes := tx.ReadBucket(nodeBucket)
@@ -878,7 +897,7 @@ func (c *KVStore) SourceNode(_ context.Context) (*models.LightningNode,
 			return ErrGraphNotFound
 		}
 
-		node, err := c.sourceNode(nodes)
+		node, err := sourceNodeWithTx(nodes)
 		if err != nil {
 			return err
 		}
@@ -895,13 +914,11 @@ func (c *KVStore) SourceNode(_ context.Context) (*models.LightningNode,
 	return source, nil
 }
 
-// sourceNode uses an existing database transaction and returns the source node
+// sourceNodeWithTx uses an existing database transaction and returns the source
-// of the graph. The source node is treated as the center node within a
+// node of the graph. The source node is treated as the center node within a
 // star-graph. This method may be used to kick off a path finding algorithm in
 // order to explore the reachability of another node based off the source node.
-func (c *KVStore) sourceNode(nodes kvdb.RBucket) (*models.LightningNode,
+func sourceNodeWithTx(nodes kvdb.RBucket) (*models.LightningNode, error) {
-	error) {
-
 	selfPub := nodes.Get(sourceKey)
 	if selfPub == nil {
 		return nil, ErrSourceNodeNotSet
@@ -1554,7 +1571,7 @@ func (c *KVStore) pruneGraphNodes(nodes kvdb.RwBucket,
 
 	// We'll retrieve the graph's source node to ensure we don't remove it
 	// even if it no longer has any open channels.
-	sourceNode, err := c.sourceNode(nodes)
+	sourceNode, err := sourceNodeWithTx(nodes)
 	if err != nil {
 		return nil, err
 	}
@@ -3240,7 +3257,7 @@ func (c *KVStore) ForEachSourceNodeChannel(cb func(chanPoint wire.OutPoint,
 			return ErrGraphNotFound
 		}
 
-		node, err := c.sourceNode(nodes)
+		node, err := sourceNodeWithTx(nodes)
 		if err != nil {
 			return err
 		}

graph/db/sql_store.go

@@ -570,7 +570,7 @@ func (s *SQLStore) AddChannelEdge(ctx context.Context,
 			alreadyExists = false
 		},
 		Do: func(tx SQLQueries) error {
-			err := insertChannel(ctx, tx, edge)
+			_, err := insertChannel(ctx, tx, edge)
 
 			// Silence ErrEdgeAlreadyExist so that the batch can
 			// succeed, but propagate the error via local state.
@@ -1452,8 +1452,8 @@ func (s *SQLStore) FilterChannelRange(startHeight, endHeight uint32,
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		dbChans, err := db.GetPublicV1ChannelsBySCID(
 			ctx, sqlc.GetPublicV1ChannelsBySCIDParams{
-				StartScid: chanIDStart[:],
+				StartScid: chanIDStart,
-				EndScid:   chanIDEnd[:],
+				EndScid:   chanIDEnd,
 			},
 		)
 		if err != nil {
@@ -1560,7 +1560,7 @@ func (s *SQLStore) MarkEdgeZombie(chanID uint64,
 		return db.UpsertZombieChannel(
 			ctx, sqlc.UpsertZombieChannelParams{
 				Version:  int16(ProtocolV1),
-				Scid:     chanIDB[:],
+				Scid:     chanIDB,
 				NodeKey1: pubKey1[:],
 				NodeKey2: pubKey2[:],
 			},
@@ -1592,7 +1592,7 @@ func (s *SQLStore) MarkEdgeLive(chanID uint64) error {
 	err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
 		res, err := db.DeleteZombieChannel(
 			ctx, sqlc.DeleteZombieChannelParams{
-				Scid:    chanIDB[:],
+				Scid:    chanIDB,
 				Version: int16(ProtocolV1),
 			},
 		)
@@ -1644,7 +1644,7 @@ func (s *SQLStore) IsZombieEdge(chanID uint64) (bool, [33]byte, [33]byte,
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		zombie, err := db.GetZombieChannel(
 			ctx, sqlc.GetZombieChannelParams{
-				Scid:    chanIDB[:],
+				Scid:    chanIDB,
 				Version: int16(ProtocolV1),
 			},
 		)
@@ -1723,7 +1723,7 @@ func (s *SQLStore) DeleteChannelEdges(strictZombiePruning, markZombie bool,
 
 			row, err := db.GetChannelBySCIDWithPolicies(
 				ctx, sqlc.GetChannelBySCIDWithPoliciesParams{
-					Scid:    chanIDB[:],
+					Scid:    chanIDB,
 					Version: int16(ProtocolV1),
 				},
 			)
@@ -1786,7 +1786,7 @@ func (s *SQLStore) DeleteChannelEdges(strictZombiePruning, markZombie bool,
 			err = db.UpsertZombieChannel(
 				ctx, sqlc.UpsertZombieChannelParams{
 					Version:  int16(ProtocolV1),
-					Scid:     chanIDB[:],
+					Scid:     chanIDB,
 					NodeKey1: nodeKey1[:],
 					NodeKey2: nodeKey2[:],
 				},
@@ -1833,14 +1833,12 @@ func (s *SQLStore) FetchChannelEdgesByID(chanID uint64) (
 		ctx              = context.TODO()
 		edge             *models.ChannelEdgeInfo
 		policy1, policy2 *models.ChannelEdgePolicy
+		chanIDB          = channelIDToBytes(chanID)
 	)
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
-		var chanIDB [8]byte
-		byteOrder.PutUint64(chanIDB[:], chanID)
-
 		row, err := db.GetChannelBySCIDWithPolicies(
 			ctx, sqlc.GetChannelBySCIDWithPoliciesParams{
-				Scid:    chanIDB[:],
+				Scid:    chanIDB,
 				Version: int16(ProtocolV1),
 			},
 		)
@@ -1849,7 +1847,7 @@ func (s *SQLStore) FetchChannelEdgesByID(chanID uint64) (
 			// index.
 			zombie, err := db.GetZombieChannel(
 				ctx, sqlc.GetZombieChannelParams{
-					Scid:    chanIDB[:],
+					Scid:    chanIDB,
 					Version: int16(ProtocolV1),
 				},
 			)
@@ -2033,13 +2031,11 @@ func (s *SQLStore) HasChannelEdge(chanID uint64) (time.Time, time.Time, bool,
 		return node1LastUpdate, node2LastUpdate, exists, isZombie, nil
 	}
 
+	chanIDB := channelIDToBytes(chanID)
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
-		var chanIDB [8]byte
-		byteOrder.PutUint64(chanIDB[:], chanID)
-
 		channel, err := db.GetChannelBySCID(
 			ctx, sqlc.GetChannelBySCIDParams{
-				Scid:    chanIDB[:],
+				Scid:    chanIDB,
 				Version: int16(ProtocolV1),
 			},
 		)
@@ -2047,7 +2043,7 @@ func (s *SQLStore) HasChannelEdge(chanID uint64) (time.Time, time.Time, bool,
 			// Check if it is a zombie channel.
 			isZombie, err = db.IsZombieChannel(
 				ctx, sqlc.IsZombieChannelParams{
-					Scid:    chanIDB[:],
+					Scid:    chanIDB,
 					Version: int16(ProtocolV1),
 				},
 			)
@@ -2179,15 +2175,14 @@ func (s *SQLStore) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, error) {
 	)
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		for _, chanID := range chanIDs {
-			var chanIDB [8]byte
+			chanIDB := channelIDToBytes(chanID)
-			byteOrder.PutUint64(chanIDB[:], chanID)
 
 			// TODO(elle): potentially optimize this by using
 			//  sqlc.slice() once that works for both SQLite and
 			//  Postgres.
 			row, err := db.GetChannelBySCIDWithPolicies(
 				ctx, sqlc.GetChannelBySCIDWithPoliciesParams{
-					Scid:    chanIDB[:],
+					Scid:    chanIDB,
 					Version: int16(ProtocolV1),
 				},
 			)
@@ -2270,8 +2265,7 @@ func (s *SQLStore) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo) ([]uint64,
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		for _, chanInfo := range chansInfo {
 			channelID := chanInfo.ShortChannelID.ToUint64()
-			var chanIDB [8]byte
+			chanIDB := channelIDToBytes(channelID)
-			byteOrder.PutUint64(chanIDB[:], channelID)
 
 			// TODO(elle): potentially optimize this by using
 			//  sqlc.slice() once that works for both SQLite and
@@ -2279,7 +2273,7 @@ func (s *SQLStore) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo) ([]uint64,
 			_, err := db.GetChannelBySCID(
 				ctx, sqlc.GetChannelBySCIDParams{
 					Version: int16(ProtocolV1),
-					Scid:    chanIDB[:],
+					Scid:    chanIDB,
 				},
 			)
 			if err == nil {
@@ -2291,7 +2285,7 @@ func (s *SQLStore) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo) ([]uint64,
 
 			isZombie, err := db.IsZombieChannel(
 				ctx, sqlc.IsZombieChannelParams{
-					Scid:    chanIDB[:],
+					Scid:    chanIDB,
 					Version: int16(ProtocolV1),
 				},
 			)
@@ -2609,18 +2603,16 @@ func (s *SQLStore) DisconnectBlockAtHeight(height uint32) (
 		endShortChanID = aliasmgr.StartingAlias
 
 		removedChans []*models.ChannelEdgeInfo
-	)
 
-	var chanIDStart [8]byte
+		chanIDStart = channelIDToBytes(startShortChanID.ToUint64())
-	byteOrder.PutUint64(chanIDStart[:], startShortChanID.ToUint64())
+		chanIDEnd   = channelIDToBytes(endShortChanID.ToUint64())
-	var chanIDEnd [8]byte
+	)
-	byteOrder.PutUint64(chanIDEnd[:], endShortChanID.ToUint64())
 
 	err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
 		rows, err := db.GetChannelsBySCIDRange(
 			ctx, sqlc.GetChannelsBySCIDRangeParams{
-				StartScid: chanIDStart[:],
+				StartScid: chanIDStart,
-				EndScid:   chanIDEnd[:],
+				EndScid:   chanIDEnd,
 			},
 		)
 		if err != nil {
@@ -2688,7 +2680,7 @@ func (s *SQLStore) AddEdgeProof(scid lnwire.ShortChannelID,
 	err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
 		res, err := db.AddV1ChannelProof(
 			ctx, sqlc.AddV1ChannelProofParams{
-				Scid:              scidBytes[:],
+				Scid:              scidBytes,
 				Node1Signature:    proof.NodeSig1Bytes,
 				Node2Signature:    proof.NodeSig2Bytes,
 				Bitcoin1Signature: proof.BitcoinSig1Bytes,
@@ -2734,7 +2726,7 @@ func (s *SQLStore) PutClosedScid(scid lnwire.ShortChannelID) error {
 	)
 
 	return s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
-		return db.InsertClosedChannel(ctx, chanIDB[:])
+		return db.InsertClosedChannel(ctx, chanIDB)
 	}, sqldb.NoOpReset)
 }
 
@@ -2750,7 +2742,7 @@ func (s *SQLStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) {
 	)
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		var err error
-		isClosed, err = db.IsClosedChannel(ctx, chanIDB[:])
+		isClosed, err = db.IsClosedChannel(ctx, chanIDB)
 		if err != nil {
 			return fmt.Errorf("unable to fetch closed channel: %w",
 				err)
@@ -3077,7 +3069,7 @@ func updateChanEdgePolicy(ctx context.Context, tx SQLQueries,
 	// abort the transaction which would abort the entire batch.
 	dbChan, err := tx.GetChannelAndNodesBySCID(
 		ctx, sqlc.GetChannelAndNodesBySCIDParams{
-			Scid:    chanIDB[:],
+			Scid:    chanIDB,
 			Version: int16(ProtocolV1),
 		},
 	)
@@ -3699,7 +3691,7 @@ type srcNodeInfo struct {
 	pub route.Vertex
 }
 
-// getSourceNode returns the DB node ID and pub key of the source node for the
+// sourceNode returns the DB node ID and pub key of the source node for the
 // specified protocol version.
 func (s *SQLStore) getSourceNode(ctx context.Context, db SQLQueries,
 	version ProtocolVersion) (int64, route.Vertex, error) {
@@ -3767,9 +3759,17 @@ func marshalExtraOpaqueData(data []byte) (map[uint64][]byte, error) {
 	return records, nil
 }
 
+// dbChanInfo holds the DB level IDs of a channel and the nodes involved in the
+// channel.
+type dbChanInfo struct {
+	channelID int64
+	node1ID   int64
+	node2ID   int64
+}
+
 // insertChannel inserts a new channel record into the database.
 func insertChannel(ctx context.Context, db SQLQueries,
-	edge *models.ChannelEdgeInfo) error {
+	edge *models.ChannelEdgeInfo) (*dbChanInfo, error) {
 
 	chanIDB := channelIDToBytes(edge.ChannelID)
 
@@ -3779,26 +3779,26 @@ func insertChannel(ctx context.Context, db SQLQueries,
 	// batch of transactions.
 	_, err := db.GetChannelBySCID(
 		ctx, sqlc.GetChannelBySCIDParams{
-			Scid:    chanIDB[:],
+			Scid:    chanIDB,
 			Version: int16(ProtocolV1),
 		},
 	)
 	if err == nil {
-		return ErrEdgeAlreadyExist
+		return nil, ErrEdgeAlreadyExist
 	} else if !errors.Is(err, sql.ErrNoRows) {
-		return fmt.Errorf("unable to fetch channel: %w", err)
+		return nil, fmt.Errorf("unable to fetch channel: %w", err)
 	}
 
 	// Make sure that at least a "shell" entry for each node is present in
 	// the nodes table.
 	node1DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey1Bytes)
 	if err != nil {
-		return fmt.Errorf("unable to create shell node: %w", err)
+		return nil, fmt.Errorf("unable to create shell node: %w", err)
 	}
 
 	node2DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey2Bytes)
 	if err != nil {
-		return fmt.Errorf("unable to create shell node: %w", err)
+		return nil, fmt.Errorf("unable to create shell node: %w", err)
 	}
 
 	var capacity sql.NullInt64
@@ -3808,7 +3808,7 @@ func insertChannel(ctx context.Context, db SQLQueries,
 
 	createParams := sqlc.CreateChannelParams{
 		Version:     int16(ProtocolV1),
-		Scid:        chanIDB[:],
+		Scid:        chanIDB,
 		NodeID1:     node1DBID,
 		NodeID2:     node2DBID,
 		Outpoint:    edge.ChannelPoint.String(),
@@ -3829,7 +3829,7 @@ func insertChannel(ctx context.Context, db SQLQueries,
 	// Insert the new channel record.
 	dbChanID, err := db.CreateChannel(ctx, createParams)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// Insert any channel features.
@@ -3837,7 +3837,7 @@ func insertChannel(ctx context.Context, db SQLQueries,
 		chanFeatures := lnwire.NewRawFeatureVector()
 		err := chanFeatures.Decode(bytes.NewReader(edge.Features))
 		if err != nil {
-			return err
+			return nil, err
 		}
 
 		fv := lnwire.NewFeatureVector(chanFeatures, lnwire.Features)
@@ -3849,7 +3849,7 @@ func insertChannel(ctx context.Context, db SQLQueries,
 				},
 			)
 			if err != nil {
-				return fmt.Errorf("unable to insert "+
+				return nil, fmt.Errorf("unable to insert "+
 					"channel(%d) feature(%v): %w", dbChanID,
 					feature, err)
 			}
@@ -3859,8 +3859,8 @@ func insertChannel(ctx context.Context, db SQLQueries,
 	// Finally, insert any extra TLV fields in the channel announcement.
 	extra, err := marshalExtraOpaqueData(edge.ExtraOpaqueData)
 	if err != nil {
-		return fmt.Errorf("unable to marshal extra opaque data: %w",
+		return nil, fmt.Errorf("unable to marshal extra opaque "+
-			err)
+			"data: %w", err)
 	}
 
 	for tlvType, value := range extra {
@@ -3872,13 +3872,17 @@ func insertChannel(ctx context.Context, db SQLQueries,
 			},
 		)
 		if err != nil {
-			return fmt.Errorf("unable to upsert channel(%d) extra "+
+			return nil, fmt.Errorf("unable to upsert "+
-				"signed field(%v): %w", edge.ChannelID,
+				"channel(%d) extra signed field(%v): %w",
-				tlvType, err)
+				edge.ChannelID, tlvType, err)
 		}
 	}
 
-	return nil
+	return &dbChanInfo{
+		channelID: dbChanID,
+		node1ID:   node1DBID,
+		node2ID:   node2DBID,
+	}, nil
 }
 
 // maybeCreateShellNode checks if a shell node entry exists for the
@@ -4455,9 +4459,9 @@ func extractChannelPolicies(row any) (*sqlc.ChannelPolicy, *sqlc.ChannelPolicy,
 
 // channelIDToBytes converts a channel ID (SCID) to a byte array
 // representation.
-func channelIDToBytes(channelID uint64) [8]byte {
+func channelIDToBytes(channelID uint64) []byte {
 	var chanIDB [8]byte
 	byteOrder.PutUint64(chanIDB[:], channelID)
 
-	return chanIDB
+	return chanIDB[:]
 }