13980275562

Committed 20 Mar 2025 10:06PM UTC coverage: 58.6% (-10.2%) from 68.789%

Build # 13980275562

Build Type

Pull #9623

github

Committed by

web-flow

Commit Message

Merge b9b960345 into 09b674508

Pull Request Pull Request #9623: Size msg test msg

Run Details

0 of 1518 new or added lines in 42 files covered. (0.0%)

26603 existing lines in 443 files now uncovered.

96807 of 165200 relevant lines covered (58.6%)

1.82 hits per line

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

0.0

/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.
UNCOV 62	func MigrateChannelToSessionIndex(tx kvdb.RwTx) error {	×
UNCOV 63	log.Infof("Migrating the tower client DB to build a new " +	×
UNCOV 64	"channel-to-session index")	×
UNCOV 65		×
UNCOV 66	sessionsBkt := tx.ReadBucket(cSessionBkt)	×
UNCOV 67	if sessionsBkt == nil {	×
68	return ErrUninitializedDB	×
69	}	×
70
UNCOV 71	chanDetailsBkt := tx.ReadWriteBucket(cChanDetailsBkt)	×
UNCOV 72	if chanDetailsBkt == nil {	×
73	return ErrUninitializedDB	×
74	}	×
75
UNCOV 76	chanIDsBkt := tx.ReadBucket(cChanIDIndexBkt)	×
UNCOV 77	if chanIDsBkt == nil {	×
78	return ErrUninitializedDB	×
79	}	×
80
81	// First gather all the new channel-to-session pairs that we want to
82	// add.
UNCOV 83	index, err := collectIndex(sessionsBkt)	×
UNCOV 84	if err != nil {	×
85	return err	×
86	}	×
87
88	// Then persist those pairs to the db.
UNCOV 89	return persistIndex(chanDetailsBkt, chanIDsBkt, index)	×
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,
UNCOV 97	error) {	×
UNCOV 98		×
UNCOV 99	index := make(map[uint64]map[uint64]bool)	×
UNCOV 100	err := sessionsBkt.ForEach(func(sessID, _ []byte) error {	×
UNCOV 101	sessionBkt := sessionsBkt.NestedReadBucket(sessID)	×
UNCOV 102	if sessionBkt == nil {	×
103	return ErrCorruptClientSession	×
104	}	×
105
UNCOV 106	ackedRanges := sessionBkt.NestedReadBucket(	×
UNCOV 107	cSessionAckRangeIndex,	×
UNCOV 108	)	×
UNCOV 109	if ackedRanges == nil {	×
110	return nil	×
111	}	×
112
UNCOV 113	sessDBIDBytes := sessionBkt.Get(cSessionDBID)	×
UNCOV 114	if sessDBIDBytes == nil {	×
115	return ErrCorruptClientSession	×
116	}	×
117
UNCOV 118	sessDBID, err := readUint64(sessDBIDBytes)	×
UNCOV 119	if err != nil {	×
120	return err	×
121	}	×
122
UNCOV 123	return ackedRanges.ForEach(func(dbChanIDBytes, _ []byte) error {	×
UNCOV 124	dbChanID, err := readUint64(dbChanIDBytes)	×
UNCOV 125	if err != nil {	×
126	return err	×
127	}	×
128
UNCOV 129	if _, ok := index[dbChanID]; !ok {	×
UNCOV 130	index[dbChanID] = make(map[uint64]bool)	×
UNCOV 131	}	×
132
UNCOV 133	index[dbChanID][sessDBID] = true	×
UNCOV 134		×
UNCOV 135	return nil	×
136	})
137	})
UNCOV 138	if err != nil {	×
139	return nil, err	×
140	}	×
141
UNCOV 142	return index, nil	×
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,
UNCOV 148	index map[uint64]map[uint64]bool) error {	×
UNCOV 149		×
UNCOV 150	for dbChanID, sessIDs := range index {	×
UNCOV 151	dbChanIDBytes, err := writeUint64(dbChanID)	×
UNCOV 152	if err != nil {	×
153	return err	×
154	}	×
155
UNCOV 156	realChanID := chanIDsBkt.Get(dbChanIDBytes)	×
UNCOV 157		×
UNCOV 158	chanBkt := chanDetailsBkt.NestedReadWriteBucket(realChanID)	×
UNCOV 159	if chanBkt == nil {	×
UNCOV 160	return fmt.Errorf("channel not found")	×
UNCOV 161	}	×
162
UNCOV 163	sessIDsBkt, err := chanBkt.CreateBucket(cChanSessions)	×
UNCOV 164	if err != nil {	×
165	return err	×
166	}	×
167
UNCOV 168	for id := range sessIDs {	×
UNCOV 169	sessID, err := writeUint64(id)	×
UNCOV 170	if err != nil {	×
171	return err	×
172	}	×
173
UNCOV 174	err = sessIDsBkt.Put(sessID, []byte{1})	×
UNCOV 175	if err != nil {	×
176	return err	×
177	}	×
178	}
179	}
180
UNCOV 181	return nil	×
182	}
183
UNCOV 184	func writeUint64(i uint64) ([]byte, error) {	×
UNCOV 185	var b bytes.Buffer	×
UNCOV 186	err := tlv.WriteVarInt(&b, i, &[8]byte{})	×
UNCOV 187	if err != nil {	×
188	return nil, err	×
189	}	×
190
UNCOV 191	return b.Bytes(), nil	×
192	}
193
UNCOV 194	func readUint64(b []byte) (uint64, error) {	×
UNCOV 195	r := bytes.NewReader(b)	×
UNCOV 196	i, err := tlv.ReadVarInt(r, &[8]byte{})	×
UNCOV 197	if err != nil {	×
198	return 0, err	×
199	}	×
200
UNCOV 201	return i, nil	×
202	}

lightningnetwork / lnd / 13980275562

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