Merge pull request #9971 from ellemouton/graphSQL16-closed-scids

[16] graph/db: SQL closed SCIDs table and last few methods
This commit is contained in:
Oliver Gugger authored and GitHub committed 2025-06-25 17:32:49 +02:00
commit fb1fef9e6c
11 files changed
+385 -20

No files matched your search

@@ -90,6 +90,7 @@ circuit. The indices are only available for forwarding events saved after v0.20.
* [7](https://github.com/lightningnetwork/lnd/pull/9937)
* [8](https://github.com/lightningnetwork/lnd/pull/9938)
* [9](https://github.com/lightningnetwork/lnd/pull/9939)
* [10](https://github.com/lightningnetwork/lnd/pull/9971)
## RPC Updates
+2 -2
View File
@@ -1128,7 +1128,7 @@ func TestAddEdgeProof(t *testing.T) {
t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t)
graph := MakeTestGraphNew(t)
// Add an edge with no proof.
node1 := createTestVertex(t)
@@ -4325,7 +4325,7 @@ func TestGraphLoading(t *testing.T) {
func TestClosedScid(t *testing.T) {
t.Parallel()
graph := MakeTestGraph(t)
graph := MakeTestGraphNew(t)
scid := lnwire.ShortChannelID{}
+2 -2
View File
@@ -2150,8 +2150,8 @@ func (c *KVStore) ChanUpdatesInHorizon(startTime,
}
if len(edgesInHorizon) > 0 {
log.Debugf("ChanUpdatesInHorizon hit percentage: %f (%d/%d)",
float64(hits)/float64(len(edgesInHorizon)), hits,
log.Debugf("ChanUpdatesInHorizon hit percentage: %.2f (%d/%d)",
float64(hits)*100/float64(len(edgesInHorizon)), hits,
len(edgesInHorizon))
} else {
log.Debugf("ChanUpdatesInHorizon returned no edges in "+
+257 -4
View File
@@ -91,6 +91,7 @@ type SQLQueries interface {
Channel queries.
*/
CreateChannel(ctx context.Context, arg sqlc.CreateChannelParams) (int64, error)
AddV1ChannelProof(ctx context.Context, arg sqlc.AddV1ChannelProofParams) (sql.Result, error)
GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.Channel, error)
GetChannelByOutpoint(ctx context.Context, outpoint string) (sqlc.GetChannelByOutpointRow, error)
GetChannelsBySCIDRange(ctx context.Context, arg sqlc.GetChannelsBySCIDRangeParams) ([]sqlc.GetChannelsBySCIDRangeRow, error)
@@ -136,6 +137,12 @@ type SQLQueries interface {
GetPruneTip(ctx context.Context) (sqlc.PruneLog, error)
UpsertPruneLogEntry(ctx context.Context, arg sqlc.UpsertPruneLogEntryParams) error
DeletePruneLogEntriesInRange(ctx context.Context, arg sqlc.DeletePruneLogEntriesInRangeParams) error
/*
Closed SCID table queries.
*/
InsertClosedChannel(ctx context.Context, scid []byte) error
IsClosedChannel(ctx context.Context, scid []byte) (bool, error)
}
// BatchedSQLQueries is a version of SQLQueries that's capable of batched
@@ -1096,8 +1103,8 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime,
}
if len(edges) > 0 {
log.Debugf("ChanUpdatesInHorizon hit percentage: %f (%d/%d)",
float64(hits)/float64(len(edges)), hits, len(edges))
log.Debugf("ChanUpdatesInHorizon hit percentage: %.2f (%d/%d)",
float64(hits)*100/float64(len(edges)), hits, len(edges))
} else {
log.Debugf("ChanUpdatesInHorizon returned no edges in "+
"horizon (%s, %s)", startTime, endTime)
@@ -1231,6 +1238,103 @@ func (s *SQLStore) ForEachNodeCached(cb func(node route.Vertex,
}, sqldb.NoOpReset)
}
// ForEachChannelCacheable 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.
//
// NOTE: this method is like ForEachChannel but fetches only the data
// required for the graph cache.
func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
*models.CachedEdgePolicy,
*models.CachedEdgePolicy) error) error {
ctx := context.TODO()
handleChannel := func(db SQLQueries,
row sqlc.ListChannelsWithPoliciesPaginatedRow) error {
node1, node2, err := buildNodeVertices(
row.Node1Pubkey, row.Node2Pubkey,
)
if err != nil {
return err
}
edge := buildCacheableChannelInfo(row.Channel, node1, node2)
dbPol1, dbPol2, err := extractChannelPolicies(row)
if err != nil {
return err
}
var pol1, pol2 *models.CachedEdgePolicy
if dbPol1 != nil {
policy1, err := buildChanPolicy(
*dbPol1, edge.ChannelID, nil, node2, true,
)
if err != nil {
return err
}
pol1 = models.NewCachedPolicy(policy1)
}
if dbPol2 != nil {
policy2, err := buildChanPolicy(
*dbPol2, edge.ChannelID, nil, node1, false,
)
if err != nil {
return err
}
pol2 = models.NewCachedPolicy(policy2)
}
if err := cb(edge, pol1, pol2); err != nil {
return err
}
return nil
}
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
lastID := int64(-1)
for {
//nolint:ll
rows, err := db.ListChannelsWithPoliciesPaginated(
ctx, sqlc.ListChannelsWithPoliciesPaginatedParams{
Version: int16(ProtocolV1),
ID: lastID,
Limit: pageSize,
},
)
if err != nil {
return err
}
if len(rows) == 0 {
break
}
for _, row := range rows {
err := handleChannel(db, row)
if err != nil {
return err
}
lastID = row.Channel.ID
}
}
return nil
}, sqldb.NoOpReset)
}
// 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.
@@ -1291,7 +1395,7 @@ func (s *SQLStore) ForEachChannel(cb func(*models.ChannelEdgeInfo,
}
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
var lastID int64
lastID := int64(-1)
for {
//nolint:ll
rows, err := db.ListChannelsWithPoliciesPaginated(
@@ -2575,6 +2679,155 @@ func (s *SQLStore) DisconnectBlockAtHeight(height uint32) (
return removedChans, nil
}
// AddEdgeProof sets the proof of an existing edge in the graph database.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) AddEdgeProof(scid lnwire.ShortChannelID,
proof *models.ChannelAuthProof) error {
var (
ctx = context.TODO()
scidBytes = channelIDToBytes(scid.ToUint64())
)
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
res, err := db.AddV1ChannelProof(
ctx, sqlc.AddV1ChannelProofParams{
Scid: scidBytes[:],
Node1Signature: proof.NodeSig1Bytes,
Node2Signature: proof.NodeSig2Bytes,
Bitcoin1Signature: proof.BitcoinSig1Bytes,
Bitcoin2Signature: proof.BitcoinSig2Bytes,
},
)
if err != nil {
return fmt.Errorf("unable to add edge proof: %w", err)
}
n, err := res.RowsAffected()
if err != nil {
return err
}
if n == 0 {
return fmt.Errorf("no rows affected when adding edge "+
"proof for SCID %v", scid)
} else if n > 1 {
return fmt.Errorf("multiple rows affected when adding "+
"edge proof for SCID %v: %d rows affected",
scid, n)
}
return nil
}, sqldb.NoOpReset)
if err != nil {
return fmt.Errorf("unable to add edge proof: %w", err)
}
return nil
}
// PutClosedScid stores a SCID for a closed channel in the database. This is so
// that we can ignore channel announcements that we know to be closed without
// having to validate them and fetch a block.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) PutClosedScid(scid lnwire.ShortChannelID) error {
var (
ctx = context.TODO()
chanIDB = channelIDToBytes(scid.ToUint64())
)
return s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
return db.InsertClosedChannel(ctx, chanIDB[:])
}, sqldb.NoOpReset)
}
// IsClosedScid checks whether a channel identified by the passed in scid is
// closed. This helps avoid having to perform expensive validation checks.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) {
var (
ctx = context.TODO()
isClosed bool
chanIDB = channelIDToBytes(scid.ToUint64())
)
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
var err error
isClosed, err = db.IsClosedChannel(ctx, chanIDB[:])
if err != nil {
return fmt.Errorf("unable to fetch closed channel: %w",
err)
}
return nil
}, sqldb.NoOpReset)
if err != nil {
return false, fmt.Errorf("unable to fetch closed channel: %w",
err)
}
return isClosed, nil
}
// GraphSession will provide the call-back with access to a NodeTraverser
// instance which can be used to perform queries against the channel graph.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) GraphSession(cb func(graph NodeTraverser) error) error {
var ctx = context.TODO()
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
return cb(newSQLNodeTraverser(db, s.cfg.ChainHash))
}, sqldb.NoOpReset)
}
// sqlNodeTraverser implements the NodeTraverser interface but with a backing
// read only transaction for a consistent view of the graph.
type sqlNodeTraverser struct {
db SQLQueries
chain chainhash.Hash
}
// A compile-time assertion to ensure that sqlNodeTraverser implements the
// NodeTraverser interface.
var _ NodeTraverser = (*sqlNodeTraverser)(nil)
// newSQLNodeTraverser creates a new instance of the sqlNodeTraverser.
func newSQLNodeTraverser(db SQLQueries,
chain chainhash.Hash) *sqlNodeTraverser {
return &sqlNodeTraverser{
db: db,
chain: chain,
}
}
// ForEachNodeDirectedChannel calls the callback for every channel of the given
// node.
//
// NOTE: Part of the NodeTraverser interface.
func (s *sqlNodeTraverser) ForEachNodeDirectedChannel(nodePub route.Vertex,
cb func(channel *DirectedChannel) error) error {
ctx := context.TODO()
return forEachNodeDirectedChannel(ctx, s.db, nodePub, cb)
}
// FetchNodeFeatures returns the features of the given node. If the node is
// unknown, assume no additional features are supported.
//
// NOTE: Part of the NodeTraverser interface.
func (s *sqlNodeTraverser) FetchNodeFeatures(nodePub route.Vertex) (
*lnwire.FeatureVector, error) {
ctx := context.TODO()
return fetchNodeFeatures(ctx, s.db, nodePub)
}
// forEachNodeDirectedChannel iterates through all channels of a given
// node, executing the passed callback on the directed edge representing the
// channel and its incoming policy. If the node is not found, no error is
@@ -2704,7 +2957,7 @@ func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries,
func forEachNodeCacheable(ctx context.Context, db SQLQueries,
cb func(nodeID int64, nodePub route.Vertex) error) error {
var lastID int64
lastID := int64(-1)
for {
nodes, err := db.ListNodeIDsAndPubKeys(
+25 -12
View File
@@ -160,7 +160,9 @@ func makeTestGraph(t *testing.T, useCache bool) (*graphdb.ChannelGraph,
kvdb.Backend, error) {
// Create channelgraph for the first time.
graph := graphdb.MakeTestGraph(t, graphdb.WithUseGraphCache(useCache))
graph := graphdb.MakeTestGraphNew(
t, graphdb.WithUseGraphCache(useCache),
)
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
@@ -289,6 +291,11 @@ func parseTestGraph(t *testing.T, useCache bool, path string) (
}
source = dbNode
// If this is the source node, we don't have to call
// AddLightningNode below since we will call
// SetSourceNode later.
continue
}
// With the node fully parsed, add it as a vertex within the
@@ -541,8 +548,8 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
channelIDs := make(map[route.Vertex]map[route.Vertex]uint64)
nodeIndex := byte(0)
addNodeWithAlias := func(alias string, features *lnwire.FeatureVector) (
*models.LightningNode, error) {
addNodeWithAlias := func(alias string,
features *lnwire.FeatureVector) error {
keyBytes := []byte{
0, 0, 0, 0, 0, 0, 0, 0,
@@ -572,18 +579,22 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
// With the node fully parsed, add it as a vertex within the
// graph.
if err := graph.AddLightningNode(ctx, dbNode); err != nil {
return nil, err
if alias == source {
err = graph.SetSourceNode(ctx, dbNode)
require.NoError(t, err)
} else {
err := graph.AddLightningNode(ctx, dbNode)
require.NoError(t, err)
}
aliasMap[alias] = dbNode.PubKeyBytes
nodeIndex++
return dbNode, nil
return nil
}
// Add the source node.
dbNode, err := addNodeWithAlias(
err = addNodeWithAlias(
source, lnwire.NewFeatureVector(
lnwire.NewRawFeatureVector(sourceFeatureBits...),
lnwire.Features,
@@ -591,10 +602,6 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
)
require.NoError(t, err)
if err = graph.SetSourceNode(ctx, dbNode); err != nil {
return nil, err
}
// Initialize variable that keeps track of the next channel id to assign
// if none is specified.
nextUnassignedChannelID := uint64(100000)
@@ -612,7 +619,7 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
features =
node.testChannelPolicy.Features
}
_, err := addNodeWithAlias(
err := addNodeWithAlias(
node.Alias, features,
)
if err != nil {
@@ -2169,6 +2176,7 @@ func runRouteFailMaxHTLC(t *testing.T, useCache bool) {
require.NoError(t, err, "unable to fetch channel edges by ID")
midEdge.MessageFlags = 1
midEdge.MaxHTLC = payAmt - 1
midEdge.LastUpdate = midEdge.LastUpdate.Add(time.Second)
err = graph.UpdateEdgePolicy(context.Background(), midEdge)
require.NoError(t, err)
@@ -2211,10 +2219,12 @@ func runRouteFailDisabledEdge(t *testing.T, useCache bool) {
_, e1, e2, err := graph.graph.FetchChannelEdgesByID(roasToPham)
require.NoError(t, err, "unable to fetch edge")
e1.ChannelFlags |= lnwire.ChanUpdateDisabled
e1.LastUpdate = e1.LastUpdate.Add(time.Second)
if err := graph.graph.UpdateEdgePolicy(ctx, e1); err != nil {
t.Fatalf("unable to update edge: %v", err)
}
e2.ChannelFlags |= lnwire.ChanUpdateDisabled
e2.LastUpdate = e2.LastUpdate.Add(time.Second)
if err := graph.graph.UpdateEdgePolicy(ctx, e2); err != nil {
t.Fatalf("unable to update edge: %v", err)
}
@@ -2232,6 +2242,7 @@ func runRouteFailDisabledEdge(t *testing.T, useCache bool) {
_, e, _, err := graph.graph.FetchChannelEdgesByID(phamToSophon)
require.NoError(t, err, "unable to fetch edge")
e.ChannelFlags |= lnwire.ChanUpdateDisabled
e.LastUpdate = e.LastUpdate.Add(time.Second)
if err := graph.graph.UpdateEdgePolicy(ctx, e); err != nil {
t.Fatalf("unable to update edge: %v", err)
}
@@ -2314,10 +2325,12 @@ func runPathSourceEdgesBandwidth(t *testing.T, useCache bool) {
_, e1, e2, err := graph.graph.FetchChannelEdgesByID(roasToSongoku)
require.NoError(t, err, "unable to fetch edge")
e1.ChannelFlags |= lnwire.ChanUpdateDisabled
e1.LastUpdate = e1.LastUpdate.Add(time.Second)
if err := graph.graph.UpdateEdgePolicy(ctx, e1); err != nil {
t.Fatalf("unable to update edge: %v", err)
}
e2.ChannelFlags |= lnwire.ChanUpdateDisabled
e2.LastUpdate = e2.LastUpdate.Add(time.Second)
if err := graph.graph.UpdateEdgePolicy(ctx, e2); err != nil {
t.Fatalf("unable to update edge: %v", err)
}
+59
View File
@@ -26,6 +26,34 @@ func (q *Queries) AddSourceNode(ctx context.Context, nodeID int64) error {
return err
}
const addV1ChannelProof = `-- name: AddV1ChannelProof :execresult
UPDATE channels
SET node_1_signature = $2,
node_2_signature = $3,
bitcoin_1_signature = $4,
bitcoin_2_signature = $5
WHERE scid = $1
AND version = 1
`
type AddV1ChannelProofParams struct {
Scid []byte
Node1Signature []byte
Node2Signature []byte
Bitcoin1Signature []byte
Bitcoin2Signature []byte
}
func (q *Queries) AddV1ChannelProof(ctx context.Context, arg AddV1ChannelProofParams) (sql.Result, error) {
return q.db.ExecContext(ctx, addV1ChannelProof,
arg.Scid,
arg.Node1Signature,
arg.Node2Signature,
arg.Bitcoin1Signature,
arg.Bitcoin2Signature,
)
}
const countZombieChannels = `-- name: CountZombieChannels :one
SELECT COUNT(*)
FROM zombie_channels
@@ -1527,6 +1555,22 @@ func (q *Queries) InsertChannelFeature(ctx context.Context, arg InsertChannelFea
return err
}
const insertClosedChannel = `-- name: InsertClosedChannel :exec
/* ─────────────────────────────────────────────
closed_scid table queries
────────────────────────────────────────────-
*/
INSERT INTO closed_scids (scid)
VALUES ($1)
ON CONFLICT (scid) DO NOTHING
`
func (q *Queries) InsertClosedChannel(ctx context.Context, scid []byte) error {
_, err := q.db.ExecContext(ctx, insertClosedChannel, scid)
return err
}
const insertNodeAddress = `-- name: InsertNodeAddress :exec
/* ─────────────────────────────────────────────
node_addresses table queries
@@ -1583,6 +1627,21 @@ func (q *Queries) InsertNodeFeature(ctx context.Context, arg InsertNodeFeaturePa
return err
}
const isClosedChannel = `-- name: IsClosedChannel :one
SELECT EXISTS (
SELECT 1
FROM closed_scids
WHERE scid = $1
)
`
func (q *Queries) IsClosedChannel(ctx context.Context, scid []byte) (bool, error) {
row := q.db.QueryRowContext(ctx, isClosedChannel, scid)
var exists bool
err := row.Scan(&exists)
return exists, err
}
const isPublicV1Node = `-- name: IsPublicV1Node :one
SELECT EXISTS (
SELECT 1
@@ -29,3 +29,4 @@ DROP TABLE IF EXISTS nodes;
DROP TABLE IF EXISTS channel_policy_extra_types;
DROP TABLE IF EXISTS zombie_channels;
DROP TABLE IF EXISTS prune_log;
DROP TABLE IF EXISTS closed_scids;
@@ -331,4 +331,9 @@ CREATE TABLE IF NOT EXISTS prune_log (
-- The block hash that the prune was performed at.
block_hash BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS closed_scids (
-- The short channel id of the channel.
scid BLOB PRIMARY KEY
);
+4
View File
@@ -78,6 +78,10 @@ type ChannelPolicyExtraType struct {
Value []byte
}
type ClosedScid struct {
Scid []byte
}
type Invoice struct {
ID int64
Hash []byte
+3
View File
@@ -12,6 +12,7 @@ import (
type Querier interface {
AddSourceNode(ctx context.Context, nodeID int64) error
AddV1ChannelProof(ctx context.Context, arg AddV1ChannelProofParams) (sql.Result, error)
ClearKVInvoiceHashIndex(ctx context.Context) error
CountZombieChannels(ctx context.Context, version int16) (int64, error)
CreateChannel(ctx context.Context, arg CreateChannelParams) (int64, error)
@@ -79,6 +80,7 @@ type Querier interface {
InsertAMPSubInvoiceHTLC(ctx context.Context, arg InsertAMPSubInvoiceHTLCParams) error
InsertChanPolicyExtraType(ctx context.Context, arg InsertChanPolicyExtraTypeParams) error
InsertChannelFeature(ctx context.Context, arg InsertChannelFeatureParams) error
InsertClosedChannel(ctx context.Context, scid []byte) error
InsertInvoice(ctx context.Context, arg InsertInvoiceParams) (int64, error)
InsertInvoiceFeature(ctx context.Context, arg InsertInvoiceFeatureParams) error
InsertInvoiceHTLC(ctx context.Context, arg InsertInvoiceHTLCParams) (int64, error)
@@ -87,6 +89,7 @@ type Querier interface {
InsertMigratedInvoice(ctx context.Context, arg InsertMigratedInvoiceParams) (int64, error)
InsertNodeAddress(ctx context.Context, arg InsertNodeAddressParams) error
InsertNodeFeature(ctx context.Context, arg InsertNodeFeatureParams) error
IsClosedChannel(ctx context.Context, scid []byte) (bool, error)
IsPublicV1Node(ctx context.Context, pubKey []byte) (bool, error)
IsZombieChannel(ctx context.Context, arg IsZombieChannelParams) (bool, error)
ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNodeIDParams) ([]ListChannelsByNodeIDRow, error)
+26
View File
@@ -208,6 +208,15 @@ INSERT INTO channels (
)
RETURNING id;
-- name: AddV1ChannelProof :execresult
UPDATE channels
SET node_1_signature = $2,
node_2_signature = $3,
bitcoin_1_signature = $4,
bitcoin_2_signature = $5
WHERE scid = $1
AND version = 1;
-- name: GetChannelsBySCIDRange :many
SELECT sqlc.embed(c),
n1.pub_key AS node1_pub_key,
@@ -700,3 +709,20 @@ LIMIT 1;
DELETE FROM prune_log
WHERE block_height >= @start_height
AND block_height <= @end_height;
/* ─────────────────────────────────────────────
closed_scid table queries
────────────────────────────────────────────-
*/
-- name: InsertClosedChannel :exec
INSERT INTO closed_scids (scid)
VALUES ($1)
ON CONFLICT (scid) DO NOTHING;
-- name: IsClosedChannel :one
SELECT EXISTS (
SELECT 1
FROM closed_scids
WHERE scid = $1
);