13586005509

Committed 28 Feb 2025 10:14AM UTC coverage: 68.629% (+9.9%) from 58.77%

Build # 13586005509

Build Type

Pull #9521

github

Committed by

web-flow

Commit Message

Merge 37d3a70a5 into 8532955b3

Pull Request Pull Request #9521: unit: remove GOACC, use Go 1.20 native coverage functionality

Run Details

129950 of 189351 relevant lines covered (68.63%)

23726.46 hits per line

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

73.95

/watchtower/wtdb/migration8/migration.go

package migration8

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

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

var (
        // cSessionBkt is a top-level bucket storing:
        //   session-id => cSessionBody -> encoded ClientSessionBody
        //                 => cSessionDBID -> db-assigned-id
        //              => cSessionCommits => seqnum -> encoded CommittedUpdate
        //              => cSessionAckRangeIndex => db-chan-id => start -> end
        //                 => cSessionRogueUpdateCount -> count
        cSessionBkt = []byte("client-session-bucket")

        // cChanIDIndexBkt is a top-level bucket storing:
        //    db-assigned-id -> channel-ID
        cChanIDIndexBkt = []byte("client-channel-id-index")

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

        // cSessionBody is a sub-bucket of cSessionBkt storing:
        //    seqnum -> encoded CommittedUpdate.
        cSessionCommits = []byte("client-session-commits")

        // cChanDetailsBkt is a top-level bucket storing:
        //   channel-id => cChannelSummary -> encoded ClientChanSummary.
        //                  => cChanDBID -> db-assigned-id
        //                 => cChanSessions => db-session-id -> 1
        //                 => cChanClosedHeight -> block-height
        //                 => cChanMaxCommitmentHeight -> commitment-height
        cChanDetailsBkt = []byte("client-channel-detail-bucket")

        cChanMaxCommitmentHeight = []byte(
                "client-channel-max-commitment-height",
        )

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

        byteOrder = binary.BigEndian
)

// MigrateChannelMaxHeights migrates the tower client db by collecting all the
// max commitment heights that have been backed up for each channel and then
// storing those heights alongside the channel info.
func MigrateChannelMaxHeights(tx kvdb.RwTx) error {
        log.Infof("Migrating the tower client DB for quick channel max " +
                "commitment height lookup")

        heights, err := collectChanMaxHeights(tx)
        if err != nil {
                return err
        }

        return writeChanMaxHeights(tx, heights)
}

// writeChanMaxHeights iterates over the given channel ID to height map and
// writes an entry under the cChanMaxCommitmentHeight key for each channel.
func writeChanMaxHeights(tx kvdb.RwTx, heights map[ChannelID]uint64) error {
        chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)
        if chanDetailsBkt == nil {
                return ErrUninitializedDB
        }

        for chanID, maxHeight := range heights {
                chanDetails := chanDetailsBkt.NestedReadWriteBucket(chanID[:])

                // If the details bucket for this channel ID does not exist,
                // it is probably a channel that has been closed and deleted
                // already. So we can skip this height.
                if chanDetails == nil {
                        continue
                }

                b, err := writeBigSize(maxHeight)
                if err != nil {
                        return err
                }

                err = chanDetails.Put(cChanMaxCommitmentHeight, b)
                if err != nil {
                        return err
                }
        }

        return nil
}

// collectChanMaxHeights iterates over all the sessions in the DB. For each
// session, it iterates over all the Acked updates and the committed updates
// to collect the maximum commitment height for each channel.
func collectChanMaxHeights(tx kvdb.RwTx) (map[ChannelID]uint64, error) {
        sessionsBkt := tx.ReadBucket(cSessionBkt)
        if sessionsBkt == nil {
                return nil, ErrUninitializedDB
        }

        chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)
        if chanIDIndexBkt == nil {
                return nil, ErrUninitializedDB
        }

        heights := make(map[ChannelID]uint64)

        // For each update we consider, we will only update the heights map if
        // the commitment height for the channel is larger than the current
        // max height stored for the channel.
        cb := func(chanID ChannelID, commitHeight uint64) {
                if commitHeight > heights[chanID] {
                        heights[chanID] = commitHeight
                }
        }

        err := sessionsBkt.ForEach(func(sessIDBytes, _ []byte) error {
                sessBkt := sessionsBkt.NestedReadBucket(sessIDBytes)
                if sessBkt == nil {
                        return fmt.Errorf("bucket not found for session %x",
                                sessIDBytes)
                }

                err := forEachCommittedUpdate(sessBkt, cb)
                if err != nil {
                        return err
                }

                return forEachAckedUpdate(sessBkt, chanIDIndexBkt, cb)
        })
        if err != nil {
                return nil, err
        }

        return heights, nil
}

// forEachCommittedUpdate iterates over all the given session's committed
// updates and calls the call-back for each.
func forEachCommittedUpdate(sessBkt kvdb.RBucket,
        cb func(chanID ChannelID, commitHeight uint64)) error {

        sessionCommits := sessBkt.NestedReadBucket(cSessionCommits)
        if sessionCommits == nil {
                return nil
        }

        return sessionCommits.ForEach(func(k, v []byte) error {
                var update CommittedUpdate
                err := update.Decode(bytes.NewReader(v))
                if err != nil {
                        return err
                }

                cb(update.BackupID.ChanID, update.BackupID.CommitHeight)

                return nil
        })
}

