13566028875

Committed 27 Feb 2025 12:09PM UTC coverage: 49.396% (-9.4%) from 58.748%

Build # 13566028875

Build Type

Pull #9555

github

Committed by

ellemouton

Commit Message

graph/db: populate the graph cache in Start instead of during construction

In this commit, we move the graph cache population logic out of the
ChannelGraph constructor and into its Start method instead.

Pull Request Pull Request #9555: graph: extract cache from CRUD [6]

Run Details

34 of 54 new or added lines in 4 files covered. (62.96%)

27464 existing lines in 436 files now uncovered.

101095 of 204664 relevant lines covered (49.4%)

1.54 hits per line

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

0.0

/channeldb/migration_01_to_11/db.go

package migration_01_to_11

import (
        "bytes"
        "encoding/binary"
        "fmt"
        "os"
        "path/filepath"
        "time"

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

const (
        dbName           = "channel.db"
        dbFilePermission = 0600
)

// migration is a function which takes a prior outdated version of the database
// instances and mutates the key/bucket structure to arrive at a more
// up-to-date version of the database.
type migration func(tx kvdb.RwTx) error

var (
        // Big endian is the preferred byte order, due to cursor scans over
        // integer keys iterating in order.
        byteOrder = binary.BigEndian
)

// DB is the primary datastore for the lnd daemon. The database stores
// information related to nodes, routing data, open/closed channels, fee
// schedules, and reputation data.
type DB struct {
        kvdb.Backend
        dbPath string
        graph  *ChannelGraph
        now    func() time.Time
}

// Open opens an existing channeldb. Any necessary schemas migrations due to
// updates will take place as necessary.
func Open(dbPath string, modifiers ...OptionModifier) (*DB, error) {
        path := filepath.Join(dbPath, dbName)

        if !fileExists(path) {
                if err := createChannelDB(dbPath); err != nil {
                        return nil, err
                }
        }

        opts := DefaultOptions()
        for _, modifier := range modifiers {
                modifier(&opts)
        }

        // Specify bbolt freelist options to reduce heap pressure in case the
        // freelist grows to be very large.
        bdb, err := kvdb.Open(
                kvdb.BoltBackendName, path,
                opts.NoFreelistSync, opts.DBTimeout,
        )
        if err != nil {
                return nil, err
        }

        chanDB := &DB{
                Backend: bdb,
                dbPath:  dbPath,
                now:     time.Now,
        }
        chanDB.graph = newChannelGraph(
                chanDB, opts.RejectCacheSize, opts.ChannelCacheSize,
        )

        return chanDB, nil
}

// createChannelDB creates and initializes a fresh version of channeldb. In
// the case that the target path has not yet been created or doesn't yet exist,
// then the path is created. Additionally, all required top-level buckets used
// within the database are created.
func createChannelDB(dbPath string) error {
        if !fileExists(dbPath) {
                if err := os.MkdirAll(dbPath, 0700); err != nil {
                        return err
                }
        }

        path := filepath.Join(dbPath, dbName)
        bdb, err := kvdb.Create(
                kvdb.BoltBackendName, path, false, kvdb.DefaultDBTimeout,
        )
        if err != nil {
                return err
        }

        err = kvdb.Update(bdb, func(tx kvdb.RwTx) error {
                if _, err := tx.CreateTopLevelBucket(openChannelBucket); err != nil {
                        return err
                }
                if _, err := tx.CreateTopLevelBucket(closedChannelBucket); err != nil {
                        return err
                }

                if _, err := tx.CreateTopLevelBucket(invoiceBucket); err != nil {
                        return err
                }

                if _, err := tx.CreateTopLevelBucket(paymentBucket); err != nil {
                        return err
                }

                nodes, err := tx.CreateTopLevelBucket(nodeBucket)
                if err != nil {
                        return err
                }
                _, err = nodes.CreateBucket(aliasIndexBucket)
                if err != nil {
                        return err
                }
                _, err = nodes.CreateBucket(nodeUpdateIndexBucket)
                if err != nil {
                        return err
                }

                edges, err := tx.CreateTopLevelBucket(edgeBucket)
                if err != nil {
                        return err
                }
                if _, err := edges.CreateBucket(edgeIndexBucket); err != nil {
                        return err
                }
                if _, err := edges.CreateBucket(edgeUpdateIndexBucket); err != nil {
                        return err
                }
                if _, err := edges.CreateBucket(channelPointBucket); err != nil {
                        return err
                }
                if _, err := edges.CreateBucket(zombieBucket); err != nil {
                        return err
                }

                graphMeta, err := tx.CreateTopLevelBucket(graphMetaBucket)
                if err != nil {
                        return err
                }
                _, err = graphMeta.CreateBucket(pruneLogBucket)
                if err != nil {
                        return err
                }

                if _, err := tx.CreateTopLevelBucket(metaBucket); err != nil {
                        return err
                }

                meta := &Meta{
                        DbVersionNumber: 0,
                }
                return putMeta(meta, tx)
        }, func() {})
        if err != nil {
                return fmt.Errorf("unable to create new channeldb")
        }

        return bdb.Close()
}

// fileExists returns true if the file exists, and false otherwise.
func fileExists(path string) bool {
        if _, err := os.Stat(path); err != nil {
                if os.IsNotExist(err) {
                        return false
                }
        }

        return true
}

// FetchClosedChannels attempts to fetch all closed channels from the database.
// The pendingOnly bool toggles if channels that aren't yet fully closed should
// be returned in the response or not. When a channel was cooperatively closed,
// it becomes fully closed after a single confirmation.  When a channel was
// forcibly closed, it will become fully closed after _all_ the pending funds
// (if any) have been swept.
func (d *DB) FetchClosedChannels(pendingOnly bool) ([]*ChannelCloseSummary, error) {
        var chanSummaries []*ChannelCloseSummary

        if err := kvdb.View(d, func(tx kvdb.RTx) error {
                closeBucket := tx.ReadBucket(closedChannelBucket)
                if closeBucket == nil {
                        return ErrNoClosedChannels
                }

                return closeBucket.ForEach(func(chanID []byte, summaryBytes []byte) error {
                        summaryReader := bytes.NewReader(summaryBytes)
                        chanSummary, err := deserializeCloseChannelSummary(summaryReader)
                        if err != nil {
                                return err
                        }

                        // If the query specified to only include pending
                        // channels, then we'll skip any channels which aren't
                        // currently pending.
                        if !chanSummary.IsPending && pendingOnly {
                                return nil
                        }

                        chanSummaries = append(chanSummaries, chanSummary)
                        return nil
                })
        }, func() {
                chanSummaries = nil
        }); err != nil {
                return nil, err
        }

        return chanSummaries, nil
}

// ChannelGraph returns a new instance of the directed channel graph.
func (d *DB) ChannelGraph() *ChannelGraph {
        return d.graph
}

1	package migration_01_to_11
2
3	import (
4	"bytes"
5	"encoding/binary"
6	"fmt"
7	"os"
8	"path/filepath"
9	"time"
10
11	"github.com/lightningnetwork/lnd/kvdb"
12	)
13
14	const (
15	dbName = "channel.db"
16	dbFilePermission = 0600
17	)
18
19	// migration is a function which takes a prior outdated version of the database
20	// instances and mutates the key/bucket structure to arrive at a more
21	// up-to-date version of the database.
22	type migration func(tx kvdb.RwTx) error
23
24	var (
25	// Big endian is the preferred byte order, due to cursor scans over
26	// integer keys iterating in order.
27	byteOrder = binary.BigEndian
28	)
29
30	// DB is the primary datastore for the lnd daemon. The database stores
31	// information related to nodes, routing data, open/closed channels, fee
32	// schedules, and reputation data.
33	type DB struct {
34	kvdb.Backend
35	dbPath string
36	graph *ChannelGraph
37	now func() time.Time
38	}
39
40	// Open opens an existing channeldb. Any necessary schemas migrations due to
41	// updates will take place as necessary.
UNCOV 42	func Open(dbPath string, modifiers ...OptionModifier) (*DB, error) {	×
UNCOV 43	path := filepath.Join(dbPath, dbName)	×
UNCOV 44		×
UNCOV 45	if !fileExists(path) {	×
UNCOV 46	if err := createChannelDB(dbPath); err != nil {	×
47	return nil, err	×
48	}	×
49	}
50
UNCOV 51	opts := DefaultOptions()	×
UNCOV 52	for _, modifier := range modifiers {	×
53	modifier(&opts)	×
54	}	×
55
56	// Specify bbolt freelist options to reduce heap pressure in case the
57	// freelist grows to be very large.
UNCOV 58	bdb, err := kvdb.Open(	×
UNCOV 59	kvdb.BoltBackendName, path,	×
UNCOV 60	opts.NoFreelistSync, opts.DBTimeout,	×
UNCOV 61	)	×
UNCOV 62	if err != nil {	×
63	return nil, err	×
64	}	×
65
UNCOV 66	chanDB := &DB{	×
UNCOV 67	Backend: bdb,	×
UNCOV 68	dbPath: dbPath,	×
UNCOV 69	now: time.Now,	×
UNCOV 70	}	×
UNCOV 71	chanDB.graph = newChannelGraph(	×
UNCOV 72	chanDB, opts.RejectCacheSize, opts.ChannelCacheSize,	×
UNCOV 73	)	×
UNCOV 74		×
UNCOV 75	return chanDB, nil	×
76	}
77
78	// createChannelDB creates and initializes a fresh version of channeldb. In
79	// the case that the target path has not yet been created or doesn't yet exist,
80	// then the path is created. Additionally, all required top-level buckets used
81	// within the database are created.
UNCOV 82	func createChannelDB(dbPath string) error {	×
UNCOV 83	if !fileExists(dbPath) {	×
84	if err := os.MkdirAll(dbPath, 0700); err != nil {	×
85	return err	×
86	}	×
87	}
88
UNCOV 89	path := filepath.Join(dbPath, dbName)	×
UNCOV 90	bdb, err := kvdb.Create(	×
UNCOV 91	kvdb.BoltBackendName, path, false, kvdb.DefaultDBTimeout,	×
UNCOV 92	)	×
UNCOV 93	if err != nil {	×
94	return err	×
95	}	×
96
UNCOV 97	err = kvdb.Update(bdb, func(tx kvdb.RwTx) error {	×
UNCOV 98	if _, err := tx.CreateTopLevelBucket(openChannelBucket); err != nil {	×
99	return err	×
100	}	×
UNCOV 101	if _, err := tx.CreateTopLevelBucket(closedChannelBucket); err != nil {	×
102	return err	×
103	}	×
104
UNCOV 105	if _, err := tx.CreateTopLevelBucket(invoiceBucket); err != nil {	×
106	return err	×
107	}	×
108
UNCOV 109	if _, err := tx.CreateTopLevelBucket(paymentBucket); err != nil {	×
110	return err	×
111	}	×
112
UNCOV 113	nodes, err := tx.CreateTopLevelBucket(nodeBucket)	×
UNCOV 114	if err != nil {	×
115	return err	×
116	}	×
UNCOV 117	_, err = nodes.CreateBucket(aliasIndexBucket)	×
UNCOV 118	if err != nil {	×
119	return err	×
120	}	×
UNCOV 121	_, err = nodes.CreateBucket(nodeUpdateIndexBucket)	×
UNCOV 122	if err != nil {	×
123	return err	×
124	}	×
125
UNCOV 126	edges, err := tx.CreateTopLevelBucket(edgeBucket)	×
UNCOV 127	if err != nil {	×
128	return err	×
129	}	×
UNCOV 130	if _, err := edges.CreateBucket(edgeIndexBucket); err != nil {	×
131	return err	×
132	}	×
UNCOV 133	if _, err := edges.CreateBucket(edgeUpdateIndexBucket); err != nil {	×
134	return err	×
135	}	×
UNCOV 136	if _, err := edges.CreateBucket(channelPointBucket); err != nil {	×
137	return err	×
138	}	×
UNCOV 139	if _, err := edges.CreateBucket(zombieBucket); err != nil {	×
140	return err	×
141	}	×
142
UNCOV 143	graphMeta, err := tx.CreateTopLevelBucket(graphMetaBucket)	×
UNCOV 144	if err != nil {	×
145	return err	×
146	}	×
UNCOV 147	_, err = graphMeta.CreateBucket(pruneLogBucket)	×
UNCOV 148	if err != nil {	×
149	return err	×
150	}	×
151
UNCOV 152	if _, err := tx.CreateTopLevelBucket(metaBucket); err != nil {	×
153	return err	×
154	}	×
155
UNCOV 156	meta := &Meta{	×
UNCOV 157	DbVersionNumber: 0,	×
UNCOV 158	}	×
UNCOV 159	return putMeta(meta, tx)	×
UNCOV 160	}, func() {})	×
UNCOV 161	if err != nil {	×
162	return fmt.Errorf("unable to create new channeldb")	×
163	}	×
164
UNCOV 165	return bdb.Close()	×
166	}
167
168	// fileExists returns true if the file exists, and false otherwise.
UNCOV 169	func fileExists(path string) bool {	×
UNCOV 170	if _, err := os.Stat(path); err != nil {	×
UNCOV 171	if os.IsNotExist(err) {	×
UNCOV 172	return false	×
UNCOV 173	}	×
174	}
175
UNCOV 176	return true	×
177	}
178
179	// FetchClosedChannels attempts to fetch all closed channels from the database.
180	// The pendingOnly bool toggles if channels that aren't yet fully closed should
181	// be returned in the response or not. When a channel was cooperatively closed,
182	// it becomes fully closed after a single confirmation. When a channel was
183	// forcibly closed, it will become fully closed after _all_ the pending funds
184	// (if any) have been swept.
UNCOV 185	func (d DB) FetchClosedChannels(pendingOnly bool) ([]ChannelCloseSummary, error) {	×
UNCOV 186	var chanSummaries []*ChannelCloseSummary	×
UNCOV 187		×
UNCOV 188	if err := kvdb.View(d, func(tx kvdb.RTx) error {	×
UNCOV 189	closeBucket := tx.ReadBucket(closedChannelBucket)	×
UNCOV 190	if closeBucket == nil {	×
191	return ErrNoClosedChannels	×
192	}	×
193
UNCOV 194	return closeBucket.ForEach(func(chanID []byte, summaryBytes []byte) error {	×
UNCOV 195	summaryReader := bytes.NewReader(summaryBytes)	×
UNCOV 196	chanSummary, err := deserializeCloseChannelSummary(summaryReader)	×
UNCOV 197	if err != nil {	×
198	return err	×
199	}	×
200
201	// If the query specified to only include pending
202	// channels, then we'll skip any channels which aren't
203	// currently pending.
UNCOV 204	if !chanSummary.IsPending && pendingOnly {	×
205	return nil	×
206	}	×
207
UNCOV 208	chanSummaries = append(chanSummaries, chanSummary)	×
UNCOV 209	return nil	×
210	})
UNCOV 211	}, func() {	×
UNCOV 212	chanSummaries = nil	×
UNCOV 213	}); err != nil {	×
214	return nil, err	×
215	}	×
216
UNCOV 217	return chanSummaries, nil	×
218	}
219
220	// ChannelGraph returns a new instance of the directed channel graph.
UNCOV 221	func (d DB) ChannelGraph() ChannelGraph {	×
UNCOV 222	return d.graph	×
UNCOV 223	}	×

lightningnetwork / lnd / 13566028875

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