15561477203

Committed 10 Jun 2025 01:54PM UTC coverage: 58.351% (-10.1%) from 68.487%

Build # 15561477203

Build Type

Pull #9356

github

Committed by

web-flow

Commit Message

Merge 6440b25db into c6d6d4c0b

Pull Request Pull Request #9356: lnrpc: add incoming/outgoing channel ids filter to forwarding history request

Run Details

33 of 36 new or added lines in 2 files covered. (91.67%)

28366 existing lines in 455 files now uncovered.

97715 of 167461 relevant lines covered (58.35%)

1.81 hits per line

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

0.0

/channeldb/migration30/migration.go

package migration30

import (
        "bytes"
        "encoding/binary"
        "fmt"
        "math"
        "sync"

        mig24 "github.com/lightningnetwork/lnd/channeldb/migration24"
        mig26 "github.com/lightningnetwork/lnd/channeldb/migration26"
        mig "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11"
        "github.com/lightningnetwork/lnd/kvdb"
)

// recordsPerTx specifies the number of records to be migrated in each database
// transaction. In the worst case, each old revocation log is 28,057 bytes.
// 20,000 records would consume 0.56 GB of ram, which is feasible for a modern
// machine.
//
// NOTE: we could've used more ram but it doesn't help with the speed of the
// migration since the most of the CPU time is used for calculating the output
// indexes.
const recordsPerTx = 20_000

// MigrateRevLogConfig is an interface that defines the config that should be
// passed to the MigrateRevocationLog function.
type MigrateRevLogConfig interface {
        // GetNoAmountData returns true if the amount data of revoked commitment
        // transactions should not be stored in the revocation log.
        GetNoAmountData() bool
}

// MigrateRevLogConfigImpl implements the MigrationRevLogConfig interface.
type MigrateRevLogConfigImpl struct {
        // NoAmountData if set to true will result in the amount data of revoked
        // commitment transactions not being stored in the revocation log.
        NoAmountData bool
}

// GetNoAmountData returns true if the amount data of revoked commitment
// transactions should not be stored in the revocation log.
func (c *MigrateRevLogConfigImpl) GetNoAmountData() bool {
        return c.NoAmountData
}

// MigrateRevocationLog migrates the old revocation logs into the newer format
// and deletes them once finished, with the deletion only happens once ALL the
// old logs have been migrates.
func MigrateRevocationLog(db kvdb.Backend, cfg MigrateRevLogConfig) error {
        log.Infof("Migrating revocation logs, might take a while...")

        var (
                err error

                // finished is used to exit the for loop.
                finished bool

                // total is the number of total records.
                total uint64

                // migrated is the number of already migrated records.
                migrated uint64
        )

        // First of all, read the stats of the revocation logs.
        total, migrated, err = logMigrationStat(db)
        if err != nil {
                return err
        }
        log.Infof("Total logs=%d, migrated=%d", total, migrated)

        // Exit early if the old logs have already been migrated and deleted.
        if total == 0 {
                log.Info("Migration already finished!")
                return nil
        }

        for {
                if finished {
                        log.Infof("Migrating old revocation logs finished, " +
                                "now checking the migration results...")
                        break
                }

                // Process the migration.
                err = kvdb.Update(db, func(tx kvdb.RwTx) error {
                        finished, err = processMigration(tx, cfg)
                        if err != nil {
                                return err
                        }
                        return nil
                }, func() {})
                if err != nil {
                        return err
                }

                // Each time we finished the above process, we'd read the stats
                // again to understand the current progress.
                total, migrated, err = logMigrationStat(db)
                if err != nil {
                        return err
                }

                // Calculate and log the progress if the progress is less than
                // one.
                progress := float64(migrated) / float64(total) * 100
                if progress >= 100 {
                        continue
                }

                log.Infof("Migration progress: %.3f%%, still have: %d",
                        progress, total-migrated)
        }

        // Before we can safety delete the old buckets, we perform a check to
        // make sure the logs are migrated as expected.
        err = kvdb.Update(db, validateMigration, func() {})
        if err != nil {
                return fmt.Errorf("validate migration failed: %w", err)
        }

        log.Info("Migration check passed, now deleting the old logs...")

        // Once the migration completes, we can now safety delete the old
        // revocation logs.
        if err := deleteOldBuckets(db); err != nil {
                return fmt.Errorf("deleteOldBuckets err: %w", err)
        }

        log.Info("Old revocation log buckets removed!")
        return nil
}

