15951470896

Committed 29 Jun 2025 04:23AM UTC coverage: 67.594% (-0.01%) from 67.606%

Build # 15951470896

Build Type

Pull #9751

github

Committed by

web-flow

Commit Message

Merge 599d9b051 into 6290edf14

Pull Request Pull Request #9751: multi: update Go to 1.23.10 and update some packages

Run Details

135088 of 199851 relevant lines covered (67.59%)

21909.44 hits per line

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

73.9

/watchtower/wtdb/migration4/client_db.go

package migration4

import (
        "bytes"
        "encoding/binary"
        "errors"
        "fmt"

        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/tlv"
)

var (
        // cChanDetailsBkt is a top-level bucket storing:
        //   channel-id => cChannelSummary -> encoded ClientChanSummary.
        //                 => cChanDBID -> db-assigned-id
        cChanDetailsBkt = []byte("client-channel-detail-bucket")

        // cChanDBID is a key used in the cChanDetailsBkt to store the
        // db-assigned-id of a channel.
        cChanDBID = []byte("client-channel-db-id")

        // cSessionBkt is a top-level bucket storing:
        //   session-id => cSessionBody -> encoded ClientSessionBody
        //              => cSessionCommits => seqnum -> encoded CommittedUpdate
        //              => cSessionAcks => seqnum -> encoded BackupID
        cSessionBkt = []byte("client-session-bucket")

        // cSessionAcks is a sub-bucket of cSessionBkt storing:
        //    seqnum -> encoded BackupID.
        cSessionAcks = []byte("client-session-acks")

        // cSessionAckRangeIndex is a sub-bucket of cSessionBkt storing:
        //    chan-id => start -> end
        cSessionAckRangeIndex = []byte("client-session-ack-range-index")

        // ErrUninitializedDB signals that top-level buckets for the database
        // have not been initialized.
        ErrUninitializedDB = errors.New("db not initialized")

        // ErrClientSessionNotFound signals that the requested client session
        // was not found in the database.
        ErrClientSessionNotFound = errors.New("client session not found")

        // ErrCorruptChanDetails signals that the clients channel detail's
        // on-disk structure deviates from what is expected.
        ErrCorruptChanDetails = errors.New("channel details corrupted")

        // ErrChannelNotRegistered signals a channel has not yet been registered
        // in the client database.
        ErrChannelNotRegistered = errors.New("channel not registered")

        // byteOrder is the default endianness used when serializing integers.
        byteOrder = binary.BigEndian

        // errExit is an error used to signal that the sessionIterator should
        // exit.
        errExit = errors.New("the exit condition has been met")
)

// DefaultSessionsPerTx is the default number of sessions that should be
// migrated per db transaction.
const DefaultSessionsPerTx = 5000

// MigrateAckedUpdates migrates the tower client DB. It takes the individual
// Acked Updates that are stored for each session and re-stores them using the
// RangeIndex representation.
func MigrateAckedUpdates(sessionsPerTx int) func(kvdb.Backend) error {
        return func(db kvdb.Backend) error {
                log.Infof("Migrating the tower client db to move all Acked " +
                        "Updates to the new Range Index representation.")

                // Migrate the old acked-updates.
                err := migrateAckedUpdates(db, sessionsPerTx)
                if err != nil {
                        return fmt.Errorf("migration failed: %w", err)
                }

                log.Infof("Migrating old session acked updates finished, now " +
                        "checking the migration results...")

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

                // Delete old acked updates.
                err = kvdb.Update(db, deleteOldAckedUpdates, func() {})
                if err != nil {
                        return fmt.Errorf("failed to delete old acked "+
                                "updates: %w", err)
                }

                return nil
        }
}

