Merge pull request #9938 from ellemouton/graphSQL14

[14] graph/db: implement more SQLStore methods
This commit is contained in:
Oliver Gugger authored and GitHub committed 2025-06-25 09:28:04 +02:00
commit 33e6f28542
6 files changed
+969 -16

No files matched your search

@@ -87,6 +87,7 @@ circuit. The indices are only available for forwarding events saved after v0.20.
* [5](https://github.com/lightningnetwork/lnd/pull/9935)
* [6](https://github.com/lightningnetwork/lnd/pull/9936)
* [7](https://github.com/lightningnetwork/lnd/pull/9937)
* [8](https://github.com/lightningnetwork/lnd/pull/9938)
## RPC Updates
+28 -15
View File
@@ -73,13 +73,11 @@ var (
)
func createLightningNode(priv *btcec.PrivateKey) *models.LightningNode {
updateTime := prand.Int63()
pub := priv.PubKey().SerializeCompressed()
n := &models.LightningNode{
HaveNodeAnnouncement: true,
AuthSigBytes: testSig.Serialize(),
LastUpdate: time.Unix(updateTime, 0),
LastUpdate: time.Unix(nextUpdateTime(), 0),
Color: color.RGBA{1, 2, 3, 0},
Alias: "kek" + hex.EncodeToString(pub),
Features: testFeatures,
@@ -392,7 +390,7 @@ func TestEdgeInsertionDeletion(t *testing.T) {
t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t)
graph := MakeTestGraphNew(t)
// We'd like to test the insertion/deletion of edges, so we create two
// vertexes to connect.
@@ -812,7 +810,7 @@ func TestEdgeInfoUpdates(t *testing.T) {
t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t)
graph := MakeTestGraphNew(t)
// We'd like to test the update of edges inserted into the database, so
// we create two vertexes to connect.
@@ -2198,7 +2196,7 @@ func TestNodeUpdatesInHorizon(t *testing.T) {
func TestFilterKnownChanIDsZombieRevival(t *testing.T) {
t.Parallel()
graph := MakeTestGraph(t)
graph := MakeTestGraphNew(t)
var (
scid1 = lnwire.ShortChannelID{BlockHeight: 1}
@@ -2264,7 +2262,7 @@ func TestFilterKnownChanIDs(t *testing.T) {
t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t)
graph := MakeTestGraphNew(t)
isZombieUpdate := func(updateTime1 time.Time,
updateTime2 time.Time) bool {
@@ -2946,7 +2944,7 @@ func TestFetchChanInfos(t *testing.T) {
t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t)
graph := MakeTestGraphNew(t)
// We'll first populate our graph with two nodes. All channels created
// below will be made between these two nodes.
@@ -3439,6 +3437,20 @@ func TestNodePruningUpdateIndexDeletion(t *testing.T) {
}
}
var (
updateTime = prand.Int63()
updateTimeMu sync.Mutex
)
func nextUpdateTime() int64 {
updateTimeMu.Lock()
defer updateTimeMu.Unlock()
updateTime++
return updateTime
}
// TestNodeIsPublic ensures that we properly detect nodes that are seen as
// public within the network graph.
func TestNodeIsPublic(t *testing.T) {
@@ -3453,19 +3465,19 @@ func TestNodeIsPublic(t *testing.T) {
// We'll need to create a separate database and channel graph for each
// participant to replicate real-world scenarios (private edges being in
// some graphs but not others, etc.).
aliceGraph := MakeTestGraph(t)
aliceGraph := MakeTestGraphNew(t)
aliceNode := createTestVertex(t)
if err := aliceGraph.SetSourceNode(ctx, aliceNode); err != nil {
t.Fatalf("unable to set source node: %v", err)
}
bobGraph := MakeTestGraph(t)
bobGraph := MakeTestGraphNew(t)
bobNode := createTestVertex(t)
if err := bobGraph.SetSourceNode(ctx, bobNode); err != nil {
t.Fatalf("unable to set source node: %v", err)
}
carolGraph := MakeTestGraph(t)
carolGraph := MakeTestGraphNew(t)
carolNode := createTestVertex(t)
if err := carolGraph.SetSourceNode(ctx, carolNode); err != nil {
t.Fatalf("unable to set source node: %v", err)
@@ -3481,13 +3493,13 @@ func TestNodeIsPublic(t *testing.T) {
graphs := []*ChannelGraph{aliceGraph, bobGraph, carolGraph}
for _, graph := range graphs {
for _, node := range nodes {
node.LastUpdate = time.Unix(nextUpdateTime(), 0)
err := graph.AddLightningNode(ctx, node)
require.NoError(t, err)
}
for _, edge := range edges {
if err := graph.AddChannelEdge(ctx, edge); err != nil {
t.Fatalf("unable to add edge: %v", err)
}
err := graph.AddChannelEdge(ctx, edge)
require.NoError(t, err)
}
}
@@ -3580,7 +3592,7 @@ func TestDisabledChannelIDs(t *testing.T) {
t.Parallel()
ctx := context.Background()
graph := MakeTestGraph(t)
graph := MakeTestGraphNew(t)
// Create first node and add it to the graph.
node1 := createTestVertex(t)
@@ -3596,6 +3608,7 @@ func TestDisabledChannelIDs(t *testing.T) {
// Adding a new channel edge to the graph.
edgeInfo, edge1, edge2 := createChannelEdge(node1, node2)
node2.LastUpdate = time.Unix(nextUpdateTime(), 0)
if err := graph.AddLightningNode(ctx, node2); err != nil {
t.Fatalf("unable to add node: %v", err)
}
+610 -1
View File
@@ -62,6 +62,7 @@ type SQLQueries interface {
GetNodesByLastUpdateRange(ctx context.Context, arg sqlc.GetNodesByLastUpdateRangeParams) ([]sqlc.Node, error)
ListNodesPaginated(ctx context.Context, arg sqlc.ListNodesPaginatedParams) ([]sqlc.Node, error)
ListNodeIDsAndPubKeys(ctx context.Context, arg sqlc.ListNodeIDsAndPubKeysParams) ([]sqlc.ListNodeIDsAndPubKeysRow, error)
IsPublicV1Node(ctx context.Context, pubKey []byte) (bool, error)
DeleteNodeByPubKey(ctx context.Context, arg sqlc.DeleteNodeByPubKeyParams) (sql.Result, error)
GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.NodeExtraType, error)
@@ -95,7 +96,9 @@ type SQLQueries interface {
ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error)
ListChannelsWithPoliciesPaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesPaginatedParams) ([]sqlc.ListChannelsWithPoliciesPaginatedRow, error)
GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg sqlc.GetChannelsByPolicyLastUpdateRangeParams) ([]sqlc.GetChannelsByPolicyLastUpdateRangeRow, error)
GetChannelByOutpointWithPolicies(ctx context.Context, arg sqlc.GetChannelByOutpointWithPoliciesParams) (sqlc.GetChannelByOutpointWithPoliciesRow, error)
GetPublicV1ChannelsBySCID(ctx context.Context, arg sqlc.GetPublicV1ChannelsBySCIDParams) ([]sqlc.Channel, error)
GetSCIDByOutpoint(ctx context.Context, arg sqlc.GetSCIDByOutpointParams) ([]byte, error)
DeleteChannel(ctx context.Context, id int64) error
CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error
@@ -106,6 +109,7 @@ type SQLQueries interface {
*/
UpsertEdgePolicy(ctx context.Context, arg sqlc.UpsertEdgePolicyParams) (int64, error)
GetChannelPolicyByChannelAndNode(ctx context.Context, arg sqlc.GetChannelPolicyByChannelAndNodeParams) (sqlc.ChannelPolicy, error)
GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error)
InsertChanPolicyExtraType(ctx context.Context, arg sqlc.InsertChanPolicyExtraTypeParams) error
GetChannelPolicyExtraTypes(ctx context.Context, arg sqlc.GetChannelPolicyExtraTypesParams) ([]sqlc.GetChannelPolicyExtraTypesRow, error)
@@ -118,6 +122,7 @@ type SQLQueries interface {
GetZombieChannel(ctx context.Context, arg sqlc.GetZombieChannelParams) (sqlc.ZombieChannel, error)
CountZombieChannels(ctx context.Context, version int16) (int64, error)
DeleteZombieChannel(ctx context.Context, arg sqlc.DeleteZombieChannelParams) (sql.Result, error)
IsZombieChannel(ctx context.Context, arg sqlc.IsZombieChannelParams) (bool, error)
}
// BatchedSQLQueries is a version of SQLQueries that's capable of batched
@@ -368,6 +373,35 @@ func (s *SQLStore) FetchNodeFeatures(nodePub route.Vertex) (
return fetchNodeFeatures(ctx, s.db, nodePub)
}
// DisabledChannelIDs returns the channel ids of disabled channels.
// A channel is disabled when two of the associated ChanelEdgePolicies
// have their disabled bit on.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) DisabledChannelIDs() ([]uint64, error) {
var (
ctx = context.TODO()
chanIDs []uint64
)
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
dbChanIDs, err := db.GetV1DisabledSCIDs(ctx)
if err != nil {
return fmt.Errorf("unable to fetch disabled "+
"channels: %w", err)
}
chanIDs = fn.Map(dbChanIDs, byteOrder.Uint64)
return nil
}, sqldb.NoOpReset)
if err != nil {
return nil, fmt.Errorf("unable to fetch disabled channels: %w",
err)
}
return chanIDs, nil
}
// LookupAlias attempts to return the alias as advertised by the target node.
//
// NOTE: part of the V1Store interface.
@@ -1671,6 +1705,501 @@ func (s *SQLStore) DeleteChannelEdges(strictZombiePruning, markZombie bool,
return deleted, nil
}
// FetchChannelEdgesByID attempts to lookup the two directed edges for the
// channel identified by the channel ID. If the channel can't be found, then
// ErrEdgeNotFound is returned. A struct which houses the general information
// for the channel itself is returned as well as two structs that contain the
// routing policies for the channel in either direction.
//
// ErrZombieEdge an be returned if the edge is currently marked as a zombie
// within the database. In this case, the ChannelEdgePolicy's will be nil, and
// the ChannelEdgeInfo will only include the public keys of each node.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) FetchChannelEdgesByID(chanID uint64) (
*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
*models.ChannelEdgePolicy, error) {
var (
ctx = context.TODO()
edge *models.ChannelEdgeInfo
policy1, policy2 *models.ChannelEdgePolicy
)
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[:],
Version: int16(ProtocolV1),
},
)
if errors.Is(err, sql.ErrNoRows) {
// First check if this edge is perhaps in the zombie
// index.
isZombie, err := db.IsZombieChannel(
ctx, sqlc.IsZombieChannelParams{
Scid: chanIDB[:],
Version: int16(ProtocolV1),
},
)
if err != nil {
return fmt.Errorf("unable to check if "+
"channel is zombie: %w", err)
} else if isZombie {
return ErrZombieEdge
}
return ErrEdgeNotFound
} else if err != nil {
return fmt.Errorf("unable to fetch channel: %w", err)
}
node1, node2, err := buildNodeVertices(
row.Node.PubKey, row.Node_2.PubKey,
)
if err != nil {
return err
}
edge, err = getAndBuildEdgeInfo(
ctx, db, s.cfg.ChainHash, row.Channel.ID, row.Channel,
node1, node2,
)
if err != nil {
return fmt.Errorf("unable to build channel info: %w",
err)
}
dbPol1, dbPol2, err := extractChannelPolicies(row)
if err != nil {
return fmt.Errorf("unable to extract channel "+
"policies: %w", err)
}
policy1, policy2, err = getAndBuildChanPolicies(
ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2,
)
if err != nil {
return fmt.Errorf("unable to build channel "+
"policies: %w", err)
}
return nil
}, sqldb.NoOpReset)
if err != nil {
return nil, nil, nil, fmt.Errorf("could not fetch channel: %w",
err)
}
return edge, policy1, policy2, nil
}
// FetchChannelEdgesByOutpoint attempts to lookup the two directed edges for
// the channel identified by the funding outpoint. If the channel can't be
// found, then ErrEdgeNotFound is returned. A struct which houses the general
// information for the channel itself is returned as well as two structs that
// contain the routing policies for the channel in either direction.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) FetchChannelEdgesByOutpoint(op *wire.OutPoint) (
*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
*models.ChannelEdgePolicy, error) {
var (
ctx = context.TODO()
edge *models.ChannelEdgeInfo
policy1, policy2 *models.ChannelEdgePolicy
)
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
row, err := db.GetChannelByOutpointWithPolicies(
ctx, sqlc.GetChannelByOutpointWithPoliciesParams{
Outpoint: op.String(),
Version: int16(ProtocolV1),
},
)
if errors.Is(err, sql.ErrNoRows) {
return ErrEdgeNotFound
} else if err != nil {
return fmt.Errorf("unable to fetch channel: %w", err)
}
node1, node2, err := buildNodeVertices(
row.Node1Pubkey, row.Node2Pubkey,
)
if err != nil {
return err
}
edge, err = getAndBuildEdgeInfo(
ctx, db, s.cfg.ChainHash, row.Channel.ID, row.Channel,
node1, node2,
)
if err != nil {
return fmt.Errorf("unable to build channel info: %w",
err)
}
dbPol1, dbPol2, err := extractChannelPolicies(row)
if err != nil {
return fmt.Errorf("unable to extract channel "+
"policies: %w", err)
}
policy1, policy2, err = getAndBuildChanPolicies(
ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2,
)
if err != nil {
return fmt.Errorf("unable to build channel "+
"policies: %w", err)
}
return nil
}, sqldb.NoOpReset)
if err != nil {
return nil, nil, nil, fmt.Errorf("could not fetch channel: %w",
err)
}
return edge, policy1, policy2, nil
}
// HasChannelEdge returns true if the database knows of a channel edge with the
// passed channel ID, and false otherwise. If an edge with that ID is found
// within the graph, then two time stamps representing the last time the edge
// was updated for both directed edges are returned along with the boolean. If
// it is not found, then the zombie index is checked and its result is returned
// as the second boolean.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) HasChannelEdge(chanID uint64) (time.Time, time.Time, bool,
bool, error) {
ctx := context.TODO()
var (
exists bool
isZombie bool
node1LastUpdate time.Time
node2LastUpdate time.Time
)
// We'll query the cache with the shared lock held to allow multiple
// readers to access values in the cache concurrently if they exist.
s.cacheMu.RLock()
if entry, ok := s.rejectCache.get(chanID); ok {
s.cacheMu.RUnlock()
node1LastUpdate = time.Unix(entry.upd1Time, 0)
node2LastUpdate = time.Unix(entry.upd2Time, 0)
exists, isZombie = entry.flags.unpack()
return node1LastUpdate, node2LastUpdate, exists, isZombie, nil
}
s.cacheMu.RUnlock()
s.cacheMu.Lock()
defer s.cacheMu.Unlock()
// The item was not found with the shared lock, so we'll acquire the
// exclusive lock and check the cache again in case another method added
// the entry to the cache while no lock was held.
if entry, ok := s.rejectCache.get(chanID); ok {
node1LastUpdate = time.Unix(entry.upd1Time, 0)
node2LastUpdate = time.Unix(entry.upd2Time, 0)
exists, isZombie = entry.flags.unpack()
return node1LastUpdate, node2LastUpdate, exists, isZombie, nil
}
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[:],
Version: int16(ProtocolV1),
},
)
if errors.Is(err, sql.ErrNoRows) {
// Check if it is a zombie channel.
isZombie, err = db.IsZombieChannel(
ctx, sqlc.IsZombieChannelParams{
Scid: chanIDB[:],
Version: int16(ProtocolV1),
},
)
if err != nil {
return fmt.Errorf("could not check if channel "+
"is zombie: %w", err)
}
return nil
} else if err != nil {
return fmt.Errorf("unable to fetch channel: %w", err)
}
exists = true
policy1, err := db.GetChannelPolicyByChannelAndNode(
ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{
Version: int16(ProtocolV1),
ChannelID: channel.ID,
NodeID: channel.NodeID1,
},
)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("unable to fetch channel policy: %w",
err)
} else if err == nil {
node1LastUpdate = time.Unix(policy1.LastUpdate.Int64, 0)
}
policy2, err := db.GetChannelPolicyByChannelAndNode(
ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{
Version: int16(ProtocolV1),
ChannelID: channel.ID,
NodeID: channel.NodeID2,
},
)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("unable to fetch channel policy: %w",
err)
} else if err == nil {
node2LastUpdate = time.Unix(policy2.LastUpdate.Int64, 0)
}
return nil
}, sqldb.NoOpReset)
if err != nil {
return time.Time{}, time.Time{}, false, false,
fmt.Errorf("unable to fetch channel: %w", err)
}
s.rejectCache.insert(chanID, rejectCacheEntry{
upd1Time: node1LastUpdate.Unix(),
upd2Time: node2LastUpdate.Unix(),
flags: packRejectFlags(exists, isZombie),
})
return node1LastUpdate, node2LastUpdate, exists, isZombie, nil
}
// ChannelID attempt to lookup the 8-byte compact channel ID which maps to the
// passed channel point (outpoint). If the passed channel doesn't exist within
// the database, then ErrEdgeNotFound is returned.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) ChannelID(chanPoint *wire.OutPoint) (uint64, error) {
var (
ctx = context.TODO()
channelID uint64
)
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
chanID, err := db.GetSCIDByOutpoint(
ctx, sqlc.GetSCIDByOutpointParams{
Outpoint: chanPoint.String(),
Version: int16(ProtocolV1),
},
)
if errors.Is(err, sql.ErrNoRows) {
return ErrEdgeNotFound
} else if err != nil {
return fmt.Errorf("unable to fetch channel ID: %w",
err)
}
channelID = byteOrder.Uint64(chanID)
return nil
}, sqldb.NoOpReset)
if err != nil {
return 0, fmt.Errorf("unable to fetch channel ID: %w", err)
}
return channelID, nil
}
// IsPublicNode is a helper method that determines whether the node with the
// given public key is seen as a public node in the graph from the graph's
// source node's point of view.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) IsPublicNode(pubKey [33]byte) (bool, error) {
ctx := context.TODO()
var isPublic bool
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
var err error
isPublic, err = db.IsPublicV1Node(ctx, pubKey[:])
return err
}, sqldb.NoOpReset)
if err != nil {
return false, fmt.Errorf("unable to check if node is "+
"public: %w", err)
}
return isPublic, nil
}
// FetchChanInfos returns the set of channel edges that correspond to the passed
// channel ID's. If an edge is the query is unknown to the database, it will
// skipped and the result will contain only those edges that exist at the time
// of the query. This can be used to respond to peer queries that are seeking to
// fill in gaps in their view of the channel graph.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) FetchChanInfos(chanIDs []uint64) ([]ChannelEdge, error) {
var (
ctx = context.TODO()
edges []ChannelEdge
)
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
for _, chanID := range chanIDs {
var chanIDB [8]byte
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[:],
Version: int16(ProtocolV1),
},
)
if errors.Is(err, sql.ErrNoRows) {
continue
} else if err != nil {
return fmt.Errorf("unable to fetch channel: %w",
err)
}
node1, node2, err := buildNodes(
ctx, db, row.Node, row.Node_2,
)
if err != nil {
return fmt.Errorf("unable to fetch nodes: %w",
err)
}
edge, err := getAndBuildEdgeInfo(
ctx, db, s.cfg.ChainHash, row.Channel.ID,
row.Channel, node1.PubKeyBytes,
node2.PubKeyBytes,
)
if err != nil {
return fmt.Errorf("unable to build "+
"channel info: %w", err)
}
dbPol1, dbPol2, err := extractChannelPolicies(row)
if err != nil {
return fmt.Errorf("unable to extract channel "+
"policies: %w", err)
}
p1, p2, err := getAndBuildChanPolicies(
ctx, db, dbPol1, dbPol2, edge.ChannelID,
node1.PubKeyBytes, node2.PubKeyBytes,
)
if err != nil {
return fmt.Errorf("unable to build channel "+
"policies: %w", err)
}
edges = append(edges, ChannelEdge{
Info: edge,
Policy1: p1,
Policy2: p2,
Node1: node1,
Node2: node2,
})
}
return nil
}, func() {
edges = nil
})
if err != nil {
return nil, fmt.Errorf("unable to fetch channels: %w", err)
}
return edges, nil
}
// FilterKnownChanIDs takes a set of channel IDs and return the subset of chan
// ID's that we don't know and are not known zombies of the passed set. In other
// words, we perform a set difference of our set of chan ID's and the ones
// passed in. This method can be used by callers to determine the set of
// channels another peer knows of that we don't. The ChannelUpdateInfos for the
// known zombies is also returned.
//
// NOTE: part of the V1Store interface.
func (s *SQLStore) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo) ([]uint64,
[]ChannelUpdateInfo, error) {
var (
ctx = context.TODO()
newChanIDs []uint64
knownZombies []ChannelUpdateInfo
)
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)
// TODO(elle): potentially optimize this by using
// sqlc.slice() once that works for both SQLite and
// Postgres.
_, err := db.GetChannelBySCID(
ctx, sqlc.GetChannelBySCIDParams{
Version: int16(ProtocolV1),
Scid: chanIDB[:],
},
)
if err == nil {
continue
} else if !errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("unable to fetch channel: %w",
err)
}
isZombie, err := db.IsZombieChannel(
ctx, sqlc.IsZombieChannelParams{
Scid: chanIDB[:],
Version: int16(ProtocolV1),
},
)
if err != nil {
return fmt.Errorf("unable to fetch zombie "+
"channel: %w", err)
}
if isZombie {
knownZombies = append(knownZombies, chanInfo)
continue
}
newChanIDs = append(newChanIDs, channelID)
}
return nil
}, func() {
newChanIDs = nil
knownZombies = nil
})
if err != nil {
return nil, nil, fmt.Errorf("unable to fetch channels: %w", err)
}
return newChanIDs, knownZombies, nil
}
// 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
@@ -3066,12 +3595,92 @@ func buildNodes(ctx context.Context, db SQLQueries, dbNode1,
// information. It returns two policies, which may be nil if the policy
// information is not present in the row.
//
//nolint:ll,dupl
//nolint:ll,dupl,funlen
func extractChannelPolicies(row any) (*sqlc.ChannelPolicy, *sqlc.ChannelPolicy,
error) {
var policy1, policy2 *sqlc.ChannelPolicy
switch r := row.(type) {
case sqlc.GetChannelByOutpointWithPoliciesRow:
if r.Policy1ID.Valid {
policy1 = &sqlc.ChannelPolicy{
ID: r.Policy1ID.Int64,
Version: r.Policy1Version.Int16,
ChannelID: r.Channel.ID,
NodeID: r.Policy1NodeID.Int64,
Timelock: r.Policy1Timelock.Int32,
FeePpm: r.Policy1FeePpm.Int64,
BaseFeeMsat: r.Policy1BaseFeeMsat.Int64,
MinHtlcMsat: r.Policy1MinHtlcMsat.Int64,
MaxHtlcMsat: r.Policy1MaxHtlcMsat,
LastUpdate: r.Policy1LastUpdate,
InboundBaseFeeMsat: r.Policy1InboundBaseFeeMsat,
InboundFeeRateMilliMsat: r.Policy1InboundFeeRateMilliMsat,
Disabled: r.Policy1Disabled,
Signature: r.Policy1Signature,
}
}
if r.Policy2ID.Valid {
policy2 = &sqlc.ChannelPolicy{
ID: r.Policy2ID.Int64,
Version: r.Policy2Version.Int16,
ChannelID: r.Channel.ID,
NodeID: r.Policy2NodeID.Int64,
Timelock: r.Policy2Timelock.Int32,
FeePpm: r.Policy2FeePpm.Int64,
BaseFeeMsat: r.Policy2BaseFeeMsat.Int64,
MinHtlcMsat: r.Policy2MinHtlcMsat.Int64,
MaxHtlcMsat: r.Policy2MaxHtlcMsat,
LastUpdate: r.Policy2LastUpdate,
InboundBaseFeeMsat: r.Policy2InboundBaseFeeMsat,
InboundFeeRateMilliMsat: r.Policy2InboundFeeRateMilliMsat,
Disabled: r.Policy2Disabled,
Signature: r.Policy2Signature,
}
}
return policy1, policy2, nil
case sqlc.GetChannelBySCIDWithPoliciesRow:
if r.Policy1ID.Valid {
policy1 = &sqlc.ChannelPolicy{
ID: r.Policy1ID.Int64,
Version: r.Policy1Version.Int16,
ChannelID: r.Channel.ID,
NodeID: r.Policy1NodeID.Int64,
Timelock: r.Policy1Timelock.Int32,
FeePpm: r.Policy1FeePpm.Int64,
BaseFeeMsat: r.Policy1BaseFeeMsat.Int64,
MinHtlcMsat: r.Policy1MinHtlcMsat.Int64,
MaxHtlcMsat: r.Policy1MaxHtlcMsat,
LastUpdate: r.Policy1LastUpdate,
InboundBaseFeeMsat: r.Policy1InboundBaseFeeMsat,
InboundFeeRateMilliMsat: r.Policy1InboundFeeRateMilliMsat,
Disabled: r.Policy1Disabled,
Signature: r.Policy1Signature,
}
}
if r.Policy2ID.Valid {
policy2 = &sqlc.ChannelPolicy{
ID: r.Policy2ID.Int64,
Version: r.Policy2Version.Int16,
ChannelID: r.Channel.ID,
NodeID: r.Policy2NodeID.Int64,
Timelock: r.Policy2Timelock.Int32,
FeePpm: r.Policy2FeePpm.Int64,
BaseFeeMsat: r.Policy2BaseFeeMsat.Int64,
MinHtlcMsat: r.Policy2MinHtlcMsat.Int64,
MaxHtlcMsat: r.Policy2MaxHtlcMsat,
LastUpdate: r.Policy2LastUpdate,
InboundBaseFeeMsat: r.Policy2InboundBaseFeeMsat,
InboundFeeRateMilliMsat: r.Policy2InboundFeeRateMilliMsat,
Disabled: r.Policy2Disabled,
Signature: r.Policy2Signature,
}
}
return policy1, policy2, nil
case sqlc.GetChannelsByPolicyLastUpdateRangeRow:
if r.Policy1ID.Valid {
policy1 = &sqlc.ChannelPolicy{
+233
View File
@@ -263,6 +263,138 @@ func (q *Queries) GetChannelAndNodesBySCID(ctx context.Context, arg GetChannelAn
return i, err
}
const getChannelByOutpointWithPolicies = `-- name: GetChannelByOutpointWithPolicies :one
SELECT
c.id, c.version, c.scid, c.node_id_1, c.node_id_2, c.outpoint, c.capacity, c.bitcoin_key_1, c.bitcoin_key_2, c.node_1_signature, c.node_2_signature, c.bitcoin_1_signature, c.bitcoin_2_signature,
n1.pub_key AS node1_pubkey,
n2.pub_key AS node2_pubkey,
-- Node 1 policy
cp1.id AS policy_1_id,
cp1.node_id AS policy_1_node_id,
cp1.version AS policy_1_version,
cp1.timelock AS policy_1_timelock,
cp1.fee_ppm AS policy_1_fee_ppm,
cp1.base_fee_msat AS policy_1_base_fee_msat,
cp1.min_htlc_msat AS policy_1_min_htlc_msat,
cp1.max_htlc_msat AS policy_1_max_htlc_msat,
cp1.last_update AS policy_1_last_update,
cp1.disabled AS policy_1_disabled,
cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat,
cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat,
cp1.signature AS policy_1_signature,
-- Node 2 policy
cp2.id AS policy_2_id,
cp2.node_id AS policy_2_node_id,
cp2.version AS policy_2_version,
cp2.timelock AS policy_2_timelock,
cp2.fee_ppm AS policy_2_fee_ppm,
cp2.base_fee_msat AS policy_2_base_fee_msat,
cp2.min_htlc_msat AS policy_2_min_htlc_msat,
cp2.max_htlc_msat AS policy_2_max_htlc_msat,
cp2.last_update AS policy_2_last_update,
cp2.disabled AS policy_2_disabled,
cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat,
cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat,
cp2.signature AS policy_2_signature
FROM channels c
JOIN nodes n1 ON c.node_id_1 = n1.id
JOIN nodes n2 ON c.node_id_2 = n2.id
LEFT JOIN channel_policies cp1
ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version
LEFT JOIN channel_policies cp2
ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version
WHERE c.outpoint = $1 AND c.version = $2
`
type GetChannelByOutpointWithPoliciesParams struct {
Outpoint string
Version int16
}
type GetChannelByOutpointWithPoliciesRow struct {
Channel Channel
Node1Pubkey []byte
Node2Pubkey []byte
Policy1ID sql.NullInt64
Policy1NodeID sql.NullInt64
Policy1Version sql.NullInt16
Policy1Timelock sql.NullInt32
Policy1FeePpm sql.NullInt64
Policy1BaseFeeMsat sql.NullInt64
Policy1MinHtlcMsat sql.NullInt64
Policy1MaxHtlcMsat sql.NullInt64
Policy1LastUpdate sql.NullInt64
Policy1Disabled sql.NullBool
Policy1InboundBaseFeeMsat sql.NullInt64
Policy1InboundFeeRateMilliMsat sql.NullInt64
Policy1Signature []byte
Policy2ID sql.NullInt64
Policy2NodeID sql.NullInt64
Policy2Version sql.NullInt16
Policy2Timelock sql.NullInt32
Policy2FeePpm sql.NullInt64
Policy2BaseFeeMsat sql.NullInt64
Policy2MinHtlcMsat sql.NullInt64
Policy2MaxHtlcMsat sql.NullInt64
Policy2LastUpdate sql.NullInt64
Policy2Disabled sql.NullBool
Policy2InboundBaseFeeMsat sql.NullInt64
Policy2InboundFeeRateMilliMsat sql.NullInt64
Policy2Signature []byte
}
func (q *Queries) GetChannelByOutpointWithPolicies(ctx context.Context, arg GetChannelByOutpointWithPoliciesParams) (GetChannelByOutpointWithPoliciesRow, error) {
row := q.db.QueryRowContext(ctx, getChannelByOutpointWithPolicies, arg.Outpoint, arg.Version)
var i GetChannelByOutpointWithPoliciesRow
err := row.Scan(
&i.Channel.ID,
&i.Channel.Version,
&i.Channel.Scid,
&i.Channel.NodeID1,
&i.Channel.NodeID2,
&i.Channel.Outpoint,
&i.Channel.Capacity,
&i.Channel.BitcoinKey1,
&i.Channel.BitcoinKey2,
&i.Channel.Node1Signature,
&i.Channel.Node2Signature,
&i.Channel.Bitcoin1Signature,
&i.Channel.Bitcoin2Signature,
&i.Node1Pubkey,
&i.Node2Pubkey,
&i.Policy1ID,
&i.Policy1NodeID,
&i.Policy1Version,
&i.Policy1Timelock,
&i.Policy1FeePpm,
&i.Policy1BaseFeeMsat,
&i.Policy1MinHtlcMsat,
&i.Policy1MaxHtlcMsat,
&i.Policy1LastUpdate,
&i.Policy1Disabled,
&i.Policy1InboundBaseFeeMsat,
&i.Policy1InboundFeeRateMilliMsat,
&i.Policy1Signature,
&i.Policy2ID,
&i.Policy2NodeID,
&i.Policy2Version,
&i.Policy2Timelock,
&i.Policy2FeePpm,
&i.Policy2BaseFeeMsat,
&i.Policy2MinHtlcMsat,
&i.Policy2MaxHtlcMsat,
&i.Policy2LastUpdate,
&i.Policy2Disabled,
&i.Policy2InboundBaseFeeMsat,
&i.Policy2InboundFeeRateMilliMsat,
&i.Policy2Signature,
)
return i, 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
@@ -1035,6 +1167,23 @@ func (q *Queries) GetPublicV1ChannelsBySCID(ctx context.Context, arg GetPublicV1
return items, nil
}
const getSCIDByOutpoint = `-- name: GetSCIDByOutpoint :one
SELECT scid from channels
WHERE outpoint = $1 AND version = $2
`
type GetSCIDByOutpointParams struct {
Outpoint string
Version int16
}
func (q *Queries) GetSCIDByOutpoint(ctx context.Context, arg GetSCIDByOutpointParams) ([]byte, error) {
row := q.db.QueryRowContext(ctx, getSCIDByOutpoint, arg.Outpoint, arg.Version)
var scid []byte
err := row.Scan(&scid)
return scid, err
}
const getSourceNodesByVersion = `-- name: GetSourceNodesByVersion :many
SELECT sn.node_id, n.pub_key
FROM source_nodes sn
@@ -1070,6 +1219,43 @@ func (q *Queries) GetSourceNodesByVersion(ctx context.Context, version int16) ([
return items, nil
}
const getV1DisabledSCIDs = `-- name: GetV1DisabledSCIDs :many
SELECT c.scid
FROM channels c
JOIN channel_policies cp ON cp.channel_id = c.id
WHERE cp.disabled = true
AND c.version = 1
GROUP BY c.scid
HAVING COUNT(*) > 1
`
// NOTE: this is V1 specific since for V1, disabled is a
// simple, single boolean. The proposed V2 policy
// structure will have a more complex disabled bit vector
// and so the query for V2 may differ.
func (q *Queries) GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error) {
rows, err := q.db.QueryContext(ctx, getV1DisabledSCIDs)
if err != nil {
return nil, err
}
defer rows.Close()
var items [][]byte
for rows.Next() {
var scid []byte
if err := rows.Scan(&scid); err != nil {
return nil, err
}
items = append(items, scid)
}
if err := rows.Close(); err != nil {
return nil, err
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const getZombieChannel = `-- name: GetZombieChannel :one
SELECT scid, version, node_key_1, node_key_2
FROM zombie_channels
@@ -1211,6 +1397,53 @@ func (q *Queries) InsertNodeFeature(ctx context.Context, arg InsertNodeFeaturePa
return err
}
const isPublicV1Node = `-- name: IsPublicV1Node :one
SELECT EXISTS (
SELECT 1
FROM channels c
JOIN nodes n ON n.id = c.node_id_1 OR n.id = c.node_id_2
-- NOTE: we hard-code the version here since the clauses
-- here that determine if a node is public is specific
-- to the V1 gossip protocol. In V1, a node is public
-- if it has a public channel and a public channel is one
-- where we have the set of signatures of the channel
-- announcement. It is enough to just check that we have
-- one of the signatures since we only ever set them
-- together.
WHERE c.version = 1
AND c.bitcoin_1_signature IS NOT NULL
AND n.pub_key = $1
)
`
func (q *Queries) IsPublicV1Node(ctx context.Context, pubKey []byte) (bool, error) {
row := q.db.QueryRowContext(ctx, isPublicV1Node, pubKey)
var exists bool
err := row.Scan(&exists)
return exists, err
}
const isZombieChannel = `-- name: IsZombieChannel :one
SELECT EXISTS (
SELECT 1
FROM zombie_channels
WHERE scid = $1
AND version = $2
) AS is_zombie
`
type IsZombieChannelParams struct {
Scid []byte
Version int16
}
func (q *Queries) IsZombieChannel(ctx context.Context, arg IsZombieChannelParams) (bool, error) {
row := q.db.QueryRowContext(ctx, isZombieChannel, arg.Scid, arg.Version)
var is_zombie bool
err := row.Scan(&is_zombie)
return is_zombie, err
}
const listChannelsByNodeID = `-- name: ListChannelsByNodeID :many
SELECT c.id, c.version, c.scid, c.node_id_1, c.node_id_2, c.outpoint, c.capacity, c.bitcoin_key_1, c.bitcoin_key_2, c.node_1_signature, c.node_2_signature, c.bitcoin_1_signature, c.bitcoin_2_signature,
n1.pub_key AS node1_pubkey,
+9
View File
@@ -31,6 +31,7 @@ type Querier interface {
FilterInvoices(ctx context.Context, arg FilterInvoicesParams) ([]Invoice, error)
GetAMPInvoiceID(ctx context.Context, setID []byte) (int64, error)
GetChannelAndNodesBySCID(ctx context.Context, arg GetChannelAndNodesBySCIDParams) (GetChannelAndNodesBySCIDRow, error)
GetChannelByOutpointWithPolicies(ctx context.Context, arg GetChannelByOutpointWithPoliciesParams) (GetChannelByOutpointWithPoliciesRow, error)
GetChannelBySCID(ctx context.Context, arg GetChannelBySCIDParams) (Channel, error)
GetChannelBySCIDWithPolicies(ctx context.Context, arg GetChannelBySCIDWithPoliciesParams) (GetChannelBySCIDWithPoliciesRow, error)
GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]GetChannelFeaturesAndExtrasRow, error)
@@ -57,7 +58,13 @@ type Querier interface {
GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error)
GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]Node, error)
GetPublicV1ChannelsBySCID(ctx context.Context, arg GetPublicV1ChannelsBySCIDParams) ([]Channel, error)
GetSCIDByOutpoint(ctx context.Context, arg GetSCIDByOutpointParams) ([]byte, error)
GetSourceNodesByVersion(ctx context.Context, version int16) ([]GetSourceNodesByVersionRow, error)
// NOTE: this is V1 specific since for V1, disabled is a
// simple, single boolean. The proposed V2 policy
// structure will have a more complex disabled bit vector
// and so the query for V2 may differ.
GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error)
GetZombieChannel(ctx context.Context, arg GetZombieChannelParams) (ZombieChannel, error)
HighestSCID(ctx context.Context, version int16) ([]byte, error)
InsertAMPSubInvoice(ctx context.Context, arg InsertAMPSubInvoiceParams) error
@@ -72,6 +79,8 @@ 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
IsPublicV1Node(ctx context.Context, pubKey []byte) (bool, error)
IsZombieChannel(ctx context.Context, arg IsZombieChannelParams) (bool, error)
ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNodeIDParams) ([]ListChannelsByNodeIDRow, error)
ListChannelsWithPoliciesPaginated(ctx context.Context, arg ListChannelsWithPoliciesPaginatedParams) ([]ListChannelsWithPoliciesPaginatedRow, error)
ListNodeIDsAndPubKeys(ctx context.Context, arg ListNodeIDsAndPubKeysParams) ([]ListNodeIDsAndPubKeysRow, error)
+88
View File
@@ -47,6 +47,24 @@ WHERE version = $1 AND id > $2
ORDER BY id
LIMIT $3;
-- name: IsPublicV1Node :one
SELECT EXISTS (
SELECT 1
FROM channels c
JOIN nodes n ON n.id = c.node_id_1 OR n.id = c.node_id_2
-- NOTE: we hard-code the version here since the clauses
-- here that determine if a node is public is specific
-- to the V1 gossip protocol. In V1, a node is public
-- if it has a public channel and a public channel is one
-- where we have the set of signatures of the channel
-- announcement. It is enough to just check that we have
-- one of the signatures since we only ever set them
-- together.
WHERE c.version = 1
AND c.bitcoin_1_signature IS NOT NULL
AND n.pub_key = $1
);
-- name: DeleteNodeByPubKey :execresult
DELETE FROM nodes
WHERE pub_key = $1
@@ -206,6 +224,10 @@ SELECT
FROM channel_extra_types cet
WHERE cet.channel_id = $1;
-- name: GetSCIDByOutpoint :one
SELECT scid from channels
WHERE outpoint = $1 AND version = $2;
-- name: GetChannelsByPolicyLastUpdateRange :many
SELECT
sqlc.embed(c),
@@ -262,6 +284,51 @@ ORDER BY
ELSE COALESCE(cp2.last_update, 0)
END ASC;
-- name: GetChannelByOutpointWithPolicies :one
SELECT
sqlc.embed(c),
n1.pub_key AS node1_pubkey,
n2.pub_key AS node2_pubkey,
-- Node 1 policy
cp1.id AS policy_1_id,
cp1.node_id AS policy_1_node_id,
cp1.version AS policy_1_version,
cp1.timelock AS policy_1_timelock,
cp1.fee_ppm AS policy_1_fee_ppm,
cp1.base_fee_msat AS policy_1_base_fee_msat,
cp1.min_htlc_msat AS policy_1_min_htlc_msat,
cp1.max_htlc_msat AS policy_1_max_htlc_msat,
cp1.last_update AS policy_1_last_update,
cp1.disabled AS policy_1_disabled,
cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat,
cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat,
cp1.signature AS policy_1_signature,
-- Node 2 policy
cp2.id AS policy_2_id,
cp2.node_id AS policy_2_node_id,
cp2.version AS policy_2_version,
cp2.timelock AS policy_2_timelock,
cp2.fee_ppm AS policy_2_fee_ppm,
cp2.base_fee_msat AS policy_2_base_fee_msat,
cp2.min_htlc_msat AS policy_2_min_htlc_msat,
cp2.max_htlc_msat AS policy_2_max_htlc_msat,
cp2.last_update AS policy_2_last_update,
cp2.disabled AS policy_2_disabled,
cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat,
cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat,
cp2.signature AS policy_2_signature
FROM channels c
JOIN nodes n1 ON c.node_id_1 = n1.id
JOIN nodes n2 ON c.node_id_2 = n2.id
LEFT JOIN channel_policies cp1
ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version
LEFT JOIN channel_policies cp2
ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version
WHERE c.outpoint = $1 AND c.version = $2;
-- name: HighestSCID :one
SELECT scid
FROM channels
@@ -506,6 +573,19 @@ JOIN channel_policy_extra_types cpet
ON cp.id = cpet.channel_policy_id
WHERE cp.id = $1 OR cp.id = $2;
-- name: GetV1DisabledSCIDs :many
SELECT c.scid
FROM channels c
JOIN channel_policies cp ON cp.channel_id = c.id
-- NOTE: this is V1 specific since for V1, disabled is a
-- simple, single boolean. The proposed V2 policy
-- structure will have a more complex disabled bit vector
-- and so the query for V2 may differ.
WHERE cp.disabled = true
AND c.version = 1
GROUP BY c.scid
HAVING COUNT(*) > 1;
-- name: DeleteChannelPolicyExtraTypes :exec
DELETE FROM channel_policy_extra_types
WHERE channel_policy_id = $1;
@@ -540,3 +620,11 @@ SELECT *
FROM zombie_channels
WHERE scid = $1
AND version = $2;
-- name: IsZombieChannel :one
SELECT EXISTS (
SELECT 1
FROM zombie_channels
WHERE scid = $1
AND version = $2
) AS is_zombie;