// processMigration finds the next un-migrated revocation logs, reads a max
// number of `recordsPerTx` records, converts them into the new revocation logs
// and save them to disk.
func processMigration(tx kvdb.RwTx, cfg MigrateRevLogConfig) (bool, error) {
        openChanBucket := tx.ReadWriteBucket(openChannelBucket)

        // If no bucket is found, we can exit early.
        if openChanBucket == nil {
                return false, fmt.Errorf("root bucket not found")
        }

        // Locate the next migration height.
        locator, err := locateNextUpdateNum(openChanBucket)
        if err != nil {
                return false, fmt.Errorf("locator got error: %w", err)
        }

        // If the returned locator is nil, we've done migrating the logs.
        if locator == nil {
                return true, nil
        }

        // Read a list of old revocation logs.
        entryMap, err := readOldRevocationLogs(openChanBucket, locator, cfg)
        if err != nil {
                return false, fmt.Errorf("read old logs err: %w", err)
        }

        // Migrate the revocation logs.
        return false, writeRevocationLogs(openChanBucket, entryMap)
}

// deleteOldBuckets iterates all the channel buckets and deletes the old
// revocation buckets.
func deleteOldBuckets(db kvdb.Backend) error {
        // locators records all the chan buckets found in the database.
        var locators []*updateLocator

        // reader is a helper closure that saves the locator found. Each
        // locator is relatively small(33+32+36+8=109 bytes), assuming 1 GB of
        // ram we can fit roughly 10 million records. Since each record
        // corresponds to a channel, we should have more than enough memory to
        // read them all.
        reader := func(_ kvdb.RwBucket, l *updateLocator) error { // nolint:unparam
                locators = append(locators, l)
                return nil
        }

        // remover is a helper closure that removes the old revocation log
        // bucket under the specified chan bucket by the given locator.
        remover := func(rootBucket kvdb.RwBucket, l *updateLocator) error {
                chanBucket, err := l.locateChanBucket(rootBucket)
                if err != nil {
                        return err
                }

                return chanBucket.DeleteNestedBucket(
                        revocationLogBucketDeprecated,
                )
        }

        // Perform the deletion in one db transaction. This should not cause
        // any memory issue as the deletion doesn't load any data from the
        // buckets.
        return kvdb.Update(db, func(tx kvdb.RwTx) error {
                openChanBucket := tx.ReadWriteBucket(openChannelBucket)

                // Exit early if there's no bucket.
                if openChanBucket == nil {
                        return nil
                }

                // Iterate the buckets to find all the locators.
                err := iterateBuckets(openChanBucket, nil, reader)
                if err != nil {
                        return err
                }

                // Iterate the locators and delete all the old revocation log
                // buckets.
                for _, l := range locators {
                        err := remover(openChanBucket, l)
                        // If the bucket doesn't exist, we can exit safety.
                        if err != nil && err != kvdb.ErrBucketNotFound {
                                return err
                        }
                }

                return nil
        }, func() {})
}

// writeRevocationLogs unwraps the entryMap and writes the new revocation logs.
func writeRevocationLogs(openChanBucket kvdb.RwBucket,
        entryMap logEntries) error {

        for locator, logs := range entryMap {
                // Find the channel bucket.
                chanBucket, err := locator.locateChanBucket(openChanBucket)
                if err != nil {
                        return fmt.Errorf("locateChanBucket err: %w", err)
                }

                // Create the new log bucket.
                logBucket, err := chanBucket.CreateBucketIfNotExists(
                        revocationLogBucket,
                )
                if err != nil {
                        return fmt.Errorf("create log bucket err: %w", err)
                }

                // Write the new logs.
                for _, entry := range logs {
                        var b bytes.Buffer
                        err := serializeRevocationLog(&b, entry.log)
                        if err != nil {
                                return err
                        }

                        logEntrykey := mig24.MakeLogKey(entry.commitHeight)
                        err = logBucket.Put(logEntrykey[:], b.Bytes())
                        if err != nil {
                                return fmt.Errorf("putRevocationLog err: %w",
                                        err)
                        }
                }
        }

        return nil
}