// migrateAckedUpdates migrates the acked updates of each session in the
// wtclient db into the new RangeIndex form. This is done over multiple db
// transactions in order to prevent the migration from taking up too much RAM.
// The sessionsPerTx parameter can be used to set the maximum number of sessions
// that should be migrated per transaction.
func migrateAckedUpdates(db kvdb.Backend, sessionsPerTx int) error {
        // Get migration progress stats.
        total, migrated, err := logMigrationStats(db)
        if err != nil {
                return err
        }
        log.Infof("Total sessions=%d, migrated=%d", total, migrated)

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

        var (
                finished bool
                startKey []byte
        )
        for {
                // Process the migration.
                err = kvdb.Update(db, func(tx kvdb.RwTx) error {
                        startKey, finished, err = processMigration(
                                tx, startKey, sessionsPerTx,
                        )

                        return err
                }, func() {})
                if err != nil {
                        return err
                }

                if finished {
                        break
                }

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

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

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

        return nil
}

func validateMigration(tx kvdb.RTx) error {
        mainSessionsBkt := tx.ReadBucket(cSessionBkt)
        if mainSessionsBkt == nil {
                return ErrUninitializedDB
        }

        chanDetailsBkt := tx.ReadBucket(cChanDetailsBkt)
        if chanDetailsBkt == nil {
                return ErrUninitializedDB
        }

        return mainSessionsBkt.ForEach(func(sessID, _ []byte) error {
                // Get the bucket for this particular session.
                sessionBkt := mainSessionsBkt.NestedReadBucket(sessID)
                if sessionBkt == nil {
                        return ErrClientSessionNotFound
                }

                // Get the bucket where any old acked updates would be stored.
                oldAcksBucket := sessionBkt.NestedReadBucket(cSessionAcks)

                // Get the bucket where any new acked updates would be stored.
                newAcksBucket := sessionBkt.NestedReadBucket(
                        cSessionAckRangeIndex,
                )

                switch {
                // If both the old and new acked updates buckets are nil, then
                // we can safely skip this session.
                case oldAcksBucket == nil && newAcksBucket == nil:
                        return nil

                case oldAcksBucket == nil:
                        return fmt.Errorf("no old acks but do have new acks")

                case newAcksBucket == nil:
                        return fmt.Errorf("no new acks but have old acks")

                default:
                }

                // Collect acked ranges for this session.
                ackedRanges := make(map[uint64]*RangeIndex)
                err := newAcksBucket.ForEach(func(dbChanID, _ []byte) error {
                        rangeIndexBkt := newAcksBucket.NestedReadBucket(
                                dbChanID,
                        )
                        if rangeIndexBkt == nil {
                                return fmt.Errorf("no acked updates bucket "+
                                        "found for channel %x", dbChanID)
                        }

                        // Read acked ranges from new bucket.
                        ri, err := readRangeIndex(rangeIndexBkt)
                        if err != nil {
                                return err
                        }

                        dbChanIDNum, err := readBigSize(dbChanID)
                        if err != nil {
                                return err
                        }

                        ackedRanges[dbChanIDNum] = ri

                        return nil
                })
                if err != nil {
                        return err
                }

                // Now we will iterate through each of the old acked updates and
                // make sure that the update appears in the new bucket.
                return oldAcksBucket.ForEach(func(_, v []byte) error {
                        var backupID BackupID
                        err := backupID.Decode(bytes.NewReader(v))
                        if err != nil {
                                return err
                        }

                        dbChanID, _, err := getDBChanID(
                                chanDetailsBkt, backupID.ChanID,
                        )
                        if err != nil {
                                return err
                        }

                        index, ok := ackedRanges[dbChanID]
                        if !ok {
                                return fmt.Errorf("no index found for this " +
                                        "channel")
                        }

                        if !index.IsInIndex(backupID.CommitHeight) {
                                return fmt.Errorf("commit height not found " +
                                        "in index")
                        }

                        return nil
                })
        })
}

func readRangeIndex(rangesBkt kvdb.RBucket) (*RangeIndex, error) {
        ranges := make(map[uint64]uint64)
        err := rangesBkt.ForEach(func(k, v []byte) error {
                start, err := readBigSize(k)
                if err != nil {
                        return err
                }

                end, err := readBigSize(v)
                if err != nil {
                        return err
                }

                ranges[start] = end

                return nil
        })
        if err != nil {
                return nil, err
        }

        return NewRangeIndex(ranges, WithSerializeUint64Fn(writeBigSize))
}

func deleteOldAckedUpdates(tx kvdb.RwTx) error {
        mainSessionsBkt := tx.ReadWriteBucket(cSessionBkt)
        if mainSessionsBkt == nil {
                return ErrUninitializedDB
        }

        return mainSessionsBkt.ForEach(func(sessID, _ []byte) error {
                // Get the bucket for this particular session.
                sessionBkt := mainSessionsBkt.NestedReadWriteBucket(
                        sessID,
                )
                if sessionBkt == nil {
                        return ErrClientSessionNotFound
                }

                // Get the bucket where any old acked updates would be stored.
                oldAcksBucket := sessionBkt.NestedReadBucket(cSessionAcks)
                if oldAcksBucket == nil {
                        return nil
                }

                // Now that we have read everything that we need to from
                // the cSessionAcks sub-bucket, we can delete it.
                return sessionBkt.DeleteNestedBucket(cSessionAcks)
        })
}

// processMigration uses the given transaction to perform a maximum of
// sessionsPerTx session migrations. If startKey is non-nil, it is used to
// determine the first session to start the migration at. The first return
// item is the key of the last session that was migrated successfully and the
// boolean is true if there are no more sessions left to migrate.
func processMigration(tx kvdb.RwTx, startKey []byte, sessionsPerTx int) ([]byte,
        bool, error) {

        chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
        if chanDetailsBkt == nil {
                return nil, false, ErrUninitializedDB
        }

        // sessionCount keeps track of the number of sessions that have been
        // migrated under the current db transaction.
        var sessionCount int

        // migrateSessionCB is a callback function that calls migrateSession
        // in order to migrate a single session. Upon success, the sessionCount
        // is incremented and is then compared against sessionsPerTx to
        // determine if we should continue migrating more sessions in this db
        // transaction.
        migrateSessionCB := func(sessionBkt kvdb.RwBucket) error {
                err := migrateSession(chanDetailsBkt, sessionBkt)
                if err != nil {
                        return err
                }

                sessionCount++

                // If we have migrated sessionsPerTx sessions in this tx, then
                // we return errExit in order to signal that this tx should be
                // committed and the migration should be continued in a new
                // transaction.
                if sessionCount >= sessionsPerTx {
                        return errExit
                }

                return nil
        }

        // Starting at startKey, iterate over the sessions in the db and migrate
        // them until either all are migrated or until the errExit signal is
        // received.
        lastKey, err := sessionIterator(tx, startKey, migrateSessionCB)
        if err != nil && errors.Is(err, errExit) {
                return lastKey, false, nil
        } else if err != nil {
                return nil, false, err
        }

        // The migration is complete.
        return nil, true, nil
}

// migrateSession migrates a single session's acked-updates to the new
// RangeIndex form.
func migrateSession(chanDetailsBkt kvdb.RBucket,
        sessionBkt kvdb.RwBucket) error {

        // Get the existing cSessionAcks bucket. If there is no such bucket,
        // then there are no acked-updates to migrate for this session.
        sessionAcks := sessionBkt.NestedReadBucket(cSessionAcks)
        if sessionAcks == nil {
                return nil
        }

        // If there is already a new cSessionAckedRangeIndex bucket, then this
        // session has already been migrated.
        sessionAckRangesBkt := sessionBkt.NestedReadBucket(
                cSessionAckRangeIndex,
        )
        if sessionAckRangesBkt != nil {
                return nil
        }

        // Otherwise, we will iterate over each of the acked-updates, and we
        // will construct a new RangeIndex for each channel.
        m := make(map[ChannelID]*RangeIndex)
        if err := sessionAcks.ForEach(func(_, v []byte) error {
                var backupID BackupID
                err := backupID.Decode(bytes.NewReader(v))
                if err != nil {
                        return err
                }

                if _, ok := m[backupID.ChanID]; !ok {
                        index, err := NewRangeIndex(nil)
                        if err != nil {
                                return err
                        }

                        m[backupID.ChanID] = index
                }

                return m[backupID.ChanID].Add(backupID.CommitHeight, nil)
        }); err != nil {
                return err
        }

        // Create a new sub-bucket that will be used to store the new RangeIndex
        // representation of the acked updates.
        ackRangeBkt, err := sessionBkt.CreateBucket(cSessionAckRangeIndex)
        if err != nil {
                return err
        }

        // Iterate over each of the new range indexes that we will add for this
        // session.
        for chanID, rangeIndex := range m {
                // Get db chanID.
                chanDetails := chanDetailsBkt.NestedReadBucket(chanID[:])
                if chanDetails == nil {
                        return ErrCorruptChanDetails
                }

                // Create a sub-bucket for this channel using the db-assigned ID
                // for the channel.
                dbChanID := chanDetails.Get(cChanDBID)
                chanAcksBkt, err := ackRangeBkt.CreateBucket(dbChanID)
                if err != nil {
                        return err
                }

                // Iterate over the range pairs that we need to add to the DB.
                for k, v := range rangeIndex.GetAllRanges() {
                        start, err := writeBigSize(k)
                        if err != nil {
                                return err
                        }

                        end, err := writeBigSize(v)
                        if err != nil {
                                return err
                        }

                        err = chanAcksBkt.Put(start, end)
                        if err != nil {
                                return err
                        }
                }
        }

        return nil
}

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

        err = kvdb.View(db, func(tx kvdb.RTx) error {
                total, unmigrated, err = getMigrationStats(tx)

                return err
        }, func() {})

        log.Debugf("Total sessions=%d, unmigrated=%d", total, unmigrated)

        return total, total - unmigrated, err
}

// getMigrationStats iterates over all sessions. It counts the total number of
// sessions as well as the total number of unmigrated sessions.
func getMigrationStats(tx kvdb.RTx) (uint64, uint64, error) {
        var (
                total      uint64
                unmigrated uint64
        )

        // Get sessions bucket.
        mainSessionsBkt := tx.ReadBucket(cSessionBkt)
        if mainSessionsBkt == nil {
                return 0, 0, ErrUninitializedDB
        }

        // Iterate over each session ID in the bucket.
        err := mainSessionsBkt.ForEach(func(sessID, _ []byte) error {
                // Get the bucket for this particular session.
                sessionBkt := mainSessionsBkt.NestedReadBucket(sessID)
                if sessionBkt == nil {
                        return ErrClientSessionNotFound
                }

                total++

                // Get the cSessionAckRangeIndex bucket.
                sessionAcksBkt := sessionBkt.NestedReadBucket(cSessionAcks)

                // Get the cSessionAckRangeIndex bucket.
                sessionAckRangesBkt := sessionBkt.NestedReadBucket(
                        cSessionAckRangeIndex,
                )

                // If both buckets do not exist, then this session is empty and
                // does not need to be migrated.
                if sessionAckRangesBkt == nil && sessionAcksBkt == nil {
                        return nil
                }

                // If the sessionAckRangesBkt is not nil, then the session has
                // already been migrated.
                if sessionAckRangesBkt != nil {
                        return nil
                }

                // Else the session has not yet been migrated.
                unmigrated++

                return nil
        })
        if err != nil {
                return 0, 0, err
        }

        return total, unmigrated, nil
}

// getDBChanID returns the db-assigned channel ID for the given real channel ID.
// It returns both the uint64 and byte representation.
func getDBChanID(chanDetailsBkt kvdb.RBucket, chanID ChannelID) (uint64,
        []byte, error) {

        chanDetails := chanDetailsBkt.NestedReadBucket(chanID[:])
        if chanDetails == nil {
                return 0, nil, ErrChannelNotRegistered
        }

        idBytes := chanDetails.Get(cChanDBID)
        if len(idBytes) == 0 {
                return 0, nil, fmt.Errorf("no db-assigned ID found for "+
                        "channel ID %s", chanID)
        }

        id, err := readBigSize(idBytes)
        if err != nil {
                return 0, nil, err
        }

        return id, idBytes, nil
}

