graph/db+sqldb: delete channels in batches

Use the new `SLICES` directive to add a DeleteChannels query which takes
a set of DB channel IDs. Then replace all our calls to DeleteChannel
with a paginated call to DeleteChannels.
This commit is contained in:
Elle Mouton committed 2025-07-22 17:17:23 +02:00
1 parent de6c030f29
commit ddc0e95eda
4 files changed
+69 -33

No files matched your search

+45 -21
View File
@@ -108,7 +108,7 @@ type SQLQueries interface {
GetChannelByOutpointWithPolicies(ctx context.Context, arg sqlc.GetChannelByOutpointWithPoliciesParams) (sqlc.GetChannelByOutpointWithPoliciesRow, error) GetChannelByOutpointWithPolicies(ctx context.Context, arg sqlc.GetChannelByOutpointWithPoliciesParams) (sqlc.GetChannelByOutpointWithPoliciesRow, error)
GetPublicV1ChannelsBySCID(ctx context.Context, arg sqlc.GetPublicV1ChannelsBySCIDParams) ([]sqlc.GraphChannel, error) GetPublicV1ChannelsBySCID(ctx context.Context, arg sqlc.GetPublicV1ChannelsBySCIDParams) ([]sqlc.GraphChannel, error)
GetSCIDByOutpoint(ctx context.Context, arg sqlc.GetSCIDByOutpointParams) ([]byte, error) GetSCIDByOutpoint(ctx context.Context, arg sqlc.GetSCIDByOutpointParams) ([]byte, error)
DeleteChannel(ctx context.Context, id int64) error DeleteChannels(ctx context.Context, ids []int64) error
CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error
InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error
@@ -1725,6 +1725,7 @@ func (s *SQLStore) DeleteChannelEdges(strictZombiePruning, markZombie bool,
deleted []*models.ChannelEdgeInfo deleted []*models.ChannelEdgeInfo
) )
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
chanIDsToDelete := make([]int64, 0, len(chanIDs))
chanCallBack := func(ctx context.Context, chanCallBack := func(ctx context.Context,
row sqlc.GetChannelsBySCIDWithPoliciesRow) error { row sqlc.GetChannelsBySCIDWithPoliciesRow) error {
@@ -1748,13 +1749,10 @@ func (s *SQLStore) DeleteChannelEdges(strictZombiePruning, markZombie bool,
return err return err
} }
err = db.DeleteChannel(ctx, row.GraphChannel.ID)
if err != nil {
return fmt.Errorf("unable to delete "+
"channel: %w", err)
}
deleted = append(deleted, info) deleted = append(deleted, info)
chanIDsToDelete = append(
chanIDsToDelete, row.GraphChannel.ID,
)
if !markZombie { if !markZombie {
return nil return nil
@@ -1809,7 +1807,7 @@ func (s *SQLStore) DeleteChannelEdges(strictZombiePruning, markZombie bool,
return ErrEdgeNotFound return ErrEdgeNotFound
} }
return nil return s.deleteChannels(ctx, db, chanIDsToDelete)
}, func() { }, func() {
deleted = nil deleted = nil
@@ -2462,6 +2460,8 @@ func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint,
prunedNodes []route.Vertex prunedNodes []route.Vertex
) )
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
var chansToDelete []int64
// Define the callback function for processing each channel. // Define the callback function for processing each channel.
channelCallback := func(ctx context.Context, channelCallback := func(ctx context.Context,
row sqlc.GetChannelsByOutpointsRow) error { row sqlc.GetChannelsByOutpointsRow) error {
@@ -2481,13 +2481,10 @@ func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint,
return err return err
} }
err = db.DeleteChannel(ctx, row.GraphChannel.ID)
if err != nil {
return fmt.Errorf("unable to delete "+
"channel: %w", err)
}
closedChans = append(closedChans, info) closedChans = append(closedChans, info)
chansToDelete = append(
chansToDelete, row.GraphChannel.ID,
)
return nil return nil
} }
@@ -2500,6 +2497,11 @@ func (s *SQLStore) PruneGraph(spentOutputs []*wire.OutPoint,
"outpoints: %w", err) "outpoints: %w", err)
} }
err = s.deleteChannels(ctx, db, chansToDelete)
if err != nil {
return fmt.Errorf("unable to delete channels: %w", err)
}
err = db.UpsertPruneLogEntry( err = db.UpsertPruneLogEntry(
ctx, sqlc.UpsertPruneLogEntryParams{ ctx, sqlc.UpsertPruneLogEntryParams{
BlockHash: blockHash[:], BlockHash: blockHash[:],
@@ -2565,6 +2567,27 @@ func (s *SQLStore) forEachChanInOutpoints(ctx context.Context, db SQLQueries,
) )
} }
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.ExecutePagedQuery(
ctx, s.cfg.PaginationCfg, dbIDs, idConverter,
queryWrapper, func(ctx context.Context, _ any) error {
return nil
},
)
}
// ChannelView returns the verifiable edge information for each active channel // ChannelView returns the verifiable edge information for each active channel
// within the known channel graph. The set of UTXOs (along with their scripts) // 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 // returned are the ones that need to be watched on chain to detect channel
@@ -2740,7 +2763,8 @@ func (s *SQLStore) DisconnectBlockAtHeight(height uint32) (
return fmt.Errorf("unable to fetch channels: %w", err) return fmt.Errorf("unable to fetch channels: %w", err)
} }
for _, row := range rows { chanIDsToDelete := make([]int64, len(rows))
for i, row := range rows {
node1, node2, err := buildNodeVertices( node1, node2, err := buildNodeVertices(
row.Node1PubKey, row.Node2PubKey, row.Node1PubKey, row.Node2PubKey,
) )
@@ -2756,15 +2780,15 @@ func (s *SQLStore) DisconnectBlockAtHeight(height uint32) (
return err return err
} }
err = db.DeleteChannel(ctx, row.GraphChannel.ID) chanIDsToDelete[i] = row.GraphChannel.ID
if err != nil {
return fmt.Errorf("unable to delete "+
"channel: %w", err)
}
removedChans = append(removedChans, channel) removedChans = append(removedChans, channel)
} }
err = s.deleteChannels(ctx, db, chanIDsToDelete)
if err != nil {
return fmt.Errorf("unable to delete channels: %w", err)
}
return db.DeletePruneLogEntriesInRange( return db.DeletePruneLogEntriesInRange(
ctx, sqlc.DeletePruneLogEntriesInRangeParams{ ctx, sqlc.DeletePruneLogEntriesInRangeParams{
StartHeight: int64(height), StartHeight: int64(height),
+20 -9
View File
@@ -143,15 +143,6 @@ func (q *Queries) CreateChannelExtraType(ctx context.Context, arg CreateChannelE
return err return err
} }
const deleteChannel = `-- name: DeleteChannel :exec
DELETE FROM graph_channels WHERE id = $1
`
func (q *Queries) DeleteChannel(ctx context.Context, id int64) error {
_, err := q.db.ExecContext(ctx, deleteChannel, id)
return err
}
const deleteChannelPolicyExtraTypes = `-- name: DeleteChannelPolicyExtraTypes :exec const deleteChannelPolicyExtraTypes = `-- name: DeleteChannelPolicyExtraTypes :exec
DELETE FROM graph_channel_policy_extra_types DELETE FROM graph_channel_policy_extra_types
WHERE channel_policy_id = $1 WHERE channel_policy_id = $1
@@ -162,6 +153,26 @@ func (q *Queries) DeleteChannelPolicyExtraTypes(ctx context.Context, channelPoli
return err return err
} }
const deleteChannels = `-- name: DeleteChannels :exec
DELETE FROM graph_channels
WHERE id IN (/*SLICE:ids*/?)
`
func (q *Queries) DeleteChannels(ctx context.Context, ids []int64) error {
query := deleteChannels
var queryParams []interface{}
if len(ids) > 0 {
for _, v := range ids {
queryParams = append(queryParams, v)
}
query = strings.Replace(query, "/*SLICE:ids*/?", makeQueryParams(len(queryParams), len(ids)), 1)
} else {
query = strings.Replace(query, "/*SLICE:ids*/?", "NULL", 1)
}
_, err := q.db.ExecContext(ctx, query, queryParams...)
return err
}
const deleteExtraNodeType = `-- name: DeleteExtraNodeType :exec const deleteExtraNodeType = `-- name: DeleteExtraNodeType :exec
DELETE FROM graph_node_extra_types DELETE FROM graph_node_extra_types
WHERE node_id = $1 WHERE node_id = $1
+1 -1
View File
@@ -18,8 +18,8 @@ type Querier interface {
CreateChannel(ctx context.Context, arg CreateChannelParams) (int64, error) CreateChannel(ctx context.Context, arg CreateChannelParams) (int64, error)
CreateChannelExtraType(ctx context.Context, arg CreateChannelExtraTypeParams) error CreateChannelExtraType(ctx context.Context, arg CreateChannelExtraTypeParams) error
DeleteCanceledInvoices(ctx context.Context) (sql.Result, error) DeleteCanceledInvoices(ctx context.Context) (sql.Result, error)
DeleteChannel(ctx context.Context, id int64) error
DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error
DeleteChannels(ctx context.Context, ids []int64) error
DeleteExtraNodeType(ctx context.Context, arg DeleteExtraNodeTypeParams) error DeleteExtraNodeType(ctx context.Context, arg DeleteExtraNodeTypeParams) error
DeleteInvoice(ctx context.Context, arg DeleteInvoiceParams) (sql.Result, error) DeleteInvoice(ctx context.Context, arg DeleteInvoiceParams) (sql.Result, error)
DeleteNode(ctx context.Context, id int64) error DeleteNode(ctx context.Context, id int64) error
+3 -2
View File
@@ -569,8 +569,9 @@ WHERE c.version = $1 AND c.id > $2
ORDER BY c.id ORDER BY c.id
LIMIT $3; LIMIT $3;
-- name: DeleteChannel :exec -- name: DeleteChannels :exec
DELETE FROM graph_channels WHERE id = $1; DELETE FROM graph_channels
WHERE id IN (sqlc.slice('ids')/*SLICE:ids*/);
/* ───────────────────────────────────────────── /* ─────────────────────────────────────────────
graph_channel_features table queries graph_channel_features table queries