// logMigrationStat reads the buckets to provide stats over current migration
// progress. The returned values are the numbers of total records and already
// migrated records.
func logMigrationStat(db kvdb.Backend) (uint64, uint64, error) {
        var (
                err error

                // total is the number of total records.
                total uint64

                // unmigrated is the number of unmigrated records.
                unmigrated uint64
        )

        err = kvdb.Update(db, func(tx kvdb.RwTx) error {
                total, unmigrated, err = fetchLogStats(tx)
                return err
        }, func() {})

        log.Debugf("Total logs=%d, unmigrated=%d", total, unmigrated)
        return total, total - unmigrated, err
}

// fetchLogStats iterates all the chan buckets to provide stats about the logs.
// The returned values are num of total records, and num of un-migrated
// records.
func fetchLogStats(tx kvdb.RwTx) (uint64, uint64, error) {
        var (
                total           uint64
                totalUnmigrated uint64
        )

        openChanBucket := tx.ReadWriteBucket(openChannelBucket)

        // If no bucket is found, we can exit early.
        if openChanBucket == nil {
                return 0, 0, fmt.Errorf("root bucket not found")
        }

        // counter is a helper closure used to count the number of records
        // based on the given bucket.
        counter := func(chanBucket kvdb.RwBucket, bucket []byte) uint64 {
                // Read the sub-bucket level 4.
                logBucket := chanBucket.NestedReadBucket(bucket)

                // Exit early if we don't have the bucket.
                if logBucket == nil {
                        return 0
                }

                // Jump to the end of the cursor.
                key, _ := logBucket.ReadCursor().Last()

                // Since the CommitHeight is a zero-based monotonically
                // increased index, its value plus one reflects the total
                // records under this chan bucket.
                lastHeight := binary.BigEndian.Uint64(key) + 1

                return lastHeight
        }

        // countTotal is a callback function used to count the total number of
        // records.
        countTotal := func(chanBucket kvdb.RwBucket, l *updateLocator) error {
                total += counter(chanBucket, revocationLogBucketDeprecated)
                return nil
        }

        // countUnmigrated is a callback function used to count the total
        // number of un-migrated records.
        countUnmigrated := func(chanBucket kvdb.RwBucket,
                l *updateLocator) error {

                totalUnmigrated += counter(
                        chanBucket, revocationLogBucketDeprecated,
                )
                return nil
        }

        // Locate the next migration height.
        locator, err := locateNextUpdateNum(openChanBucket)
        if err != nil {
                return 0, 0, fmt.Errorf("locator got error: %w", err)
        }

        // If the returned locator is not nil, we still have un-migrated
        // records so we need to count them. Otherwise we've done migrating the
        // logs.
        if locator != nil {
                err = iterateBuckets(openChanBucket, locator, countUnmigrated)
                if err != nil {
                        return 0, 0, err
                }
        }

        // Count the total number of records by supplying a nil locator.
        err = iterateBuckets(openChanBucket, nil, countTotal)
        if err != nil {
                return 0, 0, err
        }

        return total, totalUnmigrated, err
}

// logEntry houses the info needed to write a new revocation log.
type logEntry struct {
        log          *RevocationLog
        commitHeight uint64
        ourIndex     uint32
        theirIndex   uint32
        locator      *updateLocator
}

// logEntries maps a bucket locator to a list of entries under that bucket.
type logEntries map[*updateLocator][]*logEntry

// result is made of two channels that's used to send back the constructed new
// revocation log or an error.
type result struct {
        newLog  chan *logEntry
        errChan chan error
}

