16139765508

Committed 08 Jul 2025 09:44AM UTC coverage: 67.46%. First build

Build # 16139765508

Build Type

Pull #10050

github

Committed by

web-flow

Commit Message

Merge 3867d135f into cb959bddb

Pull Request Pull Request #10050: [graph mig 2]: graph/db: migrate graph channels and policies from kvdb to SQL

Run Details

13 of 235 new or added lines in 2 files covered. (5.53%)

135200 of 200414 relevant lines covered (67.46%)

21771.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/graph/db/sql_migration.go

package graphdb

import (
        "cmp"
        "context"
        "errors"
        "fmt"
        "net"
        "slices"
        "time"

        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/sqldb"
        "github.com/lightningnetwork/lnd/sqldb/sqlc"
)

// MigrateGraphToSQL migrates the graph store from a KV backend to a SQL
// backend.
//
// NOTE: this is currently not called from any code path. It is called via tests
// only for now and will be called from the main lnd binary once the
// migration is fully implemented and tested.
func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend,
        sqlDB SQLQueries, chain chainhash.Hash) error {

        log.Infof("Starting migration of the graph store from KV to SQL")
        t0 := time.Now()

        // Check if there is a graph to migrate.
        graphExists, err := checkGraphExists(kvBackend)
        if err != nil {
                return fmt.Errorf("failed to check graph existence: %w", err)
        }
        if !graphExists {
                log.Infof("No graph found in KV store, skipping the migration")
                return nil
        }

        // 1) Migrate all the nodes.
        if err := migrateNodes(ctx, kvBackend, sqlDB); err != nil {
                return fmt.Errorf("could not migrate nodes: %w", err)
        }

        // 2) Migrate the source node.
        if err := migrateSourceNode(ctx, kvBackend, sqlDB); err != nil {
                return fmt.Errorf("could not migrate source node: %w", err)
        }

        // 3) Migrate all the channels and channel policies.
        err = migrateChannelsAndPolicies(ctx, kvBackend, sqlDB, chain)
        if err != nil {
                return fmt.Errorf("could not migrate channels and policies: %w",
                        err)
        }

        log.Infof("Finished migration of the graph store from KV to SQL in %v",
                time.Since(t0))

        return nil
}

// checkGraphExists checks if the graph exists in the KV backend.
func checkGraphExists(db kvdb.Backend) (bool, error) {
        // Check if there is even a graph to migrate.
        err := db.View(func(tx kvdb.RTx) error {
                // Check for the existence of the node bucket which is a top
                // level bucket that would have been created on the initial
                // creation of the graph store.
                nodes := tx.ReadBucket(nodeBucket)
                if nodes == nil {
                        return ErrGraphNotFound
                }

                return nil
        }, func() {})
        if errors.Is(err, ErrGraphNotFound) {
                return false, nil
        } else if err != nil {
                return false, err
        }

        return true, nil
}

// migrateNodes migrates all nodes from the KV backend to the SQL database.
// This includes doing a sanity check after each migration to ensure that the
// migrated node matches the original node.
func migrateNodes(ctx context.Context, kvBackend kvdb.Backend,
        sqlDB SQLQueries) error {

        // Keep track of the number of nodes migrated and the number of
        // nodes skipped due to errors.
        var (
                count   uint64
                skipped uint64
        )

        // Loop through each node in the KV store and insert it into the SQL
        // database.
        err := forEachNode(kvBackend, func(_ kvdb.RTx,
                node *models.LightningNode) error {

                pub := node.PubKeyBytes

                // Sanity check to ensure that the node has valid extra opaque
                // data. If it does not, we'll skip it. We need to do this
                // because previously we would just persist any TLV bytes that
                // we received without validating them. Now, however, we
                // normalise the storage of extra opaque data, so we need to
                // ensure that the data is valid. We don't want to abort the
                // migration if we encounter a node with invalid extra opaque
                // data, so we'll just skip it and log a warning.
                _, err := marshalExtraOpaqueData(node.ExtraOpaqueData)
                if errors.Is(err, ErrParsingExtraTLVBytes) {
                        skipped++
                        log.Warnf("Skipping migration of node %x with invalid "+
                                "extra opaque data: %v", pub,
                                node.ExtraOpaqueData)

                        return nil
                } else if err != nil {
                        return fmt.Errorf("unable to marshal extra "+
                                "opaque data for node %x: %w", pub, err)
                }

                count++

                // TODO(elle): At this point, we should check the loaded node
                // to see if we should extract any DNS addresses from its
                // opaque type addresses. This is expected to be done in:
                // https://github.com/lightningnetwork/lnd/pull/9455.
                // This TODO is being tracked in
                //  https://github.com/lightningnetwork/lnd/issues/9795 as this
                // must be addressed before making this code path active in
                // production.

                // Write the node to the SQL database.
                id, err := upsertNode(ctx, sqlDB, node)
                if err != nil {
                        return fmt.Errorf("could not persist node(%x): %w", pub,
                                err)
                }

                // Fetch it from the SQL store and compare it against the
                // original node object to ensure the migration was successful.
                dbNode, err := sqlDB.GetNodeByPubKey(
                        ctx, sqlc.GetNodeByPubKeyParams{
                                PubKey:  node.PubKeyBytes[:],
                                Version: int16(ProtocolV1),
                        },
                )
                if err != nil {
                        return fmt.Errorf("could not get node by pubkey (%x)"+
                                "after migration: %w", pub, err)
                }

                // Sanity check: ensure the migrated node ID matches the one we
                // just inserted.
                if dbNode.ID != id {
                        return fmt.Errorf("node ID mismatch for node (%x) "+
                                "after migration: expected %d, got %d",
                                pub, id, dbNode.ID)
                }

                migratedNode, err := buildNode(ctx, sqlDB, &dbNode)
                if err != nil {
                        return fmt.Errorf("could not build migrated node "+
                                "from dbNode(db id: %d, node pub: %x): %w",
                                dbNode.ID, pub, err)
                }

                // Make sure that the node addresses are sorted before
                // comparing them to ensure that the order of addresses does
                // not affect the comparison.
                slices.SortFunc(node.Addresses, func(i, j net.Addr) int {
                        return cmp.Compare(i.String(), j.String())
                })
                slices.SortFunc(
                        migratedNode.Addresses, func(i, j net.Addr) int {
                                return cmp.Compare(i.String(), j.String())
                        },
                )

                return sqldb.CompareRecords(
                        node, migratedNode, fmt.Sprintf("node %x", pub),
                )
        })
        if err != nil {
                return fmt.Errorf("could not migrate nodes: %w", err)
        }

        log.Infof("Migrated %d nodes from KV to SQL (skipped %d nodes due to "+
                "invalid TLV streams)", count, skipped)

        return nil
}

