374 lines
9.4 KiB
Go
374 lines
9.4 KiB
Go
//go:build test_db_postgres || test_db_sqlite
|
|
|
|
package migration1
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"path"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/btcsuite/btclog/v2"
|
|
"github.com/lightningnetwork/lnd/kvdb"
|
|
"github.com/lightningnetwork/lnd/kvdb/postgres"
|
|
"github.com/lightningnetwork/lnd/kvdb/sqlbase"
|
|
"github.com/lightningnetwork/lnd/kvdb/sqlite"
|
|
"github.com/lightningnetwork/lnd/lntypes"
|
|
"github.com/lightningnetwork/lnd/payments/db/migration1/sqlc"
|
|
"github.com/lightningnetwork/lnd/sqldb"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
// TestMigrationWithExternalDB tests the migration of the payment store from a
|
|
// bolt backed channel.db or a kvdb channel.sqlite to a SQL database. Note that
|
|
// this test does not attempt to be a complete migration test for all payment
|
|
// store types but rather is added as a tool for developers and users to debug
|
|
// payment migration issues with an actual channel.db/channel.sqlite file.
|
|
//
|
|
// NOTE: To use this test, place either of those files in the
|
|
// payments/db/migration1/testdata directory, uncomment the "Skipf" line, and
|
|
// set the "fileName" variable to the name of the channel database file you
|
|
// want to use for the migration test.
|
|
func TestMigrationWithExternalDB(t *testing.T) {
|
|
ctx := context.Background()
|
|
|
|
// NOTE: comment this line out to run the test.
|
|
t.Skipf("skipping test meant for local debugging only")
|
|
|
|
// NOTE: set this to the name of the channel database file you want
|
|
// to use for the migration test. This may be either a bbolt ".db" file
|
|
// or a SQLite ".sqlite" file. If you want to migrate from a
|
|
// bbolt channel.db file, set this to "channel.db".
|
|
const fileName = "channel.db"
|
|
|
|
// NOTE: if set, this test will prefer migrating from a Postgres-backed
|
|
// kvdb source instead of a local file. Leave empty to use fileName.
|
|
const postgresKVDSN = ""
|
|
const postgresKVPfx = "channeldb"
|
|
const logSequenceOrder = false
|
|
|
|
// Determine if we are using a SQLite file or a Bolt DB file.
|
|
isSqlite := strings.HasSuffix(fileName, ".sqlite")
|
|
|
|
// Set up logging for the test.
|
|
logger := btclog.NewSLogger(btclog.NewDefaultHandler(os.Stdout))
|
|
UseLogger(logger)
|
|
|
|
// migrate runs the migration from the kvdb store to the SQL store
|
|
// and then performs a batched deep comparison of every migrated
|
|
// payment to ensure all data (including HTLC details) was preserved
|
|
// correctly.
|
|
migrate := func(t *testing.T, kvBackend kvdb.Backend) {
|
|
sqlStore := setupTestSQLDB(t)
|
|
|
|
// Run migration in a transaction.
|
|
err := sqlStore.db.ExecTx(
|
|
ctx, sqldb.WriteTxOpt(), func(tx SQLQueries) error {
|
|
migTx, ok := tx.(SQLMigrationQueries)
|
|
if !ok {
|
|
return fmt.Errorf("db does not " +
|
|
"implement SQLMigrationQueries")
|
|
}
|
|
|
|
return MigratePaymentsKVToSQL(
|
|
ctx, kvBackend, migTx, &SQLStoreConfig{
|
|
QueryCfg: sqlStore.cfg.QueryCfg,
|
|
},
|
|
)
|
|
}, sqldb.NoOpReset,
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
t.Log("========================================")
|
|
t.Log(" Deep Validation")
|
|
t.Log("========================================")
|
|
|
|
// Deep compare all migrated payments in batches.
|
|
deepValidateAllPayments(
|
|
t, ctx, kvBackend, sqlStore,
|
|
)
|
|
|
|
_ = logSequenceOrder
|
|
}
|
|
|
|
connectPostgres := func(t *testing.T, dsn, prefix string) kvdb.Backend {
|
|
dsn = strings.TrimSpace(dsn)
|
|
if dsn == "" {
|
|
t.Fatalf("missing postgres kvdb dsn")
|
|
}
|
|
|
|
prefix = strings.TrimSpace(prefix)
|
|
if prefix == "" {
|
|
prefix = "channeldb"
|
|
}
|
|
|
|
const (
|
|
timeout = 10 * time.Second
|
|
maxConns = 5
|
|
)
|
|
sqlbase.Init(maxConns)
|
|
|
|
dbCfg := &postgres.Config{
|
|
Dsn: dsn,
|
|
Timeout: timeout,
|
|
MaxConnections: maxConns,
|
|
}
|
|
|
|
kvStore, err := kvdb.Open(
|
|
kvdb.PostgresBackendName, ctx, dbCfg, prefix,
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
t.Cleanup(func() { _ = kvStore.Close() })
|
|
|
|
return kvStore
|
|
}
|
|
|
|
connectPostgresKV := func(t *testing.T) kvdb.Backend {
|
|
return connectPostgres(t, postgresKVDSN, postgresKVPfx)
|
|
}
|
|
|
|
connectBBolt := func(t *testing.T, dbPath string) kvdb.Backend {
|
|
cfg := &kvdb.BoltBackendConfig{
|
|
DBPath: dbPath,
|
|
DBFileName: fileName,
|
|
NoFreelistSync: true,
|
|
AutoCompact: false,
|
|
AutoCompactMinAge: kvdb.DefaultBoltAutoCompactMinAge,
|
|
DBTimeout: kvdb.DefaultDBTimeout,
|
|
ReadOnly: true,
|
|
}
|
|
|
|
kvStore, err := kvdb.GetBoltBackend(cfg)
|
|
require.NoError(t, err)
|
|
|
|
t.Cleanup(func() { _ = kvStore.Close() })
|
|
|
|
return kvStore
|
|
}
|
|
|
|
connectSQLite := func(t *testing.T, dbPath string) kvdb.Backend {
|
|
const (
|
|
timeout = 10 * time.Second
|
|
maxConns = 5
|
|
)
|
|
sqlbase.Init(maxConns)
|
|
|
|
cfg := &sqlite.Config{
|
|
Timeout: timeout,
|
|
BusyTimeout: timeout,
|
|
MaxConnections: maxConns,
|
|
}
|
|
|
|
kvStore, err := kvdb.Open(
|
|
kvdb.SqliteBackendName, ctx, cfg,
|
|
dbPath, fileName,
|
|
// NOTE: we use the raw string here else we get an
|
|
// import cycle if we try to import lncfg.NSChannelDB.
|
|
"channeldb",
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
t.Cleanup(func() { _ = kvStore.Close() })
|
|
|
|
return kvStore
|
|
}
|
|
|
|
tests := []struct {
|
|
name string
|
|
dbPath string
|
|
}{
|
|
{
|
|
name: "testdata",
|
|
dbPath: "testdata",
|
|
},
|
|
}
|
|
|
|
for _, test := range tests {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
if postgresKVDSN != "" {
|
|
migrate(t, connectPostgresKV(t))
|
|
return
|
|
}
|
|
|
|
chanDBPath := path.Join(test.dbPath, fileName)
|
|
t.Logf("Connecting to channel DB at: %s", chanDBPath)
|
|
|
|
connectDB := connectBBolt
|
|
if isSqlite {
|
|
connectDB = connectSQLite
|
|
}
|
|
|
|
migrate(t, connectDB(t, test.dbPath))
|
|
})
|
|
}
|
|
}
|
|
|
|
// kvPaymentRef holds a KV payment and its hash for batched deep validation.
|
|
type kvPaymentRef struct {
|
|
hash lntypes.Hash
|
|
payment *MPPayment
|
|
}
|
|
|
|
// deepValidateAllPayments iterates all KV payments in batches and performs a
|
|
// deep comparison against their SQL counterparts. For each batch, it fetches
|
|
// the full SQL payment data (including HTLCs, routes, custom records) via
|
|
// batch queries and compares field-by-field with the KV data.
|
|
func deepValidateAllPayments(t *testing.T, ctx context.Context,
|
|
kvBackend kvdb.Backend, sqlStore *SQLStore) {
|
|
|
|
t.Helper()
|
|
|
|
batchSize := int(sqlStore.cfg.QueryCfg.MaxBatchSize)
|
|
|
|
// Get total payment count from SQL so we can show progress.
|
|
var totalPayments int64
|
|
err := sqlStore.db.ExecTx(
|
|
ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
var err error
|
|
totalPayments, err = db.CountPayments(ctx)
|
|
return err
|
|
}, sqldb.NoOpReset,
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
var totalValidated int
|
|
|
|
// Iterate all KV payments, collecting them into batches. For each
|
|
// full batch, perform a batched deep comparison against SQL.
|
|
var batch []kvPaymentRef
|
|
err = kvdb.View(kvBackend, func(kvTx kvdb.RTx) error {
|
|
paymentsBucket := kvTx.ReadBucket(paymentsRootBucket)
|
|
if paymentsBucket == nil {
|
|
return nil
|
|
}
|
|
|
|
return paymentsBucket.ForEach(func(k, v []byte) error {
|
|
bucket := paymentsBucket.NestedReadBucket(k)
|
|
if bucket == nil {
|
|
return nil
|
|
}
|
|
|
|
payment, err := fetchPayment(bucket)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var hash lntypes.Hash
|
|
copy(hash[:], k)
|
|
|
|
batch = append(batch, kvPaymentRef{
|
|
hash: hash,
|
|
payment: payment,
|
|
})
|
|
|
|
if len(batch) >= batchSize {
|
|
deepCompareBatch(
|
|
t, ctx, sqlStore, batch,
|
|
)
|
|
totalValidated += len(batch)
|
|
t.Logf("Deep validated %d/%d payments",
|
|
totalValidated, totalPayments)
|
|
|
|
batch = batch[:0]
|
|
}
|
|
|
|
return nil
|
|
})
|
|
}, func() {
|
|
batch = nil
|
|
totalValidated = 0
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
// Validate any remaining payments in the last batch.
|
|
if len(batch) > 0 {
|
|
deepCompareBatch(t, ctx, sqlStore, batch)
|
|
totalValidated += len(batch)
|
|
}
|
|
|
|
t.Logf("Deep validated %d/%d payments (complete)",
|
|
totalValidated, totalPayments)
|
|
}
|
|
|
|
// deepCompareBatch performs a batched deep comparison of KV payments against
|
|
// their SQL counterparts. It fetches SQL payment data in a single read
|
|
// transaction using batch queries.
|
|
func deepCompareBatch(t *testing.T, ctx context.Context,
|
|
sqlStore *SQLStore, batch []kvPaymentRef) {
|
|
|
|
t.Helper()
|
|
|
|
err := sqlStore.db.ExecTx(
|
|
ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
|
|
// Look up each payment by hash to get the SQL row
|
|
// with payment ID.
|
|
ids := make([]int64, 0, len(batch))
|
|
rowsByHash := make(
|
|
map[lntypes.Hash]sqlc.FetchPaymentsByIDsRow,
|
|
len(batch),
|
|
)
|
|
for _, ref := range batch {
|
|
row, err := fetchPaymentByHash(
|
|
ctx, db, ref.hash,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("fetch SQL payment "+
|
|
"%x: %w", ref.hash[:8], err)
|
|
}
|
|
ids = append(ids, row.Payment.ID)
|
|
}
|
|
|
|
// Batch-fetch full payment details.
|
|
rows, err := db.FetchPaymentsByIDs(ctx, ids)
|
|
if err != nil {
|
|
return fmt.Errorf("batch fetch payments: %w",
|
|
err)
|
|
}
|
|
for _, row := range rows {
|
|
var hash lntypes.Hash
|
|
copy(hash[:], row.PaymentIdentifier)
|
|
rowsByHash[hash] = row
|
|
}
|
|
|
|
batchData, err := batchLoadPaymentDetailsData(
|
|
ctx, sqlStore.cfg.QueryCfg, db, ids,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("batch load details: %w",
|
|
err)
|
|
}
|
|
|
|
// Deep compare each payment.
|
|
for _, ref := range batch {
|
|
row, ok := rowsByHash[ref.hash]
|
|
require.True(t, ok,
|
|
"SQL payment %x not in batch "+
|
|
"results", ref.hash[:8])
|
|
|
|
sqlPayment, err := buildPaymentFromBatchData(
|
|
row, batchData,
|
|
)
|
|
require.NoError(t, err,
|
|
"build SQL payment %x",
|
|
ref.hash[:8])
|
|
|
|
normalizePaymentData(ref.payment)
|
|
normalizePaymentData(sqlPayment)
|
|
|
|
require.Equal(
|
|
t, ref.payment, sqlPayment,
|
|
"payment mismatch %x",
|
|
ref.hash[:8],
|
|
)
|
|
}
|
|
|
|
return nil
|
|
}, sqldb.NoOpReset,
|
|
)
|
|
require.NoError(t, err)
|
|
}
|