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