14193549836

Committed 01 Apr 2025 10:40AM UTC coverage: 69.046% (+0.007%) from 69.039%

Build # 14193549836

Build Type

Pull #9665

github

Committed by

web-flow

Commit Message

Merge e8825f209 into b01f4e514

Pull Request Pull Request #9665: kvdb: bump etcd libs to v3.5.12

Run Details

133439 of 193262 relevant lines covered (69.05%)

22119.45 hits per line

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

67.68

/watchtower/wtdb/migration7/client_db.go

package migration7

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

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

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

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

        // cChannelSummary is a sub-bucket of cChanDetailsBkt which stores the
        // encoded body of ClientChanSummary.
        cChannelSummary = []byte("client-channel-summary")

        // cChanSessions is a sub-bucket of cChanDetailsBkt which stores:
        //    session-id -> 1
        cChanSessions = []byte("client-channel-sessions")

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

        // cSessionDBID is a key used in the cSessionBkt to store the
        // db-assigned-d of a session.
        cSessionDBID = []byte("client-session-db-id")

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

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

        // ErrCorruptClientSession signals that the client session's on-disk
        // structure deviates from what is expected.
        ErrCorruptClientSession = errors.New("client session corrupted")

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

// MigrateChannelToSessionIndex migrates the tower client DB to add an index
// from channel-to-session. This will make it easier in future to check which
// sessions have updates for which channels.
func MigrateChannelToSessionIndex(tx kvdb.RwTx) error {
        log.Infof("Migrating the tower client DB to build a new " +
                "channel-to-session index")

        sessionsBkt := tx.ReadBucket(cSessionBkt)
        if sessionsBkt == nil {
                return ErrUninitializedDB
        }

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

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

        // First gather all the new channel-to-session pairs that we want to
        // add.
        index, err := collectIndex(sessionsBkt)
        if err != nil {
                return err
        }

        // Then persist those pairs to the db.
        return persistIndex(chanDetailsBkt, chanIDsBkt, index)
}

