Merge pull request #9939 from ellemouton/graphSQL15-prune-log

[15] graph/db: SQL prune log

docs/release-notes/release-notes-0.20.0.md

@@ -88,6 +88,7 @@ circuit. The indices are only available for forwarding events saved after v0.20.
     * [6](https://github.com/lightningnetwork/lnd/pull/9936)
     * [7](https://github.com/lightningnetwork/lnd/pull/9937)
     * [8](https://github.com/lightningnetwork/lnd/pull/9938)
+    * [9](https://github.com/lightningnetwork/lnd/pull/9939)
 
 ## RPC Updates
 

graph/db/graph_test.go

@@ -513,7 +513,7 @@ func TestDisconnectBlockAtHeight(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	sourceNode := createTestVertex(t)
 	if err := graph.SetSourceNode(ctx, sourceNode); err != nil {
@@ -1685,7 +1685,7 @@ func TestGraphPruning(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	sourceNode := createTestVertex(t)
 	if err := graph.SetSourceNode(ctx, sourceNode); err != nil {
@@ -2440,7 +2440,7 @@ func TestStressTestChannelGraphAPI(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	node1 := createTestVertex(t)
 	require.NoError(t, graph.AddLightningNode(ctx, node1))
@@ -2448,6 +2448,10 @@ func TestStressTestChannelGraphAPI(t *testing.T) {
 	node2 := createTestVertex(t)
 	require.NoError(t, graph.AddLightningNode(ctx, node2))
 
+	// We need to update the node's timestamp since this call to
+	// SetSourceNode will trigger an upsert which will only be allowed if
+	// the newest LastUpdate time is greater than the current one.
+	node1.LastUpdate = node1.LastUpdate.Add(time.Second)
 	require.NoError(t, graph.SetSourceNode(ctx, node1))
 
 	type chanInfo struct {
@@ -3291,7 +3295,7 @@ func TestPruneGraphNodes(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	// We'll start off by inserting our source node, to ensure that it's
 	// the only node left after we prune the graph.
@@ -4006,7 +4010,7 @@ func TestBatchedAddChannelEdge(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
-	graph := MakeTestGraph(t)
+	graph := MakeTestGraphNew(t)
 
 	sourceNode := createTestVertex(t)
 	require.Nil(t, graph.SetSourceNode(ctx, sourceNode))

graph/db/sql_store.go

@@ -19,6 +19,7 @@ import (
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/aliasmgr"
 	"github.com/lightningnetwork/lnd/batch"
 	"github.com/lightningnetwork/lnd/fn/v2"
 	"github.com/lightningnetwork/lnd/graph/db/models"
@@ -63,7 +64,9 @@ type SQLQueries interface {
 	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)
+	GetUnconnectedNodes(ctx context.Context) ([]sqlc.GetUnconnectedNodesRow, error)
 	DeleteNodeByPubKey(ctx context.Context, arg sqlc.DeleteNodeByPubKeyParams) (sql.Result, error)
+	DeleteNode(ctx context.Context, id int64) error
 
 	GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]sqlc.NodeExtraType, error)
 	UpsertNodeExtraType(ctx context.Context, arg sqlc.UpsertNodeExtraTypeParams) error
@@ -89,12 +92,15 @@ type SQLQueries interface {
 	*/
 	CreateChannel(ctx context.Context, arg sqlc.CreateChannelParams) (int64, error)
 	GetChannelBySCID(ctx context.Context, arg sqlc.GetChannelBySCIDParams) (sqlc.Channel, error)
+	GetChannelByOutpoint(ctx context.Context, outpoint string) (sqlc.GetChannelByOutpointRow, error)
+	GetChannelsBySCIDRange(ctx context.Context, arg sqlc.GetChannelsBySCIDRangeParams) ([]sqlc.GetChannelsBySCIDRangeRow, error)
 	GetChannelBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelBySCIDWithPoliciesParams) (sqlc.GetChannelBySCIDWithPoliciesRow, error)
 	GetChannelAndNodesBySCID(ctx context.Context, arg sqlc.GetChannelAndNodesBySCIDParams) (sqlc.GetChannelAndNodesBySCIDRow, error)
 	GetChannelFeaturesAndExtras(ctx context.Context, channelID int64) ([]sqlc.GetChannelFeaturesAndExtrasRow, error)
 	HighestSCID(ctx context.Context, version int16) ([]byte, error)
 	ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error)
 	ListChannelsWithPoliciesPaginated(ctx context.Context, arg sqlc.ListChannelsWithPoliciesPaginatedParams) ([]sqlc.ListChannelsWithPoliciesPaginatedRow, error)
+	ListChannelsPaginated(ctx context.Context, arg sqlc.ListChannelsPaginatedParams) ([]sqlc.ListChannelsPaginatedRow, 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)
@@ -123,6 +129,13 @@ type SQLQueries interface {
 	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.PruneLog, error)
+	UpsertPruneLogEntry(ctx context.Context, arg sqlc.UpsertPruneLogEntryParams) error
+	DeletePruneLogEntriesInRange(ctx context.Context, arg sqlc.DeletePruneLogEntriesInRangeParams) error
 }
 
 // BatchedSQLQueries is a version of SQLQueries that's capable of batched
@@ -2200,6 +2213,368 @@ func (s *SQLStore) FilterKnownChanIDs(chansInfo []ChannelUpdateInfo) ([]uint64,
 	return newChanIDs, knownZombies, nil
 }
 
+// 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 V1Store interface.
+func (s *SQLStore) PruneGraphNodes() ([]route.Vertex, error) {
+	var ctx = context.TODO()
+
+	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 V1Store interface.
+func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint,
+	blockHash *chainhash.Hash, blockHeight uint32) (
+	[]*models.ChannelEdgeInfo, []route.Vertex, error) {
+
+	ctx := context.TODO()
+
+	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 {
+		for _, outpoint := range spentOutputs {
+			// TODO(elle): potentially optimize this by using
+			//  sqlc.slice() once that works for both SQLite and
+			//  Postgres.
+			//
+			// NOTE: this fetches channels for all protocol
+			// versions.
+			row, err := db.GetChannelByOutpoint(
+				ctx, outpoint.String(),
+			)
+			if errors.Is(err, sql.ErrNoRows) {
+				continue
+			} 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
+			}
+
+			info, err := getAndBuildEdgeInfo(
+				ctx, db, s.cfg.ChainHash, row.Channel.ID,
+				row.Channel, node1, node2,
+			)
+			if err != nil {
+				return err
+			}
+
+			err = db.DeleteChannel(ctx, row.Channel.ID)
+			if err != nil {
+				return fmt.Errorf("unable to delete "+
+					"channel: %w", err)
+			}
+
+			closedChans = append(closedChans, info)
+		}
+
+		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.ChannelID)
+		s.chanCache.remove(channel.ChannelID)
+	}
+
+	return closedChans, prunedNodes, 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 V1Store interface.
+func (s *SQLStore) ChannelView() ([]EdgePoint, error) {
+	var (
+		ctx        = context.TODO()
+		edgePoints []EdgePoint
+	)
+
+	handleChannel := func(db SQLQueries,
+		channel sqlc.ListChannelsPaginatedRow) error {
+
+		pkScript, err := genMultiSigP2WSH(
+			channel.BitcoinKey1, channel.BitcoinKey2,
+		)
+		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
+	}
+
+	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
+		lastID := int64(-1)
+		for {
+			rows, err := db.ListChannelsPaginated(
+				ctx, sqlc.ListChannelsPaginatedParams{
+					Version: int16(ProtocolV1),
+					ID:      lastID,
+					Limit:   pageSize,
+				},
+			)
+			if err != nil {
+				return err
+			}
+
+			if len(rows) == 0 {
+				break
+			}
+
+			for _, row := range rows {
+				err := handleChannel(db, row)
+				if err != nil {
+					return err
+				}
+
+				lastID = row.ID
+			}
+		}
+
+		return nil
+	}, 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 V1Store interface.
+func (s *SQLStore) PruneTip() (*chainhash.Hash, uint32, error) {
+	var (
+		ctx       = context.TODO()
+		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) {
+
+	// Fetch all un-connected nodes from the database.
+	// NOTE: this will not include any nodes that are listed in the
+	// source table.
+	nodes, err := db.GetUnconnectedNodes(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("unable to fetch unconnected nodes: %w",
+			err)
+	}
+
+	prunedNodes := make([]route.Vertex, 0, len(nodes))
+	for _, node := range nodes {
+		// TODO(elle): update to use sqlc.slice() once that works.
+		if err = db.DeleteNode(ctx, node.ID); err != nil {
+			return nil, fmt.Errorf("unable to delete "+
+				"node(id=%d): %w", node.ID, err)
+		}
+
+		pubKey, err := route.NewVertexFromBytes(node.PubKey)
+		if err != nil {
+			return nil, fmt.Errorf("unable to parse pubkey "+
+				"for node(id=%d): %w", node.ID, err)
+		}
+
+		prunedNodes = append(prunedNodes, pubKey)
+	}
+
+	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 V1Store interface.
+func (s *SQLStore) DisconnectBlockAtHeight(height uint32) (
+	[]*models.ChannelEdgeInfo, error) {
+
+	ctx := context.TODO()
+
+	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
+	)
+
+	var chanIDStart [8]byte
+	byteOrder.PutUint64(chanIDStart[:], startShortChanID.ToUint64())
+	var chanIDEnd [8]byte
+	byteOrder.PutUint64(chanIDEnd[:], 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)
+		}
+
+		for _, row := range rows {
+			node1, node2, err := buildNodeVertices(
+				row.Node1PubKey, row.Node2PubKey,
+			)
+			if err != nil {
+				return err
+			}
+
+			channel, err := getAndBuildEdgeInfo(
+				ctx, db, s.cfg.ChainHash, row.Channel.ID,
+				row.Channel, node1, node2,
+			)
+			if err != nil {
+				return err
+			}
+
+			err = db.DeleteChannel(ctx, row.Channel.ID)
+			if err != nil {
+				return fmt.Errorf("unable to delete "+
+					"channel: %w", err)
+			}
+
+			removedChans = append(removedChans, channel)
+		}
+
+		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)
+	}
+
+	for _, channel := range removedChans {
+		s.rejectCache.remove(channel.ChannelID)
+		s.chanCache.remove(channel.ChannelID)
+	}
+
+	return removedChans, 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
@@ -3323,6 +3698,11 @@ func getAndBuildEdgeInfo(ctx context.Context, db SQLQueries,
 	chain chainhash.Hash, dbChanID int64, dbChan sqlc.Channel, node1,
 	node2 route.Vertex) (*models.ChannelEdgeInfo, error) {
 
+	if dbChan.Version != int16(ProtocolV1) {
+		return nil, fmt.Errorf("unsupported channel version: %d",
+			dbChan.Version)
+	}
+
 	fv, extras, err := getChanFeaturesAndExtras(
 		ctx, db, dbChanID,
 	)

sqldb/sqlc/graph.sql.go

@@ -149,6 +149,16 @@ func (q *Queries) DeleteExtraNodeType(ctx context.Context, arg DeleteExtraNodeTy
 	return err
 }
 
+const deleteNode = `-- name: DeleteNode :exec
+DELETE FROM nodes
+WHERE id = $1
+`
+
+func (q *Queries) DeleteNode(ctx context.Context, id int64) error {
+	_, err := q.db.ExecContext(ctx, deleteNode, id)
+	return err
+}
+
 const deleteNodeAddresses = `-- name: DeleteNodeAddresses :exec
 DELETE FROM node_addresses
 WHERE node_id = $1
@@ -190,6 +200,22 @@ func (q *Queries) DeleteNodeFeature(ctx context.Context, arg DeleteNodeFeaturePa
 	return err
 }
 
+const deletePruneLogEntriesInRange = `-- name: DeletePruneLogEntriesInRange :exec
+DELETE FROM prune_log
+WHERE block_height >= $1
+  AND block_height <= $2
+`
+
+type DeletePruneLogEntriesInRangeParams struct {
+	StartHeight int64
+	EndHeight   int64
+}
+
+func (q *Queries) DeletePruneLogEntriesInRange(ctx context.Context, arg DeletePruneLogEntriesInRangeParams) error {
+	_, err := q.db.ExecContext(ctx, deletePruneLogEntriesInRange, arg.StartHeight, arg.EndHeight)
+	return err
+}
+
 const deleteZombieChannel = `-- name: DeleteZombieChannel :execresult
 DELETE FROM zombie_channels
 WHERE scid = $1
@@ -263,6 +289,46 @@ func (q *Queries) GetChannelAndNodesBySCID(ctx context.Context, arg GetChannelAn
 	return i, err
 }
 
+const getChannelByOutpoint = `-- name: GetChannelByOutpoint :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
+FROM channels c
+    JOIN nodes n1 ON c.node_id_1 = n1.id
+    JOIN nodes n2 ON c.node_id_2 = n2.id
+WHERE c.outpoint = $1
+`
+
+type GetChannelByOutpointRow struct {
+	Channel     Channel
+	Node1Pubkey []byte
+	Node2Pubkey []byte
+}
+
+func (q *Queries) GetChannelByOutpoint(ctx context.Context, outpoint string) (GetChannelByOutpointRow, error) {
+	row := q.db.QueryRowContext(ctx, getChannelByOutpoint, outpoint)
+	var i GetChannelByOutpointRow
+	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,
+	)
+	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,
@@ -893,6 +959,67 @@ func (q *Queries) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg Ge
 	return items, nil
 }
 
+const getChannelsBySCIDRange = `-- name: GetChannelsBySCIDRange :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_pub_key,
+    n2.pub_key AS node2_pub_key
+FROM channels c
+    JOIN nodes n1 ON c.node_id_1 = n1.id
+    JOIN nodes n2 ON c.node_id_2 = n2.id
+WHERE scid >= $1
+  AND scid < $2
+`
+
+type GetChannelsBySCIDRangeParams struct {
+	StartScid []byte
+	EndScid   []byte
+}
+
+type GetChannelsBySCIDRangeRow struct {
+	Channel     Channel
+	Node1PubKey []byte
+	Node2PubKey []byte
+}
+
+func (q *Queries) GetChannelsBySCIDRange(ctx context.Context, arg GetChannelsBySCIDRangeParams) ([]GetChannelsBySCIDRangeRow, error) {
+	rows, err := q.db.QueryContext(ctx, getChannelsBySCIDRange, arg.StartScid, arg.EndScid)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetChannelsBySCIDRangeRow
+	for rows.Next() {
+		var i GetChannelsBySCIDRangeRow
+		if err := rows.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,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getExtraNodeTypes = `-- name: GetExtraNodeTypes :many
 SELECT node_id, type, value
 FROM node_extra_types
@@ -1117,6 +1244,20 @@ func (q *Queries) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByL
 	return items, nil
 }
 
+const getPruneTip = `-- name: GetPruneTip :one
+SELECT block_height, block_hash
+FROM prune_log
+ORDER BY block_height DESC
+LIMIT 1
+`
+
+func (q *Queries) GetPruneTip(ctx context.Context) (PruneLog, error) {
+	row := q.db.QueryRowContext(ctx, getPruneTip)
+	var i PruneLog
+	err := row.Scan(&i.BlockHeight, &i.BlockHash)
+	return i, err
+}
+
 const getPublicV1ChannelsBySCID = `-- name: GetPublicV1ChannelsBySCID :many
 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
@@ -1219,6 +1360,51 @@ func (q *Queries) GetSourceNodesByVersion(ctx context.Context, version int16) ([
 	return items, nil
 }
 
+const getUnconnectedNodes = `-- name: GetUnconnectedNodes :many
+SELECT n.id, n.pub_key
+FROM nodes n
+WHERE NOT EXISTS (
+    SELECT 1
+    FROM channels c
+    WHERE c.node_id_1 = n.id OR c.node_id_2 = n.id
+)
+AND NOT EXISTS (
+    SELECT 1
+    FROM source_nodes sn
+    WHERE sn.node_id = n.id
+)
+`
+
+type GetUnconnectedNodesRow struct {
+	ID     int64
+	PubKey []byte
+}
+
+// Select all nodes that do not have any channels.
+// Ignore any of our source nodes.
+func (q *Queries) GetUnconnectedNodes(ctx context.Context) ([]GetUnconnectedNodesRow, error) {
+	rows, err := q.db.QueryContext(ctx, getUnconnectedNodes)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []GetUnconnectedNodesRow
+	for rows.Next() {
+		var i GetUnconnectedNodesRow
+		if err := rows.Scan(&i.ID, &i.PubKey); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getV1DisabledSCIDs = `-- name: GetV1DisabledSCIDs :many
 SELECT c.scid
 FROM channels c
@@ -1595,6 +1781,55 @@ func (q *Queries) ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNo
 	return items, nil
 }
 
+const listChannelsPaginated = `-- name: ListChannelsPaginated :many
+SELECT id, bitcoin_key_1, bitcoin_key_2, outpoint
+FROM channels c
+WHERE c.version = $1 AND c.id > $2
+ORDER BY c.id
+LIMIT $3
+`
+
+type ListChannelsPaginatedParams struct {
+	Version int16
+	ID      int64
+	Limit   int32
+}
+
+type ListChannelsPaginatedRow struct {
+	ID          int64
+	BitcoinKey1 []byte
+	BitcoinKey2 []byte
+	Outpoint    string
+}
+
+func (q *Queries) ListChannelsPaginated(ctx context.Context, arg ListChannelsPaginatedParams) ([]ListChannelsPaginatedRow, error) {
+	rows, err := q.db.QueryContext(ctx, listChannelsPaginated, arg.Version, arg.ID, arg.Limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []ListChannelsPaginatedRow
+	for rows.Next() {
+		var i ListChannelsPaginatedRow
+		if err := rows.Scan(
+			&i.ID,
+			&i.BitcoinKey1,
+			&i.BitcoinKey2,
+			&i.Outpoint,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const listChannelsWithPoliciesPaginated = `-- name: ListChannelsWithPoliciesPaginated :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,
@@ -1978,6 +2213,31 @@ func (q *Queries) UpsertNodeExtraType(ctx context.Context, arg UpsertNodeExtraTy
 	return err
 }
 
+const upsertPruneLogEntry = `-- name: UpsertPruneLogEntry :exec
+/* ─────────────────────────────────────────────
+    prune_log table queries
+    ─────────────────────────────────────────────
+*/
+
+INSERT INTO prune_log (
+    block_height, block_hash
+) VALUES (
+    $1, $2
+)
+ON CONFLICT(block_height) DO UPDATE SET
+    block_hash = EXCLUDED.block_hash
+`
+
+type UpsertPruneLogEntryParams struct {
+	BlockHeight int64
+	BlockHash   []byte
+}
+
+func (q *Queries) UpsertPruneLogEntry(ctx context.Context, arg UpsertPruneLogEntryParams) error {
+	_, err := q.db.ExecContext(ctx, upsertPruneLogEntry, arg.BlockHeight, arg.BlockHash)
+	return err
+}
+
 const upsertZombieChannel = `-- name: UpsertZombieChannel :exec
 /* ─────────────────────────────────────────────
    zombie_channels table queries

sqldb/sqlc/migrations/000007_graph.down.sql

@@ -27,4 +27,5 @@ DROP TABLE IF EXISTS node_features;
 DROP TABLE IF EXISTS node_extra_types;
 DROP TABLE IF EXISTS nodes;
 DROP TABLE IF EXISTS channel_policy_extra_types;
-DROP TABLE IF EXISTS zombie_channels;
+DROP TABLE IF EXISTS zombie_channels;
+DROP TABLE IF EXISTS prune_log;

sqldb/sqlc/migrations/000007_graph.up.sql

@@ -320,3 +320,15 @@ CREATE TABLE IF NOT EXISTS zombie_channels (
 );
 CREATE UNIQUE INDEX IF NOT EXISTS zombie_channels_channel_id_version_idx
     ON zombie_channels(scid, version);
+
+CREATE TABLE IF NOT EXISTS prune_log (
+    -- The block height that the prune was performed at.
+    -- NOTE: we don't use INTEGER PRIMARY KEY here since that would
+    -- get transformed into an auto-incrementing type by our SQL type
+    -- replacement logic. We don't want that since this must be the
+    -- actual block height and not an auto-incrementing value.
+    block_height BIGINT PRIMARY KEY,
+
+    -- The block hash that the prune was performed at.
+    block_hash BLOB NOT NULL
+);

sqldb/sqlc/models.go

@@ -181,6 +181,11 @@ type NodeFeature struct {
 	FeatureBit int32
 }
 
+type PruneLog struct {
+	BlockHeight int64
+	BlockHash   []byte
+}
+
 type SourceNode struct {
 	NodeID int64
 }

sqldb/sqlc/querier.go

@@ -21,9 +21,11 @@ type Querier interface {
 	DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error
 	DeleteExtraNodeType(ctx context.Context, arg DeleteExtraNodeTypeParams) error
 	DeleteInvoice(ctx context.Context, arg DeleteInvoiceParams) (sql.Result, error)
+	DeleteNode(ctx context.Context, id int64) error
 	DeleteNodeAddresses(ctx context.Context, nodeID int64) error
 	DeleteNodeByPubKey(ctx context.Context, arg DeleteNodeByPubKeyParams) (sql.Result, error)
 	DeleteNodeFeature(ctx context.Context, arg DeleteNodeFeatureParams) error
+	DeletePruneLogEntriesInRange(ctx context.Context, arg DeletePruneLogEntriesInRangeParams) error
 	DeleteZombieChannel(ctx context.Context, arg DeleteZombieChannelParams) (sql.Result, error)
 	FetchAMPSubInvoiceHTLCs(ctx context.Context, arg FetchAMPSubInvoiceHTLCsParams) ([]FetchAMPSubInvoiceHTLCsRow, error)
 	FetchAMPSubInvoices(ctx context.Context, arg FetchAMPSubInvoicesParams) ([]AmpSubInvoice, error)
@@ -31,6 +33,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)
+	GetChannelByOutpoint(ctx context.Context, outpoint string) (GetChannelByOutpointRow, 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)
@@ -38,6 +41,7 @@ type Querier interface {
 	GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetChannelPolicyByChannelAndNodeParams) (ChannelPolicy, error)
 	GetChannelPolicyExtraTypes(ctx context.Context, arg GetChannelPolicyExtraTypesParams) ([]GetChannelPolicyExtraTypesRow, error)
 	GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg GetChannelsByPolicyLastUpdateRangeParams) ([]GetChannelsByPolicyLastUpdateRangeRow, error)
+	GetChannelsBySCIDRange(ctx context.Context, arg GetChannelsBySCIDRangeParams) ([]GetChannelsBySCIDRangeRow, error)
 	GetDatabaseVersion(ctx context.Context) (int32, error)
 	GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]NodeExtraType, error)
 	// This method may return more than one invoice if filter using multiple fields
@@ -57,9 +61,13 @@ type Querier interface {
 	GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error)
 	GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error)
 	GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]Node, error)
+	GetPruneTip(ctx context.Context) (PruneLog, 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)
+	// Select all nodes that do not have any channels.
+	// Ignore any of our source nodes.
+	GetUnconnectedNodes(ctx context.Context) ([]GetUnconnectedNodesRow, 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
@@ -82,6 +90,7 @@ type Querier interface {
 	IsPublicV1Node(ctx context.Context, pubKey []byte) (bool, error)
 	IsZombieChannel(ctx context.Context, arg IsZombieChannelParams) (bool, error)
 	ListChannelsByNodeID(ctx context.Context, arg ListChannelsByNodeIDParams) ([]ListChannelsByNodeIDRow, error)
+	ListChannelsPaginated(ctx context.Context, arg ListChannelsPaginatedParams) ([]ListChannelsPaginatedRow, error)
 	ListChannelsWithPoliciesPaginated(ctx context.Context, arg ListChannelsWithPoliciesPaginatedParams) ([]ListChannelsWithPoliciesPaginatedRow, error)
 	ListNodeIDsAndPubKeys(ctx context.Context, arg ListNodeIDsAndPubKeysParams) ([]ListNodeIDsAndPubKeysRow, error)
 	ListNodesPaginated(ctx context.Context, arg ListNodesPaginatedParams) ([]Node, error)
@@ -104,6 +113,7 @@ type Querier interface {
 	UpsertEdgePolicy(ctx context.Context, arg UpsertEdgePolicyParams) (int64, error)
 	UpsertNode(ctx context.Context, arg UpsertNodeParams) (int64, error)
 	UpsertNodeExtraType(ctx context.Context, arg UpsertNodeExtraTypeParams) error
+	UpsertPruneLogEntry(ctx context.Context, arg UpsertPruneLogEntryParams) error
 	UpsertZombieChannel(ctx context.Context, arg UpsertZombieChannelParams) error
 }
 

sqldb/sqlc/queries/graph.sql

@@ -65,11 +65,31 @@ SELECT EXISTS (
       AND n.pub_key = $1
 );
 
+-- name: GetUnconnectedNodes :many
+SELECT n.id, n.pub_key
+FROM nodes n
+-- Select all nodes that do not have any channels.
+WHERE NOT EXISTS (
+    SELECT 1
+    FROM channels c
+    WHERE c.node_id_1 = n.id OR c.node_id_2 = n.id
+)
+-- Ignore any of our source nodes.
+AND NOT EXISTS (
+    SELECT 1
+    FROM source_nodes sn
+    WHERE sn.node_id = n.id
+);
+
 -- name: DeleteNodeByPubKey :execresult
 DELETE FROM nodes
 WHERE pub_key = $1
   AND version = $2;
 
+-- name: DeleteNode :exec
+DELETE FROM nodes
+WHERE id = $1;
+
 /* ─────────────────────────────────────────────
    node_features table queries
    ─────────────────────────────────────────────
@@ -188,10 +208,30 @@ INSERT INTO channels (
 )
 RETURNING id;
 
+-- name: GetChannelsBySCIDRange :many
+SELECT sqlc.embed(c),
+    n1.pub_key AS node1_pub_key,
+    n2.pub_key AS node2_pub_key
+FROM channels c
+    JOIN nodes n1 ON c.node_id_1 = n1.id
+    JOIN nodes n2 ON c.node_id_2 = n2.id
+WHERE scid >= @start_scid
+  AND scid < @end_scid;
+
 -- name: GetChannelBySCID :one
 SELECT * FROM channels
 WHERE scid = $1 AND version = $2;
 
+-- name: GetChannelByOutpoint :one
+SELECT
+    sqlc.embed(c),
+    n1.pub_key AS node1_pubkey,
+    n2.pub_key AS node2_pubkey
+FROM channels c
+    JOIN nodes n1 ON c.node_id_1 = n1.id
+    JOIN nodes n2 ON c.node_id_2 = n2.id
+WHERE c.outpoint = $1;
+
 -- name: GetChannelAndNodesBySCID :one
 SELECT
     c.*,
@@ -391,6 +431,13 @@ WHERE node_1_signature IS NOT NULL
   AND scid >= @start_scid
   AND scid < @end_scid;
 
+-- name: ListChannelsPaginated :many
+SELECT id, bitcoin_key_1, bitcoin_key_2, outpoint
+FROM channels c
+WHERE c.version = $1 AND c.id > $2
+ORDER BY c.id
+LIMIT $3;
+
 -- name: ListChannelsWithPoliciesPaginated :many
 SELECT
     sqlc.embed(c),
@@ -628,3 +675,28 @@ SELECT EXISTS (
     WHERE scid = $1
     AND version = $2
 ) AS is_zombie;
+
+/* ─────────────────────────────────────────────
+    prune_log table queries
+    ─────────────────────────────────────────────
+*/
+
+-- name: UpsertPruneLogEntry :exec
+INSERT INTO prune_log (
+    block_height, block_hash
+) VALUES (
+    $1, $2
+)
+ON CONFLICT(block_height) DO UPDATE SET
+    block_hash = EXCLUDED.block_hash;
+
+-- name: GetPruneTip :one
+SELECT block_height, block_hash
+FROM prune_log
+ORDER BY block_height DESC
+LIMIT 1;
+
+-- name: DeletePruneLogEntriesInRange :exec
+DELETE FROM prune_log
+WHERE block_height >= @start_height
+  AND block_height <= @end_height;