From e5d099ef1470249c2e98f5b5e0deb5537ba986ac Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sat, 24 May 2025 14:20:28 +0200 Subject: [PATCH 1/5] sqldb/sqlc: add graph channel schemas In this commit, we define the SQL schemas for storing graph channel data. This includes a new `channels` table and a new `channel_features` table along with various indices. --- sqldb/sqlc/migrations/000007_graph.down.sql | 11 +- sqldb/sqlc/migrations/000007_graph.up.sql | 115 +++++++++++++++++++- sqldb/sqlc/models.go | 27 +++++ 3 files changed, 151 insertions(+), 2 deletions(-) diff --git a/sqldb/sqlc/migrations/000007_graph.down.sql b/sqldb/sqlc/migrations/000007_graph.down.sql index 79489d7bd..667a6099d 100644 --- a/sqldb/sqlc/migrations/000007_graph.down.sql +++ b/sqldb/sqlc/migrations/000007_graph.down.sql @@ -4,10 +4,19 @@ DROP INDEX IF EXISTS node_extra_types_unique; DROP INDEX IF EXISTS node_features_unique; DROP INDEX IF EXISTS node_addresses_unique; DROP INDEX IF EXISTS source_nodes_unique; +DROP INDEX IF EXISTS channels_node_id_1_idx; +DROP INDEX IF EXISTS channels_node_id_2_idx; +DROP INDEX IF EXISTS channels_unique; +DROP INDEX IF EXISTS channels_version_outpoint_idx; +DROP INDEX IF EXISTS channel_features_unique; +DROP INDEX IF EXISTS channel_extra_types_unique; -- Drop tables in order of reverse dependencies. DROP TABLE IF EXISTS source_nodes; DROP TABLE IF EXISTS node_addresses; DROP TABLE IF EXISTS node_features; DROP TABLE IF EXISTS node_extra_types; -DROP TABLE IF EXISTS nodes; \ No newline at end of file +DROP TABLE IF EXISTS nodes; +DROP TABLE IF EXISTS channels; +DROP TABLE IF EXISTS channel_features; +DROP TABLE IF EXISTS channel_extra_types; \ No newline at end of file diff --git a/sqldb/sqlc/migrations/000007_graph.up.sql b/sqldb/sqlc/migrations/000007_graph.up.sql index 0efc750e3..9254a5abf 100644 --- a/sqldb/sqlc/migrations/000007_graph.up.sql +++ b/sqldb/sqlc/migrations/000007_graph.up.sql @@ -94,4 +94,117 @@ CREATE TABLE IF NOT EXISTS source_nodes ( ); CREATE UNIQUE INDEX IF NOT EXISTS source_nodes_unique ON source_nodes ( node_id -); \ No newline at end of file +); + +/* ───────────────────────────────────────────── + channel data tables + ───────────────────────────────────────────── +*/ + +-- channels stores all teh channels that we are aware of in the graph. +CREATE TABLE IF NOT EXISTS channels ( + -- The db ID of the channel. + id INTEGER PRIMARY KEY, + + -- The protocol version that this node was gossiped on. + version SMALLINT NOT NULL, + + -- The channel id (short channel id) of the channel. + scid BLOB NOT NULL, + + -- A reference to a node in the nodes table for the node_1 node in + -- the channel announcement. + node_id_1 BIGINT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE, + + -- A reference to a node in the nodes table for the node_2 node in + -- the channel announcement. + node_id_2 BIGINT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE, + + -- The outpoint of the funding transaction. We chose to store this + -- in a string format for the sake of readability and user queries on + -- outpoint. + outpoint TEXT NOT NULL, + + -- The capacity of the channel in millisatoshis. This is nullable since + -- (for the v1 protocol at least), the capacity is not necessarily known as + -- it is not gossiped in the channel announcement. + capacity BIGINT, + + -- bitcoin_key_1 is the public key owned by node_1 that is used to create + -- a funding transaction. It is nullable since future protocol version may + -- not make use of this field. + bitcoin_key_1 BLOB, + + -- bitcoin_key_2 is the public key owned by node_2 that is used to create + -- the funding transaction. It is nullable since future protocol version may + -- not make use of this field. + bitcoin_key_2 BLOB, + + -- node_1_signature is the signature of the serialised channel announcement + -- using the node_1 public key. It is nullable since future protocol + -- versions may not make use of this field _and_ because for this node's own + -- channels, the signature will only be known after the initial record + -- creation. + node_1_signature BLOB, + + -- node_2_signature is the signature of the serialised channel announcement + -- using the node_2 public key. It is nullable since future protocol + -- versions may not make use of this field _and_ because for this node's own + -- channels, the signature will only be known after the initial record + -- creation. + node_2_signature BLOB, + + -- bitcoin_1_signature is the signature of the serialised channel + -- announcement using bitcion_key_1. It is nullable since future protocol + -- versions may not make use of this field _and_ because for this node's + -- own channels, the signature will only be known after the initial record + -- creation. + bitcoin_1_signature BLOB, + + -- bitcoin_2_signature is the signature of the serialised channel + -- announcement using bitcion_key_2. It is nullable since future protocol + -- versions may not make use of this field _and_ because for this node's + -- own channels, the signature will only be known after the initial record + -- creation. + bitcoin_2_signature BLOB +); +-- We'll want to lookup all the channels owned by a node, so we create +-- indexes on the node_id_1 and node_id_2 columns. +CREATE INDEX IF NOT EXISTS channels_node_id_1_idx ON channels(node_id_1); +CREATE INDEX IF NOT EXISTS channels_node_id_2_idx ON channels(node_id_2); + +-- A channel (identified by a short channel id) can only have one active +-- channel announcement per protocol version. We also order the index by +-- scid in descending order so that we have an idea of the latest channel +-- announcement we know of. +CREATE UNIQUE INDEX IF NOT EXISTS channels_unique ON channels(version, scid DESC); +CREATE INDEX IF NOT EXISTS channels_version_outpoint_idx ON channels(version, outpoint); + +-- channel_features contains the feature bits of a channel. +CREATE TABLE IF NOT EXISTS channel_features ( + -- The channel id this feature belongs to. + channel_id BIGINT NOT NULL REFERENCES channels(id) ON DELETE CASCADE, + + -- The feature bit value. + feature_bit INTEGER NOT NULL +); +CREATE UNIQUE INDEX IF NOT EXISTS channel_features_unique ON channel_features ( + channel_id, feature_bit +); + +-- channel_extra_types stores any extra TLV fields covered by a channels +-- announcement that we do not have an explicit column for in the channels +-- table. +CREATE TABLE IF NOT EXISTS channel_extra_types ( + -- The channel id this TLV field belongs to. + channel_id BIGINT NOT NULL REFERENCES channels(id) ON DELETE CASCADE, + + -- The Type field. + type BIGINT NOT NULL, + + -- The value field. + value BLOB +); +CREATE UNIQUE INDEX IF NOT EXISTS channel_extra_types_unique ON channel_extra_types ( + type, channel_id +); diff --git a/sqldb/sqlc/models.go b/sqldb/sqlc/models.go index 07269c359..3e8421336 100644 --- a/sqldb/sqlc/models.go +++ b/sqldb/sqlc/models.go @@ -28,6 +28,33 @@ type AmpSubInvoiceHtlc struct { Preimage []byte } +type Channel struct { + ID int64 + Version int16 + Scid []byte + NodeID1 int64 + NodeID2 int64 + Outpoint string + Capacity sql.NullInt64 + BitcoinKey1 []byte + BitcoinKey2 []byte + Node1Signature []byte + Node2Signature []byte + Bitcoin1Signature []byte + Bitcoin2Signature []byte +} + +type ChannelExtraType struct { + ChannelID int64 + Type int64 + Value []byte +} + +type ChannelFeature struct { + ChannelID int64 + FeatureBit int32 +} + type Invoice struct { ID int64 Hash []byte From 2a36e17f6f690060404a237514d885fb0297c9be Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 27 May 2025 08:21:11 +0200 Subject: [PATCH 2/5] graph/db: only init extra byte if not nil In preparation for having consistency with the structs created by the SQLStore and the KVStore (so that they have the same behaviour when tested by the unit tests), here we make sure not to init the ExtraOpaqueData field of the LightningNode struct unless there are actualy bytes to set. --- graph/db/graph_test.go | 1 - graph/db/kv_store.go | 6 +++++- graph/db/sql_store.go | 7 +++---- 3 files changed, 8 insertions(+), 6 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index aae7e5a7e..e644fea19 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -82,7 +82,6 @@ func createLightningNode(priv *btcec.PrivateKey) *models.LightningNode { Alias: "kek" + hex.EncodeToString(pub), Features: testFeatures, Addresses: testAddrs, - ExtraOpaqueData: make([]byte, 0), } copy(n.PubKeyBytes[:], priv.PubKey().SerializeCompressed()) diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 187fc13af..b08c86022 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -4122,7 +4122,7 @@ func deserializeLightningNode(r io.Reader) (models.LightningNode, error) { // We'll try and see if there are any opaque bytes left, if not, then // we'll ignore the EOF error and return the node as is. - node.ExtraOpaqueData, err = wire.ReadVarBytes( + extraBytes, err := wire.ReadVarBytes( r, 0, MaxAllowedExtraOpaqueBytes, "blob", ) switch { @@ -4132,6 +4132,10 @@ func deserializeLightningNode(r io.Reader) (models.LightningNode, error) { return models.LightningNode{}, err } + if len(extraBytes) > 0 { + node.ExtraOpaqueData = extraBytes + } + return node, nil } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 66a59af80..1ca00442a 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -494,10 +494,9 @@ func buildNode(ctx context.Context, db SQLQueries, dbNode *sqlc.Node) ( copy(pub[:], dbNode.PubKey) node := &models.LightningNode{ - PubKeyBytes: pub, - Features: lnwire.EmptyFeatureVector(), - LastUpdate: time.Unix(0, 0), - ExtraOpaqueData: make([]byte, 0), + PubKeyBytes: pub, + Features: lnwire.EmptyFeatureVector(), + LastUpdate: time.Unix(0, 0), } if len(dbNode.Signature) == 0 { From c5f159f48557130247b6fa3242e953aed524d751 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 27 May 2025 08:24:28 +0200 Subject: [PATCH 3/5] graph/db: expand AddChannelEdge test Expand the existing TestAddChannelEdgeShellNodes test so that we have coverage for error we expect when AddChannelEdge is called a second time if we already know of a channel. --- graph/db/graph_test.go | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index e644fea19..bd4a2d56f 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -3337,29 +3337,28 @@ func TestAddChannelEdgeShellNodes(t *testing.T) { // To start, we'll create two nodes, and only add one of them to the // channel graph. node1 := createTestVertex(t) - if err := graph.AddLightningNode(node1); err != nil { - t.Fatalf("unable to add node: %v", err) - } + require.NoError(t, graph.SetSourceNode(node1)) node2 := createTestVertex(t) // We'll now create an edge between the two nodes, as a result, node2 // should be inserted into the database as a shell node. edgeInfo, _ := createEdge(100, 0, 0, 0, node1, node2) - if err := graph.AddChannelEdge(&edgeInfo); err != nil { - t.Fatalf("unable to add edge: %v", err) - } + require.NoError(t, graph.AddChannelEdge(&edgeInfo)) // Ensure that node1 was inserted as a full node, while node2 only has // a shell node present. node1, err := graph.FetchLightningNode(node1.PubKeyBytes) require.NoError(t, err, "unable to fetch node1") - if !node1.HaveNodeAnnouncement { - t.Fatalf("have shell announcement for node1, shouldn't") - } + require.True(t, node1.HaveNodeAnnouncement) node2, err = graph.FetchLightningNode(node2.PubKeyBytes) require.NoError(t, err, "unable to fetch node2") require.False(t, node2.HaveNodeAnnouncement) + + // Show that attempting to add the channel again will result in an + // error. + err = graph.AddChannelEdge(&edgeInfo) + require.ErrorIs(t, err, ErrEdgeAlreadyExist) } // TestNodePruningUpdateIndexDeletion tests that once a node has been removed From d93d104a66e79b25b41c99be20e018e9e363c46e Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sat, 24 May 2025 15:09:37 +0200 Subject: [PATCH 4/5] graph/db+sqldb: implement AddChannelEdge on SQLStore In this commit, the `AddChannelEdge` method of the SQLStore is implemented. Like the KVStore implementation, it makes use of the available channel `batch.Scheduler` and also updates the reject and channel caches. This then lets us convert the following 2 unit tests to run against the SQL backends: - TestPartialNode - TestAddChannelEdgeShellNodes --- graph/db/graph_test.go | 4 +- graph/db/sql_store.go | 205 +++++++++++++++++++++++++++++++++++ sqldb/sqlc/graph.sql.go | 129 ++++++++++++++++++++++ sqldb/sqlc/querier.go | 4 + sqldb/sqlc/queries/graph.sql | 43 ++++++++ 5 files changed, 383 insertions(+), 2 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index bd4a2d56f..026fedbdc 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -261,7 +261,7 @@ func TestNodeInsertionAndDeletion(t *testing.T) { func TestPartialNode(t *testing.T) { t.Parallel() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) // To insert a partial node, we need to add a channel edge that has // node keys for nodes we are not yet aware @@ -3332,7 +3332,7 @@ func TestPruneGraphNodes(t *testing.T) { func TestAddChannelEdgeShellNodes(t *testing.T) { t.Parallel() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) // To start, we'll create two nodes, and only add one of them to the // channel graph. diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 1ca00442a..83749c81c 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -69,6 +69,15 @@ type SQLQueries interface { */ AddSourceNode(ctx context.Context, nodeID int64) error GetSourceNodesByVersion(ctx context.Context, version int16) ([]sqlc.GetSourceNodesByVersionRow, error) + + /* + Channel queries. + */ + CreateChannel(ctx context.Context, arg sqlc.CreateChannelParams) (int64, error) + GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.Channel, error) + + CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error + InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error } // BatchedSQLQueries is a version of SQLQueries that's capable of batched @@ -455,6 +464,54 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime, return nodes, 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 +// involved in creation of the channel, and the set of features that the channel +// supports. The chanPoint and chanID are used to uniquely identify the edge +// globally within the database. +// +// NOTE: part of the V1Store interface. +func (s *SQLStore) AddChannelEdge(edge *models.ChannelEdgeInfo, + opts ...batch.SchedulerOption) error { + + ctx := context.TODO() + + var alreadyExists bool + r := &batch.Request[SQLQueries]{ + Opts: batch.NewSchedulerOptions(opts...), + Reset: func() { + alreadyExists = false + }, + Do: func(tx SQLQueries) error { + err := insertChannel(ctx, tx, edge) + + // Silence ErrEdgeAlreadyExist so that the batch can + // succeed, but propagate the error via local state. + if errors.Is(err, ErrEdgeAlreadyExist) { + alreadyExists = true + return nil + } + + return err + }, + OnCommit: func(err error) error { + switch { + case err != nil: + return err + case alreadyExists: + return ErrEdgeAlreadyExist + default: + s.rejectCache.remove(edge.ChannelID) + s.chanCache.remove(edge.ChannelID) + return nil + } + }, + } + + return s.chanScheduler.Execute(ctx, r) +} + // getNodeByPubKey attempts to look up a target node by its public key. func getNodeByPubKey(ctx context.Context, db SQLQueries, pubKey route.Vertex) (int64, *models.LightningNode, error) { @@ -1022,3 +1079,151 @@ func marshalExtraOpaqueData(data []byte) (map[uint64][]byte, error) { return records, nil } + +// insertChannel inserts a new channel record into the database. +func insertChannel(ctx context.Context, db SQLQueries, + edge *models.ChannelEdgeInfo) error { + + var chanIDB [8]byte + byteOrder.PutUint64(chanIDB[:], edge.ChannelID) + + // Make sure that the channel doesn't already exist. We do this + // explicitly instead of relying on catching a unique constraint error + // because relying on SQL to throw that error would abort the entire + // batch of transactions. + _, err := db.GetChannelBySCID( + ctx, sqlc.GetChannelBySCIDParams{ + Scid: chanIDB[:], + Version: int16(ProtocolV1), + }, + ) + if err == nil { + return ErrEdgeAlreadyExist + } else if !errors.Is(err, sql.ErrNoRows) { + return 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) + } + + node2DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey2Bytes) + if err != nil { + return fmt.Errorf("unable to create shell node: %w", err) + } + + var capacity sql.NullInt64 + if edge.Capacity != 0 { + capacity = sqldb.SQLInt64(int64(edge.Capacity)) + } + + createParams := sqlc.CreateChannelParams{ + Version: int16(ProtocolV1), + Scid: chanIDB[:], + NodeID1: node1DBID, + NodeID2: node2DBID, + Outpoint: edge.ChannelPoint.String(), + Capacity: capacity, + BitcoinKey1: edge.BitcoinKey1Bytes[:], + BitcoinKey2: edge.BitcoinKey2Bytes[:], + } + + if edge.AuthProof != nil { + proof := edge.AuthProof + + createParams.Node1Signature = proof.NodeSig1Bytes + createParams.Node2Signature = proof.NodeSig2Bytes + createParams.Bitcoin1Signature = proof.BitcoinSig1Bytes + createParams.Bitcoin2Signature = proof.BitcoinSig2Bytes + } + + // Insert the new channel record. + dbChanID, err := db.CreateChannel(ctx, createParams) + if err != nil { + return err + } + + // Insert any channel features. + if len(edge.Features) != 0 { + chanFeatures := lnwire.NewRawFeatureVector() + err := chanFeatures.Decode(bytes.NewReader(edge.Features)) + if err != nil { + return err + } + + fv := lnwire.NewFeatureVector(chanFeatures, lnwire.Features) + for feature := range fv.Features() { + err = db.InsertChannelFeature( + ctx, sqlc.InsertChannelFeatureParams{ + ChannelID: dbChanID, + FeatureBit: int32(feature), + }, + ) + if err != nil { + return fmt.Errorf("unable to insert "+ + "channel(%d) feature(%v): %w", dbChanID, + feature, err) + } + } + } + + // 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) + } + + for tlvType, value := range extra { + err := db.CreateChannelExtraType( + ctx, sqlc.CreateChannelExtraTypeParams{ + ChannelID: dbChanID, + Type: int64(tlvType), + Value: value, + }, + ) + if err != nil { + return fmt.Errorf("unable to upsert channel(%d) extra "+ + "signed field(%v): %w", edge.ChannelID, + tlvType, err) + } + } + + return nil +} + +// maybeCreateShellNode checks if a shell node entry exists for the +// given public key. If it does not exist, then a new shell node entry is +// created. The ID of the node is returned. A shell node only has a protocol +// version and public key persisted. +func maybeCreateShellNode(ctx context.Context, db SQLQueries, + pubKey route.Vertex) (int64, error) { + + dbNode, err := db.GetNodeByPubKey( + ctx, sqlc.GetNodeByPubKeyParams{ + PubKey: pubKey[:], + Version: int16(ProtocolV1), + }, + ) + // The node exists. Return the ID. + if err == nil { + return dbNode.ID, nil + } else if !errors.Is(err, sql.ErrNoRows) { + return 0, err + } + + // Otherwise, the node does not exist, so we create a shell entry for + // it. + id, err := db.UpsertNode(ctx, sqlc.UpsertNodeParams{ + Version: int16(ProtocolV1), + PubKey: pubKey[:], + }) + if err != nil { + return 0, fmt.Errorf("unable to create shell node: %w", err) + } + + return id, nil +} diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index dcdd79053..ec730818f 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -26,6 +26,81 @@ func (q *Queries) AddSourceNode(ctx context.Context, nodeID int64) error { return err } +const createChannel = `-- name: CreateChannel :one +/* ───────────────────────────────────────────── + channels table queries + ───────────────────────────────────────────── +*/ + +INSERT INTO channels ( + version, scid, node_id_1, node_id_2, + outpoint, capacity, bitcoin_key_1, bitcoin_key_2, + node_1_signature, node_2_signature, bitcoin_1_signature, + bitcoin_2_signature +) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12 +) +RETURNING id +` + +type CreateChannelParams struct { + Version int16 + Scid []byte + NodeID1 int64 + NodeID2 int64 + Outpoint string + Capacity sql.NullInt64 + BitcoinKey1 []byte + BitcoinKey2 []byte + Node1Signature []byte + Node2Signature []byte + Bitcoin1Signature []byte + Bitcoin2Signature []byte +} + +func (q *Queries) CreateChannel(ctx context.Context, arg CreateChannelParams) (int64, error) { + row := q.db.QueryRowContext(ctx, createChannel, + arg.Version, + arg.Scid, + arg.NodeID1, + arg.NodeID2, + arg.Outpoint, + arg.Capacity, + arg.BitcoinKey1, + arg.BitcoinKey2, + arg.Node1Signature, + arg.Node2Signature, + arg.Bitcoin1Signature, + arg.Bitcoin2Signature, + ) + var id int64 + err := row.Scan(&id) + return id, err +} + +const createChannelExtraType = `-- name: CreateChannelExtraType :exec +/* ───────────────────────────────────────────── + channel_extra_types table queries + ───────────────────────────────────────────── +*/ + +INSERT INTO channel_extra_types ( + channel_id, type, value +) +VALUES ($1, $2, $3) +` + +type CreateChannelExtraTypeParams struct { + ChannelID int64 + Type int64 + Value []byte +} + +func (q *Queries) CreateChannelExtraType(ctx context.Context, arg CreateChannelExtraTypeParams) error { + _, err := q.db.ExecContext(ctx, createChannelExtraType, arg.ChannelID, arg.Type, arg.Value) + return err +} + const deleteExtraNodeType = `-- name: DeleteExtraNodeType :exec DELETE FROM node_extra_types WHERE node_id = $1 @@ -83,6 +158,37 @@ func (q *Queries) DeleteNodeFeature(ctx context.Context, arg DeleteNodeFeaturePa return err } +const getChannelBySCID = `-- name: GetChannelBySCID :one +SELECT id, version, scid, node_id_1, node_id_2, outpoint, capacity, bitcoin_key_1, bitcoin_key_2, node_1_signature, node_2_signature, bitcoin_1_signature, bitcoin_2_signature FROM channels +WHERE scid = $1 AND version = $2 +` + +type GetChannelBySCIDParams struct { + Scid []byte + Version int16 +} + +func (q *Queries) GetChannelBySCID(ctx context.Context, arg GetChannelBySCIDParams) (Channel, error) { + row := q.db.QueryRowContext(ctx, getChannelBySCID, arg.Scid, arg.Version) + var i Channel + err := row.Scan( + &i.ID, + &i.Version, + &i.Scid, + &i.NodeID1, + &i.NodeID2, + &i.Outpoint, + &i.Capacity, + &i.BitcoinKey1, + &i.BitcoinKey2, + &i.Node1Signature, + &i.Node2Signature, + &i.Bitcoin1Signature, + &i.Bitcoin2Signature, + ) + return i, err +} + const getExtraNodeTypes = `-- name: GetExtraNodeTypes :many SELECT node_id, type, value FROM node_extra_types @@ -323,6 +429,29 @@ func (q *Queries) GetSourceNodesByVersion(ctx context.Context, version int16) ([ return items, nil } +const insertChannelFeature = `-- name: InsertChannelFeature :exec +/* ───────────────────────────────────────────── + channel_features table queries + ───────────────────────────────────────────── +*/ + +INSERT INTO channel_features ( + channel_id, feature_bit +) VALUES ( + $1, $2 +) +` + +type InsertChannelFeatureParams struct { + ChannelID int64 + FeatureBit int32 +} + +func (q *Queries) InsertChannelFeature(ctx context.Context, arg InsertChannelFeatureParams) error { + _, err := q.db.ExecContext(ctx, insertChannelFeature, arg.ChannelID, arg.FeatureBit) + return err +} + const insertNodeAddress = `-- name: InsertNodeAddress :exec /* ───────────────────────────────────────────── node_addresses table queries diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 7492a21e0..e3482d524 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -13,6 +13,8 @@ import ( type Querier interface { AddSourceNode(ctx context.Context, nodeID int64) error ClearKVInvoiceHashIndex(ctx context.Context) error + CreateChannel(ctx context.Context, arg CreateChannelParams) (int64, error) + CreateChannelExtraType(ctx context.Context, arg CreateChannelExtraTypeParams) error DeleteCanceledInvoices(ctx context.Context) (sql.Result, error) DeleteExtraNodeType(ctx context.Context, arg DeleteExtraNodeTypeParams) error DeleteInvoice(ctx context.Context, arg DeleteInvoiceParams) (sql.Result, error) @@ -24,6 +26,7 @@ type Querier interface { FetchSettledAMPSubInvoices(ctx context.Context, arg FetchSettledAMPSubInvoicesParams) ([]FetchSettledAMPSubInvoicesRow, error) FilterInvoices(ctx context.Context, arg FilterInvoicesParams) ([]Invoice, error) GetAMPInvoiceID(ctx context.Context, setID []byte) (int64, error) + GetChannelBySCID(ctx context.Context, arg GetChannelBySCIDParams) (Channel, error) GetDatabaseVersion(ctx context.Context) (int32, error) GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]NodeExtraType, error) // This method may return more than one invoice if filter using multiple fields @@ -45,6 +48,7 @@ type Querier interface { GetSourceNodesByVersion(ctx context.Context, version int16) ([]GetSourceNodesByVersionRow, error) InsertAMPSubInvoice(ctx context.Context, arg InsertAMPSubInvoiceParams) error InsertAMPSubInvoiceHTLC(ctx context.Context, arg InsertAMPSubInvoiceHTLCParams) error + InsertChannelFeature(ctx context.Context, arg InsertChannelFeatureParams) error InsertInvoice(ctx context.Context, arg InsertInvoiceParams) (int64, error) InsertInvoiceFeature(ctx context.Context, arg InsertInvoiceFeatureParams) error InsertInvoiceHTLC(ctx context.Context, arg InsertInvoiceHTLCParams) (int64, error) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index a6745227e..b66aaaa75 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -132,3 +132,46 @@ SELECT sn.node_id, n.pub_key FROM source_nodes sn JOIN nodes n ON sn.node_id = n.id WHERE n.version = $1; + +/* ───────────────────────────────────────────── + channels table queries + ───────────────────────────────────────────── +*/ + +-- name: CreateChannel :one +INSERT INTO channels ( + version, scid, node_id_1, node_id_2, + outpoint, capacity, bitcoin_key_1, bitcoin_key_2, + node_1_signature, node_2_signature, bitcoin_1_signature, + bitcoin_2_signature +) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12 +) +RETURNING id; + +-- name: GetChannelBySCID :one +SELECT * FROM channels +WHERE scid = $1 AND version = $2; + +/* ───────────────────────────────────────────── + channel_features table queries + ───────────────────────────────────────────── +*/ + +-- name: InsertChannelFeature :exec +INSERT INTO channel_features ( + channel_id, feature_bit +) VALUES ( + $1, $2 +); + +/* ───────────────────────────────────────────── + channel_extra_types table queries + ───────────────────────────────────────────── +*/ + +-- name: CreateChannelExtraType :exec +INSERT INTO channel_extra_types ( + channel_id, type, value +) +VALUES ($1, $2, $3); From cf542458bdc04f3163bef2c21ec1a8fa8b1e6a01 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sat, 24 May 2025 15:21:56 +0200 Subject: [PATCH 5/5] graph/db+sqldb: add HighestChanID to SQLStore --- graph/db/graph_test.go | 2 +- graph/db/sql_store.go | 30 ++++++++++++++++++++++++++++++ sqldb/sqlc/graph.sql.go | 15 +++++++++++++++ sqldb/sqlc/querier.go | 1 + sqldb/sqlc/queries/graph.sql | 7 +++++++ 5 files changed, 54 insertions(+), 1 deletion(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 026fedbdc..6984517e0 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1861,7 +1861,7 @@ func TestGraphPruning(t *testing.T) { func TestHighestChanID(t *testing.T) { t.Parallel() - graph := MakeTestGraph(t) + graph := MakeTestGraphNew(t) // If we don't yet have any channels in the database, then we should // get a channel ID of zero if we ask for the highest channel ID. diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 83749c81c..1d296d2d2 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -75,6 +75,7 @@ type SQLQueries interface { */ CreateChannel(ctx context.Context, arg sqlc.CreateChannelParams) (int64, error) GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.Channel, error) + HighestSCID(ctx context.Context, version int16) ([]byte, error) CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error @@ -512,6 +513,35 @@ func (s *SQLStore) AddChannelEdge(edge *models.ChannelEdgeInfo, return s.chanScheduler.Execute(ctx, r) } +// HighestChanID returns the "highest" known channel ID in the channel graph. +// This represents the "newest" channel from the PoV of the chain. This method +// can be used by peers to quickly determine if their graphs are in sync. +// +// NOTE: This is part of the V1Store interface. +func (s *SQLStore) HighestChanID() (uint64, error) { + ctx := context.TODO() + + var highestChanID uint64 + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { + chanID, err := db.HighestSCID(ctx, int16(ProtocolV1)) + if errors.Is(err, sql.ErrNoRows) { + return nil + } else if err != nil { + return fmt.Errorf("unable to fetch highest chan ID: %w", + err) + } + + highestChanID = byteOrder.Uint64(chanID) + + return nil + }, sqldb.NoOpReset) + if err != nil { + return 0, fmt.Errorf("unable to fetch highest chan ID: %w", err) + } + + return highestChanID, nil +} + // getNodeByPubKey attempts to look up a target node by its public key. func getNodeByPubKey(ctx context.Context, db SQLQueries, pubKey route.Vertex) (int64, *models.LightningNode, error) { diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index ec730818f..0823132a3 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -429,6 +429,21 @@ func (q *Queries) GetSourceNodesByVersion(ctx context.Context, version int16) ([ return items, nil } +const highestSCID = `-- name: HighestSCID :one +SELECT scid +FROM channels +WHERE version = $1 +ORDER BY scid DESC +LIMIT 1 +` + +func (q *Queries) HighestSCID(ctx context.Context, version int16) ([]byte, error) { + row := q.db.QueryRowContext(ctx, highestSCID, version) + var scid []byte + err := row.Scan(&scid) + return scid, err +} + const insertChannelFeature = `-- name: InsertChannelFeature :exec /* ───────────────────────────────────────────── channel_features table queries diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index e3482d524..b905dd4c8 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -46,6 +46,7 @@ type Querier interface { GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]Node, error) GetSourceNodesByVersion(ctx context.Context, version int16) ([]GetSourceNodesByVersionRow, error) + HighestSCID(ctx context.Context, version int16) ([]byte, error) InsertAMPSubInvoice(ctx context.Context, arg InsertAMPSubInvoiceParams) error InsertAMPSubInvoiceHTLC(ctx context.Context, arg InsertAMPSubInvoiceHTLCParams) error InsertChannelFeature(ctx context.Context, arg InsertChannelFeatureParams) error diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index b66aaaa75..dd96ac2f1 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -153,6 +153,13 @@ RETURNING id; SELECT * FROM channels WHERE scid = $1 AND version = $2; +-- name: HighestSCID :one +SELECT scid +FROM channels +WHERE version = $1 +ORDER BY scid DESC +LIMIT 1; + /* ───────────────────────────────────────────── channel_features table queries ─────────────────────────────────────────────