Merge pull request #10038 from ellemouton/graphMig3-indexes

[graph mig 3]: graph/db: migrate zombies, closed SCIDs, prune log from kvdb to SQL
This commit is contained in:
Elle authored and GitHub committed 2025-07-11 10:52:34 +02:00
commit 6b326152d4
7 files changed
+597 -2

No files matched your search

@@ -106,6 +106,7 @@ circuit. The indices are only available for forwarding events saved after v0.20.
* Add graph SQL migration logic:
* [1](https://github.com/lightningnetwork/lnd/pull/10036)
* [2](https://github.com/lightningnetwork/lnd/pull/10050)
* [3](https://github.com/lightningnetwork/lnd/pull/10038)
## RPC Updates
* Previously the `RoutingPolicy` would return the inbound fee record in its
+325
View File
@@ -1,8 +1,10 @@
package graphdb
import (
"bytes"
"cmp"
"context"
"database/sql"
"errors"
"fmt"
"net"
@@ -12,6 +14,7 @@ import (
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/lightningnetwork/lnd/graph/db/models"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/sqldb"
"github.com/lightningnetwork/lnd/sqldb/sqlc"
)
@@ -55,6 +58,23 @@ func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend,
err)
}
// 4) Migrate the Prune log.
if err := migratePruneLog(ctx, kvBackend, sqlDB); err != nil {
return fmt.Errorf("could not migrate prune log: %w", err)
}
// 5) Migrate the closed SCID index.
err = migrateClosedSCIDIndex(ctx, kvBackend, sqlDB)
if err != nil {
return fmt.Errorf("could not migrate closed SCID index: %w",
err)
}
// 6) Migrate the zombie index.
if err := migrateZombieIndex(ctx, kvBackend, sqlDB); err != nil {
return fmt.Errorf("could not migrate zombie index: %w", err)
}
log.Infof("Finished migration of the graph store from KV to SQL in %v",
time.Since(t0))
@@ -506,6 +526,115 @@ func migrateSingleChannel(ctx context.Context, sqlDB SQLQueries,
return nil
}
// migratePruneLog migrates the prune log from the KV backend to the SQL
// database. It iterates over each prune log entry in the KV store, inserts it
// into the SQL database, and then verifies that the entry was inserted
// correctly by fetching it back from the SQL database and comparing it to the
// original entry.
func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend,
sqlDB SQLQueries) error {
var (
count uint64
pruneTipHeight uint32
pruneTipHash chainhash.Hash
)
// migrateSinglePruneEntry is a helper function that inserts a single
// prune log entry into the SQL database and verifies that it was
// inserted correctly.
migrateSinglePruneEntry := func(height uint32,
hash *chainhash.Hash) error {
count++
// Keep track of the prune tip height and hash.
if height > pruneTipHeight {
pruneTipHeight = height
pruneTipHash = *hash
}
err := sqlDB.UpsertPruneLogEntry(
ctx, sqlc.UpsertPruneLogEntryParams{
BlockHeight: int64(height),
BlockHash: hash[:],
},
)
if err != nil {
return fmt.Errorf("unable to insert prune log "+
"entry for height %d: %w", height, err)
}
// Now, check that the entry was inserted correctly.
migratedHash, err := sqlDB.GetPruneHashByHeight(
ctx, int64(height),
)
if err != nil {
return fmt.Errorf("could not get prune hash "+
"for height %d: %w", height, err)
}
return sqldb.CompareRecords(
hash[:], migratedHash, "prune log entry",
)
}
// Iterate over each prune log entry in the KV store and migrate it to
// the SQL database.
err := forEachPruneLogEntry(
kvBackend, func(height uint32, hash *chainhash.Hash) error {
err := migrateSinglePruneEntry(height, hash)
if err != nil {
return fmt.Errorf("could not migrate "+
"prune log entry at height %d: %w",
height, err)
}
return nil
},
)
if err != nil {
return fmt.Errorf("could not migrate prune log: %w", err)
}
// Check that the prune tip is set correctly in the SQL
// database.
pruneTip, err := sqlDB.GetPruneTip(ctx)
if errors.Is(err, sql.ErrNoRows) {
// The ErrGraphNeverPruned error is expected if no prune log
// entries were migrated from the kvdb store. Otherwise, it's
// an unexpected error.
if count == 0 {
log.Infof("No prune log entries found in KV store " +
"to migrate")
return nil
}
// Fall-through to the next error check.
}
if err != nil {
return fmt.Errorf("could not get prune tip: %w", err)
}
if pruneTip.BlockHeight != int64(pruneTipHeight) ||
!bytes.Equal(pruneTip.BlockHash, pruneTipHash[:]) {
return fmt.Errorf("prune tip mismatch after migration: "+
"expected height %d, hash %s; got height %d, "+
"hash %s", pruneTipHeight, pruneTipHash,
pruneTip.BlockHeight,
chainhash.Hash(pruneTip.BlockHash))
}
log.Infof("Migrated %d prune log entries from KV to SQL. The prune "+
"tip is: height %d, hash: %s", count, pruneTipHeight,
pruneTipHash)
return nil
}
// getAndBuildChanAndPolicies is a helper that builds the channel edge info
// and policies from the given row returned by the SQL query
// GetChannelBySCIDWithPolicies.
func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries,
row sqlc.GetChannelBySCIDWithPoliciesRow,
chain chainhash.Hash) (*models.ChannelEdgeInfo,
@@ -542,3 +671,199 @@ func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries,
return edge, policy1, policy2, nil
}
// forEachPruneLogEntry iterates over each prune log entry in the KV
// backend and calls the provided callback function for each entry.
func forEachPruneLogEntry(db kvdb.Backend, cb func(height uint32,
hash *chainhash.Hash) error) error {
return kvdb.View(db, func(tx kvdb.RTx) error {
metaBucket := tx.ReadBucket(graphMetaBucket)
if metaBucket == nil {
return ErrGraphNotFound
}
pruneBucket := metaBucket.NestedReadBucket(pruneLogBucket)
if pruneBucket == nil {
// The graph has never been pruned and so, there are no
// entries to iterate over.
return nil
}
return pruneBucket.ForEach(func(k, v []byte) error {
blockHeight := byteOrder.Uint32(k)
var blockHash chainhash.Hash
copy(blockHash[:], v)
return cb(blockHeight, &blockHash)
})
}, func() {})
}
// migrateClosedSCIDIndex migrates the closed SCID index from the KV backend to
// the SQL database. It iterates over each closed SCID in the KV store, inserts
// it into the SQL database, and then verifies that the SCID was inserted
// correctly by checking if the channel with the given SCID is seen as closed in
// the SQL database.
func migrateClosedSCIDIndex(ctx context.Context, kvBackend kvdb.Backend,
sqlDB SQLQueries) error {
var count uint64
migrateSingleClosedSCID := func(scid lnwire.ShortChannelID) error {
count++
chanIDB := channelIDToBytes(scid.ToUint64())
err := sqlDB.InsertClosedChannel(ctx, chanIDB)
if err != nil {
return fmt.Errorf("could not insert closed channel "+
"with SCID %s: %w", scid, err)
}
// Now, verify that the channel with the given SCID is
// seen as closed.
isClosed, err := sqlDB.IsClosedChannel(ctx, chanIDB)
if err != nil {
return fmt.Errorf("could not check if channel %s "+
"is closed: %w", scid, err)
}
if !isClosed {
return fmt.Errorf("channel %s should be closed, "+
"but is not", scid)
}
return nil
}
err := forEachClosedSCID(kvBackend, migrateSingleClosedSCID)
if err != nil {
return fmt.Errorf("could not migrate closed SCID index: %w",
err)
}
log.Infof("Migrated %d closed SCIDs from KV to SQL", count)
return nil
}
// migrateZombieIndex migrates the zombie index from the KV backend to
// the SQL database. It iterates over each zombie channel in the KV store,
// inserts it into the SQL database, and then verifies that the channel is
// indeed marked as a zombie channel in the SQL database.
//
// NOTE: before inserting an entry into the zombie index, the function checks
// if the channel is already marked as closed in the SQL store. If it is,
// the entry is skipped. This means that the resulting zombie index count in
// the SQL store may well be less than the count of zombie channels in the KV
// store.
func migrateZombieIndex(ctx context.Context, kvBackend kvdb.Backend,
sqlDB SQLQueries) error {
var count uint64
err := forEachZombieEntry(kvBackend, func(chanID uint64, pubKey1,
pubKey2 [33]byte) error {
chanIDB := channelIDToBytes(chanID)
// If it is in the closed SCID index, we don't need to
// add it to the zombie index.
//
// NOTE: this means that the resulting zombie index count in
// the SQL store may well be less than the count of zombie
// channels in the KV store.
isClosed, err := sqlDB.IsClosedChannel(ctx, chanIDB)
if err != nil {
return fmt.Errorf("could not check closed "+
"channel: %w", err)
}
if isClosed {
return nil
}
count++
err = sqlDB.UpsertZombieChannel(
ctx, sqlc.UpsertZombieChannelParams{
Version: int16(ProtocolV1),
Scid: chanIDB,
NodeKey1: pubKey1[:],
NodeKey2: pubKey2[:],
},
)
if err != nil {
return fmt.Errorf("could not upsert zombie "+
"channel %d: %w", chanID, err)
}
// Finally, verify that the channel is indeed marked as a
// zombie channel.
isZombie, err := sqlDB.IsZombieChannel(
ctx, sqlc.IsZombieChannelParams{
Version: int16(ProtocolV1),
Scid: chanIDB,
},
)
if err != nil {
return fmt.Errorf("could not check if "+
"channel %d is zombie: %w", chanID, err)
}
if !isZombie {
return fmt.Errorf("channel %d should be "+
"a zombie, but is not", chanID)
}
return nil
})
if err != nil {
return fmt.Errorf("could not migrate zombie index: %w", err)
}
log.Infof("Migrated %d zombie channels from KV to SQL", count)
return nil
}
// forEachZombieEntry iterates over each zombie channel entry in the
// KV backend and calls the provided callback function for each entry.
func forEachZombieEntry(db kvdb.Backend, cb func(chanID uint64, pubKey1,
pubKey2 [33]byte) error) error {
return kvdb.View(db, func(tx kvdb.RTx) error {
edges := tx.ReadBucket(edgeBucket)
if edges == nil {
return ErrGraphNoEdgesFound
}
zombieIndex := edges.NestedReadBucket(zombieBucket)
if zombieIndex == nil {
return nil
}
return zombieIndex.ForEach(func(k, v []byte) error {
var pubKey1, pubKey2 [33]byte
copy(pubKey1[:], v[:33])
copy(pubKey2[:], v[33:])
return cb(byteOrder.Uint64(k), pubKey1, pubKey2)
})
}, func() {})
}
// forEachClosedSCID iterates over each closed SCID in the KV backend and calls
// the provided callback function for each SCID.
func forEachClosedSCID(db kvdb.Backend,
cb func(lnwire.ShortChannelID) error) error {
return kvdb.View(db, func(tx kvdb.RTx) error {
closedScids := tx.ReadBucket(closedScidBucket)
if closedScids == nil {
return nil
}
return closedScids.ForEach(func(k, _ []byte) error {
return cb(lnwire.NewShortChanIDFromInt(
byteOrder.Uint64(k),
))
})
}, func() {})
}
+251 -2
View File
@@ -6,6 +6,7 @@ import (
"bytes"
"cmp"
"context"
"crypto/rand"
"errors"
"fmt"
"image/color"
@@ -21,6 +22,7 @@ import (
"github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btclog/v2"
"github.com/lightningnetwork/lnd/graph/db/models"
@@ -81,6 +83,12 @@ func TestMigrateGraphToSQL(t *testing.T) {
node2 = genPubKey(t)
)
type zombieIndexObject struct {
scid uint64
pubKey1 route.Vertex
pubKey2 route.Vertex
}
tests := []struct {
name string
write func(t *testing.T, db *KVStore, object any)
@@ -217,6 +225,90 @@ func TestMigrateGraphToSQL(t *testing.T) {
numPolicies: 3,
},
},
{
name: "prune log",
write: func(t *testing.T, db *KVStore, object any) {
var hash chainhash.Hash
_, err := rand.Read(hash[:])
require.NoError(t, err)
switch obj := object.(type) {
case *models.LightningNode:
err = db.SetSourceNode(ctx, obj)
default:
height, ok := obj.(uint32)
require.True(t, ok)
_, _, err = db.PruneGraph(
nil, &hash, height,
)
}
require.NoError(t, err)
},
objects: []any{
// The PruneGraph call requires that the source
// node be set. So that is the first object
// we will write.
&models.LightningNode{
HaveNodeAnnouncement: false,
PubKeyBytes: testPub,
},
// Now we add some block heights to prune
// the graph at.
uint32(1), uint32(2), uint32(20), uint32(3),
uint32(4),
},
expGraphStats: graphStats{
numNodes: 1,
srcNodeSet: true,
pruneTip: 20,
},
},
{
name: "closed SCID index",
write: func(t *testing.T, db *KVStore, object any) {
scid, ok := object.(lnwire.ShortChannelID)
require.True(t, ok)
err := db.PutClosedScid(scid)
require.NoError(t, err)
},
objects: []any{
lnwire.NewShortChanIDFromInt(1),
lnwire.NewShortChanIDFromInt(2),
lnwire.NewShortChanIDFromInt(3),
lnwire.NewShortChanIDFromInt(4),
},
},
{
name: "zombie index",
write: func(t *testing.T, db *KVStore, object any) {
obj, ok := object.(*zombieIndexObject)
require.True(t, ok)
err := db.MarkEdgeZombie(
obj.scid, obj.pubKey1, obj.pubKey2,
)
require.NoError(t, err)
},
objects: []any{
&zombieIndexObject{
scid: prand.Uint64(),
pubKey1: genPubKey(t),
pubKey2: genPubKey(t),
},
&zombieIndexObject{
scid: prand.Uint64(),
pubKey1: genPubKey(t),
pubKey2: genPubKey(t),
},
&zombieIndexObject{
scid: prand.Uint64(),
pubKey1: genPubKey(t),
pubKey2: genPubKey(t),
},
},
},
}
for _, test := range tests {
@@ -253,6 +345,7 @@ type graphStats struct {
srcNodeSet bool
numChannels int
numPolicies int
pruneTip int
}
// assertInSync checks that the KVStore and SQLStore both contain the same
@@ -276,6 +369,20 @@ func assertInSync(t *testing.T, kvDB *KVStore, sqlDB *SQLStore,
require.Len(t, sqlChannels, stats.numChannels)
require.Equal(t, stats.numPolicies, sqlChannels.CountPolicies())
require.Equal(t, fetchAllChannelsAndPolicies(t, kvDB), sqlChannels)
// 4) Assert prune logs match. For this one, we iterate through the
// prune log of the kvdb store and check that the entries match the
// entries in the SQL store. Then we just do a final check to ensure
// that the prune tip also matches.
checkKVPruneLogEntries(t, kvDB, sqlDB, stats.pruneTip)
// 5) Assert that the closed SCID index is also in sync. Like the prune
// log we iterate through the kvdb store and check that the entries
// match the entries in the SQL store.
checkClosedSCIDIndex(t, kvDB.db, sqlDB)
// 6) Finally, check that the zombie index is also in sync.
checkZombieIndex(t, kvDB.db, sqlDB)
}
// fetchAllNodes retrieves all nodes from the given store and returns them
@@ -379,6 +486,91 @@ func fetchAllChannelsAndPolicies(t *testing.T, store V1Store) chanSet {
return channels
}
// checkKVPruneLogEntries iterates through the prune log entries in the
// KVStore and checks that there is an entry for each in the SQLStore. It then
// does a final check to ensure that the prune tips in both stores match.
func checkKVPruneLogEntries(t *testing.T, kv *KVStore, sql *SQLStore,
expTip int) {
// Iterate through the prune log entries in the KVStore and
// check that each entry exists in the SQLStore.
err := forEachPruneLogEntry(
kv.db, func(height uint32, hash *chainhash.Hash) error {
sqlHash, err := sql.db.GetPruneHashByHeight(
context.Background(), int64(height),
)
require.NoError(t, err)
require.Equal(t, hash[:], sqlHash)
return nil
},
)
require.NoError(t, err)
kvPruneHash, kvPruneHeight, kvPruneErr := kv.PruneTip()
sqlPruneHash, sqlPruneHeight, sqlPruneErr := sql.PruneTip()
// If the prune error is ErrGraphNeverPruned, then we expect
// the SQL prune error to also be ErrGraphNeverPruned.
if errors.Is(kvPruneErr, ErrGraphNeverPruned) {
require.ErrorIs(t, sqlPruneErr, ErrGraphNeverPruned)
return
}
// Otherwise, we expect both prune errors to be nil and the
// prune hashes and heights to match.
require.NoError(t, kvPruneErr)
require.NoError(t, sqlPruneErr)
require.Equal(t, kvPruneHash[:], sqlPruneHash[:])
require.Equal(t, kvPruneHeight, sqlPruneHeight)
require.Equal(t, expTip, int(sqlPruneHeight))
}
// checkClosedSCIDIndex iterates through the closed SCID index in the
// KVStore and checks that each SCID is marked as closed in the SQLStore.
func checkClosedSCIDIndex(t *testing.T, kv kvdb.Backend, sql *SQLStore) {
err := forEachClosedSCID(kv, func(scid lnwire.ShortChannelID) error {
closed, err := sql.IsClosedScid(scid)
require.NoError(t, err)
require.True(t, closed)
return nil
})
require.NoError(t, err)
}
// checkZombieIndex iterates through the zombie index in the
// KVStore and checks that each SCID is marked as a zombie in the SQLStore.
func checkZombieIndex(t *testing.T, kv kvdb.Backend, sql *SQLStore) {
err := forEachZombieEntry(kv, func(chanID uint64, pubKey1,
pubKey2 [33]byte) error {
scid := lnwire.NewShortChanIDFromInt(chanID)
// The migration logic skips zombie entries if they are already
// present in the closed SCID index in the SQL DB. We need to
// replicate that check here.
isClosed, err := sql.IsClosedScid(scid)
require.NoError(t, err)
isZombie, _, _, err := sql.IsZombieEdge(chanID)
require.NoError(t, err)
if isClosed {
// If it's in the closed index, it should NOT be in the
// zombie index.
require.False(t, isZombie)
} else {
// If it's not in the closed index, it SHOULD be in the
// zombie index.
require.True(t, isZombie)
}
return nil
})
require.NoError(t, err)
}
// setUpKVStore initializes a new KVStore for testing.
func setUpKVStore(t *testing.T) *KVStore {
kvDB, cleanup, err := kvdb.GetTestBackend(t.TempDir(), "graph")
@@ -867,6 +1059,39 @@ func TestSQLMigrationEdgeCases(t *testing.T) {
}},
})
})
// This test covers the case where the KV store contains zombie entries
// that it also has entries for in the closed SCID index. In this case,
// the SQL store will only insert zombie entries for channels that
// are not yet closed.
t.Run("zombies and closed scids", func(t *testing.T) {
var (
n1, n2 route.Vertex
cID1 = uint64(1)
cID2 = uint64(2)
)
populateKV := func(t *testing.T, db *KVStore) {
// Mark both channels as zombies.
err := db.MarkEdgeZombie(cID1, n1, n2)
require.NoError(t, err)
err = db.MarkEdgeZombie(cID2, n1, n2)
require.NoError(t, err)
// Mark channel 1 as closed.
err = db.PutClosedScid(
lnwire.NewShortChanIDFromInt(cID1),
)
require.NoError(t, err)
}
runTestMigration(t, populateKV, dbState{
chans: make(chanSet, 0),
closed: []uint64{1},
zombies: []uint64{2},
})
})
}
// runTestMigration is a helper function that sets up the KVStore and SQLStore,
@@ -898,8 +1123,10 @@ func runTestMigration(t *testing.T, populateKV func(t *testing.T, db *KVStore),
// dbState describes the expected state of the SQLStore after a migration.
type dbState struct {
nodes []*models.LightningNode
chans chanSet
nodes []*models.LightningNode
chans chanSet
closed []uint64
zombies []uint64
}
// assertResultState asserts that the SQLStore contains the expected
@@ -910,4 +1137,26 @@ func assertResultState(t *testing.T, sql *SQLStore, expState dbState) {
require.ElementsMatch(
t, expState.chans, fetchAllChannelsAndPolicies(t, sql),
)
for _, closed := range expState.closed {
isClosed, err := sql.IsClosedScid(
lnwire.NewShortChanIDFromInt(closed),
)
require.NoError(t, err)
require.True(t, isClosed)
// Any closed SCID should NOT be in the zombie
// index.
isZombie, _, _, err := sql.IsZombieEdge(closed)
require.NoError(t, err)
require.False(t, isZombie)
}
for _, zombie := range expState.zombies {
isZombie, _, _, err := sql.IsZombieEdge(
zombie,
)
require.NoError(t, err)
require.True(t, isZombie)
}
}
+1
View File
@@ -135,6 +135,7 @@ type SQLQueries interface {
Prune log table queries.
*/
GetPruneTip(ctx context.Context) (sqlc.PruneLog, error)
GetPruneHashByHeight(ctx context.Context, blockHeight int64) ([]byte, error)
UpsertPruneLogEntry(ctx context.Context, arg sqlc.UpsertPruneLogEntryParams) error
DeletePruneLogEntriesInRange(ctx context.Context, arg sqlc.DeletePruneLogEntriesInRangeParams) error
+13
View File
@@ -1350,6 +1350,19 @@ func (q *Queries) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByL
return items, nil
}
const getPruneHashByHeight = `-- name: GetPruneHashByHeight :one
SELECT block_hash
FROM prune_log
WHERE block_height = $1
`
func (q *Queries) GetPruneHashByHeight(ctx context.Context, blockHeight int64) ([]byte, error) {
row := q.db.QueryRowContext(ctx, getPruneHashByHeight, blockHeight)
var block_hash []byte
err := row.Scan(&block_hash)
return block_hash, err
}
const getPruneTip = `-- name: GetPruneTip :one
SELECT block_height, block_hash
FROM prune_log
+1
View File
@@ -63,6 +63,7 @@ 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)
GetPruneHashByHeight(ctx context.Context, blockHeight int64) ([]byte, error)
GetPruneTip(ctx context.Context) (PruneLog, error)
GetPublicV1ChannelsBySCID(ctx context.Context, arg GetPublicV1ChannelsBySCIDParams) ([]Channel, error)
GetSCIDByOutpoint(ctx context.Context, arg GetSCIDByOutpointParams) ([]byte, error)
+5
View File
@@ -728,6 +728,11 @@ FROM prune_log
ORDER BY block_height DESC
LIMIT 1;
-- name: GetPruneHashByHeight :one
SELECT block_hash
FROM prune_log
WHERE block_height = $1;
-- name: DeletePruneLogEntriesInRange :exec
DELETE FROM prune_log
WHERE block_height >= @start_height