15561477203

Committed 10 Jun 2025 01:54PM UTC coverage: 58.351% (-10.1%) from 68.487%

Build # 15561477203

Build Type

Pull #9356

github

Committed by

web-flow

Commit Message

Merge 6440b25db into c6d6d4c0b

Pull Request Pull Request #9356: lnrpc: add incoming/outgoing channel ids filter to forwarding history request

Run Details

33 of 36 new or added lines in 2 files covered. (91.67%)

28366 existing lines in 455 files now uncovered.

97715 of 167461 relevant lines covered (58.35%)

1.81 hits per line

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

0.0

/channeldb/migration25/migration.go

package migration25

import (
        "bytes"
        "fmt"

        mig "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11"
        "github.com/lightningnetwork/lnd/kvdb"
)

var (
        // openChanBucket stores all the currently open channels. This bucket
        // has a second, nested bucket which is keyed by a node's ID. Within
        // that node ID bucket, all attributes required to track, update, and
        // close a channel are stored.
        openChannelBucket = []byte("open-chan-bucket")

        // chanInfoKey can be accessed within the bucket for a channel
        // (identified by its chanPoint). This key stores all the static
        // information for a channel which is decided at the end of  the
        // funding flow.
        chanInfoKey = []byte("chan-info-key")

        // ErrNoChanDBExists is returned when a channel bucket hasn't been
        // created.
        ErrNoChanDBExists = fmt.Errorf("channel db has not yet been created")

        // ErrNoActiveChannels  is returned when there is no active (open)
        // channels within the database.
        ErrNoActiveChannels = fmt.Errorf("no active channels exist")

        // ErrChannelNotFound is returned when we attempt to locate a channel
        // for a specific chain, but it is not found.
        ErrChannelNotFound = fmt.Errorf("channel not found")
)

// MigrateInitialBalances patches the two new fields, InitialLocalBalance and
// InitialRemoteBalance, for all the open channels. It does so by reading the
// revocation log at height 0 to learn the initial balances and then updates
// the channel's info.
// The channel info is saved in the nested bucket which is accessible via
// nodePub:chainHash:chanPoint. If any of the sub-buckets turns out to be nil,
// we will log the error and continue to process the rest.
func MigrateInitialBalances(tx kvdb.RwTx) error {
        log.Infof("Migrating initial local and remote balances...")

        openChanBucket := tx.ReadWriteBucket(openChannelBucket)

        // If no bucket is found, we can exit early.
        if openChanBucket == nil {
                return nil
        }

        // Read a list of open channels.
        channels, err := findOpenChannels(openChanBucket)
        if err != nil {
                return err
        }

        // Migrate the balances.
        for _, c := range channels {
                if err := migrateBalances(tx, c); err != nil {
                        return err
                }
        }

        return err
}

// findOpenChannels finds all open channels.
func findOpenChannels(openChanBucket kvdb.RBucket) ([]*OpenChannel, error) {
        channels := []*OpenChannel{}

        // readChannel is a helper closure that reads the channel info from the
        // channel bucket.
        readChannel := func(chainBucket kvdb.RBucket, cp []byte) error {
                c := &OpenChannel{}

                // Read the sub-bucket level 3.
                chanBucket := chainBucket.NestedReadBucket(
                        cp,
                )
                if chanBucket == nil {
                        log.Errorf("unable to read bucket for chanPoint=%x", cp)
                        return nil
                }
                // Get the old channel info.
                if err := fetchChanInfo(chanBucket, c, true); err != nil {
                        return fmt.Errorf("unable to fetch chan info: %w", err)
                }

                // Fetch the channel commitments, which are useful for freshly
                // open channels as they don't have any revocation logs and
                // their current commitments reflect the initial balances.
                if err := FetchChanCommitments(chanBucket, c); err != nil {
                        return fmt.Errorf("unable to fetch chan commits: %w",
                                err)
                }

                channels = append(channels, c)

                return nil
        }

        // Iterate the root bucket.
        err := openChanBucket.ForEach(func(nodePub, v []byte) error {
                // Ensure that this is a key the same size as a pubkey, and
                // also that it leads directly to a bucket.
                if len(nodePub) != 33 || v != nil {
                        return nil
                }

                // Read the sub-bucket level 1.
                nodeChanBucket := openChanBucket.NestedReadBucket(nodePub)
                if nodeChanBucket == nil {
                        log.Errorf("no bucket for node %x", nodePub)
                        return nil
                }

                // Iterate the bucket.
                return nodeChanBucket.ForEach(func(chainHash, _ []byte) error {
                        // Read the sub-bucket level 2.
                        chainBucket := nodeChanBucket.NestedReadBucket(
                                chainHash,
                        )
                        if chainBucket == nil {
                                log.Errorf("unable to read bucket for chain=%x",
                                        chainHash)
                                return nil
                        }

                        // Iterate the bucket.
                        return chainBucket.ForEach(func(cp, _ []byte) error {
                                return readChannel(chainBucket, cp)
                        })
                })
        })

        if err != nil {
                return nil, err
        }

        return channels, nil
}