// readOldRevocationLogs finds a list of old revocation logs and converts them
// into the new revocation logs.
func readOldRevocationLogs(openChanBucket kvdb.RwBucket,
        locator *updateLocator, cfg MigrateRevLogConfig) (logEntries, error) {

        entries := make(logEntries)
        results := make([]*result, 0)

        var wg sync.WaitGroup

        // collectLogs is a helper closure that reads all newly created
        // revocation logs sent over the result channels.
        //
        // NOTE: the order of the logs cannot be guaranteed, which is fine as
        // boltdb will take care of the orders when saving them.
        collectLogs := func() error {
                wg.Wait()

                for _, r := range results {
                        select {
                        case entry := <-r.newLog:
                                entries[entry.locator] = append(
                                        entries[entry.locator], entry,
                                )

                        case err := <-r.errChan:
                                return err
                        }
                }

                return nil
        }

        // createLog is a helper closure that constructs a new revocation log.
        //
        // NOTE: used as a goroutine.
        createLog := func(chanState *mig26.OpenChannel,
                c mig.ChannelCommitment, l *updateLocator, r *result) {

                defer wg.Done()

                // Find the output indexes.
                ourIndex, theirIndex, err := findOutputIndexes(chanState, &c)
                if err != nil {
                        r.errChan <- err
                }

                // Convert the old logs into the new logs. We do this early in
                // the read tx so the old large revocation log can be set to
                // nil here so save us some memory space.
                newLog, err := convertRevocationLog(
                        &c, ourIndex, theirIndex, cfg.GetNoAmountData(),
                )
                if err != nil {
                        r.errChan <- err
                }
                // Create the entry that will be used to create the new log.
                entry := &logEntry{
                        log:          newLog,
                        commitHeight: c.CommitHeight,
                        ourIndex:     ourIndex,
                        theirIndex:   theirIndex,
                        locator:      l,
                }

                r.newLog <- entry
        }

        // innerCb is the stepping function used when iterating the old log
        // bucket.
        innerCb := func(chanState *mig26.OpenChannel, l *updateLocator,
                _, v []byte) error {

                reader := bytes.NewReader(v)
                c, err := mig.DeserializeChanCommit(reader)
                if err != nil {
                        return err
                }

                r := &result{
                        newLog:  make(chan *logEntry, 1),
                        errChan: make(chan error, 1),
                }
                results = append(results, r)

                // We perform the log creation in a goroutine as it takes some
                // time to compute and find output indexes.
                wg.Add(1)
                go createLog(chanState, c, l, r)

                // Check the records read so far and signals exit when we've
                // reached our memory cap.
                if len(results) >= recordsPerTx {
                        return errExit
                }

                return nil
        }

        // cb is the callback function to be used when iterating the buckets.
        cb := func(chanBucket kvdb.RwBucket, l *updateLocator) error {
                // Read the open channel.
                c := &mig26.OpenChannel{}
                err := mig26.FetchChanInfo(chanBucket, c, false)
                if err != nil {
                        return fmt.Errorf("unable to fetch chan info: %w", err)
                }

                err = fetchChanRevocationState(chanBucket, c)
                if err != nil {
                        return fmt.Errorf("unable to fetch revocation "+
                                "state: %v", err)
                }

                // Read the sub-bucket level 4.
                logBucket := chanBucket.NestedReadBucket(
                        revocationLogBucketDeprecated,
                )
                // Exit early if we don't have the old bucket.
                if logBucket == nil {
                        return nil
                }

                // Init the map key when needed.
                _, ok := entries[l]
                if !ok {
                        entries[l] = make([]*logEntry, 0, recordsPerTx)
                }

                return iterator(
                        logBucket, locator.nextHeight,
                        func(k, v []byte) error {
                                // Reset the nextHeight for following chan
                                // buckets.
                                locator.nextHeight = nil
                                return innerCb(c, l, k, v)
                        },
                )
        }

        err := iterateBuckets(openChanBucket, locator, cb)
        // If there's an error and it's not exit signal, we won't collect the
        // logs from the result channels.
        if err != nil && err != errExit {
                return nil, err
        }

        // Otherwise, collect the logs.
        err = collectLogs()

        return entries, err
}