// migrateSourceNode migrates the source node from the KV backend to the
// SQL database.
func migrateSourceNode(ctx context.Context, kvdb kvdb.Backend,
        sqlDB SQLQueries) error {

        sourceNode, err := sourceNode(kvdb)
        if errors.Is(err, ErrSourceNodeNotSet) {
                // If the source node has not been set yet, we can skip this
                // migration step.
                return nil
        } else if err != nil {
                return fmt.Errorf("could not get source node from kv "+
                        "store: %w", err)
        }

        pub := sourceNode.PubKeyBytes

        // Get the DB ID of the source node by its public key. This node must
        // already exist in the SQL database, as it should have been migrated
        // in the previous node-migration step.
        id, err := sqlDB.GetNodeIDByPubKey(
                ctx, sqlc.GetNodeIDByPubKeyParams{
                        PubKey:  pub[:],
                        Version: int16(ProtocolV1),
                },
        )
        if err != nil {
                return fmt.Errorf("could not get source node ID: %w", err)
        }

        // Now we can add the source node to the SQL database.
        err = sqlDB.AddSourceNode(ctx, id)
        if err != nil {
                return fmt.Errorf("could not add source node to SQL store: %w",
                        err)
        }

        // Verify that the source node was added correctly by fetching it back
        // from the SQL database and checking that the expected DB ID and
        // pub key are returned. We don't need to do a whole node comparison
        // here, as this was already done in the previous migration step.
        srcNodes, err := sqlDB.GetSourceNodesByVersion(ctx, int16(ProtocolV1))
        if err != nil {
                return fmt.Errorf("could not get source nodes from SQL "+
                        "store: %w", err)
        }

        // The SQL store has support for multiple source nodes (for future
        // protocol versions) but this migration is purely aimed at the V1
        // store, and so we expect exactly one source node to be present.
        if len(srcNodes) != 1 {
                return fmt.Errorf("expected exactly one source node, "+
                        "got %d", len(srcNodes))
        }

        // Check that the source node ID and pub key match the original
        // source node.
        if srcNodes[0].NodeID != id {
                return fmt.Errorf("source node ID mismatch after migration: "+
                        "expected %d, got %d", id, srcNodes[0].NodeID)
        }
        err = sqldb.CompareRecords(pub[:], srcNodes[0].PubKey, "source node")
        if err != nil {
                return fmt.Errorf("source node pubkey mismatch after "+
                        "migration: %w", err)
        }

        log.Infof("Migrated source node with pubkey %x to SQL", pub[:])

        return nil
}