// callback defines a type that's used by the sessionIterator.
type callback func(bkt kvdb.RwBucket) error

// sessionIterator is a helper function that iterates over the main sessions
// bucket and performs the callback function on each individual session. If a
// seeker is specified, it will move the cursor to the given position otherwise
// it will start from the first item.
func sessionIterator(tx kvdb.RwTx, seeker []byte, cb callback) ([]byte, error) {
        // Get sessions bucket.
        mainSessionsBkt := tx.ReadWriteBucket(cSessionBkt)
        if mainSessionsBkt == nil {
                return nil, ErrUninitializedDB
        }

        c := mainSessionsBkt.ReadCursor()
        k, _ := c.First()

        // Move the cursor to the specified position if seeker is non-nil.
        if seeker != nil {
                k, _ = c.Seek(seeker)
        }

        // Start the iteration and exit on condition.
        for k := k; k != nil; k, _ = c.Next() {
                // Get the bucket for this particular session.
                bkt := mainSessionsBkt.NestedReadWriteBucket(k)
                if bkt == nil {
                        return nil, ErrClientSessionNotFound
                }

                // Call the callback function with the session's bucket.
                if err := cb(bkt); err != nil {
                        // return k, err
                        lastIndex := make([]byte, len(k))
                        copy(lastIndex, k)
                        return lastIndex, err
                }
        }

        return nil, nil
}

// writeBigSize will encode the given uint64 as a BigSize byte slice.
func writeBigSize(i uint64) ([]byte, error) {
        var b bytes.Buffer
        err := tlv.WriteVarInt(&b, i, &[8]byte{})
        if err != nil {
                return nil, err
        }

        return b.Bytes(), nil
}

// readBigSize converts the given byte slice into a uint64 and assumes that the
// bytes slice is using BigSize encoding.
func readBigSize(b []byte) (uint64, error) {
        r := bytes.NewReader(b)
        i, err := tlv.ReadVarInt(r, &[8]byte{})
        if err != nil {
                return 0, err
        }

        return i, nil
}