// collectIndex iterates through all the sessions and uses the keys in the
// cSessionAckRangeIndex bucket to collect all the channels that the session
// has updates for. The function returns a map from channel ID to session ID
// (using the db-assigned IDs for both).
func collectIndex(sessionsBkt kvdb.RBucket) (map[uint64]map[uint64]bool,
        error) {

        index := make(map[uint64]map[uint64]bool)
        err := sessionsBkt.ForEach(func(sessID, _ []byte) error {
                sessionBkt := sessionsBkt.NestedReadBucket(sessID)
                if sessionBkt == nil {
                        return ErrCorruptClientSession
                }

                ackedRanges := sessionBkt.NestedReadBucket(
                        cSessionAckRangeIndex,
                )
                if ackedRanges == nil {
                        return nil
                }

                sessDBIDBytes := sessionBkt.Get(cSessionDBID)
                if sessDBIDBytes == nil {
                        return ErrCorruptClientSession
                }

                sessDBID, err := readUint64(sessDBIDBytes)
                if err != nil {
                        return err
                }

                return ackedRanges.ForEach(func(dbChanIDBytes, _ []byte) error {
                        dbChanID, err := readUint64(dbChanIDBytes)
                        if err != nil {
                                return err
                        }

                        if _, ok := index[dbChanID]; !ok {
                                index[dbChanID] = make(map[uint64]bool)
                        }

                        index[dbChanID][sessDBID] = true

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

        return index, nil
}

// persistIndex adds the channel-to-session mapping in each channel's details
// bucket.
func persistIndex(chanDetailsBkt kvdb.RwBucket, chanIDsBkt kvdb.RBucket,
        index map[uint64]map[uint64]bool) error {

        for dbChanID, sessIDs := range index {
                dbChanIDBytes, err := writeUint64(dbChanID)
                if err != nil {
                        return err
                }

                realChanID := chanIDsBkt.Get(dbChanIDBytes)

                chanBkt := chanDetailsBkt.NestedReadWriteBucket(realChanID)
                if chanBkt == nil {
                        return fmt.Errorf("channel not found")
                }

                sessIDsBkt, err := chanBkt.CreateBucket(cChanSessions)
                if err != nil {
                        return err
                }

                for id := range sessIDs {
                        sessID, err := writeUint64(id)
                        if err != nil {
                                return err
                        }

                        err = sessIDsBkt.Put(sessID, []byte{1})
                        if err != nil {
                                return err
                        }
                }
        }

        return nil
}

func writeUint64(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
}

func readUint64(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 migration7
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	// cSessionBkt is a top-level bucket storing:
15	// session-id => cSessionBody -> encoded ClientSessionBody
16	// => cSessionDBID -> db-assigned-id
17	// => cSessionCommits => seqnum -> encoded CommittedUpdate
18	// => cSessionAckRangeIndex => chan-id => acked-index-range
19	cSessionBkt = []byte("client-session-bucket")
20
21	// cChanDetailsBkt is a top-level bucket storing:
22	// channel-id => cChannelSummary -> encoded ClientChanSummary.
23	// => cChanDBID -> db-assigned-id
24	// => cChanSessions => db-session-id -> 1
25	cChanDetailsBkt = []byte("client-channel-detail-bucket")
26
27	// cChannelSummary is a sub-bucket of cChanDetailsBkt which stores the
28	// encoded body of ClientChanSummary.
29	cChannelSummary = []byte("client-channel-summary")
30
31	// cChanSessions is a sub-bucket of cChanDetailsBkt which stores:
32	// session-id -> 1
33	cChanSessions = []byte("client-channel-sessions")
34
35	// cSessionAckRangeIndex is a sub-bucket of cSessionBkt storing:
36	// chan-id => start -> end
37	cSessionAckRangeIndex = []byte("client-session-ack-range-index")
38
39	// cSessionDBID is a key used in the cSessionBkt to store the
40	// db-assigned-d of a session.
41	cSessionDBID = []byte("client-session-db-id")
42
43	// cChanIDIndexBkt is a top-level bucket storing:
44	// db-assigned-id -> channel-ID
45	cChanIDIndexBkt = []byte("client-channel-id-index")
46
47	// ErrUninitializedDB signals that top-level buckets for the database
48	// have not been initialized.
49	ErrUninitializedDB = errors.New("db not initialized")
50
51	// ErrCorruptClientSession signals that the client session's on-disk
52	// structure deviates from what is expected.
53	ErrCorruptClientSession = errors.New("client session corrupted")
54
55	// byteOrder is the default endianness used when serializing integers.
56	byteOrder = binary.BigEndian
57	)
58
59	// MigrateChannelToSessionIndex migrates the tower client DB to add an index
60	// from channel-to-session. This will make it easier in future to check which
61	// sessions have updates for which channels.
62	func MigrateChannelToSessionIndex(tx kvdb.RwTx) error {	3✔
63	log.Infof("Migrating the tower client DB to build a new " +	3✔
64	"channel-to-session index")	3✔
65		3✔
66	sessionsBkt := tx.ReadBucket(cSessionBkt)	3✔
67	if sessionsBkt == nil {	3✔
68	return ErrUninitializedDB	×
69	}	×
70
71	chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)	3✔
72	if chanDetailsBkt == nil {	3✔
73	return ErrUninitializedDB	×
74	}	×
75
76	chanIDsBkt := tx.ReadBucket(cChanIDIndexBkt)	3✔
77	if chanIDsBkt == nil {	3✔
78	return ErrUninitializedDB	×
79	}	×
80
81	// First gather all the new channel-to-session pairs that we want to
82	// add.
83	index, err := collectIndex(sessionsBkt)	3✔
84	if err != nil {	3✔
85	return err	×
86	}	×
87
88	// Then persist those pairs to the db.
89	return persistIndex(chanDetailsBkt, chanIDsBkt, index)	3✔
90	}
91
92	// collectIndex iterates through all the sessions and uses the keys in the
93	// cSessionAckRangeIndex bucket to collect all the channels that the session
94	// has updates for. The function returns a map from channel ID to session ID
95	// (using the db-assigned IDs for both).
96	func collectIndex(sessionsBkt kvdb.RBucket) (map[uint64]map[uint64]bool,
97	error) {	3✔
98		3✔
99	index := make(map[uint64]map[uint64]bool)	3✔
100	err := sessionsBkt.ForEach(func(sessID, _ []byte) error {	7✔
101	sessionBkt := sessionsBkt.NestedReadBucket(sessID)	4✔
102	if sessionBkt == nil {	4✔
103	return ErrCorruptClientSession	×
104	}	×
105
106	ackedRanges := sessionBkt.NestedReadBucket(	4✔
107	cSessionAckRangeIndex,	4✔
108	)	4✔
109	if ackedRanges == nil {	4✔
110	return nil	×
111	}	×
112
113	sessDBIDBytes := sessionBkt.Get(cSessionDBID)	4✔
114	if sessDBIDBytes == nil {	4✔
115	return ErrCorruptClientSession	×
116	}	×
117
118	sessDBID, err := readUint64(sessDBIDBytes)	4✔
119	if err != nil {	4✔
120	return err	×
121	}	×
122
123	return ackedRanges.ForEach(func(dbChanIDBytes, _ []byte) error {	10✔
124	dbChanID, err := readUint64(dbChanIDBytes)	6✔
125	if err != nil {	6✔
126	return err	×
127	}	×
128
129	if _, ok := index[dbChanID]; !ok {	10✔
130	index[dbChanID] = make(map[uint64]bool)	4✔
131	}	4✔
132
133	index[dbChanID][sessDBID] = true	6✔
134		6✔
135	return nil	6✔
136	})
137	})
138	if err != nil {	3✔
139	return nil, err	×
140	}	×
141
142	return index, nil	3✔
143	}
144
145	// persistIndex adds the channel-to-session mapping in each channel's details
146	// bucket.
147	func persistIndex(chanDetailsBkt kvdb.RwBucket, chanIDsBkt kvdb.RBucket,
148	index map[uint64]map[uint64]bool) error {	3✔
149		3✔
150	for dbChanID, sessIDs := range index {	6✔
151	dbChanIDBytes, err := writeUint64(dbChanID)	3✔
152	if err != nil {	3✔
153	return err	×
154	}	×
155
156	realChanID := chanIDsBkt.Get(dbChanIDBytes)	3✔
157		3✔
158	chanBkt := chanDetailsBkt.NestedReadWriteBucket(realChanID)	3✔
159	if chanBkt == nil {	4✔
160	return fmt.Errorf("channel not found")	1✔
161	}	1✔
162
163	sessIDsBkt, err := chanBkt.CreateBucket(cChanSessions)	2✔
164	if err != nil {	2✔
165	return err	×
166	}	×
167
168	for id := range sessIDs {	5✔
169	sessID, err := writeUint64(id)	3✔
170	if err != nil {	3✔
171	return err	×
172	}	×
173
174	err = sessIDsBkt.Put(sessID, []byte{1})	3✔
175	if err != nil {	3✔
176	return err	×
177	}	×
178	}
179	}
180
181	return nil	2✔
182	}
183
184	func writeUint64(i uint64) ([]byte, error) {	24✔
185	var b bytes.Buffer	24✔
186	err := tlv.WriteVarInt(&b, i, &[8]byte{})	24✔
187	if err != nil {	24✔
188	return nil, err	×
189	}	×
190
191	return b.Bytes(), nil	24✔
192	}
193
194	func readUint64(b []byte) (uint64, error) {	10✔
195	r := bytes.NewReader(b)	10✔
196	i, err := tlv.ReadVarInt(r, &[8]byte{})	10✔
197	if err != nil {	10✔
198	return 0, err	×
199	}	×
200
201	return i, nil	10✔
202	}

lightningnetwork / lnd / 14193549836

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