// migrateChannelsAndPolicies migrates all channels and their policies
// from the KV backend to the SQL database.
func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend,
        sqlDB SQLQueries, chain chainhash.Hash) error {

        var (
                channelCount       uint64
                skippedChanCount   uint64
                policyCount        uint64
                skippedPolicyCount uint64
        )
        migChanPolicy := func(policy *models.ChannelEdgePolicy) error {
                // If the policy is nil, we can skip it.
                if policy == nil {
                        return nil
                }

                // Sanity check to ensure that the policy has valid extra opaque
                // data. If it does not, we'll skip it. We need to do this
                // because previously we would just persist any TLV bytes that
                // we received without validating them. Now, however, we
                // normalise the storage of extra opaque data, so we need to
                // ensure that the data is valid. We don't want to abort the
                // migration if we encounter a policy with invalid extra opaque
                // data, so we'll just skip it and log a warning.
                _, err := marshalExtraOpaqueData(policy.ExtraOpaqueData)
                if errors.Is(err, ErrParsingExtraTLVBytes) {
                        skippedPolicyCount++
                        log.Warnf("Skipping policy for channel %d with "+
                                "invalid extra opaque data: %v",
                                policy.ChannelID, policy.ExtraOpaqueData)

                        return nil
                } else if err != nil {
                        return fmt.Errorf("unable to marshal extra opaque "+
                                "data (%v): %w", policy.ExtraOpaqueData, err)
                }

                policyCount++

                _, _, _, err = updateChanEdgePolicy(ctx, sqlDB, policy)
                if err != nil {
                        return fmt.Errorf("could not migrate channel "+
                                "policy %d: %w", policy.ChannelID, err)
                }

                return nil
        }

        // Iterate over each channel in the KV store and migrate it and its
        // policies to the SQL database.
        err := forEachChannel(kvBackend, func(channel *models.ChannelEdgeInfo,
                policy1 *models.ChannelEdgePolicy,
                policy2 *models.ChannelEdgePolicy) error {

                scid := channel.ChannelID

                // Here, we do a sanity check to ensure that the chain hash of
                // the channel returned by the KV store matches the expected
                // chain hash. This is important since in the SQL store, we will
                // no longer explicitly store the chain hash in the channel
                // info, but rather rely on the chain hash LND is running with.
                // So this is our way of ensuring that LND is running on the
                // correct network at migration time.
                if channel.ChainHash != chain {
                        return fmt.Errorf("channel %d has chain hash %s, "+
                                "expected %s", scid, channel.ChainHash, chain)
                }

                // Sanity check to ensure that the channel has valid extra
                // opaque data. If it does not, we'll skip it. We need to do
                // this because previously we would just persist any TLV bytes
                // that we received without validating them. Now, however, we
                // normalise the storage of extra opaque data, so we need to
                // ensure that the data is valid. We don't want to abort the
                // migration if we encounter a channel with invalid extra opaque
                // data, so we'll just skip it and log a warning.
                _, err := marshalExtraOpaqueData(channel.ExtraOpaqueData)
                if errors.Is(err, ErrParsingExtraTLVBytes) {
                        log.Warnf("Skipping channel %d with invalid "+
                                "extra opaque data: %v", scid,
                                channel.ExtraOpaqueData)

                        skippedChanCount++

                        // If we skip a channel, we also skip its policies.
                        if policy1 != nil {
                                skippedPolicyCount++
                        }
                        if policy2 != nil {
                                skippedPolicyCount++
                        }

                        return nil
                } else if err != nil {
                        return fmt.Errorf("unable to marshal extra opaque "+
                                "data for channel %d: %w %v", scid, err,
                                channel.ExtraOpaqueData)
                }

                channelCount++
                err = migrateSingleChannel(
                        ctx, sqlDB, channel, policy1, policy2, migChanPolicy,
                )
                if err != nil {
                        return fmt.Errorf("could not migrate channel %d: %w",
                                scid, err)
                }

                return nil
        })
        if err != nil {
                return fmt.Errorf("could not migrate channels and policies: %w",
                        err)
        }

        log.Infof("Migrated %d channels and %d policies from KV to SQL "+
                "(skipped %d channels and %d policies due to invalid TLV "+
                "streams)", channelCount, policyCount, skippedChanCount,
                skippedPolicyCount)

        return nil
}

func migrateSingleChannel(ctx context.Context, sqlDB SQLQueries,
        channel *models.ChannelEdgeInfo,
        policy1, policy2 *models.ChannelEdgePolicy,
        migChanPolicy func(*models.ChannelEdgePolicy) error) error {

        scid := channel.ChannelID

        // First, migrate the channel info along with its policies.
        dbChanInfo, err := insertChannel(ctx, sqlDB, channel)
        if err != nil {
                return fmt.Errorf("could not insert record for channel %d "+
                        "in SQL store: %w", scid, err)
        }

        // Now, migrate the two channel policies.
        err = migChanPolicy(policy1)
        if err != nil {
                return fmt.Errorf("could not migrate policy1(%d): %w", scid,
                        err)
        }
        err = migChanPolicy(policy2)
        if err != nil {
                return fmt.Errorf("could not migrate policy2(%d): %w", scid,
                        err)
        }

        // Now, fetch the channel and its policies from the SQL DB.
        row, err := sqlDB.GetChannelBySCIDWithPolicies(
                ctx, sqlc.GetChannelBySCIDWithPoliciesParams{
                        Scid:    channelIDToBytes(scid),
                        Version: int16(ProtocolV1),
                },
        )
        if err != nil {
                return fmt.Errorf("could not get channel by SCID(%d): %w", scid,
                        err)
        }

        // Assert that the DB IDs for the channel and nodes are as expected
        // given the inserted channel info.
        err = sqldb.CompareRecords(
                dbChanInfo.channelID, row.Channel.ID, "channel DB ID",
        )
        if err != nil {
                return err
        }
        err = sqldb.CompareRecords(
                dbChanInfo.node1ID, row.Node.ID, "node1 DB ID",
        )
        if err != nil {
                return err
        }
        err = sqldb.CompareRecords(
                dbChanInfo.node2ID, row.Node_2.ID, "node2 DB ID",
        )
        if err != nil {
                return err
        }

        migChan, migPol1, migPol2, err := getAndBuildChanAndPolicies(
                ctx, sqlDB, row, channel.ChainHash,
        )
        if err != nil {
                return fmt.Errorf("could not build migrated channel and "+
                        "policies: %w", err)
        }

        // Finally, compare the original channel info and
        // policies with the migrated ones to ensure they match.
        if len(channel.ExtraOpaqueData) == 0 {
                channel.ExtraOpaqueData = nil
        }
        if len(migChan.ExtraOpaqueData) == 0 {
                migChan.ExtraOpaqueData = nil
        }

        err = sqldb.CompareRecords(
                channel, migChan, fmt.Sprintf("channel %d", scid),
        )
        if err != nil {
                return err
        }

        checkPolicy := func(expPolicy,
                migPolicy *models.ChannelEdgePolicy) error {

                switch {
                // Both policies are nil, nothing to compare.
                case expPolicy == nil && migPolicy == nil:
                        return nil

                // One of the policies is nil, but the other is not.
                case expPolicy == nil || migPolicy == nil:
                        return fmt.Errorf("expected both policies to be "+
                                "non-nil. Got expPolicy: %v, "+
                                "migPolicy: %v", expPolicy, migPolicy)

                // Both policies are non-nil, we can compare them.
                default:
                }

                if len(expPolicy.ExtraOpaqueData) == 0 {
                        expPolicy.ExtraOpaqueData = nil
                }
                if len(migPolicy.ExtraOpaqueData) == 0 {
                        migPolicy.ExtraOpaqueData = nil
                }

                return sqldb.CompareRecords(
                        *expPolicy, *migPolicy, "channel policy",
                )
        }

        err = checkPolicy(policy1, migPol1)
        if err != nil {
                return fmt.Errorf("policy1 mismatch for channel %d: %w", scid,
                        err)
        }

        err = checkPolicy(policy2, migPol2)
        if err != nil {
                return fmt.Errorf("policy2 mismatch for channel %d: %w", scid,
                        err)
        }

        return nil
}

