diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 87651e461..eef4f7ec5 100644 --- a/graph/db/kv_store.go +++ b/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, - error) { +func (c *KVStore) SourceNode(_ context.Context) (*models.LightningNode, 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 -// of the graph. The source node is treated as the center node within a +// sourceNodeWithTx uses an existing database transaction and returns the source +// 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, - error) { - +func sourceNodeWithTx(nodes kvdb.RBucket) (*models.LightningNode, 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 } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 067a044d3..a9de05ae1 100644 --- a/graph/db/sql_store.go +++ b/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[:], - EndScid: chanIDEnd[:], + StartScid: chanIDStart, + 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 - byteOrder.PutUint64(chanIDB[:], chanID) + chanIDB := channelIDToBytes(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 - byteOrder.PutUint64(chanIDB[:], channelID) + chanIDB := channelIDToBytes(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 - byteOrder.PutUint64(chanIDStart[:], startShortChanID.ToUint64()) - var chanIDEnd [8]byte - byteOrder.PutUint64(chanIDEnd[:], endShortChanID.ToUint64()) + chanIDStart = channelIDToBytes(startShortChanID.ToUint64()) + chanIDEnd = channelIDToBytes(endShortChanID.ToUint64()) + ) err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { rows, err := db.GetChannelsBySCIDRange( ctx, sqlc.GetChannelsBySCIDRangeParams{ - StartScid: chanIDStart[:], - EndScid: chanIDEnd[:], + StartScid: chanIDStart, + 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", - err) + return nil, fmt.Errorf("unable to marshal extra opaque "+ + "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 "+ - "signed field(%v): %w", edge.ChannelID, - tlvType, err) + return nil, fmt.Errorf("unable to upsert "+ + "channel(%d) extra signed field(%v): %w", + 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[:] }