// migrateBalances queries the revocation log at height 0 to find the initial
// balances and save them to the channel info.
func migrateBalances(tx kvdb.RwTx, c *OpenChannel) error {
        // Get the bucket.
        chanBucket, err := FetchChanBucket(tx, c)
        if err != nil {
                return err
        }
        // Get the initial balances.
        localAmt, remoteAmt, err := c.balancesAtHeight(chanBucket, 0)
        if err != nil {
                return fmt.Errorf("unable to get initial balances: %w", err)
        }

        c.InitialLocalBalance = localAmt
        c.InitialRemoteBalance = remoteAmt

        // Update the channel info.
        if err := putChanInfo(chanBucket, c, false); err != nil {
                return fmt.Errorf("unable to put chan info: %w", err)
        }

        return nil
}

// FetchChanBucket is a helper function that returns the bucket where a
// channel's data resides in given: the public key for the node, the outpoint,
// and the chainhash that the channel resides on.
func FetchChanBucket(tx kvdb.RwTx, c *OpenChannel) (kvdb.RwBucket, error) {
        // First fetch the top level bucket which stores all data related to
        // current, active channels.
        openChanBucket := tx.ReadWriteBucket(openChannelBucket)
        if openChanBucket == nil {
                return nil, ErrNoChanDBExists
        }

        // Within this top level bucket, fetch the bucket dedicated to storing
        // open channel data specific to the remote node.
        nodePub := c.IdentityPub.SerializeCompressed()
        nodeChanBucket := openChanBucket.NestedReadWriteBucket(nodePub)
        if nodeChanBucket == nil {
                return nil, ErrNoActiveChannels
        }

        // We'll then recurse down an additional layer in order to fetch the
        // bucket for this particular chain.
        chainBucket := nodeChanBucket.NestedReadWriteBucket(c.ChainHash[:])
        if chainBucket == nil {
                return nil, ErrNoActiveChannels
        }

        // With the bucket for the node and chain fetched, we can now go down
        // another level, for this channel itself.
        var chanPointBuf bytes.Buffer
        err := mig.WriteOutpoint(&chanPointBuf, &c.FundingOutpoint)
        if err != nil {
                return nil, err
        }

        chanBucket := chainBucket.NestedReadWriteBucket(chanPointBuf.Bytes())
        if chanBucket == nil {
                return nil, ErrChannelNotFound
        }

        return chanBucket, nil
}

1	package migration25
2
3	import (
4	"bytes"
5	"fmt"
6
7	mig "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11"
8	"github.com/lightningnetwork/lnd/kvdb"
9	)
10
11	var (
12	// openChanBucket stores all the currently open channels. This bucket
13	// has a second, nested bucket which is keyed by a node's ID. Within
14	// that node ID bucket, all attributes required to track, update, and
15	// close a channel are stored.
16	openChannelBucket = []byte("open-chan-bucket")
17
18	// chanInfoKey can be accessed within the bucket for a channel
19	// (identified by its chanPoint). This key stores all the static
20	// information for a channel which is decided at the end of the
21	// funding flow.
22	chanInfoKey = []byte("chan-info-key")
23
24	// ErrNoChanDBExists is returned when a channel bucket hasn't been
25	// created.
26	ErrNoChanDBExists = fmt.Errorf("channel db has not yet been created")
27
28	// ErrNoActiveChannels is returned when there is no active (open)
29	// channels within the database.
30	ErrNoActiveChannels = fmt.Errorf("no active channels exist")
31
32	// ErrChannelNotFound is returned when we attempt to locate a channel
33	// for a specific chain, but it is not found.
34	ErrChannelNotFound = fmt.Errorf("channel not found")
35	)
36
37	// MigrateInitialBalances patches the two new fields, InitialLocalBalance and
38	// InitialRemoteBalance, for all the open channels. It does so by reading the
39	// revocation log at height 0 to learn the initial balances and then updates
40	// the channel's info.
41	// The channel info is saved in the nested bucket which is accessible via
42	// nodePub:chainHash:chanPoint. If any of the sub-buckets turns out to be nil,
43	// we will log the error and continue to process the rest.
UNCOV 44	func MigrateInitialBalances(tx kvdb.RwTx) error {	×
UNCOV 45	log.Infof("Migrating initial local and remote balances...")	×
UNCOV 46		×
UNCOV 47	openChanBucket := tx.ReadWriteBucket(openChannelBucket)	×
UNCOV 48		×
UNCOV 49	// If no bucket is found, we can exit early.	×
UNCOV 50	if openChanBucket == nil {	×
51	return nil	×
52	}	×
53
54	// Read a list of open channels.
UNCOV 55	channels, err := findOpenChannels(openChanBucket)	×
UNCOV 56	if err != nil {	×
57	return err	×
58	}	×
59
60	// Migrate the balances.
UNCOV 61	for _, c := range channels {	×
UNCOV 62	if err := migrateBalances(tx, c); err != nil {	×
UNCOV 63	return err	×
UNCOV 64	}	×
65	}
66
UNCOV 67	return err	×
68	}
69
70	// findOpenChannels finds all open channels.
UNCOV 71	func findOpenChannels(openChanBucket kvdb.RBucket) ([]*OpenChannel, error) {	×
UNCOV 72	channels := []*OpenChannel{}	×
UNCOV 73		×
UNCOV 74	// readChannel is a helper closure that reads the channel info from the	×
UNCOV 75	// channel bucket.	×
UNCOV 76	readChannel := func(chainBucket kvdb.RBucket, cp []byte) error {	×
UNCOV 77	c := &OpenChannel{}	×
UNCOV 78		×
UNCOV 79	// Read the sub-bucket level 3.	×
UNCOV 80	chanBucket := chainBucket.NestedReadBucket(	×
UNCOV 81	cp,	×
UNCOV 82	)	×
UNCOV 83	if chanBucket == nil {	×
84	log.Errorf("unable to read bucket for chanPoint=%x", cp)	×
85	return nil	×
86	}	×
87	// Get the old channel info.
UNCOV 88	if err := fetchChanInfo(chanBucket, c, true); err != nil {	×
89	return fmt.Errorf("unable to fetch chan info: %w", err)	×
90	}	×
91
92	// Fetch the channel commitments, which are useful for freshly
93	// open channels as they don't have any revocation logs and
94	// their current commitments reflect the initial balances.
UNCOV 95	if err := FetchChanCommitments(chanBucket, c); err != nil {	×
96	return fmt.Errorf("unable to fetch chan commits: %w",	×
97	err)	×
98	}	×
99
UNCOV 100	channels = append(channels, c)	×
UNCOV 101		×
UNCOV 102	return nil	×
103	}
104
105	// Iterate the root bucket.
UNCOV 106	err := openChanBucket.ForEach(func(nodePub, v []byte) error {	×
UNCOV 107	// Ensure that this is a key the same size as a pubkey, and	×
UNCOV 108	// also that it leads directly to a bucket.	×
UNCOV 109	if len(nodePub) != 33 \|\| v != nil {	×
110	return nil	×
111	}	×
112
113	// Read the sub-bucket level 1.
UNCOV 114	nodeChanBucket := openChanBucket.NestedReadBucket(nodePub)	×
UNCOV 115	if nodeChanBucket == nil {	×
116	log.Errorf("no bucket for node %x", nodePub)	×
117	return nil	×
118	}	×
119
120	// Iterate the bucket.
UNCOV 121	return nodeChanBucket.ForEach(func(chainHash, _ []byte) error {	×
UNCOV 122	// Read the sub-bucket level 2.	×
UNCOV 123	chainBucket := nodeChanBucket.NestedReadBucket(	×
UNCOV 124	chainHash,	×
UNCOV 125	)	×
UNCOV 126	if chainBucket == nil {	×
127	log.Errorf("unable to read bucket for chain=%x",	×
128	chainHash)	×
129	return nil	×
130	}	×
131
132	// Iterate the bucket.
UNCOV 133	return chainBucket.ForEach(func(cp, _ []byte) error {	×
UNCOV 134	return readChannel(chainBucket, cp)	×
UNCOV 135	})	×
136	})
137	})
138
UNCOV 139	if err != nil {	×
140	return nil, err	×
141	}	×
142
UNCOV 143	return channels, nil	×
144	}
145
146	// migrateBalances queries the revocation log at height 0 to find the initial
147	// balances and save them to the channel info.
UNCOV 148	func migrateBalances(tx kvdb.RwTx, c *OpenChannel) error {	×
UNCOV 149	// Get the bucket.	×
UNCOV 150	chanBucket, err := FetchChanBucket(tx, c)	×
UNCOV 151	if err != nil {	×
152	return err	×
153	}	×
154	// Get the initial balances.
UNCOV 155	localAmt, remoteAmt, err := c.balancesAtHeight(chanBucket, 0)	×
UNCOV 156	if err != nil {	×
UNCOV 157	return fmt.Errorf("unable to get initial balances: %w", err)	×
UNCOV 158	}	×
159
UNCOV 160	c.InitialLocalBalance = localAmt	×
UNCOV 161	c.InitialRemoteBalance = remoteAmt	×
UNCOV 162		×
UNCOV 163	// Update the channel info.	×
UNCOV 164	if err := putChanInfo(chanBucket, c, false); err != nil {	×
165	return fmt.Errorf("unable to put chan info: %w", err)	×
166	}	×
167
UNCOV 168	return nil	×
169	}
170
171	// FetchChanBucket is a helper function that returns the bucket where a
172	// channel's data resides in given: the public key for the node, the outpoint,
173	// and the chainhash that the channel resides on.
UNCOV 174	func FetchChanBucket(tx kvdb.RwTx, c *OpenChannel) (kvdb.RwBucket, error) {	×
UNCOV 175	// First fetch the top level bucket which stores all data related to	×
UNCOV 176	// current, active channels.	×
UNCOV 177	openChanBucket := tx.ReadWriteBucket(openChannelBucket)	×
UNCOV 178	if openChanBucket == nil {	×
179	return nil, ErrNoChanDBExists	×
180	}	×
181
182	// Within this top level bucket, fetch the bucket dedicated to storing
183	// open channel data specific to the remote node.
UNCOV 184	nodePub := c.IdentityPub.SerializeCompressed()	×
UNCOV 185	nodeChanBucket := openChanBucket.NestedReadWriteBucket(nodePub)	×
UNCOV 186	if nodeChanBucket == nil {	×
187	return nil, ErrNoActiveChannels	×
188	}	×
189
190	// We'll then recurse down an additional layer in order to fetch the
191	// bucket for this particular chain.
UNCOV 192	chainBucket := nodeChanBucket.NestedReadWriteBucket(c.ChainHash[:])	×
UNCOV 193	if chainBucket == nil {	×
194	return nil, ErrNoActiveChannels	×
195	}	×
196
197	// With the bucket for the node and chain fetched, we can now go down
198	// another level, for this channel itself.
UNCOV 199	var chanPointBuf bytes.Buffer	×
UNCOV 200	err := mig.WriteOutpoint(&chanPointBuf, &c.FundingOutpoint)	×
UNCOV 201	if err != nil {	×
202	return nil, err	×
203	}	×
204
UNCOV 205	chanBucket := chainBucket.NestedReadWriteBucket(chanPointBuf.Bytes())	×
UNCOV 206	if chanBucket == nil {	×
207	return nil, ErrChannelNotFound	×
208	}	×
209
UNCOV 210	return chanBucket, nil	×
211	}

lightningnetwork / lnd / 15561477203

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