12349698563

Committed 16 Dec 2024 09:29AM UTC coverage: 58.55% (-0.09%) from 58.636%

Build # 12349698563

Build Type

Pull #9357

github

Committed by

GeorgeTsagk

Commit Message

contractcourt: include custom records on replayed htlc

When notifying the invoice registry for an exit hop htlc we also want to
include its custom records. The channelLink, the other caller of this
method, already populates this field. So we make sure the contest
resolver does so too.

Pull Request Pull Request #9357: contractcourt: include custom records on replayed htlc

Run Details

2 of 2 new or added lines in 1 file covered. (100.0%)

262 existing lines in 24 files now uncovered.

134243 of 229278 relevant lines covered (58.55%)

19277.11 hits per line

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

82.84

/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 / 12349698563

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