// convertRevocationLog uses the fields `CommitTx` and `Htlcs` from a
// ChannelCommitment to construct a revocation log entry.
func convertRevocationLog(commit *mig.ChannelCommitment,
        ourOutputIndex, theirOutputIndex uint32,
        noAmtData bool) (*RevocationLog, error) {

        // Sanity check that the output indexes can be safely converted.
        if ourOutputIndex > math.MaxUint16 {
                return nil, ErrOutputIndexTooBig
        }
        if theirOutputIndex > math.MaxUint16 {
                return nil, ErrOutputIndexTooBig
        }

        rl := &RevocationLog{
                OurOutputIndex:   uint16(ourOutputIndex),
                TheirOutputIndex: uint16(theirOutputIndex),
                CommitTxHash:     commit.CommitTx.TxHash(),
                HTLCEntries:      make([]*HTLCEntry, 0, len(commit.Htlcs)),
        }

        if !noAmtData {
                rl.TheirBalance = &commit.RemoteBalance
                rl.OurBalance = &commit.LocalBalance
        }

        for _, htlc := range commit.Htlcs {
                // Skip dust HTLCs.
                if htlc.OutputIndex < 0 {
                        continue
                }

                // Sanity check that the output indexes can be safely
                // converted.
                if htlc.OutputIndex > math.MaxUint16 {
                        return nil, ErrOutputIndexTooBig
                }

                entry := &HTLCEntry{
                        RHash:         htlc.RHash,
                        RefundTimeout: htlc.RefundTimeout,
                        Incoming:      htlc.Incoming,
                        OutputIndex:   uint16(htlc.OutputIndex),
                        Amt:           htlc.Amt.ToSatoshis(),
                }
                rl.HTLCEntries = append(rl.HTLCEntries, entry)
        }

        return rl, nil
}

// validateMigration checks that the data saved in the new buckets match those
// saved in the old buckets. It does so by checking the last keys saved in both
// buckets can match, given the assumption that the `CommitHeight` is
// monotonically increased value so the last key represents the total number of
// records saved.
func validateMigration(tx kvdb.RwTx) error {
        openChanBucket := tx.ReadWriteBucket(openChannelBucket)

        // If no bucket is found, we can exit early.
        if openChanBucket == nil {
                return nil
        }

        // exitWithErr is a helper closure that prepends an error message with
        // the locator info.
        exitWithErr := func(l *updateLocator, msg string) error {
                return fmt.Errorf("unmatched records found under <nodePub=%x"+
                        ", chainHash=%x, fundingOutpoint=%x>: %v", l.nodePub,
                        l.chainHash, l.fundingOutpoint, msg)
        }

        // cb is the callback function to be used when iterating the buckets.
        cb := func(chanBucket kvdb.RwBucket, l *updateLocator) error {
                // Read both the old and new revocation log buckets.
                oldBucket := chanBucket.NestedReadBucket(
                        revocationLogBucketDeprecated,
                )
                newBucket := chanBucket.NestedReadBucket(revocationLogBucket)

                // Exit early if the old bucket is nil.
                //
                // NOTE: the new bucket may not be nil here as new logs might
                // have been created using lnd@v0.15.0.
                if oldBucket == nil {
                        return nil
                }

                // Return an error if the expected new bucket cannot be found.
                if newBucket == nil {
                        return exitWithErr(l, "expected new bucket")
                }

                // Acquire the cursors.
                oldCursor := oldBucket.ReadCursor()
                newCursor := newBucket.ReadCursor()

                // Jump to the end of the cursors to do a quick check.
                newKey, _ := oldCursor.Last()
                oldKey, _ := newCursor.Last()

                // We expected the CommitHeights to be matched for nodes prior
                // to v0.15.0.
                if bytes.Equal(newKey, oldKey) {
                        return nil
                }

                // If the keys do not match, it's likely the node is running
                // v0.15.0 and have new logs created. In this case, we will
                // validate that every record in the old bucket can be found in
                // the new bucket.
                oldKey, _ = oldCursor.First()

                for {
                        // Try to locate the old key in the new bucket and we
                        // expect it to be found.
                        newKey, _ := newCursor.Seek(oldKey)

                        // If the old key is not found in the new bucket,
                        // return an error.
                        //
                        // NOTE: because Seek will return the next key when the
                        // passed key cannot be found, we need to compare the
                        // keys to deicde whether the old key is found or not.
                        if !bytes.Equal(newKey, oldKey) {
                                errMsg := fmt.Sprintf("old bucket has "+
                                        "CommitHeight=%v cannot be found in "+
                                        "new bucket", oldKey)
                                return exitWithErr(l, errMsg)
                        }

                        // Otherwise, keep iterating the old bucket.
                        oldKey, _ = oldCursor.Next()

                        // If we've done iterating, all keys have been matched
                        // and we can safely exit.
                        if oldKey == nil {
                                return nil
                        }
                }
        }

        return iterateBuckets(openChanBucket, nil, cb)
}

lightningnetwork / lnd / 15561477203

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