1	package migration4
2
3	import (
4	"bytes"
5	"encoding/binary"
6	"errors"
7	"fmt"
8
9	"github.com/lightningnetwork/lnd/kvdb"
10	"github.com/lightningnetwork/lnd/tlv"
11	)
12
13	var (
14	// cChanDetailsBkt is a top-level bucket storing:
15	// channel-id => cChannelSummary -> encoded ClientChanSummary.
16	// => cChanDBID -> db-assigned-id
17	cChanDetailsBkt = []byte("client-channel-detail-bucket")
18
19	// cChanDBID is a key used in the cChanDetailsBkt to store the
20	// db-assigned-id of a channel.
21	cChanDBID = []byte("client-channel-db-id")
22
23	// cSessionBkt is a top-level bucket storing:
24	// session-id => cSessionBody -> encoded ClientSessionBody
25	// => cSessionCommits => seqnum -> encoded CommittedUpdate
26	// => cSessionAcks => seqnum -> encoded BackupID
27	cSessionBkt = []byte("client-session-bucket")
28
29	// cSessionAcks is a sub-bucket of cSessionBkt storing:
30	// seqnum -> encoded BackupID.
31	cSessionAcks = []byte("client-session-acks")
32
33	// cSessionAckRangeIndex is a sub-bucket of cSessionBkt storing:
34	// chan-id => start -> end
35	cSessionAckRangeIndex = []byte("client-session-ack-range-index")
36
37	// ErrUninitializedDB signals that top-level buckets for the database
38	// have not been initialized.
39	ErrUninitializedDB = errors.New("db not initialized")
40
41	// ErrClientSessionNotFound signals that the requested client session
42	// was not found in the database.
43	ErrClientSessionNotFound = errors.New("client session not found")
44
45	// ErrCorruptChanDetails signals that the clients channel detail's
46	// on-disk structure deviates from what is expected.
47	ErrCorruptChanDetails = errors.New("channel details corrupted")
48
49	// ErrChannelNotRegistered signals a channel has not yet been registered
50	// in the client database.
51	ErrChannelNotRegistered = errors.New("channel not registered")
52
53	// byteOrder is the default endianness used when serializing integers.
54	byteOrder = binary.BigEndian
55
56	// errExit is an error used to signal that the sessionIterator should
57	// exit.
58	errExit = errors.New("the exit condition has been met")
59	)
60
61	// DefaultSessionsPerTx is the default number of sessions that should be
62	// migrated per db transaction.
63	const DefaultSessionsPerTx = 5000
64
65	// MigrateAckedUpdates migrates the tower client DB. It takes the individual
66	// Acked Updates that are stored for each session and re-stores them using the
67	// RangeIndex representation.
68	func MigrateAckedUpdates(sessionsPerTx int) func(kvdb.Backend) error {	20✔
69	return func(db kvdb.Backend) error {	24✔
70	log.Infof("Migrating the tower client db to move all Acked " +	4✔
71	"Updates to the new Range Index representation.")	4✔
72		4✔
73	// Migrate the old acked-updates.	4✔
74	err := migrateAckedUpdates(db, sessionsPerTx)	4✔
75	if err != nil {	5✔
76	return fmt.Errorf("migration failed: %w", err)	1✔
77	}	1✔
78
79	log.Infof("Migrating old session acked updates finished, now " +	3✔
80	"checking the migration results...")	3✔
81		3✔
82	// Before we can safety delete the old buckets, we perform a	3✔
83	// check to make sure the sessions have been migrated as	3✔
84	// expected.	3✔
85	err = kvdb.View(db, validateMigration, func() {})	6✔
86	if err != nil {	3✔
87	return fmt.Errorf("validate migration failed: %w", err)	×
88	}	×
89
90	// Delete old acked updates.
91	err = kvdb.Update(db, deleteOldAckedUpdates, func() {})	6✔
92	if err != nil {	3✔
93	return fmt.Errorf("failed to delete old acked "+	×
94	"updates: %w", err)	×
95	}	×
96
97	return nil	3✔
98	}
99	}
100
101	// migrateAckedUpdates migrates the acked updates of each session in the
102	// wtclient db into the new RangeIndex form. This is done over multiple db
103	// transactions in order to prevent the migration from taking up too much RAM.
104	// The sessionsPerTx parameter can be used to set the maximum number of sessions
105	// that should be migrated per transaction.
106	func migrateAckedUpdates(db kvdb.Backend, sessionsPerTx int) error {	4✔
107	// Get migration progress stats.	4✔
108	total, migrated, err := logMigrationStats(db)	4✔
109	if err != nil {	5✔
110	return err	1✔
111	}	1✔
112	log.Infof("Total sessions=%d, migrated=%d", total, migrated)	3✔
113		3✔
114	// Exit early if the old session acked updates have already been	3✔
115	// migrated and deleted.	3✔
116	if total == 0 {	4✔
117	log.Info("Migration already finished!")	1✔
118	return nil	1✔
119	}	1✔
120
121	var (	2✔
122	finished bool	2✔
123	startKey []byte	2✔
124	)	2✔
125	for {	6✔
126	// Process the migration.	4✔
127	err = kvdb.Update(db, func(tx kvdb.RwTx) error {	8✔
128	startKey, finished, err = processMigration(	4✔
129	tx, startKey, sessionsPerTx,	4✔
130	)	4✔
131		4✔
132	return err	4✔
133	}, func() {})	8✔
134	if err != nil {	4✔
135	return err	×
136	}	×
137
138	if finished {	4✔
139	break	×
140	}
141
142	// Each time we finished the above process, we'd read the stats
143	// again to understand the current progress.
144	total, migrated, err = logMigrationStats(db)	4✔
145	if err != nil {	4✔
146	return err	×
147	}	×
148
149	// Calculate and log the progress if the progress is less than
150	// one hundred percent.
151	progress := float64(migrated) / float64(total) * 100	4✔
152	if progress >= 100 {	6✔
153	break	2✔
154	}
155
156	log.Infof("Migration progress: %.3f%%, still have: %d",	2✔
157	progress, total-migrated)	2✔
158	}
159
160	return nil	2✔
161	}
162
163	func validateMigration(tx kvdb.RTx) error {	3✔
164	mainSessionsBkt := tx.ReadBucket(cSessionBkt)	3✔
165	if mainSessionsBkt == nil {	3✔
166	return ErrUninitializedDB	×
167	}	×
168
169	chanDetailsBkt := tx.ReadBucket(cChanDetailsBkt)	3✔
170	if chanDetailsBkt == nil {	3✔
171	return ErrUninitializedDB	×
172	}	×
173
174	return mainSessionsBkt.ForEach(func(sessID, _ []byte) error {	9✔
175	// Get the bucket for this particular session.	6✔
176	sessionBkt := mainSessionsBkt.NestedReadBucket(sessID)	6✔
177	if sessionBkt == nil {	6✔
178	return ErrClientSessionNotFound	×
179	}	×
180
181	// Get the bucket where any old acked updates would be stored.
182	oldAcksBucket := sessionBkt.NestedReadBucket(cSessionAcks)	6✔
183		6✔
184	// Get the bucket where any new acked updates would be stored.	6✔
185	newAcksBucket := sessionBkt.NestedReadBucket(	6✔
186	cSessionAckRangeIndex,	6✔
187	)	6✔
188		6✔
189	switch {	6✔
190	// If both the old and new acked updates buckets are nil, then
191	// we can safely skip this session.
192	case oldAcksBucket == nil && newAcksBucket == nil:	×
193	return nil	×
194
195	case oldAcksBucket == nil:	×
196	return fmt.Errorf("no old acks but do have new acks")	×
197
198	case newAcksBucket == nil:	×
199	return fmt.Errorf("no new acks but have old acks")	×
200
201	default:	6✔
202	}
203
204	// Collect acked ranges for this session.
205	ackedRanges := make(map[uint64]*RangeIndex)	6✔
206	err := newAcksBucket.ForEach(func(dbChanID, _ []byte) error {	16✔
207	rangeIndexBkt := newAcksBucket.NestedReadBucket(	10✔
208	dbChanID,	10✔
209	)	10✔
210	if rangeIndexBkt == nil {	10✔
211	return fmt.Errorf("no acked updates bucket "+	×
212	"found for channel %x", dbChanID)	×
213	}	×
214
215	// Read acked ranges from new bucket.
216	ri, err := readRangeIndex(rangeIndexBkt)	10✔
217	if err != nil {	10✔
218	return err	×
219	}	×
220
221	dbChanIDNum, err := readBigSize(dbChanID)	10✔
222	if err != nil {	10✔
223	return err	×
224	}	×
225
226	ackedRanges[dbChanIDNum] = ri	10✔
227		10✔
228	return nil	10✔
229	})
230	if err != nil {	6✔
231	return err	×
232	}	×
233
234	// Now we will iterate through each of the old acked updates and
235	// make sure that the update appears in the new bucket.
236	return oldAcksBucket.ForEach(func(_, v []byte) error {	26✔
237	var backupID BackupID	20✔
238	err := backupID.Decode(bytes.NewReader(v))	20✔
239	if err != nil {	20✔
240	return err	×
241	}	×
242
243	dbChanID, _, err := getDBChanID(	20✔
244	chanDetailsBkt, backupID.ChanID,	20✔
245	)	20✔
246	if err != nil {	20✔
247	return err	×
248	}	×
249
250	index, ok := ackedRanges[dbChanID]	20✔
251	if !ok {	20✔
252	return fmt.Errorf("no index found for this " +	×
253	"channel")	×
254	}	×
255
256	if !index.IsInIndex(backupID.CommitHeight) {	20✔
257	return fmt.Errorf("commit height not found " +	×
258	"in index")	×
259	}	×
260
261	return nil	20✔
262	})
263	})
264	}
265
266	func readRangeIndex(rangesBkt kvdb.RBucket) (*RangeIndex, error) {	10✔
267	ranges := make(map[uint64]uint64)	10✔
268	err := rangesBkt.ForEach(func(k, v []byte) error {	22✔
269	start, err := readBigSize(k)	12✔
270	if err != nil {	12✔
271	return err	×
272	}	×
273
274	end, err := readBigSize(v)	12✔
275	if err != nil {	12✔
276	return err	×
277	}	×
278
279	ranges[start] = end	12✔
280		12✔
281	return nil	12✔
282	})
283	if err != nil {	10✔
284	return nil, err	×
285	}	×
286
287	return NewRangeIndex(ranges, WithSerializeUint64Fn(writeBigSize))	10✔
288	}
289
290	func deleteOldAckedUpdates(tx kvdb.RwTx) error {	3✔
291	mainSessionsBkt := tx.ReadWriteBucket(cSessionBkt)	3✔
292	if mainSessionsBkt == nil {	3✔
293	return ErrUninitializedDB	×
294	}	×
295
296	return mainSessionsBkt.ForEach(func(sessID, _ []byte) error {	9✔
297	// Get the bucket for this particular session.	6✔
298	sessionBkt := mainSessionsBkt.NestedReadWriteBucket(	6✔
299	sessID,	6✔
300	)	6✔
301	if sessionBkt == nil {	6✔
302	return ErrClientSessionNotFound	×
303	}	×
304
305	// Get the bucket where any old acked updates would be stored.
306	oldAcksBucket := sessionBkt.NestedReadBucket(cSessionAcks)	6✔
307	if oldAcksBucket == nil {	6✔
308	return nil	×
309	}	×
310
311	// Now that we have read everything that we need to from
312	// the cSessionAcks sub-bucket, we can delete it.
313	return sessionBkt.DeleteNestedBucket(cSessionAcks)	6✔
314	})
315	}
316
317	// processMigration uses the given transaction to perform a maximum of
318	// sessionsPerTx session migrations. If startKey is non-nil, it is used to
319	// determine the first session to start the migration at. The first return
320	// item is the key of the last session that was migrated successfully and the
321	// boolean is true if there are no more sessions left to migrate.
322	func processMigration(tx kvdb.RwTx, startKey []byte, sessionsPerTx int) ([]byte,
323	bool, error) {	4✔
324		4✔
325	chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)	4✔
326	if chanDetailsBkt == nil {	4✔
327	return nil, false, ErrUninitializedDB	×
328	}	×
329
330	// sessionCount keeps track of the number of sessions that have been
331	// migrated under the current db transaction.
332	var sessionCount int	4✔
333		4✔
334	// migrateSessionCB is a callback function that calls migrateSession	4✔
335	// in order to migrate a single session. Upon success, the sessionCount	4✔
336	// is incremented and is then compared against sessionsPerTx to	4✔
337	// determine if we should continue migrating more sessions in this db	4✔
338	// transaction.	4✔
339	migrateSessionCB := func(sessionBkt kvdb.RwBucket) error {	12✔
340	err := migrateSession(chanDetailsBkt, sessionBkt)	8✔
341	if err != nil {	8✔
342	return err	×
343	}	×
344
345	sessionCount++	8✔
346		8✔
347	// If we have migrated sessionsPerTx sessions in this tx, then	8✔
348	// we return errExit in order to signal that this tx should be	8✔
349	// committed and the migration should be continued in a new	8✔
350	// transaction.	8✔
351	if sessionCount >= sessionsPerTx {	12✔
352	return errExit	4✔
353	}	4✔
354
355	return nil	4✔
356	}
357
358	// Starting at startKey, iterate over the sessions in the db and migrate
359	// them until either all are migrated or until the errExit signal is
360	// received.
361	lastKey, err := sessionIterator(tx, startKey, migrateSessionCB)	4✔
362	if err != nil && errors.Is(err, errExit) {	8✔
363	return lastKey, false, nil	4✔
364	} else if err != nil {	4✔
365	return nil, false, err	×
366	}	×
367
368	// The migration is complete.
369	return nil, true, nil	×
370	}
371
372	// migrateSession migrates a single session's acked-updates to the new
373	// RangeIndex form.
374	func migrateSession(chanDetailsBkt kvdb.RBucket,
375	sessionBkt kvdb.RwBucket) error {	8✔
376		8✔
377	// Get the existing cSessionAcks bucket. If there is no such bucket,	8✔
378	// then there are no acked-updates to migrate for this session.	8✔
379	sessionAcks := sessionBkt.NestedReadBucket(cSessionAcks)	8✔
380	if sessionAcks == nil {	8✔
381	return nil	×
382	}	×
383
384	// If there is already a new cSessionAckedRangeIndex bucket, then this
385	// session has already been migrated.
386	sessionAckRangesBkt := sessionBkt.NestedReadBucket(	8✔
387	cSessionAckRangeIndex,	8✔
388	)	8✔
389	if sessionAckRangesBkt != nil {	11✔
390	return nil	3✔
391	}	3✔
392
393	// Otherwise, we will iterate over each of the acked-updates, and we
394	// will construct a new RangeIndex for each channel.
395	m := make(map[ChannelID]*RangeIndex)	5✔
396	if err := sessionAcks.ForEach(func(_, v []byte) error {	20✔
397	var backupID BackupID	15✔
398	err := backupID.Decode(bytes.NewReader(v))	15✔
399	if err != nil {	15✔
400	return err	×
401	}	×
402
403	if _, ok := m[backupID.ChanID]; !ok {	23✔
404	index, err := NewRangeIndex(nil)	8✔
405	if err != nil {	8✔
406	return err	×
407	}	×
408
409	m[backupID.ChanID] = index	8✔
410	}
411
412	return m[backupID.ChanID].Add(backupID.CommitHeight, nil)	15✔
413	}); err != nil {	×
414	return err	×
415	}	×
416
417	// Create a new sub-bucket that will be used to store the new RangeIndex
418	// representation of the acked updates.
419	ackRangeBkt, err := sessionBkt.CreateBucket(cSessionAckRangeIndex)	5✔
420	if err != nil {	5✔
421	return err	×
422	}	×
423
424	// Iterate over each of the new range indexes that we will add for this
425	// session.
426	for chanID, rangeIndex := range m {	13✔
427	// Get db chanID.	8✔
428	chanDetails := chanDetailsBkt.NestedReadBucket(chanID[:])	8✔
429	if chanDetails == nil {	8✔
430	return ErrCorruptChanDetails	×
431	}	×
432
433	// Create a sub-bucket for this channel using the db-assigned ID
434	// for the channel.
435	dbChanID := chanDetails.Get(cChanDBID)	8✔
436	chanAcksBkt, err := ackRangeBkt.CreateBucket(dbChanID)	8✔
437	if err != nil {	8✔
438	return err	×
439	}	×
440
441	// Iterate over the range pairs that we need to add to the DB.
442	for k, v := range rangeIndex.GetAllRanges() {	17✔
443	start, err := writeBigSize(k)	9✔
444	if err != nil {	9✔
445	return err	×
446	}	×
447
448	end, err := writeBigSize(v)	9✔
449	if err != nil {	9✔
450	return err	×
451	}	×
452
453	err = chanAcksBkt.Put(start, end)	9✔
454	if err != nil {	9✔
455	return err	×
456	}	×
457	}
458	}
459
460	return nil	5✔
461	}
462
463	// logMigrationStats reads the buckets to provide stats over current migration
464	// progress. The returned values are the numbers of total records and already
465	// migrated records.
466	func logMigrationStats(db kvdb.Backend) (uint64, uint64, error) {	8✔
467	var (	8✔
468	err error	8✔
469	total uint64	8✔
470	unmigrated uint64	8✔
471	)	8✔
472		8✔
473	err = kvdb.View(db, func(tx kvdb.RTx) error {	16✔
474	total, unmigrated, err = getMigrationStats(tx)	8✔
475		8✔
476	return err	8✔
477	}, func() {})	16✔
478
479	log.Debugf("Total sessions=%d, unmigrated=%d", total, unmigrated)	8✔
480		8✔
481	return total, total - unmigrated, err	8✔
482	}
483
484	// getMigrationStats iterates over all sessions. It counts the total number of
485	// sessions as well as the total number of unmigrated sessions.
486	func getMigrationStats(tx kvdb.RTx) (uint64, uint64, error) {	8✔
487	var (	8✔
488	total uint64	8✔
489	unmigrated uint64	8✔
490	)	8✔
491		8✔
492	// Get sessions bucket.	8✔
493	mainSessionsBkt := tx.ReadBucket(cSessionBkt)	8✔
494	if mainSessionsBkt == nil {	8✔
495	return 0, 0, ErrUninitializedDB	×
496	}	×
497
498	// Iterate over each session ID in the bucket.
499	err := mainSessionsBkt.ForEach(func(sessID, _ []byte) error {	27✔
500	// Get the bucket for this particular session.	19✔
501	sessionBkt := mainSessionsBkt.NestedReadBucket(sessID)	19✔
502	if sessionBkt == nil {	20✔
503	return ErrClientSessionNotFound	1✔
504	}	1✔
505
506	total++	18✔
507		18✔
508	// Get the cSessionAckRangeIndex bucket.	18✔
509	sessionAcksBkt := sessionBkt.NestedReadBucket(cSessionAcks)	18✔
510		18✔
511	// Get the cSessionAckRangeIndex bucket.	18✔
512	sessionAckRangesBkt := sessionBkt.NestedReadBucket(	18✔
513	cSessionAckRangeIndex,	18✔
514	)	18✔
515		18✔
516	// If both buckets do not exist, then this session is empty and	18✔
517	// does not need to be migrated.	18✔
518	if sessionAckRangesBkt == nil && sessionAcksBkt == nil {	18✔
519	return nil	×
520	}	×
521
522	// If the sessionAckRangesBkt is not nil, then the session has
523	// already been migrated.
524	if sessionAckRangesBkt != nil {	29✔
525	return nil	11✔
526	}	11✔
527
528	// Else the session has not yet been migrated.
529	unmigrated++	7✔
530		7✔
531	return nil	7✔
532	})
533	if err != nil {	9✔
534	return 0, 0, err	1✔
535	}	1✔
536
537	return total, unmigrated, nil	7✔
538	}
539
540	// getDBChanID returns the db-assigned channel ID for the given real channel ID.
541	// It returns both the uint64 and byte representation.
542	func getDBChanID(chanDetailsBkt kvdb.RBucket, chanID ChannelID) (uint64,
543	[]byte, error) {	20✔
544		20✔
545	chanDetails := chanDetailsBkt.NestedReadBucket(chanID[:])	20✔
546	if chanDetails == nil {	20✔
547	return 0, nil, ErrChannelNotRegistered	×
548	}	×
549
550	idBytes := chanDetails.Get(cChanDBID)	20✔
551	if len(idBytes) == 0 {	20✔
552	return 0, nil, fmt.Errorf("no db-assigned ID found for "+	×
553	"channel ID %s", chanID)	×
554	}	×
555
556	id, err := readBigSize(idBytes)	20✔
557	if err != nil {	20✔
558	return 0, nil, err	×
559	}	×
560
561	return id, idBytes, nil	20✔
562	}
563
564	// callback defines a type that's used by the sessionIterator.
565	type callback func(bkt kvdb.RwBucket) error
566
567	// sessionIterator is a helper function that iterates over the main sessions
568	// bucket and performs the callback function on each individual session. If a
569	// seeker is specified, it will move the cursor to the given position otherwise
570	// it will start from the first item.
571	func sessionIterator(tx kvdb.RwTx, seeker []byte, cb callback) ([]byte, error) {	4✔
572	// Get sessions bucket.	4✔
573	mainSessionsBkt := tx.ReadWriteBucket(cSessionBkt)	4✔
574	if mainSessionsBkt == nil {	4✔
575	return nil, ErrUninitializedDB	×
576	}	×
577
578	c := mainSessionsBkt.ReadCursor()	4✔
579	k, _ := c.First()	4✔
580		4✔
581	// Move the cursor to the specified position if seeker is non-nil.	4✔
582	if seeker != nil {	6✔
583	k, _ = c.Seek(seeker)	2✔
584	}	2✔
585
586	// Start the iteration and exit on condition.
587	for k := k; k != nil; k, _ = c.Next() {	12✔
588	// Get the bucket for this particular session.	8✔
589	bkt := mainSessionsBkt.NestedReadWriteBucket(k)	8✔
590	if bkt == nil {	8✔
591	return nil, ErrClientSessionNotFound	×
592	}	×
593
594	// Call the callback function with the session's bucket.
595	if err := cb(bkt); err != nil {	12✔
596	// return k, err	4✔
597	lastIndex := make([]byte, len(k))	4✔
598	copy(lastIndex, k)	4✔
599	return lastIndex, err	4✔
600	}	4✔
601	}
602
603	return nil, nil	×
604	}
605
606	// writeBigSize will encode the given uint64 as a BigSize byte slice.
607	func writeBigSize(i uint64) ([]byte, error) {	45✔
608	var b bytes.Buffer	45✔
609	err := tlv.WriteVarInt(&b, i, &[8]byte{})	45✔
610	if err != nil {	45✔
611	return nil, err	×
612	}	×
613
614	return b.Bytes(), nil	45✔
615	}
616
617	// readBigSize converts the given byte slice into a uint64 and assumes that the
618	// bytes slice is using BigSize encoding.
619	func readBigSize(b []byte) (uint64, error) {	54✔
620	r := bytes.NewReader(b)	54✔
621	i, err := tlv.ReadVarInt(r, &[8]byte{})	54✔
622	if err != nil {	54✔
623	return 0, err	×
624	}	×
625
626	return i, nil	54✔
627	}

lightningnetwork / lnd / 15951470896

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