func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries,
        row sqlc.GetChannelBySCIDWithPoliciesRow,
        chain chainhash.Hash) (*models.ChannelEdgeInfo,
        *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) {

        node1, node2, err := buildNodeVertices(
                row.Node.PubKey, row.Node_2.PubKey,
        )
        if err != nil {
                return nil, nil, nil, err
        }

        edge, err := getAndBuildEdgeInfo(
                ctx, db, chain, row.Channel.ID, row.Channel, node1, node2,
        )
        if err != nil {
                return nil, nil, nil, fmt.Errorf("unable to build channel "+
                        "info: %w", err)
        }

        dbPol1, dbPol2, err := extractChannelPolicies(row)
        if err != nil {
                return nil, nil, nil, 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 nil, nil, nil, fmt.Errorf("unable to build channel "+
                        "policies: %w", err)
        }

        return edge, policy1, policy2, nil
}

1	package graphdb
2
3	import (
4	"cmp"
5	"context"
6	"errors"
7	"fmt"
8	"net"
9	"slices"
10	"time"
11
12	"github.com/btcsuite/btcd/chaincfg/chainhash"
13	"github.com/lightningnetwork/lnd/graph/db/models"
14	"github.com/lightningnetwork/lnd/kvdb"
15	"github.com/lightningnetwork/lnd/sqldb"
16	"github.com/lightningnetwork/lnd/sqldb/sqlc"
17	)
18
19	// MigrateGraphToSQL migrates the graph store from a KV backend to a SQL
20	// backend.
21	//
22	// NOTE: this is currently not called from any code path. It is called via tests
23	// only for now and will be called from the main lnd binary once the
24	// migration is fully implemented and tested.
25	func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend,
NEW 26	sqlDB SQLQueries, chain chainhash.Hash) error {	×
27		×
28	log.Infof("Starting migration of the graph store from KV to SQL")	×
29	t0 := time.Now()	×
30		×
31	// Check if there is a graph to migrate.	×
32	graphExists, err := checkGraphExists(kvBackend)	×
33	if err != nil {	×
34	return fmt.Errorf("failed to check graph existence: %w", err)	×
35	}	×
36	if !graphExists {	×
37	log.Infof("No graph found in KV store, skipping the migration")	×
38	return nil	×
39	}	×
40
41	// 1) Migrate all the nodes.
42	if err := migrateNodes(ctx, kvBackend, sqlDB); err != nil {	×
43	return fmt.Errorf("could not migrate nodes: %w", err)	×
44	}	×
45
46	// 2) Migrate the source node.
47	if err := migrateSourceNode(ctx, kvBackend, sqlDB); err != nil {	×
48	return fmt.Errorf("could not migrate source node: %w", err)	×
49	}	×
50
51	// 3) Migrate all the channels and channel policies.
NEW 52	err = migrateChannelsAndPolicies(ctx, kvBackend, sqlDB, chain)	×
NEW 53	if err != nil {	×
NEW 54	return fmt.Errorf("could not migrate channels and policies: %w",	×
NEW 55	err)	×
NEW 56	}	×
57
58	log.Infof("Finished migration of the graph store from KV to SQL in %v",	×
59	time.Since(t0))	×
60		×
61	return nil	×
62	}
63
64	// checkGraphExists checks if the graph exists in the KV backend.
65	func checkGraphExists(db kvdb.Backend) (bool, error) {	×
66	// Check if there is even a graph to migrate.	×
67	err := db.View(func(tx kvdb.RTx) error {	×
68	// Check for the existence of the node bucket which is a top	×
69	// level bucket that would have been created on the initial	×
70	// creation of the graph store.	×
71	nodes := tx.ReadBucket(nodeBucket)	×
72	if nodes == nil {	×
73	return ErrGraphNotFound	×
74	}	×
75
76	return nil	×
77	}, func() {})	×
78	if errors.Is(err, ErrGraphNotFound) {	×
79	return false, nil	×
80	} else if err != nil {	×
81	return false, err	×
82	}	×
83
84	return true, nil	×
85	}
86
87	// migrateNodes migrates all nodes from the KV backend to the SQL database.
88	// This includes doing a sanity check after each migration to ensure that the
89	// migrated node matches the original node.
90	func migrateNodes(ctx context.Context, kvBackend kvdb.Backend,
91	sqlDB SQLQueries) error {	×
92		×
93	// Keep track of the number of nodes migrated and the number of	×
94	// nodes skipped due to errors.	×
95	var (	×
96	count uint64	×
97	skipped uint64	×
98	)	×
99		×
100	// Loop through each node in the KV store and insert it into the SQL	×
101	// database.	×
102	err := forEachNode(kvBackend, func(_ kvdb.RTx,	×
103	node *models.LightningNode) error {	×
104		×
105	pub := node.PubKeyBytes	×
106		×
107	// Sanity check to ensure that the node has valid extra opaque	×
108	// data. If it does not, we'll skip it. We need to do this	×
109	// because previously we would just persist any TLV bytes that	×
110	// we received without validating them. Now, however, we	×
111	// normalise the storage of extra opaque data, so we need to	×
112	// ensure that the data is valid. We don't want to abort the	×
113	// migration if we encounter a node with invalid extra opaque	×
114	// data, so we'll just skip it and log a warning.	×
115	_, err := marshalExtraOpaqueData(node.ExtraOpaqueData)	×
116	if errors.Is(err, ErrParsingExtraTLVBytes) {	×
117	skipped++	×
118	log.Warnf("Skipping migration of node %x with invalid "+	×
119	"extra opaque data: %v", pub,	×
120	node.ExtraOpaqueData)	×
121		×
122	return nil	×
123	} else if err != nil {	×
124	return fmt.Errorf("unable to marshal extra "+	×
125	"opaque data for node %x: %w", pub, err)	×
126	}	×
127
128	count++	×
129		×
130	// TODO(elle): At this point, we should check the loaded node	×
131	// to see if we should extract any DNS addresses from its	×
132	// opaque type addresses. This is expected to be done in:	×
133	// https://github.com/lightningnetwork/lnd/pull/9455.	×
134	// This TODO is being tracked in	×
135	// https://github.com/lightningnetwork/lnd/issues/9795 as this	×
136	// must be addressed before making this code path active in	×
137	// production.	×
138		×
139	// Write the node to the SQL database.	×
140	id, err := upsertNode(ctx, sqlDB, node)	×
141	if err != nil {	×
142	return fmt.Errorf("could not persist node(%x): %w", pub,	×
143	err)	×
144	}	×
145
146	// Fetch it from the SQL store and compare it against the
147	// original node object to ensure the migration was successful.
148	dbNode, err := sqlDB.GetNodeByPubKey(	×
149	ctx, sqlc.GetNodeByPubKeyParams{	×
150	PubKey: node.PubKeyBytes[:],	×
151	Version: int16(ProtocolV1),	×
152	},	×
153	)	×
154	if err != nil {	×
155	return fmt.Errorf("could not get node by pubkey (%x)"+	×
156	"after migration: %w", pub, err)	×
157	}	×
158
159	// Sanity check: ensure the migrated node ID matches the one we
160	// just inserted.
161	if dbNode.ID != id {	×
162	return fmt.Errorf("node ID mismatch for node (%x) "+	×
163	"after migration: expected %d, got %d",	×
164	pub, id, dbNode.ID)	×
165	}	×
166
167	migratedNode, err := buildNode(ctx, sqlDB, &dbNode)	×
168	if err != nil {	×
169	return fmt.Errorf("could not build migrated node "+	×
170	"from dbNode(db id: %d, node pub: %x): %w",	×
171	dbNode.ID, pub, err)	×
172	}	×
173
174	// Make sure that the node addresses are sorted before
175	// comparing them to ensure that the order of addresses does
176	// not affect the comparison.
177	slices.SortFunc(node.Addresses, func(i, j net.Addr) int {	×
178	return cmp.Compare(i.String(), j.String())	×
179	})	×
180	slices.SortFunc(	×
181	migratedNode.Addresses, func(i, j net.Addr) int {	×
182	return cmp.Compare(i.String(), j.String())	×
183	},	×
184	)
185
186	return sqldb.CompareRecords(	×
187	node, migratedNode, fmt.Sprintf("node %x", pub),	×
188	)	×
189	})
190	if err != nil {	×
191	return fmt.Errorf("could not migrate nodes: %w", err)	×
192	}	×
193
194	log.Infof("Migrated %d nodes from KV to SQL (skipped %d nodes due to "+	×
195	"invalid TLV streams)", count, skipped)	×
196		×
197	return nil	×
198	}
199
200	// migrateSourceNode migrates the source node from the KV backend to the
201	// SQL database.
202	func migrateSourceNode(ctx context.Context, kvdb kvdb.Backend,
203	sqlDB SQLQueries) error {	×
204		×
205	sourceNode, err := sourceNode(kvdb)	×
206	if errors.Is(err, ErrSourceNodeNotSet) {	×
207	// If the source node has not been set yet, we can skip this	×
208	// migration step.	×
209	return nil	×
210	} else if err != nil {	×
211	return fmt.Errorf("could not get source node from kv "+	×
212	"store: %w", err)	×
213	}	×
214
215	pub := sourceNode.PubKeyBytes	×
216		×
217	// Get the DB ID of the source node by its public key. This node must	×
218	// already exist in the SQL database, as it should have been migrated	×
219	// in the previous node-migration step.	×
220	id, err := sqlDB.GetNodeIDByPubKey(	×
221	ctx, sqlc.GetNodeIDByPubKeyParams{	×
222	PubKey: pub[:],	×
223	Version: int16(ProtocolV1),	×
224	},	×
225	)	×
226	if err != nil {	×
227	return fmt.Errorf("could not get source node ID: %w", err)	×
228	}	×
229
230	// Now we can add the source node to the SQL database.
231	err = sqlDB.AddSourceNode(ctx, id)	×
232	if err != nil {	×
233	return fmt.Errorf("could not add source node to SQL store: %w",	×
234	err)	×
235	}	×
236
237	// Verify that the source node was added correctly by fetching it back
238	// from the SQL database and checking that the expected DB ID and
239	// pub key are returned. We don't need to do a whole node comparison
240	// here, as this was already done in the previous migration step.
241	srcNodes, err := sqlDB.GetSourceNodesByVersion(ctx, int16(ProtocolV1))	×
242	if err != nil {	×
243	return fmt.Errorf("could not get source nodes from SQL "+	×
244	"store: %w", err)	×
245	}	×
246
247	// The SQL store has support for multiple source nodes (for future
248	// protocol versions) but this migration is purely aimed at the V1
249	// store, and so we expect exactly one source node to be present.
250	if len(srcNodes) != 1 {	×
251	return fmt.Errorf("expected exactly one source node, "+	×
252	"got %d", len(srcNodes))	×
253	}	×
254
255	// Check that the source node ID and pub key match the original
256	// source node.
257	if srcNodes[0].NodeID != id {	×
258	return fmt.Errorf("source node ID mismatch after migration: "+	×
259	"expected %d, got %d", id, srcNodes[0].NodeID)	×
260	}	×
261	err = sqldb.CompareRecords(pub[:], srcNodes[0].PubKey, "source node")	×
262	if err != nil {	×
263	return fmt.Errorf("source node pubkey mismatch after "+	×
264	"migration: %w", err)	×
265	}	×
266
267	log.Infof("Migrated source node with pubkey %x to SQL", pub[:])	×
268		×
269	return nil	×
270	}
271
272	// migrateChannelsAndPolicies migrates all channels and their policies
273	// from the KV backend to the SQL database.
274	func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend,
NEW 275	sqlDB SQLQueries, chain chainhash.Hash) error {	×
NEW 276		×
NEW 277	var (	×
NEW 278	channelCount uint64	×
NEW 279	skippedChanCount uint64	×
NEW 280	policyCount uint64	×
NEW 281	skippedPolicyCount uint64	×
NEW 282	)	×
NEW 283	migChanPolicy := func(policy *models.ChannelEdgePolicy) error {	×
NEW 284	// If the policy is nil, we can skip it.	×
NEW 285	if policy == nil {	×
NEW 286	return nil	×
NEW 287	}	×
288
289	// Sanity check to ensure that the policy has valid extra opaque
290	// data. If it does not, we'll skip it. We need to do this
291	// because previously we would just persist any TLV bytes that
292	// we received without validating them. Now, however, we
293	// normalise the storage of extra opaque data, so we need to
294	// ensure that the data is valid. We don't want to abort the
295	// migration if we encounter a policy with invalid extra opaque
296	// data, so we'll just skip it and log a warning.
NEW 297	_, err := marshalExtraOpaqueData(policy.ExtraOpaqueData)	×
NEW 298	if errors.Is(err, ErrParsingExtraTLVBytes) {	×
NEW 299	skippedPolicyCount++	×
NEW 300	log.Warnf("Skipping policy for channel %d with "+	×
NEW 301	"invalid extra opaque data: %v",	×
NEW 302	policy.ChannelID, policy.ExtraOpaqueData)	×
NEW 303		×
NEW 304	return nil	×
NEW 305	} else if err != nil {	×
NEW 306	return fmt.Errorf("unable to marshal extra opaque "+	×
NEW 307	"data (%v): %w", policy.ExtraOpaqueData, err)	×
NEW 308	}	×
309
NEW 310	policyCount++	×
NEW 311		×
NEW 312	_, _, _, err = updateChanEdgePolicy(ctx, sqlDB, policy)	×
NEW 313	if err != nil {	×
NEW 314	return fmt.Errorf("could not migrate channel "+	×
NEW 315	"policy %d: %w", policy.ChannelID, err)	×
NEW 316	}	×
317
NEW 318	return nil	×
319	}
320
321	// Iterate over each channel in the KV store and migrate it and its
322	// policies to the SQL database.
NEW 323	err := forEachChannel(kvBackend, func(channel *models.ChannelEdgeInfo,	×
NEW 324	policy1 *models.ChannelEdgePolicy,	×
NEW 325	policy2 *models.ChannelEdgePolicy) error {	×
NEW 326		×
NEW 327	scid := channel.ChannelID	×
NEW 328		×
NEW 329	// Here, we do a sanity check to ensure that the chain hash of	×
NEW 330	// the channel returned by the KV store matches the expected	×
NEW 331	// chain hash. This is important since in the SQL store, we will	×
NEW 332	// no longer explicitly store the chain hash in the channel	×
NEW 333	// info, but rather rely on the chain hash LND is running with.	×
NEW 334	// So this is our way of ensuring that LND is running on the	×
NEW 335	// correct network at migration time.	×
NEW 336	if channel.ChainHash != chain {	×
NEW 337	return fmt.Errorf("channel %d has chain hash %s, "+	×
NEW 338	"expected %s", scid, channel.ChainHash, chain)	×
NEW 339	}	×
340
341	// Sanity check to ensure that the channel has valid extra
342	// opaque data. If it does not, we'll skip it. We need to do
343	// this because previously we would just persist any TLV bytes
344	// that we received without validating them. Now, however, we
345	// normalise the storage of extra opaque data, so we need to
346	// ensure that the data is valid. We don't want to abort the
347	// migration if we encounter a channel with invalid extra opaque
348	// data, so we'll just skip it and log a warning.
NEW 349	_, err := marshalExtraOpaqueData(channel.ExtraOpaqueData)	×
NEW 350	if errors.Is(err, ErrParsingExtraTLVBytes) {	×
NEW 351	log.Warnf("Skipping channel %d with invalid "+	×
NEW 352	"extra opaque data: %v", scid,	×
NEW 353	channel.ExtraOpaqueData)	×
NEW 354		×
NEW 355	skippedChanCount++	×
NEW 356		×
NEW 357	// If we skip a channel, we also skip its policies.	×
NEW 358	if policy1 != nil {	×
NEW 359	skippedPolicyCount++	×
NEW 360	}	×
NEW 361	if policy2 != nil {	×
NEW 362	skippedPolicyCount++	×
NEW 363	}	×
364
NEW 365	return nil	×
NEW 366	} else if err != nil {	×
NEW 367	return fmt.Errorf("unable to marshal extra opaque "+	×
NEW 368	"data for channel %d: %w %v", scid, err,	×
NEW 369	channel.ExtraOpaqueData)	×
NEW 370	}	×
371
NEW 372	channelCount++	×
NEW 373	err = migrateSingleChannel(	×
NEW 374	ctx, sqlDB, channel, policy1, policy2, migChanPolicy,	×
NEW 375	)	×
NEW 376	if err != nil {	×
NEW 377	return fmt.Errorf("could not migrate channel %d: %w",	×
NEW 378	scid, err)	×
NEW 379	}	×
380
NEW 381	return nil	×
382	})
NEW 383	if err != nil {	×
NEW 384	return fmt.Errorf("could not migrate channels and policies: %w",	×
NEW 385	err)	×
NEW 386	}	×
387
NEW 388	log.Infof("Migrated %d channels and %d policies from KV to SQL "+	×
NEW 389	"(skipped %d channels and %d policies due to invalid TLV "+	×
NEW 390	"streams)", channelCount, policyCount, skippedChanCount,	×
NEW 391	skippedPolicyCount)	×
NEW 392		×
NEW 393	return nil	×
394	}
395
396	func migrateSingleChannel(ctx context.Context, sqlDB SQLQueries,
397	channel *models.ChannelEdgeInfo,
398	policy1, policy2 *models.ChannelEdgePolicy,
NEW 399	migChanPolicy func(*models.ChannelEdgePolicy) error) error {	×
NEW 400		×
NEW 401	scid := channel.ChannelID	×
NEW 402		×
NEW 403	// First, migrate the channel info along with its policies.	×
NEW 404	dbChanInfo, err := insertChannel(ctx, sqlDB, channel)	×
NEW 405	if err != nil {	×
NEW 406	return fmt.Errorf("could not insert record for channel %d "+	×
NEW 407	"in SQL store: %w", scid, err)	×
NEW 408	}	×
409
410	// Now, migrate the two channel policies.
NEW 411	err = migChanPolicy(policy1)	×
NEW 412	if err != nil {	×
NEW 413	return fmt.Errorf("could not migrate policy1(%d): %w", scid,	×
NEW 414	err)	×
NEW 415	}	×
NEW 416	err = migChanPolicy(policy2)	×
NEW 417	if err != nil {	×
NEW 418	return fmt.Errorf("could not migrate policy2(%d): %w", scid,	×
NEW 419	err)	×
NEW 420	}	×
421
422	// Now, fetch the channel and its policies from the SQL DB.
NEW 423	row, err := sqlDB.GetChannelBySCIDWithPolicies(	×
NEW 424	ctx, sqlc.GetChannelBySCIDWithPoliciesParams{	×
NEW 425	Scid: channelIDToBytes(scid),	×
NEW 426	Version: int16(ProtocolV1),	×
NEW 427	},	×
NEW 428	)	×
NEW 429	if err != nil {	×
NEW 430	return fmt.Errorf("could not get channel by SCID(%d): %w", scid,	×
NEW 431	err)	×
NEW 432	}	×
433
434	// Assert that the DB IDs for the channel and nodes are as expected
435	// given the inserted channel info.
NEW 436	err = sqldb.CompareRecords(	×
NEW 437	dbChanInfo.channelID, row.Channel.ID, "channel DB ID",	×
NEW 438	)	×
NEW 439	if err != nil {	×
NEW 440	return err	×
NEW 441	}	×
NEW 442	err = sqldb.CompareRecords(	×
NEW 443	dbChanInfo.node1ID, row.Node.ID, "node1 DB ID",	×
NEW 444	)	×
NEW 445	if err != nil {	×
NEW 446	return err	×
NEW 447	}	×
NEW 448	err = sqldb.CompareRecords(	×
NEW 449	dbChanInfo.node2ID, row.Node_2.ID, "node2 DB ID",	×
NEW 450	)	×
NEW 451	if err != nil {	×
NEW 452	return err	×
NEW 453	}	×
454
NEW 455	migChan, migPol1, migPol2, err := getAndBuildChanAndPolicies(	×
NEW 456	ctx, sqlDB, row, channel.ChainHash,	×
NEW 457	)	×
NEW 458	if err != nil {	×
NEW 459	return fmt.Errorf("could not build migrated channel and "+	×
NEW 460	"policies: %w", err)	×
NEW 461	}	×
462
463	// Finally, compare the original channel info and
464	// policies with the migrated ones to ensure they match.
NEW 465	if len(channel.ExtraOpaqueData) == 0 {	×
NEW 466	channel.ExtraOpaqueData = nil	×
NEW 467	}	×
NEW 468	if len(migChan.ExtraOpaqueData) == 0 {	×
NEW 469	migChan.ExtraOpaqueData = nil	×
NEW 470	}	×
471
NEW 472	err = sqldb.CompareRecords(	×
NEW 473	channel, migChan, fmt.Sprintf("channel %d", scid),	×
NEW 474	)	×
NEW 475	if err != nil {	×
NEW 476	return err	×
NEW 477	}	×
478
NEW 479	checkPolicy := func(expPolicy,	×
NEW 480	migPolicy *models.ChannelEdgePolicy) error {	×
NEW 481		×
NEW 482	switch {	×
483	// Both policies are nil, nothing to compare.
NEW 484	case expPolicy == nil && migPolicy == nil:	×
NEW 485	return nil	×
486
487	// One of the policies is nil, but the other is not.
NEW 488	case expPolicy == nil \|\| migPolicy == nil:	×
NEW 489	return fmt.Errorf("expected both policies to be "+	×
NEW 490	"non-nil. Got expPolicy: %v, "+	×
NEW 491	"migPolicy: %v", expPolicy, migPolicy)	×
492
493	// Both policies are non-nil, we can compare them.
NEW 494	default:	×
495	}
496
NEW 497	if len(expPolicy.ExtraOpaqueData) == 0 {	×
NEW 498	expPolicy.ExtraOpaqueData = nil	×
NEW 499	}	×
NEW 500	if len(migPolicy.ExtraOpaqueData) == 0 {	×
NEW 501	migPolicy.ExtraOpaqueData = nil	×
NEW 502	}	×
503
NEW 504	return sqldb.CompareRecords(	×
NEW 505	expPolicy, migPolicy, "channel policy",	×
NEW 506	)	×
507	}
508
NEW 509	err = checkPolicy(policy1, migPol1)	×
NEW 510	if err != nil {	×
NEW 511	return fmt.Errorf("policy1 mismatch for channel %d: %w", scid,	×
NEW 512	err)	×
NEW 513	}	×
514
NEW 515	err = checkPolicy(policy2, migPol2)	×
NEW 516	if err != nil {	×
NEW 517	return fmt.Errorf("policy2 mismatch for channel %d: %w", scid,	×
NEW 518	err)	×
NEW 519	}	×
520
NEW 521	return nil	×
522	}
523
524	func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries,
525	row sqlc.GetChannelBySCIDWithPoliciesRow,
526	chain chainhash.Hash) (*models.ChannelEdgeInfo,
NEW 527	models.ChannelEdgePolicy, models.ChannelEdgePolicy, error) {	×
NEW 528		×
NEW 529	node1, node2, err := buildNodeVertices(	×
NEW 530	row.Node.PubKey, row.Node_2.PubKey,	×
NEW 531	)	×
NEW 532	if err != nil {	×
NEW 533	return nil, nil, nil, err	×
NEW 534	}	×
535
NEW 536	edge, err := getAndBuildEdgeInfo(	×
NEW 537	ctx, db, chain, row.Channel.ID, row.Channel, node1, node2,	×
NEW 538	)	×
NEW 539	if err != nil {	×
NEW 540	return nil, nil, nil, fmt.Errorf("unable to build channel "+	×
NEW 541	"info: %w", err)	×
NEW 542	}	×
543
NEW 544	dbPol1, dbPol2, err := extractChannelPolicies(row)	×
NEW 545	if err != nil {	×
NEW 546	return nil, nil, nil, fmt.Errorf("unable to extract channel "+	×
NEW 547	"policies: %w", err)	×
NEW 548	}	×
549
NEW 550	policy1, policy2, err := getAndBuildChanPolicies(	×
NEW 551	ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2,	×
NEW 552	)	×
NEW 553	if err != nil {	×
NEW 554	return nil, nil, nil, fmt.Errorf("unable to build channel "+	×
NEW 555	"policies: %w", err)	×
NEW 556	}	×
557
NEW 558	return edge, policy1, policy2, nil	×
559	}

lightningnetwork / lnd / 16139765508

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous