mirror of
https://github.com/lightningnetwork/lnd.git
synced 2026-08-13 12:32:48 +02:00
8047149c6a
Migrate all btcd dependencies to the new per-package v2 modules (wire/v2, txscript/v2, chaincfg/v2, chainhash/v2, btcutil/v2, psbt/v2, btcec/v2) introduced by btcd v0.26.0, and pin the tagged ecosystem versions: btcwallet v0.17.0, neutrino v0.18.0 and lightning-onion v1.4.0. The bulk of the import rewrite was produced by the scripted diff from https://github.com/btcsuite/btcd/pull/2547 (followed by 'make rpc'). The address symbols that moved out of btcutil into the new address package are imported as btcaddr where a local "address" variable would otherwise shadow them. The go.mod/go.sum updates and the remaining manual compilation fixes are folded into this single commit so it builds on its own (the migration was previously split into a reproducible scripted-diff plus follow-ups, intended to be squashed on merge).
6987 lines
197 KiB
Go
6987 lines
197 KiB
Go
package graphdb
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
color "image/color"
|
|
"iter"
|
|
"maps"
|
|
"math"
|
|
"net"
|
|
"slices"
|
|
"strconv"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/btcsuite/btcd/btcec/v2"
|
|
"github.com/btcsuite/btcd/btcutil/v2"
|
|
"github.com/btcsuite/btcd/chainhash/v2"
|
|
"github.com/btcsuite/btcd/wire/v2"
|
|
"github.com/lightningnetwork/lnd/aliasmgr"
|
|
"github.com/lightningnetwork/lnd/batch"
|
|
"github.com/lightningnetwork/lnd/fn/v2"
|
|
"github.com/lightningnetwork/lnd/graph/db/models"
|
|
"github.com/lightningnetwork/lnd/lnwire"
|
|
"github.com/lightningnetwork/lnd/routing/route"
|
|
"github.com/lightningnetwork/lnd/sqldb"
|
|
"github.com/lightningnetwork/lnd/sqldb/sqlc"
|
|
"github.com/lightningnetwork/lnd/tlv"
|
|
"github.com/lightningnetwork/lnd/tor"
|
|
)
|
|
|
|
const (
|
|
gossipV1 = lnwire.GossipVersion1
|
|
gossipV2 = lnwire.GossipVersion2
|
|
)
|
|
|
|
// SQLQueries is a subset of the sqlc.Querier interface that can be used to
|
|
// execute queries against the SQL graph tables.
|
|
//
|
|
//nolint:ll,interfacebloat
|
|
type SQLQueries interface {
|
|
/*
|
|
Node queries.
|
|
*/
|
|
UpsertNode(ctx context.Context, arg sqlc.UpsertNodeParams) (int64, error)
|
|
UpsertSourceNode(ctx context.Context, arg sqlc.UpsertSourceNodeParams) (int64, error)
|
|
GetNodeByPubKey(ctx context.Context, arg sqlc.GetNodeByPubKeyParams) (sqlc.GraphNode, error)
|
|
GetNodesByIDs(ctx context.Context, ids []int64) ([]sqlc.GraphNode, error)
|
|
GetNodeIDByPubKey(ctx context.Context, arg sqlc.GetNodeIDByPubKeyParams) (int64, error)
|
|
GetNodesByLastUpdateRange(ctx context.Context, arg sqlc.GetNodesByLastUpdateRangeParams) ([]sqlc.GraphNode, error)
|
|
GetNodesByBlockHeightRange(ctx context.Context, arg sqlc.GetNodesByBlockHeightRangeParams) ([]sqlc.GraphNode, error)
|
|
GetPublicNodesByLastUpdateRange(ctx context.Context, arg sqlc.GetPublicNodesByLastUpdateRangeParams) ([]sqlc.GraphNode, error)
|
|
ListNodesPaginated(ctx context.Context, arg sqlc.ListNodesPaginatedParams) ([]sqlc.GraphNode, error)
|
|
ListNodeIDsAndPubKeys(ctx context.Context, arg sqlc.ListNodeIDsAndPubKeysParams) ([]sqlc.ListNodeIDsAndPubKeysRow, error)
|
|
IsPublicV1Node(ctx context.Context, pubKey []byte) (bool, error)
|
|
IsPublicV2Node(ctx context.Context, pubKey []byte) (bool, error)
|
|
DeleteUnconnectedNodes(ctx context.Context) ([][]byte, error)
|
|
DeleteNodeByPubKey(ctx context.Context, arg sqlc.DeleteNodeByPubKeyParams) (sql.Result, error)
|
|
DeleteNode(ctx context.Context, id int64) error
|
|
NodeExists(ctx context.Context, arg sqlc.NodeExistsParams) (bool, error)
|
|
|
|
GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.GraphNodeExtraType, error)
|
|
GetNodeExtraTypesBatch(ctx context.Context, ids []int64) ([]sqlc.GraphNodeExtraType, error)
|
|
UpsertNodeExtraType(ctx context.Context, arg sqlc.UpsertNodeExtraTypeParams) error
|
|
DeleteExtraNodeType(ctx context.Context, arg sqlc.DeleteExtraNodeTypeParams) error
|
|
|
|
UpsertNodeAddress(ctx context.Context, arg sqlc.UpsertNodeAddressParams) error
|
|
GetNodeAddresses(ctx context.Context, nodeID int64) ([]sqlc.GetNodeAddressesRow, error)
|
|
GetNodeAddressesBatch(ctx context.Context, ids []int64) ([]sqlc.GraphNodeAddress, error)
|
|
DeleteNodeAddresses(ctx context.Context, nodeID int64) error
|
|
|
|
InsertNodeFeature(ctx context.Context, arg sqlc.InsertNodeFeatureParams) error
|
|
GetNodeFeatures(ctx context.Context, nodeID int64) ([]sqlc.GraphNodeFeature, error)
|
|
GetNodeFeaturesBatch(ctx context.Context, ids []int64) ([]sqlc.GraphNodeFeature, error)
|
|
GetNodeFeaturesByPubKey(ctx context.Context, arg sqlc.GetNodeFeaturesByPubKeyParams) ([]int32, error)
|
|
DeleteNodeFeature(ctx context.Context, arg sqlc.DeleteNodeFeatureParams) error
|
|
GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error)
|
|
GetV2DisabledSCIDs(ctx context.Context) ([][]byte, error)
|
|
|
|
/*
|
|
Source node queries.
|
|
*/
|
|
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)
|
|
AddV1ChannelProof(ctx context.Context, arg sqlc.AddV1ChannelProofParams) (sql.Result, error)
|
|
AddV2ChannelProof(ctx context.Context, arg sqlc.AddV2ChannelProofParams) (sql.Result, error)
|
|
GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.GraphChannel, error)
|
|
GetChannelsBySCIDs(ctx context.Context, arg sqlc.GetChannelsBySCIDsParams) ([]sqlc.GraphChannel, error)
|
|
GetChannelsByOutpoints(ctx context.Context, outpoints []string) ([]sqlc.GetChannelsByOutpointsRow, error)
|
|
GetChannelsBySCIDRange(ctx context.Context, arg sqlc.GetChannelsBySCIDRangeParams) ([]sqlc.GetChannelsBySCIDRangeRow, error)
|
|
GetChannelBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelBySCIDWithPoliciesParams) (sqlc.GetChannelBySCIDWithPoliciesRow, error)
|
|
GetChannelsBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelsBySCIDWithPoliciesParams) ([]sqlc.GetChannelsBySCIDWithPoliciesRow, error)
|
|
GetChannelsByIDs(ctx context.Context, ids []int64) ([]sqlc.GetChannelsByIDsRow, error)
|
|
GetChannelAndNodesBySCID(ctx context.Context, arg sqlc.GetChannelAndNodesBySCIDParams) (sqlc.GetChannelAndNodesBySCIDRow, error)
|
|
HighestSCID(ctx context.Context, version int16) ([]byte, error)
|
|
ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error)
|
|
ListChannelsForNodeIDs(ctx context.Context, arg sqlc.ListChannelsForNodeIDsParams) ([]sqlc.ListChannelsForNodeIDsRow, error)
|
|
ListChannelsWithPoliciesPaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesPaginatedParams) ([]sqlc.ListChannelsWithPoliciesPaginatedRow, error)
|
|
ListChannelsWithPoliciesForCachePaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesForCachePaginatedParams) ([]sqlc.ListChannelsWithPoliciesForCachePaginatedRow, error)
|
|
ListChannelsPaginated(ctx context.Context, arg sqlc.ListChannelsPaginatedParams) ([]sqlc.ListChannelsPaginatedRow, error)
|
|
ListChannelsPaginatedV2(ctx context.Context, arg sqlc.ListChannelsPaginatedV2Params) ([]sqlc.ListChannelsPaginatedV2Row, error)
|
|
GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg sqlc.GetChannelsByPolicyLastUpdateRangeParams) ([]sqlc.GetChannelsByPolicyLastUpdateRangeRow, error)
|
|
GetChannelsByPolicyBlockRange(ctx context.Context, arg sqlc.GetChannelsByPolicyBlockRangeParams) ([]sqlc.GetChannelsByPolicyBlockRangeRow, error)
|
|
GetChannelByOutpointWithPolicies(ctx context.Context, arg sqlc.GetChannelByOutpointWithPoliciesParams) (sqlc.GetChannelByOutpointWithPoliciesRow, error)
|
|
GetPublicV1ChannelsBySCID(ctx context.Context, arg sqlc.GetPublicV1ChannelsBySCIDParams) ([]sqlc.GraphChannel, error)
|
|
GetPublicV2ChannelsBySCID(ctx context.Context, arg sqlc.GetPublicV2ChannelsBySCIDParams) ([]sqlc.GraphChannel, error)
|
|
GetSCIDByOutpoint(ctx context.Context, arg sqlc.GetSCIDByOutpointParams) ([]byte, error)
|
|
DeleteChannels(ctx context.Context, ids []int64) error
|
|
|
|
UpsertChannelExtraType(ctx context.Context, arg sqlc.UpsertChannelExtraTypeParams) error
|
|
GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([]sqlc.GraphChannelExtraType, error)
|
|
InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error
|
|
GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]sqlc.GraphChannelFeature, error)
|
|
|
|
/*
|
|
Channel Policy table queries.
|
|
*/
|
|
UpsertEdgePolicy(ctx context.Context, arg sqlc.UpsertEdgePolicyParams) (int64, error)
|
|
GetChannelPolicyByChannelAndNode(ctx context.Context, arg sqlc.GetChannelPolicyByChannelAndNodeParams) (sqlc.GraphChannelPolicy, error)
|
|
|
|
UpsertChanPolicyExtraType(ctx context.Context, arg sqlc.UpsertChanPolicyExtraTypeParams) error
|
|
GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]sqlc.GetChannelPolicyExtraTypesBatchRow, error)
|
|
DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error
|
|
|
|
/*
|
|
Zombie index queries.
|
|
*/
|
|
UpsertZombieChannel(ctx context.Context, arg sqlc.UpsertZombieChannelParams) error
|
|
GetZombieChannel(ctx context.Context, arg sqlc.GetZombieChannelParams) (sqlc.GraphZombieChannel, error)
|
|
GetZombieChannelsSCIDs(ctx context.Context, arg sqlc.GetZombieChannelsSCIDsParams) ([]sqlc.GraphZombieChannel, 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)
|
|
|
|
/*
|
|
Prune log table queries.
|
|
*/
|
|
GetPruneTip(ctx context.Context) (sqlc.GraphPruneLog, error)
|
|
GetPruneHashByHeight(ctx context.Context, blockHeight int64) ([]byte, error)
|
|
GetPruneEntriesForHeights(ctx context.Context, heights []int64) ([]sqlc.GraphPruneLog, 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)
|
|
GetClosedChannelsSCIDs(ctx context.Context, scids [][]byte) ([][]byte, error)
|
|
|
|
/*
|
|
Migration specific queries.
|
|
|
|
NOTE: these should not be used in code other than migrations.
|
|
Once sqldbv2 is in place, these can be removed from this struct
|
|
as then migrations will have their own dedicated queries
|
|
structs.
|
|
*/
|
|
InsertNodeMig(ctx context.Context, arg sqlc.InsertNodeMigParams) (int64, error)
|
|
InsertChannelMig(ctx context.Context, arg sqlc.InsertChannelMigParams) (int64, error)
|
|
InsertEdgePolicyMig(ctx context.Context, arg sqlc.InsertEdgePolicyMigParams) (int64, error)
|
|
}
|
|
|
|
// BatchedSQLQueries is a version of SQLQueries that's capable of batched
|
|
// database operations.
|
|
type BatchedSQLQueries interface {
|
|
SQLQueries
|
|
sqldb.BatchedTx[SQLQueries]
|
|
}
|
|
|
|
// SQLStore is an implementation of the Store interface that uses a SQL
|
|
// database as the backend.
|
|
type SQLStore struct {
|
|
cfg *SQLStoreConfig
|
|
db BatchedSQLQueries
|
|
|
|
// cacheMu guards all caches (rejectCache and chanCache). If
|
|
// this mutex will be acquired at the same time as the DB mutex then
|
|
// the cacheMu MUST be acquired first to prevent deadlock.
|
|
cacheMu sync.RWMutex
|
|
rejectCache *rejectCache
|
|
chanCache *channelCache
|
|
|
|
chanScheduler batch.Scheduler[SQLQueries]
|
|
nodeScheduler batch.Scheduler[SQLQueries]
|
|
|
|
srcNodes map[lnwire.GossipVersion]*srcNodeInfo
|
|
srcNodeMu sync.Mutex
|
|
}
|
|
|
|
// A compile-time assertion to ensure that SQLStore implements the Store
|
|
// interface.
|
|
var _ Store = (*SQLStore)(nil)
|
|
|
|
// SQLStoreConfig holds the configuration for the SQLStore.
|
|
type SQLStoreConfig struct {
|
|
// ChainHash is the genesis hash for the chain that all the gossip
|
|
// messages in this store are aimed at.
|
|
ChainHash chainhash.Hash
|
|
|
|
// QueryConfig holds configuration values for SQL queries.
|
|
QueryCfg *sqldb.QueryConfig
|
|
}
|
|
|
|
// NewSQLStore creates a new SQLStore instance given an open BatchedSQLQueries
|
|
// storage backend.
|
|
func NewSQLStore(cfg *SQLStoreConfig, db BatchedSQLQueries,
|
|
options ...StoreOptionModifier) (*SQLStore, error) {
|
|
|
|
opts := DefaultOptions()
|
|
for _, o := range options {
|
|
o(opts)
|
|
}
|
|
|
|
if opts.NoMigration {
|
|
return nil, fmt.Errorf("the NoMigration option is not yet " +
|
|
"supported for SQL stores")
|
|
}
|
|
|
|
s := &SQLStore{
|
|
cfg: cfg,
|
|
db: db,
|
|
rejectCache: newRejectCache(opts.RejectCacheSize),
|
|
chanCache: newChannelCache(opts.ChannelCacheSize),
|
|
srcNodes: make(map[lnwire.GossipVersion]*srcNodeInfo),
|
|
}
|
|
|
|
s.chanScheduler = batch.NewTimeScheduler(
|
|
db, &s.cacheMu, opts.BatchCommitInterval,
|
|
)
|
|
s.nodeScheduler = batch.NewTimeScheduler(
|
|
db, nil, opts.BatchCommitInterval,
|
|
)
|
|
|
|
return s, nil
|
|
}
|
|
|
|
// AddNode adds a vertex/node to the graph database. If the node is not
|
|
// in the database from before, this will add a new, unconnected one to the
|
|
// graph. If it is present from before, this will update that node's
|
|
// information.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) AddNode(ctx context.Context,
|
|
node *models.Node, opts ...batch.SchedulerOption) error {
|
|
|
|
r := &batch.Request[SQLQueries]{
|
|
Opts: batch.NewSchedulerOptions(opts...),
|
|
Do: func(queries SQLQueries) error {
|
|
_, err := upsertNode(ctx, queries, node)
|
|
|
|
// It is possible that two of the same node
|
|
// announcements are both being processed in the same
|
|
// batch. This may case the UpsertNode conflict to
|
|
// be hit since we require at the db layer that the
|
|
// new last_update is greater than the existing
|
|
// last_update. We need to gracefully handle this here.
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
}
|
|
|
|
return err
|
|
},
|
|
}
|
|
|
|
return s.nodeScheduler.Execute(ctx, r)
|
|
}
|
|
|
|
// FetchNode attempts to look up a target node by its identity public
|
|
// key. If the node isn't found in the database, then ErrGraphNodeNotFound is
|
|
// returned.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) FetchNode(ctx context.Context, v lnwire.GossipVersion,
|
|
pubKey route.Vertex) (*models.Node, error) {
|
|
|
|
var node *models.Node
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
var err error
|
|
_, node, err = getNodeByPubKey(
|
|
ctx, s.cfg.QueryCfg, db, v, pubKey,
|
|
)
|
|
|
|
return err
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to fetch node: %w", err)
|
|
}
|
|
|
|
return node, nil
|
|
}
|
|
|
|
// HasV1Node determines if the graph has a vertex identified by the
|
|
// target node identity public key. If the node exists in the database, a
|
|
// timestamp of when the data for the node was lasted updated is returned along
|
|
// with a true boolean. Otherwise, an empty time.Time is returned with a false
|
|
// boolean.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) HasV1Node(ctx context.Context,
|
|
pubKey [33]byte) (time.Time, bool, error) {
|
|
|
|
var (
|
|
exists bool
|
|
lastUpdate time.Time
|
|
)
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
dbNode, err := db.GetNodeByPubKey(
|
|
ctx, sqlc.GetNodeByPubKeyParams{
|
|
Version: int16(gossipV1),
|
|
PubKey: pubKey[:],
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
} else if err != nil {
|
|
return fmt.Errorf("unable to fetch node: %w", err)
|
|
}
|
|
|
|
exists = true
|
|
|
|
if dbNode.LastUpdate.Valid {
|
|
lastUpdate = time.Unix(dbNode.LastUpdate.Int64, 0)
|
|
}
|
|
|
|
return nil
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return time.Time{}, false,
|
|
fmt.Errorf("unable to fetch node: %w", err)
|
|
}
|
|
|
|
return lastUpdate, exists, nil
|
|
}
|
|
|
|
// HasNode determines if the graph has a vertex identified by the
|
|
// target node identity public key.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) HasNode(ctx context.Context, v lnwire.GossipVersion,
|
|
pubKey [33]byte) (bool, error) {
|
|
|
|
var exists bool
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
var err error
|
|
exists, err = db.NodeExists(ctx, sqlc.NodeExistsParams{
|
|
Version: int16(v),
|
|
PubKey: pubKey[:],
|
|
})
|
|
|
|
return err
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return false, fmt.Errorf("unable to check if node (%x) "+
|
|
"exists: %w", pubKey, err)
|
|
}
|
|
|
|
return exists, nil
|
|
}
|
|
|
|
// AddrsForNode returns all known addresses for the target node public key
|
|
// that the graph DB is aware of. The returned boolean indicates if the
|
|
// given node is unknown to the graph DB or not.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) AddrsForNode(ctx context.Context, v lnwire.GossipVersion,
|
|
nodePub *btcec.PublicKey) (bool, []net.Addr, error) {
|
|
|
|
var (
|
|
addresses []net.Addr
|
|
known bool
|
|
)
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
// First, check if the node exists and get its DB ID if it
|
|
// does.
|
|
dbID, err := db.GetNodeIDByPubKey(
|
|
ctx, sqlc.GetNodeIDByPubKeyParams{
|
|
Version: int16(v),
|
|
PubKey: nodePub.SerializeCompressed(),
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
}
|
|
|
|
known = true
|
|
|
|
addresses, err = getNodeAddresses(ctx, db, dbID)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch node addresses: %w",
|
|
err)
|
|
}
|
|
|
|
return nil
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return false, nil, fmt.Errorf("unable to get addresses for "+
|
|
"node(%x): %w", nodePub.SerializeCompressed(), err)
|
|
}
|
|
|
|
return known, addresses, nil
|
|
}
|
|
|
|
// DeleteNode starts a new database transaction to remove a vertex/node
|
|
// from the database according to the node's public key.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) DeleteNode(ctx context.Context, v lnwire.GossipVersion,
|
|
pubKey route.Vertex) error {
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
|
res, err := db.DeleteNodeByPubKey(
|
|
ctx, sqlc.DeleteNodeByPubKeyParams{
|
|
Version: int16(v),
|
|
PubKey: pubKey[:],
|
|
},
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rows, err := res.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if rows == 0 {
|
|
return ErrGraphNodeNotFound
|
|
} else if rows > 1 {
|
|
return fmt.Errorf("deleted %d rows, expected 1", rows)
|
|
}
|
|
|
|
return err
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to delete node: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// FetchNodeFeatures returns the features of the given node. If no features are
|
|
// known for the node, an empty feature vector is returned.
|
|
//
|
|
// NOTE: this is part of the graphdb.NodeTraverser interface.
|
|
func (s *SQLStore) FetchNodeFeatures(ctx context.Context,
|
|
v lnwire.GossipVersion, nodePub route.Vertex) (*lnwire.FeatureVector,
|
|
error) {
|
|
|
|
return fetchNodeFeatures(ctx, s.db, v, 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 Store interface.
|
|
func (s *SQLStore) DisabledChannelIDs(
|
|
ctx context.Context, v lnwire.GossipVersion) ([]uint64, error) {
|
|
|
|
var (
|
|
chanIDs []uint64
|
|
)
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
var (
|
|
dbChanIDs [][]byte
|
|
err error
|
|
)
|
|
switch v {
|
|
case gossipV1:
|
|
dbChanIDs, err = db.GetV1DisabledSCIDs(ctx)
|
|
case gossipV2:
|
|
dbChanIDs, err = db.GetV2DisabledSCIDs(ctx)
|
|
default:
|
|
return fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
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 Store interface.
|
|
func (s *SQLStore) LookupAlias(ctx context.Context, v lnwire.GossipVersion,
|
|
pub *btcec.PublicKey) (string, error) {
|
|
|
|
var alias string
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
dbNode, err := db.GetNodeByPubKey(
|
|
ctx, sqlc.GetNodeByPubKeyParams{
|
|
Version: int16(v),
|
|
PubKey: pub.SerializeCompressed(),
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return ErrNodeAliasNotFound
|
|
} else if err != nil {
|
|
return fmt.Errorf("unable to fetch node: %w", err)
|
|
}
|
|
|
|
if !dbNode.Alias.Valid {
|
|
return ErrNodeAliasNotFound
|
|
}
|
|
|
|
alias = dbNode.Alias.String
|
|
|
|
return nil
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return "", fmt.Errorf("unable to look up alias: %w", err)
|
|
}
|
|
|
|
return alias, nil
|
|
}
|
|
|
|
// SourceNode 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.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) SourceNode(ctx context.Context,
|
|
v lnwire.GossipVersion) (*models.Node, error) {
|
|
|
|
var node *models.Node
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
_, nodePub, err := s.getSourceNode(ctx, db, v)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch source node: %w",
|
|
err)
|
|
}
|
|
|
|
_, node, err = getNodeByPubKey(
|
|
ctx, s.cfg.QueryCfg, db, v, nodePub,
|
|
)
|
|
|
|
return err
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to fetch source node: %w", err)
|
|
}
|
|
|
|
return node, nil
|
|
}
|
|
|
|
// SetSourceNode sets the source node within the graph database. The source
|
|
// node is to be used as the center of a star-graph within path finding
|
|
// algorithms.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) SetSourceNode(ctx context.Context,
|
|
node *models.Node) error {
|
|
|
|
return s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
|
// For the source node, we use a less strict upsert that allows
|
|
// updates even when the timestamp hasn't changed. This handles
|
|
// the race condition where multiple goroutines (e.g.,
|
|
// setSelfNode, createNewHiddenService, RPC updates) read the
|
|
// same old timestamp, independently increment it, and try to
|
|
// write concurrently. We want all parameter changes to persist,
|
|
// even if timestamps collide.
|
|
id, err := upsertSourceNode(ctx, db, node)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to upsert source node: %w",
|
|
err)
|
|
}
|
|
|
|
// Make sure that if a source node for this version is already
|
|
// set, then the ID is the same as the one we are about to set.
|
|
dbSourceNodeID, _, err := s.getSourceNode(
|
|
ctx, db, node.Version,
|
|
)
|
|
if err != nil && !errors.Is(err, ErrSourceNodeNotSet) {
|
|
return fmt.Errorf("unable to fetch source node: %w",
|
|
err)
|
|
} else if err == nil {
|
|
if dbSourceNodeID != id {
|
|
return fmt.Errorf("v1 source node already "+
|
|
"set to a different node: %d vs %d",
|
|
dbSourceNodeID, id)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
return db.AddSourceNode(ctx, id)
|
|
}, sqldb.NoOpReset)
|
|
}
|
|
|
|
// NodeUpdatesInHorizon returns all the known lightning nodes which have
|
|
// updates within the passed range for the given gossip version. This method
|
|
// can be used by two nodes to quickly determine if they have the same set of
|
|
// up-to-date node announcements.
|
|
//
|
|
// NOTE: This is part of the Store interface.
|
|
func (s *SQLStore) NodeUpdatesInHorizon(ctx context.Context,
|
|
v lnwire.GossipVersion, r NodeUpdateRange,
|
|
opts ...IteratorOption) iter.Seq2[*models.Node, error] {
|
|
|
|
if err := r.validateForVersion(v); err != nil {
|
|
return func(yield func(*models.Node, error) bool) {
|
|
_ = yield(nil, err)
|
|
}
|
|
}
|
|
|
|
cfg := defaultIteratorConfig()
|
|
for _, opt := range opts {
|
|
opt(cfg)
|
|
}
|
|
|
|
switch v {
|
|
case gossipV1:
|
|
return s.nodeUpdatesInHorizonV1(ctx, r, cfg)
|
|
|
|
case gossipV2:
|
|
return s.nodeUpdatesInHorizonV2(ctx, r, cfg)
|
|
|
|
default:
|
|
err := fmt.Errorf("unknown gossip version: %v", v)
|
|
return func(yield func(*models.Node, error) bool) {
|
|
_ = yield(nil, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// nodeUpdatesInHorizonV1 implements the v1 time-based node horizon query.
|
|
func (s *SQLStore) nodeUpdatesInHorizonV1(ctx context.Context,
|
|
r NodeUpdateRange,
|
|
cfg *iterConfig) iter.Seq2[*models.Node, error] {
|
|
|
|
startTime := r.StartTime.UnwrapOr(time.Time{})
|
|
endTime := r.EndTime.UnwrapOr(time.Time{})
|
|
|
|
return func(yield func(*models.Node, error) bool) {
|
|
var (
|
|
lastUpdateTime sql.NullInt64
|
|
lastPubKey = make([]byte, 33)
|
|
hasMore = true
|
|
)
|
|
|
|
// Each iteration, we'll read a batch amount of nodes, yield
|
|
// them, then decide is we have more or not.
|
|
for hasMore {
|
|
var batch []*models.Node
|
|
|
|
//nolint:ll
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
rows, err := nodesByLastUpdateRange(
|
|
ctx, db, cfg, startTime,
|
|
endTime, lastUpdateTime,
|
|
lastPubKey,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
hasMore = len(rows) == cfg.nodeUpdateIterBatchSize
|
|
|
|
err = forEachNodeInBatch(
|
|
ctx, s.cfg.QueryCfg, db, rows,
|
|
func(_ int64, node *models.Node) error {
|
|
batch = append(batch, node)
|
|
|
|
// Update pagination cursors
|
|
// based on the last processed
|
|
// node.
|
|
lastUpdateTime = sql.NullInt64{
|
|
Int64: node.LastUpdate.
|
|
Unix(),
|
|
Valid: true,
|
|
}
|
|
lastPubKey = node.PubKeyBytes[:]
|
|
|
|
return nil
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to build "+
|
|
"nodes: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}, func() {
|
|
batch = []*models.Node{}
|
|
})
|
|
|
|
if err != nil {
|
|
log.Errorf("NodeUpdatesInHorizon batch "+
|
|
"error: %v", err)
|
|
|
|
yield(&models.Node{}, err)
|
|
|
|
return
|
|
}
|
|
|
|
for _, node := range batch {
|
|
if !yield(node, nil) {
|
|
return
|
|
}
|
|
}
|
|
|
|
// If the batch didn't yield anything, then we're done.
|
|
if len(batch) == 0 {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// nodeUpdatesInHorizonV2 implements the v2 block-height-based node horizon
|
|
// query.
|
|
func (s *SQLStore) nodeUpdatesInHorizonV2(ctx context.Context,
|
|
r NodeUpdateRange,
|
|
cfg *iterConfig) iter.Seq2[*models.Node, error] {
|
|
|
|
startHeight := int64(r.StartHeight.UnwrapOr(0))
|
|
endHeight := int64(r.EndHeight.UnwrapOr(0))
|
|
batchSize := cfg.nodeUpdateIterBatchSize
|
|
|
|
return func(yield func(*models.Node, error) bool) {
|
|
var (
|
|
lastBlock sql.NullInt64
|
|
lastPubKey = make([]byte, 33)
|
|
hasMore = true
|
|
)
|
|
|
|
// queryNodes fetches the next page of v2 nodes in the
|
|
// block-height range.
|
|
queryNodes := func(db SQLQueries) ([]sqlc.GraphNode, error) {
|
|
return db.GetNodesByBlockHeightRange(
|
|
ctx, sqlc.GetNodesByBlockHeightRangeParams{
|
|
Version: int16(gossipV2),
|
|
StartHeight: sqldb.SQLInt64(
|
|
startHeight,
|
|
),
|
|
EndHeight: sqldb.SQLInt64(
|
|
endHeight,
|
|
),
|
|
LastBlockHeight: lastBlock,
|
|
LastPubKey: lastPubKey,
|
|
OnlyPublic: sql.NullBool{
|
|
Bool: cfg.iterPublicNodes,
|
|
Valid: true,
|
|
},
|
|
MaxResults: sqldb.SQLInt32(batchSize),
|
|
},
|
|
)
|
|
}
|
|
|
|
// processNode accumulates a node into the batch and
|
|
// advances the pagination cursors.
|
|
processNode := func(node *models.Node,
|
|
batch *[]*models.Node) error {
|
|
|
|
*batch = append(*batch, node)
|
|
|
|
lastBlock = sql.NullInt64{
|
|
Int64: int64(node.LastBlockHeight),
|
|
Valid: true,
|
|
}
|
|
lastPubKey = node.PubKeyBytes[:]
|
|
|
|
return nil
|
|
}
|
|
|
|
for hasMore {
|
|
var batch []*models.Node
|
|
|
|
err := s.db.ExecTx(
|
|
ctx, sqldb.ReadTxOpt(),
|
|
func(db SQLQueries) error {
|
|
rows, err := queryNodes(db)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
hasMore = len(rows) == batchSize
|
|
|
|
return forEachNodeInBatch(
|
|
ctx, s.cfg.QueryCfg, db,
|
|
rows, func(_ int64,
|
|
n *models.Node) error {
|
|
|
|
return processNode(
|
|
n, &batch,
|
|
)
|
|
},
|
|
)
|
|
}, func() {
|
|
batch = nil
|
|
},
|
|
)
|
|
if err != nil {
|
|
log.Errorf("NodeUpdatesInHorizon(v2) "+
|
|
"batch error: %v", err)
|
|
|
|
yield(nil, err)
|
|
|
|
return
|
|
}
|
|
|
|
for _, node := range batch {
|
|
if !yield(node, nil) {
|
|
return
|
|
}
|
|
}
|
|
|
|
if len(batch) == 0 {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// nodesByLastUpdateRange dispatches to either the all-nodes or public-only
|
|
// variant of the v1 node horizon query based on the iterator config.
|
|
func nodesByLastUpdateRange(ctx context.Context, db SQLQueries,
|
|
cfg *iterConfig, startTime, endTime time.Time,
|
|
lastUpdateTime sql.NullInt64,
|
|
lastPubKey []byte) ([]sqlc.GraphNode, error) {
|
|
|
|
if cfg.iterPublicNodes {
|
|
return db.GetPublicNodesByLastUpdateRange(
|
|
ctx, sqlc.GetPublicNodesByLastUpdateRangeParams{
|
|
StartTime: sqldb.SQLInt64(
|
|
startTime.Unix(),
|
|
),
|
|
EndTime: sqldb.SQLInt64(
|
|
endTime.Unix(),
|
|
),
|
|
LastUpdate: lastUpdateTime,
|
|
LastPubKey: lastPubKey,
|
|
MaxResults: sqldb.SQLInt32(
|
|
cfg.nodeUpdateIterBatchSize,
|
|
),
|
|
},
|
|
)
|
|
}
|
|
|
|
return db.GetNodesByLastUpdateRange(
|
|
ctx, sqlc.GetNodesByLastUpdateRangeParams{
|
|
StartTime: sqldb.SQLInt64(
|
|
startTime.Unix(),
|
|
),
|
|
EndTime: sqldb.SQLInt64(
|
|
endTime.Unix(),
|
|
),
|
|
LastUpdate: lastUpdateTime,
|
|
LastPubKey: lastPubKey,
|
|
MaxResults: sqldb.SQLInt32(
|
|
cfg.nodeUpdateIterBatchSize,
|
|
),
|
|
},
|
|
)
|
|
}
|
|
|
|
// 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 Store interface.
|
|
func (s *SQLStore) AddChannelEdge(ctx context.Context,
|
|
edge *models.ChannelEdgeInfo, opts ...batch.SchedulerOption) error {
|
|
|
|
if !isKnownGossipVersion(edge.Version) {
|
|
return fmt.Errorf("unsupported gossip version: %d",
|
|
edge.Version)
|
|
}
|
|
|
|
var alreadyExists bool
|
|
r := &batch.Request[SQLQueries]{
|
|
Opts: batch.NewSchedulerOptions(opts...),
|
|
Reset: func() {
|
|
alreadyExists = false
|
|
},
|
|
Do: func(tx SQLQueries) error {
|
|
chanIDB := channelIDToBytes(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 := tx.GetChannelBySCID(
|
|
ctx, sqlc.GetChannelBySCIDParams{
|
|
Scid: chanIDB,
|
|
Version: int16(edge.Version),
|
|
},
|
|
)
|
|
if err == nil {
|
|
alreadyExists = true
|
|
return nil
|
|
} else if !errors.Is(err, sql.ErrNoRows) {
|
|
return fmt.Errorf("unable to fetch channel: %w",
|
|
err)
|
|
}
|
|
|
|
return insertChannel(ctx, tx, edge)
|
|
},
|
|
OnCommit: func(err error) error {
|
|
switch {
|
|
case err != nil:
|
|
return err
|
|
case alreadyExists:
|
|
return ErrEdgeAlreadyExist
|
|
default:
|
|
s.rejectCache.remove(
|
|
edge.Version, edge.ChannelID,
|
|
)
|
|
s.chanCache.remove(
|
|
edge.Version, edge.ChannelID,
|
|
)
|
|
|
|
return nil
|
|
}
|
|
},
|
|
}
|
|
|
|
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 Store interface.
|
|
func (s *SQLStore) HighestChanID(ctx context.Context,
|
|
v lnwire.GossipVersion) (uint64, error) {
|
|
|
|
var highestChanID uint64
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
if !isKnownGossipVersion(v) {
|
|
return fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
|
|
chanID, err := db.HighestSCID(ctx, int16(v))
|
|
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
|
|
}
|
|
|
|
// UpdateEdgePolicy updates the edge routing policy for a single directed edge
|
|
// within the database for the referenced channel. The `flags` attribute within
|
|
// the ChannelEdgePolicy determines which of the directed edges are being
|
|
// updated. If the flag is 1, then the first node's information is being
|
|
// updated, otherwise it's the second node's information. The node ordering is
|
|
// determined by the lexicographical ordering of the identity public keys of the
|
|
// nodes on either side of the channel.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) UpdateEdgePolicy(ctx context.Context,
|
|
edge *models.ChannelEdgePolicy,
|
|
opts ...batch.SchedulerOption) (route.Vertex, route.Vertex, error) {
|
|
|
|
var (
|
|
isUpdate1 bool
|
|
edgeNotFound bool
|
|
from, to route.Vertex
|
|
)
|
|
|
|
r := &batch.Request[SQLQueries]{
|
|
Opts: batch.NewSchedulerOptions(opts...),
|
|
Reset: func() {
|
|
isUpdate1 = false
|
|
edgeNotFound = false
|
|
},
|
|
Do: func(tx SQLQueries) error {
|
|
var err error
|
|
from, to, isUpdate1, err = updateChanEdgePolicy(
|
|
ctx, tx, edge,
|
|
)
|
|
// It is possible that two of the same policy
|
|
// announcements are both being processed in the same
|
|
// batch. This may case the UpsertEdgePolicy conflict to
|
|
// be hit since we require at the db layer that the
|
|
// new last_update is greater than the existing
|
|
// last_update. We need to gracefully handle this here.
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
} else if err != nil {
|
|
log.Errorf("UpdateEdgePolicy faild: %v", err)
|
|
}
|
|
|
|
// Silence ErrEdgeNotFound so that the batch can
|
|
// succeed, but propagate the error via local state.
|
|
if errors.Is(err, ErrEdgeNotFound) {
|
|
edgeNotFound = true
|
|
return nil
|
|
}
|
|
|
|
return err
|
|
},
|
|
OnCommit: func(err error) error {
|
|
switch {
|
|
case err != nil:
|
|
return err
|
|
case edgeNotFound:
|
|
return ErrEdgeNotFound
|
|
default:
|
|
s.updateEdgeCache(edge, isUpdate1)
|
|
return nil
|
|
}
|
|
},
|
|
}
|
|
|
|
err := s.chanScheduler.Execute(ctx, r)
|
|
|
|
return from, to, err
|
|
}
|
|
|
|
// updateEdgeCache updates our reject and channel caches with the new
|
|
// edge policy information.
|
|
func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy,
|
|
isUpdate1 bool) {
|
|
|
|
// If an entry for this channel is found in reject cache, we'll modify
|
|
// the entry with the updated timestamp for the direction that was just
|
|
// written. If the edge doesn't exist, we'll load the cache entry lazily
|
|
// during the next query for this edge.
|
|
if entry, ok := s.rejectCache.get(e.Version, e.ChannelID); ok {
|
|
switch e.Version {
|
|
case gossipV1:
|
|
updateRejectCacheEntryV1(
|
|
&entry, isUpdate1, e.LastUpdate,
|
|
)
|
|
case gossipV2:
|
|
updateRejectCacheEntryV2(
|
|
&entry, isUpdate1, e.LastBlockHeight,
|
|
)
|
|
}
|
|
s.rejectCache.insert(e.Version, e.ChannelID, entry)
|
|
}
|
|
|
|
// If an entry for this channel is found in channel cache, we'll modify
|
|
// the entry with the updated policy for the direction that was just
|
|
// written. If the edge doesn't exist, we'll defer loading the info and
|
|
// policies and lazily read from disk during the next query.
|
|
if channel, ok := s.chanCache.get(e.Version, e.ChannelID); ok {
|
|
if isUpdate1 {
|
|
channel.Policy1 = e
|
|
} else {
|
|
channel.Policy2 = e
|
|
}
|
|
s.chanCache.insert(e.Version, e.ChannelID, channel)
|
|
}
|
|
}
|
|
|
|
// ForEachSourceNodeChannel iterates through all channels of the source node,
|
|
// executing the passed callback on each. The call-back is provided with the
|
|
// channel's outpoint, whether we have a policy for the channel and the channel
|
|
// peer's node information.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context,
|
|
v lnwire.GossipVersion, cb func(chanPoint wire.OutPoint,
|
|
havePolicy bool, otherNode *models.Node) error,
|
|
reset func()) error {
|
|
|
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
nodeID, nodePub, err := s.getSourceNode(ctx, db, v)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch source node: %w",
|
|
err)
|
|
}
|
|
|
|
return forEachNodeChannel(
|
|
ctx, db, s.cfg, v, nodeID,
|
|
func(info *models.ChannelEdgeInfo,
|
|
outPolicy *models.ChannelEdgePolicy,
|
|
_ *models.ChannelEdgePolicy) error {
|
|
|
|
// Fetch the other node.
|
|
var (
|
|
otherNodePub [33]byte
|
|
node1 = info.NodeKey1Bytes
|
|
node2 = info.NodeKey2Bytes
|
|
)
|
|
switch {
|
|
case bytes.Equal(node1[:], nodePub[:]):
|
|
otherNodePub = node2
|
|
case bytes.Equal(node2[:], nodePub[:]):
|
|
otherNodePub = node1
|
|
default:
|
|
return fmt.Errorf("node not " +
|
|
"participating in this channel")
|
|
}
|
|
|
|
_, otherNode, err := getNodeByPubKey(
|
|
ctx, s.cfg.QueryCfg, db, v,
|
|
otherNodePub,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch "+
|
|
"other node(%x): %w",
|
|
otherNodePub, err)
|
|
}
|
|
|
|
return cb(
|
|
info.ChannelPoint, outPolicy != nil,
|
|
otherNode,
|
|
)
|
|
},
|
|
)
|
|
}, reset)
|
|
}
|
|
|
|
// ForEachNode iterates through all the stored vertices/nodes in the graph,
|
|
// executing the passed callback with each node encountered. If the callback
|
|
// returns an error, then the transaction is aborted and the iteration stops
|
|
// early.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) ForEachNode(ctx context.Context, v lnwire.GossipVersion,
|
|
cb func(node *models.Node) error, reset func()) error {
|
|
|
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
return forEachNodePaginated(
|
|
ctx, s.cfg.QueryCfg, db,
|
|
v, func(_ context.Context, _ int64,
|
|
node *models.Node) error {
|
|
|
|
return cb(node)
|
|
},
|
|
)
|
|
}, reset)
|
|
}
|
|
|
|
// 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 callback returns an error, then the iteration
|
|
// is halted with the error propagated back up to the caller.
|
|
//
|
|
// Unknown policies are passed into the callback as nil values.
|
|
//
|
|
// NOTE: this is part of the graphdb.NodeTraverser interface.
|
|
func (s *SQLStore) ForEachNodeDirectedChannel(ctx context.Context,
|
|
v lnwire.GossipVersion, nodePub route.Vertex,
|
|
cb func(channel *DirectedChannel) error, reset func()) error {
|
|
|
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
return forEachNodeDirectedChannel(ctx, db, v, nodePub, cb)
|
|
}, reset)
|
|
}
|
|
|
|
// ForEachNodeCacheable iterates through all the stored vertices/nodes in the
|
|
// graph, executing the passed callback with each node encountered. If the
|
|
// callback returns an error, then the transaction is aborted and the iteration
|
|
// stops early.
|
|
func (s *SQLStore) ForEachNodeCacheable(ctx context.Context,
|
|
v lnwire.GossipVersion, cb func(route.Vertex,
|
|
*lnwire.FeatureVector) error, reset func()) error {
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
return forEachNodeCacheable(
|
|
ctx, s.cfg.QueryCfg, db, v,
|
|
func(_ int64, nodePub route.Vertex,
|
|
features *lnwire.FeatureVector) error {
|
|
|
|
return cb(nodePub, features)
|
|
},
|
|
)
|
|
}, reset)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch nodes: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// ForEachNodeChannel iterates through all channels of the given node,
|
|
// executing the passed callback with an edge info structure and the policies
|
|
// of each end of the channel. The first edge policy is the outgoing edge *to*
|
|
// the connecting node, while the second is the incoming edge *from* the
|
|
// connecting node. If the callback returns an error, then the iteration is
|
|
// halted with the error propagated back up to the caller.
|
|
//
|
|
// Unknown policies are passed into the callback as nil values.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) ForEachNodeChannel(ctx context.Context,
|
|
v lnwire.GossipVersion, nodePub route.Vertex,
|
|
cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
|
*models.ChannelEdgePolicy) error, reset func()) error {
|
|
|
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
dbNode, err := db.GetNodeByPubKey(
|
|
ctx, sqlc.GetNodeByPubKeyParams{
|
|
Version: int16(v),
|
|
PubKey: nodePub[:],
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
} else if err != nil {
|
|
return fmt.Errorf("unable to fetch node: %w", err)
|
|
}
|
|
|
|
return forEachNodeChannel(ctx, db, s.cfg, v, dbNode.ID, cb)
|
|
}, reset)
|
|
}
|
|
|
|
// extractMaxUpdateTime returns the maximum of the two policy update times.
|
|
// This is used for pagination cursor tracking.
|
|
func extractMaxUpdateTime(
|
|
row sqlc.GetChannelsByPolicyLastUpdateRangeRow) int64 {
|
|
|
|
switch {
|
|
case row.Policy1LastUpdate.Valid && row.Policy2LastUpdate.Valid:
|
|
return max(row.Policy1LastUpdate.Int64,
|
|
row.Policy2LastUpdate.Int64)
|
|
case row.Policy1LastUpdate.Valid:
|
|
return row.Policy1LastUpdate.Int64
|
|
case row.Policy2LastUpdate.Valid:
|
|
return row.Policy2LastUpdate.Int64
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
// buildChannelEdgeFromRow constructs a ChannelEdge from the common fields
|
|
// shared by both the v1 time-range and v2 block-height-range query rows.
|
|
// The policyRow parameter is passed to extractChannelPolicies which
|
|
// type-switches on the concrete sqlc row type.
|
|
func (s *SQLStore) buildChannelEdgeFromRow(ctx context.Context,
|
|
db SQLQueries, n1, n2 sqlc.GraphNode, ch sqlc.GraphChannel,
|
|
policyRow any) (ChannelEdge, error) {
|
|
|
|
node1, err := buildNode(ctx, s.cfg.QueryCfg, db, n1)
|
|
if err != nil {
|
|
return ChannelEdge{}, fmt.Errorf("unable to build node1: %w",
|
|
err)
|
|
}
|
|
|
|
node2, err := buildNode(ctx, s.cfg.QueryCfg, db, n2)
|
|
if err != nil {
|
|
return ChannelEdge{}, fmt.Errorf("unable to build node2: %w",
|
|
err)
|
|
}
|
|
|
|
channel, err := getAndBuildEdgeInfo(
|
|
ctx, s.cfg, db, ch, node1.PubKeyBytes,
|
|
node2.PubKeyBytes,
|
|
)
|
|
if err != nil {
|
|
return ChannelEdge{}, fmt.Errorf("unable to build "+
|
|
"channel info: %w", err)
|
|
}
|
|
|
|
dbPol1, dbPol2, err := extractChannelPolicies(policyRow)
|
|
if err != nil {
|
|
return ChannelEdge{}, fmt.Errorf("unable to extract "+
|
|
"channel policies: %w", err)
|
|
}
|
|
|
|
p1, p2, err := getAndBuildChanPolicies(
|
|
ctx, s.cfg.QueryCfg, db, dbPol1, dbPol2, channel.ChannelID,
|
|
node1.PubKeyBytes, node2.PubKeyBytes,
|
|
)
|
|
if err != nil {
|
|
return ChannelEdge{}, fmt.Errorf("unable to build "+
|
|
"channel policies: %w", err)
|
|
}
|
|
|
|
return ChannelEdge{
|
|
Info: channel,
|
|
Policy1: p1,
|
|
Policy2: p2,
|
|
Node1: node1,
|
|
Node2: node2,
|
|
}, nil
|
|
}
|
|
|
|
// extractMaxBlockHeight returns the maximum of the two policy block heights.
|
|
// This is used for pagination cursor tracking in v2 gossip queries.
|
|
func extractMaxBlockHeight(
|
|
row sqlc.GetChannelsByPolicyBlockRangeRow) int64 {
|
|
|
|
switch {
|
|
case row.Policy1BlockHeight.Valid &&
|
|
row.Policy2BlockHeight.Valid:
|
|
|
|
return max(row.Policy1BlockHeight.Int64,
|
|
row.Policy2BlockHeight.Int64)
|
|
case row.Policy1BlockHeight.Valid:
|
|
return row.Policy1BlockHeight.Int64
|
|
case row.Policy2BlockHeight.Valid:
|
|
return row.Policy2BlockHeight.Int64
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
// updateChanCacheBatch updates the channel cache with multiple edges at once.
|
|
// This method acquires the cache lock only once for the entire batch.
|
|
func (s *SQLStore) updateChanCacheBatch(v lnwire.GossipVersion,
|
|
edgesToCache map[uint64]ChannelEdge) {
|
|
|
|
if len(edgesToCache) == 0 {
|
|
return
|
|
}
|
|
|
|
s.cacheMu.Lock()
|
|
defer s.cacheMu.Unlock()
|
|
|
|
for chanID, edge := range edgesToCache {
|
|
s.chanCache.insert(v, chanID, edge)
|
|
}
|
|
}
|
|
|
|
// ChanUpdatesInHorizon returns all the known channel edges which have at least
|
|
// one edge update within the specified range for the given gossip version. For
|
|
// v1, the range is time-based with [start, end) per BOLT 07.
|
|
//
|
|
// NOTE: This is part of the Store interface.
|
|
func (s *SQLStore) ChanUpdatesInHorizon(ctx context.Context,
|
|
v lnwire.GossipVersion, r ChanUpdateRange,
|
|
opts ...IteratorOption) iter.Seq2[ChannelEdge, error] {
|
|
|
|
if err := r.validateForVersion(v); err != nil {
|
|
return func(yield func(ChannelEdge, error) bool) {
|
|
_ = yield(ChannelEdge{}, err)
|
|
}
|
|
}
|
|
|
|
cfg := defaultIteratorConfig()
|
|
for _, opt := range opts {
|
|
opt(cfg)
|
|
}
|
|
|
|
switch v {
|
|
case gossipV1:
|
|
return s.chanUpdatesInHorizonV1(ctx, r, cfg)
|
|
|
|
case gossipV2:
|
|
return s.chanUpdatesInHorizonV2(ctx, r, cfg)
|
|
|
|
default:
|
|
err := fmt.Errorf("unknown gossip version: %v", v)
|
|
return func(yield func(ChannelEdge, error) bool) {
|
|
_ = yield(ChannelEdge{}, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// chanUpdatesInHorizonV1 implements the v1 time-based channel horizon query.
|
|
func (s *SQLStore) chanUpdatesInHorizonV1(ctx context.Context,
|
|
r ChanUpdateRange,
|
|
cfg *iterConfig) iter.Seq2[ChannelEdge, error] {
|
|
|
|
startTime := r.StartTime.UnwrapOr(time.Time{})
|
|
endTime := r.EndTime.UnwrapOr(time.Time{})
|
|
|
|
return func(yield func(ChannelEdge, error) bool) {
|
|
var (
|
|
edgesSeen = make(map[uint64]struct{})
|
|
edgesToCache = make(map[uint64]ChannelEdge)
|
|
hits int
|
|
total int
|
|
lastUpdateTime sql.NullInt64
|
|
lastID sql.NullInt64
|
|
hasMore = true
|
|
)
|
|
|
|
// Each iteration, we'll read a batch amount of channel updates
|
|
// (consulting the cache along the way), yield them, then loop
|
|
// back to decide if we have any more updates to read out.
|
|
for hasMore {
|
|
var batch []ChannelEdge
|
|
|
|
// Acquire read lock before starting transaction to
|
|
// ensure consistent lock ordering (cacheMu -> DB) and
|
|
// prevent deadlock with write operations.
|
|
s.cacheMu.RLock()
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(),
|
|
func(db SQLQueries) error {
|
|
//nolint:ll
|
|
params := sqlc.GetChannelsByPolicyLastUpdateRangeParams{
|
|
Version: int16(gossipV1),
|
|
StartTime: sqldb.SQLInt64(
|
|
startTime.Unix(),
|
|
),
|
|
EndTime: sqldb.SQLInt64(
|
|
endTime.Unix(),
|
|
),
|
|
LastUpdateTime: lastUpdateTime,
|
|
LastID: lastID,
|
|
MaxResults: sql.NullInt32{
|
|
Int32: int32(
|
|
cfg.chanUpdateIterBatchSize,
|
|
),
|
|
Valid: true,
|
|
},
|
|
}
|
|
//nolint:ll
|
|
rows, err := db.GetChannelsByPolicyLastUpdateRange(
|
|
ctx, params,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
//nolint:ll
|
|
hasMore = len(rows) == cfg.chanUpdateIterBatchSize
|
|
|
|
//nolint:ll
|
|
for _, row := range rows {
|
|
lastUpdateTime = sql.NullInt64{
|
|
Int64: extractMaxUpdateTime(row),
|
|
Valid: true,
|
|
}
|
|
lastID = sql.NullInt64{
|
|
Int64: row.GraphChannel.ID,
|
|
Valid: true,
|
|
}
|
|
|
|
// Skip if we've already
|
|
// processed this channel.
|
|
chanIDInt := byteOrder.Uint64(
|
|
row.GraphChannel.Scid,
|
|
)
|
|
_, ok := edgesSeen[chanIDInt]
|
|
if ok {
|
|
continue
|
|
}
|
|
|
|
// Check cache (we already hold
|
|
// shared read lock).
|
|
channel, ok := s.chanCache.get(
|
|
gossipV1,
|
|
chanIDInt,
|
|
)
|
|
if ok {
|
|
hits++
|
|
total++
|
|
edgesSeen[chanIDInt] = struct{}{}
|
|
batch = append(batch, channel)
|
|
|
|
continue
|
|
}
|
|
|
|
chanEdge, err := s.buildChannelEdgeFromRow(
|
|
ctx, db,
|
|
row.GraphNode,
|
|
row.GraphNode_2,
|
|
row.GraphChannel,
|
|
row,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
edgesSeen[chanIDInt] = struct{}{}
|
|
edgesToCache[chanIDInt] = chanEdge
|
|
|
|
batch = append(batch, chanEdge)
|
|
|
|
total++
|
|
}
|
|
|
|
return nil
|
|
}, func() {
|
|
batch = nil
|
|
edgesSeen = make(map[uint64]struct{})
|
|
edgesToCache = make(
|
|
map[uint64]ChannelEdge,
|
|
)
|
|
})
|
|
|
|
// Release read lock after transaction completes.
|
|
s.cacheMu.RUnlock()
|
|
|
|
if err != nil {
|
|
log.Errorf("ChanUpdatesInHorizon "+
|
|
"batch error: %v", err)
|
|
|
|
yield(ChannelEdge{}, err)
|
|
|
|
return
|
|
}
|
|
|
|
for _, edge := range batch {
|
|
if !yield(edge, nil) {
|
|
return
|
|
}
|
|
}
|
|
|
|
// Update cache after successful batch yield, setting
|
|
// the cache lock only once for the entire batch.
|
|
s.updateChanCacheBatch(gossipV1, edgesToCache)
|
|
edgesToCache = make(map[uint64]ChannelEdge)
|
|
|
|
// If the batch didn't yield anything, then we're done.
|
|
if len(batch) == 0 {
|
|
break
|
|
}
|
|
}
|
|
|
|
if total > 0 {
|
|
log.Debugf("ChanUpdatesInHorizon(v1) hit "+
|
|
"percentage: %.2f (%d/%d)",
|
|
float64(hits)*100/float64(total), hits, total)
|
|
} else {
|
|
log.Debugf("ChanUpdatesInHorizon(v1) returned no " +
|
|
"edges in horizon")
|
|
}
|
|
}
|
|
}
|
|
|
|
// chanUpdatesInHorizonV2 implements the v2 block-height-based channel horizon
|
|
// query.
|
|
func (s *SQLStore) chanUpdatesInHorizonV2(ctx context.Context,
|
|
r ChanUpdateRange,
|
|
cfg *iterConfig) iter.Seq2[ChannelEdge, error] {
|
|
|
|
startHeight := int64(r.StartHeight.UnwrapOr(0))
|
|
endHeight := int64(r.EndHeight.UnwrapOr(0))
|
|
batchSize := cfg.chanUpdateIterBatchSize
|
|
|
|
return func(yield func(ChannelEdge, error) bool) {
|
|
var (
|
|
edgesSeen = make(map[uint64]struct{})
|
|
edgesToCache = make(map[uint64]ChannelEdge)
|
|
hits int
|
|
total int
|
|
lastBlockHeight sql.NullInt64
|
|
lastID sql.NullInt64
|
|
hasMore = true
|
|
)
|
|
|
|
// queryChannels fetches the next page of v2 channels in
|
|
// the block-height range.
|
|
queryChannels := func(
|
|
db SQLQueries,
|
|
) ([]sqlc.GetChannelsByPolicyBlockRangeRow, error) {
|
|
|
|
return db.GetChannelsByPolicyBlockRange(
|
|
ctx,
|
|
sqlc.GetChannelsByPolicyBlockRangeParams{
|
|
Version: int16(gossipV2),
|
|
StartHeight: sqldb.SQLInt64(
|
|
startHeight,
|
|
),
|
|
EndHeight: sqldb.SQLInt64(
|
|
endHeight,
|
|
),
|
|
LastBlockHeight: lastBlockHeight,
|
|
LastID: lastID,
|
|
MaxResults: sql.NullInt32{
|
|
Int32: int32(batchSize),
|
|
Valid: true,
|
|
},
|
|
},
|
|
)
|
|
}
|
|
|
|
// processRow handles a single channel row: updates
|
|
// pagination cursors, checks the seen set and cache, and
|
|
// builds the channel edge if needed.
|
|
processRow := func(ctx context.Context, db SQLQueries,
|
|
row sqlc.GetChannelsByPolicyBlockRangeRow,
|
|
batch *[]ChannelEdge) error {
|
|
|
|
lastBlockHeight = sql.NullInt64{
|
|
Int64: extractMaxBlockHeight(row),
|
|
Valid: true,
|
|
}
|
|
lastID = sql.NullInt64{
|
|
Int64: row.GraphChannel.ID,
|
|
Valid: true,
|
|
}
|
|
|
|
chanIDInt := byteOrder.Uint64(
|
|
row.GraphChannel.Scid,
|
|
)
|
|
if _, ok := edgesSeen[chanIDInt]; ok {
|
|
return nil
|
|
}
|
|
|
|
// Check cache (we already hold shared read
|
|
// lock).
|
|
channel, ok := s.chanCache.get(
|
|
gossipV2, chanIDInt,
|
|
)
|
|
if ok {
|
|
hits++
|
|
total++
|
|
edgesSeen[chanIDInt] = struct{}{}
|
|
*batch = append(*batch, channel)
|
|
|
|
return nil
|
|
}
|
|
|
|
chanEdge, err := s.buildChannelEdgeFromRow(
|
|
ctx, db, row.GraphNode,
|
|
row.GraphNode_2,
|
|
row.GraphChannel, row,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
edgesSeen[chanIDInt] = struct{}{}
|
|
edgesToCache[chanIDInt] = chanEdge
|
|
*batch = append(*batch, chanEdge)
|
|
total++
|
|
|
|
return nil
|
|
}
|
|
|
|
for hasMore {
|
|
var batch []ChannelEdge
|
|
|
|
s.cacheMu.RLock()
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(),
|
|
func(db SQLQueries) error {
|
|
rows, err := queryChannels(db)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
hasMore = len(rows) == batchSize
|
|
|
|
for _, row := range rows {
|
|
err := processRow(
|
|
ctx, db, row, &batch,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}, func() {
|
|
batch = nil
|
|
edgesSeen = make(
|
|
map[uint64]struct{},
|
|
)
|
|
edgesToCache = make(
|
|
map[uint64]ChannelEdge,
|
|
)
|
|
},
|
|
)
|
|
|
|
s.cacheMu.RUnlock()
|
|
|
|
if err != nil {
|
|
log.Errorf("ChanUpdatesInHorizon(v2) "+
|
|
"batch error: %v", err)
|
|
|
|
yield(ChannelEdge{}, err)
|
|
|
|
return
|
|
}
|
|
|
|
for _, edge := range batch {
|
|
if !yield(edge, nil) {
|
|
return
|
|
}
|
|
}
|
|
|
|
s.updateChanCacheBatch(gossipV2, edgesToCache)
|
|
edgesToCache = make(map[uint64]ChannelEdge)
|
|
|
|
if len(batch) == 0 {
|
|
break
|
|
}
|
|
}
|
|
|
|
if total > 0 {
|
|
log.Debugf("ChanUpdatesInHorizon(v2) hit "+
|
|
"percentage: %.2f (%d/%d)",
|
|
float64(hits)*100/float64(total), hits,
|
|
total)
|
|
} else {
|
|
log.Debugf("ChanUpdatesInHorizon(v2) returned " +
|
|
"no edges in horizon")
|
|
}
|
|
}
|
|
}
|
|
|
|
// ForEachNodeCached is similar to forEachNode, but it returns DirectedChannel
|
|
// data to the call-back.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) ForEachNodeCached(ctx context.Context,
|
|
v lnwire.GossipVersion,
|
|
cb func(ctx context.Context, node route.Vertex,
|
|
chans map[uint64]*DirectedChannel) error, reset func()) error {
|
|
|
|
type nodeCachedBatchData struct {
|
|
features map[int64][]int
|
|
chanBatchData *batchChannelData
|
|
chanMap map[int64][]sqlc.ListChannelsForNodeIDsRow
|
|
}
|
|
|
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
// pageQueryFunc is used to query the next page of nodes.
|
|
pageQueryFunc := func(ctx context.Context, lastID int64,
|
|
limit int32) ([]sqlc.ListNodeIDsAndPubKeysRow, error) {
|
|
|
|
return db.ListNodeIDsAndPubKeys(
|
|
ctx, sqlc.ListNodeIDsAndPubKeysParams{
|
|
Version: int16(v),
|
|
ID: lastID,
|
|
Limit: limit,
|
|
},
|
|
)
|
|
}
|
|
|
|
// batchDataFunc is then used to batch load the data required
|
|
// for each page of nodes.
|
|
batchDataFunc := func(ctx context.Context,
|
|
nodeIDs []int64) (*nodeCachedBatchData, error) {
|
|
|
|
// Batch load node features.
|
|
nodeFeatures, err := batchLoadNodeFeaturesHelper(
|
|
ctx, s.cfg.QueryCfg, db, nodeIDs,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load "+
|
|
"node features: %w", err)
|
|
}
|
|
|
|
// Batch load ALL unique channels for ALL nodes in this
|
|
// page.
|
|
allChannels, err := db.ListChannelsForNodeIDs(
|
|
ctx, sqlc.ListChannelsForNodeIDsParams{
|
|
Version: int16(lnwire.GossipVersion1),
|
|
Node1Ids: nodeIDs,
|
|
Node2Ids: nodeIDs,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch "+
|
|
"fetch channels for nodes: %w", err)
|
|
}
|
|
|
|
// Deduplicate channels and collect IDs.
|
|
var (
|
|
allChannelIDs []int64
|
|
allPolicyIDs []int64
|
|
)
|
|
uniqueChannels := make(
|
|
map[int64]sqlc.ListChannelsForNodeIDsRow,
|
|
)
|
|
|
|
for _, channel := range allChannels {
|
|
channelID := channel.GraphChannel.ID
|
|
|
|
// Only process each unique channel once.
|
|
_, exists := uniqueChannels[channelID]
|
|
if exists {
|
|
continue
|
|
}
|
|
|
|
uniqueChannels[channelID] = channel
|
|
allChannelIDs = append(allChannelIDs, channelID)
|
|
|
|
if channel.Policy1ID.Valid {
|
|
allPolicyIDs = append(
|
|
allPolicyIDs,
|
|
channel.Policy1ID.Int64,
|
|
)
|
|
}
|
|
if channel.Policy2ID.Valid {
|
|
allPolicyIDs = append(
|
|
allPolicyIDs,
|
|
channel.Policy2ID.Int64,
|
|
)
|
|
}
|
|
}
|
|
|
|
// Batch load channel data for all unique channels.
|
|
channelBatchData, err := batchLoadChannelData(
|
|
ctx, s.cfg.QueryCfg, db, allChannelIDs,
|
|
allPolicyIDs,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch "+
|
|
"load channel data: %w", err)
|
|
}
|
|
|
|
// Create map of node ID to channels that involve this
|
|
// node.
|
|
nodeIDSet := make(map[int64]bool)
|
|
for _, nodeID := range nodeIDs {
|
|
nodeIDSet[nodeID] = true
|
|
}
|
|
|
|
nodeChannelMap := make(
|
|
map[int64][]sqlc.ListChannelsForNodeIDsRow,
|
|
)
|
|
for _, channel := range uniqueChannels {
|
|
// Add channel to both nodes if they're in our
|
|
// current page.
|
|
node1 := channel.GraphChannel.NodeID1
|
|
if nodeIDSet[node1] {
|
|
nodeChannelMap[node1] = append(
|
|
nodeChannelMap[node1], channel,
|
|
)
|
|
}
|
|
node2 := channel.GraphChannel.NodeID2
|
|
if nodeIDSet[node2] {
|
|
nodeChannelMap[node2] = append(
|
|
nodeChannelMap[node2], channel,
|
|
)
|
|
}
|
|
}
|
|
|
|
return &nodeCachedBatchData{
|
|
features: nodeFeatures,
|
|
chanBatchData: channelBatchData,
|
|
chanMap: nodeChannelMap,
|
|
}, nil
|
|
}
|
|
|
|
// processItem is used to process each node in the current page.
|
|
processItem := func(ctx context.Context,
|
|
nodeData sqlc.ListNodeIDsAndPubKeysRow,
|
|
batchData *nodeCachedBatchData) error {
|
|
|
|
// Build feature vector for this node.
|
|
fv := lnwire.EmptyFeatureVector()
|
|
features, exists := batchData.features[nodeData.ID]
|
|
if exists {
|
|
for _, bit := range features {
|
|
fv.Set(lnwire.FeatureBit(bit))
|
|
}
|
|
}
|
|
|
|
var nodePub route.Vertex
|
|
copy(nodePub[:], nodeData.PubKey)
|
|
|
|
nodeChannels := batchData.chanMap[nodeData.ID]
|
|
|
|
toNodeCallback := func() route.Vertex {
|
|
return nodePub
|
|
}
|
|
|
|
// Build cached channels map for this node.
|
|
channels := make(map[uint64]*DirectedChannel)
|
|
for _, channelRow := range nodeChannels {
|
|
directedChan, err := buildDirectedChannel(
|
|
s.cfg.ChainHash, nodeData.ID, nodePub,
|
|
channelRow, batchData.chanBatchData, fv,
|
|
toNodeCallback,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
channels[directedChan.ChannelID] = directedChan
|
|
}
|
|
|
|
return cb(ctx, nodePub, channels)
|
|
}
|
|
|
|
return sqldb.ExecuteCollectAndBatchWithSharedDataQuery(
|
|
ctx, s.cfg.QueryCfg, int64(-1), pageQueryFunc,
|
|
func(node sqlc.ListNodeIDsAndPubKeysRow) int64 {
|
|
return node.ID
|
|
},
|
|
func(node sqlc.ListNodeIDsAndPubKeysRow) (int64,
|
|
error) {
|
|
|
|
return node.ID, nil
|
|
},
|
|
batchDataFunc, processItem,
|
|
)
|
|
}, reset)
|
|
}
|
|
|
|
// 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(ctx context.Context,
|
|
v lnwire.GossipVersion,
|
|
cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy,
|
|
*models.CachedEdgePolicy) error, reset func()) error {
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
|
|
handleChannel := func(_ context.Context,
|
|
row sqlc.ListChannelsWithPoliciesForCachePaginatedRow) error {
|
|
|
|
node1, node2, err := buildNodeVertices(
|
|
row.Node1Pubkey, row.Node2Pubkey,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
edge := buildCacheableChannelInfo(
|
|
row.Scid, row.Capacity.Int64, node1, node2,
|
|
)
|
|
|
|
dbPol1, dbPol2, err := extractChannelPolicies(row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
pol1, pol2, err := buildCachedChanPolicies(
|
|
dbPol1, dbPol2, edge.ChannelID, node1, node2,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return cb(edge, pol1, pol2)
|
|
}
|
|
|
|
extractCursor := func(
|
|
row sqlc.ListChannelsWithPoliciesForCachePaginatedRow) int64 {
|
|
|
|
return row.ID
|
|
}
|
|
|
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
//nolint:ll
|
|
queryFunc := func(ctx context.Context, lastID int64,
|
|
limit int32) ([]sqlc.ListChannelsWithPoliciesForCachePaginatedRow,
|
|
error) {
|
|
|
|
return db.ListChannelsWithPoliciesForCachePaginated(
|
|
ctx, sqlc.ListChannelsWithPoliciesForCachePaginatedParams{
|
|
Version: int16(v),
|
|
ID: lastID,
|
|
Limit: limit,
|
|
},
|
|
)
|
|
}
|
|
|
|
return sqldb.ExecutePaginatedQuery(
|
|
ctx, s.cfg.QueryCfg, int64(-1), queryFunc,
|
|
extractCursor, handleChannel,
|
|
)
|
|
}, reset)
|
|
}
|
|
|
|
// 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.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) ForEachChannel(ctx context.Context,
|
|
v lnwire.GossipVersion, cb func(*models.ChannelEdgeInfo,
|
|
*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
|
|
reset func()) error {
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
|
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
return forEachChannelWithPolicies(ctx, db, s.cfg, v, cb)
|
|
}, reset)
|
|
}
|
|
|
|
// FilterChannelRange returns the channel ID's of all known channels which were
|
|
// mined in a block height within the passed range. The channel IDs are grouped
|
|
// by their common block height. This method can be used to quickly share with a
|
|
// peer the set of channels we know of within a particular range to catch them
|
|
// up after a period of time offline. If withTimestamps is true then the
|
|
// timestamp info of the latest received channel update messages of the channel
|
|
// will be included in the response.
|
|
//
|
|
// NOTE: This is part of the Store interface.
|
|
func (s *SQLStore) FilterChannelRange(ctx context.Context,
|
|
v lnwire.GossipVersion, startHeight, endHeight uint32,
|
|
withTimestamps bool) ([]BlockChannelRange, error) {
|
|
|
|
var (
|
|
startSCID = &lnwire.ShortChannelID{
|
|
BlockHeight: startHeight,
|
|
}
|
|
endSCID = lnwire.ShortChannelID{
|
|
BlockHeight: endHeight,
|
|
TxIndex: math.MaxUint32 & 0x00ffffff,
|
|
TxPosition: math.MaxUint16,
|
|
}
|
|
chanIDStart = channelIDToBytes(startSCID.ToUint64())
|
|
chanIDEnd = channelIDToBytes(endSCID.ToUint64())
|
|
)
|
|
|
|
// 1) get all channels where channelID is between start and end chan ID.
|
|
// 2) skip if not public (ie, no channel_proof)
|
|
// 3) collect that channel.
|
|
// 4) if timestamps are wanted, fetch both policies for node 1 and node2
|
|
// and add those timestamps to the collected channel.
|
|
channelsPerBlock := make(map[uint32][]ChannelUpdateInfo)
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
var (
|
|
dbChans []sqlc.GraphChannel
|
|
chanErr error
|
|
)
|
|
|
|
switch v {
|
|
case gossipV1:
|
|
dbChans, chanErr = db.GetPublicV1ChannelsBySCID(
|
|
ctx, sqlc.GetPublicV1ChannelsBySCIDParams{
|
|
StartScid: chanIDStart,
|
|
EndScid: chanIDEnd,
|
|
},
|
|
)
|
|
case gossipV2:
|
|
dbChans, chanErr = db.GetPublicV2ChannelsBySCID(
|
|
ctx, sqlc.GetPublicV2ChannelsBySCIDParams{
|
|
StartScid: chanIDStart,
|
|
EndScid: chanIDEnd,
|
|
},
|
|
)
|
|
default:
|
|
return fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
if chanErr != nil {
|
|
return fmt.Errorf("unable to fetch channel range: %w",
|
|
chanErr)
|
|
}
|
|
|
|
for _, dbChan := range dbChans {
|
|
cid := lnwire.NewShortChanIDFromInt(
|
|
byteOrder.Uint64(dbChan.Scid),
|
|
)
|
|
|
|
var chanInfo ChannelUpdateInfo
|
|
switch v {
|
|
case gossipV1:
|
|
chanInfo = NewV1ChannelUpdateInfo(
|
|
cid, time.Time{}, time.Time{},
|
|
)
|
|
case gossipV2:
|
|
chanInfo = NewV2ChannelUpdateInfo(cid, 0, 0)
|
|
}
|
|
|
|
if !withTimestamps {
|
|
channelsPerBlock[cid.BlockHeight] = append(
|
|
channelsPerBlock[cid.BlockHeight],
|
|
chanInfo,
|
|
)
|
|
|
|
continue
|
|
}
|
|
|
|
//nolint:ll
|
|
node1Policy, err := db.GetChannelPolicyByChannelAndNode(
|
|
ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{
|
|
Version: int16(v),
|
|
ChannelID: dbChan.ID,
|
|
NodeID: dbChan.NodeID1,
|
|
},
|
|
)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return fmt.Errorf("unable to fetch node1 "+
|
|
"policy: %w", err)
|
|
} else if err == nil {
|
|
n1Update := node1Policy.LastUpdate.Int64
|
|
n1Height := node1Policy.BlockHeight.Int64
|
|
|
|
switch v {
|
|
case gossipV1:
|
|
chanInfo.Node1Freshness =
|
|
lnwire.UnixTimestamp(n1Update)
|
|
case gossipV2:
|
|
chanInfo.Node1Freshness =
|
|
lnwire.BlockHeightTimestamp(
|
|
n1Height,
|
|
)
|
|
}
|
|
}
|
|
|
|
//nolint:ll
|
|
node2Policy, err := db.GetChannelPolicyByChannelAndNode(
|
|
ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{
|
|
Version: int16(v),
|
|
ChannelID: dbChan.ID,
|
|
NodeID: dbChan.NodeID2,
|
|
},
|
|
)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return fmt.Errorf("unable to fetch node2 "+
|
|
"policy: %w", err)
|
|
} else if err == nil {
|
|
n2Update := node2Policy.LastUpdate.Int64
|
|
n2Height := node2Policy.BlockHeight.Int64
|
|
|
|
switch v {
|
|
case gossipV1:
|
|
chanInfo.Node2Freshness =
|
|
lnwire.UnixTimestamp(n2Update)
|
|
case gossipV2:
|
|
chanInfo.Node2Freshness =
|
|
lnwire.BlockHeightTimestamp(
|
|
n2Height,
|
|
)
|
|
}
|
|
}
|
|
|
|
channelsPerBlock[cid.BlockHeight] = append(
|
|
channelsPerBlock[cid.BlockHeight], chanInfo,
|
|
)
|
|
}
|
|
|
|
return nil
|
|
}, func() {
|
|
channelsPerBlock = make(map[uint32][]ChannelUpdateInfo)
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to fetch channel range: %w", err)
|
|
}
|
|
|
|
if len(channelsPerBlock) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
// Return the channel ranges in ascending block height order.
|
|
blocks := slices.Collect(maps.Keys(channelsPerBlock))
|
|
slices.Sort(blocks)
|
|
|
|
return fn.Map(blocks, func(block uint32) BlockChannelRange {
|
|
return BlockChannelRange{
|
|
Height: block,
|
|
Channels: channelsPerBlock[block],
|
|
}
|
|
}), nil
|
|
}
|
|
|
|
// MarkEdgeZombie attempts to mark a channel identified by its channel ID as a
|
|
// zombie for the given gossip version. This method is used on an ad-hoc basis,
|
|
// when channels need to be marked as zombies outside the normal pruning cycle.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) MarkEdgeZombie(ctx context.Context, v lnwire.GossipVersion,
|
|
chanID uint64, pubKey1, pubKey2 [33]byte) error {
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
|
|
s.cacheMu.Lock()
|
|
defer s.cacheMu.Unlock()
|
|
|
|
chanIDB := channelIDToBytes(chanID)
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
|
return db.UpsertZombieChannel(
|
|
ctx, sqlc.UpsertZombieChannelParams{
|
|
Version: int16(v),
|
|
Scid: chanIDB,
|
|
NodeKey1: pubKey1[:],
|
|
NodeKey2: pubKey2[:],
|
|
},
|
|
)
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to upsert zombie channel "+
|
|
"(channel_id=%d): %w", chanID, err)
|
|
}
|
|
|
|
s.rejectCache.remove(v, chanID)
|
|
s.chanCache.remove(v, chanID)
|
|
|
|
return nil
|
|
}
|
|
|
|
// MarkEdgeLive clears an edge from our zombie index for the given gossip
|
|
// version, deeming it as live.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) MarkEdgeLive(ctx context.Context,
|
|
v lnwire.GossipVersion, chanID uint64) error {
|
|
|
|
s.cacheMu.Lock()
|
|
defer s.cacheMu.Unlock()
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
|
|
chanIDB := channelIDToBytes(chanID)
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
|
res, err := db.DeleteZombieChannel(
|
|
ctx, sqlc.DeleteZombieChannelParams{
|
|
Scid: chanIDB,
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to delete zombie channel: %w",
|
|
err)
|
|
}
|
|
|
|
rows, err := res.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if rows == 0 {
|
|
return ErrZombieEdgeNotFound
|
|
} else if rows > 1 {
|
|
return fmt.Errorf("deleted %d zombie rows, "+
|
|
"expected 1", rows)
|
|
}
|
|
|
|
return nil
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to mark edge live "+
|
|
"(channel_id=%d): %w", chanID, err)
|
|
}
|
|
|
|
s.rejectCache.remove(v, chanID)
|
|
s.chanCache.remove(v, chanID)
|
|
|
|
return err
|
|
}
|
|
|
|
// IsZombieEdge returns whether the edge is considered zombie. If it is a
|
|
// zombie, then the two node public keys corresponding to this edge are also
|
|
// returned.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) IsZombieEdge(ctx context.Context, v lnwire.GossipVersion,
|
|
chanID uint64) (bool, [33]byte, [33]byte, error) {
|
|
|
|
var (
|
|
isZombie bool
|
|
pubKey1, pubKey2 route.Vertex
|
|
chanIDB = channelIDToBytes(chanID)
|
|
)
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return false, [33]byte{}, [33]byte{},
|
|
fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
zombie, err := db.GetZombieChannel(
|
|
ctx, sqlc.GetZombieChannelParams{
|
|
Scid: chanIDB,
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch zombie channel: %w",
|
|
err)
|
|
}
|
|
|
|
copy(pubKey1[:], zombie.NodeKey1)
|
|
copy(pubKey2[:], zombie.NodeKey2)
|
|
isZombie = true
|
|
|
|
return nil
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return false, route.Vertex{}, route.Vertex{},
|
|
fmt.Errorf("%w: %w (chanID=%d)",
|
|
ErrCantCheckIfZombieEdgeStr, err, chanID)
|
|
}
|
|
|
|
return isZombie, pubKey1, pubKey2, nil
|
|
}
|
|
|
|
// NumZombies returns the current number of zombie channels in the graph.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) NumZombies(
|
|
ctx context.Context, v lnwire.GossipVersion,
|
|
) (uint64, error) {
|
|
|
|
var (
|
|
numZombies uint64
|
|
)
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
count, err := db.CountZombieChannels(
|
|
ctx, int16(v),
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to count zombie channels: %w",
|
|
err)
|
|
}
|
|
|
|
numZombies = uint64(count)
|
|
|
|
return nil
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("unable to count zombies: %w", err)
|
|
}
|
|
|
|
return numZombies, nil
|
|
}
|
|
|
|
// DeleteChannelEdges removes edges with the given channel IDs from the
|
|
// database and marks them as zombies. This ensures that we're unable to re-add
|
|
// it to our database once again. If an edge does not exist within the
|
|
// database, then ErrEdgeNotFound will be returned. If strictZombiePruning is
|
|
// true, then when we mark these edges as zombies, we'll set up the keys such
|
|
// that we require the node that failed to send the fresh update to be the one
|
|
// that resurrects the channel from its zombie state. The markZombie bool
|
|
// denotes whether to mark the channel as a zombie.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) DeleteChannelEdges(ctx context.Context,
|
|
v lnwire.GossipVersion, strictZombiePruning, markZombie bool,
|
|
chanIDs ...uint64) (
|
|
[]*models.ChannelEdgeInfo, error) {
|
|
|
|
s.cacheMu.Lock()
|
|
defer s.cacheMu.Unlock()
|
|
|
|
// Keep track of which channels we end up finding so that we can
|
|
// correctly return ErrEdgeNotFound if we do not find a channel.
|
|
chanLookup := make(map[uint64]struct{}, len(chanIDs))
|
|
for _, chanID := range chanIDs {
|
|
chanLookup[chanID] = struct{}{}
|
|
}
|
|
|
|
var edges []*models.ChannelEdgeInfo
|
|
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
|
// First, collect all channel rows.
|
|
var channelRows []sqlc.GetChannelsBySCIDWithPoliciesRow
|
|
chanCallBack := func(ctx context.Context,
|
|
row sqlc.GetChannelsBySCIDWithPoliciesRow) error {
|
|
|
|
// Deleting the entry from the map indicates that we
|
|
// have found the channel.
|
|
scid := byteOrder.Uint64(row.GraphChannel.Scid)
|
|
delete(chanLookup, scid)
|
|
|
|
channelRows = append(channelRows, row)
|
|
|
|
return nil
|
|
}
|
|
|
|
err := s.forEachChanWithPoliciesInSCIDList(
|
|
ctx, db, v, chanCallBack, chanIDs,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if len(chanLookup) > 0 {
|
|
return ErrEdgeNotFound
|
|
}
|
|
|
|
if len(channelRows) == 0 {
|
|
return nil
|
|
}
|
|
|
|
// Batch build all channel edges.
|
|
var chanIDsToDelete []int64
|
|
edges, chanIDsToDelete, err = batchBuildChannelInfo(
|
|
ctx, s.cfg, db, channelRows,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if markZombie {
|
|
for i, row := range channelRows {
|
|
scid := byteOrder.Uint64(row.GraphChannel.Scid)
|
|
|
|
err := handleZombieMarking(
|
|
ctx, db, v, row, edges[i],
|
|
strictZombiePruning, scid,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to mark "+
|
|
"channel as zombie: %w", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
return s.deleteChannels(ctx, db, chanIDsToDelete)
|
|
}, func() {
|
|
edges = nil
|
|
|
|
// Re-fill the lookup map.
|
|
for _, chanID := range chanIDs {
|
|
chanLookup[chanID] = struct{}{}
|
|
}
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to delete channel edges: %w",
|
|
err)
|
|
}
|
|
|
|
for _, chanID := range chanIDs {
|
|
s.rejectCache.remove(v, chanID)
|
|
s.chanCache.remove(v, chanID)
|
|
}
|
|
|
|
return edges, 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 Store interface.
|
|
func (s *SQLStore) FetchChannelEdgesByID(ctx context.Context,
|
|
v lnwire.GossipVersion, chanID uint64) (
|
|
*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
|
*models.ChannelEdgePolicy, error) {
|
|
|
|
var (
|
|
edge *models.ChannelEdgeInfo
|
|
policy1, policy2 *models.ChannelEdgePolicy
|
|
chanIDB = channelIDToBytes(chanID)
|
|
)
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return nil, nil, nil, fmt.Errorf(
|
|
"unsupported gossip version: %d", v,
|
|
)
|
|
}
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
row, err := db.GetChannelBySCIDWithPolicies(
|
|
ctx, sqlc.GetChannelBySCIDWithPoliciesParams{
|
|
Scid: chanIDB,
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
// First check if this edge is perhaps in the zombie
|
|
// index.
|
|
zombie, err := db.GetZombieChannel(
|
|
ctx, sqlc.GetZombieChannelParams{
|
|
Scid: chanIDB,
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return ErrEdgeNotFound
|
|
} else if err != nil {
|
|
return fmt.Errorf("unable to check if "+
|
|
"channel is zombie: %w", err)
|
|
}
|
|
|
|
// At this point, we know the channel is a zombie, so
|
|
// we'll return an error indicating this, and we will
|
|
// populate the edge info with the public keys of each
|
|
// party as this is the only information we have about
|
|
// it.
|
|
node1, err := route.NewVertexFromBytes(zombie.NodeKey1)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
node2, err := route.NewVertexFromBytes(zombie.NodeKey2)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch v {
|
|
case gossipV1:
|
|
edge, err = models.NewV1Channel(
|
|
0, chainhash.Hash{}, node1,
|
|
node2, &models.ChannelV1Fields{},
|
|
)
|
|
case gossipV2:
|
|
edge, err = models.NewV2Channel(
|
|
0, chainhash.Hash{}, node1,
|
|
node2, &models.ChannelV2Fields{},
|
|
)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return ErrZombieEdge
|
|
} else if err != nil {
|
|
return fmt.Errorf("unable to fetch channel: %w", err)
|
|
}
|
|
|
|
node1, node2, err := buildNodeVertices(
|
|
row.GraphNode.PubKey, row.GraphNode_2.PubKey,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
edge, err = getAndBuildEdgeInfo(
|
|
ctx, s.cfg, db, row.GraphChannel, 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, s.cfg.QueryCfg, 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 {
|
|
// If we are returning the ErrZombieEdge, then we also need to
|
|
// return the edge info as the method comment indicates that
|
|
// this will be populated when the edge is a zombie.
|
|
return edge, 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 Store interface.
|
|
func (s *SQLStore) FetchChannelEdgesByOutpoint(ctx context.Context,
|
|
v lnwire.GossipVersion, op *wire.OutPoint) (
|
|
*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
|
*models.ChannelEdgePolicy, error) {
|
|
|
|
var (
|
|
edge *models.ChannelEdgeInfo
|
|
policy1, policy2 *models.ChannelEdgePolicy
|
|
)
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return nil, nil, nil, fmt.Errorf(
|
|
"unsupported gossip version: %d", v,
|
|
)
|
|
}
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
row, err := db.GetChannelByOutpointWithPolicies(
|
|
ctx, sqlc.GetChannelByOutpointWithPoliciesParams{
|
|
Outpoint: op.String(),
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
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, s.cfg, db, row.GraphChannel, 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, s.cfg.QueryCfg, 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
|
|
}
|
|
|
|
// HasV1ChannelEdge 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 Store interface.
|
|
func (s *SQLStore) HasV1ChannelEdge(ctx context.Context,
|
|
chanID uint64) (time.Time, time.Time, bool, bool, error) {
|
|
|
|
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(gossipV1, 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(gossipV1, 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
|
|
}
|
|
|
|
chanIDB := channelIDToBytes(chanID)
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
channel, err := db.GetChannelBySCID(
|
|
ctx, sqlc.GetChannelBySCIDParams{
|
|
Scid: chanIDB,
|
|
Version: int16(gossipV1),
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
// Check if it is a zombie channel.
|
|
isZombie, err = db.IsZombieChannel(
|
|
ctx, sqlc.IsZombieChannelParams{
|
|
Scid: chanIDB,
|
|
Version: int16(gossipV1),
|
|
},
|
|
)
|
|
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(gossipV1),
|
|
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(gossipV1),
|
|
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(
|
|
gossipV1, chanID,
|
|
newRejectCacheEntryV1(
|
|
node1LastUpdate, node2LastUpdate, exists,
|
|
isZombie,
|
|
),
|
|
)
|
|
|
|
return node1LastUpdate, node2LastUpdate, exists, isZombie, nil
|
|
}
|
|
|
|
// HasChannelEdge returns true if the database knows of a channel edge with the
|
|
// passed channel ID and gossip version, and false otherwise. If an edge with
|
|
// that ID is found within the graph, then the zombie index is checked and its
|
|
// result is returned as the second boolean.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) HasChannelEdge(ctx context.Context,
|
|
v lnwire.GossipVersion, chanID uint64) (bool, bool, error) {
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return false, false, fmt.Errorf(
|
|
"unsupported gossip version: %d", v,
|
|
)
|
|
}
|
|
|
|
var (
|
|
exists bool
|
|
isZombie bool
|
|
node1LastUpdate time.Time
|
|
node2LastUpdate time.Time
|
|
node1Block uint32
|
|
node2Block uint32
|
|
)
|
|
|
|
// 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(v, chanID); ok {
|
|
s.cacheMu.RUnlock()
|
|
exists, isZombie = entry.flags.unpack()
|
|
return 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(v, chanID); ok {
|
|
exists, isZombie = entry.flags.unpack()
|
|
return exists, isZombie, nil
|
|
}
|
|
|
|
chanIDB := channelIDToBytes(chanID)
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
channel, err := db.GetChannelBySCID(
|
|
ctx, sqlc.GetChannelBySCIDParams{
|
|
Scid: chanIDB,
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
// Check if it is a zombie channel.
|
|
isZombie, err = db.IsZombieChannel(
|
|
ctx, sqlc.IsZombieChannelParams{
|
|
Scid: chanIDB,
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
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(v),
|
|
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 {
|
|
switch v {
|
|
case gossipV1:
|
|
if policy1.LastUpdate.Valid {
|
|
node1LastUpdate = time.Unix(
|
|
policy1.LastUpdate.Int64, 0,
|
|
)
|
|
}
|
|
case gossipV2:
|
|
if policy1.BlockHeight.Valid {
|
|
node1Block = uint32(
|
|
policy1.BlockHeight.Int64,
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
policy2, err := db.GetChannelPolicyByChannelAndNode(
|
|
ctx, sqlc.GetChannelPolicyByChannelAndNodeParams{
|
|
Version: int16(v),
|
|
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 {
|
|
switch v {
|
|
case gossipV1:
|
|
if policy2.LastUpdate.Valid {
|
|
node2LastUpdate = time.Unix(
|
|
policy2.LastUpdate.Int64, 0,
|
|
)
|
|
}
|
|
case gossipV2:
|
|
if policy2.BlockHeight.Valid {
|
|
node2Block = uint32(
|
|
policy2.BlockHeight.Int64,
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return false, false,
|
|
fmt.Errorf("unable to fetch channel: %w", err)
|
|
}
|
|
|
|
var entry rejectCacheEntry
|
|
switch v {
|
|
case gossipV1:
|
|
entry = newRejectCacheEntryV1(
|
|
node1LastUpdate, node2LastUpdate, exists, isZombie,
|
|
)
|
|
case gossipV2:
|
|
entry = newRejectCacheEntryV2(
|
|
node1Block, node2Block, exists, isZombie,
|
|
)
|
|
}
|
|
s.rejectCache.insert(v, chanID, entry)
|
|
|
|
return 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 Store interface.
|
|
func (s *SQLStore) ChannelID(ctx context.Context, v lnwire.GossipVersion,
|
|
chanPoint *wire.OutPoint) (uint64, error) {
|
|
|
|
var (
|
|
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(v),
|
|
},
|
|
)
|
|
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 Store interface.
|
|
func (s *SQLStore) IsPublicNode(ctx context.Context, v lnwire.GossipVersion,
|
|
pubKey [33]byte) (bool, error) {
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return false, fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
|
|
var isPublic bool
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
var err error
|
|
switch v {
|
|
case gossipV1:
|
|
isPublic, err = db.IsPublicV1Node(ctx, pubKey[:])
|
|
case gossipV2:
|
|
isPublic, err = db.IsPublicV2Node(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 Store interface.
|
|
func (s *SQLStore) FetchChanInfos(ctx context.Context,
|
|
v lnwire.GossipVersion, chanIDs []uint64) ([]ChannelEdge, error) {
|
|
|
|
var (
|
|
edges = make(map[uint64]ChannelEdge)
|
|
)
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
if !isKnownGossipVersion(v) {
|
|
return fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
|
|
// First, collect all channel rows.
|
|
var channelRows []sqlc.GetChannelsBySCIDWithPoliciesRow
|
|
chanCallBack := func(ctx context.Context,
|
|
row sqlc.GetChannelsBySCIDWithPoliciesRow) error {
|
|
|
|
channelRows = append(channelRows, row)
|
|
return nil
|
|
}
|
|
|
|
err := s.forEachChanWithPoliciesInSCIDList(
|
|
ctx, db, v, chanCallBack, chanIDs,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if len(channelRows) == 0 {
|
|
return nil
|
|
}
|
|
|
|
// Batch build all channel edges.
|
|
chans, err := batchBuildChannelEdges(
|
|
ctx, s.cfg, db, channelRows,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to build channel edges: %w",
|
|
err)
|
|
}
|
|
|
|
for _, c := range chans {
|
|
edges[c.Info.ChannelID] = c
|
|
}
|
|
|
|
return err
|
|
}, func() {
|
|
clear(edges)
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to fetch channels: %w", err)
|
|
}
|
|
|
|
res := make([]ChannelEdge, 0, len(edges))
|
|
for _, chanID := range chanIDs {
|
|
edge, ok := edges[chanID]
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
res = append(res, edge)
|
|
}
|
|
|
|
return res, nil
|
|
}
|
|
|
|
// forEachChanWithPoliciesInSCIDList is a wrapper around the
|
|
// GetChannelsBySCIDWithPolicies query that allows us to iterate through
|
|
// channels in a paginated manner.
|
|
func (s *SQLStore) forEachChanWithPoliciesInSCIDList(ctx context.Context,
|
|
db SQLQueries, v lnwire.GossipVersion, cb func(ctx context.Context,
|
|
row sqlc.GetChannelsBySCIDWithPoliciesRow) error,
|
|
chanIDs []uint64) error {
|
|
|
|
queryWrapper := func(ctx context.Context,
|
|
scids [][]byte) ([]sqlc.GetChannelsBySCIDWithPoliciesRow,
|
|
error) {
|
|
|
|
return db.GetChannelsBySCIDWithPolicies(
|
|
ctx, sqlc.GetChannelsBySCIDWithPoliciesParams{
|
|
Version: int16(v),
|
|
Scids: scids,
|
|
},
|
|
)
|
|
}
|
|
|
|
return sqldb.ExecuteBatchQuery(
|
|
ctx, s.cfg.QueryCfg, chanIDs, channelIDToBytes, queryWrapper,
|
|
cb,
|
|
)
|
|
}
|
|
|
|
// 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 Store interface.
|
|
func (s *SQLStore) FilterKnownChanIDs(ctx context.Context,
|
|
v lnwire.GossipVersion,
|
|
chansInfo []ChannelUpdateInfo) ([]uint64, []ChannelUpdateInfo, error) {
|
|
|
|
var (
|
|
newChanIDs []uint64
|
|
knownZombies []ChannelUpdateInfo
|
|
infoLookup = make(
|
|
map[uint64]ChannelUpdateInfo, len(chansInfo),
|
|
)
|
|
)
|
|
|
|
// We first build a lookup map of the channel ID's to the
|
|
// ChannelUpdateInfo. This allows us to quickly delete channels that we
|
|
// already know about.
|
|
for _, chanInfo := range chansInfo {
|
|
infoLookup[chanInfo.ShortChannelID.ToUint64()] = chanInfo
|
|
}
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
// The call-back function deletes known channels from
|
|
// infoLookup, so that we can later check which channels are
|
|
// zombies by only looking at the remaining channels in the
|
|
// set.
|
|
cb := func(ctx context.Context,
|
|
channel sqlc.GraphChannel) error {
|
|
|
|
delete(infoLookup, byteOrder.Uint64(channel.Scid))
|
|
|
|
return nil
|
|
}
|
|
|
|
err := s.forEachChanInSCIDList(
|
|
ctx, db, v, cb, chansInfo,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to iterate "+
|
|
"channels: %w", err)
|
|
}
|
|
|
|
// We want to ensure that we deal with the channels in the
|
|
// same order that they were passed in, so we iterate over
|
|
// the original chansInfo slice and then check if that
|
|
// channel is still in the infoLookup map.
|
|
for _, chanInfo := range chansInfo {
|
|
channelID := chanInfo.ShortChannelID.ToUint64()
|
|
if _, ok := infoLookup[channelID]; !ok {
|
|
continue
|
|
}
|
|
|
|
isZombie, err := db.IsZombieChannel(
|
|
ctx, sqlc.IsZombieChannelParams{
|
|
Scid: channelIDToBytes(channelID),
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
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
|
|
// Rebuild the infoLookup map in case of a rollback.
|
|
for _, chanInfo := range chansInfo {
|
|
scid := chanInfo.ShortChannelID.ToUint64()
|
|
infoLookup[scid] = chanInfo
|
|
}
|
|
})
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("unable to fetch channels: %w", err)
|
|
}
|
|
|
|
return newChanIDs, knownZombies, nil
|
|
}
|
|
|
|
// forEachChanInSCIDList is a helper method that executes a paged query
|
|
// against the database to fetch all channels of the given gossip version that
|
|
// match the passed ChannelUpdateInfo slice. The callback function is called
|
|
// for each channel that is found.
|
|
func (s *SQLStore) forEachChanInSCIDList(ctx context.Context, db SQLQueries,
|
|
v lnwire.GossipVersion,
|
|
cb func(ctx context.Context, channel sqlc.GraphChannel) error,
|
|
chansInfo []ChannelUpdateInfo) error {
|
|
|
|
queryWrapper := func(ctx context.Context,
|
|
scids [][]byte) ([]sqlc.GraphChannel, error) {
|
|
|
|
return db.GetChannelsBySCIDs(
|
|
ctx, sqlc.GetChannelsBySCIDsParams{
|
|
Version: int16(v),
|
|
Scids: scids,
|
|
},
|
|
)
|
|
}
|
|
|
|
chanIDConverter := func(chanInfo ChannelUpdateInfo) []byte {
|
|
channelID := chanInfo.ShortChannelID.ToUint64()
|
|
|
|
return channelIDToBytes(channelID)
|
|
}
|
|
|
|
return sqldb.ExecuteBatchQuery(
|
|
ctx, s.cfg.QueryCfg, chansInfo, chanIDConverter, queryWrapper,
|
|
cb,
|
|
)
|
|
}
|
|
|
|
// PruneGraphNodes is a garbage collection method which attempts to prune out
|
|
// any nodes from the channel graph that are currently unconnected. This ensure
|
|
// that we only maintain a graph of reachable nodes. In the event that a pruned
|
|
// node gains more channels, it will be re-added back to the graph.
|
|
//
|
|
// NOTE: this prunes nodes across protocol versions. It will never prune the
|
|
// source nodes.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) PruneGraphNodes(ctx context.Context) (
|
|
[]route.Vertex, error) {
|
|
|
|
var prunedNodes []route.Vertex
|
|
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
|
var err error
|
|
prunedNodes, err = s.pruneGraphNodes(ctx, db)
|
|
|
|
return err
|
|
}, func() {
|
|
prunedNodes = nil
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to prune nodes: %w", err)
|
|
}
|
|
|
|
return prunedNodes, nil
|
|
}
|
|
|
|
// PruneGraph prunes newly closed channels from the channel graph in response
|
|
// to a new block being solved on the network. Any transactions which spend the
|
|
// funding output of any known channels within he graph will be deleted.
|
|
// Additionally, the "prune tip", or the last block which has been used to
|
|
// prune the graph is stored so callers can ensure the graph is fully in sync
|
|
// with the current UTXO state. A slice of channels that have been closed by
|
|
// the target block along with any pruned nodes are returned if the function
|
|
// succeeds without error.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) PruneGraph(ctx context.Context,
|
|
spentOutputs []*wire.OutPoint, blockHash *chainhash.Hash,
|
|
blockHeight uint32) ([]*models.ChannelEdgeInfo, []route.Vertex,
|
|
error) {
|
|
|
|
s.cacheMu.Lock()
|
|
defer s.cacheMu.Unlock()
|
|
|
|
var (
|
|
closedChans []*models.ChannelEdgeInfo
|
|
prunedNodes []route.Vertex
|
|
)
|
|
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
|
// First, collect all channel rows that need to be pruned.
|
|
var channelRows []sqlc.GetChannelsByOutpointsRow
|
|
channelCallback := func(ctx context.Context,
|
|
row sqlc.GetChannelsByOutpointsRow) error {
|
|
|
|
channelRows = append(channelRows, row)
|
|
|
|
return nil
|
|
}
|
|
|
|
err := s.forEachChanInOutpoints(
|
|
ctx, db, spentOutputs, channelCallback,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch channels by "+
|
|
"outpoints: %w", err)
|
|
}
|
|
|
|
if len(channelRows) == 0 {
|
|
// There are no channels to prune. So we can exit early
|
|
// after updating the prune log.
|
|
err = db.UpsertPruneLogEntry(
|
|
ctx, sqlc.UpsertPruneLogEntryParams{
|
|
BlockHash: blockHash[:],
|
|
BlockHeight: int64(blockHeight),
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to insert prune log "+
|
|
"entry: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Batch build all channel edges for pruning.
|
|
var chansToDelete []int64
|
|
closedChans, chansToDelete, err = batchBuildChannelInfo(
|
|
ctx, s.cfg, db, channelRows,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = s.deleteChannels(ctx, db, chansToDelete)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to delete channels: %w", err)
|
|
}
|
|
|
|
err = db.UpsertPruneLogEntry(
|
|
ctx, sqlc.UpsertPruneLogEntryParams{
|
|
BlockHash: blockHash[:],
|
|
BlockHeight: int64(blockHeight),
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to insert prune log "+
|
|
"entry: %w", err)
|
|
}
|
|
|
|
// Now that we've pruned some channels, we'll also prune any
|
|
// nodes that no longer have any channels.
|
|
prunedNodes, err = s.pruneGraphNodes(ctx, db)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to prune graph nodes: %w",
|
|
err)
|
|
}
|
|
|
|
return nil
|
|
}, func() {
|
|
prunedNodes = nil
|
|
closedChans = nil
|
|
})
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("unable to prune graph: %w", err)
|
|
}
|
|
|
|
for _, channel := range closedChans {
|
|
s.rejectCache.remove(channel.Version, channel.ChannelID)
|
|
s.chanCache.remove(channel.Version, channel.ChannelID)
|
|
}
|
|
|
|
return closedChans, prunedNodes, nil
|
|
}
|
|
|
|
// forEachChanInOutpoints is a helper function that executes a paginated
|
|
// query to fetch channels by their outpoints and applies the given call-back
|
|
// to each.
|
|
//
|
|
// NOTE: this fetches channels for all protocol versions.
|
|
func (s *SQLStore) forEachChanInOutpoints(ctx context.Context, db SQLQueries,
|
|
outpoints []*wire.OutPoint, cb func(ctx context.Context,
|
|
row sqlc.GetChannelsByOutpointsRow) error) error {
|
|
|
|
// Create a wrapper that uses the transaction's db instance to execute
|
|
// the query.
|
|
queryWrapper := func(ctx context.Context,
|
|
pageOutpoints []string) ([]sqlc.GetChannelsByOutpointsRow,
|
|
error) {
|
|
|
|
return db.GetChannelsByOutpoints(ctx, pageOutpoints)
|
|
}
|
|
|
|
// Define the conversion function from Outpoint to string.
|
|
outpointToString := func(outpoint *wire.OutPoint) string {
|
|
return outpoint.String()
|
|
}
|
|
|
|
return sqldb.ExecuteBatchQuery(
|
|
ctx, s.cfg.QueryCfg, outpoints, outpointToString,
|
|
queryWrapper, cb,
|
|
)
|
|
}
|
|
|
|
func (s *SQLStore) deleteChannels(ctx context.Context, db SQLQueries,
|
|
dbIDs []int64) error {
|
|
|
|
// Create a wrapper that uses the transaction's db instance to execute
|
|
// the query.
|
|
queryWrapper := func(ctx context.Context, ids []int64) ([]any, error) {
|
|
return nil, db.DeleteChannels(ctx, ids)
|
|
}
|
|
|
|
idConverter := func(id int64) int64 {
|
|
return id
|
|
}
|
|
|
|
return sqldb.ExecuteBatchQuery(
|
|
ctx, s.cfg.QueryCfg, dbIDs, idConverter,
|
|
queryWrapper, func(ctx context.Context, _ any) error {
|
|
return nil
|
|
},
|
|
)
|
|
}
|
|
|
|
// ChannelView returns the verifiable edge information for each active channel
|
|
// within the known channel graph. The set of UTXOs (along with their scripts)
|
|
// returned are the ones that need to be watched on chain to detect channel
|
|
// closes on the resident blockchain.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) ChannelView(ctx context.Context,
|
|
v lnwire.GossipVersion) ([]EdgePoint, error) {
|
|
|
|
var edgePoints []EdgePoint
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
switch v {
|
|
case gossipV1:
|
|
handleChannel := func(_ context.Context,
|
|
channel sqlc.ListChannelsPaginatedRow,
|
|
chanFeats map[int64][]int) error {
|
|
|
|
key1, err := route.NewVertexFromBytes(
|
|
channel.BitcoinKey1,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
key2, err := route.NewVertexFromBytes(
|
|
channel.BitcoinKey2,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Private taproot channels are currently stored
|
|
// as simple v1 channels that only need the
|
|
// taproot staging bit to reconstruct the BIP86
|
|
// funding script. They do not carry a custom
|
|
// tapscript root on this path.
|
|
//
|
|
// TODO: Remove this v1 feature-bit workaround
|
|
// once private taproot channels have been
|
|
// migrated to v2 gossip objects.
|
|
feats := lnwire.EmptyFeatureVector()
|
|
bits := chanFeats[channel.ID]
|
|
for _, bit := range bits {
|
|
feats.Set(lnwire.FeatureBit(bit))
|
|
}
|
|
|
|
edge := &models.ChannelEdgeInfo{
|
|
Version: gossipV1,
|
|
BitcoinKey1Bytes: fn.Some(key1),
|
|
BitcoinKey2Bytes: fn.Some(key2),
|
|
Features: feats,
|
|
}
|
|
|
|
pkScript, err := edge.FundingPKScript()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
op, err := wire.NewOutPointFromString(
|
|
channel.Outpoint,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
edgePoints = append(edgePoints, EdgePoint{
|
|
FundingPkScript: pkScript,
|
|
OutPoint: *op,
|
|
})
|
|
|
|
return nil
|
|
}
|
|
|
|
queryFunc := func(ctx context.Context, lastID int64,
|
|
limit int32) ([]sqlc.ListChannelsPaginatedRow,
|
|
error) {
|
|
|
|
return db.ListChannelsPaginated(
|
|
ctx, sqlc.ListChannelsPaginatedParams{
|
|
Version: int16(gossipV1),
|
|
ID: lastID,
|
|
Limit: limit,
|
|
},
|
|
)
|
|
}
|
|
|
|
extractCursor := func(
|
|
row sqlc.ListChannelsPaginatedRow) int64 {
|
|
|
|
return row.ID
|
|
}
|
|
|
|
collectID := func(
|
|
row sqlc.ListChannelsPaginatedRow) (int64,
|
|
error) {
|
|
|
|
return row.ID, nil
|
|
}
|
|
|
|
loadChannelFeatures := func(ctx context.Context,
|
|
chanIDs []int64) (map[int64][]int, error) {
|
|
|
|
return batchLoadChannelFeaturesHelper(
|
|
ctx, s.cfg.QueryCfg, db, chanIDs,
|
|
)
|
|
}
|
|
|
|
return sqldb.ExecuteCollectAndBatchWithSharedDataQuery(
|
|
ctx, s.cfg.QueryCfg, int64(-1), queryFunc,
|
|
extractCursor, collectID, loadChannelFeatures,
|
|
handleChannel,
|
|
)
|
|
|
|
case gossipV2:
|
|
handleChannel := func(_ context.Context,
|
|
channel sqlc.ListChannelsPaginatedV2Row) error {
|
|
|
|
op, err := wire.NewOutPointFromString(
|
|
channel.Outpoint,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
pkScript := channel.FundingPkScript
|
|
edgePoints = append(edgePoints, EdgePoint{
|
|
FundingPkScript: pkScript,
|
|
OutPoint: *op,
|
|
})
|
|
|
|
return nil
|
|
}
|
|
|
|
queryFunc := func(ctx context.Context, lastID int64,
|
|
limit int32) ([]sqlc.ListChannelsPaginatedV2Row,
|
|
error) {
|
|
|
|
return db.ListChannelsPaginatedV2(
|
|
ctx, sqlc.ListChannelsPaginatedV2Params{
|
|
ID: lastID,
|
|
Limit: limit,
|
|
},
|
|
)
|
|
}
|
|
|
|
extractCursor := func(
|
|
row sqlc.ListChannelsPaginatedV2Row) int64 {
|
|
|
|
return row.ID
|
|
}
|
|
|
|
return sqldb.ExecutePaginatedQuery(
|
|
ctx, s.cfg.QueryCfg, int64(-1), queryFunc,
|
|
extractCursor, handleChannel,
|
|
)
|
|
|
|
default:
|
|
return fmt.Errorf("unsupported gossip version: %d", v)
|
|
}
|
|
}, func() {
|
|
edgePoints = nil
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to fetch channel view: %w", err)
|
|
}
|
|
|
|
return edgePoints, nil
|
|
}
|
|
|
|
// PruneTip returns the block height and hash of the latest block that has been
|
|
// used to prune channels in the graph. Knowing the "prune tip" allows callers
|
|
// to tell if the graph is currently in sync with the current best known UTXO
|
|
// state.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) PruneTip(ctx context.Context) (*chainhash.Hash, uint32,
|
|
error) {
|
|
|
|
var (
|
|
tipHash chainhash.Hash
|
|
tipHeight uint32
|
|
)
|
|
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
|
pruneTip, err := db.GetPruneTip(ctx)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return ErrGraphNeverPruned
|
|
} else if err != nil {
|
|
return fmt.Errorf("unable to fetch prune tip: %w", err)
|
|
}
|
|
|
|
tipHash = chainhash.Hash(pruneTip.BlockHash)
|
|
tipHeight = uint32(pruneTip.BlockHeight)
|
|
|
|
return nil
|
|
}, sqldb.NoOpReset)
|
|
if err != nil {
|
|
return nil, 0, err
|
|
}
|
|
|
|
return &tipHash, tipHeight, nil
|
|
}
|
|
|
|
// pruneGraphNodes deletes any node in the DB that doesn't have a channel.
|
|
//
|
|
// NOTE: this prunes nodes across protocol versions. It will never prune the
|
|
// source nodes.
|
|
func (s *SQLStore) pruneGraphNodes(ctx context.Context,
|
|
db SQLQueries) ([]route.Vertex, error) {
|
|
|
|
nodeKeys, err := db.DeleteUnconnectedNodes(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to delete unconnected "+
|
|
"nodes: %w", err)
|
|
}
|
|
|
|
prunedNodes := make([]route.Vertex, len(nodeKeys))
|
|
for i, nodeKey := range nodeKeys {
|
|
pub, err := route.NewVertexFromBytes(nodeKey)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to parse pubkey "+
|
|
"from bytes: %w", err)
|
|
}
|
|
|
|
prunedNodes[i] = pub
|
|
}
|
|
|
|
return prunedNodes, nil
|
|
}
|
|
|
|
// DisconnectBlockAtHeight is used to indicate that the block specified
|
|
// by the passed height has been disconnected from the main chain. This
|
|
// will "rewind" the graph back to the height below, deleting channels
|
|
// that are no longer confirmed from the graph. The prune log will be
|
|
// set to the last prune height valid for the remaining chain.
|
|
// Channels that were removed from the graph resulting from the
|
|
// disconnected block are returned.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) DisconnectBlockAtHeight(ctx context.Context,
|
|
height uint32) ([]*models.ChannelEdgeInfo, error) {
|
|
|
|
var (
|
|
// Every channel having a ShortChannelID starting at 'height'
|
|
// will no longer be confirmed.
|
|
startShortChanID = lnwire.ShortChannelID{
|
|
BlockHeight: height,
|
|
}
|
|
|
|
// Delete everything after this height from the db up until the
|
|
// SCID alias range.
|
|
endShortChanID = aliasmgr.StartingAlias
|
|
|
|
removedChans []*models.ChannelEdgeInfo
|
|
|
|
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,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch channels: %w", err)
|
|
}
|
|
|
|
if len(rows) == 0 {
|
|
// No channels to disconnect, but still clean up prune
|
|
// log.
|
|
return db.DeletePruneLogEntriesInRange(
|
|
ctx, sqlc.DeletePruneLogEntriesInRangeParams{
|
|
StartHeight: int64(height),
|
|
EndHeight: int64(
|
|
endShortChanID.BlockHeight,
|
|
),
|
|
},
|
|
)
|
|
}
|
|
|
|
// Batch build all channel edges for disconnection.
|
|
channelEdges, chanIDsToDelete, err := batchBuildChannelInfo(
|
|
ctx, s.cfg, db, rows,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
removedChans = channelEdges
|
|
|
|
err = s.deleteChannels(ctx, db, chanIDsToDelete)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to delete channels: %w", err)
|
|
}
|
|
|
|
return db.DeletePruneLogEntriesInRange(
|
|
ctx, sqlc.DeletePruneLogEntriesInRangeParams{
|
|
StartHeight: int64(height),
|
|
EndHeight: int64(endShortChanID.BlockHeight),
|
|
},
|
|
)
|
|
}, func() {
|
|
removedChans = nil
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to disconnect block at "+
|
|
"height: %w", err)
|
|
}
|
|
|
|
s.cacheMu.Lock()
|
|
for _, channel := range removedChans {
|
|
s.rejectCache.remove(channel.Version, channel.ChannelID)
|
|
s.chanCache.remove(channel.Version, channel.ChannelID)
|
|
}
|
|
s.cacheMu.Unlock()
|
|
|
|
return removedChans, nil
|
|
}
|
|
|
|
// AddEdgeProof sets the proof of an existing edge in the graph database.
|
|
//
|
|
// NOTE: part of the Store interface.
|
|
func (s *SQLStore) AddEdgeProof(ctx context.Context,
|
|
scid lnwire.ShortChannelID, proof *models.ChannelAuthProof) error {
|
|
|
|
if !isKnownGossipVersion(proof.Version) {
|
|
return fmt.Errorf("unsupported gossip version: %d",
|
|
proof.Version)
|
|
}
|
|
|
|
var (
|
|
scidBytes = channelIDToBytes(scid.ToUint64())
|
|
)
|
|
|
|
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
|
var (
|
|
res sql.Result
|
|
err error
|
|
)
|
|
switch proof.Version {
|
|
case gossipV1:
|
|
res, err = db.AddV1ChannelProof(
|
|
ctx, sqlc.AddV1ChannelProofParams{
|
|
Scid: scidBytes,
|
|
Node1Signature: proof.NodeSig1(),
|
|
Node2Signature: proof.NodeSig2(),
|
|
Bitcoin1Signature: proof.BitcoinSig1(),
|
|
Bitcoin2Signature: proof.BitcoinSig2(),
|
|
},
|
|
)
|
|
|
|
case gossipV2:
|
|
res, err = db.AddV2ChannelProof(
|
|
ctx, sqlc.AddV2ChannelProofParams{
|
|
Scid: scidBytes,
|
|
Signature: proof.Sig(),
|
|
},
|
|
)
|
|
|
|
default:
|
|
return fmt.Errorf("unsupported gossip version: %d",
|
|
proof.Version)
|
|
}
|
|
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 Store interface.
|
|
func (s *SQLStore) PutClosedScid(ctx context.Context,
|
|
scid lnwire.ShortChannelID) error {
|
|
|
|
var (
|
|
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 Store interface.
|
|
func (s *SQLStore) IsClosedScid(ctx context.Context,
|
|
scid lnwire.ShortChannelID) (bool, error) {
|
|
|
|
var (
|
|
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 Store interface.
|
|
func (s *SQLStore) GraphSession(ctx context.Context,
|
|
cb func(graph NodeTraverser) error, reset func()) error {
|
|
|
|
return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
return cb(newSQLNodeTraverser(db, s.cfg.ChainHash))
|
|
}, reset)
|
|
}
|
|
|
|
// 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(
|
|
ctx context.Context, nodePub route.Vertex,
|
|
cb func(channel *DirectedChannel) error, _ func()) error {
|
|
|
|
return forEachNodeDirectedChannel(
|
|
ctx, s.db, lnwire.GossipVersion1, 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(ctx context.Context,
|
|
nodePub route.Vertex) (
|
|
*lnwire.FeatureVector, error) {
|
|
|
|
return fetchNodeFeatures(ctx, s.db, lnwire.GossipVersion1, 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
|
|
// returned.
|
|
func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries,
|
|
v lnwire.GossipVersion, nodePub route.Vertex,
|
|
cb func(channel *DirectedChannel) error) error {
|
|
|
|
toNodeCallback := func() route.Vertex {
|
|
return nodePub
|
|
}
|
|
|
|
dbID, err := db.GetNodeIDByPubKey(
|
|
ctx, sqlc.GetNodeIDByPubKeyParams{
|
|
Version: int16(v),
|
|
PubKey: nodePub[:],
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
} else if err != nil {
|
|
return fmt.Errorf("unable to fetch node: %w", err)
|
|
}
|
|
|
|
rows, err := db.ListChannelsByNodeID(
|
|
ctx, sqlc.ListChannelsByNodeIDParams{
|
|
Version: int16(v),
|
|
NodeID1: dbID,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch channels: %w", err)
|
|
}
|
|
|
|
// Exit early if there are no channels for this node so we don't
|
|
// do the unnecessary feature fetching.
|
|
if len(rows) == 0 {
|
|
return nil
|
|
}
|
|
|
|
features, err := getNodeFeatures(ctx, db, dbID)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch node features: %w", err)
|
|
}
|
|
|
|
for _, row := range rows {
|
|
node1, node2, err := buildNodeVertices(
|
|
row.Node1Pubkey, row.Node2Pubkey,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to build node vertices: %w",
|
|
err)
|
|
}
|
|
|
|
edge := buildCacheableChannelInfo(
|
|
row.GraphChannel.Scid, row.GraphChannel.Capacity.Int64,
|
|
node1, node2,
|
|
)
|
|
|
|
dbPol1, dbPol2, err := extractChannelPolicies(row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
p1, p2, err := buildCachedChanPolicies(
|
|
dbPol1, dbPol2, edge.ChannelID, node1, node2,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Determine the outgoing and incoming policy for this
|
|
// channel and node combo.
|
|
outPolicy, inPolicy := p1, p2
|
|
if p1 != nil && node2 == nodePub {
|
|
outPolicy, inPolicy = p2, p1
|
|
} else if p2 != nil && node1 != nodePub {
|
|
outPolicy, inPolicy = p2, p1
|
|
}
|
|
|
|
var cachedInPolicy *models.CachedEdgePolicy
|
|
if inPolicy != nil {
|
|
cachedInPolicy = inPolicy
|
|
cachedInPolicy.ToNodePubKey = toNodeCallback
|
|
cachedInPolicy.ToNodeFeatures = features
|
|
}
|
|
|
|
directedChannel := &DirectedChannel{
|
|
ChannelID: edge.ChannelID,
|
|
IsNode1: nodePub == edge.NodeKey1Bytes,
|
|
OtherNode: edge.NodeKey2Bytes,
|
|
Capacity: edge.Capacity,
|
|
OutPolicySet: outPolicy != nil,
|
|
InPolicy: cachedInPolicy,
|
|
}
|
|
if outPolicy != nil {
|
|
outPolicy.InboundFee.WhenSome(func(fee lnwire.Fee) {
|
|
directedChannel.InboundFee = fee
|
|
})
|
|
}
|
|
|
|
if nodePub == edge.NodeKey2Bytes {
|
|
directedChannel.OtherNode = edge.NodeKey1Bytes
|
|
}
|
|
|
|
if err := cb(directedChannel); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// forEachNodeCacheable fetches all node IDs and pub keys from the database,
|
|
// and executes the provided callback for each node. It does so via pagination
|
|
// along with batch loading of the node feature bits.
|
|
func forEachNodeCacheable(ctx context.Context, cfg *sqldb.QueryConfig,
|
|
db SQLQueries, v lnwire.GossipVersion,
|
|
processNode func(nodeID int64, nodePub route.Vertex,
|
|
features *lnwire.FeatureVector) error) error {
|
|
|
|
handleNode := func(_ context.Context,
|
|
dbNode sqlc.ListNodeIDsAndPubKeysRow,
|
|
featureBits map[int64][]int) error {
|
|
|
|
fv := lnwire.EmptyFeatureVector()
|
|
if features, exists := featureBits[dbNode.ID]; exists {
|
|
for _, bit := range features {
|
|
fv.Set(lnwire.FeatureBit(bit))
|
|
}
|
|
}
|
|
|
|
var pub route.Vertex
|
|
copy(pub[:], dbNode.PubKey)
|
|
|
|
return processNode(dbNode.ID, pub, fv)
|
|
}
|
|
|
|
queryFunc := func(ctx context.Context, lastID int64,
|
|
limit int32) ([]sqlc.ListNodeIDsAndPubKeysRow, error) {
|
|
|
|
return db.ListNodeIDsAndPubKeys(
|
|
ctx, sqlc.ListNodeIDsAndPubKeysParams{
|
|
Version: int16(v),
|
|
ID: lastID,
|
|
Limit: limit,
|
|
},
|
|
)
|
|
}
|
|
|
|
extractCursor := func(row sqlc.ListNodeIDsAndPubKeysRow) int64 {
|
|
return row.ID
|
|
}
|
|
|
|
collectFunc := func(node sqlc.ListNodeIDsAndPubKeysRow) (int64, error) {
|
|
return node.ID, nil
|
|
}
|
|
|
|
batchQueryFunc := func(ctx context.Context,
|
|
nodeIDs []int64) (map[int64][]int, error) {
|
|
|
|
return batchLoadNodeFeaturesHelper(ctx, cfg, db, nodeIDs)
|
|
}
|
|
|
|
return sqldb.ExecuteCollectAndBatchWithSharedDataQuery(
|
|
ctx, cfg, int64(-1), queryFunc, extractCursor, collectFunc,
|
|
batchQueryFunc, handleNode,
|
|
)
|
|
}
|
|
|
|
// forEachNodeChannel iterates through all channels of a node, executing
|
|
// the passed callback on each. The call-back is provided with the channel's
|
|
// edge information, the outgoing policy and the incoming policy for the
|
|
// channel and node combo.
|
|
func forEachNodeChannel(ctx context.Context, db SQLQueries,
|
|
cfg *SQLStoreConfig, v lnwire.GossipVersion, id int64,
|
|
cb func(*models.ChannelEdgeInfo,
|
|
*models.ChannelEdgePolicy,
|
|
*models.ChannelEdgePolicy) error) error {
|
|
|
|
// Get all the channels for this node.
|
|
rows, err := db.ListChannelsByNodeID(
|
|
ctx, sqlc.ListChannelsByNodeIDParams{
|
|
Version: int16(v),
|
|
NodeID1: id,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to fetch channels: %w", err)
|
|
}
|
|
|
|
// Collect all the channel and policy IDs.
|
|
var (
|
|
chanIDs = make([]int64, 0, len(rows))
|
|
policyIDs = make([]int64, 0, 2*len(rows))
|
|
)
|
|
for _, row := range rows {
|
|
chanIDs = append(chanIDs, row.GraphChannel.ID)
|
|
|
|
if row.Policy1ID.Valid {
|
|
policyIDs = append(policyIDs, row.Policy1ID.Int64)
|
|
}
|
|
if row.Policy2ID.Valid {
|
|
policyIDs = append(policyIDs, row.Policy2ID.Int64)
|
|
}
|
|
}
|
|
|
|
batchData, err := batchLoadChannelData(
|
|
ctx, cfg.QueryCfg, db, chanIDs, policyIDs,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to batch load channel data: %w", err)
|
|
}
|
|
|
|
// Call the call-back for each channel and its known policies.
|
|
for _, row := range rows {
|
|
node1, node2, err := buildNodeVertices(
|
|
row.Node1Pubkey, row.Node2Pubkey,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to build node vertices: %w",
|
|
err)
|
|
}
|
|
|
|
edge, err := buildEdgeInfoWithBatchData(
|
|
cfg.ChainHash, row.GraphChannel, node1, node2,
|
|
batchData,
|
|
)
|
|
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 := buildChanPoliciesWithBatchData(
|
|
dbPol1, dbPol2, edge.ChannelID, node1, node2, batchData,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to build channel "+
|
|
"policies: %w", err)
|
|
}
|
|
|
|
// Determine the outgoing and incoming policy for this
|
|
// channel and node combo.
|
|
p1ToNode := row.GraphChannel.NodeID2
|
|
p2ToNode := row.GraphChannel.NodeID1
|
|
outPolicy, inPolicy := p1, p2
|
|
if (p1 != nil && p1ToNode == id) ||
|
|
(p2 != nil && p2ToNode != id) {
|
|
|
|
outPolicy, inPolicy = p2, p1
|
|
}
|
|
|
|
if err := cb(edge, outPolicy, inPolicy); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// updateChanEdgePolicy upserts the channel policy info we have stored for
|
|
// a channel we already know of.
|
|
func updateChanEdgePolicy(ctx context.Context, tx SQLQueries,
|
|
edge *models.ChannelEdgePolicy) (route.Vertex, route.Vertex, bool,
|
|
error) {
|
|
|
|
var (
|
|
node1Pub, node2Pub route.Vertex
|
|
chanIDB = channelIDToBytes(edge.ChannelID)
|
|
version = edge.Version
|
|
)
|
|
|
|
if !isKnownGossipVersion(version) {
|
|
return node1Pub, node2Pub, false, fmt.Errorf(
|
|
"unsupported gossip version: %d", version,
|
|
)
|
|
}
|
|
|
|
// Check that this edge policy refers to a channel that we already
|
|
// know of. We do this explicitly so that we can return the appropriate
|
|
// ErrEdgeNotFound error if the channel doesn't exist, rather than
|
|
// abort the transaction which would abort the entire batch.
|
|
dbChan, err := tx.GetChannelAndNodesBySCID(
|
|
ctx, sqlc.GetChannelAndNodesBySCIDParams{
|
|
Scid: chanIDB,
|
|
Version: int16(version),
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return node1Pub, node2Pub, false, ErrEdgeNotFound
|
|
} else if err != nil {
|
|
return node1Pub, node2Pub, false, fmt.Errorf("unable to "+
|
|
"fetch channel(%v): %w", edge.ChannelID, err)
|
|
}
|
|
|
|
copy(node1Pub[:], dbChan.Node1PubKey)
|
|
copy(node2Pub[:], dbChan.Node2PubKey)
|
|
|
|
// Figure out which node this edge is from.
|
|
isNode1 := edge.IsNode1()
|
|
nodeID := dbChan.NodeID1
|
|
if !isNode1 {
|
|
nodeID = dbChan.NodeID2
|
|
}
|
|
|
|
var (
|
|
inboundBase sql.NullInt64
|
|
inboundRate sql.NullInt64
|
|
)
|
|
edge.InboundFee.WhenSome(func(fee lnwire.Fee) {
|
|
inboundRate = sqldb.SQLInt64(fee.FeeRate)
|
|
inboundBase = sqldb.SQLInt64(fee.BaseFee)
|
|
})
|
|
|
|
params := sqlc.UpsertEdgePolicyParams{
|
|
Version: int16(version),
|
|
ChannelID: dbChan.ID,
|
|
NodeID: nodeID,
|
|
Timelock: int32(edge.TimeLockDelta),
|
|
FeePpm: int64(edge.FeeProportionalMillionths),
|
|
BaseFeeMsat: int64(edge.FeeBaseMSat),
|
|
MinHtlcMsat: int64(edge.MinHTLC),
|
|
MessageFlags: sqldb.SQLInt16(edge.MessageFlags),
|
|
ChannelFlags: sqldb.SQLInt16(edge.ChannelFlags),
|
|
InboundBaseFeeMsat: inboundBase,
|
|
InboundFeeRateMilliMsat: inboundRate,
|
|
Signature: edge.SigBytes,
|
|
}
|
|
|
|
switch version {
|
|
case gossipV1:
|
|
params.LastUpdate = sqldb.SQLInt64(edge.LastUpdate.Unix())
|
|
params.Disabled = sql.NullBool{
|
|
Valid: true,
|
|
Bool: edge.IsDisabled(),
|
|
}
|
|
params.MaxHtlcMsat = sql.NullInt64{
|
|
Valid: edge.MessageFlags.HasMaxHtlc(),
|
|
Int64: int64(edge.MaxHTLC),
|
|
}
|
|
case gossipV2:
|
|
params.BlockHeight = sqldb.SQLInt64(
|
|
int64(edge.LastBlockHeight),
|
|
)
|
|
params.DisableFlags = sqldb.SQLInt16(edge.DisableFlags)
|
|
params.MaxHtlcMsat = sqldb.SQLInt64(int64(edge.MaxHTLC))
|
|
}
|
|
|
|
id, err := tx.UpsertEdgePolicy(ctx, params)
|
|
if err != nil {
|
|
return node1Pub, node2Pub, isNode1,
|
|
fmt.Errorf("unable to upsert edge policy: %w", err)
|
|
}
|
|
|
|
// Convert the flat extra opaque data into a map of TLV types to
|
|
// values.
|
|
extra := edge.ExtraSignedFields
|
|
if version == gossipV1 {
|
|
extra, err = marshalExtraOpaqueData(edge.ExtraOpaqueData)
|
|
if err != nil {
|
|
return node1Pub, node2Pub, false, fmt.Errorf(
|
|
"unable to marshal extra opaque data: %w", err,
|
|
)
|
|
}
|
|
}
|
|
|
|
// Update the channel policy's extra signed fields.
|
|
err = upsertChanPolicyExtraSignedFields(ctx, tx, id, extra)
|
|
if err != nil {
|
|
return node1Pub, node2Pub, false, fmt.Errorf("inserting chan "+
|
|
"policy extra TLVs: %w", err)
|
|
}
|
|
|
|
return node1Pub, node2Pub, isNode1, nil
|
|
}
|
|
|
|
// getNodeByPubKey attempts to look up a target node by its public key.
|
|
func getNodeByPubKey(ctx context.Context, cfg *sqldb.QueryConfig, db SQLQueries,
|
|
v lnwire.GossipVersion, pubKey route.Vertex) (int64, *models.Node,
|
|
error) {
|
|
|
|
dbNode, err := db.GetNodeByPubKey(
|
|
ctx, sqlc.GetNodeByPubKeyParams{
|
|
Version: int16(v),
|
|
PubKey: pubKey[:],
|
|
},
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return 0, nil, ErrGraphNodeNotFound
|
|
} else if err != nil {
|
|
return 0, nil, fmt.Errorf("unable to fetch node: %w", err)
|
|
}
|
|
|
|
node, err := buildNode(ctx, cfg, db, dbNode)
|
|
if err != nil {
|
|
return 0, nil, fmt.Errorf("unable to build node: %w", err)
|
|
}
|
|
|
|
return dbNode.ID, node, nil
|
|
}
|
|
|
|
// buildCacheableChannelInfo builds a models.CachedEdgeInfo instance from the
|
|
// provided parameters.
|
|
func buildCacheableChannelInfo(scid []byte, capacity int64, node1Pub,
|
|
node2Pub route.Vertex) *models.CachedEdgeInfo {
|
|
|
|
return &models.CachedEdgeInfo{
|
|
ChannelID: byteOrder.Uint64(scid),
|
|
NodeKey1Bytes: node1Pub,
|
|
NodeKey2Bytes: node2Pub,
|
|
Capacity: btcutil.Amount(capacity),
|
|
}
|
|
}
|
|
|
|
// buildNode constructs a Node instance from the given database node
|
|
// record. The node's features, addresses and extra signed fields are also
|
|
// fetched from the database and set on the node.
|
|
func buildNode(ctx context.Context, cfg *sqldb.QueryConfig, db SQLQueries,
|
|
dbNode sqlc.GraphNode) (*models.Node, error) {
|
|
|
|
data, err := batchLoadNodeData(ctx, cfg, db, []int64{dbNode.ID})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load node data: %w",
|
|
err)
|
|
}
|
|
|
|
return buildNodeWithBatchData(dbNode, data)
|
|
}
|
|
|
|
// isKnownGossipVersion checks whether the provided gossip version is known
|
|
// and supported.
|
|
func isKnownGossipVersion(v lnwire.GossipVersion) bool {
|
|
switch v {
|
|
case gossipV1:
|
|
return true
|
|
case gossipV2:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// buildNodeWithBatchData builds a models.Node instance
|
|
// from the provided sqlc.GraphNode and batchNodeData. If the node does have
|
|
// features/addresses/extra fields, then the corresponding fields are expected
|
|
// to be present in the batchNodeData.
|
|
func buildNodeWithBatchData(dbNode sqlc.GraphNode,
|
|
batchData *batchNodeData) (*models.Node, error) {
|
|
|
|
v := lnwire.GossipVersion(dbNode.Version)
|
|
|
|
if !isKnownGossipVersion(v) {
|
|
return nil, fmt.Errorf("unknown node version: %d", v)
|
|
}
|
|
|
|
pub, err := route.NewVertexFromBytes(dbNode.PubKey)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to parse pubkey: %w", err)
|
|
}
|
|
|
|
node := models.NewShellNode(v, pub)
|
|
|
|
if len(dbNode.Signature) == 0 {
|
|
return node, nil
|
|
}
|
|
|
|
node.AuthSigBytes = dbNode.Signature
|
|
|
|
if dbNode.Alias.Valid {
|
|
node.Alias = fn.Some(dbNode.Alias.String)
|
|
}
|
|
if dbNode.LastUpdate.Valid {
|
|
node.LastUpdate = time.Unix(dbNode.LastUpdate.Int64, 0)
|
|
}
|
|
if dbNode.BlockHeight.Valid {
|
|
node.LastBlockHeight = uint32(dbNode.BlockHeight.Int64)
|
|
}
|
|
|
|
if dbNode.Color.Valid {
|
|
nodeColor, err := DecodeHexColor(dbNode.Color.String)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to decode color: %w",
|
|
err)
|
|
}
|
|
|
|
node.Color = fn.Some(nodeColor)
|
|
}
|
|
|
|
// Use preloaded features.
|
|
if features, exists := batchData.features[dbNode.ID]; exists {
|
|
fv := lnwire.EmptyFeatureVector()
|
|
for _, bit := range features {
|
|
fv.Set(lnwire.FeatureBit(bit))
|
|
}
|
|
node.Features = fv
|
|
}
|
|
|
|
// Use preloaded addresses.
|
|
addresses, exists := batchData.addresses[dbNode.ID]
|
|
if exists && len(addresses) > 0 {
|
|
node.Addresses, err = buildNodeAddresses(addresses)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to build addresses "+
|
|
"for node(%d): %w", dbNode.ID, err)
|
|
}
|
|
}
|
|
|
|
// Use preloaded extra fields.
|
|
if extraFields, exists := batchData.extraFields[dbNode.ID]; exists {
|
|
if v == gossipV1 {
|
|
records := lnwire.CustomRecords(extraFields)
|
|
recs, err := records.Serialize()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to serialize "+
|
|
"extra signed fields: %w", err)
|
|
}
|
|
|
|
if len(recs) != 0 {
|
|
node.ExtraOpaqueData = recs
|
|
}
|
|
} else if len(extraFields) > 0 {
|
|
node.ExtraSignedFields = extraFields
|
|
}
|
|
}
|
|
|
|
return node, nil
|
|
}
|
|
|
|
// forEachNodeInBatch fetches all nodes in the provided batch, builds them
|
|
// with the preloaded data, and executes the provided callback for each node.
|
|
func forEachNodeInBatch(ctx context.Context, cfg *sqldb.QueryConfig,
|
|
db SQLQueries, nodes []sqlc.GraphNode,
|
|
cb func(dbID int64, node *models.Node) error) error {
|
|
|
|
// Extract node IDs for batch loading.
|
|
nodeIDs := make([]int64, len(nodes))
|
|
for i, node := range nodes {
|
|
nodeIDs[i] = node.ID
|
|
}
|
|
|
|
// Batch load all related data for this page.
|
|
batchData, err := batchLoadNodeData(ctx, cfg, db, nodeIDs)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to batch load node data: %w", err)
|
|
}
|
|
|
|
for _, dbNode := range nodes {
|
|
node, err := buildNodeWithBatchData(dbNode, batchData)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to build node(id=%d): %w",
|
|
dbNode.ID, err)
|
|
}
|
|
|
|
if err := cb(dbNode.ID, node); err != nil {
|
|
return fmt.Errorf("callback failed for node(id=%d): %w",
|
|
dbNode.ID, err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// getNodeFeatures fetches the feature bits and constructs the feature vector
|
|
// for a node with the given DB ID.
|
|
func getNodeFeatures(ctx context.Context, db SQLQueries,
|
|
nodeID int64) (*lnwire.FeatureVector, error) {
|
|
|
|
rows, err := db.GetNodeFeatures(ctx, nodeID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to get node(%d) features: %w",
|
|
nodeID, err)
|
|
}
|
|
|
|
features := lnwire.EmptyFeatureVector()
|
|
for _, feature := range rows {
|
|
features.Set(lnwire.FeatureBit(feature.FeatureBit))
|
|
}
|
|
|
|
return features, nil
|
|
}
|
|
|
|
// upsertNodeAncillaryData updates the node's features, addresses, and extra
|
|
// signed fields. This is common logic shared by upsertNode and
|
|
// upsertSourceNode.
|
|
func upsertNodeAncillaryData(ctx context.Context, db SQLQueries,
|
|
nodeID int64, node *models.Node) error {
|
|
|
|
// Update the node's features.
|
|
err := upsertNodeFeatures(ctx, db, nodeID, node.Features)
|
|
if err != nil {
|
|
return fmt.Errorf("inserting node features: %w", err)
|
|
}
|
|
|
|
// Update the node's addresses.
|
|
err = upsertNodeAddresses(ctx, db, nodeID, node.Addresses)
|
|
if err != nil {
|
|
return fmt.Errorf("inserting node addresses: %w", err)
|
|
}
|
|
|
|
// Convert the flat extra opaque data into a map of TLV types to
|
|
// values.
|
|
extra := node.ExtraSignedFields
|
|
if node.Version == gossipV1 {
|
|
extra, err = marshalExtraOpaqueData(node.ExtraOpaqueData)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to marshal extra opaque "+
|
|
"data: %w", err)
|
|
}
|
|
}
|
|
|
|
// Update the node's extra signed fields.
|
|
err = upsertNodeExtraSignedFields(ctx, db, nodeID, extra)
|
|
if err != nil {
|
|
return fmt.Errorf("inserting node extra TLVs: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// populateNodeParams populates the common node parameters from a models.Node.
|
|
// This is a helper for building UpsertNodeParams and UpsertSourceNodeParams.
|
|
func populateNodeParams(node *models.Node,
|
|
setParams func(lastUpdate, lastBlockHeight sql.NullInt64, alias,
|
|
colorStr sql.NullString, signature []byte)) error {
|
|
|
|
if !node.HaveAnnouncement() {
|
|
return nil
|
|
}
|
|
|
|
var (
|
|
alias, colorStr sql.NullString
|
|
lastUpdate, lastBlockHeight sql.NullInt64
|
|
)
|
|
node.Color.WhenSome(func(rgba color.RGBA) {
|
|
colorStr = sqldb.SQLStrValid(EncodeHexColor(rgba))
|
|
})
|
|
node.Alias.WhenSome(func(s string) {
|
|
alias = sqldb.SQLStrValid(s)
|
|
})
|
|
|
|
switch node.Version {
|
|
case gossipV1:
|
|
lastUpdate = sqldb.SQLInt64(node.LastUpdate.Unix())
|
|
|
|
case gossipV2:
|
|
lastBlockHeight = sqldb.SQLInt64(int64(node.LastBlockHeight))
|
|
|
|
default:
|
|
return fmt.Errorf("unknown gossip version: %d", node.Version)
|
|
}
|
|
|
|
setParams(
|
|
lastUpdate, lastBlockHeight, alias, colorStr, node.AuthSigBytes,
|
|
)
|
|
|
|
return nil
|
|
}
|
|
|
|
// buildNodeUpsertParams builds the parameters for upserting a node using the
|
|
// strict UpsertNode query (requires timestamp to be increasing).
|
|
func buildNodeUpsertParams(node *models.Node) (sqlc.UpsertNodeParams, error) {
|
|
params := sqlc.UpsertNodeParams{
|
|
Version: int16(node.Version),
|
|
PubKey: node.PubKeyBytes[:],
|
|
}
|
|
|
|
err := populateNodeParams(
|
|
node, func(lastUpdate, lastBlockHeight sql.NullInt64, alias,
|
|
colorStr sql.NullString,
|
|
signature []byte) {
|
|
|
|
params.LastUpdate = lastUpdate
|
|
params.BlockHeight = lastBlockHeight
|
|
params.Alias = alias
|
|
params.Color = colorStr
|
|
params.Signature = signature
|
|
},
|
|
)
|
|
|
|
return params, err
|
|
}
|
|
|
|
// buildSourceNodeUpsertParams builds the parameters for upserting the source
|
|
// node using the lenient UpsertSourceNode query (allows same timestamp).
|
|
func buildSourceNodeUpsertParams(node *models.Node) (
|
|
sqlc.UpsertSourceNodeParams, error) {
|
|
|
|
params := sqlc.UpsertSourceNodeParams{
|
|
Version: int16(node.Version),
|
|
PubKey: node.PubKeyBytes[:],
|
|
}
|
|
|
|
err := populateNodeParams(
|
|
node, func(lastUpdate, lastBlock sql.NullInt64, alias,
|
|
colorStr sql.NullString, signature []byte) {
|
|
|
|
params.BlockHeight = lastBlock
|
|
params.LastUpdate = lastUpdate
|
|
params.Alias = alias
|
|
params.Color = colorStr
|
|
params.Signature = signature
|
|
},
|
|
)
|
|
|
|
return params, err
|
|
}
|
|
|
|
// upsertSourceNode upserts the source node record into the database using a
|
|
// less strict upsert that allows updates even when the timestamp hasn't
|
|
// changed. This is necessary to handle concurrent updates to our own node
|
|
// during startup and runtime. The node's features, addresses and extra TLV
|
|
// types are also updated. The node's DB ID is returned.
|
|
func upsertSourceNode(ctx context.Context, db SQLQueries,
|
|
node *models.Node) (int64, error) {
|
|
|
|
params, err := buildSourceNodeUpsertParams(node)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
nodeID, err := db.UpsertSourceNode(ctx, params)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("upserting source node(%x): %w",
|
|
node.PubKeyBytes, err)
|
|
}
|
|
|
|
// We can exit here if we don't have the announcement yet.
|
|
if !node.HaveAnnouncement() {
|
|
return nodeID, nil
|
|
}
|
|
|
|
// Update the ancillary node data (features, addresses, extra fields).
|
|
err = upsertNodeAncillaryData(ctx, db, nodeID, node)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
return nodeID, nil
|
|
}
|
|
|
|
// upsertNode upserts the node record into the database. If the node already
|
|
// exists, then the node's information is updated. If the node doesn't exist,
|
|
// then a new node is created. The node's features, addresses and extra TLV
|
|
// types are also updated. The node's DB ID is returned.
|
|
func upsertNode(ctx context.Context, db SQLQueries,
|
|
node *models.Node) (int64, error) {
|
|
|
|
if !isKnownGossipVersion(node.Version) {
|
|
return 0, fmt.Errorf("unknown gossip version: %d", node.Version)
|
|
}
|
|
|
|
params, err := buildNodeUpsertParams(node)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
nodeID, err := db.UpsertNode(ctx, params)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("upserting node(%x): %w", node.PubKeyBytes,
|
|
err)
|
|
}
|
|
|
|
// We can exit here if we don't have the announcement yet.
|
|
if !node.HaveAnnouncement() {
|
|
return nodeID, nil
|
|
}
|
|
|
|
// Update the ancillary node data (features, addresses, extra fields).
|
|
err = upsertNodeAncillaryData(ctx, db, nodeID, node)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
return nodeID, nil
|
|
}
|
|
|
|
// upsertNodeFeatures updates the node's features node_features table. This
|
|
// includes deleting any feature bits no longer present and inserting any new
|
|
// feature bits. If the feature bit does not yet exist in the features table,
|
|
// then an entry is created in that table first.
|
|
func upsertNodeFeatures(ctx context.Context, db SQLQueries, nodeID int64,
|
|
features *lnwire.FeatureVector) error {
|
|
|
|
// Get any existing features for the node.
|
|
existingFeatures, err := db.GetNodeFeatures(ctx, nodeID)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return err
|
|
}
|
|
|
|
// Copy the nodes latest set of feature bits.
|
|
newFeatures := make(map[int32]struct{})
|
|
if features != nil {
|
|
for feature := range features.Features() {
|
|
newFeatures[int32(feature)] = struct{}{}
|
|
}
|
|
}
|
|
|
|
// For any current feature that already exists in the DB, remove it from
|
|
// the in-memory map. For any existing feature that does not exist in
|
|
// the in-memory map, delete it from the database.
|
|
for _, feature := range existingFeatures {
|
|
// The feature is still present, so there are no updates to be
|
|
// made.
|
|
if _, ok := newFeatures[feature.FeatureBit]; ok {
|
|
delete(newFeatures, feature.FeatureBit)
|
|
continue
|
|
}
|
|
|
|
// The feature is no longer present, so we remove it from the
|
|
// database.
|
|
err := db.DeleteNodeFeature(ctx, sqlc.DeleteNodeFeatureParams{
|
|
NodeID: nodeID,
|
|
FeatureBit: feature.FeatureBit,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("unable to delete node(%d) "+
|
|
"feature(%v): %w", nodeID, feature.FeatureBit,
|
|
err)
|
|
}
|
|
}
|
|
|
|
// Any remaining entries in newFeatures are new features that need to be
|
|
// added to the database for the first time.
|
|
for feature := range newFeatures {
|
|
err = db.InsertNodeFeature(ctx, sqlc.InsertNodeFeatureParams{
|
|
NodeID: nodeID,
|
|
FeatureBit: feature,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("unable to insert node(%d) "+
|
|
"feature(%v): %w", nodeID, feature, err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// fetchNodeFeatures fetches the features for a node with the given public key.
|
|
func fetchNodeFeatures(ctx context.Context, queries SQLQueries,
|
|
v lnwire.GossipVersion, nodePub route.Vertex) (*lnwire.FeatureVector,
|
|
error) {
|
|
|
|
rows, err := queries.GetNodeFeaturesByPubKey(
|
|
ctx, sqlc.GetNodeFeaturesByPubKeyParams{
|
|
PubKey: nodePub[:],
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to get node(%s) features: %w",
|
|
nodePub, err)
|
|
}
|
|
|
|
features := lnwire.EmptyFeatureVector()
|
|
for _, bit := range rows {
|
|
features.Set(lnwire.FeatureBit(bit))
|
|
}
|
|
|
|
return features, nil
|
|
}
|
|
|
|
// dbAddressType is an enum type that represents the different address types
|
|
// that we store in the node_addresses table. The address type determines how
|
|
// the address is to be serialised/deserialize.
|
|
type dbAddressType uint8
|
|
|
|
const (
|
|
addressTypeIPv4 dbAddressType = 1
|
|
addressTypeIPv6 dbAddressType = 2
|
|
addressTypeTorV2 dbAddressType = 3
|
|
addressTypeTorV3 dbAddressType = 4
|
|
addressTypeDNS dbAddressType = 5
|
|
addressTypeOpaque dbAddressType = math.MaxInt8
|
|
)
|
|
|
|
// collectAddressRecords collects the addresses from the provided
|
|
// net.Addr slice and returns a map of dbAddressType to a slice of address
|
|
// strings.
|
|
func collectAddressRecords(addresses []net.Addr) (map[dbAddressType][]string,
|
|
error) {
|
|
|
|
// Copy the nodes latest set of addresses. v2 is stored for wire
|
|
// fidelity even though lnd no longer produces it.
|
|
newAddresses := map[dbAddressType][]string{
|
|
addressTypeIPv4: {},
|
|
addressTypeIPv6: {},
|
|
addressTypeTorV2: {},
|
|
addressTypeTorV3: {},
|
|
addressTypeDNS: {},
|
|
addressTypeOpaque: {},
|
|
}
|
|
addAddr := func(t dbAddressType, addr net.Addr) {
|
|
newAddresses[t] = append(newAddresses[t], addr.String())
|
|
}
|
|
|
|
for _, address := range addresses {
|
|
switch addr := address.(type) {
|
|
case *net.TCPAddr:
|
|
if ip4 := addr.IP.To4(); ip4 != nil {
|
|
addAddr(addressTypeIPv4, addr)
|
|
} else if ip6 := addr.IP.To16(); ip6 != nil {
|
|
addAddr(addressTypeIPv6, addr)
|
|
} else {
|
|
return nil, fmt.Errorf("unhandled IP "+
|
|
"address: %v", addr)
|
|
}
|
|
|
|
case *tor.OnionAddr:
|
|
switch len(addr.OnionService) {
|
|
case tor.V2Len:
|
|
addAddr(addressTypeTorV2, addr)
|
|
case tor.V3Len:
|
|
addAddr(addressTypeTorV3, addr)
|
|
default:
|
|
return nil, fmt.Errorf("invalid length for " +
|
|
"a tor address")
|
|
}
|
|
|
|
case *lnwire.DNSAddress:
|
|
addAddr(addressTypeDNS, addr)
|
|
|
|
case *lnwire.OpaqueAddrs:
|
|
addAddr(addressTypeOpaque, addr)
|
|
|
|
default:
|
|
return nil, fmt.Errorf("unhandled address type: %T",
|
|
addr)
|
|
}
|
|
}
|
|
|
|
return newAddresses, nil
|
|
}
|
|
|
|
// upsertNodeAddresses updates the node's addresses in the database. This
|
|
// includes deleting any existing addresses and inserting the new set of
|
|
// addresses. The deletion is necessary since the ordering of the addresses may
|
|
// change, and we need to ensure that the database reflects the latest set of
|
|
// addresses so that at the time of reconstructing the node announcement, the
|
|
// order is preserved and the signature over the message remains valid.
|
|
func upsertNodeAddresses(ctx context.Context, db SQLQueries, nodeID int64,
|
|
addresses []net.Addr) error {
|
|
|
|
// Delete any existing addresses for the node. This is required since
|
|
// even if the new set of addresses is the same, the ordering may have
|
|
// changed for a given address type.
|
|
err := db.DeleteNodeAddresses(ctx, nodeID)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to delete node(%d) addresses: %w",
|
|
nodeID, err)
|
|
}
|
|
|
|
newAddresses, err := collectAddressRecords(addresses)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Any remaining entries in newAddresses are new addresses that need to
|
|
// be added to the database for the first time.
|
|
for addrType, addrList := range newAddresses {
|
|
for position, addr := range addrList {
|
|
err := db.UpsertNodeAddress(
|
|
ctx, sqlc.UpsertNodeAddressParams{
|
|
NodeID: nodeID,
|
|
Type: int16(addrType),
|
|
Address: addr,
|
|
Position: int32(position),
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to insert "+
|
|
"node(%d) address(%v): %w", nodeID,
|
|
addr, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// getNodeAddresses fetches the addresses for a node with the given DB ID.
|
|
func getNodeAddresses(ctx context.Context, db SQLQueries, id int64) ([]net.Addr,
|
|
error) {
|
|
|
|
// GetNodeAddresses ensures that the addresses for a given type are
|
|
// returned in the same order as they were inserted.
|
|
rows, err := db.GetNodeAddresses(ctx, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
addresses := make([]net.Addr, 0, len(rows))
|
|
for _, row := range rows {
|
|
address := row.Address
|
|
|
|
addr, err := parseAddress(dbAddressType(row.Type), address)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to parse address "+
|
|
"for node(%d): %v: %w", id, address, err)
|
|
}
|
|
|
|
addresses = append(addresses, addr)
|
|
}
|
|
|
|
// If we have no addresses, then we'll return nil instead of an
|
|
// empty slice.
|
|
if len(addresses) == 0 {
|
|
addresses = nil
|
|
}
|
|
|
|
return addresses, nil
|
|
}
|
|
|
|
// upsertNodeExtraSignedFields updates the node's extra signed fields in the
|
|
// database. This includes updating any existing types, inserting any new types,
|
|
// and deleting any types that are no longer present.
|
|
func upsertNodeExtraSignedFields(ctx context.Context, db SQLQueries,
|
|
nodeID int64, extraFields map[uint64][]byte) error {
|
|
|
|
// Get any existing extra signed fields for the node.
|
|
existingFields, err := db.GetExtraNodeTypes(ctx, nodeID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Make a lookup map of the existing field types so that we can use it
|
|
// to keep track of any fields we should delete.
|
|
m := make(map[uint64]bool)
|
|
for _, field := range existingFields {
|
|
m[uint64(field.Type)] = true
|
|
}
|
|
|
|
// For all the new fields, we'll upsert them and remove them from the
|
|
// map of existing fields.
|
|
for tlvType, value := range extraFields {
|
|
err = db.UpsertNodeExtraType(
|
|
ctx, sqlc.UpsertNodeExtraTypeParams{
|
|
NodeID: nodeID,
|
|
Type: int64(tlvType),
|
|
Value: value,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to upsert node(%d) extra "+
|
|
"signed field(%v): %w", nodeID, tlvType, err)
|
|
}
|
|
|
|
// Remove the field from the map of existing fields if it was
|
|
// present.
|
|
delete(m, tlvType)
|
|
}
|
|
|
|
// For all the fields that are left in the map of existing fields, we'll
|
|
// delete them as they are no longer present in the new set of fields.
|
|
for tlvType := range m {
|
|
err = db.DeleteExtraNodeType(
|
|
ctx, sqlc.DeleteExtraNodeTypeParams{
|
|
NodeID: nodeID,
|
|
Type: int64(tlvType),
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to delete node(%d) extra "+
|
|
"signed field(%v): %w", nodeID, tlvType, err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// srcNodeInfo holds the information about the source node of the graph.
|
|
type srcNodeInfo struct {
|
|
// id is the DB level ID of the source node entry in the "nodes" table.
|
|
id int64
|
|
|
|
// pub is the public key of the source node.
|
|
pub route.Vertex
|
|
}
|
|
|
|
// 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 lnwire.GossipVersion) (int64, route.Vertex, error) {
|
|
|
|
s.srcNodeMu.Lock()
|
|
defer s.srcNodeMu.Unlock()
|
|
|
|
// If we already have the source node ID and pub key cached, then
|
|
// return them.
|
|
if info, ok := s.srcNodes[version]; ok {
|
|
return info.id, info.pub, nil
|
|
}
|
|
|
|
var pubKey route.Vertex
|
|
|
|
nodes, err := db.GetSourceNodesByVersion(ctx, int16(version))
|
|
if err != nil {
|
|
return 0, pubKey, fmt.Errorf("unable to fetch source node: %w",
|
|
err)
|
|
}
|
|
|
|
if len(nodes) == 0 {
|
|
return 0, pubKey, ErrSourceNodeNotSet
|
|
} else if len(nodes) > 1 {
|
|
return 0, pubKey, fmt.Errorf("multiple source nodes for "+
|
|
"protocol %s found", version)
|
|
}
|
|
|
|
copy(pubKey[:], nodes[0].PubKey)
|
|
|
|
s.srcNodes[version] = &srcNodeInfo{
|
|
id: nodes[0].NodeID,
|
|
pub: pubKey,
|
|
}
|
|
|
|
return nodes[0].NodeID, pubKey, nil
|
|
}
|
|
|
|
// marshalExtraOpaqueData takes a flat byte slice parses it as a TLV stream.
|
|
// This then produces a map from TLV type to value. If the input is not a
|
|
// valid TLV stream, then an error is returned.
|
|
func marshalExtraOpaqueData(data []byte) (map[uint64][]byte, error) {
|
|
r := bytes.NewReader(data)
|
|
|
|
tlvStream, err := tlv.NewStream()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Since ExtraOpaqueData is provided by a potentially malicious peer,
|
|
// pass it into the P2P decoding variant.
|
|
parsedTypes, err := tlvStream.DecodeWithParsedTypesP2P(r)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%w: %w", ErrParsingExtraTLVBytes, err)
|
|
}
|
|
if len(parsedTypes) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
records := make(map[uint64][]byte)
|
|
for k, v := range parsedTypes {
|
|
records[uint64(k)] = v
|
|
}
|
|
|
|
return records, nil
|
|
}
|
|
|
|
// insertChannel inserts a new channel record into the database.
|
|
func insertChannel(ctx context.Context, db SQLQueries,
|
|
edge *models.ChannelEdgeInfo) error {
|
|
|
|
v := edge.Version
|
|
|
|
// Make sure that at least a "shell" entry for each node is present in
|
|
// the nodes table.
|
|
node1DBID, err := maybeCreateShellNode(ctx, db, v, edge.NodeKey1Bytes)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to create shell node: %w", err)
|
|
}
|
|
|
|
node2DBID, err := maybeCreateShellNode(ctx, db, v, 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(v),
|
|
Scid: channelIDToBytes(edge.ChannelID),
|
|
NodeID1: node1DBID,
|
|
NodeID2: node2DBID,
|
|
Outpoint: edge.ChannelPoint.String(),
|
|
Capacity: capacity,
|
|
}
|
|
edge.BitcoinKey1Bytes.WhenSome(func(vertex route.Vertex) {
|
|
createParams.BitcoinKey1 = vertex[:]
|
|
})
|
|
edge.BitcoinKey2Bytes.WhenSome(func(vertex route.Vertex) {
|
|
createParams.BitcoinKey2 = vertex[:]
|
|
})
|
|
edge.FundingScript.WhenSome(func(script []byte) {
|
|
createParams.FundingPkScript = script
|
|
})
|
|
edge.MerkleRootHash.WhenSome(func(hash chainhash.Hash) {
|
|
createParams.MerkleRootHash = hash[:]
|
|
})
|
|
|
|
if edge.AuthProof != nil {
|
|
proof := edge.AuthProof
|
|
|
|
createParams.Node1Signature = proof.NodeSig1()
|
|
createParams.Node2Signature = proof.NodeSig2()
|
|
createParams.Bitcoin1Signature = proof.BitcoinSig1()
|
|
createParams.Bitcoin2Signature = proof.BitcoinSig2()
|
|
createParams.Signature = proof.Sig()
|
|
}
|
|
|
|
// Insert the new channel record.
|
|
dbChanID, err := db.CreateChannel(ctx, createParams)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Insert any channel features.
|
|
for feature := range edge.Features.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 := edge.ExtraSignedFields
|
|
if v == gossipV1 {
|
|
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.UpsertChannelExtraType(
|
|
ctx, sqlc.UpsertChannelExtraTypeParams{
|
|
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,
|
|
v lnwire.GossipVersion, pubKey route.Vertex) (int64, error) {
|
|
|
|
dbNode, err := db.GetNodeByPubKey(
|
|
ctx, sqlc.GetNodeByPubKeyParams{
|
|
PubKey: pubKey[:],
|
|
Version: int16(v),
|
|
},
|
|
)
|
|
// 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(v),
|
|
PubKey: pubKey[:],
|
|
})
|
|
if err != nil {
|
|
return 0, fmt.Errorf("unable to create shell node: %w", err)
|
|
}
|
|
|
|
return id, nil
|
|
}
|
|
|
|
// upsertChanPolicyExtraSignedFields updates the policy's extra signed fields in
|
|
// the database. This includes deleting any existing types and then inserting
|
|
// the new types.
|
|
func upsertChanPolicyExtraSignedFields(ctx context.Context, db SQLQueries,
|
|
chanPolicyID int64, extraFields map[uint64][]byte) error {
|
|
|
|
// Delete all existing extra signed fields for the channel policy.
|
|
err := db.DeleteChannelPolicyExtraTypes(ctx, chanPolicyID)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to delete "+
|
|
"existing policy extra signed fields for policy %d: %w",
|
|
chanPolicyID, err)
|
|
}
|
|
|
|
// Insert all new extra signed fields for the channel policy.
|
|
for tlvType, value := range extraFields {
|
|
err = db.UpsertChanPolicyExtraType(
|
|
ctx, sqlc.UpsertChanPolicyExtraTypeParams{
|
|
ChannelPolicyID: chanPolicyID,
|
|
Type: int64(tlvType),
|
|
Value: value,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to insert "+
|
|
"channel_policy(%d) extra signed field(%v): %w",
|
|
chanPolicyID, tlvType, err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// getAndBuildEdgeInfo builds a models.ChannelEdgeInfo instance from the
|
|
// provided dbChanRow and also fetches any other required information
|
|
// to construct the edge info.
|
|
func getAndBuildEdgeInfo(ctx context.Context, cfg *SQLStoreConfig,
|
|
db SQLQueries, dbChan sqlc.GraphChannel, node1,
|
|
node2 route.Vertex) (*models.ChannelEdgeInfo, error) {
|
|
|
|
data, err := batchLoadChannelData(
|
|
ctx, cfg.QueryCfg, db, []int64{dbChan.ID}, nil,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load channel data: %w",
|
|
err)
|
|
}
|
|
|
|
return buildEdgeInfoWithBatchData(
|
|
cfg.ChainHash, dbChan, node1, node2, data,
|
|
)
|
|
}
|
|
|
|
// buildEdgeInfoWithBatchData builds edge info using pre-loaded batch data.
|
|
func buildEdgeInfoWithBatchData(chain chainhash.Hash,
|
|
dbChan sqlc.GraphChannel, node1, node2 route.Vertex,
|
|
batchData *batchChannelData) (*models.ChannelEdgeInfo, error) {
|
|
|
|
v := lnwire.GossipVersion(dbChan.Version)
|
|
if !isKnownGossipVersion(v) {
|
|
return nil, fmt.Errorf("unknown channel version: %d", v)
|
|
}
|
|
|
|
// Use pre-loaded features and extras types.
|
|
fv := lnwire.EmptyFeatureVector()
|
|
if features, exists := batchData.chanfeatures[dbChan.ID]; exists {
|
|
for _, bit := range features {
|
|
fv.Set(lnwire.FeatureBit(bit))
|
|
}
|
|
}
|
|
|
|
var extras map[uint64][]byte
|
|
channelExtras, exists := batchData.chanExtraTypes[dbChan.ID]
|
|
if exists {
|
|
extras = channelExtras
|
|
} else {
|
|
extras = make(map[uint64][]byte)
|
|
}
|
|
|
|
op, err := wire.NewOutPointFromString(dbChan.Outpoint)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Build the appropriate channel based on version.
|
|
var channel *models.ChannelEdgeInfo
|
|
switch v {
|
|
case gossipV1:
|
|
// For v1, serialize extras into ExtraOpaqueData.
|
|
recs, err := lnwire.CustomRecords(extras).Serialize()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to serialize extra "+
|
|
"signed fields: %w", err)
|
|
}
|
|
if recs == nil {
|
|
recs = make([]byte, 0)
|
|
}
|
|
|
|
// Bitcoin keys are required for v1.
|
|
btcKey1, err := route.NewVertexFromBytes(dbChan.BitcoinKey1)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
btcKey2, err := route.NewVertexFromBytes(dbChan.BitcoinKey2)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
channel, err = models.NewV1Channel(
|
|
byteOrder.Uint64(dbChan.Scid), chain, node1, node2,
|
|
&models.ChannelV1Fields{
|
|
BitcoinKey1Bytes: btcKey1,
|
|
BitcoinKey2Bytes: btcKey2,
|
|
ExtraOpaqueData: recs,
|
|
},
|
|
models.WithChannelPoint(*op),
|
|
models.WithCapacity(
|
|
btcutil.Amount(dbChan.Capacity.Int64),
|
|
),
|
|
models.WithFeatures(fv.RawFeatureVector),
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// For v1 channels, attach the auth proof if all four
|
|
// signatures are present.
|
|
if len(dbChan.Bitcoin1Signature) > 0 {
|
|
channel.AuthProof = models.NewV1ChannelAuthProof(
|
|
dbChan.Node1Signature,
|
|
dbChan.Node2Signature,
|
|
dbChan.Bitcoin1Signature,
|
|
dbChan.Bitcoin2Signature,
|
|
)
|
|
}
|
|
|
|
case gossipV2:
|
|
v2Fields := &models.ChannelV2Fields{
|
|
ExtraSignedFields: extras,
|
|
}
|
|
|
|
// For v2, bitcoin keys are optional.
|
|
if len(dbChan.BitcoinKey1) > 0 {
|
|
btcKey1, err := route.NewVertexFromBytes(
|
|
dbChan.BitcoinKey1,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
v2Fields.BitcoinKey1Bytes = fn.Some(btcKey1)
|
|
}
|
|
if len(dbChan.BitcoinKey2) > 0 {
|
|
btcKey2, err := route.NewVertexFromBytes(
|
|
dbChan.BitcoinKey2,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
v2Fields.BitcoinKey2Bytes = fn.Some(btcKey2)
|
|
}
|
|
|
|
// Parse funding script if present.
|
|
if len(dbChan.FundingPkScript) > 0 {
|
|
v2Fields.FundingScript = fn.Some(dbChan.FundingPkScript)
|
|
}
|
|
|
|
// Parse merkle root hash if present.
|
|
if len(dbChan.MerkleRootHash) > 0 {
|
|
var hash chainhash.Hash
|
|
copy(hash[:], dbChan.MerkleRootHash)
|
|
v2Fields.MerkleRootHash = fn.Some(hash)
|
|
}
|
|
|
|
opts := []models.EdgeModifier{
|
|
models.WithChannelPoint(*op),
|
|
models.WithCapacity(btcutil.Amount(
|
|
dbChan.Capacity.Int64,
|
|
)),
|
|
models.WithFeatures(fv.RawFeatureVector),
|
|
}
|
|
|
|
// For v2 channels, attach the auth proof if the signature is
|
|
// present.
|
|
if len(dbChan.Signature) > 0 {
|
|
proof := models.NewV2ChannelAuthProof(dbChan.Signature)
|
|
opts = append(opts, models.WithChanProof(proof))
|
|
}
|
|
|
|
channel, err = models.NewV2Channel(
|
|
byteOrder.Uint64(dbChan.Scid), chain, node1, node2,
|
|
v2Fields, opts...,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
default:
|
|
return nil, fmt.Errorf("unsupported channel version: %d", v)
|
|
}
|
|
|
|
return channel, nil
|
|
}
|
|
|
|
// buildNodeVertices is a helper that converts raw node public keys
|
|
// into route.Vertex instances.
|
|
func buildNodeVertices(node1Pub, node2Pub []byte) (route.Vertex,
|
|
route.Vertex, error) {
|
|
|
|
node1Vertex, err := route.NewVertexFromBytes(node1Pub)
|
|
if err != nil {
|
|
return route.Vertex{}, route.Vertex{}, fmt.Errorf("unable to "+
|
|
"create vertex from node1 pubkey: %w", err)
|
|
}
|
|
|
|
node2Vertex, err := route.NewVertexFromBytes(node2Pub)
|
|
if err != nil {
|
|
return route.Vertex{}, route.Vertex{}, fmt.Errorf("unable to "+
|
|
"create vertex from node2 pubkey: %w", err)
|
|
}
|
|
|
|
return node1Vertex, node2Vertex, nil
|
|
}
|
|
|
|
// getAndBuildChanPolicies uses the given sqlc.GraphChannelPolicy and also
|
|
// retrieves all the extra info required to build the complete
|
|
// models.ChannelEdgePolicy types. It returns two policies, which may be nil if
|
|
// the provided sqlc.GraphChannelPolicy records are nil.
|
|
func getAndBuildChanPolicies(ctx context.Context, cfg *sqldb.QueryConfig,
|
|
db SQLQueries, dbPol1, dbPol2 *sqlc.GraphChannelPolicy,
|
|
channelID uint64, node1, node2 route.Vertex) (*models.ChannelEdgePolicy,
|
|
*models.ChannelEdgePolicy, error) {
|
|
|
|
if dbPol1 == nil && dbPol2 == nil {
|
|
return nil, nil, nil
|
|
}
|
|
|
|
if dbPol1 != nil &&
|
|
!isKnownGossipVersion(lnwire.GossipVersion(dbPol1.Version)) {
|
|
|
|
return nil, nil, fmt.Errorf("unsupported policy1 version: %d",
|
|
dbPol1.Version)
|
|
}
|
|
|
|
if dbPol2 != nil &&
|
|
!isKnownGossipVersion(lnwire.GossipVersion(dbPol2.Version)) {
|
|
|
|
return nil, nil, fmt.Errorf("unsupported policy2 version: %d",
|
|
dbPol2.Version)
|
|
}
|
|
|
|
var policyIDs = make([]int64, 0, 2)
|
|
if dbPol1 != nil {
|
|
policyIDs = append(policyIDs, dbPol1.ID)
|
|
}
|
|
if dbPol2 != nil {
|
|
policyIDs = append(policyIDs, dbPol2.ID)
|
|
}
|
|
|
|
batchData, err := batchLoadChannelData(ctx, cfg, db, nil, policyIDs)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("unable to batch load channel "+
|
|
"data: %w", err)
|
|
}
|
|
|
|
pol1, err := buildChanPolicyWithBatchData(
|
|
true, dbPol1, channelID, node2, batchData,
|
|
)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("unable to build policy1: %w", err)
|
|
}
|
|
|
|
pol2, err := buildChanPolicyWithBatchData(
|
|
false, dbPol2, channelID, node1, batchData,
|
|
)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("unable to build policy2: %w", err)
|
|
}
|
|
|
|
return pol1, pol2, nil
|
|
}
|
|
|
|
// buildCachedChanPolicies builds models.CachedEdgePolicy instances from the
|
|
// provided sqlc.GraphChannelPolicy objects. If a provided policy is nil,
|
|
// then nil is returned for it.
|
|
func buildCachedChanPolicies(dbPol1, dbPol2 *sqlc.GraphChannelPolicy,
|
|
channelID uint64, node1, node2 route.Vertex) (*models.CachedEdgePolicy,
|
|
*models.CachedEdgePolicy, error) {
|
|
|
|
var p1, p2 *models.CachedEdgePolicy
|
|
if dbPol1 != nil {
|
|
policy1, err := buildChanPolicy(
|
|
true, *dbPol1, channelID, nil, node2,
|
|
)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
p1 = models.NewCachedPolicy(policy1)
|
|
}
|
|
if dbPol2 != nil {
|
|
policy2, err := buildChanPolicy(
|
|
false, *dbPol2, channelID, nil, node1,
|
|
)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
p2 = models.NewCachedPolicy(policy2)
|
|
}
|
|
|
|
return p1, p2, nil
|
|
}
|
|
|
|
// buildChanPolicy builds a models.ChannelEdgePolicy instance from the
|
|
// provided sqlc.GraphChannelPolicy and other required information.
|
|
func buildChanPolicy(isNode1 bool, dbPolicy sqlc.GraphChannelPolicy,
|
|
channelID uint64, extras map[uint64][]byte,
|
|
toNode route.Vertex) (*models.ChannelEdgePolicy, error) {
|
|
|
|
var inboundFee fn.Option[lnwire.Fee]
|
|
if dbPolicy.InboundFeeRateMilliMsat.Valid ||
|
|
dbPolicy.InboundBaseFeeMsat.Valid {
|
|
|
|
inboundFee = fn.Some(lnwire.Fee{
|
|
BaseFee: int32(dbPolicy.InboundBaseFeeMsat.Int64),
|
|
FeeRate: int32(dbPolicy.InboundFeeRateMilliMsat.Int64),
|
|
})
|
|
}
|
|
|
|
p := &models.ChannelEdgePolicy{
|
|
Version: lnwire.GossipVersion(dbPolicy.Version),
|
|
SigBytes: dbPolicy.Signature,
|
|
ChannelID: channelID,
|
|
SecondPeer: !isNode1,
|
|
TimeLockDelta: uint16(dbPolicy.Timelock),
|
|
MinHTLC: lnwire.MilliSatoshi(
|
|
dbPolicy.MinHtlcMsat,
|
|
),
|
|
MaxHTLC: lnwire.MilliSatoshi(
|
|
dbPolicy.MaxHtlcMsat.Int64,
|
|
),
|
|
FeeBaseMSat: lnwire.MilliSatoshi(
|
|
dbPolicy.BaseFeeMsat,
|
|
),
|
|
FeeProportionalMillionths: lnwire.MilliSatoshi(dbPolicy.FeePpm),
|
|
ToNode: toNode,
|
|
InboundFee: inboundFee,
|
|
}
|
|
|
|
if p.Version != gossipV2 {
|
|
recs, err := lnwire.CustomRecords(extras).Serialize()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to serialize extra "+
|
|
"signed fields: %w", err)
|
|
}
|
|
|
|
p.ExtraOpaqueData = recs
|
|
p.LastUpdate = time.Unix(dbPolicy.LastUpdate.Int64, 0)
|
|
//nolint:ll
|
|
p.MessageFlags = sqldb.ExtractSqlInt16[lnwire.ChanUpdateMsgFlags](
|
|
dbPolicy.MessageFlags,
|
|
)
|
|
//nolint:ll
|
|
p.ChannelFlags = sqldb.ExtractSqlInt16[lnwire.ChanUpdateChanFlags](
|
|
dbPolicy.ChannelFlags,
|
|
)
|
|
} else {
|
|
if dbPolicy.BlockHeight.Valid {
|
|
p.LastBlockHeight = uint32(
|
|
dbPolicy.BlockHeight.Int64,
|
|
)
|
|
}
|
|
|
|
//nolint:ll
|
|
p.DisableFlags = sqldb.ExtractSqlInt16[lnwire.ChanUpdateDisableFlags](
|
|
dbPolicy.DisableFlags,
|
|
)
|
|
p.ExtraSignedFields = extras
|
|
}
|
|
|
|
return p, nil
|
|
}
|
|
|
|
// extractChannelPolicies extracts the sqlc.GraphChannelPolicy records from the give
|
|
// row which is expected to be a sqlc type that contains channel policy
|
|
// information. It returns two policies, which may be nil if the policy
|
|
// information is not present in the row.
|
|
//
|
|
//nolint:ll,dupl,funlen
|
|
func extractChannelPolicies(row any) (*sqlc.GraphChannelPolicy,
|
|
*sqlc.GraphChannelPolicy, error) {
|
|
|
|
var policy1, policy2 *sqlc.GraphChannelPolicy
|
|
switch r := row.(type) {
|
|
case sqlc.ListChannelsWithPoliciesForCachePaginatedRow:
|
|
if r.Policy1Timelock.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
Version: r.Policy1Version.Int16,
|
|
Timelock: r.Policy1Timelock.Int32,
|
|
FeePpm: r.Policy1FeePpm.Int64,
|
|
BaseFeeMsat: r.Policy1BaseFeeMsat.Int64,
|
|
MinHtlcMsat: r.Policy1MinHtlcMsat.Int64,
|
|
MaxHtlcMsat: r.Policy1MaxHtlcMsat,
|
|
InboundBaseFeeMsat: r.Policy1InboundBaseFeeMsat,
|
|
InboundFeeRateMilliMsat: r.Policy1InboundFeeRateMilliMsat,
|
|
Disabled: r.Policy1Disabled,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2Timelock.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
Version: r.Policy2Version.Int16,
|
|
Timelock: r.Policy2Timelock.Int32,
|
|
FeePpm: r.Policy2FeePpm.Int64,
|
|
BaseFeeMsat: r.Policy2BaseFeeMsat.Int64,
|
|
MinHtlcMsat: r.Policy2MinHtlcMsat.Int64,
|
|
MaxHtlcMsat: r.Policy2MaxHtlcMsat,
|
|
InboundBaseFeeMsat: r.Policy2InboundBaseFeeMsat,
|
|
InboundFeeRateMilliMsat: r.Policy2InboundFeeRateMilliMsat,
|
|
Disabled: r.Policy2Disabled,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
case sqlc.GetChannelsBySCIDWithPoliciesRow:
|
|
if r.Policy1ID.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy1ID.Int64,
|
|
Version: r.Policy1Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
Signature: r.Policy1Signature,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2ID.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy2ID.Int64,
|
|
Version: r.Policy2Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
Signature: r.Policy2Signature,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
case sqlc.GetChannelByOutpointWithPoliciesRow:
|
|
if r.Policy1ID.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy1ID.Int64,
|
|
Version: r.Policy1Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
Signature: r.Policy1Signature,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2ID.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy2ID.Int64,
|
|
Version: r.Policy2Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
Signature: r.Policy2Signature,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
case sqlc.GetChannelBySCIDWithPoliciesRow:
|
|
if r.Policy1ID.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy1ID.Int64,
|
|
Version: r.Policy1Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
Signature: r.Policy1Signature,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2ID.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy2ID.Int64,
|
|
Version: r.Policy2Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
Signature: r.Policy2Signature,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
case sqlc.GetChannelsByPolicyLastUpdateRangeRow:
|
|
if r.Policy1ID.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy1ID.Int64,
|
|
Version: r.Policy1Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
Signature: r.Policy1Signature,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2ID.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy2ID.Int64,
|
|
Version: r.Policy2Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
Signature: r.Policy2Signature,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
case sqlc.GetChannelsByPolicyBlockRangeRow:
|
|
if r.Policy1ID.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy1ID.Int64,
|
|
Version: r.Policy1Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
Signature: r.Policy1Signature,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2ID.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy2ID.Int64,
|
|
Version: r.Policy2Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
Signature: r.Policy2Signature,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
case sqlc.ListChannelsForNodeIDsRow:
|
|
if r.Policy1ID.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy1ID.Int64,
|
|
Version: r.Policy1Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
Signature: r.Policy1Signature,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2ID.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy2ID.Int64,
|
|
Version: r.Policy2Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
Signature: r.Policy2Signature,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
case sqlc.ListChannelsByNodeIDRow:
|
|
if r.Policy1ID.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy1ID.Int64,
|
|
Version: r.Policy1Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
Signature: r.Policy1Signature,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2ID.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy2ID.Int64,
|
|
Version: r.Policy2Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
Signature: r.Policy2Signature,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
case sqlc.ListChannelsWithPoliciesPaginatedRow:
|
|
if r.Policy1ID.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy1ID.Int64,
|
|
Version: r.Policy1Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
Signature: r.Policy1Signature,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2ID.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy2ID.Int64,
|
|
Version: r.Policy2Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
Signature: r.Policy2Signature,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
case sqlc.GetChannelsByIDsRow:
|
|
if r.Policy1ID.Valid {
|
|
policy1 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy1ID.Int64,
|
|
Version: r.Policy1Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy1MessageFlags,
|
|
ChannelFlags: r.Policy1ChannelFlags,
|
|
Signature: r.Policy1Signature,
|
|
BlockHeight: r.Policy1BlockHeight,
|
|
DisableFlags: r.Policy1DisableFlags,
|
|
}
|
|
}
|
|
if r.Policy2ID.Valid {
|
|
policy2 = &sqlc.GraphChannelPolicy{
|
|
ID: r.Policy2ID.Int64,
|
|
Version: r.Policy2Version.Int16,
|
|
ChannelID: r.GraphChannel.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,
|
|
MessageFlags: r.Policy2MessageFlags,
|
|
ChannelFlags: r.Policy2ChannelFlags,
|
|
Signature: r.Policy2Signature,
|
|
BlockHeight: r.Policy2BlockHeight,
|
|
DisableFlags: r.Policy2DisableFlags,
|
|
}
|
|
}
|
|
|
|
return policy1, policy2, nil
|
|
|
|
default:
|
|
return nil, nil, fmt.Errorf("unexpected row type in "+
|
|
"extractChannelPolicies: %T", r)
|
|
}
|
|
}
|
|
|
|
// channelIDToBytes converts a channel ID (SCID) to a byte array
|
|
// representation.
|
|
func channelIDToBytes(channelID uint64) []byte {
|
|
var chanIDB [8]byte
|
|
byteOrder.PutUint64(chanIDB[:], channelID)
|
|
|
|
return chanIDB[:]
|
|
}
|
|
|
|
// buildNodeAddresses converts a slice of nodeAddress into a slice of net.Addr.
|
|
func buildNodeAddresses(addresses []nodeAddress) ([]net.Addr, error) {
|
|
if len(addresses) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
result := make([]net.Addr, 0, len(addresses))
|
|
for _, addr := range addresses {
|
|
netAddr, err := parseAddress(addr.addrType, addr.address)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to parse address %s "+
|
|
"of type %d: %w", addr.address, addr.addrType,
|
|
err)
|
|
}
|
|
if netAddr != nil {
|
|
result = append(result, netAddr)
|
|
}
|
|
}
|
|
|
|
// If we have no valid addresses, return nil instead of empty slice.
|
|
if len(result) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
// parseAddress parses the given address string based on the address type
|
|
// and returns a net.Addr instance. It supports IPv4, IPv6, Tor v2, Tor v3,
|
|
// and opaque addresses.
|
|
func parseAddress(addrType dbAddressType, address string) (net.Addr, error) {
|
|
switch addrType {
|
|
case addressTypeIPv4:
|
|
tcp, err := net.ResolveTCPAddr("tcp4", address)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
tcp.IP = tcp.IP.To4()
|
|
|
|
return tcp, nil
|
|
|
|
case addressTypeIPv6:
|
|
tcp, err := net.ResolveTCPAddr("tcp6", address)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return tcp, nil
|
|
|
|
case addressTypeTorV3, addressTypeTorV2:
|
|
service, portStr, err := net.SplitHostPort(address)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to split tor "+
|
|
"address: %v", address)
|
|
}
|
|
|
|
port, err := strconv.Atoi(portStr)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &tor.OnionAddr{
|
|
OnionService: service,
|
|
Port: port,
|
|
}, nil
|
|
|
|
case addressTypeDNS:
|
|
hostname, portStr, err := net.SplitHostPort(address)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to split DNS "+
|
|
"address: %v", address)
|
|
}
|
|
|
|
port, err := strconv.Atoi(portStr)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &lnwire.DNSAddress{
|
|
Hostname: hostname,
|
|
Port: uint16(port),
|
|
}, nil
|
|
|
|
case addressTypeOpaque:
|
|
opaque, err := hex.DecodeString(address)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to decode opaque "+
|
|
"address: %v", address)
|
|
}
|
|
|
|
return &lnwire.OpaqueAddrs{
|
|
Payload: opaque,
|
|
}, nil
|
|
|
|
default:
|
|
return nil, fmt.Errorf("unknown address type: %v", addrType)
|
|
}
|
|
}
|
|
|
|
// batchNodeData holds all the related data for a batch of nodes.
|
|
type batchNodeData struct {
|
|
// features is a map from a DB node ID to the feature bits for that
|
|
// node.
|
|
features map[int64][]int
|
|
|
|
// addresses is a map from a DB node ID to the node's addresses.
|
|
addresses map[int64][]nodeAddress
|
|
|
|
// extraFields is a map from a DB node ID to the extra signed fields
|
|
// for that node.
|
|
extraFields map[int64]map[uint64][]byte
|
|
}
|
|
|
|
// nodeAddress holds the address type, position and address string for a
|
|
// node. This is used to batch the fetching of node addresses.
|
|
type nodeAddress struct {
|
|
addrType dbAddressType
|
|
position int32
|
|
address string
|
|
}
|
|
|
|
// batchLoadNodeData loads all related data for a batch of node IDs using the
|
|
// provided SQLQueries interface. It returns a batchNodeData instance containing
|
|
// the node features, addresses and extra signed fields.
|
|
func batchLoadNodeData(ctx context.Context, cfg *sqldb.QueryConfig,
|
|
db SQLQueries, nodeIDs []int64) (*batchNodeData, error) {
|
|
|
|
// Batch load the node features.
|
|
features, err := batchLoadNodeFeaturesHelper(ctx, cfg, db, nodeIDs)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load node "+
|
|
"features: %w", err)
|
|
}
|
|
|
|
// Batch load the node addresses.
|
|
addrs, err := batchLoadNodeAddressesHelper(ctx, cfg, db, nodeIDs)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load node "+
|
|
"addresses: %w", err)
|
|
}
|
|
|
|
// Batch load the node extra signed fields.
|
|
extraTypes, err := batchLoadNodeExtraTypesHelper(ctx, cfg, db, nodeIDs)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load node extra "+
|
|
"signed fields: %w", err)
|
|
}
|
|
|
|
return &batchNodeData{
|
|
features: features,
|
|
addresses: addrs,
|
|
extraFields: extraTypes,
|
|
}, nil
|
|
}
|
|
|
|
// batchLoadNodeFeaturesHelper loads node features for a batch of node IDs
|
|
// using ExecuteBatchQuery wrapper around the GetNodeFeaturesBatch query.
|
|
func batchLoadNodeFeaturesHelper(ctx context.Context,
|
|
cfg *sqldb.QueryConfig, db SQLQueries,
|
|
nodeIDs []int64) (map[int64][]int, error) {
|
|
|
|
features := make(map[int64][]int)
|
|
|
|
return features, sqldb.ExecuteBatchQuery(
|
|
ctx, cfg, nodeIDs,
|
|
func(id int64) int64 {
|
|
return id
|
|
},
|
|
func(ctx context.Context, ids []int64) ([]sqlc.GraphNodeFeature,
|
|
error) {
|
|
|
|
return db.GetNodeFeaturesBatch(ctx, ids)
|
|
},
|
|
func(ctx context.Context, feature sqlc.GraphNodeFeature) error {
|
|
features[feature.NodeID] = append(
|
|
features[feature.NodeID],
|
|
int(feature.FeatureBit),
|
|
)
|
|
|
|
return nil
|
|
},
|
|
)
|
|
}
|
|
|
|
// batchLoadNodeAddressesHelper loads node addresses using ExecuteBatchQuery
|
|
// wrapper around the GetNodeAddressesBatch query. It returns a map from
|
|
// node ID to a slice of nodeAddress structs.
|
|
func batchLoadNodeAddressesHelper(ctx context.Context,
|
|
cfg *sqldb.QueryConfig, db SQLQueries,
|
|
nodeIDs []int64) (map[int64][]nodeAddress, error) {
|
|
|
|
addrs := make(map[int64][]nodeAddress)
|
|
|
|
return addrs, sqldb.ExecuteBatchQuery(
|
|
ctx, cfg, nodeIDs,
|
|
func(id int64) int64 {
|
|
return id
|
|
},
|
|
func(ctx context.Context, ids []int64) ([]sqlc.GraphNodeAddress,
|
|
error) {
|
|
|
|
return db.GetNodeAddressesBatch(ctx, ids)
|
|
},
|
|
func(ctx context.Context, addr sqlc.GraphNodeAddress) error {
|
|
addrs[addr.NodeID] = append(
|
|
addrs[addr.NodeID], nodeAddress{
|
|
addrType: dbAddressType(addr.Type),
|
|
position: addr.Position,
|
|
address: addr.Address,
|
|
},
|
|
)
|
|
|
|
return nil
|
|
},
|
|
)
|
|
}
|
|
|
|
// batchLoadNodeExtraTypesHelper loads node extra type bytes for a batch of
|
|
// node IDs using ExecuteBatchQuery wrapper around the GetNodeExtraTypesBatch
|
|
// query.
|
|
func batchLoadNodeExtraTypesHelper(ctx context.Context,
|
|
cfg *sqldb.QueryConfig, db SQLQueries,
|
|
nodeIDs []int64) (map[int64]map[uint64][]byte, error) {
|
|
|
|
extraFields := make(map[int64]map[uint64][]byte)
|
|
|
|
callback := func(ctx context.Context,
|
|
field sqlc.GraphNodeExtraType) error {
|
|
|
|
if extraFields[field.NodeID] == nil {
|
|
extraFields[field.NodeID] = make(map[uint64][]byte)
|
|
}
|
|
extraFields[field.NodeID][uint64(field.Type)] = field.Value
|
|
|
|
return nil
|
|
}
|
|
|
|
return extraFields, sqldb.ExecuteBatchQuery(
|
|
ctx, cfg, nodeIDs,
|
|
func(id int64) int64 {
|
|
return id
|
|
},
|
|
func(ctx context.Context, ids []int64) (
|
|
[]sqlc.GraphNodeExtraType, error) {
|
|
|
|
return db.GetNodeExtraTypesBatch(ctx, ids)
|
|
},
|
|
callback,
|
|
)
|
|
}
|
|
|
|
// buildChanPoliciesWithBatchData builds two models.ChannelEdgePolicy instances
|
|
// from the provided sqlc.GraphChannelPolicy records and the
|
|
// provided batchChannelData.
|
|
func buildChanPoliciesWithBatchData(dbPol1, dbPol2 *sqlc.GraphChannelPolicy,
|
|
channelID uint64, node1, node2 route.Vertex,
|
|
batchData *batchChannelData) (*models.ChannelEdgePolicy,
|
|
*models.ChannelEdgePolicy, error) {
|
|
|
|
pol1, err := buildChanPolicyWithBatchData(
|
|
true, dbPol1, channelID, node2, batchData,
|
|
)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("unable to build policy1: %w", err)
|
|
}
|
|
|
|
pol2, err := buildChanPolicyWithBatchData(
|
|
false, dbPol2, channelID, node1, batchData,
|
|
)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("unable to build policy2: %w", err)
|
|
}
|
|
|
|
return pol1, pol2, nil
|
|
}
|
|
|
|
// buildChanPolicyWithBatchData builds a models.ChannelEdgePolicy instance from
|
|
// the provided sqlc.GraphChannelPolicy and the provided batchChannelData.
|
|
func buildChanPolicyWithBatchData(isNode1 bool,
|
|
dbPol *sqlc.GraphChannelPolicy, channelID uint64,
|
|
toNode route.Vertex, batchData *batchChannelData) (
|
|
*models.ChannelEdgePolicy, error) {
|
|
|
|
if dbPol == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
var dbPol1Extras map[uint64][]byte
|
|
if extras, exists := batchData.policyExtras[dbPol.ID]; exists {
|
|
dbPol1Extras = extras
|
|
} else {
|
|
dbPol1Extras = make(map[uint64][]byte)
|
|
}
|
|
|
|
return buildChanPolicy(isNode1, *dbPol, channelID, dbPol1Extras, toNode)
|
|
}
|
|
|
|
// batchChannelData holds all the related data for a batch of channels.
|
|
type batchChannelData struct {
|
|
// chanFeatures is a map from DB channel ID to a slice of feature bits.
|
|
chanfeatures map[int64][]int
|
|
|
|
// chanExtras is a map from DB channel ID to a map of TLV type to
|
|
// extra signed field bytes.
|
|
chanExtraTypes map[int64]map[uint64][]byte
|
|
|
|
// policyExtras is a map from DB channel policy ID to a map of TLV type
|
|
// to extra signed field bytes.
|
|
policyExtras map[int64]map[uint64][]byte
|
|
}
|
|
|
|
// batchLoadChannelData loads all related data for batches of channels and
|
|
// policies.
|
|
func batchLoadChannelData(ctx context.Context, cfg *sqldb.QueryConfig,
|
|
db SQLQueries, channelIDs []int64,
|
|
policyIDs []int64) (*batchChannelData, error) {
|
|
|
|
batchData := &batchChannelData{
|
|
chanfeatures: make(map[int64][]int),
|
|
chanExtraTypes: make(map[int64]map[uint64][]byte),
|
|
policyExtras: make(map[int64]map[uint64][]byte),
|
|
}
|
|
|
|
// Batch load channel features and extras
|
|
var err error
|
|
if len(channelIDs) > 0 {
|
|
batchData.chanfeatures, err = batchLoadChannelFeaturesHelper(
|
|
ctx, cfg, db, channelIDs,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load "+
|
|
"channel features: %w", err)
|
|
}
|
|
|
|
batchData.chanExtraTypes, err = batchLoadChannelExtrasHelper(
|
|
ctx, cfg, db, channelIDs,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load "+
|
|
"channel extras: %w", err)
|
|
}
|
|
}
|
|
|
|
if len(policyIDs) > 0 {
|
|
policyExtras, err := batchLoadChannelPolicyExtrasHelper(
|
|
ctx, cfg, db, policyIDs,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load "+
|
|
"policy extras: %w", err)
|
|
}
|
|
batchData.policyExtras = policyExtras
|
|
}
|
|
|
|
return batchData, nil
|
|
}
|
|
|
|
// batchLoadChannelFeaturesHelper loads channel features for a batch of
|
|
// channel IDs using ExecuteBatchQuery wrapper around the
|
|
// GetChannelFeaturesBatch query. It returns a map from DB channel ID to a
|
|
// slice of feature bits.
|
|
func batchLoadChannelFeaturesHelper(ctx context.Context,
|
|
cfg *sqldb.QueryConfig, db SQLQueries,
|
|
channelIDs []int64) (map[int64][]int, error) {
|
|
|
|
features := make(map[int64][]int)
|
|
|
|
return features, sqldb.ExecuteBatchQuery(
|
|
ctx, cfg, channelIDs,
|
|
func(id int64) int64 {
|
|
return id
|
|
},
|
|
func(ctx context.Context,
|
|
ids []int64) ([]sqlc.GraphChannelFeature, error) {
|
|
|
|
return db.GetChannelFeaturesBatch(ctx, ids)
|
|
},
|
|
func(ctx context.Context,
|
|
feature sqlc.GraphChannelFeature) error {
|
|
|
|
features[feature.ChannelID] = append(
|
|
features[feature.ChannelID],
|
|
int(feature.FeatureBit),
|
|
)
|
|
|
|
return nil
|
|
},
|
|
)
|
|
}
|
|
|
|
// batchLoadChannelExtrasHelper loads channel extra types for a batch of
|
|
// channel IDs using ExecuteBatchQuery wrapper around the GetChannelExtrasBatch
|
|
// query. It returns a map from DB channel ID to a map of TLV type to extra
|
|
// signed field bytes.
|
|
func batchLoadChannelExtrasHelper(ctx context.Context,
|
|
cfg *sqldb.QueryConfig, db SQLQueries,
|
|
channelIDs []int64) (map[int64]map[uint64][]byte, error) {
|
|
|
|
extras := make(map[int64]map[uint64][]byte)
|
|
|
|
cb := func(ctx context.Context,
|
|
extra sqlc.GraphChannelExtraType) error {
|
|
|
|
if extras[extra.ChannelID] == nil {
|
|
extras[extra.ChannelID] = make(map[uint64][]byte)
|
|
}
|
|
extras[extra.ChannelID][uint64(extra.Type)] = extra.Value
|
|
|
|
return nil
|
|
}
|
|
|
|
return extras, sqldb.ExecuteBatchQuery(
|
|
ctx, cfg, channelIDs,
|
|
func(id int64) int64 {
|
|
return id
|
|
},
|
|
func(ctx context.Context,
|
|
ids []int64) ([]sqlc.GraphChannelExtraType, error) {
|
|
|
|
return db.GetChannelExtrasBatch(ctx, ids)
|
|
}, cb,
|
|
)
|
|
}
|
|
|
|
// batchLoadChannelPolicyExtrasHelper loads channel policy extra types for a
|
|
// batch of policy IDs using ExecuteBatchQuery wrapper around the
|
|
// GetChannelPolicyExtraTypesBatch query. It returns a map from DB policy ID to
|
|
// a map of TLV type to extra signed field bytes.
|
|
func batchLoadChannelPolicyExtrasHelper(ctx context.Context,
|
|
cfg *sqldb.QueryConfig, db SQLQueries,
|
|
policyIDs []int64) (map[int64]map[uint64][]byte, error) {
|
|
|
|
extras := make(map[int64]map[uint64][]byte)
|
|
|
|
return extras, sqldb.ExecuteBatchQuery(
|
|
ctx, cfg, policyIDs,
|
|
func(id int64) int64 {
|
|
return id
|
|
},
|
|
func(ctx context.Context, ids []int64) (
|
|
[]sqlc.GetChannelPolicyExtraTypesBatchRow, error) {
|
|
|
|
return db.GetChannelPolicyExtraTypesBatch(ctx, ids)
|
|
},
|
|
func(ctx context.Context,
|
|
row sqlc.GetChannelPolicyExtraTypesBatchRow) error {
|
|
|
|
if extras[row.PolicyID] == nil {
|
|
extras[row.PolicyID] = make(map[uint64][]byte)
|
|
}
|
|
extras[row.PolicyID][uint64(row.Type)] = row.Value
|
|
|
|
return nil
|
|
},
|
|
)
|
|
}
|
|
|
|
// forEachNodePaginated executes a paginated query to process each node in the
|
|
// graph. It uses the provided SQLQueries interface to fetch nodes in batches
|
|
// and applies the provided processNode function to each node.
|
|
func forEachNodePaginated(ctx context.Context, cfg *sqldb.QueryConfig,
|
|
db SQLQueries, protocol lnwire.GossipVersion,
|
|
processNode func(context.Context, int64,
|
|
*models.Node) error) error {
|
|
|
|
pageQueryFunc := func(ctx context.Context, lastID int64,
|
|
limit int32) ([]sqlc.GraphNode, error) {
|
|
|
|
return db.ListNodesPaginated(
|
|
ctx, sqlc.ListNodesPaginatedParams{
|
|
Version: int16(protocol),
|
|
ID: lastID,
|
|
Limit: limit,
|
|
},
|
|
)
|
|
}
|
|
|
|
extractPageCursor := func(node sqlc.GraphNode) int64 {
|
|
return node.ID
|
|
}
|
|
|
|
collectFunc := func(node sqlc.GraphNode) (int64, error) {
|
|
return node.ID, nil
|
|
}
|
|
|
|
batchQueryFunc := func(ctx context.Context,
|
|
nodeIDs []int64) (*batchNodeData, error) {
|
|
|
|
return batchLoadNodeData(ctx, cfg, db, nodeIDs)
|
|
}
|
|
|
|
processItem := func(ctx context.Context, dbNode sqlc.GraphNode,
|
|
batchData *batchNodeData) error {
|
|
|
|
node, err := buildNodeWithBatchData(dbNode, batchData)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to build "+
|
|
"node(id=%d): %w", dbNode.ID, err)
|
|
}
|
|
|
|
return processNode(ctx, dbNode.ID, node)
|
|
}
|
|
|
|
return sqldb.ExecuteCollectAndBatchWithSharedDataQuery(
|
|
ctx, cfg, int64(-1), pageQueryFunc, extractPageCursor,
|
|
collectFunc, batchQueryFunc, processItem,
|
|
)
|
|
}
|
|
|
|
// forEachChannelWithPolicies executes a paginated query to process each channel
|
|
// with policies in the graph.
|
|
func forEachChannelWithPolicies(ctx context.Context, db SQLQueries,
|
|
cfg *SQLStoreConfig, v lnwire.GossipVersion,
|
|
processChannel func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
|
|
*models.ChannelEdgePolicy) error) error {
|
|
|
|
type channelBatchIDs struct {
|
|
channelID int64
|
|
policyIDs []int64
|
|
}
|
|
|
|
pageQueryFunc := func(ctx context.Context, lastID int64,
|
|
limit int32) ([]sqlc.ListChannelsWithPoliciesPaginatedRow,
|
|
error) {
|
|
|
|
return db.ListChannelsWithPoliciesPaginated(
|
|
ctx, sqlc.ListChannelsWithPoliciesPaginatedParams{
|
|
Version: int16(v),
|
|
ID: lastID,
|
|
Limit: limit,
|
|
},
|
|
)
|
|
}
|
|
|
|
extractPageCursor := func(
|
|
row sqlc.ListChannelsWithPoliciesPaginatedRow) int64 {
|
|
|
|
return row.GraphChannel.ID
|
|
}
|
|
|
|
collectFunc := func(row sqlc.ListChannelsWithPoliciesPaginatedRow) (
|
|
channelBatchIDs, error) {
|
|
|
|
ids := channelBatchIDs{
|
|
channelID: row.GraphChannel.ID,
|
|
}
|
|
|
|
// Extract policy IDs from the row.
|
|
dbPol1, dbPol2, err := extractChannelPolicies(row)
|
|
if err != nil {
|
|
return ids, err
|
|
}
|
|
|
|
if dbPol1 != nil {
|
|
ids.policyIDs = append(ids.policyIDs, dbPol1.ID)
|
|
}
|
|
if dbPol2 != nil {
|
|
ids.policyIDs = append(ids.policyIDs, dbPol2.ID)
|
|
}
|
|
|
|
return ids, nil
|
|
}
|
|
|
|
batchDataFunc := func(ctx context.Context,
|
|
allIDs []channelBatchIDs) (*batchChannelData, error) {
|
|
|
|
// Separate channel IDs from policy IDs.
|
|
var (
|
|
channelIDs = make([]int64, len(allIDs))
|
|
policyIDs = make([]int64, 0, len(allIDs)*2)
|
|
)
|
|
|
|
for i, ids := range allIDs {
|
|
channelIDs[i] = ids.channelID
|
|
policyIDs = append(policyIDs, ids.policyIDs...)
|
|
}
|
|
|
|
return batchLoadChannelData(
|
|
ctx, cfg.QueryCfg, db, channelIDs, policyIDs,
|
|
)
|
|
}
|
|
|
|
processItem := func(ctx context.Context,
|
|
row sqlc.ListChannelsWithPoliciesPaginatedRow,
|
|
batchData *batchChannelData) error {
|
|
|
|
node1, node2, err := buildNodeVertices(
|
|
row.Node1Pubkey, row.Node2Pubkey,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
edge, err := buildEdgeInfoWithBatchData(
|
|
cfg.ChainHash, row.GraphChannel, node1, node2,
|
|
batchData,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to build channel info: %w",
|
|
err)
|
|
}
|
|
|
|
dbPol1, dbPol2, err := extractChannelPolicies(row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
p1, p2, err := buildChanPoliciesWithBatchData(
|
|
dbPol1, dbPol2, edge.ChannelID, node1, node2, batchData,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return processChannel(edge, p1, p2)
|
|
}
|
|
|
|
return sqldb.ExecuteCollectAndBatchWithSharedDataQuery(
|
|
ctx, cfg.QueryCfg, int64(-1), pageQueryFunc, extractPageCursor,
|
|
collectFunc, batchDataFunc, processItem,
|
|
)
|
|
}
|
|
|
|
// buildDirectedChannel builds a DirectedChannel instance from the provided
|
|
// data.
|
|
func buildDirectedChannel(chain chainhash.Hash, nodeID int64,
|
|
nodePub route.Vertex, channelRow sqlc.ListChannelsForNodeIDsRow,
|
|
channelBatchData *batchChannelData, features *lnwire.FeatureVector,
|
|
toNodeCallback func() route.Vertex) (*DirectedChannel, error) {
|
|
|
|
node1, node2, err := buildNodeVertices(
|
|
channelRow.Node1Pubkey, channelRow.Node2Pubkey,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to build node vertices: %w", err)
|
|
}
|
|
|
|
edge, err := buildEdgeInfoWithBatchData(
|
|
chain, channelRow.GraphChannel, node1, node2, channelBatchData,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to build channel info: %w", err)
|
|
}
|
|
|
|
dbPol1, dbPol2, err := extractChannelPolicies(channelRow)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to extract channel policies: %w",
|
|
err)
|
|
}
|
|
|
|
p1, p2, err := buildChanPoliciesWithBatchData(
|
|
dbPol1, dbPol2, edge.ChannelID, node1, node2,
|
|
channelBatchData,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to build channel policies: %w",
|
|
err)
|
|
}
|
|
|
|
// Determine outgoing and incoming policy for this specific node.
|
|
p1ToNode := channelRow.GraphChannel.NodeID2
|
|
p2ToNode := channelRow.GraphChannel.NodeID1
|
|
outPolicy, inPolicy := p1, p2
|
|
if (p1 != nil && p1ToNode == nodeID) ||
|
|
(p2 != nil && p2ToNode != nodeID) {
|
|
|
|
outPolicy, inPolicy = p2, p1
|
|
}
|
|
|
|
// Build cached policy.
|
|
var cachedInPolicy *models.CachedEdgePolicy
|
|
if inPolicy != nil {
|
|
cachedInPolicy = models.NewCachedPolicy(inPolicy)
|
|
cachedInPolicy.ToNodePubKey = toNodeCallback
|
|
cachedInPolicy.ToNodeFeatures = features
|
|
}
|
|
|
|
// Extract inbound fee.
|
|
var inboundFee lnwire.Fee
|
|
if outPolicy != nil {
|
|
outPolicy.InboundFee.WhenSome(func(fee lnwire.Fee) {
|
|
inboundFee = fee
|
|
})
|
|
}
|
|
|
|
// Build directed channel.
|
|
directedChannel := &DirectedChannel{
|
|
ChannelID: edge.ChannelID,
|
|
IsNode1: nodePub == edge.NodeKey1Bytes,
|
|
OtherNode: edge.NodeKey2Bytes,
|
|
Capacity: edge.Capacity,
|
|
OutPolicySet: outPolicy != nil,
|
|
InPolicy: cachedInPolicy,
|
|
InboundFee: inboundFee,
|
|
}
|
|
|
|
if nodePub == edge.NodeKey2Bytes {
|
|
directedChannel.OtherNode = edge.NodeKey1Bytes
|
|
}
|
|
|
|
return directedChannel, nil
|
|
}
|
|
|
|
// batchBuildChannelEdges builds a slice of ChannelEdge instances from the
|
|
// provided rows. It uses batch loading for channels, policies, and nodes.
|
|
func batchBuildChannelEdges[T sqlc.ChannelAndNodes](ctx context.Context,
|
|
cfg *SQLStoreConfig, db SQLQueries, rows []T) ([]ChannelEdge, error) {
|
|
|
|
var (
|
|
channelIDs = make([]int64, len(rows))
|
|
policyIDs = make([]int64, 0, len(rows)*2)
|
|
nodeIDs = make([]int64, 0, len(rows)*2)
|
|
|
|
// nodeIDSet is used to ensure we only collect unique node IDs.
|
|
nodeIDSet = make(map[int64]bool)
|
|
|
|
// edges will hold the final channel edges built from the rows.
|
|
edges = make([]ChannelEdge, 0, len(rows))
|
|
)
|
|
|
|
// Collect all IDs needed for batch loading.
|
|
for i, row := range rows {
|
|
channelIDs[i] = row.Channel().ID
|
|
|
|
// Collect policy IDs
|
|
dbPol1, dbPol2, err := extractChannelPolicies(row)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to extract channel "+
|
|
"policies: %w", err)
|
|
}
|
|
if dbPol1 != nil {
|
|
policyIDs = append(policyIDs, dbPol1.ID)
|
|
}
|
|
if dbPol2 != nil {
|
|
policyIDs = append(policyIDs, dbPol2.ID)
|
|
}
|
|
|
|
var (
|
|
node1ID = row.Node1().ID
|
|
node2ID = row.Node2().ID
|
|
)
|
|
|
|
// Collect unique node IDs.
|
|
if !nodeIDSet[node1ID] {
|
|
nodeIDs = append(nodeIDs, node1ID)
|
|
nodeIDSet[node1ID] = true
|
|
}
|
|
|
|
if !nodeIDSet[node2ID] {
|
|
nodeIDs = append(nodeIDs, node2ID)
|
|
nodeIDSet[node2ID] = true
|
|
}
|
|
}
|
|
|
|
// Batch the data for all the channels and policies.
|
|
channelBatchData, err := batchLoadChannelData(
|
|
ctx, cfg.QueryCfg, db, channelIDs, policyIDs,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load channel and "+
|
|
"policy data: %w", err)
|
|
}
|
|
|
|
// Batch the data for all the nodes.
|
|
nodeBatchData, err := batchLoadNodeData(ctx, cfg.QueryCfg, db, nodeIDs)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to batch load node data: %w",
|
|
err)
|
|
}
|
|
|
|
// Build all channel edges using batch data.
|
|
for _, row := range rows {
|
|
// Build nodes using batch data.
|
|
node1, err := buildNodeWithBatchData(row.Node1(), nodeBatchData)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to build node1: %w", err)
|
|
}
|
|
|
|
node2, err := buildNodeWithBatchData(row.Node2(), nodeBatchData)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to build node2: %w", err)
|
|
}
|
|
|
|
// Build channel info using batch data.
|
|
channel, err := buildEdgeInfoWithBatchData(
|
|
cfg.ChainHash, row.Channel(), node1.PubKeyBytes,
|
|
node2.PubKeyBytes, channelBatchData,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to build channel "+
|
|
"info: %w", err)
|
|
}
|
|
|
|
// Extract and build policies using batch data.
|
|
dbPol1, dbPol2, err := extractChannelPolicies(row)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to extract channel "+
|
|
"policies: %w", err)
|
|
}
|
|
|
|
p1, p2, err := buildChanPoliciesWithBatchData(
|
|
dbPol1, dbPol2, channel.ChannelID,
|
|
node1.PubKeyBytes, node2.PubKeyBytes, channelBatchData,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to build channel "+
|
|
"policies: %w", err)
|
|
}
|
|
|
|
edges = append(edges, ChannelEdge{
|
|
Info: channel,
|
|
Policy1: p1,
|
|
Policy2: p2,
|
|
Node1: node1,
|
|
Node2: node2,
|
|
})
|
|
}
|
|
|
|
return edges, nil
|
|
}
|
|
|
|
// batchBuildChannelInfo builds a slice of models.ChannelEdgeInfo
|
|
// instances from the provided rows using batch loading for channel data.
|
|
func batchBuildChannelInfo[T sqlc.ChannelAndNodeIDs](ctx context.Context,
|
|
cfg *SQLStoreConfig, db SQLQueries, rows []T) (
|
|
[]*models.ChannelEdgeInfo, []int64, error) {
|
|
|
|
if len(rows) == 0 {
|
|
return nil, nil, nil
|
|
}
|
|
|
|
// Collect all the channel IDs needed for batch loading.
|
|
channelIDs := make([]int64, len(rows))
|
|
for i, row := range rows {
|
|
channelIDs[i] = row.Channel().ID
|
|
}
|
|
|
|
// Batch load the channel data.
|
|
channelBatchData, err := batchLoadChannelData(
|
|
ctx, cfg.QueryCfg, db, channelIDs, nil,
|
|
)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("unable to batch load channel "+
|
|
"data: %w", err)
|
|
}
|
|
|
|
// Build all channel edges using batch data.
|
|
edges := make([]*models.ChannelEdgeInfo, 0, len(rows))
|
|
for _, row := range rows {
|
|
node1, node2, err := buildNodeVertices(
|
|
row.Node1Pub(), row.Node2Pub(),
|
|
)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
// Build channel info using batch data
|
|
info, err := buildEdgeInfoWithBatchData(
|
|
cfg.ChainHash, row.Channel(), node1, node2,
|
|
channelBatchData,
|
|
)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
edges = append(edges, info)
|
|
}
|
|
|
|
return edges, channelIDs, nil
|
|
}
|
|
|
|
// handleZombieMarking is a helper function that handles the logic of
|
|
// marking a channel as a zombie in the database. It takes into account whether
|
|
// we are in strict zombie pruning mode, and adjusts the node public keys
|
|
// accordingly based on the last update timestamps of the channel policies.
|
|
func handleZombieMarking(ctx context.Context, db SQLQueries,
|
|
v lnwire.GossipVersion,
|
|
row sqlc.GetChannelsBySCIDWithPoliciesRow, info *models.ChannelEdgeInfo,
|
|
strictZombiePruning bool, scid uint64) error {
|
|
|
|
nodeKey1, nodeKey2 := info.NodeKey1Bytes, info.NodeKey2Bytes
|
|
|
|
if strictZombiePruning {
|
|
// TODO(elle): update for V2 last update times.
|
|
if v != gossipV1 {
|
|
return fmt.Errorf("strict zombie pruning only "+
|
|
"supported for gossip v1, got %v", v)
|
|
}
|
|
|
|
var e1UpdateTime, e2UpdateTime *time.Time
|
|
if row.Policy1LastUpdate.Valid {
|
|
e1Time := time.Unix(row.Policy1LastUpdate.Int64, 0)
|
|
e1UpdateTime = &e1Time
|
|
}
|
|
if row.Policy2LastUpdate.Valid {
|
|
e2Time := time.Unix(row.Policy2LastUpdate.Int64, 0)
|
|
e2UpdateTime = &e2Time
|
|
}
|
|
|
|
nodeKey1, nodeKey2 = makeZombiePubkeys(
|
|
info.NodeKey1Bytes, info.NodeKey2Bytes, e1UpdateTime,
|
|
e2UpdateTime,
|
|
)
|
|
}
|
|
|
|
return db.UpsertZombieChannel(
|
|
ctx, sqlc.UpsertZombieChannelParams{
|
|
Version: int16(v),
|
|
Scid: channelIDToBytes(scid),
|
|
NodeKey1: nodeKey1[:],
|
|
NodeKey2: nodeKey2[:],
|
|
},
|
|
)
|
|
}
|