// forEachAckedUpdate iterates over all the given session's acked update range
// indices and calls the call-back for each.
func forEachAckedUpdate(sessBkt, chanIDIndexBkt kvdb.RBucket,
        cb func(chanID ChannelID, commitHeight uint64)) error {

        sessionAcksRanges := sessBkt.NestedReadBucket(cSessionAckRangeIndex)
        if sessionAcksRanges == nil {
                return nil
        }

        return sessionAcksRanges.ForEach(func(dbChanID, _ []byte) error {
                rangeBkt := sessionAcksRanges.NestedReadBucket(dbChanID)
                if rangeBkt == nil {
                        return nil
                }

                index, err := readRangeIndex(rangeBkt)
                if err != nil {
                        return err
                }

                chanIDBytes := chanIDIndexBkt.Get(dbChanID)
                var chanID ChannelID
                copy(chanID[:], chanIDBytes)

                cb(chanID, index.MaxHeight())

                return nil
        })
}

// readRangeIndex reads a persisted RangeIndex from the passed bucket and into
// a new in-memory RangeIndex.
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))
}

1	package migration8
2
3	import (
4	"bytes"
5	"encoding/binary"
6	"errors"
7	"fmt"
8
9	"github.com/lightningnetwork/lnd/kvdb"
10	)
11
12	var (
13	// cSessionBkt is a top-level bucket storing:
14	// session-id => cSessionBody -> encoded ClientSessionBody
15	// => cSessionDBID -> db-assigned-id
16	// => cSessionCommits => seqnum -> encoded CommittedUpdate
17	// => cSessionAckRangeIndex => db-chan-id => start -> end
18	// => cSessionRogueUpdateCount -> count
19	cSessionBkt = []byte("client-session-bucket")
20
21	// cChanIDIndexBkt is a top-level bucket storing:
22	// db-assigned-id -> channel-ID
23	cChanIDIndexBkt = []byte("client-channel-id-index")
24
25	// cSessionAckRangeIndex is a sub-bucket of cSessionBkt storing
26	// chan-id => start -> end
27	cSessionAckRangeIndex = []byte("client-session-ack-range-index")
28
29	// cSessionBody is a sub-bucket of cSessionBkt storing:
30	// seqnum -> encoded CommittedUpdate.
31	cSessionCommits = []byte("client-session-commits")
32
33	// cChanDetailsBkt is a top-level bucket storing:
34	// channel-id => cChannelSummary -> encoded ClientChanSummary.
35	// => cChanDBID -> db-assigned-id
36	// => cChanSessions => db-session-id -> 1
37	// => cChanClosedHeight -> block-height
38	// => cChanMaxCommitmentHeight -> commitment-height
39	cChanDetailsBkt = []byte("client-channel-detail-bucket")
40
41	cChanMaxCommitmentHeight = []byte(
42	"client-channel-max-commitment-height",
43	)
44
45	// ErrUninitializedDB signals that top-level buckets for the database
46	// have not been initialized.
47	ErrUninitializedDB = errors.New("db not initialized")
48
49	byteOrder = binary.BigEndian
50	)
51
52	// MigrateChannelMaxHeights migrates the tower client db by collecting all the
53	// max commitment heights that have been backed up for each channel and then
54	// storing those heights alongside the channel info.
55	func MigrateChannelMaxHeights(tx kvdb.RwTx) error {	1✔
56	log.Infof("Migrating the tower client DB for quick channel max " +	1✔
57	"commitment height lookup")	1✔
58		1✔
59	heights, err := collectChanMaxHeights(tx)	1✔
60	if err != nil {	1✔
61	return err	×
62	}	×
63
64	return writeChanMaxHeights(tx, heights)	1✔
65	}
66
67	// writeChanMaxHeights iterates over the given channel ID to height map and
68	// writes an entry under the cChanMaxCommitmentHeight key for each channel.
69	func writeChanMaxHeights(tx kvdb.RwTx, heights map[ChannelID]uint64) error {	1✔
70	chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)	1✔
71	if chanDetailsBkt == nil {	1✔
72	return ErrUninitializedDB	×
73	}	×
74
75	for chanID, maxHeight := range heights {	6✔
76	chanDetails := chanDetailsBkt.NestedReadWriteBucket(chanID[:])	5✔
77		5✔
78	// If the details bucket for this channel ID does not exist,	5✔
79	// it is probably a channel that has been closed and deleted	5✔
80	// already. So we can skip this height.	5✔
81	if chanDetails == nil {	7✔
82	continue	2✔
83	}
84
85	b, err := writeBigSize(maxHeight)	3✔
86	if err != nil {	3✔
87	return err	×
88	}	×
89
90	err = chanDetails.Put(cChanMaxCommitmentHeight, b)	3✔
91	if err != nil {	3✔
92	return err	×
93	}	×
94	}
95
96	return nil	1✔
97	}
98
99	// collectChanMaxHeights iterates over all the sessions in the DB. For each
100	// session, it iterates over all the Acked updates and the committed updates
101	// to collect the maximum commitment height for each channel.
102	func collectChanMaxHeights(tx kvdb.RwTx) (map[ChannelID]uint64, error) {	1✔
103	sessionsBkt := tx.ReadBucket(cSessionBkt)	1✔
104	if sessionsBkt == nil {	1✔
105	return nil, ErrUninitializedDB	×
106	}	×
107
108	chanIDIndexBkt := tx.ReadBucket(cChanIDIndexBkt)	1✔
109	if chanIDIndexBkt == nil {	1✔
110	return nil, ErrUninitializedDB	×
111	}	×
112
113	heights := make(map[ChannelID]uint64)	1✔
114		1✔
115	// For each update we consider, we will only update the heights map if	1✔
116	// the commitment height for the channel is larger than the current	1✔
117	// max height stored for the channel.	1✔
118	cb := func(chanID ChannelID, commitHeight uint64) {	8✔
119	if commitHeight > heights[chanID] {	13✔
120	heights[chanID] = commitHeight	6✔
121	}	6✔
122	}
123
124	err := sessionsBkt.ForEach(func(sessIDBytes, _ []byte) error {	6✔
125	sessBkt := sessionsBkt.NestedReadBucket(sessIDBytes)	5✔
126	if sessBkt == nil {	5✔
127	return fmt.Errorf("bucket not found for session %x",	×
128	sessIDBytes)	×
129	}	×
130
131	err := forEachCommittedUpdate(sessBkt, cb)	5✔
132	if err != nil {	5✔
133	return err	×
134	}	×
135
136	return forEachAckedUpdate(sessBkt, chanIDIndexBkt, cb)	5✔
137	})
138	if err != nil {	1✔
139	return nil, err	×
140	}	×
141
142	return heights, nil	1✔
143	}
144
145	// forEachCommittedUpdate iterates over all the given session's committed
146	// updates and calls the call-back for each.
147	func forEachCommittedUpdate(sessBkt kvdb.RBucket,
148	cb func(chanID ChannelID, commitHeight uint64)) error {	5✔
149		5✔
150	sessionCommits := sessBkt.NestedReadBucket(cSessionCommits)	5✔
151	if sessionCommits == nil {	7✔
152	return nil	2✔
153	}	2✔
154
155	return sessionCommits.ForEach(func(k, v []byte) error {	6✔
156	var update CommittedUpdate	3✔
157	err := update.Decode(bytes.NewReader(v))	3✔
158	if err != nil {	3✔
159	return err	×
160	}	×
161
162	cb(update.BackupID.ChanID, update.BackupID.CommitHeight)	3✔
163		3✔
164	return nil	3✔
165	})
166	}
167
168	// forEachAckedUpdate iterates over all the given session's acked update range
169	// indices and calls the call-back for each.
170	func forEachAckedUpdate(sessBkt, chanIDIndexBkt kvdb.RBucket,
171	cb func(chanID ChannelID, commitHeight uint64)) error {	5✔
172		5✔
173	sessionAcksRanges := sessBkt.NestedReadBucket(cSessionAckRangeIndex)	5✔
174	if sessionAcksRanges == nil {	7✔
175	return nil	2✔
176	}	2✔
177
178	return sessionAcksRanges.ForEach(func(dbChanID, _ []byte) error {	7✔
179	rangeBkt := sessionAcksRanges.NestedReadBucket(dbChanID)	4✔
180	if rangeBkt == nil {	4✔
181	return nil	×
182	}	×
183
184	index, err := readRangeIndex(rangeBkt)	4✔
185	if err != nil {	4✔
186	return err	×
187	}	×
188
189	chanIDBytes := chanIDIndexBkt.Get(dbChanID)	4✔
190	var chanID ChannelID	4✔
191	copy(chanID[:], chanIDBytes)	4✔
192		4✔
193	cb(chanID, index.MaxHeight())	4✔
194		4✔
195	return nil	4✔
196	})
197	}
198
199	// readRangeIndex reads a persisted RangeIndex from the passed bucket and into
200	// a new in-memory RangeIndex.
201	func readRangeIndex(rangesBkt kvdb.RBucket) (*RangeIndex, error) {	4✔
202	ranges := make(map[uint64]uint64)	4✔
203	err := rangesBkt.ForEach(func(k, v []byte) error {	10✔
204	start, err := readBigSize(k)	6✔
205	if err != nil {	6✔
206	return err	×
207	}	×
208
209	end, err := readBigSize(v)	6✔
210	if err != nil {	6✔
211	return err	×
212	}	×
213
214	ranges[start] = end	6✔
215		6✔
216	return nil	6✔
217	})
218	if err != nil {	4✔
219	return nil, err	×
220	}	×
221
222	return NewRangeIndex(ranges, WithSerializeUint64Fn(writeBigSize))	4✔
223	}

lightningnetwork / lnd / 13586005509

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